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