From 225ba2d994ea5bedeb5cc74922731afdca96fdb5 Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Fri, 10 Jul 2026 19:26:21 -0700 Subject: [PATCH] feat(orchestrator): archive and delete owned subagent subtrees Applying archive, unarchive, or delete to a thread now cascades through the subagent threads it recursively owns: the orchestrator walks the owned subtree (OwnedSubagentTree), applies the lifecycle transition to every member with a single timestamp, and cancels their active runs and pending attempts. Half-applied cascades are repaired on startup (OwnedSubagentLifecycleRepair), and checkpoints orphaned by deleted threads are cleaned up (CheckpointCleanupService). --- .../CheckpointCleanupService.test.ts | 82 ++ .../CheckpointCleanupService.ts | 70 ++ .../src/orchestration-v2/CheckpointService.ts | 58 ++ .../src/orchestration-v2/EffectOutbox.ts | 4 + .../src/orchestration-v2/EffectWorker.test.ts | 5 + .../src/orchestration-v2/EffectWorker.ts | 18 +- apps/server/src/orchestration-v2/EventSink.ts | 99 +- .../FoundationPersistence.test.ts | 562 +++++++++++ .../src/orchestration-v2/Orchestrator.ts | 891 ++++++++++++++---- .../OwnedSubagentLifecycleRepair.test.ts | 93 ++ .../OwnedSubagentLifecycleRepair.ts | 60 ++ .../OwnedSubagentTree.test.ts | 98 ++ .../src/orchestration-v2/OwnedSubagentTree.ts | 46 + .../orchestration-v2/ProjectionStore.test.ts | 57 ++ .../src/orchestration-v2/ProjectionStore.ts | 37 + .../orchestration-v2/ProviderEventIngestor.ts | 5 +- .../ProviderTurnControlService.test.ts | 1 + .../ProviderTurnStartService.test.ts | 350 +++++++ .../ProviderTurnStartService.ts | 89 +- .../RunExecutionService.test.ts | 79 ++ .../orchestration-v2/RunExecutionService.ts | 27 +- .../src/orchestration-v2/runtimeLayer.test.ts | 285 ++++++ .../src/orchestration-v2/runtimeLayer.ts | 5 + .../testkit/ProviderReplayHarness.ts | 5 + 24 files changed, 2816 insertions(+), 210 deletions(-) create mode 100644 apps/server/src/orchestration-v2/CheckpointCleanupService.test.ts create mode 100644 apps/server/src/orchestration-v2/CheckpointCleanupService.ts create mode 100644 apps/server/src/orchestration-v2/OwnedSubagentLifecycleRepair.test.ts create mode 100644 apps/server/src/orchestration-v2/OwnedSubagentLifecycleRepair.ts create mode 100644 apps/server/src/orchestration-v2/OwnedSubagentTree.test.ts create mode 100644 apps/server/src/orchestration-v2/OwnedSubagentTree.ts create mode 100644 apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts diff --git a/apps/server/src/orchestration-v2/CheckpointCleanupService.test.ts b/apps/server/src/orchestration-v2/CheckpointCleanupService.test.ts new file mode 100644 index 00000000000..942cc69cbd4 --- /dev/null +++ b/apps/server/src/orchestration-v2/CheckpointCleanupService.test.ts @@ -0,0 +1,82 @@ +import { assert, it } from "@effect/vitest"; +import { + CheckpointId, + CheckpointRef, + CheckpointScopeId, + ThreadId, + type OrchestrationV2Checkpoint, + type OrchestrationV2CheckpointScope, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { CheckpointCleanupServiceV2, layer } from "./CheckpointCleanupService.ts"; +import { CheckpointServiceV2 } from "./CheckpointService.ts"; +import { ProjectionStoreV2 } from "./ProjectionStore.ts"; + +it.effect("cleans every checkpoint scope through the latest known ordinal", () => { + const threadId = ThreadId.make("thread_checkpoint_cleanup"); + const firstScope = { + id: CheckpointScopeId.make("scope_first"), + threadId, + } as OrchestrationV2CheckpointScope; + const secondScope = { + id: CheckpointScopeId.make("scope_second"), + threadId, + } as OrchestrationV2CheckpointScope; + const checkpoint = { + id: CheckpointId.make("checkpoint_first"), + scopeId: firstScope.id, + ordinalWithinScope: 2, + ref: CheckpointRef.make("refs/t3/test/checkpoint-first"), + } as OrchestrationV2Checkpoint; + const projection = { + thread: { id: threadId }, + runs: [{ ordinal: 4 }], + checkpointScopes: [firstScope, secondScope], + checkpoints: [checkpoint], + } as unknown as OrchestrationV2ThreadProjection; + const calls: Array<{ + readonly scopeId: CheckpointScopeId; + readonly checkpointIds: ReadonlyArray; + readonly throughOrdinal: number; + }> = []; + const testLayer = layer.pipe( + Layer.provide( + Layer.merge( + Layer.mock(ProjectionStoreV2)({ + getThreadProjection: () => Effect.succeed(projection), + }), + Layer.mock(CheckpointServiceV2)({ + deleteScopeRefs: (input) => { + calls.push({ + scopeId: input.scope.id, + checkpointIds: input.checkpoints.map((item) => item.id), + throughOrdinal: input.throughOrdinal, + }); + return Effect.void; + }, + }), + ), + ), + ); + + return Effect.gen(function* () { + const cleanup = yield* CheckpointCleanupServiceV2; + yield* cleanup.cleanup(threadId); + + assert.deepEqual(calls, [ + { + scopeId: firstScope.id, + checkpointIds: [checkpoint.id], + throughOrdinal: 4, + }, + { + scopeId: secondScope.id, + checkpointIds: [], + throughOrdinal: 4, + }, + ]); + }).pipe(Effect.provide(testLayer)); +}); diff --git a/apps/server/src/orchestration-v2/CheckpointCleanupService.ts b/apps/server/src/orchestration-v2/CheckpointCleanupService.ts new file mode 100644 index 00000000000..9de8adf8ccb --- /dev/null +++ b/apps/server/src/orchestration-v2/CheckpointCleanupService.ts @@ -0,0 +1,70 @@ +import { ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; + +import { CheckpointServiceV2 } from "./CheckpointService.ts"; +import { ProjectionStoreV2 } from "./ProjectionStore.ts"; + +export class CheckpointCleanupError extends Schema.TaggedErrorClass()( + "CheckpointCleanupError", + { + threadId: ThreadId, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to clean checkpoint refs for thread ${this.threadId}.`; + } +} + +export class CheckpointCleanupServiceV2 extends Context.Service< + CheckpointCleanupServiceV2, + { + readonly cleanup: (threadId: ThreadId) => Effect.Effect; + } +>()("t3/orchestration-v2/CheckpointCleanupService/CheckpointCleanupServiceV2") {} + +export const layer: Layer.Layer< + CheckpointCleanupServiceV2, + never, + CheckpointServiceV2 | ProjectionStoreV2 +> = Layer.effect( + CheckpointCleanupServiceV2, + Effect.gen(function* () { + const checkpoints = yield* CheckpointServiceV2; + const projections = yield* ProjectionStoreV2; + + const cleanup = Effect.fn("orchestrationV2.checkpointCleanup.cleanup")(function* ( + threadId: ThreadId, + ) { + const projection = yield* projections.getThreadProjection(threadId); + const throughOrdinal = Math.max( + 0, + ...projection.runs.map((run) => run.ordinal), + ...projection.checkpoints.map((checkpoint) => checkpoint.ordinalWithinScope), + ); + + yield* Effect.forEach( + projection.checkpointScopes, + (scope) => + checkpoints.deleteScopeRefs({ + scope, + checkpoints: projection.checkpoints.filter( + (checkpoint) => checkpoint.scopeId === scope.id, + ), + throughOrdinal, + }), + { concurrency: 1, discard: true }, + ); + }); + + return CheckpointCleanupServiceV2.of({ + cleanup: (threadId) => + cleanup(threadId).pipe( + Effect.mapError((cause) => new CheckpointCleanupError({ threadId, cause })), + ), + }); + }), +); diff --git a/apps/server/src/orchestration-v2/CheckpointService.ts b/apps/server/src/orchestration-v2/CheckpointService.ts index e11caa21431..f5f06433843 100644 --- a/apps/server/src/orchestration-v2/CheckpointService.ts +++ b/apps/server/src/orchestration-v2/CheckpointService.ts @@ -103,6 +103,18 @@ export class CheckpointDeleteStaleRefsError extends Schema.TaggedErrorClass()( + "CheckpointDeleteScopeRefsError", + { + scopeId: CheckpointScopeId, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to delete checkpoint refs for scope ${this.scopeId}.`; + } +} + export const CheckpointServiceV2Error = Schema.Union([ CheckpointRootScopePrepareError, CheckpointScopeEnsureError, @@ -110,6 +122,7 @@ export const CheckpointServiceV2Error = Schema.Union([ CheckpointCaptureError, CheckpointRestoreError, CheckpointDeleteStaleRefsError, + CheckpointDeleteScopeRefsError, ]); export type CheckpointServiceV2Error = typeof CheckpointServiceV2Error.Type; @@ -150,6 +163,16 @@ export interface CheckpointServiceV2Shape { readonly scope: OrchestrationV2CheckpointScope; readonly checkpoints: ReadonlyArray; }) => Effect.Effect; + readonly deleteScopeRefs: (input: { + readonly scope: OrchestrationV2CheckpointScope; + readonly checkpoints: ReadonlyArray; + /** + * Baselines can be captured before their projection event is committed. + * Include every deterministic ordinal through this value so deletion also + * cleans refs left by an interrupted capture. + */ + readonly throughOrdinal: number; + }) => Effect.Effect; } export class CheckpointServiceV2 extends Context.Service< @@ -541,6 +564,40 @@ export const layer: Layer.Layer< ), ); + const deleteScopeRefs: CheckpointServiceV2Shape["deleteScopeRefs"] = (input) => { + const checkpointRefs = new Set( + input.checkpoints.map((checkpoint) => checkpoint.ref), + ); + for ( + let ordinalWithinScope = 0; + ordinalWithinScope <= input.throughOrdinal; + ordinalWithinScope += 1 + ) { + checkpointRefs.add( + checkpointRefForScopeOrdinal({ + scopeId: input.scope.id, + ordinalWithinScope, + }), + ); + } + + return withWorkspaceLock( + input.scope.cwd, + checkpointStore.deleteCheckpointRefs({ + cwd: input.scope.cwd, + checkpointRefs: Array.from(checkpointRefs), + }), + ).pipe( + Effect.mapError( + (cause) => + new CheckpointDeleteScopeRefsError({ + scopeId: input.scope.id, + cause, + }), + ), + ); + }; + return CheckpointServiceV2.of({ prepareRootRunScope: (input) => makeRootRunScope({ ...input, idAllocator }).pipe( @@ -559,6 +616,7 @@ export const layer: Layer.Layer< capture, restore, deleteStaleRefs, + deleteScopeRefs, } satisfies CheckpointServiceV2Shape); }), ); diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 75ead585167..cd922809bfb 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -81,6 +81,9 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ runId: RunId, scopeId: CheckpointScopeId, }), + Schema.Struct({ + type: Schema.Literal("checkpoint.cleanup"), + }), Schema.Struct({ type: Schema.Literal("terminal.cleanup"), }), @@ -95,6 +98,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ "provider-session.detach", "provider-thread.rollback", "checkpoint.capture", + "checkpoint.cleanup", "terminal.cleanup", "attachment.cleanup", ] as const satisfies ReadonlyArray; diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 5b0e4d26203..45aa1ed24e4 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -15,6 +15,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import { CheckpointCleanupServiceV2 } from "./CheckpointCleanupService.ts"; import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts"; import type { OrchestrationEffectV2 } from "./EffectOutbox.ts"; import { executorLayer, OrchestrationEffectExecutorV2 } from "./EffectWorker.ts"; @@ -125,6 +126,10 @@ function makeExecutorLayer(input: { CheckpointRollbackServiceV2, CheckpointRollbackServiceV2.of({ execute: () => Effect.void }), ), + Layer.succeed( + CheckpointCleanupServiceV2, + CheckpointCleanupServiceV2.of({ cleanup: () => Effect.void }), + ), Layer.succeed( RuntimeRequestServiceV2, RuntimeRequestServiceV2.of({ respond: () => Effect.void }), diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 612c26c44cf..097981c7fa5 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -7,10 +7,11 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import { CheckpointCleanupServiceV2 } from "./CheckpointCleanupService.ts"; +import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts"; +import { EffectOutboxV2, type OrchestrationEffectV2 } from "./EffectOutbox.ts"; import { RunFinalizationService } from "./RunFinalizationService.ts"; import { ResourceCleanupService } from "./ResourceCleanupService.ts"; -import { EffectOutboxV2, type OrchestrationEffectV2 } from "./EffectOutbox.ts"; -import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts"; import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { ProviderTurnControlServiceV2 } from "./ProviderTurnControlService.ts"; import { ProviderTurnStartServiceV2 } from "./ProviderTurnStartService.ts"; @@ -41,6 +42,7 @@ export const executorLayer: Layer.Layer< never, | ProviderSessionManagerV2 | RunFinalizationService + | CheckpointCleanupServiceV2 | CheckpointRollbackServiceV2 | ProviderTurnControlServiceV2 | ProviderTurnStartServiceV2 @@ -50,6 +52,7 @@ export const executorLayer: Layer.Layer< Effect.gen(function* () { const runFinalization = yield* RunFinalizationService; const resourceCleanup = yield* ResourceCleanupService; + const checkpointCleanup = yield* CheckpointCleanupServiceV2; const checkpointRollback = yield* CheckpointRollbackServiceV2; const providerSessions = yield* ProviderSessionManagerV2; const providerTurnControl = yield* ProviderTurnControlServiceV2; @@ -229,6 +232,17 @@ export const executorLayer: Layer.Layer< }), ), ); + case "checkpoint.cleanup": + return checkpointCleanup.cleanup(effect.threadId).pipe( + Effect.mapError( + (cause) => + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: effect.request.type, + cause, + }), + ), + ); case "terminal.cleanup": return resourceCleanup.cleanupTerminals(effect.threadId).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 5e226553e73..e02fa79880c 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -90,7 +90,16 @@ export interface EventSinkV2Shape { readonly threadId: ThreadId; readonly runId: RunId; readonly activeAttemptId: RunAttemptId; - readonly expectedStatus: OrchestrationV2Run["status"]; + readonly expectedStatus: + | OrchestrationV2Run["status"] + | ReadonlyArray; + /** + * Admit only terminal provider-turn or interrupt-result artifacts for this + * exact attempt after a restart has superseded it. The transactional + * thread lifecycle guard is still required, so archival and deletion + * always win. + */ + readonly allowSupersededAttemptTerminalArtifacts?: boolean; readonly events: ReadonlyArray; }) => Effect.Effect< { @@ -107,6 +116,7 @@ export interface EventSinkV2Shape { readonly events: ReadonlyArray; readonly effects: ReadonlyArray; readonly cancelUnsettledEffects?: { + readonly threadIds?: ReadonlyArray; readonly effectTypes: ReadonlyArray; readonly reason: string; }; @@ -258,23 +268,82 @@ const baseLayer: Layer.Layer< const result = yield* sql.withTransaction( Effect.gen(function* () { + const createdThreadIds = new Set( + input.events.flatMap((event) => + event.type === "thread.created" ? [event.threadId] : [], + ), + ); + const targetThreadIds = Array.from( + new Set([ + input.threadId, + ...input.events.flatMap((event) => + createdThreadIds.has(event.threadId) ? [] : [event.threadId], + ), + ]), + ); + const activeTargetThreads = yield* sql<{ readonly thread_id: string }>` + SELECT thread_id + FROM orchestration_v2_projection_threads + WHERE thread_id IN ${sql.in(targetThreadIds)} + AND archived_at IS NULL + AND deleted_at IS NULL + `; + if (activeTargetThreads.length !== targetThreadIds.length) { + return { + committed: false as const, + storedEvents: [] as ReadonlyArray, + }; + } const rows = yield* sql<{ readonly status: string; readonly active_attempt_id: string | null; + readonly supplied_attempt_status: string | null; }>` SELECT - status, - json_extract(payload_json, '$.activeAttemptId') AS active_attempt_id - FROM orchestration_v2_projection_runs - WHERE run_id = ${input.runId} - AND thread_id = ${input.threadId} + runs.status, + json_extract(runs.payload_json, '$.activeAttemptId') AS active_attempt_id, + supplied_attempt.status AS supplied_attempt_status + FROM orchestration_v2_projection_runs AS runs + INNER JOIN orchestration_v2_projection_threads AS threads + ON threads.thread_id = runs.thread_id + LEFT JOIN orchestration_v2_projection_run_attempts AS supplied_attempt + ON supplied_attempt.attempt_id = ${input.activeAttemptId} + AND supplied_attempt.run_id = runs.run_id + AND supplied_attempt.thread_id = runs.thread_id + WHERE runs.run_id = ${input.runId} + AND runs.thread_id = ${input.threadId} + AND threads.archived_at IS NULL + AND threads.deleted_at IS NULL LIMIT 1 `; const current = rows[0]; + const expectedStatuses = Array.isArray(input.expectedStatus) + ? input.expectedStatus + : [input.expectedStatus]; + const currentRunOwnsAttempt = + current !== undefined && + expectedStatuses.includes(current.status as OrchestrationV2Run["status"]) && + current.active_attempt_id === input.activeAttemptId; + const isSupersededAttemptTerminalArtifactWrite = + input.allowSupersededAttemptTerminalArtifacts === true && + current?.supplied_attempt_status === "superseded" && + input.events.length > 0 && + input.events.every( + (event) => + (event.type === "provider-turn.updated" && + event.payload.runAttemptId === input.activeAttemptId && + (event.payload.status === "completed" || + event.payload.status === "interrupted" || + event.payload.status === "failed" || + event.payload.status === "cancelled")) || + (event.type === "turn-item.updated" && + event.runId === input.runId && + event.payload.runId === input.runId && + event.payload.type === "run_interrupt_result"), + ); if ( current === undefined || - current.status !== input.expectedStatus || - current.active_attempt_id !== input.activeAttemptId + (!currentRunOwnsAttempt && !isSupersededAttemptTerminalArtifactWrite) ) { return { committed: false as const, @@ -359,10 +428,16 @@ const baseLayer: Layer.Layer< const cancelledEffectIds = input.cancelUnsettledEffects === undefined ? [] - : yield* effectOutbox.cancelUnsettled({ - threadId: input.threadId, - ...input.cancelUnsettledEffects, - }); + : yield* Effect.forEach( + Array.from(new Set(input.cancelUnsettledEffects.threadIds ?? [input.threadId])), + (threadId) => + effectOutbox.cancelUnsettled({ + threadId, + effectTypes: input.cancelUnsettledEffects!.effectTypes, + reason: input.cancelUnsettledEffects!.reason, + }), + { concurrency: 1 }, + ).pipe(Effect.map((effectIds) => effectIds.flat())); return { receipt, storedEvents, committed: true as const, cancelledEffectIds }; }), ); diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 48d8e460487..6e0be02573f 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -17,6 +17,7 @@ import { ProviderInstanceId, ProviderSessionId, ProviderThreadId, + ProviderTurnId, RunAttemptId, RunId, ThreadId, @@ -683,6 +684,567 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { }), ); + it.effect("admits only terminal provider updates from an exact superseded attempt", () => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const projectionStore = yield* ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:foundation-superseded-terminal-guard"); + const runId = RunId.make("run:foundation-superseded-terminal-guard"); + const oldAttemptId = RunAttemptId.make( + "run-attempt:foundation-superseded-terminal-guard:old", + ); + const replacementAttemptId = RunAttemptId.make( + "run-attempt:foundation-superseded-terminal-guard:replacement", + ); + const rootNodeId = NodeId.make("node:foundation-superseded-terminal-guard"); + const providerThreadId = ProviderThreadId.make( + "provider-thread:foundation-superseded-terminal-guard", + ); + const providerTurnId = ProviderTurnId.make( + "provider-turn:foundation-superseded-terminal-guard", + ); + const thread = makeThread(threadId, now); + const runningRun: OrchestrationV2Run = { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId, + userMessageId: MessageId.make("message:foundation-superseded-terminal-guard"), + rootNodeId, + activeAttemptId: oldAttemptId, + status: "running", + queuePosition: null, + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + const oldAttempt = { + id: oldAttemptId, + runId, + attemptOrdinal: 1, + rootNodeId, + providerInstanceId, + providerThreadId, + providerTurnId, + reason: "initial" as const, + status: "running" as const, + startedAt: now, + completedAt: null, + }; + const providerThread = { + id: providerThreadId, + driver: providerDriver, + providerInstanceId, + providerSessionId: ProviderSessionId.make( + "provider-session:foundation-superseded-terminal-guard", + ), + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "active" as const, + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }; + const runningProviderTurn = { + id: providerTurnId, + providerThreadId, + nodeId: rootNodeId, + runAttemptId: oldAttemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running" as const, + startedAt: now, + completedAt: null, + }; + yield* eventSink.write({ + events: [ + threadCreatedEvent({ + id: "event:foundation-superseded-terminal-guard:thread", + thread, + now, + }), + { + id: EventId.make("event:foundation-superseded-terminal-guard:run"), + type: "run.created", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: runningRun, + }, + { + id: EventId.make("event:foundation-superseded-terminal-guard:old-attempt"), + type: "run-attempt.created", + threadId, + runId, + nodeId: rootNodeId, + providerInstanceId, + occurredAt: now, + payload: oldAttempt, + }, + { + id: EventId.make("event:foundation-superseded-terminal-guard:provider-thread"), + type: "provider-thread.updated", + threadId, + runId, + nodeId: rootNodeId, + driver: providerDriver, + providerInstanceId, + occurredAt: now, + payload: providerThread, + }, + { + id: EventId.make("event:foundation-superseded-terminal-guard:provider-turn"), + type: "provider-turn.updated", + threadId, + runId, + nodeId: rootNodeId, + driver: providerDriver, + providerInstanceId, + occurredAt: now, + payload: runningProviderTurn, + }, + ], + }); + + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:foundation-superseded-terminal-guard:replacement-run"), + type: "run.updated", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: { + ...runningRun, + activeAttemptId: replacementAttemptId, + status: "starting", + }, + }, + { + id: EventId.make("event:foundation-superseded-terminal-guard:superseded-attempt"), + type: "run-attempt.updated", + threadId, + runId, + nodeId: rootNodeId, + providerInstanceId, + occurredAt: now, + payload: { ...oldAttempt, status: "superseded", completedAt: now }, + }, + { + id: EventId.make("event:foundation-superseded-terminal-guard:new-attempt"), + type: "run-attempt.created", + threadId, + runId, + nodeId: rootNodeId, + providerInstanceId, + occurredAt: now, + payload: { + ...oldAttempt, + id: replacementAttemptId, + attemptOrdinal: 2, + providerTurnId: null, + status: "pending", + startedAt: null, + completedAt: null, + }, + }, + ], + }); + + const nonterminal = yield* eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: oldAttemptId, + expectedStatus: ["running", "waiting"], + allowSupersededAttemptTerminalArtifacts: true, + events: [ + { + id: EventId.make("event:foundation-superseded-terminal-guard:late-running"), + type: "provider-turn.updated", + threadId, + runId, + nodeId: rootNodeId, + driver: providerDriver, + providerInstanceId, + occurredAt: now, + payload: runningProviderTurn, + }, + ], + }); + assert.isFalse(nonterminal.committed); + + const terminal = yield* eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: oldAttemptId, + expectedStatus: ["running", "waiting"], + allowSupersededAttemptTerminalArtifacts: true, + events: [ + { + id: EventId.make("event:foundation-superseded-terminal-guard:interrupted"), + type: "provider-turn.updated", + threadId, + runId, + nodeId: rootNodeId, + driver: providerDriver, + providerInstanceId, + occurredAt: now, + payload: { + ...runningProviderTurn, + status: "interrupted", + completedAt: now, + }, + }, + ], + }); + assert.isTrue(terminal.committed); + assert.equal( + (yield* projectionStore.getThreadProjection(threadId)).providerTurns[0]?.status, + "interrupted", + ); + const interruptArtifact = { + id: TurnItemId.make("turn-item:foundation-superseded-terminal-guard:interrupt"), + threadId, + runId, + nodeId: rootNodeId, + providerThreadId, + providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: 198, + status: "interrupted" as const, + title: "Interrupted", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "run_interrupt_result" as const, + message: "Run interrupted by restart", + }; + const activeArtifact = yield* eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: oldAttemptId, + expectedStatus: ["running", "waiting"], + allowSupersededAttemptTerminalArtifacts: true, + events: [ + { + id: EventId.make("event:foundation-superseded-terminal-guard:interrupt-artifact"), + type: "turn-item.updated", + threadId, + runId, + nodeId: rootNodeId, + providerInstanceId, + occurredAt: now, + payload: interruptArtifact, + }, + ], + }); + assert.isTrue(activeArtifact.committed); + + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:foundation-superseded-terminal-guard:archive"), + type: "thread.archived", + threadId, + providerInstanceId, + occurredAt: now, + payload: { ...thread, archivedAt: now, updatedAt: now }, + }, + ], + }); + const inactiveArtifact = yield* eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: oldAttemptId, + expectedStatus: ["running", "waiting"], + allowSupersededAttemptTerminalArtifacts: true, + events: [ + { + id: EventId.make("event:foundation-superseded-terminal-guard:late-interrupt-artifact"), + type: "turn-item.updated", + threadId, + runId, + nodeId: rootNodeId, + providerInstanceId, + occurredAt: now, + payload: { + ...interruptArtifact, + id: TurnItemId.make("turn-item:foundation-superseded-terminal-guard:late-interrupt"), + ordinal: 199, + }, + }, + ], + }); + assert.isFalse(inactiveArtifact.committed); + + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:foundation-superseded-terminal-guard:cleanup-run"), + type: "run.updated", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: { + ...runningRun, + activeAttemptId: replacementAttemptId, + status: "cancelled", + completedAt: now, + }, + }, + { + id: EventId.make("event:foundation-superseded-terminal-guard:cleanup-attempt"), + type: "run-attempt.updated", + threadId, + runId, + nodeId: rootNodeId, + providerInstanceId, + occurredAt: now, + payload: { + ...oldAttempt, + id: replacementAttemptId, + attemptOrdinal: 2, + providerTurnId: null, + status: "cancelled", + startedAt: null, + completedAt: now, + }, + }, + { + id: EventId.make("event:foundation-superseded-terminal-guard:cleanup-thread"), + type: "provider-thread.updated", + threadId, + runId, + nodeId: rootNodeId, + driver: providerDriver, + providerInstanceId, + occurredAt: now, + payload: { ...providerThread, status: "idle", updatedAt: now }, + }, + ], + }); + }), + ); + + it.effect("does not publish run events after the owning thread is archived", () => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:foundation-archived-run-guard"); + const runId = RunId.make("run:foundation-archived-run-guard"); + const attemptId = RunAttemptId.make("run-attempt:foundation-archived-run-guard"); + const thread = makeThread(threadId, now); + const startingRun: OrchestrationV2Run = { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:foundation-archived-run-guard"), + rootNodeId: null, + activeAttemptId: attemptId, + status: "starting", + queuePosition: null, + requestedAt: now, + startedAt: null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + yield* eventSink.write({ + events: [ + threadCreatedEvent({ id: "event:foundation-archived-run-guard:thread", thread, now }), + { + id: EventId.make("event:foundation-archived-run-guard:run"), + type: "run.created", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: startingRun, + }, + ], + }); + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:foundation-archived-run-guard:archived"), + type: "thread.archived", + threadId, + providerInstanceId, + occurredAt: now, + payload: { ...thread, archivedAt: now, updatedAt: now }, + }, + ], + }); + + const result = yield* eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: attemptId, + expectedStatus: ["starting", "waiting"], + events: [ + { + id: EventId.make("event:foundation-archived-run-guard:late-running"), + type: "run.updated", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: { ...startingRun, status: "running", startedAt: now }, + }, + ], + }); + + assert.isFalse(result.committed); + assert.deepEqual(result.storedEvents, []); + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:foundation-archived-run-guard:cleanup"), + type: "run.updated", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: { ...startingRun, status: "cancelled", completedAt: now }, + }, + ], + }); + }), + ); + + it.effect("does not publish a routed event into an archived child thread", () => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const now = yield* DateTime.now; + const rootThreadId = ThreadId.make("thread:foundation-archived-target-guard:root"); + const childThreadId = ThreadId.make("thread:foundation-archived-target-guard:child"); + const runId = RunId.make("run:foundation-archived-target-guard"); + const attemptId = RunAttemptId.make("run-attempt:foundation-archived-target-guard"); + const rootNodeId = NodeId.make("node:foundation-archived-target-guard"); + const rootThread = makeThread(rootThreadId, now); + const childThread = { + ...makeThread(childThreadId, now), + lineage: { + parentThreadId: rootThreadId, + relationshipToParent: "subagent" as const, + rootThreadId, + }, + }; + const runningRun: OrchestrationV2Run = { + id: runId, + threadId: rootThreadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:foundation-archived-target-guard:user"), + rootNodeId, + activeAttemptId: attemptId, + status: "running", + queuePosition: null, + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + yield* eventSink.write({ + events: [ + threadCreatedEvent({ + id: "event:foundation-archived-target-guard:root-thread", + thread: rootThread, + now, + }), + threadCreatedEvent({ + id: "event:foundation-archived-target-guard:child-thread", + thread: childThread, + now, + }), + { + id: EventId.make("event:foundation-archived-target-guard:run"), + type: "run.created", + threadId: rootThreadId, + runId, + providerInstanceId, + occurredAt: now, + payload: runningRun, + }, + { + id: EventId.make("event:foundation-archived-target-guard:archive-child"), + type: "thread.archived", + threadId: childThreadId, + providerInstanceId, + occurredAt: now, + payload: { ...childThread, archivedAt: now, updatedAt: now }, + }, + ], + }); + + const result = yield* eventSink.writeIfRunCurrent({ + threadId: rootThreadId, + runId, + activeAttemptId: attemptId, + expectedStatus: ["running", "waiting"], + events: [ + { + id: EventId.make("event:foundation-archived-target-guard:late-child-message"), + type: "message.updated", + threadId: childThreadId, + providerInstanceId, + occurredAt: now, + payload: { + createdBy: "agent", + creationSource: "provider", + id: MessageId.make("message:foundation-archived-target-guard:late-child"), + threadId: childThreadId, + runId: null, + nodeId: null, + role: "assistant", + text: "This late event must not revive the archived child.", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + }, + ], + }); + assert.isFalse(result.committed); + assert.deepEqual(result.storedEvents, []); + + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:foundation-archived-target-guard:cleanup-run"), + type: "run.updated", + threadId: rootThreadId, + runId, + providerInstanceId, + occurredAt: now, + payload: { ...runningRun, status: "cancelled", completedAt: now }, + }, + ], + }); + }), + ); + it.effect("interrupts a running process-bound effect when it is cancelled", () => Effect.gen(function* () { const outbox = yield* EffectOutboxV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b10e370cc5c..0f11d3a73af 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -24,6 +24,7 @@ import { type ProviderSessionId, ThreadId, } from "@t3tools/contracts"; +import * as NodeCrypto from "node:crypto"; import { modelSelectionsEqual } from "@t3tools/shared/model"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -39,9 +40,18 @@ import { CommandPolicyV2 } from "./CommandPolicy.ts"; import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts"; import { EventSinkV2 } from "./EventSink.ts"; -import type { OrchestrationEffectRequestV2, PendingOrchestrationEffectV2 } from "./EffectOutbox.ts"; +import { + PROCESS_BOUND_EFFECT_TYPES, + type OrchestrationEffectRequestV2, + type PendingOrchestrationEffectV2, +} from "./EffectOutbox.ts"; import { IdAllocatorV2 } from "./IdAllocator.ts"; import { makeKeyedSerialExecutor } from "./KeyedSerialExecutor.ts"; +import { + type OwnedSubagentLifecycleRepair, + planOwnedSubagentLifecycleRepairs, +} from "./OwnedSubagentLifecycleRepair.ts"; +import { ownedSubagentDescendants } from "./OwnedSubagentTree.ts"; import { applyToProjection, emptyProjection, ProjectionStoreV2 } from "./ProjectionStore.ts"; import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts"; @@ -411,6 +421,32 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const runtimePolicy = yield* RuntimePolicyV2; const threadForkService = yield* ThreadForkServiceV2; const threadDispatch = yield* makeKeyedSerialExecutor(); + const graphDispatch = yield* makeKeyedSerialExecutor(); + + const graphLockKeyForThread = (threadId: ThreadId) => + Effect.option(projectionStore.getThreadProjection(threadId)).pipe( + Effect.map( + Option.match({ + onNone: () => threadId, + onSome: (projection) => projection.thread.lineage.rootThreadId, + }), + ), + ); + + const graphLockKeyForCommand = (command: OrchestrationV2Command) => { + if (command.type === "thread.create") return Effect.succeed(command.threadId); + if (command.type === "thread.fork") { + return Effect.option(projectionStore.getThreadProjection(command.sourceThreadId)).pipe( + Effect.map( + Option.match({ + onNone: () => command.targetThreadId, + onSome: (projection) => projection.thread.lineage.rootThreadId, + }), + ), + ); + } + return graphLockKeyForThread(commandThreadId(command)); + }; const mapDispatchError = (command: OrchestrationV2Command) => @@ -544,6 +580,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const startNextQueuedRun = (threadId: ThreadId) => Effect.gen(function* () { const projection = yield* projectionStore.getThreadProjection(threadId); + if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) { + return; + } if (projection.runs.some(isBlockingRun)) { return; } @@ -706,7 +745,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (projection.runs.some(isBlockingRun) || nextQueuedRun(projection) === undefined) { return false; } - yield* threadDispatch.withLock(thread.id, startNextQueuedRun(thread.id)); + const graphLockKey = yield* graphLockKeyForThread(thread.id); + yield* graphDispatch.withLock( + graphLockKey, + threadDispatch.withLock(thread.id, startNextQueuedRun(thread.id)), + ); return true; }).pipe( Effect.catch((cause) => @@ -779,6 +822,453 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); }); + const loadThreadProjection = (threadId: ThreadId) => + projectionStore.getThreadProjection(threadId).pipe( + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId, + cause, + }), + ), + ); + + const sameInstant = (left: DateTime.Utc, right: DateTime.Utc): boolean => + DateTime.toEpochMillis(left) === DateTime.toEpochMillis(right); + + const lifecycleArchiveTime = ( + now: DateTime.Utc, + projections: ReadonlyArray, + ): DateTime.Utc => { + const occupied = new Set( + projections.flatMap((projection) => + projection.thread.archivedAt === null + ? [] + : [DateTime.toEpochMillis(projection.thread.archivedAt)], + ), + ); + let candidate = now; + while (occupied.has(DateTime.toEpochMillis(candidate))) { + candidate = DateTime.add(candidate, { milliseconds: 1 }); + } + return candidate; + }; + + const dispatchOwnedThreadLifecycle = Effect.fn("orchestrationV2.dispatch.ownedThreadLifecycle")( + function* ( + command: Extract< + OrchestrationV2Command, + { readonly type: "thread.archive" | "thread.unarchive" | "thread.delete" } + >, + rootProjection: OrchestrationV2ThreadProjection, + events: Ref.Ref>, + effects: Ref.Ref>, + ) { + const shell = yield* projectionStore.getShellSnapshot().pipe( + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: command.threadId, + cause, + }), + ), + ); + const descendants = ownedSubagentDescendants(command.threadId, [ + ...shell.threads, + ...shell.archivedThreads, + ]); + const descendantProjections = yield* Effect.forEach( + descendants, + ({ thread, depth }) => + loadThreadProjection(thread.id).pipe(Effect.map((projection) => ({ projection, depth }))), + { concurrency: 1 }, + ); + const subtree = [{ projection: rootProjection, depth: 0 }, ...descendantProjections]; + const ancestryOrder = subtree.toSorted((left, right) => left.depth - right.depth); + const now = yield* DateTime.now; + const archiveAt = + command.type === "thread.archive" + ? (rootProjection.thread.archivedAt ?? + lifecycleArchiveTime( + now, + subtree.map(({ projection }) => projection), + )) + : null; + const lifecycleAt = archiveAt ?? now; + const rootArchiveAt = rootProjection.thread.archivedAt; + const selected = new Set([command.threadId]); + if (command.type === "thread.archive") { + for (const { projection } of ancestryOrder.slice(1)) { + const parentThreadId = projection.thread.lineage.parentThreadId; + if ( + parentThreadId !== null && + selected.has(parentThreadId) && + projection.thread.deletedAt === null && + projection.thread.archivedAt === null + ) { + selected.add(projection.thread.id); + } + } + } else if (command.type === "thread.unarchive" && rootArchiveAt !== null) { + for (const { projection } of ancestryOrder.slice(1)) { + const parentThreadId = projection.thread.lineage.parentThreadId; + if ( + parentThreadId !== null && + selected.has(parentThreadId) && + projection.thread.deletedAt === null && + projection.thread.archivedAt !== null && + sameInstant(projection.thread.archivedAt, rootArchiveAt) + ) { + selected.add(projection.thread.id); + } + } + } else if (command.type === "thread.delete") { + for (const { projection } of ancestryOrder.slice(1)) { + selected.add(projection.thread.id); + } + } + const ordered = ancestryOrder + .filter(({ projection }) => selected.has(projection.thread.id)) + .toSorted((left, right) => + command.type === "thread.unarchive" ? left.depth - right.depth : right.depth - left.depth, + ); + const affectedThreadIds = ordered.map(({ projection }) => projection.thread.id); + const emitEvent = emit(events, command); + + if ( + command.type !== "thread.unarchive" && + rootProjection.thread.lineage.relationshipToParent === "subagent" && + rootProjection.thread.lineage.parentThreadId !== null + ) { + const ownerParentThreadId = rootProjection.thread.lineage.parentThreadId; + const ownerProjection = yield* Effect.option(loadThreadProjection(ownerParentThreadId)); + if (Option.isSome(ownerProjection) && ownerProjection.value.thread.deletedAt === null) { + const ownerTask = ownerProjection.value.subagents.find( + (candidate) => + candidate.childThreadId === rootProjection.thread.id && + ["pending", "running", "waiting"].includes(candidate.status), + ); + if (ownerTask !== undefined) { + yield* emitEvent({ + type: "subagent.updated", + threadId: ownerParentThreadId, + ...(ownerTask.runId === null ? {} : { runId: ownerTask.runId }), + nodeId: ownerTask.id, + driver: ownerTask.driver, + providerInstanceId: ownerTask.providerInstanceId, + occurredAt: lifecycleAt, + payload: { + ...ownerTask, + status: "cancelled", + completedAt: lifecycleAt, + updatedAt: lifecycleAt, + }, + }); + const ownerNode = ownerProjection.value.nodes.find( + (candidate) => candidate.id === ownerTask.id, + ); + if ( + ownerNode !== undefined && + ["pending", "running", "waiting"].includes(ownerNode.status) + ) { + yield* emitEvent({ + type: "node.updated", + threadId: ownerParentThreadId, + ...(ownerNode.runId === null ? {} : { runId: ownerNode.runId }), + nodeId: ownerNode.id, + driver: ownerTask.driver, + providerInstanceId: ownerTask.providerInstanceId, + occurredAt: lifecycleAt, + payload: { ...ownerNode, status: "cancelled", completedAt: lifecycleAt }, + }); + } + const ownerTurnItem = ownerProjection.value.turnItems.find( + (candidate) => candidate.type === "subagent" && candidate.subagentId === ownerTask.id, + ); + if ( + ownerTurnItem !== undefined && + ["pending", "running", "waiting"].includes(ownerTurnItem.status) + ) { + yield* emitEvent({ + type: "turn-item.updated", + threadId: ownerParentThreadId, + ...(ownerTurnItem.runId === null ? {} : { runId: ownerTurnItem.runId }), + ...(ownerTurnItem.nodeId === null ? {} : { nodeId: ownerTurnItem.nodeId }), + driver: ownerTask.driver, + providerInstanceId: ownerTask.providerInstanceId, + occurredAt: lifecycleAt, + payload: { + ...ownerTurnItem, + status: "cancelled", + completedAt: lifecycleAt, + updatedAt: lifecycleAt, + }, + }); + } + } + } + } + + for (const { projection } of ordered) { + const thread = projection.thread; + const shouldEmitThreadMutation = + command.type === "thread.archive" + ? thread.archivedAt === null + : command.type === "thread.unarchive" + ? selected.has(thread.id) + : true; + if (shouldEmitThreadMutation) { + const updatedThread: OrchestrationV2AppThread = + command.type === "thread.archive" + ? { ...thread, archivedAt: archiveAt!, updatedAt: archiveAt! } + : command.type === "thread.unarchive" + ? { ...thread, archivedAt: null, updatedAt: lifecycleAt } + : { + ...thread, + deletedAt: thread.deletedAt ?? lifecycleAt, + updatedAt: lifecycleAt, + }; + yield* emitEvent({ + type: + command.type === "thread.archive" + ? "thread.archived" + : command.type === "thread.unarchive" + ? "thread.unarchived" + : "thread.deleted", + threadId: thread.id, + providerInstanceId: updatedThread.providerInstanceId, + occurredAt: lifecycleAt, + payload: updatedThread, + }); + } + + if (command.type === "thread.unarchive") continue; + + const activeRunIds = new Set( + projection.runs + .filter((run) => + ["preparing", "queued", "starting", "running", "waiting"].includes(run.status), + ) + .map((run) => run.id), + ); + for (const run of projection.runs.filter((candidate) => activeRunIds.has(candidate.id))) { + yield* emitEvent({ + type: "run.updated", + threadId: thread.id, + runId: run.id, + providerInstanceId: run.providerInstanceId, + occurredAt: lifecycleAt, + payload: { + ...run, + status: "cancelled", + queuePosition: null, + completedAt: lifecycleAt, + }, + }); + } + for (const attempt of projection.attempts.filter( + (candidate) => candidate.status === "pending" || candidate.status === "running", + )) { + const run = projection.runs.find((candidate) => candidate.id === attempt.runId)!; + yield* emitEvent({ + type: "run-attempt.updated", + threadId: thread.id, + runId: attempt.runId, + nodeId: attempt.rootNodeId, + providerInstanceId: run.providerInstanceId, + occurredAt: lifecycleAt, + payload: { ...attempt, status: "cancelled", completedAt: lifecycleAt }, + }); + } + for (const node of projection.nodes.filter( + (candidate) => + candidate.runId !== null && + ["pending", "running", "waiting"].includes(candidate.status), + )) { + const run = projection.runs.find((candidate) => candidate.id === node.runId)!; + yield* emitEvent({ + type: "node.updated", + threadId: thread.id, + runId: run.id, + nodeId: node.id, + providerInstanceId: run.providerInstanceId, + occurredAt: lifecycleAt, + payload: { ...node, status: "cancelled", completedAt: lifecycleAt }, + }); + } + for (const subagent of projection.subagents.filter((candidate) => + ["pending", "running", "waiting"].includes(candidate.status), + )) { + yield* emitEvent({ + type: "subagent.updated", + threadId: thread.id, + ...(subagent.runId === null ? {} : { runId: subagent.runId }), + nodeId: subagent.id, + driver: subagent.driver, + providerInstanceId: subagent.providerInstanceId, + occurredAt: lifecycleAt, + payload: { + ...subagent, + status: "cancelled", + completedAt: lifecycleAt, + updatedAt: lifecycleAt, + }, + }); + } + for (const providerTurn of projection.providerTurns.filter( + (candidate) => candidate.status === "pending" || candidate.status === "running", + )) { + const attempt = projection.attempts.find( + (candidate) => candidate.id === providerTurn.runAttemptId, + ); + const run = projection.runs.find((candidate) => candidate.id === attempt?.runId); + yield* emitEvent({ + type: "provider-turn.updated", + threadId: thread.id, + ...(run === undefined ? {} : { runId: run.id }), + nodeId: providerTurn.nodeId, + ...(run === undefined ? {} : { providerInstanceId: run.providerInstanceId }), + occurredAt: lifecycleAt, + payload: { ...providerTurn, status: "cancelled", completedAt: lifecycleAt }, + }); + } + for (const message of projection.messages.filter((candidate) => candidate.streaming)) { + yield* emitEvent({ + type: "message.updated", + threadId: thread.id, + ...(message.runId === null ? {} : { runId: message.runId }), + ...(message.nodeId === null ? {} : { nodeId: message.nodeId }), + occurredAt: lifecycleAt, + payload: { ...message, streaming: false, updatedAt: lifecycleAt }, + }); + } + for (const item of projection.turnItems.filter((candidate) => + ["pending", "running", "waiting"].includes(candidate.status), + )) { + yield* emitEvent({ + type: "turn-item.updated", + threadId: thread.id, + ...(item.runId === null ? {} : { runId: item.runId }), + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + occurredAt: lifecycleAt, + payload: { + ...item, + status: "cancelled", + completedAt: lifecycleAt, + updatedAt: lifecycleAt, + }, + }); + } + for (const request of projection.runtimeRequests.filter( + (candidate) => candidate.status === "pending", + )) { + yield* emitEvent({ + type: "runtime-request.updated", + threadId: thread.id, + nodeId: request.nodeId, + occurredAt: lifecycleAt, + payload: { + ...request, + status: "cancelled", + responseCapability: { + type: "not_resumable", + reason: + command.type === "thread.archive" + ? "The thread was archived." + : "The thread was deleted.", + }, + resolvedAt: lifecycleAt, + }, + }); + } + for (const providerThread of projection.providerThreads.filter( + (candidate) => candidate.status === "active", + )) { + yield* emitEvent({ + type: "provider-thread.updated", + threadId: thread.id, + driver: providerThread.driver, + providerInstanceId: providerThread.providerInstanceId, + occurredAt: lifecycleAt, + payload: { ...providerThread, status: "idle", updatedAt: lifecycleAt }, + }); + } + + const liveSessions = projection.providerSessions.filter( + (session) => session.status !== "stopped" && session.status !== "error", + ); + for (const session of liveSessions) { + const detail = command.type === "thread.archive" ? "Thread archived." : "Thread deleted."; + yield* emitEvent({ + type: "provider-session.detached", + threadId: thread.id, + driver: session.driver, + providerInstanceId: session.providerInstanceId, + occurredAt: lifecycleAt, + payload: { + providerSessionId: session.id, + detachedAt: lifecycleAt, + reason: detail, + }, + }); + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:provider-session.detach:${thread.id}:${session.id}`, + commandId: command.commandId, + threadId: thread.id, + request: { + type: "provider-session.detach", + providerSessionId: session.id, + detail, + }, + } satisfies PendingOrchestrationEffectV2, + ]); + } + + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:terminal.cleanup:${thread.id}`, + commandId: command.commandId, + threadId: thread.id, + request: { type: "terminal.cleanup" }, + } satisfies PendingOrchestrationEffectV2, + ]); + + if (command.type === "thread.delete") { + const attachmentIds = Array.from( + new Set( + projection.messages.flatMap((message) => message.attachments.map((item) => item.id)), + ), + ); + if (attachmentIds.length > 0) { + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:attachment.cleanup:${thread.id}`, + commandId: command.commandId, + threadId: thread.id, + request: { type: "attachment.cleanup", attachmentIds }, + } satisfies PendingOrchestrationEffectV2, + ]); + } + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:checkpoint.cleanup:${thread.id}`, + commandId: command.commandId, + threadId: thread.id, + request: { type: "checkpoint.cleanup" }, + } satisfies PendingOrchestrationEffectV2, + ]); + } + } + + return affectedThreadIds; + }, + ); + const dispatchThreadMutation = Effect.fn("orchestrationV2.dispatch.threadMutation")(function* ( command: Extract< OrchestrationV2Command, @@ -797,15 +1287,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( - (cause) => - new OrchestratorProjectionError({ - threadId: command.threadId, - cause, - }), - ), - ); + const projection = yield* loadThreadProjection(command.threadId); const thread = projection.thread; if (thread.deletedAt !== null && command.type !== "thread.delete") { return yield* new OrchestratorDispatchError({ @@ -814,7 +1296,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} is deleted.`, }); } - if (command.type === "thread.archive" && thread.archivedAt !== null) { + if ( + command.type === "thread.archive" && + thread.archivedAt !== null && + !String(command.commandId).startsWith("command:system:owned-subagent-lifecycle-repair:") + ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, @@ -829,6 +1315,35 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } + if (command.type === "thread.unarchive") { + const parentThreadId = thread.lineage.parentThreadId; + if (thread.lineage.relationshipToParent === "subagent" && parentThreadId !== null) { + const parentProjection = yield* Effect.option(loadThreadProjection(parentThreadId)); + if (Option.isSome(parentProjection) && parentProjection.value.thread.deletedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Parent thread ${parentThreadId} is deleted; child ${command.threadId} cannot be restored.`, + }); + } + if (Option.isSome(parentProjection) && parentProjection.value.thread.archivedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Parent thread ${parentThreadId} is archived; unarchive it before restoring child ${command.threadId}.`, + }); + } + } + } + + if ( + command.type === "thread.archive" || + command.type === "thread.unarchive" || + command.type === "thread.delete" + ) { + return yield* dispatchOwnedThreadLifecycle(command, projection, events, effects); + } + const providerSwitchPlan = command.type === "thread.model-selection.set" || command.type === "provider.switch" ? yield* Effect.gen(function* () { @@ -854,12 +1369,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const now = yield* DateTime.now; const updatedThread: OrchestrationV2AppThread = (() => { switch (command.type) { - case "thread.archive": - return { ...thread, archivedAt: now, updatedAt: now }; - case "thread.unarchive": - return { ...thread, archivedAt: null, updatedAt: now }; - case "thread.delete": - return { ...thread, deletedAt: thread.deletedAt ?? now, updatedAt: now }; case "thread.metadata.update": return { ...thread, @@ -884,12 +1393,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio })(); const eventType = (() => { switch (command.type) { - case "thread.archive": - return "thread.archived" as const; - case "thread.unarchive": - return "thread.unarchived" as const; - case "thread.delete": - return "thread.deleted" as const; case "thread.metadata.update": return "thread.metadata-updated" as const; case "thread.runtime-mode.set": @@ -913,93 +1416,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: updatedThread, }); - if (command.type === "thread.delete") { - const emitEvent = emit(events, command); - const activeRunIds = new Set( - projection.runs - .filter((run) => - ["preparing", "queued", "starting", "running", "waiting"].includes(run.status), - ) - .map((run) => run.id), - ); - for (const run of projection.runs.filter((candidate) => activeRunIds.has(candidate.id))) { - yield* emitEvent({ - type: "run.updated", - threadId: command.threadId, - runId: run.id, - providerInstanceId: run.providerInstanceId, - occurredAt: now, - payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now }, - }); - } - for (const attempt of projection.attempts.filter( - (candidate) => - activeRunIds.has(candidate.runId) && - (candidate.status === "pending" || candidate.status === "running"), - )) { - const run = projection.runs.find((candidate) => candidate.id === attempt.runId)!; - yield* emitEvent({ - type: "run-attempt.updated", - threadId: command.threadId, - runId: attempt.runId, - nodeId: attempt.rootNodeId, - providerInstanceId: run.providerInstanceId, - occurredAt: now, - payload: { ...attempt, status: "cancelled", completedAt: now }, - }); - } - for (const node of projection.nodes.filter( - (candidate) => - candidate.runId !== null && - activeRunIds.has(candidate.runId) && - ["pending", "running", "waiting"].includes(candidate.status), - )) { - const run = projection.runs.find((candidate) => candidate.id === node.runId)!; - yield* emitEvent({ - type: "node.updated", - threadId: command.threadId, - runId: run.id, - nodeId: node.id, - providerInstanceId: run.providerInstanceId, - occurredAt: now, - payload: { ...node, status: "cancelled", completedAt: now }, - }); - } - for (const request of projection.runtimeRequests.filter( - (candidate) => candidate.status === "pending", - )) { - yield* emitEvent({ - type: "runtime-request.updated", - threadId: command.threadId, - nodeId: request.nodeId, - occurredAt: now, - payload: { - ...request, - status: "cancelled", - responseCapability: { - type: "not_resumable", - reason: "The thread was deleted.", - }, - resolvedAt: now, - }, - }); - } - } - const detachSessionIds = new Set( - command.type === "thread.archive" || command.type === "thread.delete" + command.type === "thread.metadata.update" && + command.worktreePath !== undefined && + command.worktreePath !== thread.worktreePath ? projection.providerSessions.map((session) => session.id) - : command.type === "thread.metadata.update" && - command.worktreePath !== undefined && - command.worktreePath !== thread.worktreePath - ? projection.providerSessions.map((session) => session.id) - : command.type === "thread.runtime-mode.set" - ? projection.providerSessions - .filter( - (session) => !session.capabilities.sessions.supportsRuntimeModeSwitchInSession, - ) - .map((session) => session.id) - : (providerSwitchPlan?.releaseProviderSessionIds ?? []), + : command.type === "thread.runtime-mode.set" + ? projection.providerSessions + .filter( + (session) => !session.capabilities.sessions.supportsRuntimeModeSwitchInSession, + ) + .map((session) => session.id) + : (providerSwitchPlan?.releaseProviderSessionIds ?? []), ); if (detachSessionIds.size > 0) { const liveSessions = projection.providerSessions.filter( @@ -1025,15 +1453,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio providerSessionId: session.id, detachedAt: now, reason: - command.type === "thread.archive" - ? "Thread archived." - : 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 = { @@ -1044,15 +1468,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio type: "provider-session.detach", providerSessionId: session.id, detail: - command.type === "thread.archive" - ? "Thread archived." - : 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.", }, } satisfies PendingOrchestrationEffectV2; yield* Ref.update(effects, (existing) => [...existing, pendingEffect]); @@ -1061,36 +1481,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); } - if (command.type === "thread.archive" || command.type === "thread.delete") { - yield* Ref.update(effects, (existing) => [ - ...existing, - { - id: `effect:${command.commandId}:terminal.cleanup`, - commandId: command.commandId, - threadId: command.threadId, - request: { type: "terminal.cleanup" }, - } satisfies PendingOrchestrationEffectV2, - ]); - } - - if (command.type === "thread.delete") { - const attachmentIds = Array.from( - new Set( - projection.messages.flatMap((message) => message.attachments.map((item) => item.id)), - ), - ); - if (attachmentIds.length > 0) { - yield* Ref.update(effects, (existing) => [ - ...existing, - { - id: `effect:${command.commandId}:attachment.cleanup`, - commandId: command.commandId, - threadId: command.threadId, - request: { type: "attachment.cleanup", attachmentIds }, - } satisfies PendingOrchestrationEffectV2, - ]); - } - } + return undefined; }); const dispatchProviderSessionDetach = Effect.fn("orchestrationV2.dispatch.providerSessionDetach")( @@ -3455,6 +3846,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); + if ( + parentProjection.thread.archivedAt !== null || + parentProjection.thread.deletedAt !== null + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Parent thread ${command.parentThreadId} is not active.`, + }); + } const parentRun = parentProjection.runs.find( (candidate) => candidate.id === command.parentRunId, ); @@ -3708,6 +4109,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); + if ( + parentProjection.thread.archivedAt !== null || + parentProjection.thread.deletedAt !== null + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Parent thread ${command.parentThreadId} is not active.`, + }); + } const targetProjection = yield* projectionStore .getThreadProjection(command.targetThreadId) .pipe( @@ -4735,6 +5146,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const finalizeAppOwnedSubagent = (childThreadId: ThreadId) => Effect.gen(function* () { const childProjection = yield* projectionStore.getThreadProjection(childThreadId); + if (childProjection.thread.deletedAt !== null) { + return; + } const forkedFrom = childProjection.thread.forkedFrom; if ( childProjection.thread.lineage.relationshipToParent !== "subagent" || @@ -4754,6 +5168,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const parentThreadId = childProjection.thread.lineage.parentThreadId; const parentProjection = yield* projectionStore.getThreadProjection(parentThreadId); + if ( + parentProjection.thread.archivedAt !== null || + parentProjection.thread.deletedAt !== null + ) { + return; + } const task = parentProjection.subagents.find( (candidate) => candidate.id === forkedFrom.nodeId && @@ -4950,6 +5370,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly events: ReadonlyArray; readonly effects: ReadonlyArray; readonly cancelUnsettledEffects?: { + readonly threadIds?: ReadonlyArray; readonly effectTypes: ReadonlyArray; readonly reason: string; }; @@ -4966,6 +5387,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const effects = yield* Ref.make>([]); let cancelUnsettledEffects: | { + readonly threadIds?: ReadonlyArray; readonly effectTypes: ReadonlyArray; readonly reason: string; } @@ -4974,9 +5396,27 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.create": yield* dispatchThreadCreate(command, events); break; - case "thread.archive": + case "thread.archive": { + const threadIds = yield* dispatchThreadMutation(command, events, effects); + cancelUnsettledEffects = { + threadIds: threadIds ?? [command.threadId], + effectTypes: PROCESS_BOUND_EFFECT_TYPES, + reason: "The owning thread subtree was archived.", + }; + break; + } + case "thread.delete": { + const threadIds = yield* dispatchThreadMutation(command, events, effects); + cancelUnsettledEffects = { + threadIds: threadIds ?? [command.threadId], + effectTypes: [...PROCESS_BOUND_EFFECT_TYPES, "checkpoint.capture"], + reason: "The owning thread subtree was deleted.", + }; + break; + } case "thread.unarchive": - case "thread.delete": + yield* dispatchThreadMutation(command, events, effects); + break; case "thread.metadata.update": case "thread.runtime-mode.set": case "thread.interaction-mode.set": @@ -5160,7 +5600,121 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); const dispatchWithReceipt = (command: OrchestrationV2Command) => - threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)); + Effect.flatMap(graphLockKeyForCommand(command), (graphLockKey) => + graphDispatch.withLock( + graphLockKey, + threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)), + ), + ); + + const lifecycleRepairCommand = ( + repair: OwnedSubagentLifecycleRepair, + discriminator: string, + ): Extract => { + const commandId = CommandId.make( + `command:system:owned-subagent-lifecycle-repair:${repair.type}:${repair.parentThreadId}:${discriminator}`, + ); + return repair.type === "delete" + ? { type: "thread.delete", commandId, threadId: repair.parentThreadId } + : { type: "thread.archive", commandId, threadId: repair.parentThreadId }; + }; + + const lifecycleRepairFingerprint = (repair: OwnedSubagentLifecycleRepair): string => + NodeCrypto.createHash("sha256") + .update(repair.childThreadIds.join("\n")) + .digest("hex") + .slice(0, 24); + + const dispatchLifecycleRepair = (repair: OwnedSubagentLifecycleRepair, discriminator: string) => + dispatchWithReceipt(lifecycleRepairCommand(repair, discriminator)).pipe(Effect.asVoid); + + const repairCreatedSubagentLifecycle = (stored: OrchestrationV2StoredEvent) => + Effect.gen(function* () { + if (stored.event.type !== "thread.created") return; + const childThreadId = stored.event.threadId; + const graphLockKey = yield* graphLockKeyForThread(childThreadId); + yield* graphDispatch.withLock( + graphLockKey, + Effect.gen(function* () { + const child = yield* loadThreadProjection(childThreadId); + const parentThreadId = child.thread.lineage.parentThreadId; + if ( + child.thread.deletedAt !== null || + child.thread.lineage.relationshipToParent !== "subagent" || + parentThreadId === null + ) { + return; + } + const parent = yield* Effect.option(loadThreadProjection(parentThreadId)); + if (Option.isNone(parent)) return; + const repairType = + parent.value.thread.deletedAt !== null + ? "delete" + : parent.value.thread.archivedAt !== null && child.thread.archivedAt === null + ? "archive" + : null; + if (repairType === null) return; + + const repair: OwnedSubagentLifecycleRepair = { + type: repairType, + parentThreadId, + childThreadIds: [childThreadId], + }; + yield* threadDispatch.withLock( + parentThreadId, + dispatchWithReceiptEffect(lifecycleRepairCommand(repair, stored.event.id)).pipe( + Effect.asVoid, + ), + ); + }), + ); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to repair newly-created owned subagent lifecycle", { + threadId: stored.event.threadId, + cause, + }), + ), + ); + + // Capture the live boundary before reading projections. Threads created + // during startup reconciliation are then replayed by the live subscription. + const lifecycleRepairAfterSequence = yield* eventSink.latestSequence().pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to read owned subagent lifecycle repair boundary", { + cause, + }).pipe(Effect.as(0)), + ), + ); + const lifecycleRecords = yield* projectionStore + .getThreadLifecycleRecords() + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to read owned subagent lifecycle records", { cause }).pipe( + Effect.as([]), + ), + ), + ); + yield* Effect.forEach( + planOwnedSubagentLifecycleRepairs(lifecycleRecords), + (repair) => + dispatchLifecycleRepair(repair, lifecycleRepairFingerprint(repair)).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to reconcile owned subagent lifecycle", { + threadId: repair.parentThreadId, + lifecycle: repair.type, + cause, + }), + ), + ), + { concurrency: 1, discard: true }, + ); + + yield* eventSink.stream({ afterSequence: lifecycleRepairAfterSequence }).pipe( + Stream.filter((stored) => stored.event.type === "thread.created"), + Stream.runForEach(repairCreatedSubagentLifecycle), + Effect.forkDetach, + ); yield* eventSink.stream().pipe( Stream.filter( @@ -5174,14 +5728,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio stored.event.payload.status === "rolled_back"), ), Stream.runForEach((stored) => - threadDispatch - .withLock( - stored.event.threadId, - finalizeAppOwnedSubagent(stored.event.threadId).pipe( - Effect.andThen(startNextQueuedRun(stored.event.threadId)), + Effect.flatMap(graphLockKeyForThread(stored.event.threadId), (graphLockKey) => + graphDispatch.withLock( + graphLockKey, + threadDispatch.withLock( + stored.event.threadId, + finalizeAppOwnedSubagent(stored.event.threadId).pipe( + Effect.andThen(startNextQueuedRun(stored.event.threadId)), + ), ), - ) - .pipe(Effect.catchCause(() => Effect.void)), + ), + ).pipe(Effect.catchCause(() => Effect.void)), ), Effect.forkDetach, ); diff --git a/apps/server/src/orchestration-v2/OwnedSubagentLifecycleRepair.test.ts b/apps/server/src/orchestration-v2/OwnedSubagentLifecycleRepair.test.ts new file mode 100644 index 00000000000..7851da1e46a --- /dev/null +++ b/apps/server/src/orchestration-v2/OwnedSubagentLifecycleRepair.test.ts @@ -0,0 +1,93 @@ +import { assert, it } from "@effect/vitest"; +import { ThreadId, type OrchestrationV2AppThread } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { planOwnedSubagentLifecycleRepairs } from "./OwnedSubagentLifecycleRepair.ts"; + +const now = DateTime.makeUnsafe("2026-07-09T12:00:00.000Z"); + +function thread(input: { + readonly id: string; + readonly parentThreadId?: string; + readonly relationshipToParent?: "fork" | "subagent"; + readonly archived?: boolean; + readonly deleted?: boolean; +}): OrchestrationV2AppThread { + const id = ThreadId.make(input.id); + return { + id, + lineage: { + parentThreadId: + input.parentThreadId === undefined ? null : ThreadId.make(input.parentThreadId), + relationshipToParent: input.relationshipToParent ?? null, + rootThreadId: id, + }, + archivedAt: input.archived === true ? now : null, + deletedAt: input.deleted === true ? now : null, + } as unknown as OrchestrationV2AppThread; +} + +it.effect("plans grouped repairs only for inconsistent owned-subagent edges", () => + Effect.sync(() => { + const deletedParent = thread({ id: "deleted-parent", deleted: true }); + const firstDeletedChild = thread({ + id: "deleted-child-a", + parentThreadId: deletedParent.id, + relationshipToParent: "subagent", + }); + const secondDeletedChild = thread({ + id: "deleted-child-b", + parentThreadId: deletedParent.id, + relationshipToParent: "subagent", + archived: true, + }); + const archivedParent = thread({ id: "archived-parent", archived: true }); + const activeArchivedChild = thread({ + id: "active-archived-child", + parentThreadId: archivedParent.id, + relationshipToParent: "subagent", + }); + const alreadyArchivedChild = thread({ + id: "already-archived-child", + parentThreadId: archivedParent.id, + relationshipToParent: "subagent", + archived: true, + }); + const fork = thread({ + id: "fork", + parentThreadId: deletedParent.id, + relationshipToParent: "fork", + }); + const missingParent = thread({ + id: "missing-parent", + parentThreadId: "absent", + relationshipToParent: "subagent", + }); + + assert.deepEqual( + planOwnedSubagentLifecycleRepairs([ + missingParent, + secondDeletedChild, + archivedParent, + firstDeletedChild, + deletedParent, + fork, + alreadyArchivedChild, + activeArchivedChild, + ]), + [ + { + type: "delete", + parentThreadId: deletedParent.id, + childThreadIds: [firstDeletedChild.id, secondDeletedChild.id], + }, + { + type: "archive", + parentThreadId: archivedParent.id, + childThreadIds: [activeArchivedChild.id], + }, + ], + ); + }), +); diff --git a/apps/server/src/orchestration-v2/OwnedSubagentLifecycleRepair.ts b/apps/server/src/orchestration-v2/OwnedSubagentLifecycleRepair.ts new file mode 100644 index 00000000000..a6aa32662ed --- /dev/null +++ b/apps/server/src/orchestration-v2/OwnedSubagentLifecycleRepair.ts @@ -0,0 +1,60 @@ +import type { OrchestrationV2AppThread, ThreadId } from "@t3tools/contracts"; + +export interface OwnedSubagentLifecycleRepair { + readonly type: "archive" | "delete"; + readonly parentThreadId: ThreadId; + readonly childThreadIds: ReadonlyArray; +} + +/** + * Finds direct ownership edges whose child lifecycle disagrees with its + * parent. Repairing the parent recursively fixes the complete affected branch; + * grouping direct children keeps the resulting command id stable and makes a + * later, newly-created orphan produce a distinct repair command. + */ +export function planOwnedSubagentLifecycleRepairs( + threads: ReadonlyArray, +): ReadonlyArray { + const threadsById = new Map(threads.map((thread) => [thread.id, thread] as const)); + const repairsByKey = new Map(); + + for (const child of threads) { + const parentThreadId = child.lineage.parentThreadId; + if ( + child.deletedAt !== null || + child.lineage.relationshipToParent !== "subagent" || + parentThreadId === null + ) { + continue; + } + const parent = threadsById.get(parentThreadId); + if (parent === undefined) continue; + + const type = + parent.deletedAt !== null + ? "delete" + : parent.archivedAt !== null && child.archivedAt === null + ? "archive" + : null; + if (type === null) continue; + + const key = `${type}:${parentThreadId}`; + const existing = repairsByKey.get(key); + repairsByKey.set(key, { + type, + parentThreadId, + childThreadIds: [...(existing?.childThreadIds ?? []), child.id], + }); + } + + return Array.from(repairsByKey.values()) + .map((repair) => ({ + ...repair, + childThreadIds: repair.childThreadIds.toSorted((left, right) => left.localeCompare(right)), + })) + .toSorted( + (left, right) => + (left.type === right.type ? 0 : left.type === "delete" ? -1 : 1) || + left.parentThreadId.localeCompare(right.parentThreadId), + ); +} diff --git a/apps/server/src/orchestration-v2/OwnedSubagentTree.test.ts b/apps/server/src/orchestration-v2/OwnedSubagentTree.test.ts new file mode 100644 index 00000000000..94881041146 --- /dev/null +++ b/apps/server/src/orchestration-v2/OwnedSubagentTree.test.ts @@ -0,0 +1,98 @@ +import { assert, it } from "@effect/vitest"; +import { + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationV2ThreadShell, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { ownedSubagentDescendants } from "./OwnedSubagentTree.ts"; + +const providerInstanceId = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId: providerInstanceId, model: "gpt-test" }; + +function shell(input: { + readonly id: string; + readonly parentThreadId?: string; + readonly relationshipToParent?: "subagent" | "fork"; + readonly rootThreadId?: string; +}): OrchestrationV2ThreadShell { + const now = DateTime.makeUnsafe("2026-07-09T12:00:00.000Z"); + const id = ThreadId.make(input.id); + return { + createdBy: "user", + creationSource: "web", + id, + projectId: ProjectId.make("project:owned-subagent-tree"), + title: input.id, + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + lineage: { + parentThreadId: + input.parentThreadId === undefined ? null : ThreadId.make(input.parentThreadId), + relationshipToParent: input.relationshipToParent ?? null, + rootThreadId: ThreadId.make(input.rootThreadId ?? input.id), + }, + forkedFrom: null, + activeProviderThreadId: null, + latestRunId: null, + activeRunId: null, + status: "idle", + pendingRuntimeRequest: null, + latestVisibleMessage: null, + latestUserMessageAt: null, + hasActionableProposedPlan: false, + itemCount: 0, + visibleItemCount: 0, + createdAt: now, + updatedAt: now, + archivedAt: null, + deletedAt: null, + }; +} + +it.effect("walks only recursively owned subagent edges", () => + Effect.sync(() => { + const root = shell({ id: "thread:root" }); + const child = shell({ + id: "thread:child", + parentThreadId: root.id, + relationshipToParent: "subagent", + rootThreadId: root.id, + }); + const grandchild = shell({ + id: "thread:grandchild", + parentThreadId: child.id, + relationshipToParent: "subagent", + rootThreadId: root.id, + }); + const fork = shell({ + id: "thread:fork", + parentThreadId: root.id, + relationshipToParent: "fork", + rootThreadId: root.id, + }); + const forkSubagent = shell({ + id: "thread:fork-subagent", + parentThreadId: fork.id, + relationshipToParent: "subagent", + rootThreadId: root.id, + }); + + assert.deepEqual( + ownedSubagentDescendants(root.id, [root, grandchild, forkSubagent, fork, child]).map( + ({ thread, depth }) => [thread.id, depth], + ), + [ + [child.id, 1], + [grandchild.id, 2], + ], + ); + }), +); diff --git a/apps/server/src/orchestration-v2/OwnedSubagentTree.ts b/apps/server/src/orchestration-v2/OwnedSubagentTree.ts new file mode 100644 index 00000000000..09fcd338d67 --- /dev/null +++ b/apps/server/src/orchestration-v2/OwnedSubagentTree.ts @@ -0,0 +1,46 @@ +import type { OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts"; + +export interface OwnedSubagentTreeNode { + readonly thread: OrchestrationV2ThreadShell; + readonly depth: number; +} + +/** + * Returns the canonical descendants owned by `rootThreadId`. + * + * Fork and context-transfer relationships are intentionally not ownership + * edges. A visited set also makes corrupted lineage cycles safe to inspect. + */ +export function ownedSubagentDescendants( + rootThreadId: ThreadId, + threads: ReadonlyArray, +): ReadonlyArray { + const childrenByParent = new Map>(); + for (const thread of threads) { + const parentThreadId = thread.lineage.parentThreadId; + if (thread.lineage.relationshipToParent !== "subagent" || parentThreadId === null) { + continue; + } + const children = childrenByParent.get(parentThreadId) ?? []; + children.push(thread); + childrenByParent.set(parentThreadId, children); + } + + const visited = new Set([rootThreadId]); + const queue: Array<{ readonly threadId: ThreadId; readonly depth: number }> = [ + { threadId: rootThreadId, depth: 0 }, + ]; + const descendants: Array = []; + for (let index = 0; index < queue.length; index += 1) { + const current = queue[index]!; + const children = childrenByParent.get(current.threadId) ?? []; + for (const child of children) { + if (visited.has(child.id)) continue; + visited.add(child.id); + const depth = current.depth + 1; + descendants.push({ thread: child, depth }); + queue.push({ threadId: child.id, depth }); + } + } + return descendants; +} diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index a9a33fa9435..b82464117e7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -38,6 +38,63 @@ const providerInstanceId = modelSelection.instanceId; const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); it.layer(TestLayer)("ProjectionStoreV2", (it) => { + it.effect("keeps deleted thread headers available for lifecycle repair", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:projection-lifecycle-records:deleted"); + const thread = { + createdBy: "user" as const, + creationSource: "web" as const, + id: threadId, + projectId: ProjectId.make("project:projection-lifecycle-records"), + title: "Deleted lifecycle record", + providerInstanceId, + modelSelection, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + deletedAt: null, + }; + + yield* projectionStore.apply({ + id: EventId.make("event:projection-lifecycle-records:created"), + type: "thread.created", + threadId, + occurredAt: now, + payload: thread, + }); + yield* projectionStore.apply({ + id: EventId.make("event:projection-lifecycle-records:deleted"), + type: "thread.deleted", + threadId, + occurredAt: now, + payload: { ...thread, deletedAt: now }, + }); + + const shell = yield* projectionStore.getShellSnapshot(); + assert.notInclude( + [...shell.threads, ...shell.archivedThreads].map((item) => item.id), + threadId, + ); + assert.include( + (yield* projectionStore.getThreadLifecycleRecords()).map((item) => item.id), + threadId, + ); + }), + ); + it.effect("projects one shared provider session into multiple thread bindings", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 9122f8dd79c..8c31a4c7b31 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,4 +1,5 @@ import type { + OrchestrationV2AppThread, OrchestrationV2ConversationMessage, OrchestrationV2DomainEvent, OrchestrationV2ProjectedTurnItem, @@ -104,6 +105,11 @@ export interface ProjectionStoreV2Shape { OrchestrationV2ThreadShellSnapshot, ProjectionStoreV2Error >; + /** Returns lifecycle headers for every thread, including archived and deleted records. */ + readonly getThreadLifecycleRecords: () => Effect.Effect< + ReadonlyArray, + ProjectionStoreV2Error + >; readonly getThreadProjection: ( threadId: ThreadId, ) => Effect.Effect; @@ -1998,6 +2004,24 @@ export const layer: Layer.Layer = const getThreadProjection: ProjectionStoreV2Shape["getThreadProjection"] = (threadId) => readProjection(threadId, new Set()); + const getThreadLifecycleRecords: ProjectionStoreV2Shape["getThreadLifecycleRecords"] = () => + sql<{ readonly payload_json: string }>` + SELECT payload_json + FROM orchestration_v2_projection_threads + ORDER BY created_at ASC, thread_id ASC + `.pipe( + Effect.flatMap((rows) => + Effect.forEach(rows, (row) => decodeThreadPayload(row.payload_json)), + ), + Effect.mapError( + (cause) => + new ProjectionStoreReadError({ + threadId: ThreadId.make("thread:lifecycle-records"), + cause, + }), + ), + ); + const getThreadSnapshot: ProjectionStoreV2Shape["getThreadSnapshot"] = (threadId) => sql .withTransaction( @@ -2200,6 +2224,7 @@ export const layer: Layer.Layer = return { apply, getShellSnapshot, + getThreadLifecycleRecords, getThreadProjection, getThreadSnapshot, } satisfies ProjectionStoreV2Shape; @@ -2252,6 +2277,18 @@ export const layerMemory: Layer.Layer = Layer.effect( archivedThreads: visible.filter((thread) => thread.archivedAt !== null), }; }), + getThreadLifecycleRecords: () => + Ref.get(replayState).pipe( + Effect.map((state) => + [...state.projections.values()] + .map((projection) => projection.thread) + .toSorted( + (left, right) => + DateTime.toEpochMillis(left.createdAt) - + DateTime.toEpochMillis(right.createdAt) || left.id.localeCompare(right.id), + ), + ), + ), getThreadProjection: (threadId) => Effect.gen(function* () { const existing = (yield* Ref.get(replayState)).projections; diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 91e4a974390..d24e0f6cded 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -79,7 +79,10 @@ export interface ProviderEventIngestorV2Shape { readonly writeIfRunCurrent?: { readonly runId: RunId; readonly activeAttemptId: RunAttemptId; - readonly expectedStatus: OrchestrationV2Run["status"]; + readonly expectedStatus: + | OrchestrationV2Run["status"] + | ReadonlyArray; + readonly allowSupersededAttemptTerminalArtifacts?: boolean; }; }, ) => Effect.Effect, ProviderEventIngestorV2Error>; diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index dc5e7232c98..76c3514a171 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -212,6 +212,7 @@ it.effect( ProjectionStoreV2.of({ apply: () => Effect.void, getShellSnapshot: () => Effect.die("unused getShellSnapshot"), + getThreadLifecycleRecords: () => Effect.die("unused getThreadLifecycleRecords"), getThreadProjection: () => Ref.get(projection), getThreadSnapshot: () => Effect.die("unused getThreadSnapshot"), }), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts new file mode 100644 index 00000000000..38411eba8a6 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -0,0 +1,350 @@ +import { assert, it } from "@effect/vitest"; +import { + CheckpointScopeId, + MessageId, + type ModelSelection, + NodeId, + type OrchestrationV2ThreadProjection, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ProviderThreadId, + RunAttemptId, + RunId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; + +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts"; +import { EventSinkV2 } from "./EventSink.ts"; +import { layer as idAllocatorLayer } from "./IdAllocator.ts"; +import { ProjectionStoreV2 } from "./ProjectionStore.ts"; +import type { ProviderAdapterV2SessionRuntime } from "./ProviderAdapter.ts"; +import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; +import { + layer as providerTurnStartLayer, + ProviderTurnStartServiceV2, +} from "./ProviderTurnStartService.ts"; +import { RunExecutionServiceV2 } from "./RunExecutionService.ts"; +import { layer as runtimePolicyLayer } from "./RuntimePolicy.ts"; + +const driver = ProviderDriverKind.make("codex"); +const providerInstanceId = ProviderInstanceId.make("codex"); +const modelSelection = { + instanceId: providerInstanceId, + model: "gpt-5.4", +} satisfies ModelSelection; + +type LifecycleMutation = "archive" | "delete"; +type RacePoint = "after_open" | "before_running_write"; + +function makeProjection(input: { + readonly now: DateTime.Utc; + readonly threadId: ThreadId; + readonly runId: RunId; + readonly attemptId: RunAttemptId; + readonly providerSessionId: ProviderSessionId; + readonly providerThreadId: ProviderThreadId; +}): OrchestrationV2ThreadProjection { + const rootNodeId = NodeId.make(`node:${input.runId}`); + const messageId = MessageId.make(`message:${input.runId}`); + const checkpointScopeId = CheckpointScopeId.make(`checkpoint-scope:${input.runId}`); + return { + thread: { + createdBy: "user", + creationSource: "web", + id: input.threadId, + projectId: ProjectId.make(`project:${input.threadId}`), + title: "Provider start lifecycle race", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: "/workspace", + activeProviderThreadId: input.providerThreadId, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: input.threadId, + }, + forkedFrom: null, + createdAt: input.now, + updatedAt: input.now, + archivedAt: null, + deletedAt: null, + }, + runs: [ + { + id: input.runId, + threadId: input.threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: input.providerThreadId, + userMessageId: messageId, + rootNodeId, + activeAttemptId: input.attemptId, + status: "starting", + queuePosition: null, + requestedAt: input.now, + startedAt: null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + ], + attempts: [ + { + id: input.attemptId, + runId: input.runId, + attemptOrdinal: 1, + rootNodeId, + providerInstanceId, + providerThreadId: input.providerThreadId, + providerTurnId: null, + reason: "initial", + status: "pending", + startedAt: null, + completedAt: null, + }, + ], + nodes: [ + { + id: rootNodeId, + threadId: input.threadId, + runId: input.runId, + parentNodeId: null, + rootNodeId, + kind: "root_turn", + status: "pending", + countsForRun: true, + providerThreadId: input.providerThreadId, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId, + startedAt: null, + completedAt: null, + }, + ], + subagents: [], + providerSessions: [], + providerThreads: [ + { + id: input.providerThreadId, + driver, + providerInstanceId, + providerSessionId: input.providerSessionId, + appThreadId: input.threadId, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "not_loaded", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: input.now, + updatedAt: input.now, + }, + ], + providerTurns: [], + runtimeRequests: [], + messages: [ + { + createdBy: "user", + creationSource: "web", + id: messageId, + threadId: input.threadId, + runId: input.runId, + nodeId: rootNodeId, + role: "user", + text: "Start the provider turn.", + attachments: [], + streaming: false, + createdAt: input.now, + updatedAt: input.now, + }, + ], + plans: [], + turnItems: [], + checkpointScopes: [ + { + id: checkpointScopeId, + threadId: input.threadId, + runId: input.runId, + nodeId: rootNodeId, + parentScopeId: null, + providerThreadId: input.providerThreadId, + kind: "root_run", + ordinalWithinParent: 0, + advancesAppRunCount: true, + cwd: "/workspace", + createdAt: input.now, + }, + ], + checkpoints: [], + contextHandoffs: [], + contextTransfers: [], + visibleTurnItems: [], + updatedAt: input.now, + }; +} + +function runLifecycleRaceTest(input: { + readonly lifecycle: LifecycleMutation; + readonly racePoint: RacePoint; +}) { + return Effect.gen(function* () { + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:provider-start:${input.lifecycle}:${input.racePoint}`); + const runId = RunId.make(`run:provider-start:${input.lifecycle}:${input.racePoint}`); + const attemptId = RunAttemptId.make( + `attempt:provider-start:${input.lifecycle}:${input.racePoint}`, + ); + const providerSessionId = ProviderSessionId.make( + `session:provider-start:${input.lifecycle}:${input.racePoint}`, + ); + const providerThreadId = ProviderThreadId.make( + `provider-thread:provider-start:${input.lifecycle}:${input.racePoint}`, + ); + const projection = yield* Ref.make( + makeProjection({ + now, + threadId, + runId, + attemptId, + providerSessionId, + providerThreadId, + }), + ); + const openCount = yield* Ref.make(0); + const ensureThreadCount = yield* Ref.make(0); + const detachCount = yield* Ref.make(0); + const runningWriteCount = yield* Ref.make(0); + const executionCount = yield* Ref.make(0); + + const mutateLifecycle = Ref.update(projection, (current) => ({ + ...current, + thread: { + ...current.thread, + archivedAt: input.lifecycle === "archive" ? now : null, + deletedAt: input.lifecycle === "delete" ? now : null, + updatedAt: now, + }, + ...(input.lifecycle === "delete" + ? { + runs: current.runs.map((run) => + run.id === runId ? { ...run, status: "cancelled" as const, completedAt: now } : run, + ), + } + : {}), + })); + const providerSession = { + id: providerSessionId, + driver, + providerInstanceId, + status: "ready" as const, + cwd: "/workspace", + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }; + const runtime: ProviderAdapterV2SessionRuntime = { + instanceId: providerInstanceId, + driver, + providerSessionId, + providerSession, + events: Stream.empty, + ensureThread: () => + Effect.gen(function* () { + yield* Ref.update(ensureThreadCount, (count) => count + 1); + return (yield* Ref.get(projection)).providerThreads[0]!; + }), + resumeThread: () => Effect.die("resumeThread is unused"), + startTurn: () => Effect.die("startTurn is unused"), + steerTurn: () => Effect.die("steerTurn is unused"), + interruptTurn: () => Effect.die("interruptTurn is unused"), + respondToRuntimeRequest: () => Effect.die("respondToRuntimeRequest is unused"), + readThreadSnapshot: () => Effect.die("readThreadSnapshot is unused"), + rollbackThread: () => Effect.die("rollbackThread is unused"), + forkThread: () => Effect.die("forkThread is unused"), + }; + const dependencies = Layer.mergeAll( + Layer.mock(EventSinkV2)({ + write: () => Effect.succeed([]), + writeIfRunCurrent: () => + Effect.gen(function* () { + yield* Ref.update(runningWriteCount, (count) => count + 1); + if (input.racePoint === "before_running_write") { + yield* mutateLifecycle; + return { committed: false as const, storedEvents: [] }; + } + return { committed: true as const, storedEvents: [] }; + }), + }), + Layer.mock(ContextHandoffServiceV2)({}), + idAllocatorLayer, + Layer.mock(ProjectionStoreV2)({ + getThreadProjection: () => Ref.get(projection), + }), + Layer.succeed( + ProviderSessionManagerV2, + ProviderSessionManagerV2.of({ + shutdown: Effect.void, + open: () => + Effect.gen(function* () { + yield* Ref.update(openCount, (count) => count + 1); + if (input.racePoint === "after_open") { + yield* mutateLifecycle; + } + return runtime; + }), + get: () => Effect.succeed(Option.none()), + close: () => Effect.void, + release: () => Effect.void, + detach: () => Ref.update(detachCount, (count) => count + 1), + }), + ), + Layer.mock(RunExecutionServiceV2)({ + startRootRun: () => Ref.update(executionCount, (count) => count + 1), + }), + runtimePolicyLayer, + ); + const testLayer = providerTurnStartLayer.pipe(Layer.provide(dependencies)); + + yield* Effect.gen(function* () { + const service = yield* ProviderTurnStartServiceV2; + yield* service.start({ threadId, runId }); + }).pipe(Effect.provide(testLayer)); + + assert.equal(yield* Ref.get(openCount), 1); + assert.equal(yield* Ref.get(detachCount), 1); + assert.equal(yield* Ref.get(executionCount), 0); + if (input.racePoint === "after_open") { + assert.equal(yield* Ref.get(ensureThreadCount), 0); + assert.equal(yield* Ref.get(runningWriteCount), 0); + } else { + assert.equal(yield* Ref.get(ensureThreadCount), 1); + assert.equal(yield* Ref.get(runningWriteCount), 1); + } + }); +} + +it.effect("detaches a provider session when deletion wins immediately after open", () => + runLifecycleRaceTest({ lifecycle: "delete", racePoint: "after_open" }), +); + +it.effect("detaches a provider session when archival wins the running-state commit", () => + runLifecycleRaceTest({ lifecycle: "archive", racePoint: "before_running_write" }), +); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index b8bd2b2fe84..ee4eaad3a9a 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -12,6 +12,7 @@ import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import { EventSinkV2 } from "./EventSink.ts"; @@ -78,15 +79,28 @@ export const layer: Layer.Layer< if (run === undefined) { return yield* new ProviderTurnStartError({ runId, cause: `Run ${runId} was not found.` }); } + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === run.providerThreadId, + ); + if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) { + if ( + providerThread?.providerSessionId !== null && + providerThread?.providerSessionId !== undefined + ) { + yield* providerSessions.detach({ + providerSessionId: providerThread.providerSessionId, + threadId: projection.thread.id, + detail: "Provider turn start skipped because the thread is archived or deleted.", + }); + } + return; + } if (run.status !== "starting") { // The effect is idempotent once the run has advanced or terminalized. return; } const rootNode = projection.nodes.find((candidate) => candidate.id === run.rootNodeId); const attempt = projection.attempts.find((candidate) => candidate.id === run.activeAttemptId); - const providerThread = projection.providerThreads.find( - (candidate) => candidate.id === run.providerThreadId, - ); const message = projection.messages.find((candidate) => candidate.id === run.userMessageId); const checkpointScope = projection.checkpointScopes.find( (candidate) => candidate.id === rootNode?.checkpointScopeId, @@ -125,19 +139,48 @@ export const layer: Layer.Layer< }); } const providerSessionId = providerThread.providerSessionId; - const isCurrentAttemptInStatus = ( - expectedStatus: OrchestrationV2Run["status"], - ): Effect.Effect => - projectionStore.getThreadProjection(projection.thread.id).pipe( - Effect.map((current) => { - const currentRun = current.runs.find((candidate) => candidate.id === run.id); - return ( - currentRun?.activeAttemptId === attempt.id && currentRun.status === expectedStatus + type CurrentAttemptState = "current" | "run_stale" | "thread_inactive"; + const getCurrentAttemptState = Effect.fn( + "orchestrationV2.providerTurnStart.getCurrentAttemptState", + )( + function* ( + expectedStatus: OrchestrationV2Run["status"], + ): Effect.fn.Return { + const currentOption = yield* Effect.option( + projectionStore.getThreadProjection(projection.thread.id), + ); + if (Option.isNone(currentOption)) return "run_stale"; + const current = currentOption.value; + if (current.thread.archivedAt !== null || current.thread.deletedAt !== null) { + return "thread_inactive"; + } + const currentRun = current.runs.find((candidate) => candidate.id === run.id); + return currentRun?.activeAttemptId === attempt.id && currentRun.status === expectedStatus + ? "current" + : "run_stale"; + }, + Effect.catchCause(() => Effect.succeed("run_stale" as const)), + ); + const detachInactiveThreadSession = Effect.fn( + "orchestrationV2.providerTurnStart.detachInactiveThreadSession", + )(function* (detail: string) { + yield* providerSessions.detach({ + providerSessionId, + threadId: projection.thread.id, + detail, + }); + }); + const continueIfCurrent = Effect.fn("orchestrationV2.providerTurnStart.continueIfCurrent")( + function* (expectedStatus: OrchestrationV2Run["status"]) { + const state = yield* getCurrentAttemptState(expectedStatus); + if (state === "thread_inactive") { + yield* detachInactiveThreadSession( + "Provider turn start aborted because the thread was archived or deleted.", ); - }), - Effect.catchCause(() => Effect.succeed(false)), - ); - + } + return state === "current"; + }, + ); const resolvedRuntimePolicy = yield* runtimePolicy.resolve({ thread: projection.thread, modelSelection: run.modelSelection, @@ -154,6 +197,9 @@ export const layer: Layer.Layer< ? {} : { resumeFromSession: existingSessionProjection }), }); + if (!(yield* continueIfCurrent("starting"))) { + return; + } let effectiveHandoffs = handoffs; const loadedProviderThread = yield* Effect.gen(function* () { if (nativeForkTransfer !== undefined) { @@ -279,7 +325,7 @@ export const layer: Layer.Layer< }); return replacement; }); - if (!(yield* isCurrentAttemptInStatus("starting"))) { + if (!(yield* continueIfCurrent("starting"))) { return; } const now = yield* DateTime.now; @@ -401,6 +447,12 @@ export const layer: Layer.Layer< events, }); if (!runningWrite.committed) { + const state = yield* getCurrentAttemptState("starting"); + if (state === "thread_inactive") { + yield* detachInactiveThreadSession( + "Provider turn start lost a race with thread archival or deletion.", + ); + } return; } yield* runExecution.startRootRun({ @@ -427,12 +479,15 @@ export const layer: Layer.Layer< .filter((turn) => turn.providerThreadId === providerThread.id) .map((turn) => turn.ordinal), ) + 1, - shouldStartProviderTurn: () => isCurrentAttemptInStatus("running"), + shouldStartProviderTurn: () => + getCurrentAttemptState("running").pipe(Effect.map((state) => state === "current")), shouldFinalizeRun: () => projectionStore.getThreadProjection(projection.thread.id).pipe( Effect.map((current) => { const currentRun = current.runs.find((candidate) => candidate.id === run.id); return ( + current.thread.archivedAt === null && + current.thread.deletedAt === null && currentRun?.activeAttemptId === attempt.id && (currentRun.status === "starting" || currentRun.status === "running") ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index a588bb25bcd..b0ca560af88 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -272,7 +272,20 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes", const childNodeId = NodeId.make("node:run-execution-late-child:child"); const subagentNodeId = NodeId.make("node:run-execution-late-child:subagent"); const childMessageIngested = yield* Deferred.make(); + const allEventsIngested = yield* Deferred.make(); const order = yield* Ref.make>([]); + const guardedEvents = yield* Ref.make< + ReadonlyArray<{ + readonly type: ProviderAdapterV2Event["type"]; + readonly runId: RunId | undefined; + readonly activeAttemptId: RunAttemptId | undefined; + readonly expectedStatus: + | OrchestrationV2Run["status"] + | ReadonlyArray + | undefined; + readonly allowSupersededAttemptTerminalArtifacts: boolean | undefined; + }> + >([]); const testLayer = runExecutionServiceLayer.pipe( Layer.provide( Layer.mergeAll( @@ -296,6 +309,17 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes", Layer.mock(ProviderEventIngestorV2)({ ingestNormalized: (input) => Effect.gen(function* () { + yield* Ref.update(guardedEvents, (current) => [ + ...current, + { + type: input.event.type, + runId: input.writeIfRunCurrent?.runId, + activeAttemptId: input.writeIfRunCurrent?.activeAttemptId, + expectedStatus: input.writeIfRunCurrent?.expectedStatus, + allowSupersededAttemptTerminalArtifacts: + input.writeIfRunCurrent?.allowSupersededAttemptTerminalArtifacts, + }, + ]); if ( input.event.type === "message.updated" && input.event.message.threadId === childThreadId @@ -303,6 +327,12 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes", yield* Ref.update(order, (current) => [...current, "child-message"]); yield* Deferred.succeed(childMessageIngested, undefined).pipe(Effect.ignore); } + if ( + input.event.type === "subagent.updated" && + input.event.subagent.status === "completed" + ) { + yield* Deferred.succeed(allEventsIngested, undefined).pipe(Effect.ignore); + } return []; }), }), @@ -311,6 +341,18 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes", ), ); const events: ReadonlyArray = [ + { + type: "provider_turn.updated", + driver, + threadId, + providerTurn: { + id: rootProviderTurnId, + providerThreadId, + nodeId: rootNodeId, + runAttemptId: attemptId, + status: "running", + }, + } as ProviderAdapterV2Event, { type: "app_thread.created", driver, @@ -353,6 +395,18 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes", status: "running", }, } as ProviderAdapterV2Event, + { + type: "provider_turn.updated", + driver, + threadId, + providerTurn: { + id: rootProviderTurnId, + providerThreadId, + nodeId: rootNodeId, + runAttemptId: attemptId, + status: "completed", + }, + } as ProviderAdapterV2Event, { type: "turn.terminal", driver, @@ -455,7 +509,32 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes", const observed = yield* Deferred.await(childMessageIngested).pipe( Effect.timeoutOption("2 seconds"), ); + const completed = yield* Deferred.await(allEventsIngested).pipe( + Effect.timeoutOption("2 seconds"), + ); assert.isTrue(Option.isSome(observed), "child message was not ingested after root terminal"); + assert.isTrue(Option.isSome(completed), "provider event stream did not finish ingesting"); assert.deepEqual(yield* Ref.get(order), ["root-finalized", "child-message"]); + const guards = yield* Ref.get(guardedEvents); + assert.equal(guards.length, events.length); + for (const guard of guards) { + assert.equal(guard.runId, runId, `${guard.type} must retain root run ownership`); + assert.equal(guard.activeAttemptId, attemptId, `${guard.type} must retain attempt ownership`); + assert.deepEqual( + guard.expectedStatus, + ["running", "waiting"], + `${guard.type} must require an active root`, + ); + const matchingEvent = events[guards.indexOf(guard)]; + const isExactTerminalRootUpdate = + matchingEvent?.type === "provider_turn.updated" && + matchingEvent.providerTurn.runAttemptId === attemptId && + matchingEvent.providerTurn.status === "completed"; + assert.equal( + guard.allowSupersededAttemptTerminalArtifacts === true, + isExactTerminalRootUpdate, + `${guard.type} must narrowly mark only an exact terminal root update`, + ); + } }), ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index c3da3c886fd..768917e9af6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -321,7 +321,12 @@ export const layer: Layer.Layer< // it terminalized this attempt as superseded. Preserve that domain // status while retaining the provider's interruption artifact. if (input.terminal.status === "interrupted") { - yield* eventSink.write({ + yield* eventSink.writeIfRunCurrent({ + threadId: input.run.threadId, + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedStatus: ["running", "waiting"], + allowSupersededAttemptTerminalArtifacts: true, events: [ { id: yield* idAllocator.allocate.event({ threadId: input.run.threadId }), @@ -648,16 +653,16 @@ export const layer: Layer.Layer< runId: input.run.id, nodeId: input.rootNode.id, event, - ...(event.type === "provider_thread.updated" && - event.providerThread.id === input.providerThread.id - ? { - writeIfRunCurrent: { - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedStatus: "running" as const, - }, - } - : {}), + writeIfRunCurrent: { + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedStatus: ["running", "waiting"], + ...(event.type === "provider_turn.updated" && + event.providerTurn.runAttemptId === input.attempt.id && + isTerminalProviderTurnStatus(event.providerTurn.status) + ? { allowSupersededAttemptTerminalArtifacts: true } + : {}), + }, }); storedEventCount = storedEvents.length; } diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 6f7420d3624..6ebebefde50 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -266,6 +266,291 @@ it.layer(TestLayer)("OrchestrationV2LayerLive", (it) => { }), ); + it.effect("cascades owned subagent lifecycle without crossing independent threads", () => + Effect.gen(function* () { + const applicationEngine = yield* OrchestrationEngineService; + const orchestrator = yield* OrchestratorV2; + const sql = yield* SqlClient.SqlClient; + const projectId = ProjectId.make("runtime-layer-subagent-lifecycle-project"); + const rootThreadId = ThreadId.make("runtime-layer-subagent-lifecycle-root"); + const independentThreadId = ThreadId.make("runtime-layer-subagent-lifecycle-independent"); + + yield* applicationEngine.dispatch({ + type: "project.create", + commandId: CommandId.make("runtime-layer-subagent-lifecycle-project-create"), + projectId, + title: "Subagent lifecycle project", + workspaceRoot: "/tmp/runtime-layer-subagent-lifecycle-project", + defaultModelSelection: modelSelection, + scripts: [], + createdAt: "2026-07-09T00:00:00.000Z", + }); + + const createThread = (threadId: ThreadId, title: string) => + orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:create:${threadId}`), + threadId, + projectId, + title, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + const startRun = (threadId: ThreadId, suffix: string) => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:message:${suffix}`), + threadId, + messageId: MessageId.make(`message:${suffix}`), + text: `Run ${suffix}`, + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + + yield* createThread(rootThreadId, "Lifecycle root"); + yield* createThread(independentThreadId, "Independent thread"); + yield* startRun(rootThreadId, "root"); + const rootBeforeDelegate = yield* orchestrator.getThreadProjection(rootThreadId); + const rootRun = rootBeforeDelegate.runs[0]; + const rootNode = rootBeforeDelegate.nodes.find((node) => node.kind === "root_turn"); + assert.isDefined(rootRun); + assert.isDefined(rootNode); + + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:delegate:child"), + parentThreadId: rootThreadId, + parentRunId: rootRun.id, + parentNodeId: rootNode.id, + task: "Create the child result", + title: "Owned child", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + }); + const rootAfterDelegate = yield* orchestrator.getThreadProjection(rootThreadId); + const childThreadId = rootAfterDelegate.subagents[0]?.childThreadId; + assert.isNotNull(childThreadId); + assert.isDefined(childThreadId); + + const childBeforeDelegate = yield* orchestrator.getThreadProjection(childThreadId); + const childRun = childBeforeDelegate.runs[0]; + const childNode = childBeforeDelegate.nodes.find((node) => node.kind === "root_turn"); + assert.isDefined(childRun); + assert.isDefined(childNode); + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:delegate:grandchild"), + parentThreadId: childThreadId, + parentRunId: childRun.id, + parentNodeId: childNode.id, + task: "Create the grandchild result", + title: "Owned grandchild", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + }); + const childAfterDelegate = yield* orchestrator.getThreadProjection(childThreadId); + const grandchildThreadId = childAfterDelegate.subagents[0]?.childThreadId; + assert.isNotNull(grandchildThreadId); + assert.isDefined(grandchildThreadId); + + yield* orchestrator.dispatch({ + type: "delegated_task.request", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:delegate:deleted-grandchild"), + parentThreadId: childThreadId, + parentRunId: childRun.id, + parentNodeId: childNode.id, + task: "Create the grandchild that will be deleted directly", + title: "Deleted grandchild", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + }); + const childWithDeletedGrandchild = yield* orchestrator.getThreadProjection(childThreadId); + const deletedGrandchildTask = childWithDeletedGrandchild.subagents.find( + (subagent) => subagent.childThreadId !== grandchildThreadId, + ); + const deletedGrandchildThreadId = deletedGrandchildTask?.childThreadId; + assert.isDefined(deletedGrandchildTask); + assert.isNotNull(deletedGrandchildThreadId); + assert.isDefined(deletedGrandchildThreadId); + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make("command:delete:grandchild-branch"), + threadId: deletedGrandchildThreadId, + }); + const childAfterGrandchildDelete = yield* orchestrator.getThreadProjection(childThreadId); + assert.equal( + childAfterGrandchildDelete.subagents.find( + (subagent) => subagent.id === deletedGrandchildTask.id, + )?.status, + "cancelled", + ); + assert.equal( + childAfterGrandchildDelete.nodes.find((node) => node.id === deletedGrandchildTask.id) + ?.status, + "cancelled", + ); + assert.equal( + childAfterGrandchildDelete.turnItems.find( + (item) => item.type === "subagent" && item.subagentId === deletedGrandchildTask.id, + )?.status, + "cancelled", + ); + + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive:grandchild"), + threadId: grandchildThreadId, + }); + const legacyParentDeletedAt = "2026-07-09T01:00:00.000Z"; + yield* sql` + UPDATE orchestration_v2_projection_threads + SET + deleted_at = ${legacyParentDeletedAt}, + payload_json = json_set(payload_json, '$.deletedAt', ${legacyParentDeletedAt}) + WHERE thread_id = ${childThreadId} + `; + const deletedParentUnarchiveError = yield* orchestrator + .dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("command:unarchive:deleted-parent-child"), + threadId: grandchildThreadId, + }) + .pipe(Effect.flip); + assert.equal(deletedParentUnarchiveError._tag, "OrchestratorDispatchError"); + yield* sql` + UPDATE orchestration_v2_projection_threads + SET + deleted_at = NULL, + payload_json = json_set(payload_json, '$.deletedAt', NULL) + WHERE thread_id = ${childThreadId} + `; + const rootArchive = yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive:root-subtree"), + threadId: rootThreadId, + }); + + const archivedRoot = yield* orchestrator.getThreadProjection(rootThreadId); + const archivedChild = yield* orchestrator.getThreadProjection(childThreadId); + const archivedGrandchild = yield* orchestrator.getThreadProjection(grandchildThreadId); + assert.isNotNull(archivedRoot.thread.archivedAt); + assert.deepEqual(archivedChild.thread.archivedAt, archivedRoot.thread.archivedAt); + assert.notDeepEqual(archivedGrandchild.thread.archivedAt, archivedRoot.thread.archivedAt); + assert.equal(archivedRoot.runs[0]?.status, "cancelled"); + assert.equal(archivedChild.runs[0]?.status, "cancelled"); + assert.equal(archivedGrandchild.runs[0]?.status, "cancelled"); + assert.equal(archivedRoot.subagents[0]?.status, "cancelled"); + assert.equal( + archivedRoot.nodes.find((node) => node.kind === "subagent")?.status, + "cancelled", + ); + assert.equal( + archivedRoot.turnItems.find((item) => item.type === "subagent")?.status, + "cancelled", + ); + assert.equal(archivedChild.subagents[0]?.status, "cancelled"); + assert.deepEqual( + rootArchive.storedEvents + .filter((stored) => stored.event.type === "thread.archived") + .map((stored) => stored.event.threadId) + .toSorted(), + [childThreadId, rootThreadId].toSorted(), + ); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + yield* Effect.yieldNow; + assert.isFalse( + (yield* orchestrator.getThreadProjection(rootThreadId)).contextTransfers.some( + (transfer) => transfer.type === "subagent_result", + ), + ); + + const childUnarchiveError = yield* orchestrator + .dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("command:unarchive:managed-child"), + threadId: childThreadId, + }) + .pipe(Effect.flip); + assert.equal(childUnarchiveError._tag, "OrchestratorDispatchError"); + + const grandchildUnarchiveError = yield* orchestrator + .dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("command:unarchive:independently-archived-grandchild"), + threadId: grandchildThreadId, + }) + .pipe(Effect.flip); + assert.equal(grandchildUnarchiveError._tag, "OrchestratorDispatchError"); + + yield* orchestrator.dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("command:unarchive:root-subtree"), + threadId: rootThreadId, + }); + assert.isNull((yield* orchestrator.getThreadProjection(rootThreadId)).thread.archivedAt); + assert.isNull((yield* orchestrator.getThreadProjection(childThreadId)).thread.archivedAt); + assert.isNotNull( + (yield* orchestrator.getThreadProjection(grandchildThreadId)).thread.archivedAt, + ); + yield* startRun(childThreadId, "child-after-unarchive"); + assert.equal( + (yield* orchestrator.getThreadProjection(childThreadId)).runs.at(-1)?.status, + "starting", + ); + + const deletion = yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make("command:delete:root-subtree"), + threadId: rootThreadId, + }); + assert.deepEqual( + deletion.storedEvents + .filter((stored) => stored.event.type === "thread.deleted") + .map((stored) => stored.event.threadId) + .toSorted(), + [rootThreadId, childThreadId, grandchildThreadId].toSorted(), + ); + for (const threadId of [rootThreadId, childThreadId, grandchildThreadId]) { + assert.isNotNull((yield* orchestrator.getThreadProjection(threadId)).thread.deletedAt); + } + yield* Effect.yieldNow; + yield* Effect.yieldNow; + yield* Effect.yieldNow; + assert.isFalse( + (yield* orchestrator.getThreadProjection(rootThreadId)).contextTransfers.some( + (transfer) => transfer.type === "subagent_result", + ), + ); + const shell = yield* orchestrator.getShellSnapshot(); + assert.include( + [...shell.threads, ...shell.archivedThreads].map((thread) => thread.id), + independentThreadId, + ); + assert.notInclude( + [...shell.threads, ...shell.archivedThreads].map((thread) => thread.id), + rootThreadId, + ); + }).pipe(Effect.provide(SharedApplicationDataPlaneTestLayer)), + ); + it.effect("persists rejected command receipts across retries", () => Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 33a3b796a51..f62fac230dd 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -7,6 +7,7 @@ import { ProjectionProjectRepositoryLive } from "../persistence/Layers/Projectio import { layer as projectServiceLayer } from "../project/ProjectService.ts"; import { layer as projectSetupScriptRunnerLayer } from "../project/ProjectSetupScriptRunner.ts"; import { layer as checkpointCaptureServiceLayer } from "./CheckpointCaptureService.ts"; +import { layer as checkpointCleanupServiceLayer } from "./CheckpointCleanupService.ts"; import { layer as checkpointServiceLayer } from "./CheckpointService.ts"; import { layer as checkpointRollbackServiceLayer } from "./CheckpointRollbackService.ts"; import { layer as commandPolicyLayer } from "./CommandPolicy.ts"; @@ -72,6 +73,9 @@ const providerEventIngestorProvided = providerEventIngestorLayer.pipe( ); const checkpointServiceProvided = checkpointServiceLayer.pipe(Layer.provide(idAllocatorLayer)); +const checkpointCleanupServiceProvided = checkpointCleanupServiceLayer.pipe( + Layer.provide(Layer.merge(checkpointServiceProvided, projectionStoreLayer)), +); const contextHandoffServiceProvided = contextHandoffServiceLayer.pipe( Layer.provide(idAllocatorLayer), ); @@ -153,6 +157,7 @@ const effectExecutorProvided = effectExecutorLayer.pipe( Layer.provide( Layer.mergeAll( runFinalizationServiceProvided, + checkpointCleanupServiceProvided, checkpointRollbackServiceProvided, providerSessionManagerProvided, providerTurnControlServiceProvided, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index b2f5c2cb444..e66975490d0 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -17,6 +17,7 @@ import { layer as mcpSessionRegistryTestLayer } from "../../mcp/McpSessionRegist import * as VcsDriverRegistry from "../../vcs/VcsDriverRegistry.ts"; import * as VcsProcess from "../../vcs/VcsProcess.ts"; import { layer as checkpointCaptureServiceLayer } from "../CheckpointCaptureService.ts"; +import { layer as checkpointCleanupServiceLayer } from "../CheckpointCleanupService.ts"; import { layer as checkpointServiceLayer } from "../CheckpointService.ts"; import { layer as checkpointRollbackServiceLayer } from "../CheckpointRollbackService.ts"; import { layer as commandPolicyLayer } from "../CommandPolicy.ts"; @@ -257,6 +258,9 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( const checkpointServiceProvided = checkpointServiceLayer.pipe( Layer.provide(Layer.mergeAll(checkpointStoreLayer, idAllocatorLayer)), ); + const checkpointCleanupServiceProvided = checkpointCleanupServiceLayer.pipe( + Layer.provide(Layer.merge(checkpointServiceProvided, storesLayer)), + ); const contextHandoffServiceProvided = contextHandoffServiceLayer.pipe( Layer.provide(idAllocatorLayer), ); @@ -337,6 +341,7 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( Layer.provide( Layer.mergeAll( runFinalizationServiceProvided, + checkpointCleanupServiceProvided, checkpointRollbackServiceProvided, providerSessionManagerProvided, providerTurnControlServiceProvided,