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 3be77bba74c7c..083888b3de5dc 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -31,6 +31,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"; @@ -88,6 +89,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 0c42ac73ff9cb..64097de0608ad 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -171,6 +171,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 acf3cc27f6211..7fbef560758ca 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -2251,12 +2251,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 3157f55cacfa6..a78592a564dea 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -610,6 +610,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; @@ -1159,6 +1161,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 = @@ -11450,22 +11453,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; @@ -11758,10 +11765,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, [ @@ -11826,7 +11838,35 @@ 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 || !isActiveRun(run)) + ) + 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; + 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, ); @@ -11930,6 +11970,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") @@ -11961,7 +12079,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", @@ -12077,9 +12198,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; }; @@ -13461,6 +13584,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; }; @@ -13478,6 +13602,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; } @@ -13678,6 +13803,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) { @@ -13817,6 +13945,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; @@ -13964,6 +14093,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio isUnobservedAcknowledgement(command) || 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. @@ -14721,6 +14852,7 @@ export const layer: Layer.Layer< | ProjectStore.ProjectStoreV2 | ProviderAdapterRegistryV2 | ProviderSessionManagerV2 + | RunExecutionService.RunExecutionServiceV2 | ProviderSwitchServiceV2 | ProjectionStoreV2 | RuntimePolicyV2 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 b4fc69727c1de..2bd3e0d0b845a 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1947,8 +1947,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 }))); @@ -1978,7 +1980,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({ @@ -2010,11 +2015,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); }); @@ -3113,6 +3122,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 0fb31e81a0732..6e9d5aa04d248 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -234,9 +234,8 @@ 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; readonly idleGeneration: number; @@ -259,6 +258,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; @@ -547,64 +556,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); @@ -1029,9 +1026,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( @@ -1091,19 +1090,28 @@ 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, 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, @@ -1888,24 +1896,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 @@ -1927,9 +1935,8 @@ export const layerWithOptions = ( const decorateRuntime = ( runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime, - eventSubscribers: Ref.Ref< - ReadonlyMap> - >, + eventSubscribers: LiveSessionEntry["eventSubscribers"], + closeObservers: LiveSessionEntry["closeObservers"], ): ProviderAdapter.ProviderAdapterV2SessionRuntime => { const providerSessionId = runtime.providerSessionId; const subscribeEvents = makeEventSubscription(eventSubscribers); @@ -1950,6 +1957,22 @@ export const layerWithOptions = ( return runtime.providerSession; }, 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)), ), @@ -2340,11 +2363,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", { @@ -2507,10 +2526,14 @@ export const layerWithOptions = ( attributes: { provider: adapter.driver, operation: "open" }, }), ); - const eventSubscribers = yield* Ref.make< - ReadonlyMap> - >(new Map()); - const exposedRuntime = decorateRuntime(runtime, eventSubscribers); + const eventSubscribers = yield* Ref.make({ + queues: new Map(), + terminal: null, + }); + 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(), @@ -2526,6 +2549,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 new file mode 100644 index 0000000000000..0724c2a71fdd0 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderStartStop.integration.test.ts @@ -0,0 +1,804 @@ +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 Cause from "effect/Cause"; +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 ProviderSessionManager from "./ProviderSessionManager.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", + "unsupported-compact", + "prompt-refusal", + "attachment-refusal", + "manager-release", + "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(); + 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 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" }; + 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* () { + 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, + ) => + Effect.gen(function* () { + if (scenario === "prompt-refusal" || scenario === "attachment-refusal") { + return yield* new ProviderAdapter.ProviderAdapterTurnStartError({ + driver, + threadId: turn.threadId, + providerThreadId: turn.providerThread.id, + runId: turn.runId, + 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); + 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; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + 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 (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; + 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)); + }); + 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 === "unsupported-compact" || + scenario === "prompt-refusal" || + scenario === "attachment-refusal" || + lateSubscription + ) { + 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, []); + if (lateSubscription) assert.isTrue(yield* Deferred.isDone(lateSubscriptionClosed)); + return; + } + 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"), + 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")!; + 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) => + 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"); + 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( + 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 98652036118fc..d402684f6d1c2 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; @@ -271,19 +254,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"], { @@ -1596,7 +1586,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.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index e25b4c7ab556a..cfc4d12740bf6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -60,8 +60,15 @@ 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 mockEventSink = (implementation: Partial) => Layer.mock(EventSink.EventSinkV2)({ consumeProviderEvents: (source, process, after) => @@ -84,7 +91,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, @@ -899,7 +906,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, @@ -940,7 +947,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, @@ -1142,7 +1149,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, @@ -1262,7 +1269,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, @@ -1435,7 +1442,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, { @@ -1662,12 +1669,8 @@ it.effect("ingests new member threads of an owned child after the root turn term }, }, }); + yield* Deferred.await(childMessageIngested); }).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. const recorded = yield* Ref.get(origins); @@ -1705,7 +1708,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, @@ -1777,7 +1780,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}`), @@ -1812,9 +1815,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]] : [], @@ -2150,7 +2154,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, @@ -2401,11 +2405,9 @@ it.effect( }, }, }); + yield* Deferred.await(ingestionDone); }).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); const subagentEvents = events.flatMap((event) => event.type === "subagent.updated" ? [event] : [], @@ -3237,7 +3239,7 @@ function captureRootRunTermination(input: { } } }); - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( Layer.mock(McpAppModelContext.McpAppModelContext)({ @@ -3788,7 +3790,7 @@ function runBackgroundItemScenario( true, ) : undefined; - const layerTest = RunExecutionService.layer.pipe( + const layerTest = layerRunExecution.pipe( Layer.provide( Layer.mergeAll( McpAppModelContext.layerEmpty, @@ -3923,11 +3925,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); }); } diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index e981ab7362d2d..f836229943787 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 Queue from "effect/Queue"; @@ -48,9 +50,12 @@ 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, + refusedBeforePrompt, makeProviderFailure, makeProviderFailureTurnItem, } from "@t3tools/provider-core/server/failure"; @@ -563,6 +568,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; @@ -591,7 +607,67 @@ 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; +}; + +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", + { + 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; + unsubscribeClosed?: Effect.Effect; +} + 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; @@ -605,6 +681,8 @@ export class RunExecutionServiceV2 extends Context.Service< export const layer: Layer.Layer< RunExecutionServiceV2, never, + | ThreadCommandExecutor.ThreadCommandExecutor + | ProjectionStore.ProjectionStoreV2 | CheckpointService.CheckpointServiceV2 | EventSink.EventSinkV2 | IdAllocator.IdAllocatorV2 @@ -617,6 +695,23 @@ 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, + }; + 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; const idAllocator = yield* IdAllocator.IdAllocatorV2; @@ -894,7 +989,112 @@ 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* new NativeStartOwnershipError({ ...key, state: "missing" }); + } + 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* new NativeStartOwnershipError({ ...key, state: "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, + }) + .pipe( + Effect.andThen(Deferred.await(owner.terminal)), + Effect.raceFirst(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, @@ -1011,6 +1211,53 @@ 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(), + 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); @@ -1134,6 +1381,7 @@ export const layer: Layer.Layer< idleFlushClosed = true; } 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* ( @@ -1547,6 +1795,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 = assistantStreaming.filter( @@ -1597,9 +1847,25 @@ export const layer: Layer.Layer< yield* trackChildLifecycle(event, deliveredEvent !== null); yield* scheduleIdleFlush; }); + // Observe closure before the ingestion coordinator can block on guarded writes. + const rawProviderEvents = eventSubscription.events.pipe( + Stream.tap((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 providerEventFiber = yield* eventSink .consumeProviderEvents( - eventSubscription.events.pipe( + rawProviderEvents.pipe( Stream.merge(Stream.fromQueue(idleFlushTicks), { haltStrategy: "left" }), ), (event) => @@ -1661,6 +1927,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) => @@ -1749,11 +2016,14 @@ 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), Effect.ensuring(eventSubscription.close), - Effect.forkDetach, + Effect.forkIn(scope), ); // A provider turn is a sign that the session is still alive. Keep @@ -1765,16 +2035,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. @@ -1803,33 +2074,42 @@ 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 && compactThread !== undefined + ? compactThread(turnInput) + : 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(() => @@ -1839,51 +2119,105 @@ 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)) || refusedBeforePrompt(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* () { + 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, + 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 ( + owner.revoked || + (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.andThen(owner.unsubscribeClosed ?? Effect.void), + 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 742616f27bd3e..e255f0a5f874f 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)({ @@ -591,7 +598,7 @@ it.effect.each([ const dispatched: Array = []; // The notification starts after the completed run was observed, before settlement commits. let raced = false; - const layerTest = ThreadManagementService.layer.pipe( + const layerTest = layerThreadManagement.pipe( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ @@ -757,7 +764,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), @@ -838,7 +845,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 de12768ac7b17..60e0cb697ad6d 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"; @@ -576,6 +577,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; @@ -1216,6 +1218,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 }; @@ -1242,13 +1250,24 @@ const make = Effect.gen(function* () { yield* providerTurn === undefined && !active ? Effect.void : providerTurn === undefined - ? orchestrator.dispatch({ - type: "run.interrupt", - commandId, - threadId: input.threadId, - runId: run.id, - 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, @@ -1650,6 +1669,7 @@ const layerLegacyV1ThreadImporterNoop = Layer.succeed( export const layer: Layer.Layer< ThreadManagementService, never, + | RunExecutionService.RunExecutionServiceV2 | Orchestrator.OrchestratorV2 | ProviderSessionManager.ProviderSessionManagerV2 | ServerSettings.ServerSettingsService @@ -1662,6 +1682,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/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; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index ba1dadad1864c..8353082d444c6 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -151,6 +151,8 @@ const layerProviderAuthServiceProvided = ProviderAuthService.layer.pipe( const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, + ProjectionStore.layer, McpAppModelContext.layer, PreviewAgentActivity.layer, layerCheckpointServiceProvided, @@ -252,6 +254,7 @@ const layerThreadManagementProvided = ThreadManagementService.layerWithLegacyImp Layer.provide( Layer.mergeAll( layerOrchestratorProvided, + layerRunExecutionServiceProvided, layerLegacyV1ThreadImporterProvided, layerCommandReceiptStoreProvided, layerProviderSessionManagerProvided, @@ -323,6 +326,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 ae0d5ef346d34..9154787c4b79a 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -256,6 +256,7 @@ export function layerProviderReplay { @@ -289,6 +290,7 @@ export function layerWithRegistry( | EffectWorker.OrchestrationEffectWorkerV2 | EventSink.EventSinkV2 | ProviderSessionManager.ProviderSessionManagerV2 + | RunExecutionService.RunExecutionServiceV2 | CommandOutputRecorder.CommandOutputRecorder, Error | MigrationError | PlatformError.PlatformError | SqlError > { @@ -403,6 +405,7 @@ export function layerWithRegistry( const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( Layer.provide( Layer.mergeAll( + ThreadCommandExecutor.layer, McpAppModelContext.layerEmpty, PreviewAgentActivity.layer, layerCheckpointServiceProvided, @@ -524,6 +527,7 @@ export function layerWithRegistry( layerProviderSessionManagerProvided, layerProviderTurnControlServiceProvided, layerProviderTurnStartServiceProvided, + layerRunExecutionServiceProvided, layerRuntimeRequestServiceProvided, layerThreadTitleRegenerationTest, layerServerSettings, @@ -536,6 +540,7 @@ export function layerWithRegistry( ); const layerReplayRuntime = Layer.mergeAll( layerOrchestratorProvided, + layerRunExecutionServiceProvided, layerProviderSessionManagerProvided, layerEffectWorkerProvided, layerEventSinkProvided, 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 83eee8f66a5dc..8fd59fc411700 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -250,7 +250,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/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index fc38ffd4da9bb..9538db3fd7396 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3531,6 +3531,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"), diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index a2c2c429fa88c..c38b0d2aa6499 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -8083,7 +8083,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 { @@ -8105,6 +8114,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 0c43deb4f7804..360f512ad3118 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -323,6 +323,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", { @@ -633,6 +645,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..186819b8c0962 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,23 @@ export class ContextHandoffBackgroundWorkError extends Schema.TaggedError + 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; const MAX_PROVIDER_FAILURE_CODE_LENGTH = 128;