1import { assert, it } from "@effect/vitest";
2import {
3 CheckpointId,
4 CheckpointRef,
5 CheckpointScopeId,
6 MessageId,
7 type ModelSelection,
8 NodeId,
9 type OrchestrationV2AppThread,
10 type OrchestrationV2DomainEvent,
11 type OrchestrationV2ExecutionNode,
12 type OrchestrationV2PlanArtifact,
13 type OrchestrationV2RuntimeRequest,
14 type OrchestrationV2ProviderThread,
15 type OrchestrationV2Run,
16 type OrchestrationV2TurnItem,
17 ProviderDriverKind,
18 ProviderInstanceId,
19 PlanId,
20 type ProviderThreadId,
21 ProviderTurnId,
22 RunAttemptId,
23 RunId,
24 RuntimeRequestId,
25 TurnItemId,
26} from "@t3tools/contracts";
27import * as DateTime from "effect/DateTime";
28import * as Deferred from "effect/Deferred";
29import * as Fiber from "effect/Fiber";
30import * as Effect from "effect/Effect";
31import * as Layer from "effect/Layer";
32import * as SqlClient from "effect/sql/SqlClient";
33import * as Stream from "effect/Stream";
34import * as TestClock from "effect/testing/TestClock";
35
36import { toolOutputImages } from "@t3tools/shared/toolOutput";
37import * as SqlitePersistence from "../persistence/Sqlite.ts";
38import * as EventSink from "./EventSink.ts";
39import * as EventStore from "./EventStore.ts";
40import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
41import * as ProjectionStore from "./ProjectionStore.ts";
42import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
43import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
44import { makeProviderFailure } from "@t3tools/provider-core/server/failure";
45import {
46 makeProviderEventRoutingState,
47 type ProviderEventRouteIdentity,
48 routeProviderEvent,
49 selectInheritedBackgroundTurnItems,
50} from "./RunExecutionService.ts";
51
52const layerTestDatabase = SqlitePersistence.layerMemory;
53const layerTestStores = Layer.merge(EventStore.layer, ProjectionStore.layer).pipe(
54 Layer.provide(layerTestDatabase),
55);
56
57const layerTestEventSink = EventSink.layer.pipe(
58 Layer.provide(Layer.mergeAll(layerTestStores, layerTestDatabase)),
59);
60
61const layerTest = Layer.mergeAll(
62 layerTestDatabase,
63 layerTestStores,
64 layerTestEventSink,
65 IdAllocator.layer,
66 ThreadCommandExecutor.layer,
67 ProviderEventIngestor.layer.pipe(
68 Layer.provide(
69 Layer.mergeAll(
70 layerTestStores,
71 layerTestEventSink,
72 IdAllocator.layer,
73 ThreadCommandExecutor.layer,
74 ),
75 ),
76 ),
77);
78const modelSelection = {
79 instanceId: ProviderInstanceId.make("codex"),
80 model: "gpt-5.4",
81} satisfies ModelSelection;
82const CODEX_DRIVER = ProviderDriverKind.make("codex");
83
84function threadCreatedEvent(
85 now: DateTime.Utc,
86): Effect.Effect<
87 OrchestrationV2DomainEvent,
88 IdAllocator.IdAllocatorV2Error,
89 IdAllocator.IdAllocatorV2
90> {
91 return Effect.gen(function* () {
92 const idAllocator = yield* IdAllocator.IdAllocatorV2;
93 const projectId = yield* idAllocator.allocate.project({
94 fixtureName: "provider-event-ingestor",
95 });
96 const threadId = yield* idAllocator.allocate.thread({
97 fixtureName: "provider-event-ingestor",
98 projectId,
99 });
100 const providerThreadId = idAllocator.derive.providerThread({
101 driver: CODEX_DRIVER,
102 nativeThreadId: "native-thread",
103 });
104 const thread: OrchestrationV2AppThread = {
105 createdBy: "user",
106 creationSource: "web",
107 id: threadId,
108 projectId,
109 title: "Provider event ingestor",
110 providerInstanceId: modelSelection.instanceId,
111 modelSelection: modelSelection,
112 runtimeMode: "full-access",
113 interactionMode: "default",
114 branch: null,
115 worktreePath: null,
116 branchPullRequest: null,
117 activeOrderKey: null,
118 activeProviderThreadId: providerThreadId,
119 lineage: {
120 parentThreadId: null,
121 relationshipToParent: null,
122 rootThreadId: threadId,
123 },
124 forkedFrom: null,
125 createdAt: now,
126 updatedAt: now,
127 archivedAt: null,
128 settledOverride: null,
129 settledAt: null,
130 lastVisitedAt: null,
131 deletedAt: null,
132 };
133
134 return {
135 id: yield* idAllocator.allocate.event({ threadId }),
136 type: "thread.created",
137 threadId,
138 occurredAt: now,
139 payload: thread,
140 };
141 });
142}
143
144const layer = it.layer(layerTest);
145
146layer("ProviderEventIngestorV2", (it) => {
147 it.effect("normalizes provider events through the real event log and projection store", () =>
148 Effect.gen(function* () {
149 const now = yield* DateTime.now;
150 const eventSink = yield* EventSink.EventSinkV2;
151 const eventStore = yield* EventStore.EventStoreV2;
152 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
153 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
154 const idAllocator = yield* IdAllocator.IdAllocatorV2;
155 const threadEvent = yield* threadCreatedEvent(now);
156 const providerSessionId = yield* idAllocator.allocate.providerSession({
157 providerInstanceId: modelSelection.instanceId,
158 threadId: threadEvent.threadId,
159 });
160 const providerThread: OrchestrationV2ProviderThread = {
161 id: idAllocator.derive.providerThread({
162 driver: CODEX_DRIVER,
163 nativeThreadId: "native-thread",
164 }),
165 driver: CODEX_DRIVER,
166 providerInstanceId: modelSelection.instanceId,
167 providerSessionId,
168 appThreadId: threadEvent.threadId,
169 ownerNodeId: null,
170 nativeThreadRef: {
171 driver: CODEX_DRIVER,
172 nativeId: "native-thread",
173 strength: "strong",
174 },
175 nativeConversationHeadRef: null,
176 status: "idle",
177 firstRunOrdinal: null,
178 lastRunOrdinal: null,
179 handoffIds: [],
180 forkedFrom: null,
181 createdAt: now,
182 updatedAt: now,
183 };
184
185 yield* eventSink.write({ events: [threadEvent] });
186 const storedEvents = yield* ingestor.ingestNormalized({
187 providerSessionId,
188 providerInstanceId: modelSelection.instanceId,
189 threadId: threadEvent.threadId,
190 event: {
191 type: "provider_thread.updated",
192 driver: CODEX_DRIVER,
193 providerThread,
194 },
195 });
196
197 const projection = yield* projectionStore.getThreadProjection(threadEvent.threadId);
198 const storedDomainEvents = yield* eventStore.read({}).pipe(Stream.runCollect);
199 const afterFirstEvent = yield* eventStore
200 .read({ afterSequence: 1, threadId: threadEvent.threadId })
201 .pipe(Stream.runCollect);
202 const latestThreadSequence = yield* eventStore.latestSequence({
203 threadId: threadEvent.threadId,
204 });
205
206 assert.equal(storedEvents.length, 1);
207 assert.equal(storedEvents[0]?.event.type, "provider-thread.updated");
208 assert.deepEqual(
209 projection.providerThreads.map((thread) => thread.id),
210 [providerThread.id],
211 );
212 assert.deepEqual(
213 Array.from(storedDomainEvents).map((stored) => stored.event.type),
214 ["thread.created", "provider-thread.updated"],
215 );
216 assert.deepEqual(
217 Array.from(storedDomainEvents).map((stored) => stored.sequence),
218 [1, 2],
219 );
220 assert.deepEqual(
221 Array.from(afterFirstEvent).map((stored) => stored.event.type),
222 ["provider-thread.updated"],
223 );
224 assert.equal(latestThreadSequence, 2);
225 }),
226 );
227
228 it.effect("skips a node update only when it matches the stored node", () =>
229 Effect.gen(function* () {
230 const now = yield* DateTime.now;
231 const eventSink = yield* EventSink.EventSinkV2;
232 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
233 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
234 const idAllocator = yield* IdAllocator.IdAllocatorV2;
235 const threadEvent = yield* threadCreatedEvent(now);
236 const threadId = threadEvent.threadId;
237 yield* eventSink.write({ events: [threadEvent] });
238 const providerSessionId = yield* idAllocator.allocate.providerSession({
239 providerInstanceId: modelSelection.instanceId,
240 threadId,
241 });
242 const nodeId = NodeId.make(`${threadId}:approval`);
243 const node: OrchestrationV2ExecutionNode = {
244 id: nodeId,
245 threadId,
246 runId: null,
247 parentNodeId: null,
248 rootNodeId: nodeId,
249 kind: "approval_request",
250 status: "waiting",
251 countsForRun: false,
252 providerThreadId: null,
253 providerTurnId: null,
254 nativeItemRef: null,
255 runtimeRequestId: null,
256 checkpointScopeId: null,
257 startedAt: now,
258 completedAt: null,
259 };
260 const ingestNode = (sent: OrchestrationV2ExecutionNode) =>
261 ingestor.ingestNormalized({
262 providerSessionId,
263 providerInstanceId: modelSelection.instanceId,
264 threadId,
265 event: { type: "node.updated", driver: CODEX_DRIVER, node: sent },
266 });
267
268 assert.lengthOf(yield* ingestNode(node), 1);
269 assert.isEmpty(yield* ingestNode({ ...node }));
270 // The server resolves the request, then the provider asks again with the same node.
271 yield* eventSink.write({
272 events: [
273 {
274 id: yield* idAllocator.allocate.event({ threadId }),
275 type: "node.updated",
276 threadId,
277 occurredAt: now,
278 payload: { ...node, status: "completed", completedAt: now },
279 },
280 ],
281 });
282 assert.lengthOf(yield* ingestNode(node), 1);
283 const projection = yield* projectionStore.getThreadProjection(threadId);
284 assert.equal(projection.nodes.find((entry) => entry.id === nodeId)?.status, "waiting");
285 }),
286 );
287
288 it.effect("rolls back runs a provider rewound off its active branch", () =>
289 Effect.gen(function* () {
290 const now = yield* DateTime.now;
291 const eventSink = yield* EventSink.EventSinkV2;
292 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
293 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
294 const idAllocator = yield* IdAllocator.IdAllocatorV2;
295 const threadEvent = yield* threadCreatedEvent(now);
296 const threadId = threadEvent.threadId;
297 const providerSessionId = yield* idAllocator.allocate.providerSession({
298 providerInstanceId: modelSelection.instanceId,
299 threadId,
300 });
301 const rewoundThreadId = idAllocator.derive.providerThread({
302 driver: CODEX_DRIVER,
303 nativeThreadId: "rewound-thread",
304 });
305 const otherThreadId = idAllocator.derive.providerThread({
306 driver: CODEX_DRIVER,
307 nativeThreadId: "other-thread",
308 });
309 const scopeId = CheckpointScopeId.make("checkpoint-scope:rewound");
310 // A run as execution leaves it: attempt, provider turn, root node, and a
311 // checkpoint once completed. A null native turn id is a weak ref.
312 const seedRun = Effect.fnUntraced(function* (input: {
313 readonly ordinal: number;
314 readonly providerThreadId: ProviderThreadId;
315 readonly status: "completed" | "interrupted" | "running";
316 readonly nativeTurnId: string | null;
317 }) {
318 const runId = RunId.make(`run:rewound:${input.ordinal}`);
319 const attemptId = RunAttemptId.make(`attempt:rewound:${input.ordinal}`);
320 const nodeId = NodeId.make(`node:rewound:${input.ordinal}`);
321 const providerTurnId = ProviderTurnId.make(`provider-turn:rewound:${input.ordinal}`);
322 const completedAt = input.status === "running" ? null : now;
323 const eventId = () => idAllocator.allocate.event({ threadId });
324 const events: Array<OrchestrationV2DomainEvent> = [
325 {
326 id: yield* eventId(),
327 type: "run.created",
328 threadId,
329 occurredAt: now,
330 payload: {
331 id: runId,
332 threadId,
333 ordinal: input.ordinal,
334 providerInstanceId: modelSelection.instanceId,
335 modelSelection,
336 providerThreadId: input.providerThreadId,
337 userMessageId: MessageId.make(`message:rewound:${input.ordinal}`),
338 rootNodeId: nodeId,
339 activeAttemptId: attemptId,
340 status: input.status,
341 requestedAt: now,
342 startedAt: now,
343 completedAt,
344 checkpointId: null,
345 contextHandoffId: null,
346 },
347 },
348 {
349 id: yield* eventId(),
350 type: "run-attempt.created",
351 threadId,
352 occurredAt: now,
353 payload: {
354 id: attemptId,
355 runId,
356 attemptOrdinal: 1,
357 rootNodeId: nodeId,
358 providerInstanceId: modelSelection.instanceId,
359 providerThreadId: input.providerThreadId,
360 providerTurnId,
361 reason: "initial",
362 status: input.status,
363 startedAt: now,
364 completedAt,
365 },
366 },
367 {
368 id: yield* eventId(),
369 type: "provider-turn.updated",
370 threadId,
371 occurredAt: now,
372 payload: {
373 id: providerTurnId,
374 providerThreadId: input.providerThreadId,
375 nodeId,
376 runAttemptId: attemptId,
377 nativeTurnRef:
378 input.nativeTurnId === null
379 ? {
380 driver: CODEX_DRIVER,
381 nativeId: `synthetic:${input.ordinal}`,
382 strength: "weak",
383 }
384 : { driver: CODEX_DRIVER, nativeId: input.nativeTurnId, strength: "strong" },
385 ordinal: input.ordinal,
386 status: input.status,
387 startedAt: now,
388 completedAt,
389 },
390 },
391 {
392 id: yield* eventId(),
393 type: "node.updated",
394 threadId,
395 occurredAt: now,
396 payload: {
397 id: nodeId,
398 threadId,
399 runId,
400 parentNodeId: null,
401 rootNodeId: nodeId,
402 kind: "root_turn",
403 status: input.status,
404 countsForRun: true,
405 providerThreadId: input.providerThreadId,
406 providerTurnId,
407 nativeItemRef: null,
408 runtimeRequestId: null,
409 checkpointScopeId: scopeId,
410 startedAt: now,
411 completedAt,
412 },
413 },
414 ...(input.status === "completed"
415 ? [
416 {
417 id: yield* eventId(),
418 type: "checkpoint.captured" as const,
419 threadId,
420 occurredAt: now,
421 payload: {
422 id: CheckpointId.make(`checkpoint:rewound:${input.ordinal}`),
423 threadId,
424 scopeId,
425 runId,
426 nodeId,
427 parentCheckpointId: null,
428 ordinalWithinScope: input.ordinal,
429 appRunOrdinal: input.ordinal,
430 ref: CheckpointRef.make(`refs/t3/checkpoints/rewound/${input.ordinal}`),
431 status: "ready" as const,
432 files: [],
433 capturedAt: now,
434 },
435 },
436 ]
437 : []),
438 ];
439 yield* eventSink.write({ events });
440 return runId;
441 });
442
443 yield* eventSink.write({ events: [threadEvent] });
444 const otherProviderRun = yield* seedRun({
445 ordinal: 1,
446 providerThreadId: otherThreadId,
447 status: "completed",
448 nativeTurnId: "other-turn",
449 });
450 const keptRun = yield* seedRun({
451 ordinal: 2,
452 providerThreadId: rewoundThreadId,
453 status: "completed",
454 nativeTurnId: "kept-turn",
455 });
456 const abandonedRun = yield* seedRun({
457 ordinal: 3,
458 providerThreadId: rewoundThreadId,
459 status: "completed",
460 nativeTurnId: "abandoned-turn",
461 });
462 const unlocatedRun = yield* seedRun({
463 ordinal: 4,
464 providerThreadId: rewoundThreadId,
465 status: "interrupted",
466 nativeTurnId: null,
467 });
468 const rewindingRun = yield* seedRun({
469 ordinal: 5,
470 providerThreadId: rewoundThreadId,
471 status: "running",
472 nativeTurnId: null,
473 });
474
475 yield* ingestor.ingestNormalized({
476 providerSessionId,
477 providerInstanceId: modelSelection.instanceId,
478 threadId,
479 runId: rewindingRun,
480 event: {
481 type: "provider_thread.updated",
482 driver: CODEX_DRIVER,
483 providerThread: {
484 id: rewoundThreadId,
485 driver: CODEX_DRIVER,
486 providerInstanceId: modelSelection.instanceId,
487 providerSessionId,
488 appThreadId: threadId,
489 ownerNodeId: null,
490 nativeThreadRef: {
491 driver: CODEX_DRIVER,
492 nativeId: "rewound-thread",
493 strength: "strong",
494 },
495 nativeConversationHeadRef: {
496 driver: CODEX_DRIVER,
497 nativeId: "kept-turn-reply",
498 strength: "strong",
499 },
500 status: "idle",
501 firstRunOrdinal: 2,
502 lastRunOrdinal: 5,
503 handoffIds: [],
504 forkedFrom: null,
505 createdAt: now,
506 updatedAt: now,
507 },
508 retainedNativeTurnIds: ["turn-started-outside-t3", "kept-turn"],
509 },
510 });
511
512 const projection = yield* projectionStore.getThreadProjection(threadId);
513 const runIds = [otherProviderRun, keptRun, abandonedRun, unlocatedRun, rewindingRun];
514 assert.deepEqual(
515 runIds.map((runId) => projection.runs.find((run) => run.id === runId)?.status),
516 ["completed", "completed", "rolled_back", "rolled_back", "running"],
517 );
518 assert.deepEqual(
519 runIds.map((runId) => projection.nodes.find((node) => node.runId === runId)?.status),
520 ["completed", "completed", "rolled_back", "rolled_back", "running"],
521 );
522 assert.deepEqual(
523 [otherProviderRun, keptRun, abandonedRun].map(
524 (runId) =>
525 projection.checkpoints.find((checkpoint) => checkpoint.runId === runId)?.status,
526 ),
527 ["ready", "ready", "stale"],
528 );
529 assert.equal(
530 projection.providerThreads.find((thread) => thread.id === rewoundThreadId)?.lastRunOrdinal,
531 5,
532 );
533 }),
534 );
535
536 it.effect("carries plan-step durations through consecutive and restarted ingestion", () =>
537 Effect.gen(function* () {
538 yield* TestClock.setTime(Date.parse("2026-09-07T00:00:00.000Z"));
539 const now = yield* DateTime.now;
540 const eventSink = yield* EventSink.EventSinkV2;
541 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
542 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
543 const idAllocator = yield* IdAllocator.IdAllocatorV2;
544 const threadEvent = yield* threadCreatedEvent(now);
545 const providerSessionId = yield* idAllocator.allocate.providerSession({
546 providerInstanceId: modelSelection.instanceId,
547 threadId: threadEvent.threadId,
548 });
549 const planId = PlanId.make("plan:provider-event-duration");
550 const nodeId = NodeId.make("node:provider-event-duration");
551 type TodoListPlan = Extract<OrchestrationV2PlanArtifact, { readonly kind: "todo_list" }>;
552 const plan = (steps: TodoListPlan["steps"]): TodoListPlan => ({
553 id: planId,
554 threadId: threadEvent.threadId,
555 runId: null,
556 nodeId,
557 kind: "todo_list",
558 status: "active",
559 steps,
560 });
561 const ingest = (
562 service: ProviderEventIngestor.ProviderEventIngestorV2["Service"],
563 steps: TodoListPlan["steps"],
564 ) =>
565 service.ingestNormalized({
566 providerSessionId,
567 providerInstanceId: modelSelection.instanceId,
568 threadId: threadEvent.threadId,
569 event: { type: "plan.updated", driver: CODEX_DRIVER, plan: plan(steps) },
570 });
571
572 yield* eventSink.write({ events: [threadEvent] });
573 yield* ingest(ingestor, [
574 { id: "duplicate-a", text: "Verify", status: "running" },
575 { id: "duplicate-b", text: "Verify", status: "pending" },
576 { id: "fallback", text: "Report", status: "pending" },
577 ]);
578 yield* TestClock.adjust("3 seconds");
579 yield* ingest(ingestor, [
580 { id: "duplicate-a", text: "Verify", status: "completed" },
581 { id: "duplicate-b", text: "Verify", status: "pending" },
582 { id: "fallback", text: "Report", status: "pending" },
583 ]);
584
585 const restartedIngestor = yield* ProviderEventIngestor.ProviderEventIngestorV2.pipe(
586 Effect.provide(
587 Layer.fresh(ProviderEventIngestor.layer).pipe(
588 Layer.provide(
589 Layer.succeed(ProjectionStore.ProjectionStoreV2, {
590 ...projectionStore,
591 getThreadProjection: () => Effect.die("Plan timing must not load thread history"),
592 }),
593 ),
594 ),
595 ),
596 );
597 yield* TestClock.adjust("4 seconds");
598 yield* ingest(restartedIngestor, [
599 { id: "duplicate-a", text: "Verify", status: "completed" },
600 { id: "duplicate-b", text: "Verify", status: "completed" },
601 { id: "fallback", text: "Report", status: "pending" },
602 ]);
603 yield* TestClock.adjust("5 seconds");
604 yield* ingest(restartedIngestor, [
605 { id: "duplicate-a", text: "Verify", status: "completed" },
606 { id: "duplicate-b", text: "Verify", status: "completed" },
607 { id: "fallback", text: "Report", status: "completed" },
608 ]);
609
610 const projection = yield* projectionStore.getThreadProjection(threadEvent.threadId);
611 const persisted = projection.plans.find(
612 (candidate): candidate is TodoListPlan =>
613 candidate.kind === "todo_list" && candidate.id === planId,
614 );
615 assert.deepEqual(
616 persisted?.steps.map(({ id, text, status, durationMs }) => ({
617 id,
618 text,
619 status,
620 durationMs,
621 })),
622 [
623 { id: "duplicate-a", text: "Verify", status: "completed", durationMs: 3_000 },
624 { id: "duplicate-b", text: "Verify", status: "completed", durationMs: 4_000 },
625 { id: "fallback", text: "Report", status: "completed", durationMs: 5_000 },
626 ],
627 );
628
629 yield* ingest(restartedIngestor, [
630 { id: "duplicate-a", text: "Inserted task", status: "running" },
631 { id: "duplicate-b", text: "Verify", status: "completed" },
632 { id: "fallback", text: "Different completed task", status: "completed" },
633 ]);
634 yield* TestClock.adjust("2 seconds");
635 yield* ingest(restartedIngestor, [
636 { id: "duplicate-a", text: "Inserted task", status: "completed" },
637 { id: "duplicate-b", text: "Verify", status: "completed" },
638 { id: "fallback", text: "Different completed task", status: "completed" },
639 ]);
640 const updated = yield* projectionStore.getThreadProjection(threadEvent.threadId);
641 const changedPlan = updated.plans.find(
642 (candidate): candidate is TodoListPlan =>
643 candidate.kind === "todo_list" && candidate.id === planId,
644 );
645 assert.deepEqual(
646 changedPlan?.steps.map((step) => step.durationMs),
647 [2_000, 4_000, undefined],
648 );
649 }),
650 );
651
652 it.effect(
653 "treats successful provider terminal markers as non-persisted orchestration control signals",
654 () =>
655 Effect.gen(function* () {
656 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
657 const idAllocator = yield* IdAllocator.IdAllocatorV2;
658 const projectId = yield* idAllocator.allocate.project({
659 fixtureName: "provider-event-terminal",
660 });
661 const threadId = yield* idAllocator.allocate.thread({
662 fixtureName: "provider-event-terminal",
663 projectId,
664 });
665 const providerSessionId = yield* idAllocator.allocate.providerSession({
666 providerInstanceId: modelSelection.instanceId,
667 threadId,
668 });
669 const normalized = yield* ingestor.normalize({
670 providerSessionId,
671 providerInstanceId: modelSelection.instanceId,
672 threadId,
673 event: {
674 type: "turn.terminal",
675 driver: CODEX_DRIVER,
676 providerThreadId: idAllocator.derive.providerThread({
677 driver: CODEX_DRIVER,
678 nativeThreadId: "native-thread",
679 }),
680 providerTurnId: idAllocator.derive.providerTurn({
681 driver: CODEX_DRIVER,
682 nativeTurnId: "native-turn",
683 }),
684 runOrdinal: 1,
685 status: "completed",
686 failure: null,
687 threadDisposition: "reusable",
688 },
689 });
690
691 assert.deepEqual(normalized, []);
692 }),
693 );
694
695 it.effect("persists an interrupted run's inherited terminal through the live run router", () =>
696 Effect.gen(function* () {
697 const now = yield* DateTime.now;
698 const eventSink = yield* EventSink.EventSinkV2;
699 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
700 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
701 const idAllocator = yield* IdAllocator.IdAllocatorV2;
702 const threadEvent = yield* threadCreatedEvent(now);
703 const priorRunId = RunId.make("run:provider-event-inherited:prior");
704 const currentRunId = RunId.make("run:provider-event-inherited:current");
705 const itemId = TurnItemId.make("turn-item:provider-event-inherited");
706 const providerSessionId = yield* idAllocator.allocate.providerSession({
707 providerInstanceId: modelSelection.instanceId,
708 threadId: threadEvent.threadId,
709 });
710 const providerThreadId = idAllocator.derive.providerThread({
711 driver: CODEX_DRIVER,
712 nativeThreadId: "native-thread-inherited",
713 });
714 const providerTurnId = idAllocator.derive.providerTurn({
715 driver: CODEX_DRIVER,
716 nativeTurnId: "native-turn-inherited",
717 });
718 const runningItem = {
719 id: itemId,
720 threadId: threadEvent.threadId,
721 runId: priorRunId,
722 nodeId: NodeId.make("node:provider-event-inherited"),
723 providerThreadId,
724 providerTurnId,
725 nativeItemRef: null,
726 parentItemId: null,
727 ordinal: 101,
728 status: "running",
729 title: "Inherited background command",
730 startedAt: now,
731 completedAt: null,
732 updatedAt: now,
733 type: "command_execution",
734 input: "sleep 60",
735 } satisfies OrchestrationV2TurnItem;
736 const terminalItem = {
737 ...runningItem,
738 status: "completed" as const,
739 completedAt: now,
740 updatedAt: now,
741 };
742
743 yield* eventSink.write({ events: [threadEvent] });
744 yield* ingestor.ingestNormalized({
745 providerSessionId,
746 providerInstanceId: modelSelection.instanceId,
747 threadId: threadEvent.threadId,
748 runId: priorRunId,
749 event: { type: "turn_item.updated", driver: CODEX_DRIVER, turnItem: runningItem },
750 });
751
752 const identity: ProviderEventRouteIdentity = {
753 threadId: threadEvent.threadId,
754 runId: currentRunId,
755 attemptId: RunAttemptId.make("attempt:provider-event-inherited:current"),
756 providerThreadId,
757 };
758 const inheritedBackgroundTurnItems = selectInheritedBackgroundTurnItems({
759 threadId: threadEvent.threadId,
760 currentProviderThreadId: providerThreadId,
761 currentRunOrdinal: 2,
762 runs: [
763 {
764 id: priorRunId,
765 threadId: threadEvent.threadId,
766 ordinal: 1,
767 status: "interrupted",
768 } as OrchestrationV2Run,
769 {
770 id: currentRunId,
771 threadId: threadEvent.threadId,
772 ordinal: 2,
773 status: "running",
774 } as OrchestrationV2Run,
775 ],
776 turnItems: [runningItem],
777 });
778 const routeState = makeProviderEventRoutingState({
779 identity,
780 inheritedBackgroundTurnItems,
781 providerTurnId: null,
782 });
783 const terminalEvent = {
784 type: "turn_item.updated",
785 driver: CODEX_DRIVER,
786 turnItem: terminalItem,
787 } as const;
788 const [accepted] = routeProviderEvent(terminalEvent, identity, routeState);
789 assert.isTrue(accepted);
790
791 const stored = yield* ingestor.ingestNormalized({
792 providerSessionId,
793 providerInstanceId: modelSelection.instanceId,
794 threadId: threadEvent.threadId,
795 runId: currentRunId,
796 event: terminalEvent,
797 });
798 const projection = yield* projectionStore.getThreadProjection(threadEvent.threadId);
799 const persisted = projection.turnItems.find((item) => item.id === itemId);
800
801 assert.equal(stored.length, 1);
802 assert.equal(stored[0]?.event.type, "turn-item.updated");
803 assert.equal(persisted?.runId, priorRunId);
804 assert.equal(persisted?.threadId, threadEvent.threadId);
805 assert.equal(persisted?.status, "completed");
806 }),
807 );
808
809 it.effect("persists a completed run's late background terminal exactly once", () =>
810 Effect.gen(function* () {
811 const now = yield* DateTime.now;
812 const eventSink = yield* EventSink.EventSinkV2;
813 const eventStore = yield* EventStore.EventStoreV2;
814 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
815 const idAllocator = yield* IdAllocator.IdAllocatorV2;
816 const threadEvent = yield* threadCreatedEvent(now);
817 const priorRunId = RunId.make("run:provider-event-completed:prior");
818 const currentRunId = RunId.make("run:provider-event-completed:current");
819 const itemId = TurnItemId.make("turn-item:provider-event-completed");
820 const providerSessionId = yield* idAllocator.allocate.providerSession({
821 providerInstanceId: modelSelection.instanceId,
822 threadId: threadEvent.threadId,
823 });
824 const providerThreadId = idAllocator.derive.providerThread({
825 driver: CODEX_DRIVER,
826 nativeThreadId: "native-thread-completed",
827 });
828 const providerTurnId = idAllocator.derive.providerTurn({
829 driver: CODEX_DRIVER,
830 nativeTurnId: "native-turn-completed",
831 });
832 const runningItem = {
833 id: itemId,
834 threadId: threadEvent.threadId,
835 runId: priorRunId,
836 nodeId: NodeId.make("node:provider-event-completed"),
837 providerThreadId,
838 providerTurnId,
839 nativeItemRef: null,
840 parentItemId: null,
841 ordinal: 101,
842 status: "running",
843 title: "Completed run background command",
844 startedAt: now,
845 completedAt: null,
846 updatedAt: now,
847 type: "command_execution",
848 input: "sleep 60",
849 } satisfies OrchestrationV2TurnItem;
850 const terminalEvent = {
851 type: "turn_item.updated",
852 driver: CODEX_DRIVER,
853 turnItem: {
854 ...runningItem,
855 status: "completed" as const,
856 completedAt: now,
857 updatedAt: now,
858 },
859 } as const;
860
861 yield* eventSink.write({ events: [threadEvent] });
862 yield* ingestor.ingestNormalized({
863 providerSessionId,
864 providerInstanceId: modelSelection.instanceId,
865 threadId: threadEvent.threadId,
866 runId: priorRunId,
867 event: { type: "turn_item.updated", driver: CODEX_DRIVER, turnItem: runningItem },
868 });
869
870 const priorIdentity: ProviderEventRouteIdentity = {
871 threadId: threadEvent.threadId,
872 runId: priorRunId,
873 attemptId: RunAttemptId.make("attempt:provider-event-completed:prior"),
874 providerThreadId,
875 };
876 const currentIdentity: ProviderEventRouteIdentity = {
877 threadId: threadEvent.threadId,
878 runId: currentRunId,
879 attemptId: RunAttemptId.make("attempt:provider-event-completed:current"),
880 providerThreadId,
881 };
882 const inheritedBackgroundTurnItems = selectInheritedBackgroundTurnItems({
883 threadId: threadEvent.threadId,
884 currentProviderThreadId: providerThreadId,
885 currentRunOrdinal: 2,
886 runs: [
887 {
888 id: priorRunId,
889 threadId: threadEvent.threadId,
890 ordinal: 1,
891 status: "completed",
892 } as OrchestrationV2Run,
893 {
894 id: currentRunId,
895 threadId: threadEvent.threadId,
896 ordinal: 2,
897 status: "running",
898 } as OrchestrationV2Run,
899 ],
900 turnItems: [runningItem],
901 });
902 const routers = [
903 {
904 identity: priorIdentity,
905 state: makeProviderEventRoutingState({
906 identity: priorIdentity,
907 providerTurnId: providerTurnId,
908 }),
909 },
910 {
911 identity: currentIdentity,
912 state: makeProviderEventRoutingState({
913 identity: currentIdentity,
914 inheritedBackgroundTurnItems,
915 providerTurnId: null,
916 }),
917 },
918 ];
919 const acceptedRouters = routers.filter(
920 ({ identity, state }) => routeProviderEvent(terminalEvent, identity, state)[0],
921 );
922
923 yield* Effect.forEach(
924 acceptedRouters,
925 ({ identity }) =>
926 ingestor.ingestNormalized({
927 providerSessionId,
928 providerInstanceId: modelSelection.instanceId,
929 threadId: threadEvent.threadId,
930 runId: identity.runId,
931 event: terminalEvent,
932 }),
933 { concurrency: 1 },
934 );
935
936 const storedEvents = yield* eventStore
937 .read({ threadId: threadEvent.threadId })
938 .pipe(Stream.runCollect);
939 const storedTerminals = Array.from(storedEvents).filter(
940 (stored) =>
941 stored.event.type === "turn-item.updated" &&
942 stored.event.payload.id === itemId &&
943 stored.event.payload.status === "completed",
944 );
945
946 assert.equal(storedTerminals.length, 1);
947 assert.equal(acceptedRouters.length, 1);
948 assert.equal(acceptedRouters[0]?.identity.runId, priorRunId);
949 }),
950 );
951
952 it.effect.each(["completed", "interrupted", "failed", "cancelled", "control"] as const)(
953 "dismisses only native questions when a provider turn ends with %s",
954 (terminal) =>
955 Effect.gen(function* () {
956 const now = yield* DateTime.now;
957 const eventSink = yield* EventSink.EventSinkV2;
958 const eventStore = yield* EventStore.EventStoreV2;
959 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
960 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
961 const idAllocator = yield* IdAllocator.IdAllocatorV2;
962 const threadEvent = yield* threadCreatedEvent(now);
963 const threadId = threadEvent.threadId;
964 const providerSessionId = yield* idAllocator.allocate.providerSession({
965 providerInstanceId: modelSelection.instanceId,
966 threadId,
967 });
968 const providerThreadId = idAllocator.derive.providerThread({
969 driver: CODEX_DRIVER,
970 nativeThreadId: `${threadId}:questions`,
971 });
972 const providerTurnId = idAllocator.derive.providerTurn({
973 driver: CODEX_DRIVER,
974 nativeTurnId: `${threadId}:questions`,
975 });
976 const otherTurnId = idAllocator.derive.providerTurn({
977 driver: CODEX_DRIVER,
978 nativeTurnId: `${threadId}:other-turn`,
979 });
980 const specs = [
981 { key: "native", responseCapability: { type: "live", providerSessionId } },
982 {
983 key: "unavailable",
984 responseCapability: { type: "not_resumable", reason: "Turn ended" },
985 },
986 { key: "message", responseCapability: { type: "message" } },
987 { key: "answered", responseCapability: { type: "live", providerSessionId } },
988 { key: "other-turn", responseCapability: { type: "live", providerSessionId } },
989 { key: "approval", responseCapability: { type: "live", providerSessionId } },
990 ] as const;
991 const fixtures = specs.map((spec, ordinal) => {
992 const nodeId = NodeId.make(`${threadId}:${spec.key}`);
993 const resolved = spec.key === "answered";
994 const request: OrchestrationV2RuntimeRequest = {
995 id: RuntimeRequestId.make(`${threadId}:${spec.key}`),
996 nodeId,
997 providerTurnId: spec.key === "other-turn" ? otherTurnId : providerTurnId,
998 nativeRequestRef: null,
999 kind: spec.key === "approval" ? "command" : "user_input",
1000 status: resolved ? "resolved" : "pending",
1001 responseCapability: spec.responseCapability,
1002 createdAt: now,
1003 resolvedAt: resolved ? now : null,
1004 };
1005 const node: OrchestrationV2ExecutionNode = {
1006 id: nodeId,
1007 threadId,
1008 runId: null,
1009 parentNodeId: null,
1010 rootNodeId: nodeId,
1011 kind: spec.key === "approval" ? "approval_request" : "user_input_request",
1012 status: resolved ? "completed" : "waiting",
1013 countsForRun: false,
1014 providerThreadId,
1015 providerTurnId: request.providerTurnId,
1016 nativeItemRef: null,
1017 runtimeRequestId: request.id,
1018 checkpointScopeId: null,
1019 startedAt: now,
1020 completedAt: resolved ? now : null,
1021 };
1022 const item: OrchestrationV2TurnItem = {
1023 id: TurnItemId.make(`${threadId}:${spec.key}`),
1024 threadId,
1025 runId: null,
1026 nodeId,
1027 providerThreadId,
1028 providerTurnId: request.providerTurnId,
1029 nativeItemRef: null,
1030 parentItemId: null,
1031 ordinal,
1032 status: resolved ? "completed" : "waiting",
1033 title: "Which option?",
1034 startedAt: now,
1035 completedAt: resolved ? now : null,
1036 updatedAt: now,
1037 ...(spec.key === "approval"
1038 ? { type: "approval_request", requestId: request.id, requestKind: "command" }
1039 : {
1040 type: "user_input_request",
1041 requestId: request.id,
1042 questions: [],
1043 ...(spec.key === "message" ? { responseMode: "message" as const } : {}),
1044 }),
1045 };
1046 return { key: spec.key, request, node, item };
1047 });
1048 const seedEvents: Array<OrchestrationV2DomainEvent> = [threadEvent];
1049 for (const fixture of fixtures) {
1050 for (const payload of [
1051 { type: "runtime-request.updated" as const, payload: fixture.request },
1052 { type: "node.updated" as const, payload: fixture.node },
1053 { type: "turn-item.updated" as const, payload: fixture.item },
1054 ]) {
1055 seedEvents.push({
1056 id: yield* idAllocator.allocate.event({ threadId }),
1057 threadId,
1058 occurredAt: now,
1059 ...payload,
1060 });
1061 }
1062 }
1063 yield* eventSink.write({ events: seedEvents });
1064 const input = {
1065 providerSessionId,
1066 providerInstanceId: modelSelection.instanceId,
1067 threadId,
1068 event:
1069 terminal === "control"
1070 ? {
1071 type: "turn.terminal" as const,
1072 driver: CODEX_DRIVER,
1073 providerThreadId,
1074 providerTurnId,
1075 runOrdinal: 1,
1076 status: "completed" as const,
1077 failure: null,
1078 threadDisposition: "reusable" as const,
1079 }
1080 : {
1081 type: "provider_turn.updated" as const,
1082 driver: CODEX_DRIVER,
1083 providerTurn: {
1084 id: providerTurnId,
1085 providerThreadId,
1086 nodeId: NodeId.make(`${threadId}:root`),
1087 runAttemptId: null,
1088 nativeTurnRef: null,
1089 ordinal: 1,
1090 status: terminal,
1091 startedAt: now,
1092 completedAt: now,
1093 },
1094 },
1095 };
1096 const stored = yield* ingestor.ingestNormalized(input);
1097 const projection = yield* projectionStore.getThreadProjection(threadId);
1098 for (const fixture of fixtures) {
1099 const closed = fixture.key === "native" || fixture.key === "unavailable";
1100 const request = projection.runtimeRequests.find(
1101 (item) => item.id === fixture.request.id,
1102 )!;
1103 const node = projection.nodes.find((item) => item.id === fixture.node.id)!;
1104 const item = projection.turnItems.find((item) => item.id === fixture.item.id)!;
1105 if (closed) {
1106 assert.equal(request.status, "cancelled");
1107 assert.isNotNull(request.resolvedAt);
1108 assert.equal(node.status, "cancelled");
1109 assert.isNotNull(node.completedAt);
1110 assert.equal(item.status, "cancelled");
1111 assert.isNotNull(item.completedAt);
1112 } else {
1113 assert.equal(request.status, fixture.request.status);
1114 assert.equal(node.status, fixture.node.status);
1115 assert.equal(item.status, fixture.item.status);
1116 }
1117 }
1118 assert.equal(
1119 stored.filter((entry) => entry.event.type === "runtime-request.updated").length,
1120 2,
1121 );
1122 assert.isEmpty(
1123 (yield* projectionStore.getPendingNativeUserInputs(threadId, providerTurnId))
1124 .runtimeRequests,
1125 );
1126 const replayed = yield* eventStore.read({ threadId }).pipe(Stream.runCollect);
1127 assert.equal(
1128 replayed.filter(
1129 (entry) =>
1130 entry.event.type === "runtime-request.updated" &&
1131 entry.event.payload.status === "cancelled",
1132 ).length,
1133 2,
1134 );
1135 const repeated = yield* ingestor.ingestNormalized(input);
1136 assert.isFalse(repeated.some((entry) => entry.event.type === "runtime-request.updated"));
1137 }),
1138 );
1139
1140 it.effect(
1141 "preserves an answer committed after terminal normalization reads a pending question",
1142 () =>
1143 Effect.scoped(
1144 Effect.gen(function* () {
1145 const eventSink = yield* EventSink.EventSinkV2;
1146 const eventStore = yield* EventStore.EventStoreV2;
1147 const projections = yield* ProjectionStore.ProjectionStoreV2;
1148 const idAllocator = yield* IdAllocator.IdAllocatorV2;
1149 const now = yield* DateTime.now;
1150 const threadEvent = yield* threadCreatedEvent(now);
1151 const threadId = threadEvent.threadId;
1152 const providerSessionId = yield* idAllocator.allocate.providerSession({
1153 providerInstanceId: modelSelection.instanceId,
1154 threadId,
1155 });
1156 const providerThreadId = idAllocator.derive.providerThread({
1157 driver: CODEX_DRIVER,
1158 nativeThreadId: `${threadId}:race`,
1159 });
1160 const providerTurnId = idAllocator.derive.providerTurn({
1161 driver: CODEX_DRIVER,
1162 nativeTurnId: `${threadId}:race`,
1163 });
1164 const nodeId = NodeId.make(`${threadId}:question`);
1165 const request: OrchestrationV2RuntimeRequest = {
1166 id: RuntimeRequestId.make(`${threadId}:question`),
1167 nodeId,
1168 providerTurnId,
1169 nativeRequestRef: null,
1170 kind: "user_input",
1171 status: "pending",
1172 responseCapability: { type: "live", providerSessionId },
1173 createdAt: now,
1174 resolvedAt: null,
1175 };
1176 const node: OrchestrationV2ExecutionNode = {
1177 id: nodeId,
1178 threadId,
1179 runId: null,
1180 parentNodeId: null,
1181 rootNodeId: nodeId,
1182 kind: "user_input_request",
1183 status: "waiting",
1184 countsForRun: false,
1185 providerThreadId,
1186 providerTurnId,
1187 nativeItemRef: null,
1188 runtimeRequestId: request.id,
1189 checkpointScopeId: null,
1190 startedAt: now,
1191 completedAt: null,
1192 };
1193 const item: OrchestrationV2TurnItem = {
1194 id: TurnItemId.make(`${threadId}:question`),
1195 threadId,
1196 runId: null,
1197 nodeId,
1198 providerThreadId,
1199 providerTurnId,
1200 nativeItemRef: null,
1201 parentItemId: null,
1202 ordinal: 1,
1203 status: "waiting",
1204 title: "Which option?",
1205 startedAt: now,
1206 completedAt: null,
1207 updatedAt: now,
1208 type: "user_input_request",
1209 requestId: request.id,
1210 questions: [],
1211 };
1212 const seedEvents: Array<OrchestrationV2DomainEvent> = [threadEvent];
1213 for (const payload of [
1214 { type: "runtime-request.updated" as const, payload: request },
1215 { type: "node.updated" as const, payload: node },
1216 { type: "turn-item.updated" as const, payload: item },
1217 ])
1218 seedEvents.push({
1219 id: yield* idAllocator.allocate.event({ threadId }),
1220 threadId,
1221 occurredAt: now,
1222 ...payload,
1223 });
1224 yield* eventSink.write({ events: seedEvents });
1225 const normalized = yield* Deferred.make<void>();
1226 const releaseTerminalWrite = yield* Deferred.make<void>();
1227 const gatedSink = EventSink.EventSinkV2.of({
1228 ...eventSink,
1229 write: (input) =>
1230 Deferred.succeed(normalized, undefined).pipe(
1231 Effect.andThen(Deferred.await(releaseTerminalWrite)),
1232 Effect.andThen(eventSink.write(input)),
1233 ),
1234 });
1235 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2.pipe(
1236 Effect.provide(Layer.fresh(ProviderEventIngestor.layer)),
1237 Effect.provideService(EventSink.EventSinkV2, gatedSink),
1238 );
1239 const terminal = yield* ingestor
1240 .ingestNormalized({
1241 providerSessionId,
1242 providerInstanceId: modelSelection.instanceId,
1243 threadId,
1244 event: {
1245 type: "provider_turn.updated",
1246 driver: CODEX_DRIVER,
1247 providerTurn: {
1248 id: providerTurnId,
1249 providerThreadId,
1250 nodeId,
1251 runAttemptId: null,
1252 nativeTurnRef: null,
1253 ordinal: 1,
1254 status: "completed",
1255 startedAt: now,
1256 completedAt: now,
1257 },
1258 },
1259 })
1260 .pipe(Effect.forkScoped);
1261 yield* Deferred.await(normalized);
1262 const answers = { decision: "Use the existing workspace" };
1263 const responseEvents: Array<OrchestrationV2DomainEvent> = [];
1264 for (const payload of [
1265 {
1266 type: "runtime-request.updated" as const,
1267 payload: { ...request, status: "resolved" as const, answers, resolvedAt: now },
1268 },
1269 {
1270 type: "node.updated" as const,
1271 payload: { ...node, status: "completed" as const, completedAt: now },
1272 },
1273 {
1274 type: "turn-item.updated" as const,
1275 payload: { ...item, status: "completed" as const, completedAt: now },
1276 },
1277 ])
1278 responseEvents.push({
1279 id: yield* idAllocator.allocate.event({ threadId }),
1280 threadId,
1281 occurredAt: now,
1282 ...payload,
1283 });
1284 yield* eventSink.write({ events: responseEvents });
1285 yield* Deferred.succeed(releaseTerminalWrite, undefined);
1286 const committedTerminal = yield* Fiber.join(terminal);
1287 assert.deepEqual(
1288 committedTerminal.map((stored) => stored.event.type),
1289 ["provider-turn.updated"],
1290 );
1291 const projection = yield* projections.getThreadProjection(threadId);
1292 const answered = projection.runtimeRequests.find((entry) => entry.id === request.id)!;
1293 assert.equal(answered.status, "resolved");
1294 assert.deepEqual(answered.answers, answers);
1295 assert.equal(projection.nodes.find((entry) => entry.id === nodeId)!.status, "completed");
1296 assert.equal(
1297 projection.turnItems.find((entry) => entry.id === item.id)!.status,
1298 "completed",
1299 );
1300 const history = yield* eventStore.read({ threadId }).pipe(Stream.runCollect);
1301 assert.isFalse(
1302 history.some(
1303 (stored) =>
1304 stored.event.type === "runtime-request.updated" &&
1305 stored.event.payload.status === "cancelled",
1306 ),
1307 );
1308 }),
1309 ),
1310 );
1311
1312 it.effect("persists a failed provider terminal as one expected error item", () =>
1313 Effect.gen(function* () {
1314 const now = yield* DateTime.now;
1315 const retryStartedAt = DateTime.makeUnsafe(DateTime.toEpochMillis(now) - 5_000);
1316 const eventSink = yield* EventSink.EventSinkV2;
1317 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1318 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
1319 const idAllocator = yield* IdAllocator.IdAllocatorV2;
1320 const threadEvent = yield* threadCreatedEvent(now);
1321 const providerSessionId = yield* idAllocator.allocate.providerSession({
1322 providerInstanceId: modelSelection.instanceId,
1323 threadId: threadEvent.threadId,
1324 });
1325 const providerThreadId = idAllocator.derive.providerThread({
1326 driver: CODEX_DRIVER,
1327 nativeThreadId: "native-thread-failed",
1328 });
1329 const providerTurnId = idAllocator.derive.providerTurn({
1330 driver: CODEX_DRIVER,
1331 nativeTurnId: "native-turn-failed",
1332 });
1333
1334 yield* eventSink.write({ events: [threadEvent] });
1335 const stored = yield* ingestor.ingestNormalized({
1336 providerSessionId,
1337 providerInstanceId: modelSelection.instanceId,
1338 threadId: threadEvent.threadId,
1339 event: {
1340 type: "turn.terminal",
1341 driver: CODEX_DRIVER,
1342 providerThreadId,
1343 providerTurnId,
1344 runOrdinal: 1,
1345 failureItemOrdinal: 102,
1346 status: "failed",
1347 failure: makeProviderFailure({
1348 message: "Invalid reasoning effort.",
1349 code: "invalid_request",
1350 class: "validation_error",
1351 }),
1352 retry: {
1353 attempt: 3,
1354 maxAttempts: 3,
1355 retryDelayMs: 2_000,
1356 },
1357 retryStartedAt,
1358 threadDisposition: "reusable",
1359 },
1360 });
1361
1362 const projection = yield* projectionStore.getThreadProjection(threadEvent.threadId);
1363 const errorItems = projection.visibleTurnItems.filter(
1364 (candidate) => candidate.item.type === "error",
1365 );
1366
1367 assert.equal(stored.length, 1);
1368 assert.equal(stored[0]?.event.type, "turn-item.updated");
1369 assert.equal(errorItems.length, 1);
1370 const errorItem = errorItems[0]?.item;
1371 assert.equal(errorItem?.type, "error");
1372 if (errorItem?.type !== "error") return;
1373 assert.equal(errorItem.failure.message, "Invalid reasoning effort.");
1374 assert.equal(errorItem.failure.code, "invalid_request");
1375 assert.deepEqual(errorItem.retry, {
1376 attempt: 3,
1377 maxAttempts: 3,
1378 retryDelayMs: 2_000,
1379 });
1380 const errorStartedAt = errorItem.startedAt;
1381 assert.ok(errorStartedAt);
1382 assert.equal(DateTime.toEpochMillis(errorStartedAt), DateTime.toEpochMillis(retryStartedAt));
1383 assert.equal(errorItem.providerThreadId, providerThreadId);
1384 assert.equal(errorItem.providerTurnId, providerTurnId);
1385 }),
1386 );
1387
1388 it.effect("stores tool image bytes only where a tool-output-image asset serves them", () =>
1389 Effect.gen(function* () {
1390 const now = yield* DateTime.now;
1391 const eventSink = yield* EventSink.EventSinkV2;
1392 const eventStore = yield* EventStore.EventStoreV2;
1393 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1394 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
1395 const idAllocator = yield* IdAllocator.IdAllocatorV2;
1396 const threadEvent = yield* threadCreatedEvent(now);
1397 const providerSessionId = yield* idAllocator.allocate.providerSession({
1398 providerInstanceId: modelSelection.instanceId,
1399 threadId: threadEvent.threadId,
1400 });
1401 const readBase64 = Buffer.alloc(30_000, 7).toString("base64");
1402 const screenshotBase64 = Buffer.alloc(20_000, 9).toString("base64");
1403 const toolItem = (
1404 id: string,
1405 ordinal: number,
1406 toolName: string,
1407 output: unknown,
1408 ): OrchestrationV2TurnItem => ({
1409 id: TurnItemId.make(id),
1410 threadId: threadEvent.threadId,
1411 runId: null,
1412 nodeId: null,
1413 providerThreadId: null,
1414 providerTurnId: null,
1415 nativeItemRef: null,
1416 parentItemId: null,
1417 ordinal,
1418 status: "completed",
1419 title: toolName,
1420 startedAt: now,
1421 completedAt: now,
1422 updatedAt: now,
1423 type: "dynamic_tool",
1424 toolName,
1425 input: {},
1426 output,
1427 });
1428 const read = toolItem("turn-item:read-image", 1, "Read", {
1429 type: "image",
1430 file: { base64: readBase64, type: "image/png", originalSize: 30_000 },
1431 });
1432 const screenshot = toolItem("turn-item:screenshot", 2, "mcp__t3-code__device_screenshot", {
1433 content: [
1434 {
1435 type: "image",
1436 source: { type: "base64", media_type: "image/png", data: screenshotBase64 },
1437 },
1438 ],
1439 });
1440
1441 yield* eventSink.write({ events: [threadEvent] });
1442 for (const turnItem of [read, screenshot]) {
1443 yield* ingestor.ingestNormalized({
1444 providerSessionId,
1445 providerInstanceId: modelSelection.instanceId,
1446 threadId: threadEvent.threadId,
1447 event: { type: "turn_item.updated", driver: CODEX_DRIVER, turnItem },
1448 });
1449 }
1450
1451 const storedEvents = yield* eventStore
1452 .read({ threadId: threadEvent.threadId, eventType: "turn-item.updated" })
1453 .pipe(Stream.runCollect);
1454 const storedJson = JSON.stringify(Array.from(storedEvents, (stored) => stored.event));
1455 const projectedRead = yield* projectionStore.getTurnItem({
1456 threadId: threadEvent.threadId,
1457 itemId: read.id,
1458 });
1459 const projectedScreenshot = yield* projectionStore.getTurnItem({
1460 threadId: threadEvent.threadId,
1461 itemId: screenshot.id,
1462 });
1463
1464 assert.equal(storedJson.includes(readBase64), false);
1465 assert.equal(storedJson.includes(screenshotBase64), true);
1466 assert.deepEqual(projectedRead?.type === "dynamic_tool" ? projectedRead.output : null, {
1467 type: "image",
1468 file: { type: "image/png", originalSize: 30_000, sizeBytes: 30_000 },
1469 });
1470 assert.deepEqual(
1471 toolOutputImages(
1472 projectedScreenshot?.type === "dynamic_tool" ? projectedScreenshot.output : null,
1473 ),
1474 [{ mimeType: "image/png", data: screenshotBase64 }],
1475 );
1476 }),
1477 );
1478
1479 it.effect("stores a bounded copy of a large file-change diff in both SQLite tables", () =>
1480 Effect.gen(function* () {
1481 const now = yield* DateTime.now;
1482 const eventSink = yield* EventSink.EventSinkV2;
1483 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1484 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
1485 const idAllocator = yield* IdAllocator.IdAllocatorV2;
1486 const threadEvent = yield* threadCreatedEvent(now);
1487 const providerSessionId = yield* idAllocator.allocate.providerSession({
1488 providerInstanceId: modelSelection.instanceId,
1489 threadId: threadEvent.threadId,
1490 });
1491 const largeDiff = `@@ -0,0 +1 @@\n${"+generated line\n".repeat(150_000)}`;
1492 const fileChange = (
1493 id: string,
1494 ordinal: number,
1495 diffStr: string,
1496 ): OrchestrationV2TurnItem => ({
1497 id: TurnItemId.make(id),
1498 threadId: threadEvent.threadId,
1499 runId: null,
1500 nodeId: null,
1501 providerThreadId: null,
1502 providerTurnId: null,
1503 nativeItemRef: null,
1504 parentItemId: null,
1505 ordinal,
1506 status: "completed",
1507 title: null,
1508 startedAt: now,
1509 completedAt: now,
1510 updatedAt: now,
1511 type: "file_change",
1512 fileName: "src/generated.ts",
1513 diffStr,
1514 });
1515 const large = fileChange("turn-item:large-diff", 1, largeDiff);
1516 const small = fileChange("turn-item:small-diff", 2, "@@ -1 +1 @@\n-old\n+new\n");
1517
1518 yield* eventSink.write({ events: [threadEvent] });
1519 for (const turnItem of [large, small]) {
1520 yield* ingestor.ingestNormalized({
1521 providerSessionId,
1522 providerInstanceId: modelSelection.instanceId,
1523 threadId: threadEvent.threadId,
1524 event: { type: "turn_item.updated", driver: CODEX_DRIVER, turnItem },
1525 });
1526 }
1527
1528 const sql = yield* SqlClient.SqlClient;
1529 const storedBytes = yield* sql<{ readonly events: number; readonly projection: number }>`
1530 SELECT
1531 (SELECT SUM(LENGTH(payload_json)) FROM orchestration_events
1532 WHERE stream_id = ${threadEvent.threadId} AND event_type = 'turn-item.updated') AS events,
1533 (SELECT SUM(LENGTH(payload_json)) FROM orchestration_v2_projection_turn_items
1534 WHERE thread_id = ${threadEvent.threadId}) AS projection
1535 `;
1536 const projectedLarge = yield* projectionStore.getTurnItem({
1537 threadId: threadEvent.threadId,
1538 itemId: large.id,
1539 });
1540 const projectedSmall = yield* projectionStore.getTurnItem({
1541 threadId: threadEvent.threadId,
1542 itemId: small.id,
1543 });
1544
1545 assert.isAbove(largeDiff.length, 2_000_000);
1546 assert.isBelow(storedBytes[0]!.events, 100_000);
1547 assert.isBelow(storedBytes[0]!.projection, 100_000);
1548 assert.isTrue(projectedLarge?.type === "file_change");
1549 if (projectedLarge?.type !== "file_change") return;
1550 assert.equal(projectedLarge.fileName, "src/generated.ts");
1551 assert.isBelow(projectedLarge.diffStr?.length ?? 0, 40_000);
1552 assert.isTrue(largeDiff.startsWith(projectedLarge.diffStr?.split("\n…")[0] ?? "-"));
1553 assert.deepEqual(projectedSmall, small);
1554 }),
1555 );
1556
1557 it.effect("routes provider-owned child artifacts to their child app thread", () =>
1558 Effect.gen(function* () {
1559 const now = yield* DateTime.now;
1560 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
1561 const idAllocator = yield* IdAllocator.IdAllocatorV2;
1562 const rootEvent = yield* threadCreatedEvent(now);
1563 if (rootEvent.type !== "thread.created") {
1564 throw new Error("Expected a thread.created fixture event");
1565 }
1566 const childThreadId = idAllocator.derive.threadFromProviderThread({
1567 driver: CODEX_DRIVER,
1568 nativeThreadId: "native-subagent-thread",
1569 });
1570 const childRootNodeId = NodeId.make("node:subagent-root");
1571 const childThread: OrchestrationV2AppThread = {
1572 ...rootEvent.payload,
1573 id: childThreadId,
1574 title: "inspect package",
1575 activeProviderThreadId: null,
1576 lineage: {
1577 parentThreadId: rootEvent.threadId,
1578 relationshipToParent: "subagent",
1579 rootThreadId: rootEvent.threadId,
1580 },
1581 forkedFrom: {
1582 type: "node",
1583 nodeId: NodeId.make("node:parent-subagent"),
1584 },
1585 };
1586 const providerSessionId = yield* idAllocator.allocate.providerSession({
1587 providerInstanceId: modelSelection.instanceId,
1588 threadId: rootEvent.threadId,
1589 });
1590
1591 const threadEvents = yield* ingestor.normalize({
1592 providerSessionId,
1593 providerInstanceId: modelSelection.instanceId,
1594 threadId: rootEvent.threadId,
1595 event: {
1596 type: "app_thread.created",
1597 driver: CODEX_DRIVER,
1598 appThread: childThread,
1599 },
1600 });
1601 const messageEvents = yield* ingestor.normalize({
1602 providerSessionId,
1603 providerInstanceId: modelSelection.instanceId,
1604 threadId: rootEvent.threadId,
1605 event: {
1606 type: "message.updated",
1607 driver: CODEX_DRIVER,
1608 message: {
1609 createdBy: "agent",
1610 creationSource: "provider",
1611 id: MessageId.make("message:subagent-response"),
1612 threadId: childThreadId,
1613 runId: null,
1614 nodeId: childRootNodeId,
1615 role: "assistant",
1616 text: "Subagent result",
1617 attachments: [],
1618 streaming: false,
1619 createdAt: now,
1620 updatedAt: now,
1621 },
1622 },
1623 });
1624
1625 assert.equal(threadEvents[0]?.type, "thread.created");
1626 assert.equal(threadEvents[0]?.threadId, childThreadId);
1627 assert.equal(messageEvents[0]?.type, "message.updated");
1628 assert.equal(messageEvents[0]?.threadId, childThreadId);
1629 }),
1630 );
1631
1632 it.effect("updates provider child metadata without recreating its saved thread", () =>
1633 Effect.gen(function* () {
1634 const now = yield* DateTime.now;
1635 const eventSink = yield* EventSink.EventSinkV2;
1636 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1637 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
1638 const idAllocator = yield* IdAllocator.IdAllocatorV2;
1639 const rootEvent = yield* threadCreatedEvent(now);
1640 if (rootEvent.type !== "thread.created") throw new Error("Expected thread fixture");
1641 const childThreadId = idAllocator.derive.threadFromProviderThread({
1642 driver: CODEX_DRIVER,
1643 nativeThreadId: "workflow-member-metadata",
1644 });
1645 const providerSessionId = yield* idAllocator.allocate.providerSession({
1646 providerInstanceId: modelSelection.instanceId,
1647 threadId: rootEvent.threadId,
1648 });
1649 const offered = {
1650 ...rootEvent.payload,
1651 id: childThreadId,
1652 title: "Inspect checkout",
1653 activeProviderThreadId: null,
1654 lineage: {
1655 parentThreadId: rootEvent.threadId,
1656 relationshipToParent: "subagent" as const,
1657 rootThreadId: rootEvent.threadId,
1658 },
1659 };
1660 const ingest = (appThread: OrchestrationV2AppThread) =>
1661 ingestor.ingestNormalized({
1662 providerSessionId,
1663 providerInstanceId: modelSelection.instanceId,
1664 threadId: rootEvent.threadId,
1665 event: { type: "app_thread.created", driver: CODEX_DRIVER, appThread },
1666 });
1667 yield* eventSink.write({ events: [rootEvent] });
1668 const created = yield* Effect.all([ingest(offered), ingest(offered)], {
1669 concurrency: "unbounded",
1670 });
1671 assert.deepEqual(
1672 created.flat().map((stored) => stored.event.type),
1673 ["thread.created"],
1674 );
1675 const saved = {
1676 ...offered,
1677 archivedAt: now,
1678 deletedAt: now,
1679 activeProviderThreadId: rootEvent.payload.activeProviderThreadId,
1680 branch: "user/checkout",
1681 };
1682 yield* eventSink.write({
1683 events: [
1684 {
1685 ...rootEvent,
1686 id: yield* idAllocator.allocate.event({ threadId: childThreadId }),
1687 threadId: childThreadId,
1688 type: "thread.deleted",
1689 payload: saved,
1690 },
1691 ],
1692 });
1693 for (const metadata of [
1694 { title: "Review checkout", modelSelection },
1695 { title: "Review checkout", modelSelection: { ...modelSelection, model: "gpt-6.1-sol" } },
1696 ]) {
1697 const updated = yield* ingest({ ...offered, ...metadata });
1698 assert.deepEqual(
1699 updated.map((stored) => stored.event.type),
1700 ["thread.metadata-updated"],
1701 );
1702 const thread = yield* projectionStore.getThread(childThreadId);
1703 assert.deepEqual(thread, { ...saved, ...metadata, updatedAt: thread.updatedAt });
1704 }
1705 assert.deepEqual(
1706 yield* ingest({
1707 ...offered,
1708 title: "Review checkout",
1709 modelSelection: { ...modelSelection, model: "gpt-6.1-sol" },
1710 }),
1711 [],
1712 );
1713 }),
1714 );
1715
1716 it.effect("moves a native subagent's thread to the model its provider reports later", () =>
1717 Effect.gen(function* () {
1718 const now = yield* DateTime.now;
1719 const eventSink = yield* EventSink.EventSinkV2;
1720 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1721 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
1722 const idAllocator = yield* IdAllocator.IdAllocatorV2;
1723 const rootEvent = yield* threadCreatedEvent(now);
1724 if (rootEvent.type !== "thread.created") {
1725 throw new Error("Expected a thread.created fixture event");
1726 }
1727 const childThreadId = idAllocator.derive.threadFromProviderThread({
1728 driver: CODEX_DRIVER,
1729 nativeThreadId: "native-late-model-subagent",
1730 });
1731 const providerSessionId = yield* idAllocator.allocate.providerSession({
1732 providerInstanceId: modelSelection.instanceId,
1733 threadId: rootEvent.threadId,
1734 });
1735 const ingest = (event: ProviderEventIngestor.ProviderEventIngestInput["event"]) =>
1736 ingestor.ingestNormalized({
1737 providerSessionId,
1738 providerInstanceId: modelSelection.instanceId,
1739 threadId: rootEvent.threadId,
1740 event,
1741 });
1742 yield* eventSink.write({ events: [rootEvent] });
1743 // The subagent's thread starts on the parent's model and options.
1744 yield* ingest({
1745 type: "app_thread.created",
1746 driver: CODEX_DRIVER,
1747 appThread: {
1748 ...rootEvent.payload,
1749 id: childThreadId,
1750 title: "review design",
1751 modelSelection: {
1752 ...modelSelection,
1753 options: [{ id: "reasoningEffort", value: "xhigh" }],
1754 },
1755 activeProviderThreadId: null,
1756 lineage: {
1757 parentThreadId: rootEvent.threadId,
1758 relationshipToParent: "subagent",
1759 rootThreadId: rootEvent.threadId,
1760 },
1761 },
1762 });
1763 const subagentUpdated = {
1764 type: "subagent.updated",
1765 driver: CODEX_DRIVER,
1766 subagent: {
1767 id: NodeId.make("node:late-model-subagent"),
1768 threadId: rootEvent.threadId,
1769 runId: null,
1770 parentNodeId: NodeId.make("node:root"),
1771 origin: "provider_native",
1772 createdBy: "agent",
1773 driver: CODEX_DRIVER,
1774 providerInstanceId: modelSelection.instanceId,
1775 providerThreadId: null,
1776 childThreadId,
1777 nativeTaskRef: null,
1778 prompt: "Review the design",
1779 title: "review design",
1780 model: "gpt-6.1-sol",
1781 status: "running",
1782 result: null,
1783 startedAt: now,
1784 completedAt: null,
1785 updatedAt: now,
1786 },
1787 } satisfies ProviderEventIngestor.ProviderEventIngestInput["event"];
1788
1789 const first = yield* ingest(subagentUpdated);
1790 const repeated = yield* ingest(subagentUpdated);
1791 const childThread = yield* projectionStore.getThread(childThreadId);
1792
1793 assert.deepEqual(
1794 first.map((stored) => [stored.event.type, stored.event.threadId]),
1795 [
1796 ["subagent.updated", rootEvent.threadId],
1797 ["thread.model-selection-updated", childThreadId],
1798 ],
1799 );
1800 assert.deepEqual(
1801 repeated.map((stored) => stored.event.type),
1802 ["subagent.updated"],
1803 );
1804 assert.deepEqual(childThread.modelSelection, {
1805 instanceId: modelSelection.instanceId,
1806 model: "gpt-6.1-sol",
1807 });
1808
1809 // A selection the provider reports, effort included, replaces the thread's.
1810 const reportedSelection = {
1811 instanceId: modelSelection.instanceId,
1812 model: "gpt-6.1-sol",
1813 options: [{ id: "reasoningEffort", value: "low" }],
1814 };
1815 const reported = yield* ingest({
1816 ...subagentUpdated,
1817 subagent: { ...subagentUpdated.subagent, modelSelection: reportedSelection },
1818 });
1819 assert.deepEqual(
1820 reported.map((stored) => stored.event.type),
1821 ["subagent.updated", "thread.model-selection-updated"],
1822 );
1823 assert.deepEqual(
1824 (yield* projectionStore.getThread(childThreadId)).modelSelection,
1825 reportedSelection,
1826 );
1827
1828 // A subagent recovered after a restart reports its model but no
1829 // selection; the same model keeps the effort the thread holds.
1830 const recovered = yield* ingest(subagentUpdated);
1831 assert.deepEqual(
1832 recovered.map((stored) => stored.event.type),
1833 ["subagent.updated"],
1834 );
1835 assert.deepEqual(
1836 (yield* projectionStore.getThread(childThreadId)).modelSelection,
1837 reportedSelection,
1838 );
1839 }),
1840 );
1841});