1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 EventId,
5 MessageId,
6 NodeId,
7 ProjectId,
8 ProviderDriverKind,
9 ProviderInstanceId,
10 ProviderThreadId,
11 ProviderTurnId,
12 RunAttemptId,
13 RunId,
14 ThreadId,
15 TurnItemId,
16 type OrchestrationV2DomainEvent,
17} from "@t3tools/contracts";
18import * as DateTime from "effect/DateTime";
19import * as Effect from "effect/Effect";
20import * as Fiber from "effect/Fiber";
21import * as Queue from "effect/Queue";
22import * as Stream from "effect/Stream";
23import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
24import * as EffectWorker from "./EffectWorker.ts";
25import * as EventSink from "./EventSink.ts";
26import * as Orchestrator from "./Orchestrator.ts";
27import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
28import * as ProviderSessionManager from "./ProviderSessionManager.ts";
29import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
30import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
31import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
32
33const driver = ProviderDriverKind.make("codex");
34const instanceId = ProviderInstanceId.make("codex");
35const modelSelection = { instanceId, model: "test-model" };
36
37// Codex turns leave commands running, then the thread moves to another
38// provider thread (a provider switch). Stop must end all of the Codex work,
39// including when it selects an older resumed run on the other provider thread.
40const stopEarlierBackgroundWork = ({
41 failedStart = false,
42 stopWithQueue,
43 olderStart = false,
44 stalledRun,
45 workflowMember = false,
46}: {
47 readonly failedStart?: boolean;
48 readonly stopWithQueue?: "thread.stop" | "run.interrupt";
49 readonly olderStart?: boolean;
50 readonly workflowMember?: boolean;
51 readonly stalledRun?:
52 | "missing-session"
53 | "missing-session-terminal"
54 | "returned-interrupt"
55 | "returned-interrupt-terminal"
56 | "superseded-attempt";
57}) =>
58 Effect.scoped(
59 Effect.gen(function* () {
60 const cwd = yield* checkpointWorkspace("background-work-stop");
61 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
62 const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = [];
63 const interrupts: ProviderAdapter.ProviderAdapterV2InterruptInput[] = [];
64 const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
65 instanceId,
66 driver,
67 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
68 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
69 openSession: (input) =>
70 Effect.gen(function* () {
71 const now = yield* DateTime.now;
72 return {
73 instanceId,
74 driver,
75 providerSessionId: input.providerSessionId,
76 providerSession: {
77 id: input.providerSessionId,
78 driver,
79 providerInstanceId: instanceId,
80 status: "ready",
81 cwd,
82 model: modelSelection.model,
83 capabilities: CodexProviderCapabilitiesV2,
84 createdAt: now,
85 updatedAt: now,
86 lastError: null,
87 },
88 events: Stream.fromQueue(events),
89 ensureThread: ({ threadId }) =>
90 Effect.succeed({
91 id: ProviderThreadId.make(`provider-thread:codex:${threadId}`),
92 driver,
93 providerInstanceId: instanceId,
94 providerSessionId: input.providerSessionId,
95 appThreadId: threadId,
96 ownerNodeId: null,
97 nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
98 nativeConversationHeadRef: null,
99 status: "idle",
100 firstRunOrdinal: null,
101 lastRunOrdinal: null,
102 handoffIds: [],
103 forkedFrom: null,
104 createdAt: now,
105 updatedAt: now,
106 }),
107 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
108 startTurn: (turn) =>
109 Effect.gen(function* () {
110 started.push(turn);
111 yield* Queue.offer(events, {
112 type: "provider_turn.updated",
113 driver,
114 providerTurn: {
115 id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`),
116 providerThreadId: turn.providerThread.id,
117 nodeId: turn.rootNodeId,
118 runAttemptId: turn.attemptId,
119 nativeTurnRef: {
120 driver,
121 nativeId: `native:${turn.attemptId}`,
122 strength: "strong",
123 },
124 ordinal: turn.providerTurnOrdinal,
125 status: "running",
126 startedAt: now,
127 completedAt: null,
128 },
129 });
130 }),
131 steerTurn: () => Effect.die("unused"),
132 interruptTurn: (interrupt) =>
133 Effect.sync(() => {
134 interrupts.push(interrupt);
135 }),
136 respondToRuntimeRequest: () => Effect.die("unused"),
137 readThreadSnapshot: () => Effect.die("unused"),
138 rollbackThread: () => Effect.die("unused"),
139 forkThread: () => Effect.die("unused"),
140 };
141 }),
142 };
143 yield* Effect.gen(function* () {
144 const orchestrator = yield* Orchestrator.OrchestratorV2;
145 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
146 const sink = yield* EventSink.EventSinkV2;
147 const threadId = ThreadId.make("thread:background-work-stop");
148 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
149 orchestrator.streamDomainEvents.pipe(
150 Stream.filter(predicate),
151 Stream.take(1),
152 Stream.runDrain,
153 Effect.forkScoped,
154 );
155 yield* orchestrator.dispatch({
156 type: "thread.create",
157 commandId: CommandId.make("create"),
158 threadId,
159 projectId: ProjectId.make("project:background-work-stop"),
160 title: "Background work stop",
161 modelSelection,
162 runtimeMode: "full-access",
163 interactionMode: "default",
164 branch: null,
165 worktreePath: cwd,
166 createdBy: "user",
167 creationSource: "web",
168 });
169 const running = yield* watch(
170 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
171 );
172 yield* orchestrator.dispatch({
173 type: "message.dispatch",
174 commandId: CommandId.make("start-dev-server"),
175 threadId,
176 messageId: MessageId.make("message:start-dev-server"),
177 text: "Start the dev server",
178 attachments: [],
179 dispatchMode: { type: "start_immediately" },
180 createdBy: "user",
181 creationSource: "web",
182 });
183 yield* worker.drain();
184 yield* Fiber.join(running);
185 const first = started[0]!;
186 const codexTurn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!;
187 const devServerId = TurnItemId.make("turn-item:dev-server");
188 const now = yield* DateTime.now;
189 yield* sink.write({
190 events: [
191 {
192 id: EventId.make("dev-server"),
193 type: "turn-item.updated",
194 threadId,
195 runId: first.runId,
196 occurredAt: now,
197 payload: {
198 id: devServerId,
199 threadId,
200 runId: first.runId,
201 nodeId: first.rootNodeId,
202 providerThreadId: codexTurn.providerThreadId,
203 providerTurnId: codexTurn.id,
204 nativeItemRef: null,
205 parentItemId: null,
206 ordinal: 100,
207 status: "running",
208 title: null,
209 startedAt: now,
210 completedAt: null,
211 updatedAt: now,
212 type: "command_execution",
213 input: "vp run dev --share",
214 },
215 },
216 ],
217 });
218 if (stalledRun !== undefined) {
219 const before = yield* orchestrator.getThreadProjection(threadId);
220 const run = before.runs[0]!;
221 const node = before.nodes.find((candidate) => candidate.id === run.rootNodeId)!;
222 const attempt = before.attempts[0]!;
223 if (stalledRun === "missing-session" || stalledRun === "missing-session-terminal") {
224 const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
225 const failed = yield* watch(
226 (event) => event.type === "run.updated" && event.payload.status === "failed",
227 );
228 yield* sessions.release({
229 providerSessionId: first.providerThread.providerSessionId!,
230 reason: "runtime_error",
231 });
232 yield* Fiber.join(failed);
233 // Disk exhaustion can lose these terminal writes. Restore that stale state.
234 yield* sink.write({
235 events: [
236 {
237 id: EventId.make("stale-run"),
238 type: "run.updated",
239 threadId,
240 occurredAt: now,
241 payload: run,
242 },
243 {
244 id: EventId.make("stale-node"),
245 type: "node.updated",
246 threadId,
247 occurredAt: now,
248 payload: node,
249 },
250 {
251 id: EventId.make("stale-attempt"),
252 type: "run-attempt.updated",
253 threadId,
254 occurredAt: now,
255 payload: attempt,
256 },
257 ],
258 });
259 }
260 const messageId = MessageId.make("partial-output");
261 const item = before.turnItems.find((candidate) => candidate.id === devServerId)!;
262 assert.ok(item.type === "command_execution");
263 const terminalProviderTurn = stalledRun.endsWith("-terminal");
264 if (terminalProviderTurn) {
265 yield* sink.write({
266 events: [
267 {
268 id: EventId.make("terminal-provider-turn"),
269 type: "provider-turn.updated",
270 threadId,
271 occurredAt: now,
272 payload: { ...codexTurn, status: "completed", completedAt: now },
273 },
274 ...(stalledRun === "missing-session-terminal"
275 ? [
276 {
277 id: EventId.make("completed-dev-server"),
278 type: "turn-item.updated" as const,
279 threadId,
280 occurredAt: now,
281 payload: { ...item, status: "completed" as const, completedAt: now },
282 },
283 ]
284 : []),
285 ],
286 });
287 }
288 yield* sink.write({
289 events: [
290 {
291 id: EventId.make("partial-message"),
292 type: "message.updated",
293 threadId,
294 runId: run.id,
295 occurredAt: now,
296 payload: {
297 id: messageId,
298 threadId,
299 runId: run.id,
300 nodeId: node.id,
301 role: "assistant",
302 text: "Partial output",
303 attachments: [],
304 streaming: true,
305 createdBy: "agent",
306 creationSource: "provider",
307 createdAt: now,
308 updatedAt: now,
309 },
310 },
311 {
312 id: EventId.make("partial-item"),
313 type: "turn-item.updated",
314 threadId,
315 runId: run.id,
316 occurredAt: now,
317 payload: {
318 ...item,
319 providerThreadId: codexTurn.providerThreadId,
320 providerTurnId: codexTurn.id,
321 id: TurnItemId.make("partial-output"),
322 type: "assistant_message",
323 messageId,
324 text: "Partial output",
325 streaming: true,
326 },
327 },
328 ],
329 });
330 if (stalledRun === "superseded-attempt") {
331 yield* sink.write({
332 events: [
333 {
334 id: EventId.make("new-attempt"),
335 type: "run.updated",
336 threadId,
337 occurredAt: now,
338 payload: { ...run, activeAttemptId: RunAttemptId.make("new-attempt") },
339 },
340 ],
341 });
342 }
343 yield* orchestrator.dispatch(
344 stalledRun === "superseded-attempt"
345 ? {
346 type: "thread.background-work.settle",
347 commandId: CommandId.make("late-settle"),
348 threadId,
349 providerThreadId: codexTurn.providerThreadId,
350 providerTurnId: codexTurn.id,
351 }
352 : {
353 type: "thread.stop",
354 commandId: CommandId.make("stop-stalled-run"),
355 threadId,
356 },
357 );
358 yield* worker.drain();
359 const after = yield* orchestrator.getThreadProjection(threadId);
360 const interrupted = stalledRun !== "superseded-attempt";
361 assert.equal(after.runs[0]?.status, interrupted ? "interrupted" : "running");
362 assert.equal(after.attempts[0]?.status, interrupted ? "interrupted" : "running");
363 assert.equal(
364 after.providerTurns[0]?.status,
365 terminalProviderTurn ? "completed" : interrupted ? "interrupted" : "running",
366 );
367 assert.equal(
368 after.nodes.find((candidate) => candidate.id === node.id)?.status,
369 interrupted ? "interrupted" : "running",
370 );
371 const output = after.messages.find((message) => message.id === messageId)!;
372 assert.equal(output.streaming, !interrupted);
373 assert.equal(output.text, "Partial output");
374 assert.equal(
375 after.turnItems.find((candidate) => candidate.id === devServerId)?.status,
376 stalledRun === "missing-session-terminal"
377 ? "completed"
378 : interrupted
379 ? "interrupted"
380 : "running",
381 );
382 assert.equal(
383 after.turnItems.find((candidate) => candidate.type === "assistant_message")?.status,
384 interrupted ? "interrupted" : "running",
385 );
386 assert.equal(
387 after.turnItems.filter((candidate) => candidate.type === "run_interrupt_result").length,
388 interrupted ? 1 : 0,
389 );
390 assert.isEmpty(after.runs.filter((candidate) => candidate.status === "waiting"));
391 return;
392 }
393 const settled = yield* watch(
394 (event) =>
395 event.type === "run.updated" &&
396 event.payload.id === first.runId &&
397 event.payload.status === "waiting",
398 );
399 yield* Queue.offer(events, {
400 type: "provider_turn.updated",
401 driver,
402 providerTurn: { ...codexTurn, status: "completed", completedAt: now },
403 });
404 yield* Queue.offer(events, {
405 type: "turn.terminal",
406 driver,
407 providerThreadId: codexTurn.providerThreadId,
408 providerTurnId: codexTurn.id,
409 runOrdinal: first.runOrdinal,
410 status: "completed",
411 failure: null,
412 threadDisposition: "reusable",
413 });
414 yield* Fiber.join(settled);
415 yield* worker.drain();
416
417 const codexProviderThread = (yield* orchestrator.getThreadProjection(threadId))
418 .providerThreads[0]!;
419 // A native subagent's own thread, still showing its runless root turn as working.
420 const subagentThread = (name: string) =>
421 Effect.gen(function* () {
422 const subagentThreadId = ThreadId.make(`thread:${name}`);
423 const rootId = NodeId.make(`node:${name}-root`);
424 yield* orchestrator.dispatch({
425 type: "thread.create",
426 commandId: CommandId.make(`create-${name}`),
427 threadId: subagentThreadId,
428 projectId: ProjectId.make("project:background-work-stop"),
429 title: name,
430 modelSelection,
431 runtimeMode: "full-access",
432 interactionMode: "default",
433 branch: null,
434 worktreePath: cwd,
435 createdBy: "agent",
436 creationSource: "provider",
437 });
438 yield* sink.write({
439 events: [
440 {
441 id: EventId.make(`${name}-root`),
442 type: "node.updated",
443 threadId: subagentThreadId,
444 occurredAt: now,
445 payload: {
446 id: rootId,
447 threadId: subagentThreadId,
448 runId: null,
449 parentNodeId: null,
450 rootNodeId: rootId,
451 kind: "root_turn",
452 status: "running",
453 countsForRun: false,
454 providerThreadId: null,
455 providerTurnId: null,
456 nativeItemRef: null,
457 runtimeRequestId: null,
458 checkpointScopeId: null,
459 startedAt: now,
460 completedAt: null,
461 },
462 },
463 ],
464 });
465 return { threadId: subagentThreadId, rootId };
466 });
467 const { threadId: reviewerThreadId, rootId: reviewerRootId } =
468 yield* subagentThread("reviewer");
469 // The reviewer starts a tester, which runs on a thread of its own.
470 const tester = yield* subagentThread("tester");
471 const nestedSubagentId = NodeId.make("subagent:nested");
472 const nestedItemId = TurnItemId.make("turn-item:nested-subagent");
473 const nested = {
474 threadId: reviewerThreadId,
475 runId: null,
476 origin: "provider_native",
477 driver,
478 providerInstanceId: instanceId,
479 providerThreadId: null,
480 childThreadId: tester.threadId,
481 prompt: "Check the tests",
482 title: null,
483 status: "running",
484 result: null,
485 startedAt: now,
486 completedAt: null,
487 updatedAt: now,
488 } as const;
489 yield* sink.write({
490 events: [
491 // A subagent the reviewer started, which the provider records on the
492 // reviewer's own thread.
493 ...(workflowMember
494 ? []
495 : [
496 {
497 id: EventId.make("nested-subagent-item"),
498 type: "turn-item.updated",
499 threadId: reviewerThreadId,
500 occurredAt: now,
501 payload: {
502 ...nested,
503 id: nestedItemId,
504 nodeId: nestedSubagentId,
505 providerTurnId: null,
506 nativeItemRef: null,
507 parentItemId: null,
508 ordinal: 1,
509 type: "subagent",
510 subagentId: nestedSubagentId,
511 },
512 } as const,
513 ]),
514 {
515 id: EventId.make("nested-subagent"),
516 type: "subagent.updated",
517 threadId: reviewerThreadId,
518 occurredAt: now,
519 payload: {
520 ...nested,
521 id: nestedSubagentId,
522 parentNodeId: reviewerRootId,
523 createdBy: "agent",
524 nativeTaskRef: workflowMember
525 ? { driver, nativeId: "workflow:member", strength: "strong" }
526 : null,
527 model: null,
528 },
529 },
530 ],
531 });
532 if (workflowMember) {
533 // Workflow members have execution nodes and child roots, but no subagent turn item.
534 const memberRoot = (yield* orchestrator.getThreadProjection(tester.threadId)).nodes[0]!;
535 yield* sink.write({
536 events: [
537 {
538 id: EventId.make("workflow-member-node"),
539 type: "node.updated",
540 threadId: reviewerThreadId,
541 occurredAt: now,
542 payload: {
543 ...memberRoot,
544 id: nestedSubagentId,
545 threadId: reviewerThreadId,
546 parentNodeId: reviewerRootId,
547 rootNodeId: reviewerRootId,
548 kind: "subagent",
549 },
550 },
551 ],
552 });
553 }
554 // A settled run with its root node, attempt, provider turn and,
555 // optionally, a command or native subagent it left running.
556 const settledRun = (input: {
557 readonly ordinal: number;
558 readonly providerThreadId: ProviderThreadId;
559 readonly runningItem?: {
560 readonly id: TurnItemId;
561 readonly kind: "command" | "subagent";
562 };
563 }) => {
564 const runId = RunId.make(`run:${input.ordinal}`);
565 const attemptId = RunAttemptId.make(`attempt:${input.ordinal}`);
566 const nodeId = NodeId.make(`node:${input.ordinal}`);
567 const providerTurnId = ProviderTurnId.make(`provider-turn:${input.ordinal}`);
568 const events: Array<OrchestrationV2DomainEvent> = [
569 {
570 id: EventId.make(`run:${input.ordinal}`),
571 type: "run.created",
572 threadId,
573 occurredAt: now,
574 payload: {
575 id: runId,
576 threadId,
577 ordinal: input.ordinal,
578 providerInstanceId: instanceId,
579 modelSelection,
580 providerThreadId: input.providerThreadId,
581 userMessageId: MessageId.make(`message:${input.ordinal}`),
582 rootNodeId: nodeId,
583 activeAttemptId: attemptId,
584 status: "completed",
585 requestedAt: now,
586 startedAt: now,
587 completedAt: now,
588 checkpointId: null,
589 contextHandoffId: null,
590 },
591 },
592 {
593 id: EventId.make(`node:${input.ordinal}`),
594 type: "node.updated",
595 threadId,
596 runId,
597 occurredAt: now,
598 payload: {
599 id: nodeId,
600 threadId,
601 runId,
602 parentNodeId: null,
603 rootNodeId: nodeId,
604 kind: "root_turn",
605 status: "completed",
606 countsForRun: true,
607 providerThreadId: input.providerThreadId,
608 providerTurnId,
609 nativeItemRef: null,
610 runtimeRequestId: null,
611 checkpointScopeId: null,
612 startedAt: now,
613 completedAt: now,
614 },
615 },
616 {
617 id: EventId.make(`attempt:${input.ordinal}`),
618 type: "run-attempt.created",
619 threadId,
620 occurredAt: now,
621 payload: {
622 id: attemptId,
623 runId,
624 attemptOrdinal: 1,
625 rootNodeId: nodeId,
626 providerInstanceId: instanceId,
627 providerThreadId: input.providerThreadId,
628 providerTurnId,
629 reason: "initial",
630 status: "completed",
631 startedAt: now,
632 completedAt: now,
633 },
634 },
635 {
636 id: EventId.make(`provider-turn:${input.ordinal}`),
637 type: "provider-turn.updated",
638 threadId,
639 occurredAt: now,
640 payload: {
641 id: providerTurnId,
642 providerThreadId: input.providerThreadId,
643 nodeId,
644 runAttemptId: attemptId,
645 nativeTurnRef: null,
646 ordinal: input.ordinal,
647 status: "completed",
648 startedAt: now,
649 completedAt: now,
650 },
651 },
652 ];
653 const item = input.runningItem;
654 if (item !== undefined) {
655 const base = {
656 id: item.id,
657 threadId,
658 runId,
659 nodeId,
660 providerThreadId: input.providerThreadId,
661 providerTurnId,
662 nativeItemRef: null,
663 parentItemId: null,
664 ordinal: input.ordinal * 100,
665 status: "running",
666 title: null,
667 startedAt: now,
668 completedAt: null,
669 updatedAt: now,
670 } as const;
671 events.push({
672 id: EventId.make(`item:${input.ordinal}`),
673 type: "turn-item.updated",
674 threadId,
675 runId,
676 occurredAt: now,
677 payload:
678 item.kind === "command"
679 ? { ...base, type: "command_execution", input: "vp run test --watch" }
680 : {
681 ...base,
682 // A native subagent item names its own provider thread
683 // but its parent's provider turn.
684 providerThreadId: ProviderThreadId.make("provider-thread:codex-subagent"),
685 type: "subagent",
686 subagentId: NodeId.make(`subagent:${input.ordinal}`),
687 origin: "provider_native",
688 driver,
689 providerInstanceId: instanceId,
690 childThreadId: reviewerThreadId,
691 prompt: "Review the change",
692 result: null,
693 },
694 });
695 if (item.kind === "subagent") {
696 events.push({
697 id: EventId.make(`subagent:${input.ordinal}`),
698 type: "subagent.updated",
699 threadId,
700 runId,
701 occurredAt: now,
702 payload: {
703 ...base,
704 id: NodeId.make(`subagent:${input.ordinal}`),
705 parentNodeId: nodeId,
706 origin: "provider_native",
707 createdBy: "agent",
708 driver,
709 providerInstanceId: instanceId,
710 providerThreadId: null,
711 childThreadId: reviewerThreadId,
712 nativeTaskRef: null,
713 prompt: "Review the change",
714 model: null,
715 result: null,
716 },
717 });
718 }
719 }
720 return { runId, providerTurnId, events };
721 };
722
723 // Later Codex runs leave a second command and a native subagent. Then
724 // the thread moves on to another provider thread, which also has a
725 // live session.
726 const watcherId = TurnItemId.make("turn-item:watcher");
727 const otherProviderThreadId = ProviderThreadId.make("provider-thread:other");
728 const watcherRun = settledRun({
729 ordinal: 2,
730 providerThreadId: olderStart ? otherProviderThreadId : codexProviderThread.id,
731 ...(olderStart ? {} : { runningItem: { id: watcherId, kind: "command" as const } }),
732 });
733 const reviewerId = TurnItemId.make("turn-item:reviewer");
734 const reviewerRun = settledRun({
735 ordinal: 3,
736 providerThreadId: codexProviderThread.id,
737 runningItem: { id: reviewerId, kind: "subagent" },
738 });
739 const latestRun = settledRun({ ordinal: 4, providerThreadId: otherProviderThreadId });
740 yield* sink.write({
741 events: [
742 ...watcherRun.events,
743 ...reviewerRun.events,
744 {
745 id: EventId.make("provider-thread:other"),
746 type: "provider-thread.updated",
747 threadId,
748 occurredAt: now,
749 payload: {
750 ...codexProviderThread,
751 id: otherProviderThreadId,
752 firstRunOrdinal: 4,
753 lastRunOrdinal: 4,
754 },
755 },
756 ...latestRun.events,
757 ],
758 });
759
760 const failedRun = settledRun({ ordinal: 5, providerThreadId: otherProviderThreadId });
761 if (failedStart) {
762 yield* sink.write({
763 events: failedRun.events.flatMap((event): Array<OrchestrationV2DomainEvent> => {
764 switch (event.type) {
765 case "provider-turn.updated":
766 return [];
767 case "run.created":
768 return [{ ...event, payload: { ...event.payload, status: "failed" } }];
769 case "node.updated":
770 return [
771 {
772 ...event,
773 payload: { ...event.payload, status: "failed", providerTurnId: null },
774 },
775 ];
776 case "run-attempt.created":
777 return [
778 {
779 ...event,
780 payload: { ...event.payload, status: "failed", providerTurnId: null },
781 },
782 ];
783 default:
784 return [event];
785 }
786 }),
787 });
788 }
789
790 if (stopWithQueue !== undefined) {
791 const owner = (yield* orchestrator.getThreadProjection(threadId)).runs.find(
792 (run) => run.id === latestRun.runId,
793 )!;
794 // A message queues while checkpointing finishes, then the user holds the queue.
795 yield* sink.write({
796 events: [
797 {
798 id: EventId.make("latest-run-waiting"),
799 type: "run.updated",
800 threadId,
801 runId: owner.id,
802 occurredAt: now,
803 payload: { ...owner, status: "waiting", completedAt: null },
804 },
805 ],
806 });
807 yield* orchestrator.dispatch({
808 type: "message.dispatch",
809 commandId: CommandId.make("queue-follow-up"),
810 threadId,
811 messageId: MessageId.make("message:queue-follow-up"),
812 text: "Follow up after the background work",
813 attachments: [],
814 dispatchMode: { type: "queue_after_active" },
815 createdBy: "user",
816 creationSource: "web",
817 });
818 const queued = (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)!;
819 assert.equal(queued.status, "queued");
820 yield* sink.write({
821 events: [
822 {
823 id: EventId.make("queue-held"),
824 type: "run.updated",
825 threadId,
826 runId: queued.id,
827 occurredAt: now,
828 payload: { ...queued, queueHeld: true },
829 },
830 {
831 id: EventId.make("latest-run-settled"),
832 type: "run.updated",
833 threadId,
834 runId: owner.id,
835 occurredAt: now,
836 payload: owner,
837 },
838 ],
839 });
840 }
841
842 if (olderStart) {
843 const older = (yield* orchestrator.getThreadProjection(threadId)).runs.find(
844 (run) => run.id === watcherRun.runId,
845 )!;
846 // A resumed queue can start a lower-ordinal run after later runs have ended.
847 yield* sink.write({
848 events: [
849 {
850 id: EventId.make("older-run-starting"),
851 type: "run.updated",
852 threadId,
853 runId: older.id,
854 occurredAt: now,
855 payload: { ...older, status: "starting", startedAt: null, completedAt: null },
856 },
857 ],
858 });
859 }
860
861 yield* orchestrator.dispatch(
862 stopWithQueue === "thread.stop"
863 ? {
864 type: "thread.stop",
865 commandId: CommandId.make("stop-background-work"),
866 threadId,
867 }
868 : {
869 type: "run.interrupt",
870 commandId: CommandId.make("stop-background-work"),
871 threadId,
872 runId: olderStart
873 ? watcherRun.runId
874 : failedStart
875 ? failedRun.runId
876 : latestRun.runId,
877 ...(stopWithQueue === undefined ? {} : { holdQueue: true }),
878 },
879 );
880 yield* worker.drain();
881
882 // The Codex interrupt targets its latest pending work, the subagent's parent turn,
883 // so its settlement covers the Codex background work.
884 assert.sameDeepMembers(
885 interrupts.map((interrupt) => [interrupt.providerThread.id, interrupt.providerTurnId]),
886 [
887 ...(olderStart ? [] : [[otherProviderThreadId, latestRun.providerTurnId]]),
888 [codexProviderThread.id, reviewerRun.providerTurnId],
889 ],
890 );
891 const after = yield* orchestrator.getThreadProjection(threadId);
892 if (olderStart) {
893 assert.equal(
894 after.runs.find((run) => run.id === watcherRun.runId)?.status,
895 "interrupted",
896 );
897 }
898 if (stopWithQueue !== undefined) {
899 assert.equal(after.runs.at(-1)?.status, "queued");
900 assert.equal(after.runs.at(-1)?.queueHeld, true);
901 assert.lengthOf(started, 1);
902 }
903 assert.deepEqual(
904 [devServerId, ...(olderStart ? [] : [watcherId]), reviewerId].map(
905 (id) => after.turnItems.find((item) => item.id === id)?.status,
906 ),
907 olderStart
908 ? ["interrupted", "interrupted"]
909 : ["interrupted", "interrupted", "interrupted"],
910 );
911 // The reviewer died with its provider process, so it no longer reads as running.
912 assert.equal(after.subagents[0]?.status, "interrupted");
913 const reviewerThread = yield* orchestrator.getThreadProjection(reviewerThreadId);
914 assert.equal(reviewerThread.nodes[0]?.status, "interrupted");
915 assert.equal(reviewerThread.subagents[0]?.status, "interrupted");
916 if (workflowMember) {
917 assert.isUndefined(reviewerThread.turnItems.find((item) => item.id === nestedItemId));
918 assert.equal(
919 reviewerThread.nodes.find((node) => node.id === nestedSubagentId)?.status,
920 "interrupted",
921 );
922 } else {
923 assert.equal(
924 reviewerThread.turnItems.find((item) => item.id === nestedItemId)?.status,
925 "interrupted",
926 );
927 }
928 const testerThread = yield* orchestrator.getThreadProjection(tester.threadId);
929 assert.equal(testerThread.nodes[0]?.status, "interrupted");
930 }).pipe(
931 Effect.provide(
932 ProviderReplayHarness.layerWithRegistry(
933 { name: "background-work-stop" },
934 ProviderAdapterRegistry.layerSingle(adapter),
935 { runEffectWorker: false },
936 ),
937 ),
938 );
939 }),
940 );
941
942it.effect("Stop reaches background work an earlier provider thread still runs", () =>
943 stopEarlierBackgroundWork({}),
944);
945
946it.effect("Stop settles workflow members without subagent turn items", () =>
947 stopEarlierBackgroundWork({ workflowMember: true }),
948);
949
950it.effect(
951 "Stop reaches earlier background work after the newest run fails before provider start",
952 () => stopEarlierBackgroundWork({ failedStart: true }),
953);
954
955it.effect.each(["thread.stop", "run.interrupt"] as const)(
956 "%s stops background work when a later message is queued",
957 (stopType) => stopEarlierBackgroundWork({ stopWithQueue: stopType }),
958);
959
960it.effect.each(["thread.stop", "run.interrupt"] as const)(
961 "%s reaches later background work when an older run is starting",
962 (stopType) => stopEarlierBackgroundWork({ stopWithQueue: stopType, olderStart: true }),
963);
964
965it.effect.each([
966 "missing-session",
967 "missing-session-terminal",
968 "returned-interrupt",
969 "returned-interrupt-terminal",
970 "superseded-attempt",
971] as const)("Stop recovers a stalled run after %s without changing a newer attempt", (stalledRun) =>
972 stopEarlierBackgroundWork({ stalledRun }),
973);