diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 4a2e99f3ebc0..ff6b344212cc 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/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) => { diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index fb2fadde5e63..9ca24a51e254 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/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({ diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 6183dbc66168..b1dada84c8ba 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/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"); }), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index edfee6fbdd9e..0fdad96ad76f 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -463,6 +463,50 @@ function mapThreadPullRequests( }; } +function mapCommandThreadRow( + row: Schema.Schema.Type, + input: { + readonly repositoryIdentity: OrchestrationProject["repositoryIdentity"]; + readonly proposedPlans: ReadonlyArray; + readonly pullRequests: ReadonlyArray; + readonly latestTurn: OrchestrationLatestTurn | null; + readonly session: OrchestrationSession | null; + }, +): OrchestrationThread { + return { + id: row.threadId, + projectId: row.projectId, + title: row.title, + modelSelection: row.modelSelection, + runtimeMode: row.runtimeMode, + interactionMode: row.interactionMode, + branch: row.branch, + worktreePath: row.worktreePath, + ...mapThreadPullRequests(input.pullRequests, row.projectId, input.repositoryIdentity), + branchPullRequest: row.branchPullRequest, + latestTurn: input.latestTurn, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + archivedAt: row.archivedAt, + settledOverride: row.settledOverride, + settledAt: row.settledAt, + unsettledAt: row.unsettledAt, + snoozedUntil: row.snoozedUntil, + snoozedAt: row.snoozedAt, + pinnedAt: row.pinnedAt, + pinOrderKey: row.pinOrderKey ?? null, + activeOrderKey: row.activeOrderKey ?? null, + titleRegeneration: mapTitleRegeneration(row), + titleState: row.titleState, + deletedAt: row.deletedAt, + messages: [], + proposedPlans: input.proposedPlans, + activities: [], + checkpoints: [], + session: input.session, + }; +} + function mapThreadActivityRow( row: Schema.Schema.Type, ): OrchestrationThreadActivity { @@ -1198,6 +1242,48 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const getThreadRowById = SqlSchema.findOneOption({ + Request: ThreadIdLookupInput, + Result: ProjectionThreadDbRowSchema, + execute: ({ threadId }) => + sql` + SELECT + thread_id AS "threadId", + project_id AS "projectId", + title, + title_state_json AS "titleState", + model_selection_json AS "modelSelection", + runtime_mode AS "runtimeMode", + interaction_mode AS "interactionMode", + branch, + worktree_path AS "worktreePath", + linked_pull_request_json AS "linkedPullRequest", + branch_pull_request_json AS "branchPullRequest", + latest_turn_id AS "latestTurnId", + created_at AS "createdAt", + updated_at AS "updatedAt", + archived_at AS "archivedAt", + settled_override AS "settledOverride", + settled_at AS "settledAt", + unsettled_at AS "unsettledAt", + snoozed_until AS "snoozedUntil", + snoozed_at AS "snoozedAt", + pinned_at AS "pinnedAt", + pin_order_key AS "pinOrderKey", + active_order_key AS "activeOrderKey", + title_regeneration_request_id AS "titleRegenerationRequestId", + title_regeneration_started_at AS "titleRegenerationStartedAt", + latest_user_message_at AS "latestUserMessageAt", + pending_approval_count AS "pendingApprovalCount", + pending_user_input_count AS "pendingUserInputCount", + has_actionable_proposed_plan AS "hasActionableProposedPlan", + deleted_at AS "deletedAt" + FROM projection_threads + WHERE thread_id = ${threadId} + LIMIT 1 + `, + }); + const getActiveThreadRowById = SqlSchema.findOneOption({ Request: ThreadIdLookupInput, Result: ProjectionThreadDbRowSchema, @@ -1610,6 +1696,30 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const getCommandLatestTurnRowByThread = SqlSchema.findOneOption({ + Request: ThreadIdLookupInput, + Result: ProjectionLatestTurnDbRowSchema, + execute: ({ threadId }) => + sql` + SELECT + turns.thread_id AS "threadId", + turns.turn_id AS "turnId", + turns.state, + turns.requested_at AS "requestedAt", + turns.started_at AS "startedAt", + turns.completed_at AS "completedAt", + turns.assistant_message_id AS "assistantMessageId", + turns.source_proposed_plan_thread_id AS "sourceProposedPlanThreadId", + turns.source_proposed_plan_id AS "sourceProposedPlanId" + FROM projection_threads threads + JOIN projection_turns turns + ON turns.thread_id = threads.thread_id + AND turns.turn_id = threads.latest_turn_id + WHERE threads.thread_id = ${threadId} + LIMIT 1 + `, + }); + const listCheckpointRowsByThread = SqlSchema.findAll({ Request: ThreadIdLookupInput, Result: ProjectionCheckpointDbRowSchema, @@ -2334,7 +2444,7 @@ pending_approval_requests AS ( }), ); - const getCommandReadModel: ProjectionSnapshotQueryShape["getCommandReadModel"] = () => + const getFullCommandReadModel = () => sql .withTransaction( Effect.all([ @@ -2520,42 +2630,15 @@ pending_approval_requests AS ( if (!row) { continue; } - threads.push({ - id: row.threadId, - projectId: row.projectId, - title: row.title, - modelSelection: row.modelSelection, - runtimeMode: row.runtimeMode, - interactionMode: row.interactionMode, - branch: row.branch, - worktreePath: row.worktreePath, - ...mapThreadPullRequests( - pullRequestsByThread.get(row.threadId) ?? [], - row.projectId, - repositoryIdentities.get(row.projectId), - ), - branchPullRequest: row.branchPullRequest, - latestTurn: latestTurnByThread.get(row.threadId) ?? null, - createdAt: row.createdAt, - updatedAt: row.updatedAt, - archivedAt: row.archivedAt, - settledOverride: row.settledOverride, - settledAt: row.settledAt, - unsettledAt: row.unsettledAt, - snoozedUntil: row.snoozedUntil, - snoozedAt: row.snoozedAt, - pinnedAt: row.pinnedAt, - pinOrderKey: row.pinOrderKey ?? null, - activeOrderKey: row.activeOrderKey ?? null, - titleRegeneration: mapTitleRegeneration(row), - titleState: row.titleState, - deletedAt: row.deletedAt, - messages: [], - proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], - activities: [], - checkpoints: [], - session: sessionByThread.get(row.threadId) ?? null, - }); + threads.push( + mapCommandThreadRow(row, { + repositoryIdentity: repositoryIdentities.get(row.projectId) ?? null, + proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], + pullRequests: pullRequestsByThread.get(row.threadId) ?? [], + latestTurn: latestTurnByThread.get(row.threadId) ?? null, + session: sessionByThread.get(row.threadId) ?? null, + }), + ); } return { @@ -3041,6 +3124,103 @@ pending_approval_requests AS ( ), ); + const getCommandThreadById = Effect.fn("ProjectionSnapshotQuery.getCommandThreadById")(function* ( + threadId: ThreadId, + ) { + const [threadRow, proposedPlanRows, pullRequestRows, latestTurnRow, sessionRow] = yield* sql + .withTransaction( + Effect.all([ + getThreadRowById({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getCommandThreadById:getThread:query", + "ProjectionSnapshotQuery.getCommandThreadById:getThread:decodeRow", + ), + ), + ), + listThreadProposedPlanRowsByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getCommandThreadById:listPlans:query", + "ProjectionSnapshotQuery.getCommandThreadById:listPlans:decodeRows", + ), + ), + ), + listThreadPullRequestRowsByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getCommandThreadById:listPullRequests:query", + "ProjectionSnapshotQuery.getCommandThreadById:listPullRequests:decodeRows", + ), + ), + ), + getCommandLatestTurnRowByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getCommandThreadById:getLatestTurn:query", + "ProjectionSnapshotQuery.getCommandThreadById:getLatestTurn:decodeRow", + ), + ), + ), + getThreadSessionRowByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getCommandThreadById:getSession:query", + "ProjectionSnapshotQuery.getCommandThreadById:getSession:decodeRow", + ), + ), + ), + ]), + ) + .pipe( + Effect.mapError((error) => + isPersistenceError(error) + ? error + : toPersistenceSqlError("ProjectionSnapshotQuery.getCommandThreadById:transaction")( + error, + ), + ), + ); + + if (Option.isNone(threadRow)) { + return Option.none(); + } + + const pullRequests = pullRequestRows.map(mapPullRequestRow); + const repositoryIdentity = + pullRequests.length === 0 + ? null + : (Option.getOrNull(yield* getProjectShellById(threadRow.value.projectId)) + ?.repositoryIdentity ?? null); + + return Option.some( + mapCommandThreadRow(threadRow.value, { + repositoryIdentity, + proposedPlans: proposedPlanRows.map(mapProposedPlanRow), + pullRequests, + latestTurn: Option.getOrNull(Option.map(latestTurnRow, mapLatestTurn)), + session: Option.getOrNull(Option.map(sessionRow, mapSessionRow)), + }), + ); + }); + + const getCommandReadModel: ProjectionSnapshotQueryShape["getCommandReadModel"] = (query) => { + if (query?.threadId === undefined) { + return getFullCommandReadModel(); + } + return getCommandThreadById(query.threadId).pipe( + Effect.map((thread) => ({ + snapshotSequence: 0, + projects: [], + threads: Option.toArray(thread), + updatedAt: Option.match(thread, { + onNone: () => "1970-01-01T00:00:00.000Z", + onSome: (value) => value.updatedAt, + }), + })), + ); + }; + const getFirstActiveThreadIdByProjectId: ProjectionSnapshotQueryShape["getFirstActiveThreadIdByProjectId"] = (projectId) => getFirstActiveThreadIdByProject({ projectId }).pipe( diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 46f269a0cf40..11df2dbf4f8e 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -95,11 +95,11 @@ export interface ProjectionSnapshotQueryShape { /** * Read the lightweight command snapshot used to bootstrap the in-memory * orchestration engine without hydrating message/activity/checkpoint bodies. + * Pass a thread id to read only that thread, including archived rows. */ - readonly getCommandReadModel: () => Effect.Effect< - OrchestrationReadModel, - ProjectionRepositoryError - >; + readonly getCommandReadModel: (query?: { + readonly threadId?: ThreadId; + }) => Effect.Effect; /** * Read the latest orchestration projection snapshot.