fix(server): hydrate threads created by sibling processes

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -130,6 +130,172 @@ const hasMetricSnapshot = (
   );
 
 describe("OrchestrationEngine", () => {
+  it("hydrates a thread projected by another server before dispatching to it", async () => {
+    const directory = await NodeFSP.mkdtemp(
+      NodePath.join(NodeOS.tmpdir(), "t3-shared-orchestration-"),
+    );
+    const databasePath = NodePath.join(directory, "state.sqlite");
+    const staleSystem = await createOrchestrationSystem(databasePath);
+    const writerSystem = await createOrchestrationSystem(databasePath);
+    const projectId = asProjectId("shared-project");
+    const threadId = ThreadId.make("shared-thread");
+
+    try {
+      await writerSystem.run(
+        writerSystem.engine.dispatch({
+          type: "project.create",
+          commandId: CommandId.make("cmd-shared-project-create"),
+          projectId,
+          title: "Shared project",
+          workspaceRoot: "/tmp/shared-project",
+          defaultModelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5-codex",
+          },
+          createdAt: now(),
+        }),
+      );
+      await writerSystem.run(
+        writerSystem.engine.dispatch({
+          type: "thread.create",
+          commandId: CommandId.make("cmd-shared-thread-create"),
+          threadId,
+          projectId,
+          title: "Shared thread",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5-codex",
+          },
+          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+          runtimeMode: "approval-required",
+          branch: null,
+          worktreePath: null,
+          createdAt: now(),
+        }),
+      );
+
+      // Advance this process past the sibling's events without projecting them
+      // onto its command model, matching two live servers sharing one database.
+      await staleSystem.run(
+        staleSystem.engine.dispatch({
+          type: "project.create",
+          commandId: CommandId.make("cmd-stale-project-create"),
+          projectId: asProjectId("stale-project"),
+          title: "Stale process project",
+          workspaceRoot: "/tmp/stale-project",
+          createdAt: now(),
+        }),
+      );
+
+      const result = await staleSystem.run(
+        staleSystem.engine.dispatch({
+          type: "thread.turn.start",
+          commandId: CommandId.make("cmd-shared-turn-start"),
+          threadId,
+          message: {
+            messageId: asMessageId("msg-shared-turn-start"),
+            role: "user",
+            text: "sent through the other server",
+            attachments: [],
+          },
+          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+          runtimeMode: "approval-required",
+          createdAt: now(),
+        }),
+      );
+
+      expect(result.sequence).toBe(5);
+      const thread = await staleSystem.readThread(threadId);
+      expect(Option.getOrThrow(thread).messages).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({
+            id: asMessageId("msg-shared-turn-start"),
+            text: "sent through the other server",
+          }),
+        ]),
+      );
+    } finally {
+      await writerSystem.dispose();
+      await staleSystem.dispose();
+      await NodeFSP.rm(directory, { recursive: true, force: true });
+    }
+  });
+
+  it("hydrates an archived thread projected by another server before unarchiving it", async () => {
+    const directory = await NodeFSP.mkdtemp(
+      NodePath.join(NodeOS.tmpdir(), "t3-shared-orchestration-archive-"),
+    );
+    const databasePath = NodePath.join(directory, "state.sqlite");
+    const staleSystem = await createOrchestrationSystem(databasePath);
+    const writerSystem = await createOrchestrationSystem(databasePath);
+    const projectId = asProjectId("shared-archive-project");
+    const threadId = ThreadId.make("shared-archive-thread");
+
+    try {
+      await writerSystem.run(
+        writerSystem.engine.dispatch({
+          type: "project.create",
+          commandId: CommandId.make("cmd-shared-archive-project-create"),
+          projectId,
+          title: "Shared archive project",
+          workspaceRoot: "/tmp/shared-archive-project",
+          createdAt: now(),
+        }),
+      );
+      await writerSystem.run(
+        writerSystem.engine.dispatch({
+          type: "thread.create",
+          commandId: CommandId.make("cmd-shared-archive-thread-create"),
+          threadId,
+          projectId,
+          title: "Shared archived thread",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5-codex",
+          },
+          interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+          runtimeMode: "approval-required",
+          branch: null,
+          worktreePath: null,
+          createdAt: now(),
+        }),
+      );
+      await writerSystem.run(
+        writerSystem.engine.dispatch({
+          type: "thread.archive",
+          commandId: CommandId.make("cmd-shared-archive-thread-archive"),
+          threadId,
+        }),
+      );
+
+      await staleSystem.run(
+        staleSystem.engine.dispatch({
+          type: "project.create",
+          commandId: CommandId.make("cmd-stale-archive-project-create"),
+          projectId: asProjectId("stale-archive-project"),
+          title: "Stale archive process project",
+          workspaceRoot: "/tmp/stale-archive-project",
+          createdAt: now(),
+        }),
+      );
+
+      const result = await staleSystem.run(
+        staleSystem.engine.dispatch({
+          type: "thread.unarchive",
+          commandId: CommandId.make("cmd-shared-thread-unarchive"),
+          threadId,
+        }),
+      );
+
+      expect(result.sequence).toBe(5);
+      expect(Option.getOrThrow(await staleSystem.readThread(threadId)).archivedAt).toBeNull();
+    } finally {
+      await writerSystem.dispose();
+      await staleSystem.dispose();
+      await NodeFSP.rm(directory, { recursive: true, force: true });
+    }
+  });
+
   it.each(["running", "stopped"] as const)(
     "sends async answers with a %s session and rejects old duplicate replies",
     async (status) => {

apps/server/src/orchestration/Layers/OrchestrationEngine.ts

@@ -61,10 +61,15 @@ interface CommandEnvelope {
   startedAtMs: number;
 }
 
-function commandToAggregateRef(command: OrchestrationCommand): {
-  readonly aggregateKind: "project" | "thread";
-  readonly aggregateId: ProjectId | ThreadId;
-} {
+function commandToAggregateRef(command: OrchestrationCommand):
+  | {
+      readonly aggregateKind: "project";
+      readonly aggregateId: ProjectId;
+    }
+  | {
+      readonly aggregateKind: "thread";
+      readonly aggregateId: ThreadId;
+    } {
   switch (command.type) {
     case "project.create":
     case "project.meta.update":
@@ -171,6 +176,30 @@ const makeOrchestrationEngine = Effect.gen(function* () {
           });
         }
 
+        // Multiple server processes can share one T3 home. Projection tables make a
+        // thread created by a sibling process visible to clients, but that creation
+        // event is absent from this process's command model. Hydrate only on a miss
+        // so commands for the projected thread do not fail until this server restarts.
+        if (
+          aggregateRef.aggregateKind === "thread" &&
+          envelope.command.type !== "thread.create" &&
+          !commandReadModel.threads.some((thread) => thread.id === aggregateRef.aggregateId)
+        ) {
+          const projectedCommandModel = yield* projectionSnapshotQuery.getCommandReadModel({
+            threadId: aggregateRef.aggregateId,
+          });
+          const projectedThread = projectedCommandModel.threads[0];
+          if (projectedThread !== undefined) {
+            commandReadModel = {
+              ...commandReadModel,
+              threads: [...commandReadModel.threads, projectedThread],
+            };
+            yield* Effect.logDebug("hydrated missing orchestration command thread").pipe(
+              Effect.annotateLogs({ threadId: aggregateRef.aggregateId }),
+            );
+          }
+        }
+
         if (
           envelope.command.type === "thread.auto-settle" &&
           (yield* eventStore.hasEventAfter({

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts

@@ -649,6 +649,15 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
         commandReadModel.threads[0]?.linkedPullRequest,
         snapshot.threads[0]?.linkedPullRequest,
       );
+      const targetedCommandReadModel = yield* snapshotQuery.getCommandReadModel({
+        threadId: ThreadId.make("thread-1"),
+      });
+      const commandThread = targetedCommandReadModel.threads[0];
+      assert.deepEqual(commandThread?.messages, []);
+      assert.deepEqual(commandThread?.activities, []);
+      assert.deepEqual(commandThread?.checkpoints, []);
+      assert.deepEqual(commandThread?.proposedPlans, snapshot.threads[0]?.proposedPlans);
+      assert.deepEqual(commandThread?.pullRequests, expectedPullRequests);
 
       // Without link rows the legacy field is omitted, whatever the old JSON
       // column still holds.
@@ -765,6 +774,16 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
         assert.equal(changedContext.value.session?.providerInstanceId, "claude-secondary");
         assert.equal(changedContext.value.session?.lastError, "Starting another session");
       }
+
+      yield* sql`
+        UPDATE projection_threads
+        SET archived_at = '2026-02-24T00:00:09.000Z'
+        WHERE thread_id = 'thread-1'
+      `;
+      const archivedCommandReadModel = yield* snapshotQuery.getCommandReadModel({
+        threadId: ThreadId.make("thread-1"),
+      });
+      assert.equal(archivedCommandReadModel.threads[0]?.archivedAt, "2026-02-24T00:00:09.000Z");
     }),
   );