diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 0f0c90bca934..d3435afe774f 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/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 new file mode 100644 index 000000000000..659294cdba68 --- /dev/null +++ b/apps/server/src/orchestration-v2/DeferredOrganizationRecovery.integration.test.ts @@ -0,0 +1,599 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + MessageId, + type ModelSelection, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + 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"; +import { makeLayer as makeProviderAdapterRegistryLayer } from "./ProviderAdapterRegistry.ts"; +import * as ProjectionStore from "./ProjectionStore.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("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("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("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* () { + 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 unreadableThreadId = ThreadId.make("thread:deferred-organization-recovery-unreadable"); + + const runIds = yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.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* Orchestrator.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* Orchestrator.OrchestratorV2; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:deferred-runtime-recovery:create"), + threadId, + projectId: ProjectId.make("project:deferred-organization-runtime-recovery"), + title: "Deferred organization runtime 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-runtime-recovery:active"), + threadId, + messageId: MessageId.make("message:deferred-runtime-recovery:active"), + text: "Settle only after this run completes.", + 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-runtime-recovery:schedule"), + threadId, + runId: activeRun.id, + action: "settle", + }); + return activeRun.id; + }).pipe(Effect.provide(runtimeLayer("deferred-runtime-recovery:first-runtime"))), + ); + + const recovered = yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const providerRuntimeRecovery = yield* ProviderRuntimeRecoveryService; + yield* runV2RecoveryPhase({ + recoverProviderRuntime: providerRuntimeRecovery.recover, + recoverDeferredOrganization: orchestrator.recoverDeferredOrganization, + }); + return yield* orchestrator.getThreadProjection(threadId); + }).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); + }).pipe(Effect.provide(NodeServices.layer)), + ), +); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 92d9213c3ec5..dde750194f19 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,30 @@ 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", { @@ -178,6 +204,7 @@ export interface OrchestratorV2DispatchResult { export interface OrchestratorV2Shape { readonly resumeQueuedRuns: Effect.Effect; + readonly recoverDeferredOrganization: Effect.Effect; readonly dispatch: ( command: OrchestrationV2Command, ) => Effect.Effect; @@ -257,6 +284,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": @@ -1432,7 +1462,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: movesForward ? { ...thread, lastVisitedAt: visitedAt.value } : thread, }); }); - const dispatchThreadMutation = Effect.fn("orchestrationV2.dispatch.threadMutation")(function* ( command: Extract< OrchestrationV2Command, @@ -1458,6 +1487,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( @@ -1758,6 +1788,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": @@ -1799,9 +1833,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) { @@ -1969,6 +2003,149 @@ 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>, + effects: 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; + } + 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, + { clearDeferredOrganization: true }, + ); + }, + ); + const dispatchProviderSessionDetach = Effect.fn("orchestrationV2.dispatch.providerSessionDetach")( function* ( command: Extract, @@ -6991,6 +7168,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, effects); + break; case "provider-session.detach": yield* dispatchProviderSessionDetach(command, events, effects); break; @@ -7115,7 +7297,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) @@ -7197,6 +7389,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; @@ -7211,19 +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, 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, + }), ), ); @@ -7249,6 +7465,48 @@ 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. + const recoverDeferredOrganization = projectionStore.getShellSnapshot().pipe( + Effect.flatMap((shell) => + Effect.forEach( + [...shell.threads, ...shell.archivedThreads].filter( + (thread) => thread.deferredOrganization != null, + ), + (thread) => + runDeferredOrganizationRepair( + thread.id, + 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.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: ThreadId.make("thread:shell"), + cause, + }), + ), + ); yield* resumeQueuedRuns.pipe( Effect.tap((resumed) => resumed === 0 @@ -7345,6 +7603,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return OrchestratorV2.of({ resumeQueuedRuns, + recoverDeferredOrganization, dispatch: dispatchWithReceipt, getThreadProjection: (threadId) => projectionStore @@ -7446,6 +7705,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/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..abebc90966eb 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -10,10 +10,12 @@ import { NodeId, RuntimeRequestId, TurnItemId, + type OrchestrationV2Run, type ModelSelection, ProjectId, ProviderDriverKind, ProviderInstanceId, + ProviderSessionId, ProviderThreadId, RunId, ThreadId, @@ -22,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"; @@ -55,7 +58,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, @@ -96,12 +100,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, @@ -128,7 +169,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()), @@ -868,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; @@ -1432,6 +1473,398 @@ 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; + const effectWorker = yield* OrchestrationEffectWorkerV2; + const providerSessions = yield* ProviderSessionManagerV2; + 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: workspacePath, + }); + 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); + const providerSessionId = ProviderSessionId.make( + "runtime-layer-deferred-settle-provider-session", + ); + yield* providerSessions.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy: { + cwd: workspacePath, + runtimeMode: "full-access", + interactionMode: "default", + }, + }); + 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); + 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}`), + threadId, + runId: run.id, + }; + const retry = yield* orchestrator.dispatch(applyCommand); + 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.", + }, + ], + ); + 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); + 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 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; + const effectWorker = yield* OrchestrationEffectWorkerV2; + const providerSessions = yield* ProviderSessionManagerV2; + + 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: workspacePath, + }); + 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"); + const providerSessionId = ProviderSessionId.make( + "runtime-layer-deferred-archive-provider-session", + ); + yield* providerSessions.open({ + threadId: archived.threadId, + providerSessionId, + modelSelection, + runtimePolicy: { + cwd: workspacePath, + runtimeMode: "full-access", + interactionMode: "default", + }, + }); + 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); + 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; + readonly status: string; + }>` + SELECT payload_json, status + 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" }, + ], + ); + 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({ + 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; readonly enableLegacyTokenStreaming?: boolean; readonly runEffectWorker?: boolean; + readonly decorateProjectionStore?: ( + service: ProjectionStoreV2["Service"], + ) => ProjectionStoreV2["Service"]; readonly replayGate?: ProviderReplayGate; } = {}, -): Layer.Layer { +): Layer.Layer< + OrchestratorV2 | ProviderRuntimeRecoveryService | CommandReceiptStoreV2 | EventSinkV2, + Error | MigrationError | PlatformError.PlatformError | SqlError +> { const registryLayer = harness.makeProviderAdapterRegistryLayer( scenario.transcript, options.replayGate === undefined ? {} : { replayGate: options.replayGate }, @@ -239,8 +252,14 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( >; readonly enableLegacyTokenStreaming?: boolean; readonly runEffectWorker?: boolean; + readonly decorateProjectionStore?: ( + service: ProjectionStoreV2["Service"], + ) => ProjectionStoreV2["Service"]; } = {}, -): Layer.Layer { +): Layer.Layer< + OrchestratorV2 | ProviderRuntimeRecoveryService | CommandReceiptStoreV2 | EventSinkV2, + Error | MigrationError | PlatformError.PlatformError | SqlError +> { const serverConfigLayer = Layer.effect( ServerConfig, makeReplayServerConfig(scenario.name).pipe(Effect.orDie), @@ -255,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)), ); @@ -396,18 +423,30 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( ), ), ); - const replayRuntime = Layer.merge(orchestratorProvided, effectWorkerProvided).pipe( - Layer.provide(worktreeRepairDependenciesTestLayer), - Layer.provide(NodeServices.layer), + const providerRuntimeRecoveryProvided = providerRuntimeRecoveryLayer.pipe( + Layer.provide( + Layer.mergeAll(effectWorkerProvided, storesLayer, eventSinkProvided, idAllocatorLayer), + ), + Layer.provide(serverSettingsLayer), ); + const replayRuntime = Layer.mergeAll( + orchestratorProvided, + effectWorkerProvided, + providerRuntimeRecoveryProvided, + ).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 orchestratorProvided.pipe(Layer.provide(NodeServices.layer)); + return Layer.mergeAll( + orchestratorProvided, + providerRuntimeRecoveryProvided, + commandReceiptStoreProvided, + eventSinkProvided, + ).pipe(Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer)); } - return Layer.effect( + const orchestratorWithWorker = Layer.effect( OrchestratorV2, Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; @@ -415,4 +454,10 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( return orchestrator; }), ).pipe(Layer.provide(replayRuntime)); + return Layer.mergeAll( + orchestratorWithWorker, + providerRuntimeRecoveryProvided, + commandReceiptStoreProvided, + eventSinkProvided, + ).pipe(Layer.provide(worktreeRepairDependenciesTestLayer), Layer.provide(NodeServices.layer)); } 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/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index cc332eb6a85d..da8d9b0da56c 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -325,6 +325,21 @@ cannot settle. A pending approval or user-input request also blocks archive through MCP. Every successful outcome returns the resulting durable organization state. +### `t3_thread_defer_organization` + +Stores an organization intent for the calling thread and binds it to the +calling run. The supported deferred actions are settle and archive. The +terminal-run handler applies the intent only after that exact run completes +successfully and the serialized decision sees no newer run, queued or active +work, pending runtime request, or title regeneration. Otherwise the intent is +cleared without changing settlement or archive state. + +`operation: "read"` returns the current durable intent, and +`operation: "cancel"` clears it. Schedule, cancel, and the terminal application +all use command receipts. Startup recovery examines persisted intents so a +terminal or stale intent is applied or discarded after a server restart; it +does not require a polling daemon. + ### `t3_thread_delete` Permanently deletes one thread in the calling project and defaults to the diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 65c9c7e977e6..902145414b00 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -42,6 +42,12 @@ settle, archive, or change read state in the current project. T3 Code applies th same rules as the sidebar. For example, an agent cannot settle a thread while work 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 that creates a +newer run, or an unresolved request, keeps the thread visible and discards the stale +organization request. + ## Settle finished work Choose **Settle thread** from its menu to move finished work out of the active list 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 68e1ac01a318..a42f90adb263 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -296,6 +296,13 @@ export const OrchestrationV2ProviderCapabilities = Schema.Struct({ }); export type OrchestrationV2ProviderCapabilities = typeof OrchestrationV2ProviderCapabilities.Type; +export const OrchestrationV2DeferredOrganization = Schema.Struct({ + runId: RunId, + action: Schema.Literals(["settle", "archive"]), + requestedAt: Schema.DateTimeUtc, +}); +export type OrchestrationV2DeferredOrganization = typeof OrchestrationV2DeferredOrganization.Type; + export const OrchestrationV2AppThread = Schema.Struct({ ...OrchestrationV2CreationFields, id: ThreadId, @@ -343,6 +350,7 @@ export const OrchestrationV2AppThread = Schema.Struct({ lastVisitedAt: Schema.NullOr(Schema.DateTimeUtc).pipe( Schema.withDecodingDefault(Effect.succeed(null)), ), + deferredOrganization: Schema.optional(Schema.NullOr(OrchestrationV2DeferredOrganization)), /** In-flight title regeneration marker; cleared when a new title lands. */ titleRegeneration: Schema.optional( Schema.NullOr( @@ -1371,6 +1379,7 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ * back to their local visited state when the field is absent. */ lastVisitedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), + deferredOrganization: Schema.optional(Schema.NullOr(OrchestrationV2DeferredOrganization)), /** In-flight title regeneration marker; null/absent when no request is pending. */ titleRegeneration: Schema.optional( Schema.NullOr( @@ -1459,6 +1468,14 @@ export const OrchestrationV2AppThreadJson = OrchestrationV2AppThread.mapFields(( lastVisitedAt: Schema.NullOr(Schema.DateTimeUtcFromString).pipe( Schema.withDecodingDefault(Effect.succeed(null)), ), + deferredOrganization: Schema.optional( + Schema.NullOr( + OrchestrationV2DeferredOrganization.mapFields((deferredFields) => ({ + ...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, @@ -2350,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", @@ -2644,7 +2700,7 @@ export class OrchestrationGetWorkflowScriptError extends Schema.TaggedErrorClass export const OrchestrationV2RpcSchemas = { dispatchCommand: { - input: OrchestrationV2Command, + input: OrchestrationV2ClientCommand, output: OrchestrationV2DispatchCommandResult, }, getTurnDiff: { diff --git a/packages/contracts/src/orchestratorMcp.test.ts b/packages/contracts/src/orchestratorMcp.test.ts index dc8a88a9faab..339795885afa 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,31 @@ 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(); + 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 524684af1713..924ae6f437c4 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -530,6 +530,44 @@ 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.Struct({ + operation: Schema.Literals(["read", "schedule", "cancel"]).annotate({ + description: + "Read the current intent, schedule settle/archive after this run, or cancel the intent.", + }), + action: Schema.optional( + Schema.Literals(["settle", "archive"]).annotate({ + description: + "Required for schedule only. Applies after this calling run completes safely with no newer, queued, or blocked work.", + }), + ), + 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; + +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" },