sync: port upstream OpenCode, timestamps, client probe (#3608, #4112, #4132, #4134, #4006, #4137)

apps/server/src/environment/ServerEnvironment.test.ts

@@ -67,6 +67,7 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => {
 
       expect(first.environmentId).toBe(second.environmentId);
       expect(second.capabilities.repositoryIdentity).toBe(true);
+      expect(second.capabilities.connectionProbe).toBe(true);
     }),
   );
 

apps/server/src/environment/ServerEnvironment.ts

@@ -135,6 +135,7 @@ export const make = Effect.gen(function* () {
     serverVersion: packageJson.version,
     capabilities: {
       repositoryIdentity: true,
+      connectionProbe: true,
     },
   };
 

apps/server/src/orchestration/Normalizer.test.ts

@@ -0,0 +1,73 @@
+import { describe, expect, it } from "vite-plus/test";
+import {
+  CommandId,
+  type ClientOrchestrationCommand,
+  MessageId,
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+} from "@t3tools/contracts";
+
+import { canonicalizeClientCommandTimestamps } from "./Normalizer.ts";
+
+const clientCreatedAt = "2031-01-01T00:00:00.000Z";
+const serverReceivedAt = "2026-07-18T00:00:00.000Z";
+
+describe("canonicalizeClientCommandTimestamps", () => {
+  it("replaces a client command timestamp with the server receipt timestamp", () => {
+    const command: ClientOrchestrationCommand = {
+      type: "project.create",
+      commandId: CommandId.make("command-1"),
+      projectId: ProjectId.make("project-1"),
+      title: "Clock-safe project",
+      workspaceRoot: "/tmp/clock-safe-project",
+      createdAt: clientCreatedAt,
+    };
+
+    expect(canonicalizeClientCommandTimestamps(command, serverReceivedAt)).toEqual({
+      ...command,
+      createdAt: serverReceivedAt,
+    });
+  });
+
+  it("replaces both timestamps when the first turn bootstraps a thread", () => {
+    const command: ClientOrchestrationCommand = {
+      type: "thread.turn.start",
+      commandId: CommandId.make("command-2"),
+      threadId: ThreadId.make("thread-1"),
+      message: {
+        messageId: MessageId.make("message-1"),
+        role: "user",
+        text: "Start a thread",
+        attachments: [],
+      },
+      runtimeMode: "full-access",
+      interactionMode: "default",
+      bootstrap: {
+        createThread: {
+          projectId: ProjectId.make("project-1"),
+          title: "Clock-safe thread",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5.4",
+          },
+          runtimeMode: "full-access",
+          interactionMode: "default",
+          branch: null,
+          worktreePath: null,
+          createdAt: clientCreatedAt,
+        },
+      },
+      createdAt: clientCreatedAt,
+    };
+
+    const result = canonicalizeClientCommandTimestamps(command, serverReceivedAt);
+
+    expect(result.type).toBe("thread.turn.start");
+    if (result.type !== "thread.turn.start") {
+      throw new Error("Expected a thread.turn.start command");
+    }
+    expect(result.createdAt).toBe(serverReceivedAt);
+    expect(result.bootstrap?.createThread?.createdAt).toBe(serverReceivedAt);
+  });
+});

apps/server/src/orchestration/Normalizer.ts

@@ -1,8 +1,10 @@
+import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Path from "effect/Path";
 import {
   type ClientOrchestrationCommand,
+  type IsoDateTime,
   type OrchestrationCommand,
   OrchestrationDispatchCommandError,
   PROVIDER_SEND_TURN_MAX_IMAGE_BYTES,
@@ -13,8 +15,38 @@ import { ServerConfig } from "../config.ts";
 import { parseBase64DataUrl } from "../imageMime.ts";
 import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
 
+export const canonicalizeClientCommandTimestamps = (
+  command: ClientOrchestrationCommand,
+  receivedAt: IsoDateTime,
+): ClientOrchestrationCommand => {
+  const canonicalCommand =
+    "createdAt" in command
+      ? {
+          ...command,
+          createdAt: receivedAt,
+        }
+      : command;
+
+  if (canonicalCommand.type !== "thread.turn.start" || !canonicalCommand.bootstrap?.createThread) {
+    return canonicalCommand;
+  }
+
+  return {
+    ...canonicalCommand,
+    bootstrap: {
+      ...canonicalCommand.bootstrap,
+      createThread: {
+        ...canonicalCommand.bootstrap.createThread,
+        createdAt: receivedAt,
+      },
+    },
+  };
+};
+
 export const normalizeDispatchCommand = (command: ClientOrchestrationCommand) =>
   Effect.gen(function* () {
+    const receivedAt = DateTime.formatIso(yield* DateTime.now);
+    const canonicalCommand = canonicalizeClientCommandTimestamps(command, receivedAt);
     const fileSystem = yield* FileSystem.FileSystem;
     const path = yield* Path.Path;
     const serverConfig = yield* ServerConfig;
@@ -47,30 +79,33 @@ export const normalizeDispatchCommand = (command: ClientOrchestrationCommand) =>
           ),
         );
 
-    if (command.type === "project.create") {
+    if (canonicalCommand.type === "project.create") {
       return {
-        ...command,
+        ...canonicalCommand,
         workspaceRoot: yield* normalizeProjectWorkspaceRootForCreate(
-          command.workspaceRoot,
-          command.createWorkspaceRootIfMissing,
+          canonicalCommand.workspaceRoot,
+          canonicalCommand.createWorkspaceRootIfMissing,
         ),
-        createWorkspaceRootIfMissing: command.createWorkspaceRootIfMissing === true,
+        createWorkspaceRootIfMissing: canonicalCommand.createWorkspaceRootIfMissing === true,
       } satisfies OrchestrationCommand;
     }
 
-    if (command.type === "project.meta.update" && command.workspaceRoot !== undefined) {
+    if (
+      canonicalCommand.type === "project.meta.update" &&
+      canonicalCommand.workspaceRoot !== undefined
+    ) {
       return {
-        ...command,
-        workspaceRoot: yield* normalizeProjectWorkspaceRoot(command.workspaceRoot),
+        ...canonicalCommand,
+        workspaceRoot: yield* normalizeProjectWorkspaceRoot(canonicalCommand.workspaceRoot),
       } satisfies OrchestrationCommand;
     }
 
-    if (command.type !== "thread.turn.start") {
-      return command as OrchestrationCommand;
+    if (canonicalCommand.type !== "thread.turn.start") {
+      return canonicalCommand as OrchestrationCommand;
     }
 
     const normalizedAttachments = yield* Effect.forEach(
-      command.message.attachments,
+      canonicalCommand.message.attachments,
       (attachment) =>
         Effect.gen(function* () {
           const parsed = parseBase64DataUrl(attachment.dataUrl);
@@ -87,7 +122,7 @@ export const normalizeDispatchCommand = (command: ClientOrchestrationCommand) =>
             });
           }
 
-          const attachmentId = createAttachmentId(command.threadId);
+          const attachmentId = createAttachmentId(canonicalCommand.threadId);
           if (!attachmentId) {
             return yield* new OrchestrationDispatchCommandError({
               message: "Failed to create a safe attachment id.",
@@ -135,9 +170,9 @@ export const normalizeDispatchCommand = (command: ClientOrchestrationCommand) =>
     );
 
     return {
-      ...command,
+      ...canonicalCommand,
       message: {
-        ...command.message,
+        ...canonicalCommand.message,
         attachments: normalizedAttachments,
       },
     } satisfies OrchestrationCommand;

apps/server/src/provider/Layers/OpenCodeAdapter.test.ts

@@ -248,6 +248,38 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => {
     }),
   );
 
+  it.effect("fails sendTurn for missing sessions through the typed error channel", () =>
+    Effect.gen(function* () {
+      const adapter = yield* OpenCodeAdapter;
+      const result = yield* adapter
+        .sendTurn({
+          threadId: asThreadId("thread-opencode-missing-send"),
+          input: "hello",
+          attachments: [],
+        })
+        .pipe(Effect.result);
+
+      NodeAssert.equal(result._tag, "Failure");
+      NodeAssert.equal(result.failure._tag, "ProviderAdapterSessionNotFoundError");
+      NodeAssert.equal(result.failure.provider, "opencode");
+      NodeAssert.equal(result.failure.threadId, "thread-opencode-missing-send");
+    }),
+  );
+
+  it.effect("fails stopSession for missing sessions through the typed error channel", () =>
+    Effect.gen(function* () {
+      const adapter = yield* OpenCodeAdapter;
+      const result = yield* adapter
+        .stopSession(asThreadId("thread-opencode-missing-stop"))
+        .pipe(Effect.result);
+
+      NodeAssert.equal(result._tag, "Failure");
+      NodeAssert.equal(result.failure._tag, "ProviderAdapterSessionNotFoundError");
+      NodeAssert.equal(result.failure.provider, "opencode");
+      NodeAssert.equal(result.failure.threadId, "thread-opencode-missing-stop");
+    }),
+  );
+
   it.effect("stops a configured-server session without trying to own server lifecycle", () =>
     Effect.gen(function* () {
       const adapter = yield* OpenCodeAdapter;

apps/server/src/provider/Layers/OpenCodeAdapter.ts

@@ -228,28 +228,25 @@ function appendTurnItem(
   resolveTurnSnapshot(context, turnId).items.push(item);
 }
 
-function ensureSessionContext(
+const ensureSessionContext = Effect.fn("ensureSessionContext")(function* (
   sessions: ReadonlyMap<ThreadId, OpenCodeSessionContext>,
   threadId: ThreadId,
-): OpenCodeSessionContext {
+) {
   const session = sessions.get(threadId);
   if (!session) {
-    throw new ProviderAdapterSessionNotFoundError({
+    return yield* new ProviderAdapterSessionNotFoundError({
       provider: PROVIDER,
       threadId,
     });
   }
-  // `ensureSessionContext` is a sync gate used from both sync helpers and
-  // Effect bodies. `Ref.getUnsafe` is an atomic read of the backing cell —
-  // no fiber suspension required, which keeps this callable everywhere.
-  if (Ref.getUnsafe(session.stopped)) {
-    throw new ProviderAdapterSessionClosedError({
+  if (yield* Ref.get(session.stopped)) {
+    return yield* new ProviderAdapterSessionClosedError({
       provider: PROVIDER,
       threadId,
     });
   }
   return session;
-}
+});
 
 function normalizeQuestionRequest(request: QuestionRequest): ReadonlyArray<UserInputQuestion> {
   return request.questions.map((question, index) => ({
@@ -1167,7 +1164,7 @@ export function makeOpenCodeAdapter(
     );
 
     const sendTurn: OpenCodeAdapterShape["sendTurn"] = Effect.fn("sendTurn")(function* (input) {
-      const context = ensureSessionContext(sessions, input.threadId);
+      const context = yield* ensureSessionContext(sessions, input.threadId);
       // A sendTurn while a turn is active is a steer: OpenCode queues the
       // prompt into the busy session and the work continues as one turn, so
       // the active turn id is reused instead of opening a new turn.
@@ -1291,7 +1288,7 @@ export function makeOpenCodeAdapter(
 
     const interruptTurn: OpenCodeAdapterShape["interruptTurn"] = Effect.fn("interruptTurn")(
       function* (threadId, turnId) {
-        const context = ensureSessionContext(sessions, threadId);
+        const context = yield* ensureSessionContext(sessions, threadId);
         yield* runOpenCodeSdk("session.abort", () =>
           context.client.session.abort({ sessionID: context.openCodeSessionId }),
         ).pipe(Effect.mapError(toRequestError));
@@ -1313,7 +1310,7 @@ export function makeOpenCodeAdapter(
     const respondToRequest: OpenCodeAdapterShape["respondToRequest"] = Effect.fn(
       "respondToRequest",
     )(function* (threadId, requestId, decision) {
-      const context = ensureSessionContext(sessions, threadId);
+      const context = yield* ensureSessionContext(sessions, threadId);
       if (!context.pendingPermissions.has(requestId)) {
         return yield* new ProviderAdapterRequestError({
           provider: PROVIDER,
@@ -1333,7 +1330,7 @@ export function makeOpenCodeAdapter(
     const respondToUserInput: OpenCodeAdapterShape["respondToUserInput"] = Effect.fn(
       "respondToUserInput",
     )(function* (threadId, requestId, answers) {
-      const context = ensureSessionContext(sessions, threadId);
+      const context = yield* ensureSessionContext(sessions, threadId);
       const request = context.pendingQuestions.get(requestId);
       if (!request) {
         return yield* new ProviderAdapterRequestError({
@@ -1355,7 +1352,7 @@ export function makeOpenCodeAdapter(
       function* (threadId) {
         const context = sessions.get(threadId);
         if (!context) {
-          throw new ProviderAdapterSessionNotFoundError({
+          return yield* new ProviderAdapterSessionNotFoundError({
             provider: PROVIDER,
             threadId,
           });
@@ -1385,7 +1382,7 @@ export function makeOpenCodeAdapter(
 
     const readThread: OpenCodeAdapterShape["readThread"] = Effect.fn("readThread")(
       function* (threadId) {
-        const context = ensureSessionContext(sessions, threadId);
+        const context = yield* ensureSessionContext(sessions, threadId);
         const messages = yield* runOpenCodeSdk("session.messages", () =>
           context.client.session.messages({
             sessionID: context.openCodeSessionId,
@@ -1411,7 +1408,7 @@ export function makeOpenCodeAdapter(
 
     const rollbackThread: OpenCodeAdapterShape["rollbackThread"] = Effect.fn("rollbackThread")(
       function* (threadId, numTurns) {
-        const context = ensureSessionContext(sessions, threadId);
+        const context = yield* ensureSessionContext(sessions, threadId);
         const messages = yield* runOpenCodeSdk("session.messages", () =>
           context.client.session.messages({
             sessionID: context.openCodeSessionId,

apps/server/src/provider/opencodeRuntime.ts

@@ -37,7 +37,7 @@ const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.UnknownFromJ
 const OPENCODE_EMPTY_CONFIG_CONTENT = "{}";
 
 const OPENCODE_SERVER_READY_PREFIX = "opencode server listening";
-const DEFAULT_OPENCODE_SERVER_TIMEOUT_MS = 5_000;
+const DEFAULT_OPENCODE_SERVER_TIMEOUT_MS = 30_000;
 const DEFAULT_HOSTNAME = "127.0.0.1";
 export interface OpenCodeServerProcess {
   readonly url: string;

apps/server/src/ws.ts

@@ -284,6 +284,7 @@ const RPC_REQUIRED_SCOPE = new Map<string, AuthEnvironmentScope>([
   [ORCHESTRATION_WS_METHODS.subscribeShell, AuthOrchestrationReadScope],
   [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, AuthOrchestrationReadScope],
   [ORCHESTRATION_WS_METHODS.subscribeThread, AuthOrchestrationReadScope],
+  [WS_METHODS.serverProbe, AuthOrchestrationReadScope],
   [WS_METHODS.serverGetConfig, AuthOrchestrationReadScope],
   [WS_METHODS.serverRefreshProviders, AuthOrchestrationOperateScope],
   [WS_METHODS.serverUpdateProvider, AuthOrchestrationOperateScope],
@@ -1258,6 +1259,10 @@ const makeWsRpcLayer = (
             }),
             { "rpc.aggregate": "orchestration" },
           ),
+        [WS_METHODS.serverProbe]: (_input) =>
+          observeRpcEffect(WS_METHODS.serverProbe, Effect.succeed({}), {
+            "rpc.aggregate": "server",
+          }),
         [WS_METHODS.serverGetConfig]: (_input) =>
           observeRpcEffect(WS_METHODS.serverGetConfig, loadServerConfig, {
             "rpc.aggregate": "server",

packages/client-runtime/src/connection/supervisor.test.ts

@@ -652,7 +652,7 @@ describe("EnvironmentSupervisor", () => {
     }).pipe(Effect.provide(TestClock.layer())),
   );
 
-  it.effect("keeps a healthy session when the application becomes active", () =>
+  it.effect("probes the active session without reconnecting on application activation", () =>
     Effect.gen(function* () {
       const probeCount = yield* Ref.make(0);
       const probeCalled = yield* Deferred.make<void>();