From 75945890aff0341a488917729d8fb028bbe44729 Mon Sep 17 00:00:00 2001 From: Chinoman10 <8300763+Chinoman10@users.noreply.github.com> Date: Thu, 5 Nov 2026 19:26:22 +0000 Subject: [PATCH 1/9] fix(server): retain native start ownership across Stop acceptance Linearize exact-attempt native ownership with Stop on the existing thread lock. Keep native start and ingestion service-scoped when the outbox wait is cancelled, and await exact native interrupt plus terminal confirmation. Scope start/restart cancellation to the target run. Preserve interrupt failures, timeout markers, explicit ForceStop revocation and missing-identity settlement without changing the provider manager API. Verify the accepted red witness, nine deterministic synthetic race cases, 177 focused tests and check:changed (331 tests and all affected checks). No live activation or publication. Model: openai-codex/gpt-6.1-sol. Harness: Pi. --- .../src/orchestration-v2/EffectOutbox.ts | 10 +- .../src/orchestration-v2/EffectWorker.test.ts | 2 + .../src/orchestration-v2/EffectWorker.ts | 46 ++ apps/server/src/orchestration-v2/EventSink.ts | 1 + .../FoundationPersistence.test.ts | 21 +- .../src/orchestration-v2/Orchestrator.ts | 146 ++++- .../ProviderStartStop.integration.test.ts | 606 ++++++++++++++++++ .../ProviderTurnStartService.ts | 33 +- .../RunExecutionService.test.ts | 28 +- .../orchestration-v2/RunExecutionService.ts | 412 +++++++++--- .../RunFinalizationQueue.test.ts | 2 + .../ThreadManagementService.test.ts | 25 +- .../ThreadManagementService.ts | 13 +- .../src/orchestration-v2/runtimeLayer.ts | 4 + .../testkit/ProviderReplayHarness.ts | 5 + packages/contracts/src/orchestrationV2.ts | 15 + 16 files changed, 1242 insertions(+), 127 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 92e1e01ad43f2..293ec2a0ed3b9 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -44,6 +44,11 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ type: Schema.Literal("provider-turn.start"), runId: RunId, }), + Schema.Struct({ + type: Schema.Literal("provider-turn.interrupt-start"), + runId: RunId, + attemptId: RunAttemptId, + }), Schema.Struct({ type: Schema.Literal("provider-turn.interrupt"), providerSessionId: ProviderSessionId, @@ -138,6 +143,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ export const PROCESS_BOUND_EFFECT_TYPES = [ "provider-turn.start", "provider-turn.interrupt", + "provider-turn.interrupt-start", "provider-turn.steer", "provider-turn.restart", "runtime-request.respond", @@ -209,6 +215,7 @@ export interface EffectOutboxV2Shape { ) => Effect.Effect, EffectOutboxError>; readonly cancelUnsettled: (input: { readonly threadId: ThreadId; + readonly runId?: RunId; readonly effectTypes: ReadonlyArray; readonly reason: string; }) => Effect.Effect, EffectOutboxError>; @@ -437,7 +444,7 @@ export const layer: Layer.Layer = La ), ), cancelUnsettled: Effect.fn("EffectOutbox.cancelUnsettled")( - function* ({ threadId, effectTypes, reason }) { + function* ({ threadId, runId, effectTypes, reason }) { if (effectTypes.length === 0) return []; const now = DateTime.formatIso(yield* DateTime.now); const rows = yield* sql<{ readonly effect_id: string }>` @@ -451,6 +458,7 @@ export const layer: Layer.Layer = La WHERE thread_id = ${threadId} AND status IN ('pending', 'running') AND effect_type IN ${sql.in(effectTypes)} + AND (${runId ?? null} IS NULL OR json_extract(payload_json, '$.runId') = ${runId ?? null}) RETURNING effect_id `; return rows.map(({ effect_id }) => effect_id); diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 2329f2aa20cf0..572cf44588b03 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -30,6 +30,7 @@ import * as RunFinalizationService from "./RunFinalizationService.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderTurnControlService from "./ProviderTurnControlService.ts"; import * as ProviderTurnStartService from "./ProviderTurnStartService.ts"; +import * as RunExecutionService from "./RunExecutionService.ts"; import * as RuntimeRequestService from "./RuntimeRequestService.ts"; import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; @@ -87,6 +88,7 @@ function layerExecutorFor(input: { }) { const record = (event: string) => Ref.update(input.events, (events) => [...events, event]); const layerDependencies = Layer.mergeAll( + Layer.mock(RunExecutionService.RunExecutionServiceV2)({}), Layer.succeed( ProviderTurnControlService.ProviderTurnControlServiceV2, ProviderTurnControlService.ProviderTurnControlServiceV2.of({ diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index c5b5ea9b383a9..1a33a2e829ddc 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -23,6 +23,7 @@ import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderTurnControlService from "./ProviderTurnControlService.ts"; import * as ProviderTurnStartService from "./ProviderTurnStartService.ts"; +import * as RunExecutionService from "./RunExecutionService.ts"; import * as RuntimeRequestService from "./RuntimeRequestService.ts"; import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; @@ -76,6 +77,7 @@ export const layerExecutor: Layer.Layer< | CheckpointRollbackService.CheckpointRollbackServiceV2 | ProviderTurnControlService.ProviderTurnControlServiceV2 | ProviderTurnStartService.ProviderTurnStartServiceV2 + | RunExecutionService.RunExecutionServiceV2 | RuntimeRequestService.RuntimeRequestServiceV2 | ThreadTitleRegenerationService.ThreadTitleRegenerationService | ThreadManagementService.ThreadManagementService @@ -89,6 +91,7 @@ export const layerExecutor: Layer.Layer< const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2; const providerTurnControl = yield* ProviderTurnControlService.ProviderTurnControlServiceV2; const providerTurnStart = yield* ProviderTurnStartService.ProviderTurnStartServiceV2; + const runExecution = yield* RunExecutionService.RunExecutionServiceV2; const runtimeRequests = yield* RuntimeRequestService.RuntimeRequestServiceV2; const threadTitleRegeneration = yield* ThreadTitleRegenerationService.ThreadTitleRegenerationService; @@ -151,6 +154,49 @@ export const layerExecutor: Layer.Layer< }), ), ); + case "provider-turn.interrupt-start": { + const request = effect.request; + return runExecution + .interruptOwnedStart({ + threadId: effect.threadId, + runId: request.runId, + attemptId: request.attemptId, + }) + .pipe( + Effect.map((target) => ({ unresponsive: false, target }) as const), + Effect.catchIf( + (cause) => cause.unresponsive, + () => Effect.succeed({ unresponsive: true, target: undefined } as const), + ), + Effect.andThen(({ unresponsive, target }) => + unresponsive + ? threads.dispatch({ + type: "run.interrupt.unresponsive", + commandId: CommandId.make(`${effect.id}:unresponsive`), + threadId: effect.threadId, + runId: request.runId, + attemptId: request.attemptId, + }) + : target === undefined + ? Effect.void + : threads.dispatch({ + type: "thread.background-work.settle", + commandId: CommandId.make(`${effect.id}:background-work-settled`), + threadId: effect.threadId, + providerThreadId: target.providerThreadId, + providerTurnId: target.providerTurnId, + }), + ), + Effect.mapError( + (cause) => + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: request.type, + cause, + }), + ), + ); + } case "provider-turn.interrupt": { const request = effect.request; return providerTurnControl diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index b210d3c9840ae..45de4fb75e11c 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -143,6 +143,7 @@ export interface EventSinkV2Shape { readonly events: ReadonlyArray; readonly effects: ReadonlyArray; readonly cancelUnsettledEffects?: { + readonly runId?: RunId; readonly effectTypes: ReadonlyArray; readonly reason: string; }; diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 475e0a0e4ae7b..2d47bddf9e23c 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -1845,12 +1845,31 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; const workerFiber = yield* worker.runOnce.pipe(Effect.forkChild); yield* Deferred.await(started); + const replacementId = `${effectId}:replacement`; + yield* outbox.enqueue([ + { + id: replacementId, + commandId, + threadId, + request: { type: "provider-turn.start", runId: RunId.make("run:replacement") }, + }, + ]); const cancelledEffectIds = yield* outbox.cancelUnsettled({ threadId, - effectTypes: ["provider-turn.start"], + runId: RunId.make("run:foundation-cancel-running-effect"), + effectTypes: ["provider-turn.start", "provider-turn.restart"], reason: "The owning run was interrupted.", }); assert.deepEqual(cancelledEffectIds, [effectId]); + assert.equal(Option.getOrThrow(yield* outbox.get(replacementId)).status, "pending"); + yield* outbox.signalCancellations( + yield* outbox.cancelUnsettled({ + threadId, + runId: RunId.make("run:replacement"), + effectTypes: ["provider-turn.start"], + reason: "Test cleanup.", + }), + ); yield* outbox.signalCancellations(cancelledEffectIds); assert.isTrue(yield* Fiber.join(workerFiber)); yield* Deferred.await(interrupted); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a6b876db1de96..34721012b5c9f 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -604,6 +604,8 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "checkpoint.rollback.fail": case "thread.background-work.settle": case "thread.interrupt.unresponsive": + case "run.interrupt.unresponsive": + case "run.force-stop.settle": case "thread.stop": case "provider.switch": return command.threadId; @@ -1069,6 +1071,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const runtimePolicy = yield* RuntimePolicyV2; const threadForkService = yield* ThreadForkServiceV2; const threadDispatch = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const runExecution = yield* RunExecutionService.RunExecutionServiceV2; // Read once: the policy is fixed for the server's life, and dispatch takes it as an input. const { managedPolicy } = serverSettings; const runtimeModeCeiling: RuntimeModeCeiling | undefined = @@ -11014,22 +11017,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio function* ( command: Extract< OrchestrationV2InternalCommand, - { readonly type: "thread.interrupt.unresponsive" } + { readonly type: "thread.interrupt.unresponsive" | "run.interrupt.unresponsive" } >, events: Ref.Ref>, ) { - const { providerTurn } = yield* projectionStore - .getProviderControlContext(command.threadId, { - providerThreadId: command.providerThreadId, - providerTurnId: command.providerTurnId, - }) - .pipe(mapDispatchError(command)); + const providerTurn = + command.type === "thread.interrupt.unresponsive" + ? (yield* projectionStore + .getProviderControlContext(command.threadId, { + providerThreadId: command.providerThreadId, + providerTurnId: command.providerTurnId, + }) + .pipe(mapDispatchError(command))).providerTurn + : undefined; const { runs } = yield* projectionStore .getThreadRecords(command.threadId, ["runs"]) .pipe(mapDispatchError(command)); - const run = runs.find( - (candidate) => - providerTurn !== undefined && candidate.activeAttemptId === providerTurn.runAttemptId, + const run = runs.find((candidate) => + command.type === "run.interrupt.unresponsive" + ? candidate.id === command.runId && candidate.activeAttemptId === command.attemptId + : providerTurn !== undefined && candidate.activeAttemptId === providerTurn.runAttemptId, ); // A settled run still counts: Stop may have targeted its background work. if (run === undefined) return; @@ -11322,10 +11329,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); const dispatchRunInterrupt = Effect.fn("Orchestrator.dispatchRunInterrupt")(function* ( - command: Extract, + incoming: Extract< + OrchestrationV2ServerCommand, + { readonly type: "run.interrupt" | "run.force-stop.settle" } + >, events: Ref.Ref>, effects: Ref.Ref>, ) { + const command: Extract = + incoming.type === "run.force-stop.settle" ? { ...incoming, type: "run.interrupt" } : incoming; const projection = yield* loadProjectionForCommand( command, [ @@ -11390,6 +11402,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Run ${command.runId} is not interruptible.`, }); } + if ( + incoming.type === "run.force-stop.settle" && + (run.activeAttemptId !== incoming.attemptId || + !["preparing", "starting", "running"].includes(run.status)) + ) + return undefined; + const ownership = + run.activeAttemptId === null + ? Option.none() + : yield* runExecution.getStartOwnership({ + threadId: command.threadId, + runId: run.id, + attemptId: run.activeAttemptId, + }); + const interruptOwnedStart = + incoming.type !== "run.force-stop.settle" && + Option.isSome(ownership) && + (providerTurn === undefined || ownership.value.startPending); const now = yield* DateTime.now; const completionMessage = projection.messages.find( (candidate) => candidate.id === run.userMessageId, @@ -11494,6 +11524,84 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } } + if (interruptOwnedStart && run.activeAttemptId !== null) { + const session = + providerThread.providerSessionId === null + ? Option.none() + : yield* providerSessions + .get(providerThread.providerSessionId) + .pipe(mapDispatchError(command)); + if (Option.isNone(session)) + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Owned start has no live provider session.", + }); + yield* mapDispatchError(command)( + commandPolicy.ensureInterrupt({ + commandId: command.commandId, + threadId: command.threadId, + providerInstanceId: run.providerInstanceId, + capabilities: session.value.providerSession.capabilities, + }), + ); + yield* emitEvent({ + type: "turn-item.updated", + threadId: command.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: interruptRequestItem, + }); + yield* stopRemainingWork(); + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:provider-turn.interrupt-start:${run.activeAttemptId}`, + commandId: command.commandId, + threadId: command.threadId, + request: { + type: "provider-turn.interrupt-start", + runId: run.id, + attemptId: run.activeAttemptId!, + }, + } satisfies PendingOrchestrationEffectV2, + ...otherProviderInterrupts, + ]); + return { + runId: run.id, + effectTypes: ["provider-turn.start", "provider-turn.restart"] as const, + reason: `Stop requested for owned start ${run.id}.`, + }; + } + + if (incoming.type === "run.force-stop.settle" && providerTurn !== undefined) { + yield* emitEvent({ + type: "turn-item.updated", + threadId: command.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: interruptRequestItem, + }); + yield* settleInterruptedRun({ command, projection, providerTurn, events, effects, now }); + yield* settleBackgroundWork({ + command, + events, + projection, + stoppedProviderThreadId: providerThread.id, + throughRunOrdinal: run.ordinal, + now, + }); + return { + runId: run.id, + effectTypes: ["provider-turn.start", "provider-turn.restart"] as const, + reason: incoming.reason, + }; + } + if ( providerTurn === undefined && (run.status === "preparing" || run.status === "starting" || run.status === "running") @@ -11525,7 +11633,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio completedAt: now, updatedAt: now, type: "run_interrupt_result", - message: "Run interrupted before provider start", + message: + incoming.type === "run.force-stop.settle" + ? incoming.reason + : "Run interrupted before provider start", }; yield* emitEvent({ type: "turn-item.updated", @@ -11641,9 +11752,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* stopRemainingWork(); yield* Ref.update(effects, (existing) => [...existing, ...otherProviderInterrupts]); return { + runId: run.id, effectTypes: ["provider-turn.start", "provider-turn.restart"], reason: `Run ${run.id} was interrupted before its provider turn started.`, } satisfies { + readonly runId?: RunId; readonly effectTypes: ReadonlyArray; readonly reason: string; }; @@ -13122,6 +13235,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly events: ReadonlyArray; readonly effects: ReadonlyArray; readonly cancelUnsettledEffects?: { + readonly runId?: RunId; readonly effectTypes: ReadonlyArray; readonly reason: string; }; @@ -13139,6 +13253,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const effects = yield* Ref.make>([]); let cancelUnsettledEffects: | { + readonly runId?: RunId; readonly effectTypes: ReadonlyArray; readonly reason: string; } @@ -13339,6 +13454,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "subagent.stop": yield* dispatchSubagentStop(command, events, effects); break; + case "run.force-stop.settle": + cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects); + break; case "run.interrupt": cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects); if (command.holdQueue === true) { @@ -13478,6 +13596,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.background-work.settle": yield* dispatchBackgroundWorkSettle(command, events, effects); break; + case "run.interrupt.unresponsive": case "thread.interrupt.unresponsive": yield* dispatchInterruptUnresponsive(command, events); break; @@ -13618,6 +13737,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio planned.events.length > 0 || command.type === "thread.background-work.settle" || command.type === "thread.interrupt.unresponsive" || + command.type === "run.interrupt.unresponsive" || + command.type === "run.force-stop.settle" || command.type === "thread.stop" || command.type === "subagent.stop" || // Closing a share again, or after it ended, is not an error. @@ -14337,6 +14458,7 @@ export const layer: Layer.Layer< | ProjectStore.ProjectStoreV2 | ProviderAdapterRegistryV2 | ProviderSessionManagerV2 + | RunExecutionService.RunExecutionServiceV2 | ProviderSwitchServiceV2 | ProjectionStoreV2 | RuntimePolicyV2 diff --git a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts new file mode 100644 index 0000000000000..034cfc8eaa6a8 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts @@ -0,0 +1,606 @@ +import { assert, it, vi } from "@effect/vitest"; +import { + CommandId, + MessageId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + RunAttemptId, + ThreadId, + type OrchestrationV2ProviderTurn, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Config from "effect/Config"; +import * as Logger from "effect/Logger"; +import * as References from "effect/References"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import * as EffectOutbox from "./EffectOutbox.ts"; +import * as EffectWorker from "./EffectWorker.ts"; +import * as EventSink from "./EventSink.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as RunExecutionService from "./RunExecutionService.ts"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; + +it.effect.each([ + "succeeded", + "pending-ingestion", + "accepted-start-error", + "queued-promotion", + "before-ownership", + "interrupt-error", + "interrupt-timeout", + "identity-timeout", + "identity-timeout-force", +] as const)("Stop crosses native start: %s", (scenario) => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace(`start-stop-${scenario}`); + const events = yield* Queue.unbounded(); + const entered = yield* Deferred.make(); + const accept = yield* Deferred.make(); + const nativeAccepted = yield* Deferred.make(); + const releaseIdentity = yield* Deferred.make(); + const releaseStart = yield* Deferred.make(); + const snapshotted = yield* Deferred.make(); + const releaseSnapshot = yield* Deferred.make(); + const ingestionHeld = yield* Deferred.make(); + const releaseIngestion = yield* Deferred.make(); + const interruptCalled = yield* Deferred.make(); + const interruptCancelled = yield* Deferred.make(); + const releaseTerminal = yield* Deferred.make(); + const ownedWait = yield* Deferred.make(); + const beforeOwnership = yield* Deferred.make(); + const driver = ProviderDriverKind.make("codex"); + const instanceId = ProviderInstanceId.make("codex"); + const modelSelection = { instanceId, model: "synthetic" }; + const threadId = ThreadId.make(`thread:start-stop:${scenario}`); + const interrupts: ProviderTurnId[] = []; + const milestones: string[] = []; + let active: OrchestrationV2ProviderTurn | undefined; + let starts = 0; + const startPending = + scenario !== "succeeded" && + scenario !== "queued-promotion" && + scenario !== "accepted-start-error"; + const identityDelayed = + scenario === "identity-timeout" || scenario === "identity-timeout-force"; + const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { + instanceId, + driver, + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (input) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const startTurn: ProviderAdapter.ProviderAdapterV2SessionRuntime["startTurn"] = ( + turn, + ) => + Effect.gen(function* () { + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(accept); + assert.isUndefined(active, "native turns must never overlap"); + starts++; + active = { + id: ProviderTurnId.make(`native-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: { + driver, + nativeId: `native:${turn.attemptId}`, + strength: "strong", + }, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }; + milestones.push(`accepted:${active.id}`); + yield* Deferred.succeed(nativeAccepted, undefined); + if (identityDelayed) yield* Deferred.await(releaseIdentity); + // Another attempt and a sibling on this same session cannot supply our identity. + if (starts === 1) { + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + ...active, + id: ProviderTurnId.make("foreign:old"), + runAttemptId: RunAttemptId.make("foreign:old"), + }, + }); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + ...active, + id: ProviderTurnId.make("foreign:sibling"), + providerThreadId: ProviderThreadId.make("foreign:sibling"), + runAttemptId: RunAttemptId.make("foreign:sibling"), + }, + }); + } + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: active, + }); + if (scenario === "accepted-start-error" && starts === 1) + return yield* new ProviderAdapter.ProviderAdapterTurnStartError({ + driver, + threadId: turn.threadId, + providerThreadId: turn.providerThread.id, + runId: turn.runId, + cause: "synthetic lost start response after acceptance", + }); + if (startPending && starts === 1) yield* Deferred.await(releaseStart); + }); + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd, + model: "synthetic", + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.fromQueue(events), + ensureThread: ({ threadId }) => + Effect.succeed({ + id: ProviderThreadId.make(`provider:${threadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: input.providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: "synthetic-thread", strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn, + compactThread: startTurn, + interruptTurn: (target) => + Effect.gen(function* () { + interrupts.push(target.providerTurnId); + assert.equal( + target.providerTurnId, + active?.id, + "interrupt must target the accepted native identity", + ); + assert.isTrue(target.requestRuntimeRestart); + milestones.push(`interrupt:${target.providerTurnId}`); + yield* Deferred.succeed(interruptCalled, undefined); + if (scenario === "interrupt-error") + return yield* new ProviderAdapter.ProviderAdapterInterruptError({ + driver, + providerThreadId: target.providerThread.id, + providerTurnId: target.providerTurnId, + cause: "synthetic interrupt rejection", + }); + if (scenario === "interrupt-timeout") + return yield* Effect.never.pipe( + Effect.ensuring(Deferred.succeed(interruptCancelled, undefined)), + ); + yield* Deferred.await(releaseTerminal); + const turn = active!; + active = undefined; + const completedAt = yield* DateTime.now; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "interrupted", completedAt }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: 1, + status: "interrupted", + failure: null, + threadDisposition: "reusable", + }); + }), + steerTurn: () => Effect.die("unused"), + respondToRuntimeRequest: () => Effect.die("unused"), + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: () => Effect.die("unused"), + forkThread: () => Effect.die("unused"), + }; + }), + }; + const databaseLayer = SqlitePersistence.layerMemory; + const base = Layer.mergeAll( + ProviderReplayHarness.layerWithRegistry( + { name: `start-stop-${scenario}` }, + ProviderAdapterRegistry.layerFromAdapters([adapter]), + { runEffectWorker: false, databaseLayer }, + ), + ProjectionStore.layer.pipe(Layer.provide(databaseLayer)), + EffectOutbox.layer.pipe(Layer.provide(databaseLayer)), + CommandReceiptStore.layer.pipe(Layer.provide(databaseLayer)), + ServerSettings.layerTest(), + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const store = yield* ProjectionStore.ProjectionStoreV2; + const sink = yield* EventSink.EventSinkV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const execution = yield* RunExecutionService.RunExecutionServiceV2; + const threads = yield* ThreadManagementService.ThreadManagementService; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:start-stop"), + title: "Synthetic acceptance race", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const send = (id: string, queued = false) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(id), + text: queued ? "/goal ship it" : id, + attachments: [], + dispatchMode: { type: queued ? "queue_after_active" : "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const originalStart = execution.startRootRun; + let staleInput: RunExecutionService.RunExecutionServiceV2StartRootRunInput | undefined; + using _beforeOwnership = vi.spyOn(execution, "startRootRun").mockImplementation((input) => { + if (scenario !== "before-ownership") return originalStart(input); + staleInput = input; + return Deferred.succeed(beforeOwnership, undefined).pipe(Effect.andThen(Effect.never)); + }); + const originalWrite = sink.write; + using _ingestion = vi.spyOn(sink, "write").mockImplementation((input) => { + const hold = + scenario === "pending-ingestion" && + input.events.some( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ); + return ( + hold + ? Deferred.succeed(ingestionHeld, undefined).pipe( + Effect.andThen(Deferred.await(releaseIngestion)), + ) + : Effect.void + ).pipe(Effect.andThen(originalWrite(input))); + }); + yield* send("first"); + const startup = yield* worker.drain(1).pipe(Effect.forkScoped); + if (scenario === "before-ownership") { + yield* Deferred.await(beforeOwnership); + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("stop"), + threadId, + }); + yield* Fiber.join(startup); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs[0]?.status, + "interrupted", + ); + yield* originalStart(staleInput!); + assert.equal(starts, 0); + return; + } + yield* Deferred.await(entered); + milestones.push("entered"); + if (scenario === "queued-promotion") yield* send("promoted", true); + const initial = yield* orchestrator.getThreadProjection(threadId); + const run = initial.runs.find((run) => run.status === "running")!; + const original = store.getThreadRecords; + let armed = true; + using _snapshot = vi.spyOn(store, "getThreadRecords").mockImplementation((...args) => + original(...args).pipe( + Effect.tap((snapshot) => { + if ( + !armed || + !args[1].includes("providerTurns") || + !args[1].includes("attempts") || + args[1].includes("providerSessions") || + snapshot.providerTurns.length !== 0 + ) + return Effect.void; + armed = false; + return Deferred.succeed(snapshotted, undefined).pipe( + Effect.andThen(Deferred.await(releaseSnapshot)), + ); + }), + ), + ); + const stop = yield* ( + scenario === "queued-promotion" + ? orchestrator.dispatch({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make("stop"), + threadId, + queuedRunId: initial.runs.find((run) => run.status === "queued")!.id, + targetRunId: run.id, + }) + : orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("stop"), + threadId, + }) + ).pipe(Effect.forkScoped); + yield* Deferred.await(snapshotted); + milestones.push("snapshot"); + const accepted = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Deferred.succeed(accept, undefined); + yield* Deferred.await(nativeAccepted); + if (!startPending) { + yield* Fiber.join(startup); + milestones.push("start-returned"); + yield* Fiber.join(accepted); + } else if (scenario === "pending-ingestion") yield* Deferred.await(ingestionHeld); + yield* Deferred.succeed(releaseSnapshot, undefined); + yield* Fiber.join(stop); + milestones.push(`stop-request:${run.id}`); + if (startPending) yield* Fiber.join(startup); + const stopping = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + stopping.runs.find((candidate) => candidate.id === run.id)?.status, + "running", + "Stop must not report terminal while the native turn is unresolved", + ); + assert.isTrue(stopping.turnItems.some((item) => item.type === "run_interrupt_request")); + assert.isFalse(stopping.turnItems.some((item) => item.type === "run_interrupt_result")); + const stoppedNativeTurn = active!; + const stopEffects = yield* outbox.listByCommandId(CommandId.make("stop")); + assert.equal( + stopEffects.filter((effect) => effect.request.type === "provider-turn.interrupt-start") + .length, + 1, + ); + const originalWait = execution.interruptOwnedStart; + using _wait = vi + .spyOn(execution, "interruptOwnedStart") + .mockImplementation((key) => + Deferred.succeed(ownedWait, undefined).pipe(Effect.andThen(originalWait(key))), + ); + const interrupt = yield* worker.drain(1).pipe(Effect.forkScoped); + yield* Deferred.await(ownedWait); + milestones.push("owned-wait"); + if (identityDelayed || scenario === "interrupt-timeout") { + if (scenario === "interrupt-timeout") yield* Deferred.await(interruptCalled); + yield* TestClock.adjust("15 seconds"); + yield* Fiber.join(interrupt); + if (scenario === "interrupt-timeout") yield* Deferred.await(interruptCancelled); + const unresolved = yield* orchestrator.getThreadProjection(threadId); + assert.equal(unresolved.runs[0]?.status, "running"); + assert.isTrue( + unresolved.turnItems.some( + (item) => item.type === "run_interrupt_request" && item.unresponsiveAt !== undefined, + ), + ); + assert.isFalse(unresolved.turnItems.some((item) => item.type === "run_interrupt_result")); + assert.isTrue( + Option.isSome( + yield* execution.getStartOwnership({ + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }), + ), + ); + if (scenario === "identity-timeout") { + yield* Deferred.succeed(releaseIdentity, undefined); + yield* Fiber.join(accepted); + assert.deepEqual(interrupts, []); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs[0]?.status, + "running", + ); + } + if (scenario === "identity-timeout-force") { + assert.deepEqual(interrupts, []); + const outcome = yield* threads.forceStopThread({ threadId, runId: run.id }); + assert.equal(outcome.outcome, "released"); + const forced = yield* orchestrator.getThreadProjection(threadId); + assert.equal(forced.runs[0]?.status, "interrupted"); + assert.isTrue( + forced.turnItems.some( + (item) => + item.type === "run_interrupt_result" && item.message === "Force stopped by user", + ), + ); + } + return; + } + yield* Deferred.await(interruptCalled); + if (scenario === "interrupt-error") { + yield* Fiber.join(interrupt); + const row = (yield* outbox.listByCommandId(CommandId.make("stop"))).find( + (effect) => effect.request.type === "provider-turn.interrupt-start", + )!; + assert.equal(row.status, "pending"); + assert.include(row.lastError!, "synthetic interrupt rejection"); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs[0]?.status, + "running", + ); + return; + } + assert.isTrue( + Option.isSome( + yield* execution.getStartOwnership({ + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }), + ), + ); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (candidate) => candidate.id === run.id, + )?.status, + "running", + ); + assert.deepEqual(interrupts, [active!.id]); + const terminal = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === run.id && + event.payload.status === "interrupted", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Deferred.succeed(releaseIngestion, undefined); + yield* Deferred.succeed(releaseTerminal, undefined); + yield* Fiber.join(interrupt); + yield* Fiber.join(terminal); + yield* Fiber.join(accepted); + milestones.push(`terminal:${run.id}`); + yield* Deferred.succeed(releaseStart, undefined); + if (scenario !== "queued-promotion") yield* send("replacement"); + const replacementAccepted = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.status === "running" && + event.payload.runAttemptId !== run.activeAttemptId, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* worker.drain(); + yield* Fiber.join(replacementAccepted); + assert.equal(starts, 2); + assert.isDefined(active); + const replacementId = active!.id; + if (scenario === "pending-ingestion") { + const afterLateTerminal = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && event.payload.id === replacementId, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + ...stoppedNativeTurn, + status: "interrupted", + completedAt: yield* DateTime.now, + }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: stoppedNativeTurn.providerThreadId, + providerTurnId: stoppedNativeTurn.id, + runOrdinal: 1, + status: "interrupted", + failure: null, + threadDisposition: "reusable", + }); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: active!, + }); + yield* Fiber.join(afterLateTerminal); + } + assert.isUndefined( + yield* execution.interruptOwnedStart({ + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }), + ); + yield* execution.revokeOwnedStart({ + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }); + yield* orchestrator.dispatch({ + type: "run.interrupt.unresponsive", + commandId: CommandId.make("stale-timeout"), + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }); + yield* orchestrator.dispatch({ + type: "run.force-stop.settle", + commandId: CommandId.make("stale-force"), + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + reason: "stale", + }); + assert.equal(active!.id, replacementId); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs.findLast( + (candidate) => candidate.status === "running", + )?.ordinal, + 2, + ); + assert.equal(interrupts.length, 1); + if (yield* Config.Boolean("T3_START_STOP_TRANSCRIPT").pipe(Config.withDefault(false))) + yield* Effect.logInfo({ synthetic: true, scenario, milestones, starts, interrupts }).pipe( + Effect.provide(Logger.layer([Logger.consoleJson])), + Effect.provideService(References.MinimumLogLevel, "Info"), + ); + }).pipe(Effect.provide(ThreadManagementService.layer.pipe(Layer.provideMerge(base)))); + }), + ), +); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 1d6e34b4a8652..4089075142b9a 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -271,19 +271,26 @@ export const layer: Layer.Layer< }), ), hasUnpairedRunInterruptRequest: () => - projectionStore - .hasUnpairedRunInterruptRequest( - input.threadId, - idAllocator.derive.runSignalTurnItem({ - runId: input.runId, - signal: "interrupt-request", - }), - idAllocator.derive.runSignalTurnItem({ - runId: input.runId, - signal: "interrupt-result", - }), - ) - .pipe(Effect.catchCause(() => Effect.succeed(false))), + projectionStore.getRuntimeRecoveryProjection(input.threadId).pipe( + Effect.flatMap((current) => + current.runs.some( + (run) => run.id === input.runId && run.activeAttemptId === input.attemptId, + ) + ? projectionStore.hasUnpairedRunInterruptRequest( + input.threadId, + idAllocator.derive.runSignalTurnItem({ + runId: input.runId, + signal: "interrupt-request", + }), + idAllocator.derive.runSignalTurnItem({ + runId: input.runId, + signal: "interrupt-result", + }), + ) + : Effect.succeed(false), + ), + Effect.catchCause(() => Effect.succeed(false)), + ), loadRunInterruptReason: () => projectionStore .getThreadRecords(input.threadId, ["turnItems"], { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index f3f1542146c66..bf434507c6186 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -53,8 +53,16 @@ import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts"; import * as PreviewAgentActivity from "../preview/PreviewAgentActivity.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; + import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; +const layerRunExecution = RunExecutionService.layer.pipe( + Layer.provide( + Layer.merge(ThreadCommandExecutor.layer, Layer.mock(ProjectionStore.ProjectionStoreV2)({})), + ), +); + const driver = ProviderDriverKind.make("codex"); const rootAppThread = (id: ThreadId) => ({ @@ -66,7 +74,7 @@ const layerTurnItemPositionsTest = Layer.mock(TurnItemPositionStore.TurnItemPosi allocate: () => Effect.succeed(1), }); -const layerRunExecutionTest = RunExecutionService.layer.pipe( +const layerRunExecutionTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, @@ -881,7 +889,7 @@ it.effect("passes the thread's MCP app context to the provider under a safe key" ]); }).pipe( Effect.provide( - RunExecutionService.layer.pipe( + layerRunExecution.pipe( Layer.provide( Layer.mergeAll( PreviewAgentActivity.layer, @@ -922,7 +930,7 @@ it.effect("fails the run when its ownership check cannot be read before calling const runId = RunId.make("run:run-execution-start-guard-read"); const attemptId = RunAttemptId.make("attempt:run-execution-start-guard-read"); const providerInstanceId = ProviderInstanceId.make("codex"); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, @@ -1124,7 +1132,7 @@ it.effect("starts the provider when checkpoint baseline capture fails", () => effects: ReadonlyArray; }> >([]); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, @@ -1244,7 +1252,7 @@ it.effect.each(["failure", "interruption", "stale-attempt", "start-guard"] as co const refreshes = yield* Ref.make(0); const guardedWrites = yield* Ref.make(0); const writes = yield* Ref.make>>([]); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, @@ -1417,7 +1425,7 @@ it.effect("ingests new member threads of an owned child after the root turn term const order = yield* Ref.make>([]); const origins = yield* Ref.make>([]); const requests = yield* Ref.make>([]); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( Layer.succeed(ProviderUsageAnalytics.ProviderTurnAnalytics, { @@ -1687,7 +1695,7 @@ it.effect("ends a nested native subagent and its thread when the root run is int const nestedId = NodeId.make(`node:${key}:nested`); const grandchildRootId = NodeId.make(`node:${key}:grandchild-root`); const finalized = yield* Deferred.make>(); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, @@ -2098,7 +2106,7 @@ it.effect( const written = yield* Ref.make>([]); const ingested = yield* Ref.make>([]); const ingestionDone = yield* Deferred.make(); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, @@ -3185,7 +3193,7 @@ function captureRootRunTermination(input: { } } }); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( Layer.mock(McpAppModelContext.McpAppModelContext)({ @@ -3708,7 +3716,7 @@ function runBackgroundItemScenario( const previewActivity = yield* PreviewAgentActivity.PreviewAgentActivity.pipe( Effect.provide(PreviewAgentActivity.layer), ); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 6a52fac447926..1b6e5403eb404 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -18,18 +18,20 @@ import { type OrchestrationV2Subagent, type OrchestrationV2TurnItem, type ProviderSessionId, + RunId, + RunAttemptId, + ThreadId, type ProviderThreadId, type ProviderTurnId, - type RunAttemptId, - type ThreadId, type TurnItemId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; +import * as Deferred from "effect/Deferred"; +import * as Option from "effect/Option"; import * as Effect from "effect/Effect"; -import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; @@ -47,7 +49,9 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; import * as ProviderUsageAnalytics from "../telemetry/ProviderUsageAnalytics.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; import type { ProjectionStoreV2Error } from "./ProjectionStore.ts"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import { knownProviderFailureMessage, makeProviderFailure, @@ -553,6 +557,17 @@ export const RunExecutionServiceV2Error = Schema.Union([ ]); export type RunExecutionServiceV2Error = typeof RunExecutionServiceV2Error.Type; +const isRunExecutionStartError = Schema.is(RunExecutionStartError); +// This structured rejection precedes turn/start, unlike a lost start response. +const isRejectedGoal = Schema.is( + Schema.Struct({ + cause: Schema.TaggedStruct("CodexAppServerRequestError", { + code: Schema.Literal(-32600), + method: Schema.Literal("thread/goal/set"), + }), + }), +); + export interface RunExecutionServiceV2StartRootRunInput { readonly commandId: CommandId; readonly appThread: OrchestrationV2AppThread; @@ -581,7 +596,50 @@ export interface RunExecutionServiceV2StartRootRunInput { readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy; } +type StartOwnerKey = { + readonly threadId: ThreadId; + readonly runId: RunId; + readonly attemptId: RunAttemptId; +}; +type StartInterruptTarget = { + readonly providerSessionId: ProviderSessionId; + readonly providerThreadId: ProviderThreadId; + readonly providerTurnId: ProviderTurnId; +}; + +export class RunExecutionInterruptError extends Schema.TaggedError()( + "RunExecutionInterruptError", + { + threadId: ThreadId, + runId: RunId, + attemptId: RunAttemptId, + cause: Schema.Defect(), + unresponsive: Schema.Boolean, + }, +) { + override get message(): string { + return "Could not confirm interruption of the owned provider start."; + } +} + +interface StartOwner { + readonly input: RunExecutionServiceV2StartRootRunInput; + readonly accepted: Deferred.Deferred; + readonly terminal: Deferred.Deferred; + startPending: boolean; + revoked: boolean; + turn?: OrchestrationV2ProviderTurn; + fiber?: Fiber.Fiber; +} + export interface RunExecutionServiceV2Shape { + readonly getStartOwnership: ( + key: StartOwnerKey, + ) => Effect.Effect>; + readonly interruptOwnedStart: ( + key: StartOwnerKey, + ) => Effect.Effect; + readonly revokeOwnedStart: (key: StartOwnerKey) => Effect.Effect; readonly startRootRun: ( input: RunExecutionServiceV2StartRootRunInput, ) => Effect.Effect; @@ -595,6 +653,8 @@ export class RunExecutionServiceV2 extends Context.Service< export const layer: Layer.Layer< RunExecutionServiceV2, never, + | ThreadCommandExecutor.ThreadCommandExecutor + | ProjectionStore.ProjectionStoreV2 | CheckpointService.CheckpointServiceV2 | EventSink.EventSinkV2 | IdAllocator.IdAllocatorV2 @@ -607,6 +667,22 @@ export const layer: Layer.Layer< RunExecutionServiceV2, Effect.gen(function* () { const scope = yield* Effect.scope; + const threadDispatch = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const owners = new Map(); + const ownerId = (key: StartOwnerKey) => + JSON.stringify([key.threadId, key.runId, key.attemptId]); + const releaseOwner = Effect.fn("RunExecutionService.releaseOwner")(function* ( + owner: StartOwner, + ) { + const key = { + threadId: owner.input.run.threadId, + runId: owner.input.run.id, + attemptId: owner.input.attemptId, + }; + yield* Deferred.succeed(owner.terminal, undefined); + if (owners.get(ownerId(key)) === owner) owners.delete(ownerId(key)); + }); const checkpointService = yield* CheckpointService.CheckpointServiceV2; const eventSink = yield* EventSink.EventSinkV2; const idAllocator = yield* IdAllocator.IdAllocatorV2; @@ -884,7 +960,108 @@ export const layer: Layer.Layer< ); return RunExecutionServiceV2.of({ + getStartOwnership: Effect.fn("RunExecutionService.getStartOwnership")((key) => + Effect.sync(() => { + const owner = owners.get(ownerId(key)); + return owner === undefined + ? Option.none() + : Option.some({ startPending: owner.startPending }); + }), + ), + revokeOwnedStart: Effect.fn("RunExecutionService.revokeOwnedStart")((key) => + Effect.sync(() => { + const owner = owners.get(ownerId(key)); + if (owner === undefined) return; + owner.revoked = true; + if (owner.startPending && owner.fiber !== undefined) { + // Request only: provider I/O must never be awaited under the thread lock. + owner.fiber.interruptUnsafe(); + } + }), + ), + interruptOwnedStart: Effect.fn("RunExecutionService.interruptOwnedStart")(function* (key) { + const wait = Effect.gen(function* () { + const owner = owners.get(ownerId(key)); + if (owner === undefined) { + const { attempts, providerTurns } = yield* projectionStore.getThreadRecords( + key.threadId, + ["attempts", "providerTurns"], + ); + const attempt = attempts.find( + (attempt) => attempt.id === key.attemptId && attempt.runId === key.runId, + ); + if ( + attempt !== undefined && + ["completed", "interrupted", "failed", "cancelled"].includes(attempt.status) && + providerTurns + .filter((turn) => turn.runAttemptId === key.attemptId) + .every((turn) => isTerminalProviderTurnStatus(turn.status)) + ) + return; + return yield* Effect.fail("Missing unresolved native-start owner"); + } + const accepted = yield* Deferred.await(owner.accepted).pipe( + Effect.raceFirst(Deferred.await(owner.terminal).pipe(Effect.as(undefined))), + ); + if (accepted === undefined) return; + const current = yield* projectionStore.getThreadRecords(key.threadId, [ + "runs", + "providerThreads", + ]); + const run = current.runs.find((run) => run.id === key.runId); + const binding = current.providerThreads.find( + (thread) => thread.id === owner.input.providerThread.id, + ); + if ( + run?.activeAttemptId !== key.attemptId || + binding?.providerSessionId !== owner.input.providerSessionId || + run.providerThreadId !== binding.id || + binding.providerInstanceId !== owner.input.session.instanceId || + binding.nativeThreadRef?.nativeId !== + owner.input.providerThread.nativeThreadRef?.nativeId + ) { + return yield* Effect.fail("Native-start binding was superseded"); + } + if (yield* Deferred.isDone(owner.terminal)) return; + const turn = owner.turn ?? accepted; + yield* owner.input.session.interruptTurn({ + providerThread: owner.input.providerThread, + providerTurnId: turn.id, + requestRuntimeRestart: true, + }); + yield* Deferred.await(owner.terminal); + return { + providerSessionId: owner.input.providerSessionId, + providerThreadId: owner.input.providerThread.id, + providerTurnId: turn.id, + }; + }); + return yield* wait.pipe( + Effect.timeout("15 seconds"), + Effect.mapError( + (cause) => + new RunExecutionInterruptError({ + ...key, + cause, + unresponsive: + typeof cause === "object" && + cause !== null && + "_tag" in cause && + (cause._tag === "TimeoutError" || + cause._tag === "ProviderAdapterInterruptTimeoutError"), + }), + ), + ); + }), startRootRun: Effect.fn("RunExecutionService.startRootRun")(function* (input) { + const existingOwner = owners.get( + ownerId({ + threadId: input.run.threadId, + runId: input.run.id, + attemptId: input.attemptId, + }), + ); + if (existingOwner?.fiber !== undefined) return yield* Fiber.join(existingOwner.fiber); const rootTurnOrigin: ProviderUsageAnalytics.ProviderTurnOrigin = { createdBy: input.message.createdBy, creationSource: input.message.creationSource, @@ -1001,6 +1178,13 @@ export const layer: Layer.Layer< if (responseStreamingMode === null) { return; } + const owner: StartOwner = { + input, + accepted: yield* Deferred.make(), + terminal: yield* Deferred.make(), + startPending: true, + revoked: false, + }; const terminalEvent = yield* Ref.make(null); const latestTurnItemOrdinal = yield* Ref.make(input.providerTurnOrdinal * 100); const latestProviderThread = yield* Ref.make(input.providerThread); @@ -1117,6 +1301,7 @@ export const layer: Layer.Layer< yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection()); } yield* Ref.set(rootRunFinalized, true); + yield* releaseOwner(owner); }); // Runs for every provider event, including streamed text updates; avoid a per-event span. const trackChildLifecycle = Effect.fnUntraced(function* ( @@ -1412,6 +1597,8 @@ export const layer: Layer.Layer< event.providerTurn.providerThreadId === input.providerThread.id ) { yield* Ref.set(latestRootProviderTurn, event.providerTurn); + owner.turn = event.providerTurn; + yield* Deferred.succeed(owner.accepted, event.providerTurn); } let storedEventCount = 0; const deliveredEvent = filterAssistantEvent( @@ -1504,6 +1691,12 @@ export const layer: Layer.Layer< Math.max(current, event.turnItem.ordinal), ); } + if ( + event.type === "provider_session.updated" && + event.providerSession.id === input.providerSessionId && + event.providerSession.status === "stopped" + ) + yield* releaseOwner(owner); if (event.type === "turn.terminal") { yield* Ref.set(terminalEvent, event); yield* Ref.set(rootTerminalSeen, true); @@ -1664,7 +1857,7 @@ export const layer: Layer.Layer< ), Effect.ensuring(endPreviewActivity), Effect.ensuring(eventSubscription.close), - Effect.forkDetach, + Effect.forkIn(scope), ); // A provider turn is a sign that the session is still alive. Keep @@ -1676,16 +1869,17 @@ export const layer: Layer.Layer< message: "Failed to refresh MCP credential liveness.", }), ); - yield* touchMcpCredential; - const appContext = (yield* mcpAppModelContext - .forThread(input.run.threadId) - .pipe( - Effect.catch((cause) => - Effect.logWarning("Failed to read MCP app model context.", { cause }).pipe( - Effect.as([]), - ), + yield* touchMcpCredential.pipe( + Effect.onInterrupt(() => Fiber.interrupt(providerEventFiber)), + ); + const appContext = (yield* mcpAppModelContext.forThread(input.run.threadId).pipe( + Effect.catch((cause) => + Effect.logWarning("Failed to read MCP app model context.", { cause }).pipe( + Effect.as([]), ), - )).map((entry) => ({ + ), + Effect.onInterrupt(() => Fiber.interrupt(providerEventFiber)), + )).map((entry) => ({ // The item id alone is unique and needs no escaping; server and // tool names are free text that would break the tag Codex wraps // the context in. @@ -1714,33 +1908,54 @@ export const layer: Layer.Layer< runtimePolicy: input.runtimePolicy, ...(appContext.length === 0 ? {} : { appContext }), }; - // Check after the context read: ingestion can settle the run while - // it awaits. A failed ownership read fails the start below. - const shouldStart = - input.shouldStartProviderTurn === undefined - ? Exit.succeed(true) - : yield* Effect.exit(input.shouldStartProviderTurn()); - if (Exit.isSuccess(shouldStart) && !shouldStart.value) { - yield* Fiber.interrupt(providerEventFiber); - return; - } - const compact = input.message.attachments.length === 0 && input.message.text.trim().toLowerCase() === "/compact"; - const startTurn = compact - ? (input.session.compactThread?.(turnInput) ?? - Effect.fail( - new ProviderAdapter.ProviderAdapterTurnStartError({ - driver: input.session.driver, - threadId: input.run.threadId, - providerThreadId: input.providerThread.id, - runId: input.run.id, - cause: "This provider does not support context compaction.", - }), - )) - : input.session.startTurn(turnInput); - yield* Effect.andThen(shouldStart, startTurn).pipe( + const startTurn = Effect.suspend(() => + compact + ? (input.session.compactThread?.(turnInput) ?? + Effect.fail( + new ProviderAdapter.ProviderAdapterTurnStartError({ + driver: input.session.driver, + threadId: input.run.threadId, + providerThreadId: input.providerThread.id, + runId: input.run.id, + cause: "This provider does not support context compaction.", + }), + )) + : input.session.startTurn(turnInput), + ); + const failBeforeNativeStart = Effect.fn("RunExecutionService.failBeforeNativeStart")( + function* (cause: Cause.Cause) { + yield* Fiber.interrupt(providerEventFiber); + yield* writeFinalRunEvents({ + run: input.run, + rootNode: input.rootNode, + checkpointScope: input.checkpointScope, + providerThread: yield* Ref.get(latestProviderThread), + attempt: input.attempt, + writeIfRunCurrent: { activeAttemptId: input.attempt.id, expectedStatus: "running" }, + openRunOwnedSubagents: yield* Ref.get(openRunOwnedSubagents), + terminal: makeFailedTerminalEvent( + makeProviderFailure({ cause: Cause.squash(cause), class: "provider_error" }), + (yield* Ref.get(latestTurnItemOrdinal)) + 1, + ), + failureItemPersisted: false, + refreshAfterTurn, + }).pipe( + Effect.mapError( + (writeCause) => + new RunExecutionStartError({ + commandId: input.commandId, + runId: input.run.id, + cause: { start: cause, write: writeCause }, + }), + ), + ); + yield* releaseOwner(owner); + }, + ); + const nativeStart = Effect.suspend(() => (owner.revoked ? Effect.void : startTurn)).pipe( // Wait for acceptance so abandoned startup cannot leave a refresher // on the already-detached ingestion fiber. Effect.tap(() => @@ -1750,51 +1965,88 @@ export const layer: Layer.Layer< ), ), Effect.catchCause((cause) => - Effect.logError("orchestration V2 provider turn start failed", { - runId: input.run.id, - cause, - }).pipe( - Effect.andThen(Fiber.interrupt(providerEventFiber)), - Effect.andThen(Ref.get(latestProviderThread)), - Effect.flatMap((providerThread) => - Ref.get(latestTurnItemOrdinal).pipe( - Effect.flatMap((latestItemOrdinal) => - Ref.get(openRunOwnedSubagents).pipe( - Effect.flatMap((openSubagents) => - writeFinalRunEvents({ - run: input.run, - rootNode: input.rootNode, - checkpointScope: input.checkpointScope, - providerThread, - attempt: input.attempt, - writeIfRunCurrent: { - activeAttemptId: input.attempt.id, - expectedStatus: "running", - }, - openRunOwnedSubagents: openSubagents, - terminal: makeFailedTerminalEvent( - makeProviderFailure({ - cause: Cause.squash(cause), - class: Exit.isFailure(shouldStart) ? "unknown" : "provider_error", - }), - latestItemOrdinal + 1, - ), - failureItemPersisted: false, - refreshAfterTurn, + Effect.gen(function* () { + if ( + !owner.revoked && + !(yield* Deferred.isDone(owner.accepted)) && + isRejectedGoal(Cause.squash(cause)) + ) + return yield* failBeforeNativeStart(cause); + // A failed response is not proof that native work never started. + return yield* new RunExecutionStartError({ + commandId: input.commandId, + runId: input.run.id, + cause, + }); + }), + ), + Effect.ensuring( + Effect.sync(() => { + owner.startPending = false; + }), + ), + ); + let claimed = false; + let reused = false; + yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const fiber = yield* restore( + threadDispatch.withLock( + input.run.threadId, + Effect.gen(function* () { + const existing = owners.get( + ownerId({ + threadId: input.run.threadId, + runId: input.run.id, + attemptId: input.attemptId, + }), + ); + if (existing !== undefined) { + reused = true; + return existing.fiber; + } + if ( + input.shouldStartProviderTurn !== undefined && + !(yield* input.shouldStartProviderTurn()) + ) + return undefined; + return yield* Effect.uninterruptibleMask((restoreStart) => + Effect.gen(function* () { + owners.set( + ownerId({ + threadId: input.run.threadId, + runId: input.run.id, + attemptId: input.attemptId, }), - ), - ), - ), - ), - ), - Effect.mapError( - (writeCause) => - new RunExecutionStartError({ - commandId: input.commandId, - runId: input.run.id, - cause: { start: cause, write: writeCause }, - }), + owner, + ); + claimed = true; + owner.fiber = yield* restoreStart(nativeStart).pipe(Effect.forkIn(scope)); + return owner.fiber; + }), + ); + }), ), + ); + if (fiber !== undefined) yield* restore(Fiber.join(fiber)); + }), + ).pipe( + Effect.catchCauseIf( + (cause) => !claimed && !reused && !Cause.hasInterruptsOnly(cause), + failBeforeNativeStart, + ), + Effect.mapError((cause) => + isRunExecutionStartError(cause) + ? cause + : new RunExecutionStartError({ + commandId: input.commandId, + runId: input.run.id, + cause, + }), + ), + Effect.ensuring( + Effect.suspend(() => + claimed ? Effect.void : Fiber.interrupt(providerEventFiber).pipe(Effect.asVoid), ), ), ); diff --git a/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts b/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts index c2b9e0aabcd5b..fc46baa46cd4f 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts @@ -13,6 +13,7 @@ import * as CheckpointCapture from "./CheckpointCaptureService.ts"; import * as CheckpointRollback from "./CheckpointRollbackService.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import * as EffectWorker from "./EffectWorker.ts"; +import * as RunExecutionService from "./RunExecutionService.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderSessions from "./ProviderSessionManager.ts"; import * as ProviderTurnControl from "./ProviderTurnControlService.ts"; @@ -84,6 +85,7 @@ it.effect.each([ Layer.provide( Layer.mergeAll( finalization, + Layer.mock(RunExecutionService.RunExecutionServiceV2)({}), Layer.mock(ProviderSessions.ProviderSessionManagerV2)({}), Layer.mock(ProviderTurnControl.ProviderTurnControlServiceV2)({}), Layer.mock(CheckpointRollback.CheckpointRollbackServiceV2)({}), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 9285f72d46f56..9127c1c8cf693 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -37,6 +37,13 @@ import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; +import * as RunExecutionService from "./RunExecutionService.ts"; + +const layerThreadManagement = ThreadManagementService.layer.pipe( + Layer.provide( + Layer.mock(RunExecutionService.RunExecutionServiceV2)({ revokeOwnedStart: () => Effect.void }), + ), +); it("stamps authoritative provenance on commands that create threads or messages", () => { const command: OrchestrationV2Command = { @@ -82,7 +89,7 @@ const layerReceiptsTest = CommandReceiptStore.layer.pipe( const layerDispatchChokepointTest = (settingsLayer: ReturnType) => (dispatched: Array) => - ThreadManagementService.layer.pipe( + layerThreadManagement.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ @@ -214,7 +221,7 @@ it.effect("waitForThread follows a restart continuation instead of the cut run", let pending = true; const stored = (sequence: number, event: object) => ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent; - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ @@ -291,7 +298,7 @@ it.effect("waitForThread can wait past an ended run for the thread's delegated w let subagents = [{ status: "running" }]; const stored = (sequence: number, event: object) => ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent; - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ @@ -369,7 +376,7 @@ it.effect("waitForThread settles delegated work on a thread rolled back past its startedAt: DateTime.makeUnsafe(0), completedAt: DateTime.makeUnsafe(1_000), } as unknown as OrchestrationV2Run; - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ @@ -422,7 +429,7 @@ it.effect("waitForThread returns the cut run once its continuation is declined", const subscribed = yield* Deferred.make(); const declined = yield* Deferred.make(); let pending = true; - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ @@ -511,7 +518,7 @@ it.effect.each(["live", "no-session", "unsupported-runtime"] as const)( }), }), } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime; - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide( Layer.mergeAll( Layer.mock(Orchestrator.OrchestratorV2)({ @@ -568,7 +575,7 @@ it.effect.each([ const threadId = ThreadId.make("thread:thread-management:settle-after-run"); const runId = RunId.make("run:thread-management:settle-after-run"); const dispatched: Array = []; - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ @@ -653,7 +660,7 @@ it.effect.each([ const dispatched: Array = []; const teardowns: Array = []; const tornDown = yield* Deferred.make(); - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide( Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ isLive: () => Effect.succeed(true), @@ -734,7 +741,7 @@ it.effect("Force stop refuses once a newer run has started, before killing anyth Effect.gen(function* () { const threadId = ThreadId.make("thread:thread-management:force-stop-superseded"); let killed = false; - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 7aff0137562f3..d51d52cb6ee4d 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -62,6 +62,7 @@ import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import { projectedRowEncodedBytes } from "./threadHistoryPaging.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; +import * as RunExecutionService from "./RunExecutionService.ts"; import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimit.ts"; import { projectTurnItemForDetail } from "./WireProjection.ts"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; @@ -571,6 +572,7 @@ const isForceStopThreadError = Schema.is(OrchestrationV2ForceStopThreadError); const make = Effect.gen(function* () { const threadDispatch = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const runExecution = yield* RunExecutionService.RunExecutionServiceV2; const orchestrator = yield* Orchestrator.OrchestratorV2; const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; const layerScope = yield* Effect.scope; @@ -1142,6 +1144,12 @@ const make = Effect.gen(function* () { reason: "superseded", }); } + if (target?.activeAttemptId != null) + yield* runExecution.revokeOwnedStart({ + threadId: input.threadId, + runId: target.id, + attemptId: target.activeAttemptId, + }); // Shared scopes are never tracked per thread, so this only kills the thread's own agent. const killed = yield* agentScope.kill(input.threadId); return { records, target, killed }; @@ -1169,10 +1177,11 @@ const make = Effect.gen(function* () { ? Effect.void : providerTurn === undefined ? orchestrator.dispatch({ - type: "run.interrupt", + type: "run.force-stop.settle", commandId, threadId: input.threadId, runId: run.id, + attemptId: run.activeAttemptId!, reason: FORCE_STOP_REASON, }) : orchestrator.dispatch({ @@ -1575,6 +1584,7 @@ const layerLegacyV1ThreadImporterNoop = Layer.succeed( export const layer: Layer.Layer< ThreadManagementService, never, + | RunExecutionService.RunExecutionServiceV2 | Orchestrator.OrchestratorV2 | ProviderSessionManager.ProviderSessionManagerV2 | ServerSettings.ServerSettingsService @@ -1587,6 +1597,7 @@ export const layer: Layer.Layer< export const layerWithLegacyImporter: Layer.Layer< ThreadManagementService, never, + | RunExecutionService.RunExecutionServiceV2 | LegacyV1ThreadImporter.LegacyV1ThreadImporter | Orchestrator.OrchestratorV2 | ProviderSessionManager.ProviderSessionManagerV2 diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 1b54627595372..fea21a005526c 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -150,6 +150,8 @@ const layerProviderAuthServiceProvided = ProviderAuthService.layer.pipe( const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, + ProjectionStore.layer, McpAppModelContext.layer, PreviewAgentActivity.layer, layerCheckpointServiceProvided, @@ -251,6 +253,7 @@ const layerThreadManagementProvided = ThreadManagementService.layerWithLegacyImp Layer.provide( Layer.mergeAll( layerOrchestratorProvided, + layerRunExecutionServiceProvided, layerLegacyV1ThreadImporterProvided, layerCommandReceiptStoreProvided, layerProviderSessionManagerProvided, @@ -322,6 +325,7 @@ const layerEffectExecutorProvided = EffectWorker.layerExecutor.pipe( layerProviderSessionManagerProvided, layerProviderTurnControlServiceProvided, layerProviderTurnStartServiceProvided, + layerRunExecutionServiceProvided, layerRuntimeRequestServiceProvided, layerThreadTitleRegenerationProvided, layerThreadManagementProvided, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index e4cac1bd4adfc..cf656d41045d3 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -254,6 +254,7 @@ export function layerProviderReplay { @@ -287,6 +288,7 @@ export function layerWithRegistry( | EffectWorker.OrchestrationEffectWorkerV2 | EventSink.EventSinkV2 | ProviderSessionManager.ProviderSessionManagerV2 + | RunExecutionService.RunExecutionServiceV2 | CommandOutputRecorder.CommandOutputRecorder, Error | MigrationError | PlatformError.PlatformError | SqlError > { @@ -401,6 +403,7 @@ export function layerWithRegistry( const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, PreviewAgentActivity.layer, layerCheckpointServiceProvided, @@ -522,6 +525,7 @@ export function layerWithRegistry( layerProviderSessionManagerProvided, layerProviderTurnControlServiceProvided, layerProviderTurnStartServiceProvided, + layerRunExecutionServiceProvided, layerRuntimeRequestServiceProvided, layerThreadTitleRegenerationTest, layerServerSettings, @@ -534,6 +538,7 @@ export function layerWithRegistry( ); const layerReplayRuntime = Layer.mergeAll( layerOrchestratorProvided, + layerRunExecutionServiceProvided, layerProviderSessionManagerProvided, layerEffectWorkerProvided, layerEventSinkProvided, diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 5eb2b6ed4bfdd..d8a6204bd4a53 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3490,6 +3490,21 @@ const OrchestrationV2InternalCommand = Schema.Union([ threadId: ThreadId, requestId: RuntimeRequestId, }), + Schema.Struct({ + type: Schema.Literal("run.interrupt.unresponsive"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + attemptId: RunAttemptId, + }), + Schema.Struct({ + type: Schema.Literal("run.force-stop.settle"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + attemptId: RunAttemptId, + reason: Schema.String, + }), /** The provider did not confirm a Stop of this turn in time. */ Schema.Struct({ type: Schema.Literal("thread.interrupt.unresponsive"), From a9e9f7150f3244d5deb703003bb92760c1b3089c Mon Sep 17 00:00:00 2001 From: Chinoman10 <8300763+Chinoman10@users.noreply.github.com> Date: Thu, 5 Nov 2026 20:22:53 +0000 Subject: [PATCH 2/9] fix(server): settle refused starts and reconcile session teardown Settle unsupported compaction before claiming ownership and share the established pre-prompt refusal predicate. Preserve consumers after ambiguous acceptance errors. Revoke pending starts on confirmed exact-runtime teardown, await guarded failure finalization, and remove ownership before waking interrupt waiters. Add refusal and teardown regressions to the existing race table. Verification: check:changed against 337c5858c33b passed all 36 package typechecks and 400 tests. External witnesses failed before the correction and passed afterward. Model: openai-codex/gpt-6.1-sol. Harness: Pi. --- .../ProviderSessionManager.ts | 42 +++++- .../ProviderStartStop.integration.test.ts | 129 +++++++++++++++++- .../ProviderTurnStartService.ts | 19 +-- .../orchestration-v2/RunExecutionService.ts | 129 +++++++++++++----- .../src/server/ProviderAdapter.ts | 2 + packages/provider-core/src/server/failure.ts | 14 ++ 6 files changed, 279 insertions(+), 56 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index f06011676c667..23b996c364966 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -226,6 +226,7 @@ interface LiveSessionEntry { readonly eventSubscribers: Ref.Ref< ReadonlyMap> >; + readonly closeObservers: Ref.Ref> | null>; readonly requestEventPermit: Semaphore.Semaphore; readonly scope: Scope.Closeable; readonly idleGeneration: number; @@ -971,9 +972,11 @@ export const layerWithOptions = ( const closeScopeWithin = Effect.fn("ProviderSessionManager.closeScopeWithin")(function* ( scope: Scope.Closeable, annotations: { readonly providerSessionId?: ProviderSessionId; readonly reason: string }, + onClosed: Effect.Effect = Effect.void, ) { const closeFiber = yield* Scope.close(scope, Exit.void).pipe( Effect.exit, + Effect.tap((exit) => (Exit.isSuccess(exit) ? onClosed : Effect.void)), Effect.forkDetach({ startImmediately: true }), ); const closeExit = yield* Fiber.join(closeFiber).pipe( @@ -1039,10 +1042,18 @@ export const layerWithOptions = ( input.detail ?? `Provider session released: ${input.reason}.`, ); } - const closeExit = yield* closeScopeWithin(entry.scope, { - providerSessionId: input.providerSessionId, - reason: input.reason, - }); + const closeExit = yield* closeScopeWithin( + entry.scope, + { providerSessionId: input.providerSessionId, reason: input.reason }, + Ref.getAndSet(entry.closeObservers, null).pipe( + Effect.flatMap((observers) => + Effect.forEach(observers ?? [], (observer) => observer, { + concurrency: "unbounded", + discard: true, + }), + ), + ), + ); const records = { entry, reason: input.reason, @@ -1695,6 +1706,7 @@ export const layerWithOptions = ( eventSubscribers: Ref.Ref< ReadonlyMap> >, + closeObservers: LiveSessionEntry["closeObservers"], ): ProviderAdapter.ProviderAdapterV2SessionRuntime => { const providerSessionId = runtime.providerSessionId; const subscribeEvents = makeEventSubscription(eventSubscribers); @@ -1711,6 +1723,22 @@ export const layerWithOptions = ( return { ...runtime, subscribeEvents, + onSessionClosed: (observer) => + Ref.modify(closeObservers, (current) => + current === null + ? ([true, current] as const) + : ([false, new Set([...current, observer])] as const), + ).pipe( + Effect.tap((closed) => (closed ? observer : Effect.void)), + Effect.as( + Ref.update(closeObservers, (current) => { + if (current === null) return null; + const next = new Set(current); + next.delete(observer); + return next; + }), + ), + ), events: Stream.unwrap( subscribeEvents.pipe(Effect.map((subscription) => subscription.events)), ), @@ -2232,7 +2260,10 @@ export const layerWithOptions = ( const eventSubscribers = yield* Ref.make< ReadonlyMap> >(new Map()); - const exposedRuntime = decorateRuntime(runtime, eventSubscribers); + const closeObservers = yield* Ref.make> | null>( + new Set(), + ); + const exposedRuntime = decorateRuntime(runtime, eventSubscribers, closeObservers); const entry: LiveSessionEntry = { attachedThreadIds: new Set([input.threadId]), loadedProviderThreadKeyByThread: new Map(), @@ -2246,6 +2277,7 @@ export const layerWithOptions = ( runtime, exposedRuntime, eventSubscribers, + closeObservers, requestEventPermit: yield* Semaphore.make(1), scope: sessionScope, idleGeneration: 0, diff --git a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts index 034cfc8eaa6a8..d9710ecfcd7bd 100644 --- a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts @@ -12,6 +12,7 @@ import { type OrchestrationV2ProviderTurn, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Cause from "effect/Cause"; import * as Config from "effect/Config"; import * as Logger from "effect/Logger"; import * as References from "effect/References"; @@ -33,6 +34,7 @@ import * as EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; @@ -49,11 +51,17 @@ it.effect.each([ "interrupt-timeout", "identity-timeout", "identity-timeout-force", + "unsupported-compact", + "prompt-refusal", + "manager-release", + "manager-release-interrupt", + "provider-stopped", + "stream-ended", ] as const)("Stop crosses native start: %s", (scenario) => Effect.scoped( Effect.gen(function* () { const cwd = yield* checkpointWorkspace(`start-stop-${scenario}`); - const events = yield* Queue.unbounded(); + const events = yield* Queue.unbounded(); const entered = yield* Deferred.make(); const accept = yield* Deferred.make(); const nativeAccepted = yield* Deferred.make(); @@ -94,6 +102,18 @@ it.effect.each([ turn, ) => Effect.gen(function* () { + if (scenario === "prompt-refusal") { + return yield* new ProviderAdapter.ProviderAdapterTurnStartError({ + driver, + threadId: turn.threadId, + providerThreadId: turn.providerThread.id, + runId: turn.runId, + cause: { + _tag: "ClaudeAppendSystemPromptFileError", + path: "/task-owned/missing-file", + }, + }); + } yield* Deferred.succeed(entered, undefined); yield* Deferred.await(accept); assert.isUndefined(active, "native turns must never overlap"); @@ -290,6 +310,15 @@ it.effect.each([ const originalStart = execution.startRootRun; let staleInput: RunExecutionService.RunExecutionServiceV2StartRootRunInput | undefined; using _beforeOwnership = vi.spyOn(execution, "startRootRun").mockImplementation((input) => { + if (scenario === "unsupported-compact") { + const session = { ...input.session }; + delete session.compactThread; + return originalStart({ + ...input, + message: { ...input.message, text: "/compact" }, + session, + }); + } if (scenario !== "before-ownership") return originalStart(input); staleInput = input; return Deferred.succeed(beforeOwnership, undefined).pipe(Effect.andThen(Effect.never)); @@ -312,6 +341,26 @@ it.effect.each([ }); yield* send("first"); const startup = yield* worker.drain(1).pipe(Effect.forkScoped); + if (scenario === "unsupported-compact" || scenario === "prompt-refusal") { + yield* Fiber.join(startup); + const failed = yield* orchestrator.getThreadProjection(threadId); + const run = failed.runs[0]!; + assert.equal(run.status, "failed"); + assert.equal(failed.attempts[0]?.status, "failed"); + assert.equal(failed.nodes.find((node) => node.id === run.rootNodeId)?.status, "failed"); + assert.isTrue( + Option.isNone( + yield* execution.getStartOwnership({ + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }), + ), + ); + assert.equal(starts, 0); + assert.deepEqual(interrupts, []); + return; + } if (scenario === "before-ownership") { yield* Deferred.await(beforeOwnership); yield* orchestrator.dispatch({ @@ -333,6 +382,75 @@ it.effect.each([ if (scenario === "queued-promotion") yield* send("promoted", true); const initial = yield* orchestrator.getThreadProjection(threadId); const run = initial.runs.find((run) => run.status === "running")!; + if ( + [ + "manager-release", + "manager-release-interrupt", + "provider-stopped", + "stream-ended", + ].includes(scenario) + ) { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const providerSessionId = initial.providerThreads.find( + (thread) => thread.id === run.providerThreadId, + )!.providerSessionId!; + const runtime = Option.getOrThrow(yield* manager.get(providerSessionId)); + if (scenario === "manager-release-interrupt") { + const accepted = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Deferred.succeed(accept, undefined); + yield* Fiber.join(accepted); + } + const waiter = yield* Deferred.succeed(ownedWait, undefined).pipe( + Effect.andThen( + execution.interruptOwnedStart({ + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }), + ), + Effect.forkScoped, + ); + yield* Deferred.await(ownedWait); + if (scenario === "manager-release-interrupt") yield* Deferred.await(interruptCalled); + if (scenario === "provider-stopped" || scenario === "stream-ended") { + if (scenario === "provider-stopped") + yield* Queue.offer(events, { + type: "provider_session.updated", + driver, + providerSession: { ...runtime.providerSession, status: "stopped" }, + }); + yield* Queue.end(events); + } else { + yield* manager.release({ providerSessionId, reason: "manual_shutdown" }); + } + yield* Fiber.join(waiter); + yield* Fiber.join(startup); + assert.isTrue( + Option.isNone( + yield* execution.getStartOwnership({ + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }), + ), + ); + const failed = yield* orchestrator.getThreadProjection(threadId); + assert.equal(failed.runs[0]?.status, "failed"); + assert.equal(failed.attempts[0]?.status, "failed"); + assert.isFalse(failed.turnItems.some((item) => item.type === "run_interrupt_result")); + yield* Deferred.succeed(accept, undefined); + yield* Deferred.succeed(releaseStart, undefined); + assert.equal(starts, scenario === "manager-release-interrupt" ? 1 : 0); + return; + } const original = store.getThreadRecords; let armed = true; using _snapshot = vi.spyOn(store, "getThreadRecords").mockImplementation((...args) => @@ -448,6 +566,15 @@ it.effect.each([ assert.deepEqual(interrupts, []); const outcome = yield* threads.forceStopThread({ threadId, runId: run.id }); assert.equal(outcome.outcome, "released"); + assert.isTrue( + Option.isNone( + yield* execution.getStartOwnership({ + threadId, + runId: run.id, + attemptId: run.activeAttemptId!, + }), + ), + ); const forced = yield* orchestrator.getThreadProjection(threadId); assert.equal(forced.runs[0]?.status, "interrupted"); assert.isTrue( diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 4089075142b9a..861a3f79a5905 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -22,7 +22,6 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; -import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; import * as ServerConfig from "../config.ts"; @@ -54,6 +53,7 @@ import { ContextHandoffBackgroundWorkError, knownProviderFailureMessage, makeProviderFailure, + refusedBeforePrompt, } from "@t3tools/provider-core/server/failure"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; @@ -98,23 +98,6 @@ const isProviderAdapterOpenSessionError = Schema.is( ProviderAdapter.ProviderAdapterOpenSessionError, ); -/** - * Claude refuses these before it reads the prompt: replacing a process running background work, - * launch arguments whose appended system prompt or settings cannot be read, or MCP registration - * that never answers while the query opens. - */ -const PROMPT_REFUSAL_TAGS = [ - "ClaudeBackgroundWorkBlocksQueryReplacementError", - "ClaudeAppendSystemPromptFileError", - "ClaudeAppendSystemPromptMissingValueError", - "ClaudeSettingsLaunchArgError", - "ClaudeMcpRegistrationTimeoutError", -] as const; - -const refusedBeforePrompt = (error: unknown): boolean => - PROMPT_REFUSAL_TAGS.some((tag) => Predicate.isTagged(error, tag)) || - (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); - export interface ProviderTurnStartServiceV2Shape { readonly start: (input: { readonly threadId: ThreadId; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 1b6e5403eb404..390f62e3984dc 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -54,6 +54,7 @@ import type { ProjectionStoreV2Error } from "./ProjectionStore.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import { knownProviderFailureMessage, + refusedBeforePrompt, makeProviderFailure, makeProviderFailureTurnItem, } from "@t3tools/provider-core/server/failure"; @@ -630,6 +631,7 @@ interface StartOwner { revoked: boolean; turn?: OrchestrationV2ProviderTurn; fiber?: Fiber.Fiber; + unsubscribeClosed?: Effect.Effect; } export interface RunExecutionServiceV2Shape { @@ -680,8 +682,9 @@ export const layer: Layer.Layer< runId: owner.input.run.id, attemptId: owner.input.attemptId, }; - yield* Deferred.succeed(owner.terminal, undefined); if (owners.get(ownerId(key)) === owner) owners.delete(ownerId(key)); + yield* owner.unsubscribeClosed ?? Effect.void; + yield* Deferred.succeed(owner.terminal, undefined); }); const checkpointService = yield* CheckpointService.CheckpointServiceV2; const eventSink = yield* EventSink.EventSinkV2; @@ -1024,12 +1027,16 @@ export const layer: Layer.Layer< } if (yield* Deferred.isDone(owner.terminal)) return; const turn = owner.turn ?? accepted; - yield* owner.input.session.interruptTurn({ - providerThread: owner.input.providerThread, - providerTurnId: turn.id, - requestRuntimeRestart: true, - }); - yield* Deferred.await(owner.terminal); + yield* owner.input.session + .interruptTurn({ + providerThread: owner.input.providerThread, + providerTurnId: turn.id, + requestRuntimeRestart: true, + }) + .pipe( + Effect.andThen(Deferred.await(owner.terminal)), + Effect.raceFirst(Deferred.await(owner.terminal)), + ); return { providerSessionId: owner.input.providerSessionId, providerThreadId: owner.input.providerThread.id, @@ -1178,6 +1185,46 @@ export const layer: Layer.Layer< if (responseStreamingMode === null) { return; } + const compact = + input.message.attachments.length === 0 && + input.message.text.trim().toLowerCase() === "/compact"; + const compactThread = input.session.compactThread; + if (compact && compactThread === undefined) { + yield* writeFinalRunEvents({ + run: input.run, + rootNode: input.rootNode, + checkpointScope: input.checkpointScope, + providerThread: input.providerThread, + attempt: input.attempt, + writeIfRunCurrent: { activeAttemptId: input.attemptId, expectedStatus: "running" }, + terminal: makeFailedTerminalEvent( + makeProviderFailure({ + cause: new ProviderAdapter.ProviderAdapterTurnStartError({ + driver: input.session.driver, + threadId: input.run.threadId, + providerThreadId: input.providerThread.id, + runId: input.run.id, + cause: "This provider does not support context compaction.", + }), + class: "provider_error", + }), + input.providerTurnOrdinal * 100 + 1, + ), + failureItemPersisted: false, + refreshAfterTurn, + }).pipe( + Effect.mapError( + (cause) => + new RunExecutionStartError({ + commandId: input.commandId, + runId: input.run.id, + cause, + }), + ), + ); + return; + } + const sessionClosed = yield* Deferred.make(); const owner: StartOwner = { input, accepted: yield* Deferred.make(), @@ -1510,6 +1557,15 @@ export const layer: Layer.Layer< const providerEventFiber = yield* eventSubscription.events.pipe( Stream.mapEffect((event) => Effect.gen(function* () { + if ( + event.type === "provider_session.updated" && + event.providerSession.id === input.providerSessionId && + event.providerSession.status === "stopped" + ) { + owner.revoked = true; + owner.fiber?.interruptUnsafe(); + yield* Deferred.succeed(sessionClosed, undefined); + } const previous = yield* Ref.get(eventRouting); const [accepted, routing] = routeProviderEvent(event, routeIdentity, previous); yield* Ref.set(eventRouting, routing); @@ -1691,12 +1747,6 @@ export const layer: Layer.Layer< Math.max(current, event.turnItem.ordinal), ); } - if ( - event.type === "provider_session.updated" && - event.providerSession.id === input.providerSessionId && - event.providerSession.status === "stopped" - ) - yield* releaseOwner(owner); if (event.type === "turn.terminal") { yield* Ref.set(terminalEvent, event); yield* Ref.set(rootTerminalSeen, true); @@ -1718,7 +1768,13 @@ export const layer: Layer.Layer< if (yield* agentTurnsSettled) yield* endPreviewActivity; }), ), - Stream.takeUntilEffect(() => shouldStopProviderEventIngestion), + Stream.takeUntilEffect(() => + Deferred.isDone(sessionClosed).pipe( + Effect.flatMap((closed) => + closed ? Effect.succeed(true) : shouldStopProviderEventIngestion, + ), + ), + ), Stream.runDrain, Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })), Effect.flatMap(() => @@ -1765,6 +1821,7 @@ export const layer: Layer.Layer< }) .pipe( Effect.as(true), + Effect.raceFirst(Deferred.await(sessionClosed).pipe(Effect.as(true))), Effect.catchCauseIf( (interruptCause) => !Cause.hasInterruptsOnly(interruptCause), (interruptCause) => @@ -1853,6 +1910,9 @@ export const layer: Layer.Layer< Effect.forkIn(scope), Effect.flatMap(Fiber.join), ); + if ((yield* Deferred.isDone(sessionClosed)) || abandonedProviderTurn !== null) { + yield* releaseOwner(owner); + } }), ), Effect.ensuring(endPreviewActivity), @@ -1908,21 +1968,9 @@ export const layer: Layer.Layer< runtimePolicy: input.runtimePolicy, ...(appContext.length === 0 ? {} : { appContext }), }; - const compact = - input.message.attachments.length === 0 && - input.message.text.trim().toLowerCase() === "/compact"; const startTurn = Effect.suspend(() => - compact - ? (input.session.compactThread?.(turnInput) ?? - Effect.fail( - new ProviderAdapter.ProviderAdapterTurnStartError({ - driver: input.session.driver, - threadId: input.run.threadId, - providerThreadId: input.providerThread.id, - runId: input.run.id, - cause: "This provider does not support context compaction.", - }), - )) + compact && compactThread !== undefined + ? compactThread(turnInput) : input.session.startTurn(turnInput), ); const failBeforeNativeStart = Effect.fn("RunExecutionService.failBeforeNativeStart")( @@ -1969,7 +2017,7 @@ export const layer: Layer.Layer< if ( !owner.revoked && !(yield* Deferred.isDone(owner.accepted)) && - isRejectedGoal(Cause.squash(cause)) + (isRejectedGoal(Cause.squash(cause)) || refusedBeforePrompt(Cause.squash(cause))) ) return yield* failBeforeNativeStart(cause); // A failed response is not proof that native work never started. @@ -1990,6 +2038,17 @@ export const layer: Layer.Layer< let reused = false; yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { + if (input.session.onSessionClosed !== undefined) { + owner.unsubscribeClosed = yield* input.session.onSessionClosed( + Effect.gen(function* () { + owner.revoked = true; + owner.fiber?.interruptUnsafe(); + yield* Deferred.succeed(sessionClosed, undefined); + yield* Fiber.await(providerEventFiber); + yield* releaseOwner(owner); + }), + ); + } const fiber = yield* restore( threadDispatch.withLock( input.run.threadId, @@ -2006,8 +2065,9 @@ export const layer: Layer.Layer< return existing.fiber; } if ( - input.shouldStartProviderTurn !== undefined && - !(yield* input.shouldStartProviderTurn()) + owner.revoked || + (input.shouldStartProviderTurn !== undefined && + !(yield* input.shouldStartProviderTurn())) ) return undefined; return yield* Effect.uninterruptibleMask((restoreStart) => @@ -2046,7 +2106,12 @@ export const layer: Layer.Layer< ), Effect.ensuring( Effect.suspend(() => - claimed ? Effect.void : Fiber.interrupt(providerEventFiber).pipe(Effect.asVoid), + claimed + ? Effect.void + : Fiber.interrupt(providerEventFiber).pipe( + Effect.andThen(owner.unsubscribeClosed ?? Effect.void), + Effect.asVoid, + ), ), ), ); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 9ea7597e6d087..7a3bfe440992b 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -612,6 +612,8 @@ export interface ProviderAdapterV2SessionRuntime { readonly events: Stream.Stream; /** Manager runtimes broadcast independent subscriptions; adapters may expose only a single-consumer stream. */ readonly subscribeEvents?: Effect.Effect; + /** Manager-owned notification after this exact runtime's scope closes successfully; returns unsubscribe. */ + readonly onSessionClosed?: (observer: Effect.Effect) => Effect.Effect>; readonly hasPendingBackgroundWork?: Effect.Effect; /** Root-subscription probe; replacement checks opt in to subagents owned by the live process. */ readonly hasPendingBackgroundWorkForThread?: ( diff --git a/packages/provider-core/src/server/failure.ts b/packages/provider-core/src/server/failure.ts index edaf0aa97a743..578345d887e54 100644 --- a/packages/provider-core/src/server/failure.ts +++ b/packages/provider-core/src/server/failure.ts @@ -12,6 +12,7 @@ import type { } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Cause from "effect/Cause"; +import * as Predicate from "effect/Predicate"; import * as IdAllocator from "./IdAllocator.ts"; import type { ProviderAdapterFailureReason } from "./ProviderAdapter.ts"; @@ -35,6 +36,19 @@ export class ContextHandoffBackgroundWorkError extends Schema.TaggedError + PROMPT_REFUSAL_TAGS.some((tag) => Predicate.isTagged(error, tag)) || + (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); + export const MAX_PROVIDER_FAILURE_MESSAGE_LENGTH = 4_096; const MAX_PROVIDER_FAILURE_CODE_LENGTH = 128; From dbbec135de8ed2c1203216e267a59167945d81ed Mon Sep 17 00:00:00 2001 From: Chinoman10 <8300763+Chinoman10@users.noreply.github.com> Date: Fri, 6 Nov 2026 08:46:14 +0000 Subject: [PATCH 3/9] fix(server): wake late subscribers after runtime teardown Wake an idle event source from the exact runtime close signal before event processing, preserving guarded reconciliation writes and runtimes without an external close observer. Cover retained-runtime startup after completed release and subscription between teardown broadcast and a held scope finalizer. Both must settle failed without native calls, retained ownership, or an open subscription. --- .../ProviderStartStop.integration.test.ts | 50 ++++++++++++++++++- .../orchestration-v2/RunExecutionService.ts | 7 ++- 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts index d9710ecfcd7bd..379c59e5f05f5 100644 --- a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts @@ -57,9 +57,12 @@ it.effect.each([ "manager-release-interrupt", "provider-stopped", "stream-ended", + "released-before-subscribe", + "released-during-subscribe", ] as const)("Stop crosses native start: %s", (scenario) => Effect.scoped( Effect.gen(function* () { + const scope = yield* Effect.scope; const cwd = yield* checkpointWorkspace(`start-stop-${scenario}`); const events = yield* Queue.unbounded(); const entered = yield* Deferred.make(); @@ -76,6 +79,12 @@ it.effect.each([ const releaseTerminal = yield* Deferred.make(); const ownedWait = yield* Deferred.make(); const beforeOwnership = yield* Deferred.make(); + const closeHeld = yield* Deferred.make(); + const releaseClose = yield* Deferred.make(); + const lateSubscribed = yield* Deferred.make(); + const lateSubscriptionClosed = yield* Deferred.make(); + const lateSubscription = + scenario === "released-before-subscribe" || scenario === "released-during-subscribe"; const driver = ProviderDriverKind.make("codex"); const instanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId, model: "synthetic" }; @@ -97,6 +106,12 @@ it.effect.each([ planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), openSession: (input) => Effect.gen(function* () { + if (scenario === "released-during-subscribe") + yield* Effect.addFinalizer(() => + Deferred.succeed(closeHeld, undefined).pipe( + Effect.andThen(Deferred.await(releaseClose)), + ), + ); const now = yield* DateTime.now; const startTurn: ProviderAdapter.ProviderAdapterV2SessionRuntime["startTurn"] = ( turn, @@ -281,6 +296,7 @@ it.effect.each([ const outbox = yield* EffectOutbox.EffectOutboxV2; const execution = yield* RunExecutionService.RunExecutionServiceV2; const threads = yield* ThreadManagementService.ThreadManagementService; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; yield* orchestrator.dispatch({ type: "thread.create", commandId: CommandId.make("create"), @@ -310,6 +326,33 @@ it.effect.each([ const originalStart = execution.startRootRun; let staleInput: RunExecutionService.RunExecutionServiceV2StartRootRunInput | undefined; using _beforeOwnership = vi.spyOn(execution, "startRootRun").mockImplementation((input) => { + if (lateSubscription) + return Effect.gen(function* () { + const released = yield* manager + .release({ providerSessionId: input.providerSessionId, reason: "manual_shutdown" }) + .pipe(Effect.orDie, Effect.forkIn(scope)); + if (scenario === "released-before-subscribe") yield* Fiber.join(released); + else yield* Deferred.await(closeHeld); + const startup = yield* originalStart({ + ...input, + session: { + ...input.session, + subscribeEvents: input.session.subscribeEvents!.pipe( + Effect.map((subscription) => ({ + ...subscription, + close: subscription.close.pipe( + Effect.tap(() => Deferred.succeed(lateSubscriptionClosed, undefined)), + ), + })), + Effect.tap(() => Deferred.succeed(lateSubscribed, undefined)), + ), + }, + }).pipe(Effect.forkIn(scope)); + yield* Deferred.await(lateSubscribed); + yield* Deferred.succeed(releaseClose, undefined); + yield* Fiber.join(released); + yield* Fiber.join(startup); + }); if (scenario === "unsupported-compact") { const session = { ...input.session }; delete session.compactThread; @@ -341,7 +384,11 @@ it.effect.each([ }); yield* send("first"); const startup = yield* worker.drain(1).pipe(Effect.forkScoped); - if (scenario === "unsupported-compact" || scenario === "prompt-refusal") { + if ( + scenario === "unsupported-compact" || + scenario === "prompt-refusal" || + lateSubscription + ) { yield* Fiber.join(startup); const failed = yield* orchestrator.getThreadProjection(threadId); const run = failed.runs[0]!; @@ -359,6 +406,7 @@ it.effect.each([ ); assert.equal(starts, 0); assert.deepEqual(interrupts, []); + if (lateSubscription) assert.isTrue(yield* Deferred.isDone(lateSubscriptionClosed)); return; } if (scenario === "before-ownership") { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 390f62e3984dc..edca124a47585 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1554,7 +1554,12 @@ export const layer: Layer.Layer< // Ends once the root turn and the children it drives settle. Background commands such // as a dev server keep this consumer alive afterwards and must not pin preview focus. const endPreviewActivity = yield* previewAgentActivity.begin(input.run.threadId); - const providerEventFiber = yield* eventSubscription.events.pipe( + // Only an out-of-stream closure observer needs to wake an idle subscription. + const providerEvents = + input.session.onSessionClosed === undefined + ? eventSubscription.events + : eventSubscription.events.pipe(Stream.interruptWhen(Deferred.await(sessionClosed))); + const providerEventFiber = yield* providerEvents.pipe( Stream.mapEffect((event) => Effect.gen(function* () { if ( From 9954e9b8f72f54974b07ff4a3489c12434a5bba3 Mon Sep 17 00:00:00 2001 From: Chinoman10 <8300763+Chinoman10@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:39:53 +0000 Subject: [PATCH 4/9] fix(server): drain provider events before releasing startup ownership Retain session subscription termination atomically so buffered terminal and background events drain and late subscribers receive the same closure. Stop cancelling the ingestion source independently of its commit coordinator. Mark ACP turn configuration and context-handoff delivery failures at their definite pre-prompt boundary, preserving ambiguous native-start ownership. Include attachment refusal and process-bound recovery corrections, with late-subscription and provider-switch coverage. Verified pinned check:changed (499 tests) and targeted CI adapter/replay/batching suites (213 tests). Implemented with openai-codex/gpt-6.1-sol in Pi. --- .../ProviderRuntimeRecoveryService.test.ts | 1 + .../ProviderSessionManager.test.ts | 20 ++- .../ProviderSessionManager.ts | 132 ++++++++---------- .../ProviderStartStop.integration.test.ts | 19 ++- .../ProviderTurnStartService.ts | 10 +- .../orchestration-v2/RunExecutionService.ts | 11 +- .../ProviderSwitch.integration.test.ts | 16 ++- packages/provider-acp/src/server/adapter.ts | 16 ++- .../src/server/ProviderAdapter.ts | 12 ++ packages/provider-core/src/server/failure.ts | 4 + 10 files changed, 150 insertions(+), 91 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 642a642a80df1..a76b5e3ab5cd1 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -773,6 +773,7 @@ it.effect( assert.deepEqual(committedInput?.cancelUnsettledEffects?.effectTypes, [ "provider-turn.start", "provider-turn.interrupt", + "provider-turn.interrupt-start", "provider-turn.steer", "provider-turn.restart", "runtime-request.respond", diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 896de803a38f6..e3f25105f2c1b 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1946,8 +1946,10 @@ it.effect("ProviderSessionManagerV2 interrupts event subscriptions on server shu yield* manager.shutdown; - const exit = yield* bufferedSubscription.events.pipe(Stream.runCollect, Effect.exit); - assert.isTrue(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)); + for (const subscription of [bufferedSubscription, yield* runtime.subscribeEvents!]) { + const exit = yield* subscription.events.pipe(Stream.runCollect, Effect.exit); + assert.isTrue(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)); + } }); yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 }))); @@ -1977,7 +1979,10 @@ it.effect("ProviderSessionManagerV2 drains subscribers when the provider stops", runtimePolicy, }); const subscription = yield* runtime.subscribeEvents!; - const collected = yield* subscription.events.pipe(Stream.runCollect, Effect.forkScoped); + const closed = yield* Deferred.make(); + const unsubscribe = yield* runtime.onSessionClosed!( + Deferred.succeed(closed, undefined).pipe(Effect.asVoid), + ); const adapterQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId)); assert.isDefined(adapterQueue); const providerThreadId = idAllocator.derive.providerThread({ @@ -2009,11 +2014,15 @@ it.effect("ProviderSessionManagerV2 drains subscribers when the provider stops", }); yield* Queue.end(adapterQueue!); - const events = Array.from(yield* Fiber.join(collected)); + yield* Deferred.await(closed); + const events = Array.from(yield* subscription.events.pipe(Stream.runCollect)); assert.deepEqual( events.map((event) => event.type), ["turn.terminal", "provider_session.updated"], ); + const late = yield* runtime.subscribeEvents!; + assert.deepEqual(Array.from(yield* late.events.pipe(Stream.runCollect)), []); + yield* unsubscribe; assert.isTrue(Option.isNone(yield* manager.get(providerSessionId))); assert.equal((yield* Ref.get(state)).closeCount, 1); }); @@ -3055,6 +3064,9 @@ it.effect("ProviderSessionManagerV2 releases sessions when provider event stream const projection = yield* projectionStore.getThreadProjection(threadId); assert.isTrue(Option.isNone(liveSession)); + const late = yield* runtime.subscribeEvents!; + const lateExit = yield* late.events.pipe(Stream.runCollect, Effect.exit); + assert.isTrue(Exit.isFailure(lateExit) && !Cause.hasInterruptsOnly(lateExit.cause)); assert.equal(runtimeState.closeCount, 1); assert.equal(projection.providerSessions.at(-1)?.status, "error"); assert.equal( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index fa1866477e4fb..9ec3e2b10a8cf 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -223,9 +223,7 @@ interface LiveSessionEntry { readonly supportsMultipleProviderThreads: boolean; readonly runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime; readonly exposedRuntime: ProviderAdapter.ProviderAdapterV2SessionRuntime; - readonly eventSubscribers: Ref.Ref< - ReadonlyMap> - >; + readonly eventSubscribers: Ref.Ref; readonly closeObservers: Ref.Ref> | null>; readonly requestEventPermit: Semaphore.Semaphore; readonly scope: Scope.Closeable; @@ -249,6 +247,16 @@ type ProviderSessionEventSignal = readonly cause: Cause.Cause; }; +type ProviderSessionEventTermination = + | { readonly type: "end" } + | { readonly type: "interrupt" } + | Extract; + +interface ProviderSessionSubscribers { + readonly queues: ReadonlyMap>; + readonly terminal: ProviderSessionEventTermination | null; +} + export interface ProviderSessionManagerV2LayerOptions { readonly idleTimeoutMs?: number; readonly maxIdlePinMs?: number; @@ -519,64 +527,52 @@ export const layerWithOptions = ( ); const publishToSubscribers = ( - subscribers: Ref.Ref< - ReadonlyMap> - >, + subscribers: LiveSessionEntry["eventSubscribers"], signal: ProviderSessionEventSignal, ) => Ref.get(subscribers).pipe( Effect.flatMap((current) => - Effect.forEach(current.values(), (queue) => Queue.offer(queue, signal), { + Effect.forEach(current.queues.values(), (queue) => Queue.offer(queue, signal), { discard: true, }), ), ); - const failSubscribers = Effect.fn("ProviderSessionManager.failSubscribers")(function* ( - entry: LiveSessionEntry, - detail: string, - ) { - const error = new ProviderAdapter.ProviderAdapterEventStreamError({ - driver: entry.runtime.driver, - providerSessionId: entry.runtime.providerSessionId, - cause: detail, - }); - const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map()); - yield* Effect.forEach( - subscribers.values(), - (queue) => - Queue.offer(queue, { - type: "failure", - cause: Cause.fail(error), - }), - { discard: true }, - ); - }); + const terminateQueue = ( + queue: Queue.Queue, + terminal: ProviderSessionEventTermination, + ) => + terminal.type === "interrupt" + ? Queue.clear(queue).pipe(Effect.andThen(Queue.interrupt(queue)), Effect.asVoid) + : (terminal.type === "failure" ? Queue.offer(queue, terminal) : Effect.void).pipe( + Effect.andThen(Queue.end(queue)), + Effect.asVoid, + ); - // Server shutdown interrupts subscribers rather than ending them: a - // cleanly ended stream without a terminal fails its run, while an - // interrupted one leaves run settlement to restart recovery. - const interruptSubscribers = Effect.fn("ProviderSessionManager.interruptSubscribers")( - function* (entry: LiveSessionEntry) { - const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map()); - yield* Effect.forEach( - subscribers.values(), - (queue) => Queue.clear(queue).pipe(Effect.andThen(Queue.interrupt(queue))), - { discard: true }, + const terminateSubscribers = Effect.fn("ProviderSessionManager.terminateSubscribers")( + function* (entry: LiveSessionEntry, terminal: ProviderSessionEventTermination) { + const queues = yield* Ref.modify(entry.eventSubscribers, (current) => + current.terminal === null + ? [current.queues, { queues: new Map(), terminal }] + : [current.queues, current], ); + yield* Effect.forEach(queues.values(), (queue) => terminateQueue(queue, terminal), { + discard: true, + }); }, ); - // Server shutdown intentionally clears them; a provider-announced Stop - // must let consumers drain them before the stream completes. - const endSubscribers = Effect.fn("ProviderSessionManager.endSubscribers")(function* ( - entry: LiveSessionEntry, - ) { - const subscribers = yield* Ref.getAndSet(entry.eventSubscribers, new Map()); - yield* Effect.forEach(subscribers.values(), (queue) => Queue.end(queue), { - discard: true, + const failSubscribers = (entry: LiveSessionEntry, detail: string) => + terminateSubscribers(entry, { + type: "failure", + cause: Cause.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: entry.runtime.driver, + providerSessionId: entry.runtime.providerSessionId, + cause: detail, + }), + ), }); - }); const cancelIdleFiber = (fiber: Fiber.Fiber | null) => fiber === null ? Effect.void : Fiber.interrupt(fiber).pipe(Effect.ignore); @@ -1046,9 +1042,10 @@ export const layerWithOptions = ( yield* cancelIdleFiber(entry.idleFiber); } if (input.gracefulSubscribers === true) { - yield* endSubscribers(entry); + yield* terminateSubscribers(entry, { type: "end" }); } else if (input.reason === "server_shutdown") { - yield* interruptSubscribers(entry); + // Shutdown discards buffered work; interruption leaves settlement to recovery. + yield* terminateSubscribers(entry, { type: "interrupt" }); } else { yield* failSubscribers( entry, @@ -1677,24 +1674,24 @@ export const layerWithOptions = ( const makeEventSubscription = Effect.fn("ProviderSessionManager.makeEventSubscription")( function* ( - subscribers: Ref.Ref< - ReadonlyMap> - >, + subscribers: LiveSessionEntry["eventSubscribers"], ): Effect.fn.Return { const queue = yield* Queue.unbounded(); const subscriberId = yield* Ref.getAndUpdate(nextSubscriberId, (value) => value + 1); - yield* Ref.update(subscribers, (current) => { - const updated = new Map(current); - updated.set(subscriberId, queue); - return updated; + const terminal = yield* Ref.modify(subscribers, (current) => { + if (current.terminal !== null) return [current.terminal, current] as const; + const queues = new Map(current.queues); + queues.set(subscriberId, queue); + return [null, { ...current, queues }] as const; }); + if (terminal !== null) yield* terminateQueue(queue, terminal); const close = Ref.modify(subscribers, (current) => { - if (!current.has(subscriberId)) { + if (!current.queues.has(subscriberId)) { return [false, current] as const; } - const updated = new Map(current); - updated.delete(subscriberId); - return [true, updated] as const; + const queues = new Map(current.queues); + queues.delete(subscriberId); + return [true, { ...current, queues }] as const; }).pipe( Effect.flatMap((removed) => removed @@ -1716,9 +1713,7 @@ export const layerWithOptions = ( const decorateRuntime = ( runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime, - eventSubscribers: Ref.Ref< - ReadonlyMap> - >, + eventSubscribers: LiveSessionEntry["eventSubscribers"], closeObservers: LiveSessionEntry["closeObservers"], ): ProviderAdapter.ProviderAdapterV2SessionRuntime => { const providerSessionId = runtime.providerSessionId; @@ -2116,11 +2111,7 @@ export const layerWithOptions = ( cause: "Provider event stream ended unexpectedly.", }), ); - yield* publishToSubscribers(entry.eventSubscribers, { - type: "failure", - cause, - }); - yield* Ref.set(entry.eventSubscribers, new Map()); + yield* terminateSubscribers(entry, { type: "failure", cause }); // lastError is the thread banner, so it gets the same bounded // message as a failed turn; the raw cause survives only in the log. yield* Effect.logWarning("orchestration-v2.provider-session-runtime-failed", { @@ -2279,9 +2270,10 @@ export const layerWithOptions = ( attributes: { provider: adapter.driver, operation: "open" }, }), ); - const eventSubscribers = yield* Ref.make< - ReadonlyMap> - >(new Map()); + const eventSubscribers = yield* Ref.make({ + queues: new Map(), + terminal: null, + }); const closeObservers = yield* Ref.make> | null>( new Set(), ); diff --git a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts index 379c59e5f05f5..434aa4aec8246 100644 --- a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts @@ -53,6 +53,7 @@ it.effect.each([ "identity-timeout-force", "unsupported-compact", "prompt-refusal", + "attachment-refusal", "manager-release", "manager-release-interrupt", "provider-stopped", @@ -117,16 +118,23 @@ it.effect.each([ turn, ) => Effect.gen(function* () { - if (scenario === "prompt-refusal") { + if (scenario === "prompt-refusal" || scenario === "attachment-refusal") { return yield* new ProviderAdapter.ProviderAdapterTurnStartError({ driver, threadId: turn.threadId, providerThreadId: turn.providerThread.id, runId: turn.runId, - cause: { - _tag: "ClaudeAppendSystemPromptFileError", - path: "/task-owned/missing-file", - }, + cause: + scenario === "attachment-refusal" + ? new ProviderAdapter.ProviderAdapterProtocolError({ + driver, + reason: "attachment_unreadable", + detail: "Synthetic unreadable attachment before prompt delivery", + }) + : { + _tag: "ClaudeAppendSystemPromptFileError", + path: "/task-owned/missing-file", + }, }); } yield* Deferred.succeed(entered, undefined); @@ -387,6 +395,7 @@ it.effect.each([ if ( scenario === "unsupported-compact" || scenario === "prompt-refusal" || + scenario === "attachment-refusal" || lateSubscription ) { yield* Fiber.join(startup); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 1a8c8e48a3941..9766e575e902b 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1582,7 +1582,15 @@ export const layer: Layer.Layer< }), }), persist: persistHandoff, - }); + }).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterTurnPreparationError({ + driver: session.driver, + cause, + }), + ), + ); if (!(yield* isCurrentAttemptInStatus("running"))) return; const start = compact ? session.compactThread! : session.startTurn; // Like history, the side chat context waits for a turn that sends a prompt. diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index af45cb93421d9..a0ae284919fdd 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1763,23 +1763,16 @@ export const layer: Layer.Layer< }), ), ); - // Only an out-of-stream closure observer needs to wake an idle subscription. - const providerEvents = - input.session.onSessionClosed === undefined - ? rawProviderEvents - : rawProviderEvents.pipe(Stream.interruptWhen(Deferred.await(sessionClosed))); const providerEventFiber = yield* eventSink .consumeProviderEvents( - providerEvents, + rawProviderEvents, (event) => Effect.gen(function* () { yield* ingestEvent(yield* routeEvent(event)); }), Effect.gen(function* () { if (yield* agentTurnsSettled) yield* endPreviewActivity; - return ( - (yield* Deferred.isDone(sessionClosed)) || (yield* shouldStopProviderEventIngestion) - ); + return yield* shouldStopProviderEventIngestion; }), Ref.get(rootTerminalSeen).pipe(Effect.map((seen) => !seen)), ) diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index e9b2636998386..888fe8322ada1 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -249,7 +249,21 @@ function makeTestAdapter(input: { input.failStartOnce !== undefined && (yield* Ref.getAndSet(input.failStartOnce, false)) ) - return yield* unimplemented(input.driver, "turn start failed after injection"); + return yield* unimplemented(input.driver, "turn start failed after injection").pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterTurnStartError({ + driver: input.driver, + threadId: turnInput.threadId, + providerThreadId: turnInput.providerThread.id, + runId: turnInput.runId, + cause: new ProviderAdapter.ProviderAdapterTurnPreparationError({ + driver: input.driver, + cause, + }), + }), + ), + ); if ( input.refuseStarts !== undefined && (yield* Ref.getAndUpdate(input.refuseStarts, (left) => Math.max(0, left - 1))) > 0 diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index f10be3c952dc8..fcf1a888b9145 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -8052,7 +8052,16 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const activated = yield* activateSession(requestedSessionId, turnInput.threadId); yield* Ref.set(activeSessionId, activated.sessionId); yield* setActiveSessionSetup(activated); - yield* configureSession(activated, turnInput.modelSelection, turnInput.runtimePolicy); + yield* configureSession( + activated, + turnInput.modelSelection, + turnInput.runtimePolicy, + ).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterTurnPreparationError({ driver, cause }), + ), + ); yield* Ref.set(activeSelection, turnInput.modelSelection); yield* Ref.set(activeInteractionMode, turnInput.runtimePolicy.interactionMode); } else { @@ -8074,6 +8083,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( currentSessionSetup, turnInput.modelSelection, turnInput.runtimePolicy, + ).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterTurnPreparationError({ driver, cause }), + ), ); yield* Ref.set(activeSelection, turnInput.modelSelection); yield* Ref.set(activeInteractionMode, turnInput.runtimePolicy.interactionMode); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 7a3bfe440992b..4a76f097dc656 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -306,6 +306,18 @@ export class ProviderAdapterForkThreadError extends Schema.TaggedError()( + "ProviderAdapterTurnPreparationError", + { + driver: ProviderDriverKind, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return `Failed to prepare a turn on ${this.driver}.`; + } +} + export class ProviderAdapterTurnStartError extends Schema.TaggedError()( "ProviderAdapterTurnStartError", { diff --git a/packages/provider-core/src/server/failure.ts b/packages/provider-core/src/server/failure.ts index 578345d887e54..186819b8c0962 100644 --- a/packages/provider-core/src/server/failure.ts +++ b/packages/provider-core/src/server/failure.ts @@ -46,7 +46,11 @@ const PROMPT_REFUSAL_TAGS = [ ] as const; export const refusedBeforePrompt = (error: unknown): boolean => + Predicate.isTagged(error, "ProviderAdapterTurnPreparationError") || PROMPT_REFUSAL_TAGS.some((tag) => Predicate.isTagged(error, tag)) || + (Predicate.isTagged(error, "ProviderAdapterProtocolError") && + Predicate.hasProperty(error, "reason") && + error.reason === "attachment_unreadable") || (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); export const MAX_PROVIDER_FAILURE_MESSAGE_LENGTH = 4_096; From bac71650033ae560a28f7bf51b997b399b75ba9a Mon Sep 17 00:00:00 2001 From: Chinoman10 <8300763+Chinoman10@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:57:06 +0000 Subject: [PATCH 5/9] refactor(server): type native start ownership failures Carry the exact thread, run and attempt plus missing/superseded state in a private tagged domain error instead of rejecting with bare strings. Keep interruption timeout classification and control behavior unchanged. Assert structured missing-owner diagnostics in the existing pre-ownership Stop regression. --- .../ProviderStartStop.integration.test.ts | 14 +++++++++++++ .../orchestration-v2/RunExecutionService.ts | 20 +++++++++++++++++-- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts index 434aa4aec8246..0724c2a71fdd0 100644 --- a/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts @@ -420,6 +420,20 @@ it.effect.each([ } if (scenario === "before-ownership") { yield* Deferred.await(beforeOwnership); + const pending = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const missing = yield* execution + .interruptOwnedStart({ + threadId, + runId: pending.id, + attemptId: pending.activeAttemptId!, + }) + .pipe(Effect.flip); + assert.isFalse(missing.unresponsive); + assert.nestedPropertyVal(missing, "cause._tag", "NativeStartOwnershipError"); + assert.nestedPropertyVal(missing, "cause.state", "missing"); + assert.nestedPropertyVal(missing, "cause.threadId", threadId); + assert.nestedPropertyVal(missing, "cause.runId", pending.id); + assert.nestedPropertyVal(missing, "cause.attemptId", pending.activeAttemptId); yield* orchestrator.dispatch({ type: "thread.stop", commandId: CommandId.make("stop"), diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index a0ae284919fdd..7e0ffab042096 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -608,6 +608,22 @@ type StartInterruptTarget = { readonly providerTurnId: ProviderTurnId; }; +class NativeStartOwnershipError extends Schema.TaggedError()( + "NativeStartOwnershipError", + { + threadId: ThreadId, + runId: RunId, + attemptId: RunAttemptId, + state: Schema.Literals(["missing", "superseded"]), + }, +) { + override get message(): string { + return this.state === "missing" + ? "Missing unresolved native-start owner" + : "Native-start binding was superseded"; + } +} + export class RunExecutionInterruptError extends Schema.TaggedError()( "RunExecutionInterruptError", { @@ -1001,7 +1017,7 @@ export const layer: Layer.Layer< .every((turn) => isTerminalProviderTurnStatus(turn.status)) ) return; - return yield* Effect.fail("Missing unresolved native-start owner"); + return yield* new NativeStartOwnershipError({ ...key, state: "missing" }); } const accepted = yield* Deferred.await(owner.accepted).pipe( Effect.raceFirst(Deferred.await(owner.terminal).pipe(Effect.as(undefined))), @@ -1023,7 +1039,7 @@ export const layer: Layer.Layer< binding.nativeThreadRef?.nativeId !== owner.input.providerThread.nativeThreadRef?.nativeId ) { - return yield* Effect.fail("Native-start binding was superseded"); + return yield* new NativeStartOwnershipError({ ...key, state: "superseded" }); } if (yield* Deferred.isDone(owner.terminal)) return; const turn = owner.turn ?? accepted; From a73a17034c3b9f09d5cc4e2535e62cac82585818 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 10 Nov 2026 12:21:50 +0000 Subject: [PATCH 6/9] test(server): keep the run execution layer open until ingestion settles --- .../RunExecutionService.test.ts | 23 ++++++++----------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index b2c03952a49b9..c7a5dd6cf3171 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1618,7 +1618,7 @@ it.effect("ingests new member threads of an owned child after the root turn term } as ProviderAdapter.ProviderAdapterV2Event, ]; - yield* Effect.gen(function* () { + const observed = yield* Effect.gen(function* () { const runExecution = yield* RunExecutionService.RunExecutionServiceV2; yield* runExecution.startRootRun({ commandId: CommandId.make("command:run-execution-late-child"), @@ -1669,11 +1669,8 @@ it.effect("ingests new member threads of an owned child after the root turn term }, }, }); + return yield* Deferred.await(childMessageIngested).pipe(Effect.timeoutOption("2 seconds")); }).pipe(Effect.provide(layerTest)); - - const observed = yield* Deferred.await(childMessageIngested).pipe( - Effect.timeoutOption("2 seconds"), - ); assert.isTrue(Option.isSome(observed), "child message was not ingested after root terminal"); assert.deepEqual(yield* Ref.get(order), ["root-finalized", "member-thread", "child-message"]); // The run's message started the root; the parent agent started the child's turn. @@ -1784,7 +1781,7 @@ it.effect("ends a nested native subagent and its thread when the root run is int }, ]; - yield* Effect.gen(function* () { + const finalizedEvents = yield* Effect.gen(function* () { const runExecution = yield* RunExecutionService.RunExecutionServiceV2; yield* runExecution.startRootRun({ commandId: CommandId.make(`command:${key}`), @@ -1819,9 +1816,10 @@ it.effect("ends a nested native subagent and its thread when the root run is int sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false }, }, }); + return yield* Deferred.await(finalized); }).pipe(Effect.provide(layerTest)); - const ended = (yield* Deferred.await(finalized)).flatMap((event) => + const ended = finalizedEvents.flatMap((event) => event.type === "subagent.updated" || event.type === "node.updated" ? [[event.type, event.payload.id, event.payload.status]] : [], @@ -2265,7 +2263,7 @@ it.effect( nodeId: unrelatedChildNode.id, }; - yield* Effect.gen(function* () { + const closed = yield* Effect.gen(function* () { const runExecution = yield* RunExecutionService.RunExecutionServiceV2; yield* runExecution.startRootRun({ commandId: CommandId.make("command:subagent-interrupt-cascade"), @@ -2408,9 +2406,8 @@ it.effect( }, }, }); + return yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds")); }).pipe(Effect.provide(layerTest)); - - const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds")); assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish"); const events = yield* Ref.get(written); @@ -3930,11 +3927,11 @@ function runBackgroundItemScenario( }, }, }); + // The run's ingestion lives in the service scope, so await it before the layer closes. + // No timeout: tests advance the TestClock, which would fire it early. + yield* Deferred.await(ingestionDone); }).pipe(Effect.provide(layerTest)); - // No timeout: tests advance the TestClock, which would fire it early. - yield* Deferred.await(ingestionDone); - return yield* Ref.get(observed); }); } From 062b32f37870d6f76f1e8595fde60ae1e3f3257b Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 10 Nov 2026 12:44:41 +0000 Subject: [PATCH 7/9] test(server): await run ingestion milestones without a timeout --- .../src/orchestration-v2/RunExecutionService.test.ts | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index c7a5dd6cf3171..cfc4d12740bf6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1618,7 +1618,7 @@ it.effect("ingests new member threads of an owned child after the root turn term } as ProviderAdapter.ProviderAdapterV2Event, ]; - const observed = yield* Effect.gen(function* () { + yield* Effect.gen(function* () { const runExecution = yield* RunExecutionService.RunExecutionServiceV2; yield* runExecution.startRootRun({ commandId: CommandId.make("command:run-execution-late-child"), @@ -1669,9 +1669,8 @@ it.effect("ingests new member threads of an owned child after the root turn term }, }, }); - return yield* Deferred.await(childMessageIngested).pipe(Effect.timeoutOption("2 seconds")); + yield* Deferred.await(childMessageIngested); }).pipe(Effect.provide(layerTest)); - assert.isTrue(Option.isSome(observed), "child message was not ingested after root terminal"); assert.deepEqual(yield* Ref.get(order), ["root-finalized", "member-thread", "child-message"]); // The run's message started the root; the parent agent started the child's turn. const recorded = yield* Ref.get(origins); @@ -2263,7 +2262,7 @@ it.effect( nodeId: unrelatedChildNode.id, }; - const closed = yield* Effect.gen(function* () { + yield* Effect.gen(function* () { const runExecution = yield* RunExecutionService.RunExecutionServiceV2; yield* runExecution.startRootRun({ commandId: CommandId.make("command:subagent-interrupt-cascade"), @@ -2406,9 +2405,8 @@ it.effect( }, }, }); - return yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds")); + yield* Deferred.await(ingestionDone); }).pipe(Effect.provide(layerTest)); - assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish"); const events = yield* Ref.get(written); const subagentEvents = events.flatMap((event) => From 945c5efb334d14e44438f9831e23cd629dc66b58 Mon Sep 17 00:00:00 2001 From: Chinoman10 <8300763+Chinoman10@users.noreply.github.com> Date: Tue, 10 Nov 2026 15:29:34 +0000 Subject: [PATCH 8/9] fix(server): settle force-stopped background work during checkpointing Use the existing active-run predicate and background settlement for a waiting run, retaining the exact-attempt guard and run-ordinal bounds. Leave checkpoint capture, root completion, and run finalization untouched. Add a regression that fails on the PR base, checks stale-attempt isolation, and preserves the pending checkpoint effect. Verified check:changed, lifecycle/finalization suites, and fresh independent review. --- .../src/orchestration-v2/Orchestrator.ts | 14 +- .../src/orchestration-v2/ThreadStop.test.ts | 128 ++++++++++++++++++ 2 files changed, 140 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e61d71e2a408a..d946a3844188e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11871,8 +11871,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } if ( incoming.type === "run.force-stop.settle" && - (run.activeAttemptId !== incoming.attemptId || - !["preparing", "starting", "running"].includes(run.status)) + (run.activeAttemptId !== incoming.attemptId || !isActiveRun(run)) ) return undefined; const ownership = @@ -11888,6 +11887,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Option.isSome(ownership) && (providerTurn === undefined || ownership.value.startPending); const now = yield* DateTime.now; + if (incoming.type === "run.force-stop.settle" && run.status === "waiting") { + yield* settleBackgroundWork({ + command, + events, + projection, + stoppedProviderThreadId: providerThread.id, + throughRunOrdinal: run.ordinal, + now, + }); + return undefined; + } const completionMessage = projection.messages.find( (candidate) => candidate.id === run.userMessageId, ); diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 6b3ea08a86449..5d460e4bb7e71 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1,5 +1,6 @@ import { assert, it } from "@effect/vitest"; import { + CheckpointScopeId, CommandId, ContextTransferId, EventId, @@ -13,6 +14,7 @@ import { ProviderThreadId, ProviderTurnId, RuntimeRequestId, + RunAttemptId, ScheduledTaskId, ThreadId, TurnItemId, @@ -21,6 +23,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; @@ -1553,6 +1556,131 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) }).pipe(Effect.provide(layerTest)), ); +it.effect("Force stop settles background work without interrupting checkpoint capture", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const threadId = ThreadId.make("thread:force-stop-checkpoint"); + yield* createWatchingThread(threadId, 21); + yield* send(threadId, "work", "start_immediately"); + yield* delegate(threadId, "native-background"); + const before = yield* orchestrator.getThreadProjection(threadId); + const run = before.runs[0]!; + const attempt = before.attempts[0]!; + const root = before.nodes.find((node) => node.id === run.rootNodeId)!; + const subagent = before.subagents[0]!; + const item = before.turnItems.find((item) => item.type === "subagent")!; + const now = yield* DateTime.now; + const scopeId = CheckpointScopeId.make("scope:force-stop-checkpoint"); + const providerTurnId = ProviderTurnId.make("turn:force-stop-checkpoint"); + const apply = (event: Omit) => + projections.apply({ + ...event, + id: EventId.make(`event:force-stop-checkpoint:${event.type}`), + threadId, + } as OrchestrationV2DomainEvent); + yield* apply({ + type: "run.updated", + runId: run.id, + occurredAt: now, + payload: { ...run, status: "waiting", startedAt: now }, + }); + yield* apply({ + type: "run-attempt.updated", + occurredAt: now, + payload: { ...attempt, status: "completed", startedAt: now, completedAt: now }, + }); + yield* apply({ + type: "node.updated", + nodeId: root.id, + occurredAt: now, + payload: { ...root, status: "completed", checkpointScopeId: scopeId, completedAt: now }, + }); + yield* apply({ + type: "provider-turn.updated", + occurredAt: now, + payload: { + id: providerTurnId, + providerThreadId: run.providerThreadId!, + nodeId: root.id, + runAttemptId: attempt.id, + nativeTurnRef: null, + ordinal: 1, + status: "completed", + startedAt: now, + completedAt: now, + }, + }); + yield* apply({ + type: "subagent.updated", + nodeId: subagent.id, + occurredAt: now, + payload: { ...subagent, origin: "provider_native", childThreadId: null, status: "running" }, + }); + yield* apply({ + type: "turn-item.updated", + runId: run.id, + occurredAt: now, + payload: { + ...item, + origin: "provider_native", + childThreadId: null, + providerThreadId: run.providerThreadId, + providerTurnId, + status: "running", + }, + }); + const captureId = `effect:checkpoint.capture:${run.id}`; + yield* outbox.enqueue([ + { + id: captureId, + commandId: CommandId.make("checkpoint:force-stop-checkpoint"), + threadId, + request: { type: "checkpoint.capture", runId: run.id, scopeId }, + }, + ]); + const captureBefore = Option.getOrThrow(yield* outbox.get(captureId)); + assert.equal(captureBefore.status, "pending"); + const waiting = yield* orchestrator.getThreadProjection(threadId); + const settle = { + type: "run.force-stop.settle" as const, + threadId, + runId: run.id, + attemptId: attempt.id, + reason: "Force stopped by user", + }; + const stale = yield* orchestrator.dispatch({ + ...settle, + commandId: CommandId.make("force-stop-checkpoint:stale-attempt"), + attemptId: RunAttemptId.make("attempt:stale"), + }); + assert.lengthOf(stale.storedEvents, 0); + assert.deepEqual(yield* orchestrator.getThreadProjection(threadId), waiting); + + yield* orchestrator.dispatch({ + ...settle, + commandId: CommandId.make("force-stop-checkpoint:settle"), + }); + const settled = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + settled.turnItems.find((candidate) => candidate.id === item.id)?.status, + "interrupted", + ); + assert.equal( + settled.subagents.find((candidate) => candidate.id === subagent.id)?.status, + "interrupted", + ); + assert.deepEqual(settled.runs, waiting.runs); + assert.deepEqual(settled.attempts, waiting.attempts); + assert.deepEqual( + settled.nodes.find((node) => node.id === root.id), + waiting.nodes.find((node) => node.id === root.id), + ); + assert.deepEqual(Option.getOrThrow(yield* outbox.get(captureId)), captureBefore); + }).pipe(Effect.provide(layerTest)), +); + it.effect("an unanswered Stop is flagged until Force stop ends the run with its own reason", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From fbaa763edd9b0102bbd9b0846dbbaf42af755689 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 10 Nov 2026 23:26:56 +0000 Subject: [PATCH 9/9] fix(server): force stop without an attempt interrupts the run --- .../ThreadManagementService.ts | 26 +++++++++++++------ 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 514737dc9db92..60e0cb697ad6d 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1250,14 +1250,24 @@ const make = Effect.gen(function* () { yield* providerTurn === undefined && !active ? Effect.void : providerTurn === undefined - ? orchestrator.dispatch({ - type: "run.force-stop.settle", - commandId, - threadId: input.threadId, - runId: run.id, - attemptId: run.activeAttemptId!, - reason: FORCE_STOP_REASON, - }) + ? orchestrator.dispatch( + run.activeAttemptId === null + ? { + type: "run.interrupt", + commandId, + threadId: input.threadId, + runId: run.id, + reason: FORCE_STOP_REASON, + } + : { + type: "run.force-stop.settle", + commandId, + threadId: input.threadId, + runId: run.id, + attemptId: run.activeAttemptId, + reason: FORCE_STOP_REASON, + }, + ) : orchestrator.dispatch({ type: "thread.background-work.settle", commandId,