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