authorgravatar for saiansh2525@gmail.comYash Singh <saiansh2525@gmail.com> 2026-10-06 04:04:13-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-06 04:04:13-05:00
loga7a2230c33fbec9d6616a5de26f09e1fa8856c05
tree8df259fe3fa9332b330903604dc59c7000971570
parent0290c4a04e52ca4e93be7a6fb94c14bcc7b53728
signature Signed by PGP key B5690EEEBB952194

fix(orchestration-v2): let Stop recover stalled runs (#15442)

10 files changed, 623 insertions(+), 50 deletions(-)

apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts+193
......@@ -31,6 +31,7 @@ import type {
3131 ProviderAdapterV2TurnInput,
3232} from "./ProviderAdapter.ts";
3333import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
34import * as ProviderSessionManager from "./ProviderSessionManager.ts";
3435import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
3536import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts";
3637
......@@ -45,10 +46,17 @@ const stopEarlierBackgroundWork = ({
4546 failedStart = false,
4647 stopWithQueue,
4748 olderStart = false,
49 stalledRun,
4850}: {
4951 readonly failedStart?: boolean;
5052 readonly stopWithQueue?: "thread.stop" | "run.interrupt";
5153 readonly olderStart?: boolean;
54 readonly stalledRun?:
55 | "missing-session"
56 | "missing-session-terminal"
57 | "returned-interrupt"
58 | "returned-interrupt-terminal"
59 | "superseded-attempt";
5260}) =>
5361 Effect.scoped(
5462 Effect.gen(function* () {
......@@ -210,6 +218,181 @@ const stopEarlierBackgroundWork = ({
210218 },
211219 ],
212220 });
221 if (stalledRun !== undefined) {
222 const before = yield* orchestrator.getThreadProjection(threadId);
223 const run = before.runs[0]!;
224 const node = before.nodes.find((candidate) => candidate.id === run.rootNodeId)!;
225 const attempt = before.attempts[0]!;
226 if (stalledRun === "missing-session" || stalledRun === "missing-session-terminal") {
227 const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
228 const failed = yield* watch(
229 (event) => event.type === "run.updated" && event.payload.status === "failed",
230 );
231 yield* sessions.release({
232 providerSessionId: first.providerThread.providerSessionId!,
233 reason: "runtime_error",
234 });
235 yield* Fiber.join(failed);
236 // Disk exhaustion can lose these terminal writes. Restore that stale state.
237 yield* sink.write({
238 events: [
239 {
240 id: EventId.make("stale-run"),
241 type: "run.updated",
242 threadId,
243 occurredAt: now,
244 payload: run,
245 },
246 {
247 id: EventId.make("stale-node"),
248 type: "node.updated",
249 threadId,
250 occurredAt: now,
251 payload: node,
252 },
253 {
254 id: EventId.make("stale-attempt"),
255 type: "run-attempt.updated",
256 threadId,
257 occurredAt: now,
258 payload: attempt,
259 },
260 ],
261 });
262 }
263 const messageId = MessageId.make("partial-output");
264 const item = before.turnItems.find((candidate) => candidate.id === devServerId)!;
265 assert.ok(item.type === "command_execution");
266 const terminalProviderTurn = stalledRun.endsWith("-terminal");
267 if (terminalProviderTurn) {
268 yield* sink.write({
269 events: [
270 {
271 id: EventId.make("terminal-provider-turn"),
272 type: "provider-turn.updated",
273 threadId,
274 occurredAt: now,
275 payload: { ...codexTurn, status: "completed", completedAt: now },
276 },
277 ...(stalledRun === "missing-session-terminal"
278 ? [
279 {
280 id: EventId.make("completed-dev-server"),
281 type: "turn-item.updated" as const,
282 threadId,
283 occurredAt: now,
284 payload: { ...item, status: "completed" as const, completedAt: now },
285 },
286 ]
287 : []),
288 ],
289 });
290 }
291 yield* sink.write({
292 events: [
293 {
294 id: EventId.make("partial-message"),
295 type: "message.updated",
296 threadId,
297 runId: run.id,
298 occurredAt: now,
299 payload: {
300 id: messageId,
301 threadId,
302 runId: run.id,
303 nodeId: node.id,
304 role: "assistant",
305 text: "Partial output",
306 attachments: [],
307 streaming: true,
308 createdBy: "agent",
309 creationSource: "provider",
310 createdAt: now,
311 updatedAt: now,
312 },
313 },
314 {
315 id: EventId.make("partial-item"),
316 type: "turn-item.updated",
317 threadId,
318 runId: run.id,
319 occurredAt: now,
320 payload: {
321 ...item,
322 providerThreadId: codexTurn.providerThreadId,
323 providerTurnId: codexTurn.id,
324 id: TurnItemId.make("partial-output"),
325 type: "assistant_message",
326 messageId,
327 text: "Partial output",
328 streaming: true,
329 },
330 },
331 ],
332 });
333 if (stalledRun === "superseded-attempt") {
334 yield* sink.write({
335 events: [
336 {
337 id: EventId.make("new-attempt"),
338 type: "run.updated",
339 threadId,
340 occurredAt: now,
341 payload: { ...run, activeAttemptId: RunAttemptId.make("new-attempt") },
342 },
343 ],
344 });
345 }
346 yield* orchestrator.dispatch(
347 stalledRun === "superseded-attempt"
348 ? {
349 type: "thread.background-work.settle",
350 commandId: CommandId.make("late-settle"),
351 threadId,
352 providerThreadId: codexTurn.providerThreadId,
353 providerTurnId: codexTurn.id,
354 }
355 : {
356 type: "thread.stop",
357 commandId: CommandId.make("stop-stalled-run"),
358 threadId,
359 },
360 );
361 yield* worker.drain();
362 const after = yield* orchestrator.getThreadProjection(threadId);
363 const interrupted = stalledRun !== "superseded-attempt";
364 assert.equal(after.runs[0]?.status, interrupted ? "interrupted" : "running");
365 assert.equal(after.attempts[0]?.status, interrupted ? "interrupted" : "running");
366 assert.equal(
367 after.providerTurns[0]?.status,
368 terminalProviderTurn ? "completed" : interrupted ? "interrupted" : "running",
369 );
370 assert.equal(
371 after.nodes.find((candidate) => candidate.id === node.id)?.status,
372 interrupted ? "interrupted" : "running",
373 );
374 const output = after.messages.find((message) => message.id === messageId)!;
375 assert.equal(output.streaming, !interrupted);
376 assert.equal(output.text, "Partial output");
377 assert.equal(
378 after.turnItems.find((candidate) => candidate.id === devServerId)?.status,
379 stalledRun === "missing-session-terminal"
380 ? "completed"
381 : interrupted
382 ? "interrupted"
383 : "running",
384 );
385 assert.equal(
386 after.turnItems.find((candidate) => candidate.type === "assistant_message")?.status,
387 interrupted ? "interrupted" : "running",
388 );
389 assert.equal(
390 after.turnItems.filter((candidate) => candidate.type === "run_interrupt_result").length,
391 interrupted ? 1 : 0,
392 );
393 assert.isEmpty(after.runs.filter((candidate) => candidate.status === "waiting"));
394 return;
395 }
213396 const settled = yield* watch(
214397 (event) =>
215398 event.type === "run.updated" &&
......@@ -599,3 +782,13 @@ it.effect.each(["thread.stop", "run.interrupt"] as const)(
599782 "%s reaches later background work when an older run is starting",
600783 (stopType) => stopEarlierBackgroundWork({ stopWithQueue: stopType, olderStart: true }),
601784);
785
786it.effect.each([
787 "missing-session",
788 "missing-session-terminal",
789 "returned-interrupt",
790 "returned-interrupt-terminal",
791 "superseded-attempt",
792] as const)("Stop recovers a stalled run after %s without changing a newer attempt", (stalledRun) =>
793 stopEarlierBackgroundWork({ stalledRun }),
794);
apps/server/src/orchestration-v2/EffectWorker.test.ts+38-1
......@@ -81,13 +81,14 @@ function layerExecutorFor(input: {
8181 readonly failFirstStart?: Ref.Ref<boolean>;
8282 readonly threads?: Partial<ThreadManagementService.ThreadManagementService["Service"]>;
8383 readonly continueAfterRestart?: boolean;
84 readonly interrupt?: ProviderTurnControlService.ProviderTurnControlServiceV2Shape["interrupt"];
8485}) {
8586 const record = (event: string) => Ref.update(input.events, (events) => [...events, event]);
8687 const layerDependencies = Layer.mergeAll(
8788 Layer.succeed(
8889 ProviderTurnControlService.ProviderTurnControlServiceV2,
8990 ProviderTurnControlService.ProviderTurnControlServiceV2.of({
90 interrupt: () => Effect.void,
91 interrupt: input.interrupt ?? (() => Effect.void),
9192 steer: () => Effect.void,
9293 interruptAndAwaitTerminal: (request) =>
9394 record(
......@@ -193,6 +194,42 @@ it("does not retry pure interrupt races where the turn is already gone", () => {
193194 );
194195});
195196
197it.effect("settles a stopped run when its adapter has already lost the native turn", () =>
198 Effect.gen(function* () {
199 const now = yield* DateTime.now;
200 const events = yield* Ref.make<ReadonlyArray<string>>([]);
201 const layer = layerExecutorFor({
202 events,
203 interrupt: () =>
204 new ProviderTurnControlService.ProviderTurnControlError({
205 threadId,
206 operation: "interrupt",
207 providerTurnId,
208 cause: "Provider turn is not active.",
209 }),
210 threads: {
211 dispatch: (command) =>
212 Ref.update(events, (current) => [...current, command.type]).pipe(
213 Effect.as({ sequence: 1, storedEvents: [] }),
214 ),
215 },
216 });
217 yield* Effect.gen(function* () {
218 const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
219 yield* executor.execute({
220 ...restartEffect(now, { type: "detach" }),
221 request: {
222 type: "provider-turn.interrupt",
223 providerSessionId: oldSessionId,
224 providerThreadId,
225 providerTurnId,
226 },
227 });
228 }).pipe(Effect.provide(layer));
229 assert.deepEqual(yield* Ref.get(events), ["thread.background-work.settle"]);
230 }),
231);
232
196233it.effect("requeues a claim when a pre-execution worker check fails", () =>
197234 Effect.gen(function* () {
198235 const now = DateTime.formatIso(yield* DateTime.now);
apps/server/src/orchestration-v2/EffectWorker.ts+8
......@@ -171,6 +171,14 @@ export const layerExecutor: Layer.Layer<
171171 providerTurnId: effect.request.providerTurnId,
172172 })
173173 .pipe(
174 Effect.catch((cause) =>
175 isNonRetryableProviderTurnControlFailure(
176 effect.request.type,
177 Cause.pretty(Cause.fail(cause)),
178 )
179 ? Effect.void
180 : Effect.fail(cause),
181 ),
174182 // The provider has stopped what it still ran and reported it.
175183 // Whatever the thread still shows on that provider thread is
176184 // work no process will report on, so the Stop ends it too.
apps/server/src/orchestration-v2/EventSink.ts+10-1
......@@ -90,6 +90,7 @@ export interface EventSinkV2Shape {
9090 readonly activeAttemptId: RunAttemptId;
9191 readonly expectedStatus: OrchestrationV2Run["status"];
9292 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
93 readonly effects?: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
9394 }) => Effect.Effect<
9495 {
9596 readonly committed: boolean;
......@@ -437,9 +438,17 @@ const layerBase: Layer.Layer<
437438 events: normalized,
438439 });
439440 yield* applyStoredEvents(storedEvents);
441 yield* effectOutbox.enqueue(input.effects ?? []);
440442 return { committed: true as const, storedEvents };
441443 }),
442 (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void),
444 (result) =>
445 Effect.gen(function* () {
446 if (!result.committed) return;
447 if (input.effects !== undefined && input.effects.length > 0) {
448 yield* effectOutbox.notifyAvailable(input.effects.length);
449 }
450 yield* publishStoredEvents(result.storedEvents);
451 }),
443452 );
444453 },
445454 );
apps/server/src/orchestration-v2/FoundationPersistence.test.ts+28-1
......@@ -1525,6 +1525,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
15251525 Effect.gen(function* () {
15261526 const eventSink = yield* EventSink.EventSinkV2;
15271527 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1528 const outbox = yield* EffectOutbox.EffectOutboxV2;
15281529 const now = yield* DateTime.now;
15291530 const threadId = ThreadId.make("thread:foundation-stale-provider-start");
15301531 const runId = RunId.make("run:foundation-stale-provider-start");
......@@ -1567,6 +1568,30 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
15671568 ],
15681569 });
15691570
1571 const captureEffect = {
1572 id: "effect:foundation-current-capture",
1573 commandId: CommandId.make("command:foundation-current-capture"),
1574 threadId,
1575 request: {
1576 type: "checkpoint.capture" as const,
1577 runId,
1578 scopeId: CheckpointScopeId.make("scope:foundation-current-capture"),
1579 },
1580 };
1581 assert.isTrue(
1582 (yield* eventSink.writeIfRunCurrent({
1583 threadId,
1584 runId,
1585 activeAttemptId: attemptId,
1586 expectedStatus: "starting",
1587 events: [],
1588 effects: [captureEffect],
1589 })).committed,
1590 );
1591 assert.isTrue(Option.isSome(yield* outbox.get(captureEffect.id)));
1592 yield* outbox.awaitAvailable;
1593 const staleCaptureEffect = { ...captureEffect, id: "effect:foundation-stale-capture" };
1594
15701595 const reachedPrecommitGap = yield* Deferred.make<void>();
15711596 const releaseStaleStart = yield* Deferred.make<void>();
15721597 const providerStartCount = yield* Ref.make(0);
......@@ -1578,6 +1603,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
15781603 runId,
15791604 activeAttemptId: attemptId,
15801605 expectedStatus: "starting",
1606 effects: [staleCaptureEffect],
15811607 events: [
15821608 {
15831609 id: EventId.make("event:foundation-stale-provider-start:running"),
......@@ -1619,11 +1645,12 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
16191645
16201646 const staleResult = yield* Fiber.join(staleStartFiber);
16211647 assert.isFalse(staleResult.committed);
1648 assert.isTrue(Option.isNone(yield* outbox.get(staleCaptureEffect.id)));
16221649 assert.deepEqual(staleResult.storedEvents, []);
16231650 assert.equal(yield* Ref.get(providerStartCount), 0);
16241651 const projection = yield* projectionStore.getThreadProjection(threadId);
16251652 assert.equal(projection.runs[0]?.status, "cancelled");
1626 }),
1653 }).pipe(Effect.provide(Layer.fresh(layerTest))),
16271654 );
16281655
16291656 it.effect("guards post-terminal provider-thread writes by attempt and run ordinal", () =>
apps/server/src/orchestration-v2/Orchestrator.ts+267-38
......@@ -102,6 +102,7 @@ import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
102102import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts";
103103import { ProviderContinuationRequests } from "./ProviderContinuationRequests.ts";
104104import { makeProviderFailure } from "./ProviderFailure.ts";
105import * as RunExecutionService from "./RunExecutionService.ts";
105106import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
106107import { ProviderSwitchServiceV2 } from "./ProviderSwitchService.ts";
107108import { isAutomaticCompletionRun, queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts";
......@@ -8116,9 +8117,204 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
81168117 driver: providerThread.driver,
81178118 providerInstanceId: providerThread.providerInstanceId,
81188119 occurredAt: input.now,
8119 payload: { ...providerThread, pendingBackgroundTasks: [], updatedAt: input.now },
8120 payload: {
8121 ...providerThread,
8122 status:
8123 providerThread.status === "active" &&
8124 providerThread.lastRunOrdinal !== null &&
8125 providerThread.lastRunOrdinal <= input.throughRunOrdinal
8126 ? "idle"
8127 : providerThread.status,
8128 pendingBackgroundTasks: [],
8129 updatedAt: input.now,
8130 },
8131 });
8132 }
8133 });
8134
8135 // A successful Stop may have no terminal event to ingest: the session or
8136 // native turn already died, or its final write failed when storage filled.
8137 const settleInterruptedRun = (input: {
8138 readonly command: Extract<
8139 OrchestrationV2ServerCommand,
8140 { readonly type: "run.interrupt" | "thread.background-work.settle" }
8141 >;
8142 readonly projection: Pick<
8143 OrchestrationV2ThreadProjection,
8144 | "runs"
8145 | "attempts"
8146 | "nodes"
8147 | "subagents"
8148 | "runtimeRequests"
8149 | "messages"
8150 | "turnItems"
8151 | "providerThreads"
8152 >;
8153 readonly providerTurn: OrchestrationV2ProviderTurn;
8154 readonly events: Ref.Ref<Array<OrchestrationV2DomainEvent>>;
8155 readonly effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>;
8156 readonly now: DateTime.Utc;
8157 }) =>
8158 Effect.gen(function* () {
8159 const attempt = input.projection.attempts.find(
8160 (candidate) => candidate.id === input.providerTurn.runAttemptId,
8161 );
8162 const run = input.projection.runs.find((candidate) => candidate.id === attempt?.runId);
8163 const rootNode = input.projection.nodes.find((candidate) => candidate.id === run?.rootNodeId);
8164 if (
8165 attempt === undefined ||
8166 run === undefined ||
8167 rootNode === undefined ||
8168 run.activeAttemptId !== attempt.id ||
8169 run.status !== "running" ||
8170 attempt.status !== "running"
8171 )
8172 return;
8173 const emitEvent = emit(input.events, input.command);
8174 const base = {
8175 threadId: input.command.threadId,
8176 runId: run.id,
8177 nodeId: rootNode.id,
8178 providerInstanceId: run.providerInstanceId,
8179 occurredAt: input.now,
8180 };
8181 if (input.providerTurn.status === "running") {
8182 yield* emitEvent({
8183 ...base,
8184 type: "provider-turn.updated",
8185 payload: {
8186 ...input.providerTurn,
8187 status: "interrupted",
8188 completedAt: input.now,
8189 },
8190 });
8191 }
8192 yield* emitEvent({
8193 ...base,
8194 type: "run-attempt.updated",
8195 payload: {
8196 ...attempt,
8197 status: "interrupted",
8198 completedAt: input.now,
8199 },
8200 });
8201 const providerThread = input.projection.providerThreads.find(
8202 (candidate) => candidate.id === input.providerTurn.providerThreadId,
8203 );
8204 if (providerThread?.status === "active" && providerThread.lastRunOrdinal === run.ordinal) {
8205 yield* emitEvent({
8206 ...base,
8207 type: "provider-thread.updated",
8208 payload: { ...providerThread, status: "idle", updatedAt: input.now },
8209 });
8210 }
8211 for (const message of input.projection.messages) {
8212 if (message.runId !== run.id || !message.streaming) continue;
8213 yield* emitEvent({
8214 ...base,
8215 ...(message.nodeId === null ? {} : { nodeId: message.nodeId }),
8216 type: "message.updated",
8217 payload: { ...message, streaming: false, updatedAt: input.now },
8218 });
8219 }
8220 const delegatedNodeIds = new Set(
8221 input.projection.subagents
8222 .filter((subagent) => subagent.origin === "app_owned")
8223 .map((subagent) => subagent.id),
8224 );
8225 const cascaded = yield* RunExecutionService.cascadeTerminalizeRunOwnedSubagents({
8226 run,
8227 status: "interrupted",
8228 completedAt: input.now,
8229 allocateEventId: () =>
8230 idAllocator.allocate.event({
8231 threadId: input.command.threadId,
8232 commandId: input.command.commandId,
8233 }),
8234 open: {
8235 subagents: new Map(
8236 input.projection.subagents
8237 .filter((task) => task.runId === run.id && task.origin !== "app_owned")
8238 .map((task) => [task.id, task]),
8239 ),
8240 nodes: new Map(
8241 input.projection.nodes
8242 .filter((node) => node.runId === run.id && !delegatedNodeIds.has(node.id))
8243 .map((node) => [node.id, node]),
8244 ),
8245 turnItems: new Map(),
8246 childTurnItems: new Map(
8247 input.projection.turnItems
8248 .filter(
8249 (item) =>
8250 item.runId === run.id &&
8251 (item.type === "assistant_message" || item.type === "reasoning"),
8252 )
8253 .map((item) => [item.id, item]),
8254 ),
8255 linkedChildThreadIds: new Set([run.threadId]),
8256 },
8257 }).pipe(mapDispatchError(input.command));
8258 yield* Ref.update(input.events, (events) => [...events, ...cascaded]);
8259 for (const request of input.projection.runtimeRequests) {
8260 if (
8261 request.status !== "pending" ||
8262 !input.projection.nodes.some(
8263 (node) => node.id === request.nodeId && node.runId === run.id,
8264 )
8265 )
8266 continue;
8267 yield* emitEvent({
8268 ...base,
8269 nodeId: request.nodeId,
8270 type: "runtime-request.updated",
8271 payload: {
8272 ...request,
8273 status: "cancelled",
8274 resolvedAt: input.now,
8275 responseCapability: { type: "not_resumable", reason: "The run was interrupted." },
8276 },
8277 });
8278 }
8279 if (providerThread !== undefined) {
8280 yield* emitEvent({
8281 ...base,
8282 type: "turn-item.updated",
8283 payload: RunExecutionService.makeInterruptResultTurnItem({
8284 idAllocator,
8285 run,
8286 rootNode,
8287 providerThread,
8288 completedAt: input.now,
8289 }),
81208290 });
81218291 }
8292 const { delegatedCompletion: _delegatedCompletion, ...runWithoutDelegatedCompletion } = run;
8293 yield* emitEvent({
8294 ...base,
8295 type: "run.updated",
8296 payload: {
8297 ...runWithoutDelegatedCompletion,
8298 status: "interrupted",
8299 completedAt: input.now,
8300 },
8301 });
8302 const scopeId = rootNode.checkpointScopeId;
8303 if (scopeId !== null) {
8304 yield* Ref.update(input.effects, (effects) => [
8305 ...effects,
8306 {
8307 id: `effect:checkpoint.capture:${run.id}`,
8308 commandId: CommandId.make(`command:effect:checkpoint.capture:${run.id}`),
8309 threadId: input.command.threadId,
8310 request: {
8311 type: "checkpoint.capture",
8312 runId: run.id,
8313 scopeId,
8314 },
8315 } satisfies PendingOrchestrationEffectV2,
8316 ]);
8317 }
81228318 });
81238319
81248320 /**
......@@ -8280,14 +8476,29 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
82808476 { readonly type: "thread.background-work.settle" }
82818477 >,
82828478 events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
8479 effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
82838480 ) =>
82848481 Effect.gen(function* () {
82858482 const [projection, stopped] = yield* Effect.all([
82868483 projectionStore.getThreadRecords(
82878484 command.threadId,
8288 ["runs", "attempts", "turnItems", "providerThreads"],
8485 [
8486 "runs",
8487 "attempts",
8488 "nodes",
8489 "subagents",
8490 "runtimeRequests",
8491 "turnItems",
8492 "providerThreads",
8493 ],
82898494 {
8290 turnItemTypes: ["command_execution", "dynamic_tool", "subagent"],
8495 turnItemTypes: [
8496 "command_execution",
8497 "dynamic_tool",
8498 "subagent",
8499 "assistant_message",
8500 "reasoning",
8501 ],
82918502 turnItemStatuses: ["pending", "running", "waiting"],
82928503 },
82938504 ),
......@@ -8304,12 +8515,35 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83048515 (attempt) => attempt.id === stopped.providerTurn?.runAttemptId,
83058516 )?.runId;
83068517 const stoppedRun = projection.runs.find((run) => run.id === stoppedRunId);
8307 // A new turn may have started since Stop; its work is not this Stop's.
83088518 if (
83098519 stoppedRun === undefined ||
8520 (["preparing", "starting", "running"].includes(stoppedRun.status) &&
8521 stoppedRun.activeAttemptId !== stopped.providerTurn?.runAttemptId)
8522 )
8523 return;
8524 const now = yield* DateTime.now;
8525 if (stopped.providerTurn !== undefined && stoppedRun.status === "running") {
8526 const output = yield* projectionStore
8527 .getThreadRecords(command.threadId, ["messages"], {
8528 messageRunIds: [stoppedRun.id],
8529 messageRoles: ["assistant"],
8530 })
8531 .pipe(mapDispatchError(command));
8532 yield* settleInterruptedRun({
8533 command,
8534 projection: { ...projection, messages: output.messages },
8535 providerTurn: stopped.providerTurn,
8536 events,
8537 effects,
8538 now,
8539 });
8540 }
8541 // A new turn may have started since Stop; its work is not this Stop's.
8542 if (
83108543 projection.runs.some(
83118544 (run) =>
8312 run.status === "preparing" || run.status === "starting" || run.status === "running",
8545 run.id !== stoppedRun.id &&
8546 (run.status === "preparing" || run.status === "starting" || run.status === "running"),
83138547 )
83148548 ) {
83158549 return;
......@@ -8320,7 +8554,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83208554 projection,
83218555 stoppedProviderThreadId: command.providerThreadId,
83228556 throughRunOrdinal: stoppedRun.ordinal,
8323 now: yield* DateTime.now,
8557 now,
83248558 });
83258559 });
83268560
......@@ -8341,11 +8575,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83418575 "attempts",
83428576 "turnItems",
83438577 "subagents",
8578 "runtimeRequests",
83448579 ],
83458580 // Thread-wide, like the Waiting strip: a settled thread's background
83468581 // work can belong to an earlier run than the one Stop targets.
83478582 {
8348 turnItemTypes: ["command_execution", "dynamic_tool", "subagent"],
8583 turnItemTypes: [
8584 "command_execution",
8585 "dynamic_tool",
8586 "subagent",
8587 "assistant_message",
8588 "reasoning",
8589 ],
83498590 turnItemStatuses: ["pending", "running", "waiting"],
83508591 },
83518592 );
......@@ -8373,7 +8614,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83738614 projection.providerTurns.findLast(
83748615 (candidate) =>
83758616 candidate.runAttemptId === run?.activeAttemptId &&
8376 (candidate.status === "running" || hasBackgroundWork),
8617 (run?.status === "running" || candidate.status === "running" || hasBackgroundWork),
83778618 ) ??
83788619 (hasBackgroundWork
83798620 ? projection.providerTurns.findLast(
......@@ -8623,18 +8864,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
86238864 cause: `Run ${command.runId} is not interruptible.`,
86248865 });
86258866 }
8626 // Stop on a settled thread's background work. Its process may be gone
8627 // (released, restarted) and only the projection still shows the work;
8628 // the settle follow-up ends whatever no provider reports ending.
8629 const settleOnly =
8630 providerTurn.status !== "running" &&
8631 (providerThread.providerSessionId === null ||
8632 Option.isNone(
8633 yield* providerSessions
8634 .get(providerThread.providerSessionId)
8635 .pipe(Effect.orElseSucceed(() => Option.none())),
8636 ));
8637 if (settleOnly) {
8867 // A dead session cannot report a terminal event for either a running
8868 // turn or a settled turn's background work. Stop ends both locally.
8869 const sessionOption =
8870 providerThread.providerSessionId === null
8871 ? Option.none()
8872 : yield* providerSessions.get(providerThread.providerSessionId).pipe(
8873 Effect.mapError(
8874 (cause) =>
8875 new OrchestratorProviderAdapterError({
8876 commandId: command.commandId,
8877 providerInstanceId: run.providerInstanceId,
8878 cause,
8879 }),
8880 ),
8881 );
8882 if (Option.isNone(sessionOption)) {
86388883 yield* emitEvent({
86398884 type: "turn-item.updated",
86408885 threadId: command.threadId,
......@@ -8645,6 +8890,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
86458890 payload: interruptRequestItem,
86468891 });
86478892 yield* stopRemainingWork();
8893 yield* settleInterruptedRun({ command, projection, providerTurn, events, effects, now });
86488894 yield* settleBackgroundWork({
86498895 command,
86508896 events,
......@@ -8664,23 +8910,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
86648910 });
86658911 }
86668912 const providerSessionId = providerThread.providerSessionId;
8667 const sessionOption = yield* providerSessions.get(providerSessionId).pipe(
8668 Effect.mapError(
8669 (cause) =>
8670 new OrchestratorProviderAdapterError({
8671 commandId: command.commandId,
8672 providerInstanceId: run.providerInstanceId,
8673 cause,
8674 }),
8675 ),
8676 );
8677 if (Option.isNone(sessionOption)) {
8678 return yield* new OrchestratorDispatchError({
8679 commandId: command.commandId,
8680 commandType: command.type,
8681 cause: `Provider session ${providerThread.providerSessionId} is not active.`,
8682 });
8683 }
86848913 yield* enforceCommandPolicy(command)(
86858914 commandPolicy.ensureInterrupt({
86868915 commandId: command.commandId,
......@@ -10079,7 +10308,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1007910308 yield* dispatchCheckpointRollbackFail(command, events);
1008010309 break;
1008110310 case "thread.background-work.settle":
10082 yield* dispatchBackgroundWorkSettle(command, events);
10311 yield* dispatchBackgroundWorkSettle(command, events, effects);
1008310312 break;
1008410313 case "thread.fork":
1008510314 yield* dispatchThreadFork(command, events);
apps/server/src/orchestration-v2/ProviderTurnControlService.ts+14
......@@ -12,6 +12,7 @@ import * as Effect from "effect/Effect";
1212import * as Layer from "effect/Layer";
1313import * as Option from "effect/Option";
1414import * as Schema from "effect/Schema";
15import * as NodeTimersPromises from "node:timers/promises";
1516
1617import * as ProjectionStore from "./ProjectionStore.ts";
1718import * as ProviderSessionManager from "./ProviderSessionManager.ts";
......@@ -185,6 +186,19 @@ export const layer: Layer.Layer<
185186 providerTurnId: loaded.providerTurn.id,
186187 requestRuntimeRestart: true,
187188 });
189 // Give native terminal ingestion time to finish before the Stop
190 // follow-up repairs a run whose provider no longer reports on it.
191 const deadline = performance.now() + 2_000;
192 while (loaded.providerTurn.status === "running" && performance.now() < deadline) {
193 const current = yield* projections.getProviderControlContext(input.threadId, input);
194 if (
195 current.providerTurn?.status !== "running" &&
196 current.attempt?.status !== "running"
197 ) {
198 break;
199 }
200 yield* Effect.promise(() => NodeTimersPromises.setTimeout(10));
201 }
188202 }).pipe(
189203 Effect.mapError((cause) =>
190204 isProviderTurnControlError(cause)
apps/server/src/orchestration-v2/RunExecutionService.test.ts+49-6
......@@ -3243,9 +3243,27 @@ it.effect("omits run_interrupt_result when superseded attempt request is already
32433243 }),
32443244);
32453245
3246it.effect("does not overwrite Stop when ownership changes after the finalization read", () =>
3247 Effect.gen(function* () {
3248 const { written, observed, submittedEffects, committedEffects } =
3249 yield* captureRootRunTermination({
3250 key: "stop-wins-finalization-gap",
3251 shouldFinalizeRun: () => Effect.succeed(true),
3252 rejectTerminalWrite: true,
3253 });
3254 assert.deepEqual(written, []);
3255 assert.deepEqual(observed, []);
3256 assert.deepEqual(
3257 submittedEffects.map((effect) => effect.request.type),
3258 ["checkpoint.capture"],
3259 );
3260 assert.deepEqual(committedEffects, []);
3261 }),
3262);
3263
32463264it.effect("emits run_interrupt_result when hard-stop finalizes the active attempt", () =>
32473265 Effect.gen(function* () {
3248 const { written, observed } = yield* captureRootRunTermination({
3266 const { written, observed, committedEffects } = yield* captureRootRunTermination({
32493267 key: "hard-stop",
32503268 shouldFinalizeRun: () => Effect.succeed(true),
32513269 });
......@@ -3254,6 +3272,10 @@ it.effect("emits run_interrupt_result when hard-stop finalizes the active attemp
32543272 ["run_interrupt_result"],
32553273 );
32563274 assert.deepEqual(observed, ["run:interrupted", "pull-requests-refreshed"]);
3275 assert.deepEqual(
3276 committedEffects.map((effect) => effect.request.type),
3277 ["checkpoint.capture"],
3278 );
32573279 }),
32583280);
32593281
......@@ -3382,6 +3404,7 @@ it.effect("keeps completed runs completed when pull request refresh fails", () =
33823404function captureRootRunTermination(input: {
33833405 readonly key: string;
33843406 readonly shouldFinalizeRun: () => Effect.Effect<boolean, ProjectionStore.ProjectionStoreV2Error>;
3407 readonly rejectTerminalWrite?: boolean;
33853408 readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect<boolean, never>;
33863409 readonly seedOpenSubagent?: boolean;
33873410 readonly events?: (
......@@ -3402,6 +3425,8 @@ function captureRootRunTermination(input: {
34023425 });
34033426 const writtenItems = yield* Ref.make<ReadonlyArray<OrchestrationV2TurnItem>>([]);
34043427 const observed = yield* Ref.make<ReadonlyArray<string>>([]);
3428 const submittedEffects = yield* Ref.make<ReadonlyArray<PendingOrchestrationEffectV2>>([]);
3429 const committedEffects = yield* Ref.make<ReadonlyArray<PendingOrchestrationEffectV2>>([]);
34053430 const ingestionDone = yield* Deferred.make<void>();
34063431 const captureTurnItem = (payload: OrchestrationV2TurnItem) =>
34073432 Ref.update(writtenItems, (current) => [...current, payload]);
......@@ -3430,11 +3455,24 @@ function captureRootRunTermination(input: {
34303455 }
34313456 return [];
34323457 }),
3433 writeWithEffects: (payload) => captureFinalEvents(payload.events).pipe(Effect.as([])),
3458 writeWithEffects: (payload) =>
3459 Effect.gen(function* () {
3460 yield* Ref.update(submittedEffects, (current) => [...current, ...payload.effects]);
3461 yield* Ref.update(committedEffects, (current) => [...current, ...payload.effects]);
3462 yield* captureFinalEvents(payload.events);
3463 return [];
3464 }),
34343465 writeIfRunCurrent: (payload) =>
3435 captureFinalEvents(payload.events).pipe(
3436 Effect.as({ committed: true, storedEvents: [] }),
3437 ),
3466 Effect.gen(function* () {
3467 const effects = payload.effects ?? [];
3468 yield* Ref.update(submittedEffects, (current) => [...current, ...effects]);
3469 if (input.rejectTerminalWrite === true) {
3470 return { committed: false, storedEvents: [] };
3471 }
3472 yield* Ref.update(committedEffects, (current) => [...current, ...effects]);
3473 yield* captureFinalEvents(payload.events);
3474 return { committed: true, storedEvents: [] };
3475 }),
34383476 }),
34393477 IdAllocator.layer,
34403478 Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
......@@ -3543,7 +3581,12 @@ function captureRootRunTermination(input: {
35433581 }).pipe(Effect.provide(layerTest));
35443582
35453583 yield* Deferred.await(ingestionDone);
3546 return { written: yield* Ref.get(writtenItems), observed: yield* Ref.get(observed) };
3584 return {
3585 written: yield* Ref.get(writtenItems),
3586 observed: yield* Ref.get(observed),
3587 submittedEffects: yield* Ref.get(submittedEffects),
3588 committedEffects: yield* Ref.get(committedEffects),
3589 };
35473590 });
35483591}
35493592
apps/server/src/orchestration-v2/RunExecutionService.ts+11-2
......@@ -783,6 +783,7 @@ export const layer: Layer.Layer<
783783 activeAttemptId: input.writeIfRunCurrent.activeAttemptId,
784784 expectedStatus: input.writeIfRunCurrent.expectedStatus,
785785 events: finalization.events,
786 effects: finalization.effects,
786787 });
787788 if (!result.committed) {
788789 return;
......@@ -971,7 +972,15 @@ export const layer: Layer.Layer<
971972 attempt: input.attempt,
972973 ...(input.shouldFinalizeRun === undefined
973974 ? {}
974 : { shouldFinalizeRun: input.shouldFinalizeRun }),
975 : {
976 shouldFinalizeRun: input.shouldFinalizeRun,
977 // Stop may commit between the ownership read and this
978 // terminal write. Gate the events and checkpoint together.
979 writeIfRunCurrent: {
980 activeAttemptId: input.attempt.id,
981 expectedStatus: "running" as const,
982 },
983 }),
975984 ...(input.hasUnpairedRunInterruptRequest === undefined
976985 ? {}
977986 : {
......@@ -1437,7 +1446,7 @@ export const layer: Layer.Layer<
14371446 }),
14381447);
14391448
1440function makeInterruptResultTurnItem(input: {
1449export function makeInterruptResultTurnItem(input: {
14411450 readonly idAllocator: IdAllocator.IdAllocatorV2Shape;
14421451 readonly run: OrchestrationV2Run;
14431452 readonly rootNode: OrchestrationV2ExecutionNode;
apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts+5-1
......@@ -261,7 +261,10 @@ export function layerWithRegistry<Error>(
261261 readonly continueThreadsAfterServerUpdate?: boolean;
262262 } = {},
263263): Layer.Layer<
264 Orchestrator.OrchestratorV2 | EffectWorker.OrchestrationEffectWorkerV2 | EventSink.EventSinkV2,
264 | Orchestrator.OrchestratorV2
265 | EffectWorker.OrchestrationEffectWorkerV2
266 | EventSink.EventSinkV2
267 | ProviderSessionManager.ProviderSessionManagerV2,
265268 Error | MigrationError | PlatformError.PlatformError | SqlError
266269> {
267270 const layerServerConfig = Layer.effect(
......@@ -474,6 +477,7 @@ export function layerWithRegistry<Error>(
474477 );
475478 const layerReplayRuntime = Layer.mergeAll(
476479 layerOrchestratorProvided,
480 layerProviderSessionManagerProvided,
477481 layerEffectWorkerProvided,
478482 layerEventSinkProvided,
479483 layerContinuationWorkerProvided,