From 859a559e8a3f9f7c149369795c96a726f37396cf Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 14:53:43 -0700 Subject: [PATCH 01/10] feat(mcp): defer thread organization safely --- apps/server/src/mcp/OrchestratorMcpService.ts | 77 +++++ ...OrchestratorMcpToolkit.integration.test.ts | 32 +++ .../src/mcp/toolkits/orchestrator/handlers.ts | 6 + .../mcp/toolkits/orchestrator/tools.test.ts | 7 + .../src/mcp/toolkits/orchestrator/tools.ts | 15 + .../src/orchestration-v2/Orchestrator.ts | 221 +++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 2 + .../src/orchestration-v2/runtimeLayer.test.ts | 268 ++++++++++++++++++ .../testkit/OrchestratorScenario.ts | 3 + .../orchestrator-mcp-server.md | 15 + docs/user/thread-sidebar.md | 5 + packages/contracts/src/orchestrationV2.ts | 43 +++ .../contracts/src/orchestratorMcp.test.ts | 26 ++ packages/contracts/src/orchestratorMcp.ts | 35 +++ .../shared/src/t3McpToolPresentation.test.ts | 4 + packages/shared/src/t3McpToolPresentation.ts | 1 + 16 files changed, 760 insertions(+) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 0f0c90bca934..5f94370c95cb 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -30,6 +30,8 @@ import { type OrchestratorMcpTaskCancelResult, type OrchestratorMcpUpdateScheduledTaskInput, type OrchestratorMcpThreadDetail, + type OrchestratorMcpThreadDeferOrganizationInput, + type OrchestratorMcpThreadDeferOrganizationResult, type OrchestratorMcpThreadInterruptInput, type OrchestratorMcpThreadInterruptResult, type OrchestratorMcpThreadDeleteInput, @@ -158,6 +160,10 @@ export interface OrchestratorMcpServiceShape { scope: McpInvocationScope, input: OrchestratorMcpThreadOrganizeInput, ) => Effect.Effect; + readonly deferThreadOrganization: ( + scope: McpInvocationScope, + input: OrchestratorMcpThreadDeferOrganizationInput, + ) => Effect.Effect; readonly deleteThread: ( scope: McpInvocationScope, input: OrchestratorMcpThreadDeleteInput, @@ -680,6 +686,23 @@ function organizationState( }; } +function deferredOrganizationResult( + projection: OrchestrationV2ThreadProjection, +): OrchestratorMcpThreadDeferOrganizationResult { + const intent = projection.thread.deferredOrganization; + return { + threadId: projection.thread.id, + intent: + intent == null + ? null + : { + runId: intent.runId, + action: intent.action, + requestedAt: DateTime.formatIso(intent.requestedAt), + }, + }; +} + function organizationCommand(input: { readonly action: OrchestratorMcpThreadOrganizeAction; readonly commandId: CommandId; @@ -1833,6 +1856,60 @@ const make = Effect.gen(function* () { ); return { outcomes } satisfies OrchestratorMcpThreadOrganizeResult; }), + deferThreadOrganization: (scope, input) => + Effect.gen(function* () { + yield* requireCapability(scope); + const parent = yield* loadProjection(scope.threadId); + if (input.operation === "read") return deferredOrganizationResult(parent); + + const key = yield* requestKey(input.clientRequestId); + const command: OrchestrationV2Command = + input.operation === "cancel" + ? { + type: "thread.organization.defer.cancel", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: "thread-defer-organization-cancel", + }), + threadId: scope.threadId, + } + : yield* Effect.gen(function* () { + const parentRun = latestActiveRun(parent); + if ( + parentRun === undefined || + parentRun.rootNodeId === null || + parentRun.providerInstanceId !== scope.providerInstanceId + ) { + return yield* failure( + "parent_not_active", + "Deferred organization requires an active run owned by this MCP provider session.", + ); + } + return { + type: "thread.organization.defer", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: `thread-defer-organization-${input.action}`, + }), + threadId: scope.threadId, + runId: parentRun.id, + action: input.action, + } satisfies OrchestrationV2Command; + }); + yield* threadManagement + .dispatch(command) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to ${input.operation} deferred organization for thread ${scope.threadId}: ${errorMessage(error)}`, + ), + ), + ); + return deferredOrganizationResult(yield* loadProjection(scope.threadId)); + }), deleteThread: (scope, input) => Effect.gen(function* () { const threadId = input.threadId ?? scope.threadId; diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 4c294158c901..bc3421bd5b07 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -16,6 +16,7 @@ import { OrchestratorMcpTaskCancelResult, OrchestratorMcpThreadInterruptResult, OrchestratorMcpThreadDeleteResult, + OrchestratorMcpThreadDeferOrganizationResult, OrchestratorMcpThreadListResult, OrchestratorMcpThreadReadResult, OrchestratorMcpThreadOrganizeResult, @@ -100,6 +101,9 @@ const decodeThreadInterruptResult = Schema.decodeUnknownEffect( OrchestratorMcpThreadInterruptResult, ); const decodeThreadDeleteResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadDeleteResult); +const decodeThreadDeferOrganizationResult = Schema.decodeUnknownEffect( + OrchestratorMcpThreadDeferOrganizationResult, +); const decodeThreadListResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadListResult); const decodeThreadReadResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadReadResult); const decodeThreadOrganizeResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadOrganizeResult); @@ -1230,6 +1234,10 @@ describe("orchestrator MCP toolkit", () => { ({ tool }) => tool.name === "t3_thread_organize", ); expect(threadOrganizeTool?.tool.annotations?.destructiveHint).toBe(true); + const threadDeferOrganizationTool = server.tools.find( + ({ tool }) => tool.name === "t3_thread_defer_organization", + ); + expect(threadDeferOrganizationTool?.tool.annotations?.destructiveHint).toBe(true); const threadDeleteTool = server.tools.find( ({ tool }) => tool.name === "t3_thread_delete", ); @@ -1279,6 +1287,30 @@ describe("orchestrator MCP toolkit", () => { ]), }); + const scheduledOrganization = yield* decodeThreadDeferOrganizationResult( + (yield* invoke("t3_thread_defer_organization", { + operation: "schedule", + action: "settle", + clientRequestId: "defer-parent-settlement", + })).structuredContent, + ).pipe(Effect.orDie); + expect(scheduledOrganization).toMatchObject({ + threadId: parentThreadId, + intent: { runId: parentRun.id, action: "settle", requestedAt: expect.any(String) }, + }); + const readOrganization = yield* decodeThreadDeferOrganizationResult( + (yield* invoke("t3_thread_defer_organization", { operation: "read" })) + .structuredContent, + ).pipe(Effect.orDie); + expect(readOrganization).toEqual(scheduledOrganization); + const cancelledOrganization = yield* decodeThreadDeferOrganizationResult( + (yield* invoke("t3_thread_defer_organization", { + operation: "cancel", + clientRequestId: "cancel-parent-settlement", + })).structuredContent, + ).pipe(Effect.orDie); + expect(cancelledOrganization).toEqual({ threadId: parentThreadId, intent: null }); + const scheduleTool = server.tools.find(({ tool }) => tool.name === "schedule_task"); expect(scheduleTool?.tool.annotations?.destructiveHint).toBe(true); const scheduleCall = yield* invoke("schedule_task", { diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts index 00209c367ca4..85cf5d3d46f9 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -104,6 +104,12 @@ const handlers = { const service = yield* OrchestratorMcpService; return yield* service.organizeThreads(scope, input); }), + t3_thread_defer_organization: (input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext; + const service = yield* OrchestratorMcpService; + return yield* service.deferThreadOrganization(scope, input); + }), t3_thread_delete: (input) => Effect.gen(function* () { const scope = yield* McpInvocationContext; diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts index de85b6f72e0c..bfec178b2457 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts @@ -5,6 +5,7 @@ import { CreateThreadsTool, DelegateTaskTool, ScheduleTaskTool, + ThreadDeferOrganizationTool, ThreadUpdateTool, } from "./tools.ts"; @@ -61,6 +62,12 @@ describe("orchestrator MCP tool guidance", () => { assert.include(ScheduleTaskTool.description ?? "", "nextRunAt"); }); + it("describes safe, calling-run-bound deferred organization", () => { + assert.include(ThreadDeferOrganizationTool.description ?? "", "THIS calling thread"); + assert.include(ThreadDeferOrganizationTool.description ?? "", "current run"); + assert.include(ThreadDeferOrganizationTool.description ?? "", "approval-blocked"); + }); + it("publishes thread metadata actions from an object-root schema", () => { const schema = Tool.getJsonSchema(ThreadUpdateTool) as { readonly type?: unknown; diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index ab4a961935d5..34cb8d3ec7b8 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -19,6 +19,8 @@ import { OrchestratorMcpThreadInterruptResult, OrchestratorMcpThreadDeleteInput, OrchestratorMcpThreadDeleteResult, + OrchestratorMcpThreadDeferOrganizationInput, + OrchestratorMcpThreadDeferOrganizationResult, OrchestratorMcpThreadListInput, OrchestratorMcpThreadListResult, OrchestratorMcpThreadReadInput, @@ -226,6 +228,18 @@ export const ThreadOrganizeTool = Tool.make("t3_thread_organize", { .annotate(Tool.Title, "Organize T3 threads") .annotate(Tool.Destructive, true); +export const ThreadDeferOrganizationTool = Tool.make("t3_thread_defer_organization", { + description: + "Schedule settlement or archival of THIS calling thread after the current run completes safely. The intent is durable and applies only when that run completes successfully with no newer, queued, active, approval-blocked, or title-regeneration work. Otherwise it is discarded. Use operation='read' to inspect the current intent or operation='cancel' to remove it. clientRequestId makes schedule and cancel retries idempotent.", + parameters: OrchestratorMcpThreadDeferOrganizationInput, + success: OrchestratorMcpThreadDeferOrganizationResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Defer thread organization") + .annotate(Tool.Destructive, true); + export const ThreadDeleteTool = Tool.make("t3_thread_delete", { description: "Permanently delete one T3 thread in the calling project, defaulting to this thread. This cancels its active work and removes it from thread listings. clientRequestId makes retries idempotent.", @@ -292,6 +306,7 @@ export const OrchestratorToolkit = Toolkit.make( ThreadReadTool, ThreadUpdateTool, ThreadOrganizeTool, + ThreadDeferOrganizationTool, ThreadDeleteTool, ThreadSendTool, ThreadWaitTool, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 92d9213c3ec5..afe00b017b48 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -257,6 +257,9 @@ function commandThreadId(command: OrchestrationV2Command): ThreadId { case "thread.pin.reorder": case "thread.visit": case "thread.mark-unread": + case "thread.organization.defer": + case "thread.organization.defer.cancel": + case "thread.organization.defer.apply": case "thread.metadata.update": case "thread.title.regeneration.complete": case "thread.runtime-mode.set": @@ -1969,6 +1972,168 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); + const dispatchDeferredOrganization = Effect.fn("orchestrationV2.dispatch.deferredOrganization")( + function* ( + command: Extract< + OrchestrationV2Command, + { + readonly type: + | "thread.organization.defer" + | "thread.organization.defer.cancel" + | "thread.organization.defer.apply"; + } + >, + events: Ref.Ref>, + ) { + const projection = yield* projectionStore.getThreadProjection(command.threadId).pipe( + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: command.threadId, + cause, + }), + ), + ); + const thread = projection.thread; + if (thread.deletedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} is deleted.`, + }); + } + const now = yield* DateTime.now; + if (command.type === "thread.organization.defer") { + if (thread.archivedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} is archived.`, + }); + } + const activeRuns = projection.runs + .filter((run) => ["preparing", "starting", "running", "waiting"].includes(run.status)) + .toSorted((left, right) => right.ordinal - left.ordinal); + if (activeRuns[0]?.id !== command.runId) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Run ${command.runId} is not the active run for thread ${command.threadId}.`, + }); + } + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { + ...thread, + deferredOrganization: { + runId: command.runId, + action: command.action, + requestedAt: now, + }, + updatedAt: now, + }, + }); + return; + } + + if (command.type === "thread.organization.defer.cancel") { + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { + ...thread, + deferredOrganization: null, + updatedAt: thread.deferredOrganization == null ? thread.updatedAt : now, + }, + }); + return; + } + + const intent = thread.deferredOrganization; + if (intent == null || intent.runId !== command.runId) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} has no deferred organization intent for run ${command.runId}.`, + }); + } + const boundRun = projection.runs.find((run) => run.id === command.runId); + const hasNewerRun = + boundRun !== undefined && projection.runs.some((run) => run.ordinal > boundRun.ordinal); + const hasPendingWork = + projection.runs.some((run) => + ["preparing", "queued", "starting", "running", "waiting"].includes(run.status), + ) || projection.runtimeRequests.some((request) => request.status === "pending"); + const canApply = + boundRun?.status === "completed" && + !hasNewerRun && + !hasPendingWork && + thread.titleRegeneration == null && + thread.archivedAt === null; + if (!canApply) { + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { ...thread, deferredOrganization: null, updatedAt: now }, + }); + return; + } + if (intent.action === "settle") { + yield* emit( + events, + command, + )({ + type: "thread.settled", + threadId: command.threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { + ...thread, + deferredOrganization: null, + settledOverride: "settled", + settledAt: now, + pinnedAt: null, + pinOrderKey: null, + updatedAt: now, + }, + }); + return; + } + yield* emit( + events, + command, + )({ + type: "thread.archived", + threadId: command.threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { + ...thread, + deferredOrganization: null, + archivedAt: now, + titleRegeneration: null, + updatedAt: now, + }, + }); + }, + ); + const dispatchProviderSessionDetach = Effect.fn("orchestrationV2.dispatch.providerSessionDetach")( function* ( command: Extract, @@ -6991,6 +7156,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "provider.switch": yield* dispatchThreadMutation(command, events, effects); break; + case "thread.organization.defer": + case "thread.organization.defer.cancel": + case "thread.organization.defer.apply": + yield* dispatchDeferredOrganization(command, events); + break; case "provider-session.detach": yield* dispatchProviderSessionDetach(command, events, effects); break; @@ -7197,6 +7367,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const dispatchWithReceipt = (command: OrchestrationV2Command) => threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)); + const applyDeferredOrganization = (threadId: ThreadId, runId: RunId) => + Effect.gen(function* () { + const projection = yield* projectionStore.getThreadProjection(threadId); + if (projection.thread.deferredOrganization?.runId !== runId) return; + yield* dispatchWithReceiptEffect({ + type: "thread.organization.defer.apply", + commandId: CommandId.make(`command:system:thread-organization-defer:${threadId}:${runId}`), + threadId, + runId, + }); + }); + const handleTerminalRun = (stored: OrchestrationV2StoredEvent) => Effect.gen(function* () { const threadId = stored.event.threadId; @@ -7215,6 +7397,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio threadId, finalizeDelegatedCompletionDelivery(threadId, stored.event.payload.id), ); + yield* threadDispatch.withLock( + threadId, + applyDeferredOrganization(threadId, stored.event.payload.id), + ); } yield* threadDispatch.withLock(threadId, startNextQueuedRun(threadId)); }).pipe( @@ -7249,6 +7435,41 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // The high-water subscription deliberately skips history, so recover the // two terminal side effects from current projections instead: one queued // run per idle thread, plus any app-owned child result not yet transferred. + yield* projectionStore.getShellSnapshot().pipe( + Effect.flatMap((shell) => + Effect.forEach( + [...shell.threads, ...shell.archivedThreads].filter( + (thread) => thread.deferredOrganization != null, + ), + (thread) => + threadDispatch.withLock( + thread.id, + Effect.gen(function* () { + const projection = yield* projectionStore.getThreadProjection(thread.id); + const intent = projection.thread.deferredOrganization; + if (intent == null) return; + const boundRun = projection.runs.find((run) => run.id === intent.runId); + const hasNewerRun = + boundRun !== undefined && + projection.runs.some((run) => run.ordinal > boundRun.ordinal); + if ( + boundRun === undefined || + hasNewerRun || + ["completed", "failed", "cancelled", "interrupted", "rolled_back"].includes( + boundRun.status, + ) + ) { + yield* applyDeferredOrganization(thread.id, intent.runId); + } + }), + ), + { concurrency: 8, discard: true }, + ), + ), + Effect.catchCause((cause) => + Effect.logWarning("Failed to recover deferred thread organization", { cause }), + ), + ); yield* resumeQueuedRuns.pipe( Effect.tap((resumed) => resumed === 0 diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 5d0ae69bfa5a..fc873818ec2c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1123,6 +1123,7 @@ export function threadShellFromProjection( pinnedAt: projection.thread.pinnedAt ?? null, pinOrderKey: projection.thread.pinOrderKey ?? null, lastVisitedAt: projection.thread.lastVisitedAt, + deferredOrganization: projection.thread.deferredOrganization ?? null, titleRegeneration: projection.thread.titleRegeneration ?? null, deletedAt: projection.thread.deletedAt, }; @@ -1303,6 +1304,7 @@ function shellFromState(input: { pinnedAt: input.state.thread.pinnedAt ?? null, pinOrderKey: input.state.thread.pinOrderKey ?? null, lastVisitedAt: input.state.thread.lastVisitedAt, + deferredOrganization: input.state.thread.deferredOrganization ?? null, titleRegeneration: input.state.thread.titleRegeneration ?? null, deletedAt: input.state.thread.deletedAt, }; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 53fb901bac01..4c02a2a9bf4b 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -10,6 +10,7 @@ import { NodeId, RuntimeRequestId, TurnItemId, + type OrchestrationV2Run, type ModelSelection, ProjectId, ProviderDriverKind, @@ -1432,6 +1433,273 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("applies deferred settlement after safe completion and replays its receipt", () => + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const eventSink = yield* EventSinkV2; + const maintenance = yield* ProjectionMaintenanceV2; + const threadId = ThreadId.make("runtime-layer-deferred-settle-thread"); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-deferred-settle-create"), + threadId, + projectId: ProjectId.make("runtime-layer-deferred-settle-project"), + title: "Deferred settlement", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/tmp/runtime-layer-deferred-settle", + }); + yield* orchestrator.dispatch({ + type: "thread.pin", + commandId: CommandId.make("runtime-layer-deferred-settle-pin"), + threadId, + orderKey: "a0", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-deferred-settle-message"), + threadId, + messageId: MessageId.make("runtime-layer-deferred-settle-message"), + text: "Settle this thread after completion.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const before = yield* orchestrator.getThreadProjection(threadId); + const run = before.runs[0]; + assert.isDefined(run); + yield* orchestrator.dispatch({ + type: "thread.organization.defer", + commandId: CommandId.make("runtime-layer-deferred-settle-schedule"), + threadId, + runId: run.id, + action: "settle", + }); + const scheduledShell = yield* orchestrator.getThreadShell(threadId); + assert.isNotNull(scheduledShell); + assert.deepEqual(scheduledShell.deferredOrganization, { + runId: run.id, + action: "settle", + requestedAt: (yield* orchestrator.getThreadProjection(threadId)).thread.deferredOrganization + ?.requestedAt, + }); + + const settledEvents = yield* Queue.unbounded(); + const afterSequence = yield* orchestrator.getThreadEventSequence(threadId); + yield* eventSink.stream({ threadId, afterSequence }).pipe( + Stream.runForEach((stored) => + stored.event.type === "thread.settled" + ? Queue.offer(settledEvents, stored.sequence) + : Effect.void, + ), + Effect.forkScoped, + ); + yield* Effect.yieldNow; + const completedAt = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("runtime-layer-deferred-settle-completed"), + type: "run.updated", + threadId, + runId: run.id, + ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), + providerInstanceId: run.providerInstanceId, + occurredAt: completedAt, + payload: { ...run, status: "completed", completedAt }, + }, + ], + }); + const settledSequence = yield* Queue.take(settledEvents); + + const settled = yield* orchestrator.getThreadProjection(threadId); + assert.equal(settled.thread.settledOverride, "settled"); + assert.isNull(settled.thread.pinnedAt); + assert.isNull(settled.thread.deferredOrganization); + const applyCommand = { + type: "thread.organization.defer.apply" as const, + commandId: CommandId.make(`command:system:thread-organization-defer:${threadId}:${run.id}`), + threadId, + runId: run.id, + }; + const retry = yield* orchestrator.dispatch(applyCommand); + assert.equal(retry.sequence, settledSequence); + assert.equal(retry.storedEvents[0]?.event.type, "thread.settled"); + + const rebuilt = yield* maintenance.rebuild; + assert.isTrue(rebuilt.valid); + const replayed = yield* orchestrator.getThreadProjection(threadId); + assert.equal(replayed.thread.settledOverride, "settled"); + assert.isNull(replayed.thread.deferredOrganization); + }), + ); + + it.effect("archives after safe completion and discards stale or blocked deferred intents", () => + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const eventSink = yield* EventSinkV2; + + const createStartingThread = (suffix: string) => + Effect.gen(function* () { + const threadId = ThreadId.make(`runtime-layer-deferred-${suffix}-thread`); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`runtime-layer-deferred-${suffix}-create`), + threadId, + projectId: ProjectId.make(`runtime-layer-deferred-${suffix}-project`), + title: `Deferred ${suffix}`, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: `/tmp/runtime-layer-deferred-${suffix}`, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`runtime-layer-deferred-${suffix}-message`), + threadId, + messageId: MessageId.make(`runtime-layer-deferred-${suffix}-message`), + text: `Finish deferred ${suffix}.`, + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]; + assert.isDefined(run); + return { threadId, run }; + }); + + const completeAndAwait = (input: { + readonly suffix: string; + readonly threadId: ThreadId; + readonly run: OrchestrationV2Run; + readonly expectedEvent: "thread.archived" | "thread.metadata-updated"; + }) => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const afterSequence = yield* orchestrator.getThreadEventSequence(input.threadId); + yield* eventSink.stream({ threadId: input.threadId, afterSequence }).pipe( + Stream.runForEach((stored) => + stored.event.type === input.expectedEvent + ? Queue.offer(events, undefined) + : Effect.void, + ), + Effect.forkScoped, + ); + yield* Effect.yieldNow; + const completedAt = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make(`runtime-layer-deferred-${input.suffix}-completed`), + type: "run.updated", + threadId: input.threadId, + runId: input.run.id, + ...(input.run.rootNodeId === null ? {} : { nodeId: input.run.rootNodeId }), + providerInstanceId: input.run.providerInstanceId, + occurredAt: completedAt, + payload: { ...input.run, status: "completed", completedAt }, + }, + ], + }); + yield* Queue.take(events); + }); + + const archived = yield* createStartingThread("archive"); + yield* orchestrator.dispatch({ + type: "thread.organization.defer", + commandId: CommandId.make("runtime-layer-deferred-archive-schedule"), + threadId: archived.threadId, + runId: archived.run.id, + action: "archive", + }); + yield* completeAndAwait({ ...archived, suffix: "archive", expectedEvent: "thread.archived" }); + const archivedProjection = yield* orchestrator.getThreadProjection(archived.threadId); + assert.isNotNull(archivedProjection.thread.archivedAt); + assert.isNull(archivedProjection.thread.deferredOrganization); + + const stale = yield* createStartingThread("stale"); + yield* orchestrator.dispatch({ + type: "thread.organization.defer", + commandId: CommandId.make("runtime-layer-deferred-stale-schedule"), + threadId: stale.threadId, + runId: stale.run.id, + action: "settle", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-deferred-stale-queued"), + threadId: stale.threadId, + messageId: MessageId.make("runtime-layer-deferred-stale-queued"), + text: "Newer queued work makes the intent stale.", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + }); + yield* completeAndAwait({ + ...stale, + suffix: "stale", + expectedEvent: "thread.metadata-updated", + }); + const staleProjection = yield* orchestrator.getThreadProjection(stale.threadId); + assert.isNull(staleProjection.thread.deferredOrganization); + assert.isNull(staleProjection.thread.settledOverride); + + const blocked = yield* createStartingThread("blocked"); + yield* orchestrator.dispatch({ + type: "thread.organization.defer", + commandId: CommandId.make("runtime-layer-deferred-blocked-schedule"), + threadId: blocked.threadId, + runId: blocked.run.id, + action: "settle", + }); + const requestTime = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("runtime-layer-deferred-blocked-request"), + type: "runtime-request.updated", + threadId: blocked.threadId, + occurredAt: requestTime, + payload: { + id: RuntimeRequestId.make("runtime-layer-deferred-blocked-request"), + nodeId: NodeId.make("runtime-layer-deferred-blocked-node"), + providerTurnId: null, + nativeRequestRef: null, + kind: "command", + status: "pending", + responseCapability: { type: "not_resumable", reason: "test request" }, + createdAt: requestTime, + resolvedAt: null, + }, + }, + ], + }); + yield* completeAndAwait({ + ...blocked, + suffix: "blocked", + expectedEvent: "thread.metadata-updated", + }); + const blockedProjection = yield* orchestrator.getThreadProjection(blocked.threadId); + assert.isNull(blockedProjection.thread.deferredOrganization); + assert.isNull(blockedProjection.thread.settledOverride); + assert.equal(blockedProjection.runtimeRequests[0]?.status, "pending"); + }), + ); + it.effect("promotes only one queued run after each terminal run", () => Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts index ee479d8eb872..e73272e5103d 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts @@ -127,6 +127,9 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray ({ + ...deferredFields, + requestedAt: Schema.DateTimeUtcFromString, + })), + ), + ), titleRegeneration: Schema.optional( Schema.NullOr( Schema.Struct({ @@ -1851,6 +1868,14 @@ export const OrchestrationV2ThreadShellJson = OrchestrationV2ThreadShell.mapFiel snoozedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtcFromString)), pinnedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtcFromString)), lastVisitedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtcFromString)), + deferredOrganization: Schema.optional( + Schema.NullOr( + OrchestrationV2DeferredOrganization.mapFields((deferredFields) => ({ + ...deferredFields, + requestedAt: Schema.DateTimeUtcFromString, + })), + ), + ), titleRegeneration: Schema.optional( Schema.NullOr( Schema.Struct({ @@ -2131,6 +2156,24 @@ export const OrchestrationV2Command = Schema.Union([ commandId: CommandId, threadId: ThreadId, }), + Schema.Struct({ + type: Schema.Literal("thread.organization.defer"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + action: Schema.Literals(["settle", "archive"]), + }), + Schema.Struct({ + type: Schema.Literal("thread.organization.defer.cancel"), + commandId: CommandId, + threadId: ThreadId, + }), + Schema.Struct({ + type: Schema.Literal("thread.organization.defer.apply"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + }), Schema.Struct({ type: Schema.Literal("thread.metadata.update"), commandId: CommandId, diff --git a/packages/contracts/src/orchestratorMcp.test.ts b/packages/contracts/src/orchestratorMcp.test.ts index dc8a88a9faab..982da87085e0 100644 --- a/packages/contracts/src/orchestratorMcp.test.ts +++ b/packages/contracts/src/orchestratorMcp.test.ts @@ -6,6 +6,7 @@ import { OrchestratorMcpDelegateTaskInput, OrchestratorMcpDelegateTaskResult, OrchestratorMcpThreadInterruptInput, + OrchestratorMcpThreadDeferOrganizationInput, OrchestratorMcpThreadListInput, OrchestratorMcpThreadOrganizeInput, OrchestratorMcpThreadReadInput, @@ -18,6 +19,9 @@ const decodeCreateThreadsInput = Schema.decodeUnknownSync(OrchestratorMcpCreateT const decodeDelegateTaskInput = Schema.decodeUnknownSync(OrchestratorMcpDelegateTaskInput); const decodeDelegateTaskResult = Schema.decodeUnknownSync(OrchestratorMcpDelegateTaskResult); const decodeThreadInterruptInput = Schema.decodeUnknownSync(OrchestratorMcpThreadInterruptInput); +const decodeThreadDeferOrganizationInput = Schema.decodeUnknownSync( + OrchestratorMcpThreadDeferOrganizationInput, +); const decodeThreadListInput = Schema.decodeUnknownSync(OrchestratorMcpThreadListInput); const decodeThreadOrganizeInput = Schema.decodeUnknownSync(OrchestratorMcpThreadOrganizeInput); const decodeThreadReadInput = Schema.decodeUnknownSync(OrchestratorMcpThreadReadInput); @@ -215,4 +219,26 @@ describe("orchestrator MCP contracts", () => { }), ).toThrow(); }); + + it("decodes durable after-run organization operations", () => { + expect(decodeThreadDeferOrganizationInput({ operation: "read" })).toEqual({ + operation: "read", + }); + expect( + decodeThreadDeferOrganizationInput({ + operation: "schedule", + action: "archive", + clientRequestId: "archive-after-run", + }), + ).toMatchObject({ operation: "schedule", action: "archive" }); + expect( + decodeThreadDeferOrganizationInput({ + operation: "cancel", + clientRequestId: "cancel-after-run", + }), + ).toMatchObject({ operation: "cancel" }); + expect(() => + decodeThreadDeferOrganizationInput({ operation: "schedule", action: "delete" }), + ).toThrow(); + }); }); diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 524684af1713..925fade28ef3 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -530,6 +530,41 @@ export const OrchestratorMcpThreadOrganizeResult = Schema.Struct({ }); export type OrchestratorMcpThreadOrganizeResult = typeof OrchestratorMcpThreadOrganizeResult.Type; +export const OrchestratorMcpDeferredOrganizationIntent = Schema.Struct({ + runId: RunId, + action: Schema.Literals(["settle", "archive"]), + requestedAt: IsoDateTime, +}); +export type OrchestratorMcpDeferredOrganizationIntent = + typeof OrchestratorMcpDeferredOrganizationIntent.Type; + +export const OrchestratorMcpThreadDeferOrganizationInput = Schema.Union([ + Schema.Struct({ + operation: Schema.Literal("read"), + }), + Schema.Struct({ + operation: Schema.Literal("schedule"), + action: Schema.Literals(["settle", "archive"]).annotate({ + description: + "Organization action to apply only after this calling run completes safely with no newer, queued, or blocked work.", + }), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), + }), + Schema.Struct({ + operation: Schema.Literal("cancel"), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), + }), +]); +export type OrchestratorMcpThreadDeferOrganizationInput = + typeof OrchestratorMcpThreadDeferOrganizationInput.Type; + +export const OrchestratorMcpThreadDeferOrganizationResult = Schema.Struct({ + threadId: ThreadId, + intent: Schema.NullOr(OrchestratorMcpDeferredOrganizationIntent), +}); +export type OrchestratorMcpThreadDeferOrganizationResult = + typeof OrchestratorMcpThreadDeferOrganizationResult.Type; + export const OrchestratorMcpThreadDeleteInput = Schema.Struct({ threadId: Schema.optional(ThreadId), clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), diff --git a/packages/shared/src/t3McpToolPresentation.test.ts b/packages/shared/src/t3McpToolPresentation.test.ts index fb18c22a9280..8dbde300aae2 100644 --- a/packages/shared/src/t3McpToolPresentation.test.ts +++ b/packages/shared/src/t3McpToolPresentation.test.ts @@ -12,6 +12,10 @@ describe("resolveT3McpToolPresentation", () => { displayName: "Organize T3 threads", logo: "t3-code", }); + expect(resolveT3McpToolPresentation("mcp__t3-code__t3_thread_defer_organization")).toEqual({ + displayName: "Defer thread organization", + logo: "t3-code", + }); }); it("pretty prints Codex T3 MCP tool names", () => { diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index 4a8b80502bd0..0b779829299f 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -50,6 +50,7 @@ const T3_MCP_TOOLS: Record< t3_thread_read: { displayName: "Read a T3 thread", summaryAction: "thread-read" }, t3_thread_update: { displayName: "Update T3 thread metadata" }, t3_thread_organize: { displayName: "Organize T3 threads" }, + t3_thread_defer_organization: { displayName: "Defer thread organization" }, t3_thread_delete: { displayName: "Delete a T3 thread" }, t3_thread_send: { displayName: "Send to a T3 thread", summaryAction: "thread-send" }, t3_thread_wait: { displayName: "Wait for a T3 thread", summaryAction: "thread-wait" }, From 31a279b8ad7bcac3feb7e0c5e07414084cc946d0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 14:55:13 -0700 Subject: [PATCH 02/10] test(mcp): cover deferred organization recovery --- ...edOrganizationRecovery.integration.test.ts | 129 ++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts diff --git a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts new file mode 100644 index 000000000000..96b5e5aa53f3 --- /dev/null +++ b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts @@ -0,0 +1,129 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + CommandId, + MessageId, + type ModelSelection, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; + +import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import { OrchestratorV2 } from "./Orchestrator.ts"; +import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; +import { makeLayer as makeProviderAdapterRegistryLayer } from "./ProviderAdapterRegistry.ts"; +import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; + +const modelSelection = { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", +} satisfies ModelSelection; + +const adapter = { + instanceId: modelSelection.instanceId, + driver: ProviderDriverKind.make("codex"), + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: () => Effect.die("provider sessions are not used in recovery coverage"), +} as ProviderAdapterV2Shape; + +it.effect("discards a stale deferred organization intent after runtime restart", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const tempDir = yield* Effect.acquireRelease( + fs.makeTempDirectory({ prefix: "t3-deferred-organization-recovery-" }), + (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie), + ); + const databaseLayer = makeSqlitePersistenceLive(path.join(tempDir, "state.sqlite")).pipe( + Layer.provide(NodeServices.layer), + ); + const registryLayer = makeProviderAdapterRegistryLayer([adapter]); + const runtimeLayer = (name: string) => + makeOrchestratorV2ReplayLayerWithRegistry( + { name, runtimePolicyOverride: { cwd: tempDir } }, + registryLayer, + { databaseLayer, runEffectWorker: false }, + ); + const threadId = ThreadId.make("thread:deferred-organization-recovery"); + + const runIds = yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-recovery:create"), + threadId, + projectId: ProjectId.make("project:deferred-organization-recovery"), + title: "Deferred organization recovery", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: tempDir, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-recovery:active"), + threadId, + messageId: MessageId.make("message:deferred-recovery:active"), + text: "Keep this run active.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const activeRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]; + assert.isDefined(activeRun); + yield* orchestrator.dispatch({ + type: "thread.organization.defer", + commandId: CommandId.make("command:deferred-recovery:schedule"), + threadId, + runId: activeRun.id, + action: "settle", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-recovery:queued"), + threadId, + messageId: MessageId.make("message:deferred-recovery:queued"), + text: "This newer run makes the intent stale.", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + }); + const seeded = yield* orchestrator.getThreadProjection(threadId); + assert.equal(seeded.thread.deferredOrganization?.runId, activeRun.id); + const queuedRun = seeded.runs.find((run) => run.status === "queued"); + assert.isDefined(queuedRun); + return { activeRunId: activeRun.id, queuedRunId: queuedRun.id }; + }).pipe(Effect.provide(runtimeLayer("deferred-organization:first-runtime"))), + ); + + const recovered = yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + return yield* orchestrator.getThreadProjection(threadId); + }).pipe(Effect.provide(runtimeLayer("deferred-organization:second-runtime"))), + ); + + assert.isNull(recovered.thread.deferredOrganization); + assert.isNull(recovered.thread.settledOverride); + assert.equal(recovered.runs.find((run) => run.id === runIds.activeRunId)?.status, "starting"); + assert.equal(recovered.runs.find((run) => run.id === runIds.queuedRunId)?.status, "queued"); + }).pipe(Effect.provide(NodeServices.layer)), + ), +); From 72ca28015ed463461d0ce7636db20f5c0ce8ab46 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 15:03:42 -0700 Subject: [PATCH 03/10] fix(orchestration): clean up deferred terminal state --- .../src/orchestration-v2/Orchestrator.ts | 187 ++++++++++++------ .../src/orchestration-v2/runtimeLayer.test.ts | 107 +++++++++- 2 files changed, 227 insertions(+), 67 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index afe00b017b48..a586fc48656f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1436,6 +1436,63 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); }); + const queueProviderSessionDetaches = Effect.fn( + "orchestrationV2.dispatch.queueProviderSessionDetaches", + )(function* (input: { + readonly command: OrchestrationV2Command; + readonly threadId: ThreadId; + readonly events: Ref.Ref>; + readonly effects: Ref.Ref>; + readonly projection: OrchestrationV2ThreadProjection; + readonly providerSessionIds: ReadonlySet; + readonly occurredAt: DateTime.Utc; + readonly detail: string; + readonly revokeMcpCredential: boolean; + }) { + const liveSessions = input.projection.providerSessions.filter( + (session) => + input.providerSessionIds.has(session.id) && + session.status !== "stopped" && + session.status !== "error", + ); + yield* Effect.forEach( + liveSessions, + (session) => + Effect.gen(function* () { + yield* emit( + input.events, + input.command, + )({ + type: "provider-session.detached", + threadId: input.threadId, + driver: session.driver, + providerInstanceId: session.providerInstanceId, + occurredAt: input.occurredAt, + payload: { + providerSessionId: session.id, + detachedAt: input.occurredAt, + reason: input.detail, + }, + }); + yield* Ref.update(input.effects, (existing) => [ + ...existing, + { + id: `effect:${input.command.commandId}:provider-session.detach:${session.id}`, + commandId: input.command.commandId, + threadId: input.threadId, + request: { + type: "provider-session.detach", + providerSessionId: session.id, + detail: input.detail, + ...(input.revokeMcpCredential ? { revokeMcpCredential: true } : {}), + }, + } satisfies PendingOrchestrationEffectV2, + ]); + }), + { concurrency: 1, discard: true }, + ); + }); + const dispatchThreadMutation = Effect.fn("orchestrationV2.dispatch.threadMutation")(function* ( command: Extract< OrchestrationV2Command, @@ -1881,7 +1938,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // decided serially against the projection, so a turn start that // re-engages the thread cannot race this detach. const detachSessionIds = new Set( - command.type === "thread.archive" || command.type === "thread.settle" + command.type === "thread.archive" || + command.type === "thread.delete" || + command.type === "thread.settle" ? projection.providerSessions.map((session) => session.id) : command.type === "thread.metadata.update" && command.worktreePath !== undefined && @@ -1896,70 +1955,31 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : (providerSwitchPlan?.releaseProviderSessionIds ?? []), ); if (detachSessionIds.size > 0) { - const liveSessions = projection.providerSessions.filter( - (session) => - detachSessionIds.has(session.id) && - session.status !== "stopped" && - session.status !== "error", - ); - yield* Effect.forEach( - liveSessions, - (session) => - Effect.gen(function* () { - yield* emit( - events, - command, - )({ - type: "provider-session.detached", - threadId: command.threadId, - driver: session.driver, - providerInstanceId: session.providerInstanceId, - occurredAt: now, - payload: { - providerSessionId: session.id, - detachedAt: now, - reason: - command.type === "thread.archive" - ? "Thread archived." - : command.type === "thread.settle" - ? "Thread settled." - : command.type === "thread.metadata.update" - ? "Workspace changed." - : command.type === "thread.runtime-mode.set" - ? "Runtime mode changed." - : "Provider or model selection changed.", - }, - }); - const pendingEffect = { - id: `effect:${command.commandId}:provider-session.detach:${session.id}`, - commandId: command.commandId, - threadId: command.threadId, - request: { - type: "provider-session.detach", - providerSessionId: session.id, - detail: - command.type === "thread.archive" - ? "Thread archived." - : command.type === "thread.settle" - ? "Thread settled." - : command.type === "thread.metadata.update" - ? "Workspace changed." - : command.type === "thread.runtime-mode.set" - ? "Runtime mode changed." - : "Provider or model selection changed.", - // Terminal detaches revoke the thread's MCP credentials; other - // detach reasons keep them so a re-attaching provider process - // stays authorized. - ...(command.type === "thread.archive" ? { revokeMcpCredential: true } : {}), - }, - } satisfies PendingOrchestrationEffectV2; - yield* Ref.update(effects, (existing) => [...existing, pendingEffect]); - }), - { concurrency: 1, discard: true }, - ); + yield* queueProviderSessionDetaches({ + command, + threadId: command.threadId, + events, + effects, + projection, + providerSessionIds: detachSessionIds, + occurredAt: now, + detail: + command.type === "thread.archive" + ? "Thread archived." + : command.type === "thread.settle" + ? "Thread settled." + : command.type === "thread.delete" + ? "Thread deleted." + : command.type === "thread.metadata.update" + ? "Workspace changed." + : command.type === "thread.runtime-mode.set" + ? "Runtime mode changed." + : "Provider or model selection changed.", + revokeMcpCredential: command.type === "thread.archive" || command.type === "thread.delete", + }); } - if (command.type === "thread.archive") { + if (command.type === "thread.archive" || command.type === "thread.delete") { yield* Ref.update(effects, (existing) => [ ...existing, { @@ -1984,6 +2004,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } >, events: Ref.Ref>, + effects: Ref.Ref>, ) { const projection = yield* projectionStore.getThreadProjection(command.threadId).pipe( Effect.mapError( @@ -2113,6 +2134,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio updatedAt: now, }, }); + yield* queueProviderSessionDetaches({ + command, + threadId: command.threadId, + events, + effects, + projection, + providerSessionIds: new Set(projection.providerSessions.map((session) => session.id)), + occurredAt: now, + detail: "Thread settled.", + revokeMcpCredential: false, + }); return; } yield* emit( @@ -2131,6 +2163,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio updatedAt: now, }, }); + yield* disposeAllDelegatedCompletionCohorts({ + command, + events, + projection, + now, + cancelQueuedDelivery: false, + }); + yield* queueProviderSessionDetaches({ + command, + threadId: command.threadId, + events, + effects, + projection, + providerSessionIds: new Set(projection.providerSessions.map((session) => session.id)), + occurredAt: now, + detail: "Thread archived.", + revokeMcpCredential: true, + }); + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:terminal.cleanup`, + commandId: command.commandId, + threadId: command.threadId, + request: { type: "terminal.cleanup" }, + } satisfies PendingOrchestrationEffectV2, + ]); }, ); @@ -7159,7 +7218,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.organization.defer": case "thread.organization.defer.cancel": case "thread.organization.defer.apply": - yield* dispatchDeferredOrganization(command, events); + yield* dispatchDeferredOrganization(command, events, effects); break; case "provider-session.detach": yield* dispatchProviderSessionDetach(command, events, effects); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 4c02a2a9bf4b..8dc51389ee76 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -15,6 +15,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + ProviderSessionId, ProviderThreadId, RunId, ThreadId, @@ -129,7 +130,7 @@ const TestProviderInstanceRegistry = Layer.succeed(ProviderInstanceRegistry, { const TestLayer = Layer.merge(OrchestrationV2LayerLive, OrchestrationV2EventSinkLayerLive).pipe( Layer.provide(mcpSessionRegistryTestLayer), - Layer.provide(SqlitePersistenceMemory), + Layer.provideMerge(SqlitePersistenceMemory), Layer.provide(CheckpointStoreTestLayer), Layer.provide(ServerConfigLayer), Layer.provide(ServerSettingsService.layerTest()), @@ -1437,6 +1438,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; const eventSink = yield* EventSinkV2; + const sql = yield* SqlClient.SqlClient; const maintenance = yield* ProjectionMaintenanceV2; const threadId = ThreadId.make("runtime-layer-deferred-settle-thread"); @@ -1475,6 +1477,34 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { const before = yield* orchestrator.getThreadProjection(threadId); const run = before.runs[0]; assert.isDefined(run); + const providerSessionId = ProviderSessionId.make( + "runtime-layer-deferred-settle-provider-session", + ); + const providerSessionAttachedAt = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("runtime-layer-deferred-settle-provider-session-attached"), + type: "provider-session.attached", + threadId, + driver, + providerInstanceId: modelSelection.instanceId, + occurredAt: providerSessionAttachedAt, + payload: { + id: providerSessionId, + driver, + providerInstanceId: modelSelection.instanceId, + status: "ready", + cwd: "/tmp/runtime-layer-deferred-settle", + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: providerSessionAttachedAt, + updatedAt: providerSessionAttachedAt, + lastError: null, + }, + }, + ], + }); yield* orchestrator.dispatch({ type: "thread.organization.defer", commandId: CommandId.make("runtime-layer-deferred-settle-schedule"), @@ -1523,6 +1553,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { assert.equal(settled.thread.settledOverride, "settled"); assert.isNull(settled.thread.pinnedAt); assert.isNull(settled.thread.deferredOrganization); + assert.lengthOf(settled.providerSessions, 0); const applyCommand = { type: "thread.organization.defer.apply" as const, commandId: CommandId.make(`command:system:thread-organization-defer:${threadId}:${run.id}`), @@ -1530,8 +1561,27 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { runId: run.id, }; const retry = yield* orchestrator.dispatch(applyCommand); - assert.equal(retry.sequence, settledSequence); - assert.equal(retry.storedEvents[0]?.event.type, "thread.settled"); + assert.equal(retry.sequence, settledSequence + 1); + assert.deepEqual( + retry.storedEvents.map((stored) => stored.event.type), + ["thread.settled", "provider-session.detached"], + ); + const settleEffects = yield* sql<{ readonly payload_json: string }>` + SELECT payload_json + FROM orchestration_v2_effect_outbox + WHERE command_id = ${applyCommand.commandId} + ORDER BY effect_id ASC + `; + assert.deepEqual( + settleEffects.map((effect) => JSON.parse(effect.payload_json)), + [ + { + type: "provider-session.detach", + providerSessionId, + detail: "Thread settled.", + }, + ], + ); const rebuilt = yield* maintenance.rebuild; assert.isTrue(rebuilt.valid); @@ -1545,6 +1595,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; const eventSink = yield* EventSinkV2; + const sql = yield* SqlClient.SqlClient; const createStartingThread = (suffix: string) => Effect.gen(function* () { @@ -1617,6 +1668,34 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { }); const archived = yield* createStartingThread("archive"); + const providerSessionId = ProviderSessionId.make( + "runtime-layer-deferred-archive-provider-session", + ); + const providerSessionAttachedAt = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("runtime-layer-deferred-archive-provider-session-attached"), + type: "provider-session.attached", + threadId: archived.threadId, + driver, + providerInstanceId: modelSelection.instanceId, + occurredAt: providerSessionAttachedAt, + payload: { + id: providerSessionId, + driver, + providerInstanceId: modelSelection.instanceId, + status: "ready", + cwd: "/tmp/runtime-layer-deferred-archive", + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: providerSessionAttachedAt, + updatedAt: providerSessionAttachedAt, + lastError: null, + }, + }, + ], + }); yield* orchestrator.dispatch({ type: "thread.organization.defer", commandId: CommandId.make("runtime-layer-deferred-archive-schedule"), @@ -1628,6 +1707,28 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { const archivedProjection = yield* orchestrator.getThreadProjection(archived.threadId); assert.isNotNull(archivedProjection.thread.archivedAt); assert.isNull(archivedProjection.thread.deferredOrganization); + assert.lengthOf(archivedProjection.providerSessions, 0); + const archiveApplyCommandId = CommandId.make( + `command:system:thread-organization-defer:${archived.threadId}:${archived.run.id}`, + ); + const archiveEffects = yield* sql<{ readonly payload_json: string }>` + SELECT payload_json + FROM orchestration_v2_effect_outbox + WHERE command_id = ${archiveApplyCommandId} + ORDER BY effect_id ASC + `; + assert.deepEqual( + archiveEffects.map((effect) => JSON.parse(effect.payload_json)), + [ + { + type: "provider-session.detach", + providerSessionId, + detail: "Thread archived.", + revokeMcpCredential: true, + }, + { type: "terminal.cleanup" }, + ], + ); const stale = yield* createStartingThread("stale"); yield* orchestrator.dispatch({ From 097b4135760ba8b9bcbbb06f469dc0c3c9e162fd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 15:12:49 -0700 Subject: [PATCH 04/10] fix(orchestration): harden deferred organization recovery --- apps/server/src/mcp/OrchestratorMcpService.ts | 2 +- .../toolkits/worktree/registration.test.ts | 11 +- ...edOrganizationRecovery.integration.test.ts | 165 +++++++-- .../src/orchestration-v2/Orchestrator.ts | 314 ++++++++---------- .../src/orchestration-v2/runtimeLayer.test.ts | 161 ++++++--- .../testkit/ProviderReplayHarness.ts | 32 +- apps/server/src/serverRuntimeStartup.ts | 21 +- .../contracts/src/orchestratorMcp.test.ts | 5 + packages/contracts/src/orchestratorMcp.ts | 29 +- 9 files changed, 459 insertions(+), 281 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 5f94370c95cb..d3435afe774f 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1895,7 +1895,7 @@ const make = Effect.gen(function* () { }), threadId: scope.threadId, runId: parentRun.id, - action: input.action, + action: input.action!, } satisfies OrchestrationV2Command; }); yield* threadManagement diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index 3300a869fe67..477d6cfde083 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -37,7 +37,10 @@ const ToolsListPayload = Schema.fromJsonString( tools: Schema.Array( Schema.Struct({ name: Schema.String, - inputSchema: Schema.Struct({ type: Schema.optional(Schema.String) }), + inputSchema: Schema.Struct({ + type: Schema.optional(Schema.String), + properties: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), + }), annotations: Schema.optional( Schema.Struct({ readOnlyHint: Schema.optional(Schema.Boolean), @@ -114,6 +117,7 @@ it.effect("production mcp layer lists worktree tools over http", () => // than replacing them. expect(toolNames).toContain("preview_status"); expect(toolNames).toContain("delegate_task"); + expect(toolNames).toContain("t3_thread_defer_organization"); // The handoff tool mutates thread state, reaches the network (origin // fetch), and runs project setup scripts, so its MCP hints must not @@ -132,6 +136,11 @@ it.effect("production mcp layer lists worktree tools over http", () => for (const tool of tools) { expect(tool.inputSchema.type, `inputSchema.type of ${tool.name}`).toBe("object"); } + const deferredOrganization = tools.find( + (tool) => tool.name === "t3_thread_defer_organization", + ); + expect(deferredOrganization?.inputSchema.properties).toHaveProperty("operation"); + expect(deferredOrganization?.inputSchema.properties).toHaveProperty("action"); }), ).pipe(Effect.provide(Layer.mergeAll(NodeHttpServer.layerTest, NodeServices.layer))), ); diff --git a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts index 96b5e5aa53f3..1724149aa622 100644 --- a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts @@ -13,10 +13,13 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite.ts"; +import { runV2RecoveryPhase } from "../serverRuntimeStartup.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import { OrchestratorV2 } from "./Orchestrator.ts"; +import { ProviderRuntimeRecoveryService } from "./ProviderRuntimeRecoveryService.ts"; import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; import { makeLayer as makeProviderAdapterRegistryLayer } from "./ProviderAdapterRegistry.ts"; import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; @@ -54,18 +57,136 @@ it.effect("discards a stale deferred organization intent after runtime restart", { databaseLayer, runEffectWorker: false }, ); const threadId = ThreadId.make("thread:deferred-organization-recovery"); + const unreadableThreadId = ThreadId.make("thread:deferred-organization-recovery-unreadable"); const runIds = yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const seed = (targetThreadId: ThreadId, suffix: string) => + Effect.gen(function* () { + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:deferred-recovery:${suffix}:create`), + threadId: targetThreadId, + projectId: ProjectId.make("project:deferred-organization-recovery"), + title: `Deferred organization recovery ${suffix}`, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: tempDir, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:deferred-recovery:${suffix}:active`), + threadId: targetThreadId, + messageId: MessageId.make(`message:deferred-recovery:${suffix}:active`), + text: "Keep this run active.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const activeRun = (yield* orchestrator.getThreadProjection(targetThreadId)).runs[0]; + assert.isDefined(activeRun); + yield* orchestrator.dispatch({ + type: "thread.organization.defer", + commandId: CommandId.make(`command:deferred-recovery:${suffix}:schedule`), + threadId: targetThreadId, + runId: activeRun.id, + action: "settle", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:deferred-recovery:${suffix}:queued`), + threadId: targetThreadId, + messageId: MessageId.make(`message:deferred-recovery:${suffix}:queued`), + text: "This newer run makes the intent stale.", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + }); + const seeded = yield* orchestrator.getThreadProjection(targetThreadId); + assert.equal(seeded.thread.deferredOrganization?.runId, activeRun.id); + const queuedRun = seeded.runs.find((run) => run.status === "queued"); + assert.isDefined(queuedRun); + return { activeRunId: activeRun.id, queuedRunId: queuedRun.id }; + }); + + const unreadable = yield* seed(unreadableThreadId, "unreadable"); + const recoverable = yield* seed(threadId, "recoverable"); + return { unreadable, recoverable }; + }).pipe(Effect.provide(runtimeLayer("deferred-organization:first-runtime"))), + ); + + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql` + UPDATE orchestration_v2_projection_runs + SET payload_json = '{not-json' + WHERE run_id = ${runIds.unreadable.activeRunId} + `; + }).pipe(Effect.provide(databaseLayer)); + + const recovered = yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + yield* orchestrator.recoverDeferredOrganization; + return yield* orchestrator.getThreadProjection(threadId); + }).pipe(Effect.provide(runtimeLayer("deferred-organization:second-runtime"))), + ); + + assert.isNull(recovered.thread.deferredOrganization); + assert.isNull(recovered.thread.settledOverride); + assert.equal( + recovered.runs.find((run) => run.id === runIds.recoverable.activeRunId)?.status, + "starting", + ); + assert.equal( + recovered.runs.find((run) => run.id === runIds.recoverable.queuedRunId)?.status, + "queued", + ); + }).pipe(Effect.provide(NodeServices.layer)), + ), +); + +it.effect("discards an active-run intent after startup runtime reconciliation", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const tempDir = yield* Effect.acquireRelease( + fs.makeTempDirectory({ prefix: "t3-deferred-organization-runtime-recovery-" }), + (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie), + ); + const databaseLayer = makeSqlitePersistenceLive(path.join(tempDir, "state.sqlite")).pipe( + Layer.provide(NodeServices.layer), + ); + const registryLayer = makeProviderAdapterRegistryLayer([adapter]); + const runtimeLayer = (name: string) => + makeOrchestratorV2ReplayLayerWithRegistry( + { name, runtimePolicyOverride: { cwd: tempDir } }, + registryLayer, + { databaseLayer, runEffectWorker: false }, + ); + const threadId = ThreadId.make("thread:deferred-organization-runtime-recovery"); + + const runId = yield* Effect.scoped( Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; yield* orchestrator.dispatch({ type: "thread.create", createdBy: "user", creationSource: "web", - commandId: CommandId.make("command:deferred-recovery:create"), + commandId: CommandId.make("command:deferred-runtime-recovery:create"), threadId, - projectId: ProjectId.make("project:deferred-organization-recovery"), - title: "Deferred organization recovery", + projectId: ProjectId.make("project:deferred-organization-runtime-recovery"), + title: "Deferred organization runtime recovery", modelSelection, runtimeMode: "full-access", interactionMode: "default", @@ -76,10 +197,10 @@ it.effect("discards a stale deferred organization intent after runtime restart", type: "message.dispatch", createdBy: "user", creationSource: "web", - commandId: CommandId.make("command:deferred-recovery:active"), + commandId: CommandId.make("command:deferred-runtime-recovery:active"), threadId, - messageId: MessageId.make("message:deferred-recovery:active"), - text: "Keep this run active.", + messageId: MessageId.make("message:deferred-runtime-recovery:active"), + text: "Settle only after this run completes.", attachments: [], modelSelection, dispatchMode: { type: "start_immediately" }, @@ -88,42 +209,30 @@ it.effect("discards a stale deferred organization intent after runtime restart", assert.isDefined(activeRun); yield* orchestrator.dispatch({ type: "thread.organization.defer", - commandId: CommandId.make("command:deferred-recovery:schedule"), + commandId: CommandId.make("command:deferred-runtime-recovery:schedule"), threadId, runId: activeRun.id, action: "settle", }); - yield* orchestrator.dispatch({ - type: "message.dispatch", - createdBy: "user", - creationSource: "web", - commandId: CommandId.make("command:deferred-recovery:queued"), - threadId, - messageId: MessageId.make("message:deferred-recovery:queued"), - text: "This newer run makes the intent stale.", - attachments: [], - modelSelection, - dispatchMode: { type: "queue_after_active" }, - }); - const seeded = yield* orchestrator.getThreadProjection(threadId); - assert.equal(seeded.thread.deferredOrganization?.runId, activeRun.id); - const queuedRun = seeded.runs.find((run) => run.status === "queued"); - assert.isDefined(queuedRun); - return { activeRunId: activeRun.id, queuedRunId: queuedRun.id }; - }).pipe(Effect.provide(runtimeLayer("deferred-organization:first-runtime"))), + return activeRun.id; + }).pipe(Effect.provide(runtimeLayer("deferred-runtime-recovery:first-runtime"))), ); const recovered = yield* Effect.scoped( Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; + const providerRuntimeRecovery = yield* ProviderRuntimeRecoveryService; + yield* runV2RecoveryPhase({ + recoverProviderRuntime: providerRuntimeRecovery.recover, + recoverDeferredOrganization: orchestrator.recoverDeferredOrganization, + }); return yield* orchestrator.getThreadProjection(threadId); - }).pipe(Effect.provide(runtimeLayer("deferred-organization:second-runtime"))), + }).pipe(Effect.provide(runtimeLayer("deferred-runtime-recovery:second-runtime"))), ); + assert.equal(recovered.runs.find((run) => run.id === runId)?.status, "cancelled"); assert.isNull(recovered.thread.deferredOrganization); assert.isNull(recovered.thread.settledOverride); - assert.equal(recovered.runs.find((run) => run.id === runIds.activeRunId)?.status, "starting"); - assert.equal(recovered.runs.find((run) => run.id === runIds.queuedRunId)?.status, "queued"); }).pipe(Effect.provide(NodeServices.layer)), ), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a586fc48656f..3a3e360f659e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -178,6 +178,7 @@ export interface OrchestratorV2DispatchResult { export interface OrchestratorV2Shape { readonly resumeQueuedRuns: Effect.Effect; + readonly recoverDeferredOrganization: Effect.Effect; readonly dispatch: ( command: OrchestrationV2Command, ) => Effect.Effect; @@ -1435,64 +1436,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: movesForward ? { ...thread, lastVisitedAt: visitedAt.value } : thread, }); }); - - const queueProviderSessionDetaches = Effect.fn( - "orchestrationV2.dispatch.queueProviderSessionDetaches", - )(function* (input: { - readonly command: OrchestrationV2Command; - readonly threadId: ThreadId; - readonly events: Ref.Ref>; - readonly effects: Ref.Ref>; - readonly projection: OrchestrationV2ThreadProjection; - readonly providerSessionIds: ReadonlySet; - readonly occurredAt: DateTime.Utc; - readonly detail: string; - readonly revokeMcpCredential: boolean; - }) { - const liveSessions = input.projection.providerSessions.filter( - (session) => - input.providerSessionIds.has(session.id) && - session.status !== "stopped" && - session.status !== "error", - ); - yield* Effect.forEach( - liveSessions, - (session) => - Effect.gen(function* () { - yield* emit( - input.events, - input.command, - )({ - type: "provider-session.detached", - threadId: input.threadId, - driver: session.driver, - providerInstanceId: session.providerInstanceId, - occurredAt: input.occurredAt, - payload: { - providerSessionId: session.id, - detachedAt: input.occurredAt, - reason: input.detail, - }, - }); - yield* Ref.update(input.effects, (existing) => [ - ...existing, - { - id: `effect:${input.command.commandId}:provider-session.detach:${session.id}`, - commandId: input.command.commandId, - threadId: input.threadId, - request: { - type: "provider-session.detach", - providerSessionId: session.id, - detail: input.detail, - ...(input.revokeMcpCredential ? { revokeMcpCredential: true } : {}), - }, - } satisfies PendingOrchestrationEffectV2, - ]); - }), - { concurrency: 1, discard: true }, - ); - }); - const dispatchThreadMutation = Effect.fn("orchestrationV2.dispatch.threadMutation")(function* ( command: Extract< OrchestrationV2Command, @@ -1518,6 +1461,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio >, events: Ref.Ref>, effects: Ref.Ref>, + options?: { readonly clearDeferredOrganization?: boolean }, ) { const projection = yield* projectionStore.getThreadProjection(command.threadId).pipe( Effect.mapError( @@ -1818,6 +1762,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }; } })(); + const committedThread = + options?.clearDeferredOrganization === true + ? { ...updatedThread, deferredOrganization: null } + : updatedThread; const eventType = (() => { switch (command.type) { case "thread.archive": @@ -1859,9 +1807,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio )({ type: eventType, threadId: command.threadId, - providerInstanceId: updatedThread.providerInstanceId, + providerInstanceId: committedThread.providerInstanceId, occurredAt: now, - payload: updatedThread, + payload: committedThread, }); if (command.type === "thread.metadata.update" && command.regenerateTitle === true) { @@ -1955,28 +1903,73 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : (providerSwitchPlan?.releaseProviderSessionIds ?? []), ); if (detachSessionIds.size > 0) { - yield* queueProviderSessionDetaches({ - command, - threadId: command.threadId, - events, - effects, - projection, - providerSessionIds: detachSessionIds, - occurredAt: now, - detail: - command.type === "thread.archive" - ? "Thread archived." - : command.type === "thread.settle" - ? "Thread settled." - : command.type === "thread.delete" - ? "Thread deleted." - : command.type === "thread.metadata.update" - ? "Workspace changed." - : command.type === "thread.runtime-mode.set" - ? "Runtime mode changed." - : "Provider or model selection changed.", - revokeMcpCredential: command.type === "thread.archive" || command.type === "thread.delete", - }); + const liveSessions = projection.providerSessions.filter( + (session) => + detachSessionIds.has(session.id) && + session.status !== "stopped" && + session.status !== "error", + ); + yield* Effect.forEach( + liveSessions, + (session) => + Effect.gen(function* () { + yield* emit( + events, + command, + )({ + type: "provider-session.detached", + threadId: command.threadId, + driver: session.driver, + providerInstanceId: session.providerInstanceId, + occurredAt: now, + payload: { + providerSessionId: session.id, + detachedAt: now, + reason: + command.type === "thread.archive" + ? "Thread archived." + : command.type === "thread.settle" + ? "Thread settled." + : command.type === "thread.delete" + ? "Thread deleted." + : command.type === "thread.metadata.update" + ? "Workspace changed." + : command.type === "thread.runtime-mode.set" + ? "Runtime mode changed." + : "Provider or model selection changed.", + }, + }); + const pendingEffect = { + id: `effect:${command.commandId}:provider-session.detach:${session.id}`, + commandId: command.commandId, + threadId: command.threadId, + request: { + type: "provider-session.detach", + providerSessionId: session.id, + detail: + command.type === "thread.archive" + ? "Thread archived." + : command.type === "thread.settle" + ? "Thread settled." + : command.type === "thread.delete" + ? "Thread deleted." + : command.type === "thread.metadata.update" + ? "Workspace changed." + : command.type === "thread.runtime-mode.set" + ? "Runtime mode changed." + : "Provider or model selection changed.", + // Terminal detaches revoke the thread's MCP credentials; other + // detach reasons keep them so a re-attaching provider process + // stays authorized. + ...(command.type === "thread.archive" || command.type === "thread.delete" + ? { revokeMcpCredential: true } + : {}), + }, + } satisfies PendingOrchestrationEffectV2; + yield* Ref.update(effects, (existing) => [...existing, pendingEffect]); + }), + { concurrency: 1, discard: true }, + ); } if (command.type === "thread.archive" || command.type === "thread.delete") { @@ -2115,81 +2108,23 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); return; } - if (intent.action === "settle") { - yield* emit( - events, - command, - )({ - type: "thread.settled", - threadId: command.threadId, - providerInstanceId: thread.providerInstanceId, - occurredAt: now, - payload: { - ...thread, - deferredOrganization: null, - settledOverride: "settled", - settledAt: now, - pinnedAt: null, - pinOrderKey: null, - updatedAt: now, - }, - }); - yield* queueProviderSessionDetaches({ - command, - threadId: command.threadId, - events, - effects, - projection, - providerSessionIds: new Set(projection.providerSessions.map((session) => session.id)), - occurredAt: now, - detail: "Thread settled.", - revokeMcpCredential: false, - }); - return; - } - yield* emit( - events, - command, - )({ - type: "thread.archived", - threadId: command.threadId, - providerInstanceId: thread.providerInstanceId, - occurredAt: now, - payload: { - ...thread, - deferredOrganization: null, - archivedAt: now, - titleRegeneration: null, - updatedAt: now, - }, - }); - yield* disposeAllDelegatedCompletionCohorts({ - command, - events, - projection, - now, - cancelQueuedDelivery: false, - }); - yield* queueProviderSessionDetaches({ - command, - threadId: command.threadId, + yield* dispatchThreadMutation( + intent.action === "settle" + ? { + type: "thread.settle", + commandId: command.commandId, + threadId: command.threadId, + } + : { + type: "thread.archive", + commandId: command.commandId, + threadId: command.threadId, + requireNoPendingRuntimeRequests: true, + }, events, effects, - projection, - providerSessionIds: new Set(projection.providerSessions.map((session) => session.id)), - occurredAt: now, - detail: "Thread archived.", - revokeMcpCredential: true, - }); - yield* Ref.update(effects, (existing) => [ - ...existing, - { - id: `effect:${command.commandId}:terminal.cleanup`, - commandId: command.commandId, - threadId: command.threadId, - request: { type: "terminal.cleanup" }, - } satisfies PendingOrchestrationEffectV2, - ]); + { clearDeferredOrganization: true }, + ); }, ); @@ -7494,39 +7429,52 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // The high-water subscription deliberately skips history, so recover the // two terminal side effects from current projections instead: one queued // run per idle thread, plus any app-owned child result not yet transferred. - yield* projectionStore.getShellSnapshot().pipe( + const recoverDeferredOrganization = projectionStore.getShellSnapshot().pipe( Effect.flatMap((shell) => Effect.forEach( [...shell.threads, ...shell.archivedThreads].filter( (thread) => thread.deferredOrganization != null, ), (thread) => - threadDispatch.withLock( - thread.id, - Effect.gen(function* () { - const projection = yield* projectionStore.getThreadProjection(thread.id); - const intent = projection.thread.deferredOrganization; - if (intent == null) return; - const boundRun = projection.runs.find((run) => run.id === intent.runId); - const hasNewerRun = - boundRun !== undefined && - projection.runs.some((run) => run.ordinal > boundRun.ordinal); - if ( - boundRun === undefined || - hasNewerRun || - ["completed", "failed", "cancelled", "interrupted", "rolled_back"].includes( - boundRun.status, - ) - ) { - yield* applyDeferredOrganization(thread.id, intent.runId); - } - }), - ), + threadDispatch + .withLock( + thread.id, + Effect.gen(function* () { + const projection = yield* projectionStore.getThreadProjection(thread.id); + const intent = projection.thread.deferredOrganization; + if (intent == null) return; + const boundRun = projection.runs.find((run) => run.id === intent.runId); + const hasNewerRun = + boundRun !== undefined && + projection.runs.some((run) => run.ordinal > boundRun.ordinal); + if ( + boundRun === undefined || + hasNewerRun || + ["completed", "failed", "cancelled", "interrupted", "rolled_back"].includes( + boundRun.status, + ) + ) { + yield* applyDeferredOrganization(thread.id, intent.runId); + } + }), + ) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to recover deferred thread organization", { + threadId: thread.id, + cause, + }), + ), + ), { concurrency: 8, discard: true }, ), ), - Effect.catchCause((cause) => - Effect.logWarning("Failed to recover deferred thread organization", { cause }), + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: ThreadId.make("thread:shell"), + cause, + }), ), ); yield* resumeQueuedRuns.pipe( @@ -7625,6 +7573,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return OrchestratorV2.of({ resumeQueuedRuns, + recoverDeferredOrganization, dispatch: dispatchWithReceipt, getThreadProjection: (threadId) => projectionStore @@ -7726,6 +7675,13 @@ export const layerUnavailable: Layer.Layer = Layer.succeed( cause: "Orchestration V2 live runtime is not configured.", }), ), + recoverDeferredOrganization: Effect.fail( + new OrchestratorDispatchError({ + commandId: CommandId.make("command:system:recover-deferred-organization"), + commandType: "thread.organization.defer.apply", + cause: "Orchestration V2 live runtime is not configured.", + }), + ), dispatch: (command) => Effect.fail( new OrchestratorDispatchError({ diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 8dc51389ee76..65a6be2947ba 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -57,7 +57,8 @@ import { import { OrchestrationEffectWorkerV2 } from "./EffectWorker.ts"; import { EventSinkV2 } from "./EventSink.ts"; import { ProjectionMaintenanceV2 } from "./ProjectionMaintenance.ts"; -import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; +import type { ProviderAdapterV2SessionRuntime, ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; +import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { OrchestrationV2EventSinkLayerLive, OrchestrationV2LayerLive, @@ -98,12 +99,49 @@ const ProjectServiceTestLayer = Layer.mock(ProjectService.ProjectService)({ }); const driver = ProviderDriverKind.make("codex"); +const lifecycleProviderCapabilities = { + ...CodexProviderCapabilitiesV2, + sessions: { + ...CodexProviderCapabilitiesV2.sessions, + supportsMultipleProviderThreadsPerSession: false, + }, +}; const orchestrationAdapter = { instanceId: modelSelection.instanceId, driver, - getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + getCapabilities: () => Effect.succeed(lifecycleProviderCapabilities), planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), - openSession: () => Effect.die("sessions are not used by lifecycle tests"), + openSession: (input) => + Effect.gen(function* () { + const now = yield* DateTime.now; + return { + instanceId: modelSelection.instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: modelSelection.instanceId, + status: "ready", + cwd: input.runtimePolicy.cwd ?? process.cwd(), + model: input.modelSelection.model ?? "gpt-5.4", + capabilities: lifecycleProviderCapabilities, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.never, + ensureThread: () => Effect.die("ensureThread is unused by lifecycle tests"), + resumeThread: () => Effect.die("resumeThread is unused by lifecycle tests"), + startTurn: () => Effect.void, + steerTurn: () => Effect.void, + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.void, + readThreadSnapshot: () => Effect.die("readThreadSnapshot is unused by lifecycle tests"), + rollbackThread: () => Effect.die("rollbackThread is unused by lifecycle tests"), + forkThread: () => Effect.die("forkThread is unused by lifecycle tests"), + } satisfies ProviderAdapterV2SessionRuntime; + }), } as ProviderAdapterV2Shape; const providerInstance = { instanceId: modelSelection.instanceId, @@ -1439,6 +1477,8 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { const orchestrator = yield* OrchestratorV2; const eventSink = yield* EventSinkV2; const sql = yield* SqlClient.SqlClient; + const effectWorker = yield* OrchestrationEffectWorkerV2; + const providerSessions = yield* ProviderSessionManagerV2; const maintenance = yield* ProjectionMaintenanceV2; const threadId = ThreadId.make("runtime-layer-deferred-settle-thread"); @@ -1480,30 +1520,15 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { const providerSessionId = ProviderSessionId.make( "runtime-layer-deferred-settle-provider-session", ); - const providerSessionAttachedAt = yield* DateTime.now; - yield* eventSink.write({ - events: [ - { - id: EventId.make("runtime-layer-deferred-settle-provider-session-attached"), - type: "provider-session.attached", - threadId, - driver, - providerInstanceId: modelSelection.instanceId, - occurredAt: providerSessionAttachedAt, - payload: { - id: providerSessionId, - driver, - providerInstanceId: modelSelection.instanceId, - status: "ready", - cwd: "/tmp/runtime-layer-deferred-settle", - model: modelSelection.model, - capabilities: CodexProviderCapabilitiesV2, - createdAt: providerSessionAttachedAt, - updatedAt: providerSessionAttachedAt, - lastError: null, - }, - }, - ], + yield* providerSessions.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy: { + cwd: "/tmp/runtime-layer-deferred-settle", + runtimeMode: "full-access", + interactionMode: "default", + }, }); yield* orchestrator.dispatch({ type: "thread.organization.defer", @@ -1582,6 +1607,19 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { }, ], ); + assert.isAtLeast(yield* effectWorker.drain(), 1); + assert.isTrue(Option.isNone(yield* providerSessions.get(providerSessionId))); + const completedSettleEffects = yield* sql<{ readonly status: string }>` + SELECT status + FROM orchestration_v2_effect_outbox + WHERE command_id = ${applyCommand.commandId} + ORDER BY effect_id ASC + `; + assert.deepEqual( + completedSettleEffects.map((effect) => effect.status), + ["succeeded"], + ); + assert.equal(yield* effectWorker.drain(), 0); const rebuilt = yield* maintenance.rebuild; assert.isTrue(rebuilt.valid); @@ -1596,6 +1634,8 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { const orchestrator = yield* OrchestratorV2; const eventSink = yield* EventSinkV2; const sql = yield* SqlClient.SqlClient; + const effectWorker = yield* OrchestrationEffectWorkerV2; + const providerSessions = yield* ProviderSessionManagerV2; const createStartingThread = (suffix: string) => Effect.gen(function* () { @@ -1671,30 +1711,15 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { const providerSessionId = ProviderSessionId.make( "runtime-layer-deferred-archive-provider-session", ); - const providerSessionAttachedAt = yield* DateTime.now; - yield* eventSink.write({ - events: [ - { - id: EventId.make("runtime-layer-deferred-archive-provider-session-attached"), - type: "provider-session.attached", - threadId: archived.threadId, - driver, - providerInstanceId: modelSelection.instanceId, - occurredAt: providerSessionAttachedAt, - payload: { - id: providerSessionId, - driver, - providerInstanceId: modelSelection.instanceId, - status: "ready", - cwd: "/tmp/runtime-layer-deferred-archive", - model: modelSelection.model, - capabilities: CodexProviderCapabilitiesV2, - createdAt: providerSessionAttachedAt, - updatedAt: providerSessionAttachedAt, - lastError: null, - }, - }, - ], + yield* providerSessions.open({ + threadId: archived.threadId, + providerSessionId, + modelSelection, + runtimePolicy: { + cwd: "/tmp/runtime-layer-deferred-archive", + runtimeMode: "full-access", + interactionMode: "default", + }, }); yield* orchestrator.dispatch({ type: "thread.organization.defer", @@ -1711,8 +1736,11 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { const archiveApplyCommandId = CommandId.make( `command:system:thread-organization-defer:${archived.threadId}:${archived.run.id}`, ); - const archiveEffects = yield* sql<{ readonly payload_json: string }>` - SELECT payload_json + const archiveEffects = yield* sql<{ + readonly payload_json: string; + readonly status: string; + }>` + SELECT payload_json, status FROM orchestration_v2_effect_outbox WHERE command_id = ${archiveApplyCommandId} ORDER BY effect_id ASC @@ -1729,6 +1757,33 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { { type: "terminal.cleanup" }, ], ); + assert.deepEqual( + archiveEffects.map((effect) => effect.status), + ["pending", "pending"], + ); + assert.isAtLeast(yield* effectWorker.drain(), 2); + assert.isTrue(Option.isNone(yield* providerSessions.get(providerSessionId))); + const archiveRetry = yield* orchestrator.dispatch({ + type: "thread.organization.defer.apply", + commandId: archiveApplyCommandId, + threadId: archived.threadId, + runId: archived.run.id, + }); + assert.deepEqual( + archiveRetry.storedEvents.map((stored) => stored.event.type), + ["thread.archived", "provider-session.detached"], + ); + assert.equal(yield* effectWorker.drain(), 0); + const completedArchiveEffects = yield* sql<{ readonly status: string }>` + SELECT status + FROM orchestration_v2_effect_outbox + WHERE command_id = ${archiveApplyCommandId} + ORDER BY effect_id ASC + `; + assert.deepEqual( + completedArchiveEffects.map((effect) => effect.status), + ["succeeded", "succeeded"], + ); const stale = yield* createStartingThread("stale"); yield* orchestrator.dispatch({ diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 73750176fd39..5f011100dc02 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -37,6 +37,10 @@ import { layer as projectionStoreLayer } from "../ProjectionStore.ts"; import { OrchestratorV2, type OrchestratorV2Error } from "../Orchestrator.ts"; import { ProviderAdapterRegistryV2 } from "../ProviderAdapterRegistry.ts"; import { ProviderAuthService } from "../../provider/Services/ProviderAuthService.ts"; +import { + ProviderRuntimeRecoveryService, + layer as providerRuntimeRecoveryLayer, +} from "../ProviderRuntimeRecoveryService.ts"; import { layer as providerEventIngestorLayer } from "../ProviderEventIngestor.ts"; import { layerWithOptions as providerSessionManagerLayerWithOptions } from "../ProviderSessionManager.ts"; import { layer as providerSwitchServiceLayer } from "../ProviderSwitchService.ts"; @@ -221,7 +225,10 @@ export function makeOrchestratorV2ProviderReplayLayer< readonly runEffectWorker?: boolean; readonly replayGate?: ProviderReplayGate; } = {}, -): Layer.Layer { +): Layer.Layer< + OrchestratorV2 | ProviderRuntimeRecoveryService, + Error | MigrationError | PlatformError.PlatformError | SqlError +> { const registryLayer = harness.makeProviderAdapterRegistryLayer( scenario.transcript, options.replayGate === undefined ? {} : { replayGate: options.replayGate }, @@ -240,7 +247,10 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( readonly enableLegacyTokenStreaming?: boolean; readonly runEffectWorker?: boolean; } = {}, -): Layer.Layer { +): Layer.Layer< + OrchestratorV2 | ProviderRuntimeRecoveryService, + Error | MigrationError | PlatformError.PlatformError | SqlError +> { const serverConfigLayer = Layer.effect( ServerConfig, makeReplayServerConfig(scenario.name).pipe(Effect.orDie), @@ -396,7 +406,16 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( ), ), ); - const replayRuntime = Layer.merge(orchestratorProvided, effectWorkerProvided).pipe( + const providerRuntimeRecoveryProvided = providerRuntimeRecoveryLayer.pipe( + Layer.provide( + Layer.mergeAll(effectWorkerProvided, storesLayer, eventSinkProvided, idAllocatorLayer), + ), + ); + const replayRuntime = Layer.mergeAll( + orchestratorProvided, + effectWorkerProvided, + providerRuntimeRecoveryProvided, + ).pipe( Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer), ); @@ -405,9 +424,11 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( // orchestrator. Keeping this acquisition in the replay layer makes the // outbox lifecycle explicit and prevents test-only command-side draining. if (options.runEffectWorker === false) { - return orchestratorProvided.pipe(Layer.provide(NodeServices.layer)); + return Layer.merge(orchestratorProvided, providerRuntimeRecoveryProvided).pipe( + Layer.provide(NodeServices.layer), + ); } - return Layer.effect( + const orchestratorWithWorker = Layer.effect( OrchestratorV2, Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; @@ -415,4 +436,5 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( return orchestrator; }), ).pipe(Layer.provide(replayRuntime)); + return Layer.merge(orchestratorWithWorker, providerRuntimeRecoveryProvided); } diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 0ed8529355b6..e9c80eabd04b 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -31,6 +31,7 @@ import * as EffectWorker from "./orchestration-v2/EffectWorker.ts"; import * as LegacyV1ThreadImporter from "./orchestration-v2/LegacyV1ThreadImporter.ts"; import * as ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; import * as ProviderRuntimeRecovery from "./orchestration-v2/ProviderRuntimeRecoveryService.ts"; +import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; import * as ThreadLaunch from "./orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; @@ -426,6 +427,17 @@ export function runOrderedV2StartupPhases< }); } +export const runV2RecoveryPhase = < + Recovery, + RecoveryError, + RecoveryContext, + DeferredError, + DeferredContext, +>(input: { + readonly recoverProviderRuntime: Effect.Effect; + readonly recoverDeferredOrganization: Effect.Effect; +}) => input.recoverProviderRuntime.pipe(Effect.tap(() => input.recoverDeferredOrganization)); + export const make = (options?: StartupOptions) => Effect.gen(function* () { const serverConfig = yield* ServerConfig.ServerConfig; @@ -433,6 +445,7 @@ export const make = (options?: StartupOptions) => const projectionMaintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; const legacyV1ThreadImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; const providerRuntimeRecovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const orchestrator = yield* Orchestrator.OrchestratorV2; const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2; const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay; const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents; @@ -552,7 +565,13 @@ export const make = (options?: StartupOptions) => "orchestration-v2.projections.rebuild", projectionMaintenance.rebuild, ), - recover: runStartupPhase("orchestration-v2.recovery", providerRuntimeRecovery.recover), + recover: runStartupPhase( + "orchestration-v2.recovery", + runV2RecoveryPhase({ + recoverProviderRuntime: providerRuntimeRecovery.recover, + recoverDeferredOrganization: orchestrator.recoverDeferredOrganization, + }), + ), startEffectWorker: runStartupPhase( "orchestration-v2.effect-worker.start", startEffectWorkerWithRelay({ diff --git a/packages/contracts/src/orchestratorMcp.test.ts b/packages/contracts/src/orchestratorMcp.test.ts index 982da87085e0..339795885afa 100644 --- a/packages/contracts/src/orchestratorMcp.test.ts +++ b/packages/contracts/src/orchestratorMcp.test.ts @@ -240,5 +240,10 @@ describe("orchestrator MCP contracts", () => { expect(() => decodeThreadDeferOrganizationInput({ operation: "schedule", action: "delete" }), ).toThrow(); + expect(() => decodeThreadDeferOrganizationInput({ operation: "schedule" })).toThrow(); + expect(() => + decodeThreadDeferOrganizationInput({ operation: "read", action: "settle" }), + ).toThrow(); + expect(() => decodeThreadDeferOrganizationInput({ operation: "later" })).toThrow(); }); }); diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 925fade28ef3..924ae6f437c4 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -538,23 +538,26 @@ export const OrchestratorMcpDeferredOrganizationIntent = Schema.Struct({ export type OrchestratorMcpDeferredOrganizationIntent = typeof OrchestratorMcpDeferredOrganizationIntent.Type; -export const OrchestratorMcpThreadDeferOrganizationInput = Schema.Union([ - Schema.Struct({ - operation: Schema.Literal("read"), +export const OrchestratorMcpThreadDeferOrganizationInput = Schema.Struct({ + operation: Schema.Literals(["read", "schedule", "cancel"]).annotate({ + description: + "Read the current intent, schedule settle/archive after this run, or cancel the intent.", }), - Schema.Struct({ - operation: Schema.Literal("schedule"), - action: Schema.Literals(["settle", "archive"]).annotate({ + action: Schema.optional( + Schema.Literals(["settle", "archive"]).annotate({ description: - "Organization action to apply only after this calling run completes safely with no newer, queued, or blocked work.", + "Required for schedule only. Applies after this calling run completes safely with no newer, queued, or blocked work.", }), - clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), - }), - Schema.Struct({ - operation: Schema.Literal("cancel"), - clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), + ), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), +}).check( + Schema.makeFilter((input) => { + if (input.operation === "schedule") { + return input.action !== undefined || "Schedule requires an action."; + } + return input.action === undefined || `${input.operation} does not accept an action.`; }), -]); +); export type OrchestratorMcpThreadDeferOrganizationInput = typeof OrchestratorMcpThreadDeferOrganizationInput.Type; From 63289692b4af221dc9a868322829d47e0150a6e9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 10:13:46 -0700 Subject: [PATCH 05/10] test(orchestration): provide replay recovery dependencies --- .../orchestration-v2/testkit/ProviderReplayHarness.ts | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 5f011100dc02..3058e88798bf 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -415,16 +415,14 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( orchestratorProvided, effectWorkerProvided, providerRuntimeRecoveryProvided, - ).pipe( - Layer.provide(worktreeRepairDependenciesTestLayer), - Layer.provide(NodeServices.layer), - ); + ).pipe(Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer)); // Build the daemon from the exact worker instance exposed alongside the // orchestrator. Keeping this acquisition in the replay layer makes the // outbox lifecycle explicit and prevents test-only command-side draining. if (options.runEffectWorker === false) { return Layer.merge(orchestratorProvided, providerRuntimeRecoveryProvided).pipe( + Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer), ); } @@ -436,5 +434,8 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( return orchestrator; }), ).pipe(Layer.provide(replayRuntime)); - return Layer.merge(orchestratorWithWorker, providerRuntimeRecoveryProvided); + return Layer.merge(orchestratorWithWorker, providerRuntimeRecoveryProvided).pipe( + Layer.provide(worktreeRepairDependenciesTestLayer), + Layer.provide(NodeServices.layer), + ); } From 76244404aea27c4d4c05a717e81c2d35f7d8f829 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 10:37:07 -0700 Subject: [PATCH 06/10] fix(orchestration): preserve recovery interruption --- ...edOrganizationRecovery.integration.test.ts | 42 ++++++++++++++++--- .../src/orchestration-v2/Orchestrator.ts | 39 ++++++++++++----- 2 files changed, 66 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts index 1724149aa622..0bf461a0be0d 100644 --- a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts @@ -9,19 +9,23 @@ import { ProviderInstanceId, ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite.ts"; import { runV2RecoveryPhase } from "../serverRuntimeStartup.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; -import { OrchestratorV2 } from "./Orchestrator.ts"; +import * as Orchestrator from "./Orchestrator.ts"; import { ProviderRuntimeRecoveryService } from "./ProviderRuntimeRecoveryService.ts"; import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; import { makeLayer as makeProviderAdapterRegistryLayer } from "./ProviderAdapterRegistry.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; const modelSelection = { @@ -37,6 +41,34 @@ const adapter = { openSession: () => Effect.die("provider sessions are not used in recovery coverage"), } as ProviderAdapterV2Shape; +it.effect("retries one typed deferred repair failure without swallowing interruption", () => + Effect.gen(function* () { + const attempts = yield* Ref.make(0); + yield* Orchestrator.runDeferredOrganizationRepair( + ThreadId.make("thread:deferred-organization-transient-repair"), + Effect.gen(function* () { + const attempt = yield* Ref.updateAndGet(attempts, (current) => current + 1); + if (attempt === 1) { + return yield* new ProjectionStore.ProjectionStoreReadError({ + threadId: ThreadId.make("thread:deferred-organization-transient-repair"), + cause: "transient projection failure", + }); + } + }), + ); + assert.equal(yield* Ref.get(attempts), 2); + + const interruption = yield* Orchestrator.runDeferredOrganizationRepair( + ThreadId.make("thread:deferred-organization-interruption"), + Effect.interrupt, + ).pipe(Effect.exit); + assert.isTrue(Exit.isFailure(interruption)); + if (Exit.isFailure(interruption)) { + assert.isTrue(Cause.hasInterruptsOnly(interruption.cause)); + } + }), +); + it.effect("discards a stale deferred organization intent after runtime restart", () => Effect.scoped( Effect.gen(function* () { @@ -61,7 +93,7 @@ it.effect("discards a stale deferred organization intent after runtime restart", const runIds = yield* Effect.scoped( Effect.gen(function* () { - const orchestrator = yield* OrchestratorV2; + const orchestrator = yield* Orchestrator.OrchestratorV2; const seed = (targetThreadId: ThreadId, suffix: string) => Effect.gen(function* () { yield* orchestrator.dispatch({ @@ -135,7 +167,7 @@ it.effect("discards a stale deferred organization intent after runtime restart", const recovered = yield* Effect.scoped( Effect.gen(function* () { - const orchestrator = yield* OrchestratorV2; + const orchestrator = yield* Orchestrator.OrchestratorV2; yield* orchestrator.recoverDeferredOrganization; return yield* orchestrator.getThreadProjection(threadId); }).pipe(Effect.provide(runtimeLayer("deferred-organization:second-runtime"))), @@ -178,7 +210,7 @@ it.effect("discards an active-run intent after startup runtime reconciliation", const runId = yield* Effect.scoped( Effect.gen(function* () { - const orchestrator = yield* OrchestratorV2; + const orchestrator = yield* Orchestrator.OrchestratorV2; yield* orchestrator.dispatch({ type: "thread.create", createdBy: "user", @@ -220,7 +252,7 @@ it.effect("discards an active-run intent after startup runtime reconciliation", const recovered = yield* Effect.scoped( Effect.gen(function* () { - const orchestrator = yield* OrchestratorV2; + const orchestrator = yield* Orchestrator.OrchestratorV2; const providerRuntimeRecovery = yield* ProviderRuntimeRecoveryService; yield* runV2RecoveryPhase({ recoverProviderRuntime: providerRuntimeRecovery.recover, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3a3e360f659e..df49aaaa3215 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -30,6 +30,7 @@ import { ThreadId, } from "@t3tools/contracts"; import { modelSelectionsEqual } from "@t3tools/shared/model"; +import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -54,6 +55,7 @@ import { applyToProjection, emptyProjection, isTurnItemAtOrBeforeRun, + ProjectionStoreReadError, ProjectionStoreV2, type ProjectionCheckpointContext, } from "./ProjectionStore.ts"; @@ -97,6 +99,29 @@ export class OrchestratorProjectionError extends Schema.TaggedErrorClass( + threadId: ThreadId, + repair: Effect.Effect, +) => + repair.pipe( + Effect.retry({ + times: 1, + while: (error) => isProjectionStoreReadError(error) || isOrchestratorDispatchError(error), + }), + Effect.asVoid, + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("Failed to recover deferred thread organization", { + threadId, + cause, + }), + ), + ); + export class OrchestratorDomainEventStreamError extends Schema.TaggedErrorClass()( "OrchestratorDomainEventStreamError", { @@ -7436,8 +7461,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (thread) => thread.deferredOrganization != null, ), (thread) => - threadDispatch - .withLock( + runDeferredOrganizationRepair( + thread.id, + threadDispatch.withLock( thread.id, Effect.gen(function* () { const projection = yield* projectionStore.getThreadProjection(thread.id); @@ -7457,15 +7483,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* applyDeferredOrganization(thread.id, intent.runId); } }), - ) - .pipe( - Effect.catchCause((cause) => - Effect.logWarning("Failed to recover deferred thread organization", { - threadId: thread.id, - cause, - }), - ), ), + ), { concurrency: 8, discard: true }, ), ), From 923a2d19240b154987ab5f8e3cc8f3819c93e38d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 13:26:42 -0700 Subject: [PATCH 07/10] fix(orchestration): retry deferred projection reads --- ...ferredOrganizationRecovery.integration.test.ts | 15 +++++++++++++++ apps/server/src/orchestration-v2/Orchestrator.ts | 6 +++++- docs/user/thread-sidebar.md | 5 +++-- 3 files changed, 23 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts index 0bf461a0be0d..532aed2923e3 100644 --- a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts @@ -58,6 +58,21 @@ it.effect("retries one typed deferred repair failure without swallowing interrup ); assert.equal(yield* Ref.get(attempts), 2); + const mappedAttempts = yield* Ref.make(0); + yield* Orchestrator.runDeferredOrganizationRepair( + ThreadId.make("thread:deferred-organization-transient-projection"), + Effect.gen(function* () { + const attempt = yield* Ref.updateAndGet(mappedAttempts, (current) => current + 1); + if (attempt === 1) { + return yield* new Orchestrator.OrchestratorProjectionError({ + threadId: ThreadId.make("thread:deferred-organization-transient-projection"), + cause: "transient mapped projection failure", + }); + } + }), + ); + assert.equal(yield* Ref.get(mappedAttempts), 2); + const interruption = yield* Orchestrator.runDeferredOrganizationRepair( ThreadId.make("thread:deferred-organization-interruption"), Effect.interrupt, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index df49aaaa3215..e0ddee698a6b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -101,6 +101,7 @@ export class OrchestratorProjectionError extends Schema.TaggedErrorClass( threadId: ThreadId, @@ -109,7 +110,10 @@ export const runDeferredOrganizationRepair = ( repair.pipe( Effect.retry({ times: 1, - while: (error) => isProjectionStoreReadError(error) || isOrchestratorDispatchError(error), + while: (error) => + isProjectionStoreReadError(error) || + isOrchestratorDispatchError(error) || + isOrchestratorProjectionError(error), }), Effect.asVoid, Effect.catchCause((cause) => diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 04de4e3874a1..902145414b00 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -44,8 +44,9 @@ or a user request is still pending. An agent can also ask T3 Code to settle or archive its own thread after the current run finishes. The server applies the saved request only when that run succeeds and -no newer, queued, or approval-blocked work remains. A new message or unresolved -request keeps the thread visible and discards the stale organization request. +no newer, queued, or approval-blocked work remains. A new message that creates a +newer run, or an unresolved request, keeps the thread visible and discards the stale +organization request. ## Settle finished work From 4b24547abc7181b0547a596ba794acc106e435a2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 13:34:56 -0700 Subject: [PATCH 08/10] fix(orchestration): retry deferred apply reads before receipts --- ...edOrganizationRecovery.integration.test.ts | 157 ++++++++++++++++-- .../src/orchestration-v2/Orchestrator.ts | 17 +- .../testkit/ProviderReplayHarness.ts | 47 ++++-- 3 files changed, 187 insertions(+), 34 deletions(-) diff --git a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts index 532aed2923e3..f52acab2b70e 100644 --- a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts @@ -14,6 +14,7 @@ import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Ref from "effect/Ref"; import * as SqlClient from "effect/unstable/sql/SqlClient"; @@ -21,6 +22,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite.ts"; import { runV2RecoveryPhase } from "../serverRuntimeStartup.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; import { ProviderRuntimeRecoveryService } from "./ProviderRuntimeRecoveryService.ts"; import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; @@ -58,21 +60,6 @@ it.effect("retries one typed deferred repair failure without swallowing interrup ); assert.equal(yield* Ref.get(attempts), 2); - const mappedAttempts = yield* Ref.make(0); - yield* Orchestrator.runDeferredOrganizationRepair( - ThreadId.make("thread:deferred-organization-transient-projection"), - Effect.gen(function* () { - const attempt = yield* Ref.updateAndGet(mappedAttempts, (current) => current + 1); - if (attempt === 1) { - return yield* new Orchestrator.OrchestratorProjectionError({ - threadId: ThreadId.make("thread:deferred-organization-transient-projection"), - cause: "transient mapped projection failure", - }); - } - }), - ); - assert.equal(yield* Ref.get(mappedAttempts), 2); - const interruption = yield* Orchestrator.runDeferredOrganizationRepair( ThreadId.make("thread:deferred-organization-interruption"), Effect.interrupt, @@ -84,6 +71,146 @@ it.effect("retries one typed deferred repair failure without swallowing interrup }), ); +it.effect("retries a transient deferred apply read before recording its receipt", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const tempDir = yield* Effect.acquireRelease( + fs.makeTempDirectory({ prefix: "t3-deferred-organization-receipt-" }), + (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie), + ); + const databaseLayer = makeSqlitePersistenceLive(path.join(tempDir, "state.sqlite")).pipe( + Layer.provide(NodeServices.layer), + ); + const registryLayer = makeProviderAdapterRegistryLayer([adapter]); + const threadId = ThreadId.make("thread:deferred-organization-receipt"); + + const runId = yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-organization-receipt:create"), + threadId, + projectId: ProjectId.make("project:deferred-organization-receipt"), + title: "Deferred organization receipt", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: tempDir, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-organization-receipt:active"), + threadId, + messageId: MessageId.make("message:deferred-organization-receipt:active"), + text: "Keep this run active.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const activeRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]; + assert.isDefined(activeRun); + yield* orchestrator.dispatch({ + type: "thread.organization.defer", + commandId: CommandId.make("command:deferred-organization-receipt:schedule"), + threadId, + runId: activeRun.id, + action: "settle", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-organization-receipt:queued"), + threadId, + messageId: MessageId.make("message:deferred-organization-receipt:queued"), + text: "Make the deferred intent stale.", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + }); + return activeRun.id; + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "deferred-organization-receipt:first-runtime", + runtimePolicyOverride: { cwd: tempDir }, + }, + registryLayer, + { databaseLayer, runEffectWorker: false }, + ), + ), + ), + ); + const commandId = CommandId.make( + `command:system:thread-organization-defer:${threadId}:${runId}`, + ); + + const armed = yield* Ref.make(false); + const projectionReads = yield* Ref.make(0); + const decorateProjectionStore = (store: ProjectionStore.ProjectionStoreV2["Service"]) => + ProjectionStore.ProjectionStoreV2.of({ + ...store, + getThreadProjection: (requestedThreadId) => + Ref.get(armed).pipe( + Effect.flatMap((isArmed) => + isArmed + ? Ref.updateAndGet(projectionReads, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === 3 + ? Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId: requestedThreadId, + cause: "simulated transient deferred apply read failure", + }), + ) + : store.getThreadProjection(requestedThreadId), + ), + ) + : store.getThreadProjection(requestedThreadId), + ), + ), + }); + + yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const receipts = yield* CommandReceiptStoreV2; + yield* Ref.set(armed, true); + yield* orchestrator.recoverDeferredOrganization; + + const receipt = yield* receipts.getByCommandId(commandId); + assert.isTrue(Option.isSome(receipt)); + if (Option.isSome(receipt)) assert.equal(receipt.value.status, "accepted"); + assert.equal(yield* Ref.get(projectionReads), 4); + assert.isNull( + (yield* orchestrator.getThreadProjection(threadId)).thread.deferredOrganization, + ); + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "deferred-organization-receipt:second-runtime", + runtimePolicyOverride: { cwd: tempDir }, + }, + registryLayer, + { databaseLayer, runEffectWorker: false, decorateProjectionStore }, + ), + ), + ), + ); + }).pipe(Effect.provide(NodeServices.layer)), + ), +); + it.effect("discards a stale deferred organization intent after runtime restart", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e0ddee698a6b..25b3bf8e28ec 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -110,10 +110,7 @@ export const runDeferredOrganizationRepair = ( repair.pipe( Effect.retry({ times: 1, - while: (error) => - isProjectionStoreReadError(error) || - isOrchestratorDispatchError(error) || - isOrchestratorProjectionError(error), + while: (error) => isProjectionStoreReadError(error) || isOrchestratorDispatchError(error), }), Effect.asVoid, Effect.catchCause((cause) => @@ -7308,7 +7305,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } satisfies OrchestratorV2DispatchResult; } - const plan = yield* dispatchOnce(command).pipe( + const planAttempt = dispatchOnce(command); + const plan = yield* ( + command.type === "thread.organization.defer.apply" + ? planAttempt.pipe( + Effect.retry({ + times: 1, + while: isOrchestratorProjectionError, + }), + ) + : planAttempt + ).pipe( Effect.flatMap((planned) => planned.events.length > 0 ? Effect.succeed(planned) diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 3058e88798bf..10fae802cb9e 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -21,7 +21,10 @@ import { layer as checkpointCaptureServiceLayer } from "../CheckpointCaptureServ import { layer as checkpointServiceLayer } from "../CheckpointService.ts"; import { layer as checkpointRollbackServiceLayer } from "../CheckpointRollbackService.ts"; import { layer as commandPolicyLayer } from "../CommandPolicy.ts"; -import { layer as commandReceiptStoreLayer } from "../CommandReceiptStore.ts"; +import { + CommandReceiptStoreV2, + layer as commandReceiptStoreLayer, +} from "../CommandReceiptStore.ts"; import { layer as contextHandoffServiceLayer } from "../ContextHandoffService.ts"; import { layer as effectOutboxLayer } from "../EffectOutbox.ts"; import { @@ -33,7 +36,7 @@ import { layerFromStores as eventSinkLayer } from "../EventSink.ts"; import { layer as eventStoreLayer } from "../EventStore.ts"; import { layer as idAllocatorLayer } from "../IdAllocator.ts"; import { layer as orchestratorLayer } from "../Orchestrator.ts"; -import { layer as projectionStoreLayer } from "../ProjectionStore.ts"; +import { layer as projectionStoreLayer, ProjectionStoreV2 } from "../ProjectionStore.ts"; import { OrchestratorV2, type OrchestratorV2Error } from "../Orchestrator.ts"; import { ProviderAdapterRegistryV2 } from "../ProviderAdapterRegistry.ts"; import { ProviderAuthService } from "../../provider/Services/ProviderAuthService.ts"; @@ -223,10 +226,13 @@ export function makeOrchestratorV2ProviderReplayLayer< >; readonly enableLegacyTokenStreaming?: boolean; readonly runEffectWorker?: boolean; + readonly decorateProjectionStore?: ( + service: ProjectionStoreV2["Service"], + ) => ProjectionStoreV2["Service"]; readonly replayGate?: ProviderReplayGate; } = {}, ): Layer.Layer< - OrchestratorV2 | ProviderRuntimeRecoveryService, + OrchestratorV2 | ProviderRuntimeRecoveryService | CommandReceiptStoreV2, Error | MigrationError | PlatformError.PlatformError | SqlError > { const registryLayer = harness.makeProviderAdapterRegistryLayer( @@ -246,9 +252,12 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( >; readonly enableLegacyTokenStreaming?: boolean; readonly runEffectWorker?: boolean; + readonly decorateProjectionStore?: ( + service: ProjectionStoreV2["Service"], + ) => ProjectionStoreV2["Service"]; } = {}, ): Layer.Layer< - OrchestratorV2 | ProviderRuntimeRecoveryService, + OrchestratorV2 | ProviderRuntimeRecoveryService | CommandReceiptStoreV2, Error | MigrationError | PlatformError.PlatformError | SqlError > { const serverConfigLayer = Layer.effect( @@ -265,13 +274,21 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( const serverSettingsLayer = ServerSettingsService.layerTest({ enableLegacyTokenStreaming: options.enableLegacyTokenStreaming ?? false, }).pipe(Layer.orDie); - const storesLayer = Layer.mergeAll( + const otherStoresLayer = Layer.mergeAll( eventStoreLayer, - projectionStoreLayer, commandReceiptStoreLayer, effectOutboxLayer, turnItemPositionStoreLayer, ).pipe(Layer.provide(databaseLayer)); + const projectionStoreProvided = projectionStoreLayer.pipe(Layer.provide(databaseLayer)); + const selectedProjectionStoreLayer = + options.decorateProjectionStore === undefined + ? projectionStoreProvided + : Layer.effect( + ProjectionStoreV2, + ProjectionStoreV2.pipe(Effect.map(options.decorateProjectionStore)), + ).pipe(Layer.provide(projectionStoreProvided)); + const storesLayer = Layer.merge(otherStoresLayer, selectedProjectionStoreLayer); const eventSinkProvided = eventSinkLayer.pipe( Layer.provide(Layer.mergeAll(storesLayer, databaseLayer)), ); @@ -421,10 +438,11 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( // orchestrator. Keeping this acquisition in the replay layer makes the // outbox lifecycle explicit and prevents test-only command-side draining. if (options.runEffectWorker === false) { - return Layer.merge(orchestratorProvided, providerRuntimeRecoveryProvided).pipe( - Layer.provide(worktreeRepairDependenciesTestLayer), - Layer.provide(NodeServices.layer), - ); + return Layer.mergeAll( + orchestratorProvided, + providerRuntimeRecoveryProvided, + commandReceiptStoreProvided, + ).pipe(Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer)); } const orchestratorWithWorker = Layer.effect( OrchestratorV2, @@ -434,8 +452,9 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( return orchestrator; }), ).pipe(Layer.provide(replayRuntime)); - return Layer.merge(orchestratorWithWorker, providerRuntimeRecoveryProvided).pipe( - Layer.provide(worktreeRepairDependenciesTestLayer), - Layer.provide(NodeServices.layer), - ); + return Layer.mergeAll( + orchestratorWithWorker, + providerRuntimeRecoveryProvided, + commandReceiptStoreProvided, + ).pipe(Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer)); } From 04bbc3329fd97bfee70edf9c5e8e16e0214ca792 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 4 Sep 2026 23:04:11 -0700 Subject: [PATCH 09/10] fix(orchestration): preserve v2 thread deletion planning --- .../src/orchestration-v2/Orchestrator.ts | 34 +++++++------------ .../src/orchestration-v2/runtimeLayer.test.ts | 19 ++++++++--- .../testkit/ProviderReplayHarness.ts | 1 + 3 files changed, 28 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 25b3bf8e28ec..d4295aac9a6d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1912,9 +1912,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // decided serially against the projection, so a turn start that // re-engages the thread cannot race this detach. const detachSessionIds = new Set( - command.type === "thread.archive" || - command.type === "thread.delete" || - command.type === "thread.settle" + command.type === "thread.archive" || command.type === "thread.settle" ? projection.providerSessions.map((session) => session.id) : command.type === "thread.metadata.update" && command.worktreePath !== undefined && @@ -1956,13 +1954,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? "Thread archived." : command.type === "thread.settle" ? "Thread settled." - : command.type === "thread.delete" - ? "Thread deleted." - : command.type === "thread.metadata.update" - ? "Workspace changed." - : command.type === "thread.runtime-mode.set" - ? "Runtime mode changed." - : "Provider or model selection changed.", + : command.type === "thread.metadata.update" + ? "Workspace changed." + : command.type === "thread.runtime-mode.set" + ? "Runtime mode changed." + : "Provider or model selection changed.", }, }); const pendingEffect = { @@ -1977,19 +1973,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? "Thread archived." : command.type === "thread.settle" ? "Thread settled." - : command.type === "thread.delete" - ? "Thread deleted." - : command.type === "thread.metadata.update" - ? "Workspace changed." - : command.type === "thread.runtime-mode.set" - ? "Runtime mode changed." - : "Provider or model selection changed.", + : command.type === "thread.metadata.update" + ? "Workspace changed." + : command.type === "thread.runtime-mode.set" + ? "Runtime mode changed." + : "Provider or model selection changed.", // Terminal detaches revoke the thread's MCP credentials; other // detach reasons keep them so a re-attaching provider process // stays authorized. - ...(command.type === "thread.archive" || command.type === "thread.delete" - ? { revokeMcpCredential: true } - : {}), + ...(command.type === "thread.archive" ? { revokeMcpCredential: true } : {}), }, } satisfies PendingOrchestrationEffectV2; yield* Ref.update(effects, (existing) => [...existing, pendingEffect]); @@ -1998,7 +1990,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); } - if (command.type === "thread.archive" || command.type === "thread.delete") { + if (command.type === "thread.archive") { yield* Ref.update(effects, (existing) => [ ...existing, { diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 65a6be2947ba..abebc90966eb 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -24,6 +24,7 @@ import * as Effect from "effect/Effect"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; @@ -908,7 +909,7 @@ it.layer(LegacyImportTestLayer)("OrchestrationV2 legacy import", (it) => { ); }); -it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { +it.layer(Layer.merge(TestLayer, NodeServices.layer))("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("applies lifecycle commands idempotently and emits archive/removal shell deltas", () => Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; @@ -1474,6 +1475,10 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("applies deferred settlement after safe completion and replays its receipt", () => Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const workspacePath = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-runtime-layer-deferred-settle-", + }); const orchestrator = yield* OrchestratorV2; const eventSink = yield* EventSinkV2; const sql = yield* SqlClient.SqlClient; @@ -1494,7 +1499,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { runtimeMode: "full-access", interactionMode: "default", branch: null, - worktreePath: "/tmp/runtime-layer-deferred-settle", + worktreePath: workspacePath, }); yield* orchestrator.dispatch({ type: "thread.pin", @@ -1525,7 +1530,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { providerSessionId, modelSelection, runtimePolicy: { - cwd: "/tmp/runtime-layer-deferred-settle", + cwd: workspacePath, runtimeMode: "full-access", interactionMode: "default", }, @@ -1631,6 +1636,10 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("archives after safe completion and discards stale or blocked deferred intents", () => Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const workspacePath = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-runtime-layer-deferred-organization-", + }); const orchestrator = yield* OrchestratorV2; const eventSink = yield* EventSinkV2; const sql = yield* SqlClient.SqlClient; @@ -1652,7 +1661,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { runtimeMode: "full-access", interactionMode: "default", branch: null, - worktreePath: `/tmp/runtime-layer-deferred-${suffix}`, + worktreePath: workspacePath, }); yield* orchestrator.dispatch({ type: "message.dispatch", @@ -1716,7 +1725,7 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { providerSessionId, modelSelection, runtimePolicy: { - cwd: "/tmp/runtime-layer-deferred-archive", + cwd: workspacePath, runtimeMode: "full-access", interactionMode: "default", }, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 10fae802cb9e..b39c5a1b5d2f 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -427,6 +427,7 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( Layer.provide( Layer.mergeAll(effectWorkerProvided, storesLayer, eventSinkProvided, idAllocatorLayer), ), + Layer.provide(serverSettingsLayer), ); const replayRuntime = Layer.mergeAll( orchestratorProvided, From 6ff0395a3cf73f8b756c352b492b3d5a205c0959 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 4 Sep 2026 23:39:31 -0700 Subject: [PATCH 10/10] fix(orchestration): protect deferred organization lifecycle --- ...edOrganizationRecovery.integration.test.ts | 187 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 26 ++- .../testkit/ProviderReplayHarness.ts | 8 +- .../contracts/src/orchestrationV2.test.ts | 23 +++ packages/contracts/src/orchestrationV2.ts | 15 +- 5 files changed, 246 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts index f52acab2b70e..659294cdba68 100644 --- a/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts @@ -2,6 +2,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { CommandId, + EventId, MessageId, type ModelSelection, ProjectId, @@ -10,19 +11,23 @@ import { ThreadId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Ref from "effect/Ref"; import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as Stream from "effect/Stream"; import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite.ts"; import { runV2RecoveryPhase } from "../serverRuntimeStartup.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; +import { EventSinkV2 } from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; import { ProviderRuntimeRecoveryService } from "./ProviderRuntimeRecoveryService.ts"; import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; @@ -211,6 +216,188 @@ it.effect("retries a transient deferred apply read before recording its receipt" ), ); +it.effect("promotes queued work when terminal deferred apply fails", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const tempDir = yield* Effect.acquireRelease( + fs.makeTempDirectory({ prefix: "t3-deferred-organization-terminal-" }), + (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie), + ); + const databaseLayer = makeSqlitePersistenceLive(path.join(tempDir, "state.sqlite")).pipe( + Layer.provide(NodeServices.layer), + ); + const registryLayer = makeProviderAdapterRegistryLayer([adapter]); + const threadId = ThreadId.make("thread:deferred-organization-terminal"); + + const seeded = yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-organization-terminal:create"), + threadId, + projectId: ProjectId.make("project:deferred-organization-terminal"), + title: "Deferred organization terminal failure", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: tempDir, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-organization-terminal:active"), + threadId, + messageId: MessageId.make("message:deferred-organization-terminal:active"), + text: "Complete this run before applying organization.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const activeRun = (yield* orchestrator.getThreadProjection(threadId)).runs[0]; + assert.isDefined(activeRun); + yield* orchestrator.dispatch({ + type: "thread.organization.defer", + commandId: CommandId.make("command:deferred-organization-terminal:schedule"), + threadId, + runId: activeRun.id, + action: "settle", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-organization-terminal:queued"), + threadId, + messageId: MessageId.make("message:deferred-organization-terminal:queued"), + text: "Promote this run after the active run ends.", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + }); + const projection = yield* orchestrator.getThreadProjection(threadId); + const queuedRun = projection.runs.find((run) => run.status === "queued"); + assert.isDefined(queuedRun); + return { activeRun, queuedRun }; + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "deferred-organization-terminal:first-runtime", + runtimePolicyOverride: { cwd: tempDir }, + }, + registryLayer, + { databaseLayer, runEffectWorker: false }, + ), + ), + ), + ); + + const armed = yield* Ref.make(false); + const projectionReads = yield* Ref.make(0); + const decorateProjectionStore = (store: ProjectionStore.ProjectionStoreV2["Service"]) => + ProjectionStore.ProjectionStoreV2.of({ + ...store, + getThreadProjection: (requestedThreadId) => + Ref.get(armed).pipe( + Effect.flatMap((isArmed) => + isArmed + ? Ref.updateAndGet(projectionReads, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === 3 + ? Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ + threadId: requestedThreadId, + cause: "simulated terminal deferred apply failure", + }), + ) + : store.getThreadProjection(requestedThreadId), + ), + ) + : store.getThreadProjection(requestedThreadId), + ), + ), + }); + + yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSinkV2; + const receipts = yield* CommandReceiptStoreV2; + const afterSequence = yield* eventSink.latestSequence({ threadId }); + const promoted = yield* eventSink.stream({ threadId, afterSequence }).pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + stored.event.payload.id === seeded.queuedRun.id && + stored.event.payload.status === "starting", + ), + Stream.runHead, + Effect.forkChild, + ); + yield* Ref.set(armed, true); + const completedAt = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:deferred-organization-terminal:completed"), + type: "run.updated", + threadId, + runId: seeded.activeRun.id, + ...(seeded.activeRun.rootNodeId === null + ? {} + : { nodeId: seeded.activeRun.rootNodeId }), + providerInstanceId: seeded.activeRun.providerInstanceId, + occurredAt: completedAt, + payload: { + ...seeded.activeRun, + status: "completed", + completedAt, + }, + }, + ], + }); + assert.isTrue(Option.isSome(yield* Fiber.join(promoted))); + yield* Ref.set(armed, false); + + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + projection.runs.find((run) => run.id === seeded.queuedRun.id)?.status, + "starting", + ); + assert.equal(yield* Ref.get(projectionReads), 4); + assert.isTrue( + Option.isNone( + yield* receipts.getByCommandId( + CommandId.make( + `command:system:thread-organization-defer:${threadId}:${seeded.activeRun.id}`, + ), + ), + ), + ); + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "deferred-organization-terminal:second-runtime", + runtimePolicyOverride: { cwd: tempDir }, + }, + registryLayer, + { databaseLayer, runEffectWorker: false, decorateProjectionStore }, + ), + ), + ), + ); + }).pipe(Effect.provide(NodeServices.layer)), + ), +); + it.effect("discards a stale deferred organization intent after runtime restart", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index d4295aac9a6d..dde750194f19 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -7415,23 +7415,31 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId)); } if (stored.event.type === "run.updated") { + const runId = stored.event.payload.id; yield* threadDispatch.withLock( threadId, - finalizeDelegatedCompletionDelivery(threadId, stored.event.payload.id), + finalizeDelegatedCompletionDelivery(threadId, runId), ); - yield* threadDispatch.withLock( - threadId, - applyDeferredOrganization(threadId, stored.event.payload.id), + yield* threadDispatch.withLock(threadId, applyDeferredOrganization(threadId, runId)).pipe( + Effect.catch((cause) => + Effect.logWarning("Failed to apply deferred thread organization", { + threadId, + runId, + cause, + }), + ), ); } yield* threadDispatch.withLock(threadId, startNextQueuedRun(threadId)); }).pipe( Effect.catchCause((cause) => - Effect.logWarning("Failed to react to terminal V2 run", { - threadId: stored.event.threadId, - sequence: stored.sequence, - cause, - }), + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("Failed to react to terminal V2 run", { + threadId: stored.event.threadId, + sequence: stored.sequence, + cause, + }), ), ); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index b39c5a1b5d2f..981ac1b539c4 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -32,7 +32,7 @@ import { layer as effectWorkerLayer, runDaemon as runEffectWorkerDaemon, } from "../EffectWorker.ts"; -import { layerFromStores as eventSinkLayer } from "../EventSink.ts"; +import { EventSinkV2, layerFromStores as eventSinkLayer } from "../EventSink.ts"; import { layer as eventStoreLayer } from "../EventStore.ts"; import { layer as idAllocatorLayer } from "../IdAllocator.ts"; import { layer as orchestratorLayer } from "../Orchestrator.ts"; @@ -232,7 +232,7 @@ export function makeOrchestratorV2ProviderReplayLayer< readonly replayGate?: ProviderReplayGate; } = {}, ): Layer.Layer< - OrchestratorV2 | ProviderRuntimeRecoveryService | CommandReceiptStoreV2, + OrchestratorV2 | ProviderRuntimeRecoveryService | CommandReceiptStoreV2 | EventSinkV2, Error | MigrationError | PlatformError.PlatformError | SqlError > { const registryLayer = harness.makeProviderAdapterRegistryLayer( @@ -257,7 +257,7 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( ) => ProjectionStoreV2["Service"]; } = {}, ): Layer.Layer< - OrchestratorV2 | ProviderRuntimeRecoveryService | CommandReceiptStoreV2, + OrchestratorV2 | ProviderRuntimeRecoveryService | CommandReceiptStoreV2 | EventSinkV2, Error | MigrationError | PlatformError.PlatformError | SqlError > { const serverConfigLayer = Layer.effect( @@ -443,6 +443,7 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( orchestratorProvided, providerRuntimeRecoveryProvided, commandReceiptStoreProvided, + eventSinkProvided, ).pipe(Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer)); } const orchestratorWithWorker = Layer.effect( @@ -457,5 +458,6 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( orchestratorWithWorker, providerRuntimeRecoveryProvided, commandReceiptStoreProvided, + eventSinkProvided, ).pipe(Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer)); } diff --git a/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index 21ec7e75e214..34a8bb16fb6e 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -26,6 +26,7 @@ import { OrchestrationV2DomainEvent, OrchestrationV2ProviderThread, OrchestrationV2ProviderThreadJson, + OrchestrationV2RpcSchemas, OrchestrationV2ShellSnapshot, OrchestrationV2Subagent, OrchestrationV2ThreadProjection, @@ -43,6 +44,9 @@ const LegacyShellStreamItem = Schema.Union([ ]); const decodeLegacyShellStreamItem = Schema.decodeUnknownSync(LegacyShellStreamItem); const decodeOrchestrationV2Command = Schema.decodeUnknownSync(OrchestrationV2Command); +const decodeOrchestrationV2ClientCommand = Schema.decodeUnknownSync( + OrchestrationV2RpcSchemas.dispatchCommand.input, +); const decodeOrchestrationV2TurnItem = Schema.decodeUnknownSync(OrchestrationV2TurnItem); const decodeOrchestrationV2CheckpointScope = Schema.decodeUnknownSync( OrchestrationV2CheckpointScope, @@ -61,6 +65,25 @@ const decodeOrchestrationV2ProviderThread = Schema.decodeUnknownSync(Orchestrati const decodeOrchestrationV2ThreadShell = Schema.decodeUnknownSync(OrchestrationV2ThreadShell); describe("orchestration V2 contracts", () => { + it("keeps deferred organization apply internal to the server", () => { + const applyCommand = { + type: "thread.organization.defer.apply", + commandId: "command:deferred-organization-apply", + threadId: "thread:deferred-organization-apply", + runId: "run:deferred-organization-apply", + }; + + expect(decodeOrchestrationV2Command(applyCommand).type).toBe("thread.organization.defer.apply"); + expect(() => decodeOrchestrationV2ClientCommand(applyCommand)).toThrow(); + expect( + decodeOrchestrationV2ClientCommand({ + type: "thread.organization.defer.cancel", + commandId: "command:deferred-organization-cancel", + threadId: "thread:deferred-organization-apply", + }).type, + ).toBe("thread.organization.defer.cancel"); + }); + it("lets legacy snapshot decoders ignore enrichment metadata", () => { const decoded = decodeLegacyShellStreamItem({ kind: "snapshot", diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index a7e07ffaf0f1..a42f90adb263 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2393,6 +2393,19 @@ export const OrchestrationV2Command = Schema.Union([ ]); export type OrchestrationV2Command = typeof OrchestrationV2Command.Type; +export type OrchestrationV2ClientCommand = Exclude< + OrchestrationV2Command, + { readonly type: "thread.organization.defer.apply" } +>; + +export const OrchestrationV2ClientCommand = OrchestrationV2Command.pipe( + Schema.refine( + (command): command is OrchestrationV2ClientCommand => + command.type !== "thread.organization.defer.apply", + { expected: "a client-dispatchable orchestration command" }, + ), +); + export const ORCHESTRATION_V2_WS_METHODS = { dispatchCommand: "orchestration.dispatchCommand", getTurnDiff: "orchestration.getTurnDiff", @@ -2687,7 +2700,7 @@ export class OrchestrationGetWorkflowScriptError extends Schema.TaggedErrorClass export const OrchestrationV2RpcSchemas = { dispatchCommand: { - input: OrchestrationV2Command, + input: OrchestrationV2ClientCommand, output: OrchestrationV2DispatchCommandResult, }, getTurnDiff: {