1import { assert, it, vi } from "@effect/vitest";
2import {
3 EnvironmentId,
4 EventId,
5 CommandId,
6 CheckpointId,
7 CheckpointRef,
8 CheckpointScopeId,
9 MessageId,
10 type ModelSelection,
11 type OrchestrationV2ProviderThread,
12 type OrchestrationV2ThreadShellSnapshot,
13 NodeId,
14 ProjectId,
15 ProviderDriverKind,
16 ProviderInstanceId,
17 ProviderSessionId,
18 ProviderThreadId,
19 ProviderTurnId,
20 RuntimeRequestId,
21 RunAttemptId,
22 RunId,
23 ThreadId,
24 TurnItemId,
25 type ThreadPullRequestLink,
26} from "@t3tools/contracts";
27import * as DateTime from "effect/DateTime";
28import * as Effect from "effect/Effect";
29import * as Layer from "effect/Layer";
30import * as Schema from "effect/Schema";
31import * as Tracer from "effect/Tracer";
32import * as SqlClient from "effect/sql/SqlClient";
33import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness";
34
35import * as SqlitePersistence from "../persistence/Sqlite.ts";
36import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
37import * as ProjectionStore from "./ProjectionStore.ts";
38import {
39 buildBoundedThreadProjection,
40 decodeThreadHistoryCursor,
41 selectHistoryPageFromCursor,
42 THREAD_HISTORY_PAGE_POLICY,
43} from "./threadHistoryPaging.ts";
44
45const layerTest = Layer.mergeAll(
46 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
47 SqlitePersistence.layerMemory,
48);
49const modelSelection = {
50 instanceId: ProviderInstanceId.make("codex"),
51 model: "gpt-5.4",
52} satisfies ModelSelection;
53const driver = ProviderDriverKind.make("codex");
54const providerInstanceId = modelSelection.instanceId;
55const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
56
57const traceSqlStatements = () => {
58 const statements: Array<string> = [];
59 const tracer = Tracer.make({
60 span(options) {
61 const span = new Tracer.NativeSpan(options);
62 const end = span.end.bind(span);
63 span.end = (endTime, exit) => {
64 end(endTime, exit);
65 const query = span.attributes.get("db.query.text");
66 if (typeof query === "string") statements.push(query);
67 };
68 return span;
69 },
70 });
71 return { statements, tracer };
72};
73
74const addRolledBackRecoveryCandidate = Effect.fn("addRolledBackRecoveryCandidate")(function* (
75 suffix: string,
76) {
77 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
78 const now = yield* DateTime.now;
79 const threadId = ThreadId.make(`thread:${suffix}:rolled-back`);
80 const runId = RunId.make(`run:${suffix}:rolled-back`);
81 const rootNodeId = NodeId.make(`node:${suffix}:rolled-back`);
82 const run = {
83 id: runId,
84 threadId,
85 ordinal: 1,
86 providerInstanceId,
87 modelSelection,
88 providerThreadId: null,
89 userMessageId: MessageId.make(`message:${suffix}:rolled-back`),
90 rootNodeId,
91 activeAttemptId: null,
92 status: "running" as const,
93 requestedAt: now,
94 startedAt: now,
95 completedAt: null,
96 checkpointId: null,
97 contextHandoffId: null,
98 };
99
100 yield* projectionStore.apply({
101 id: EventId.make(`event:${suffix}:thread-created`),
102 type: "thread.created",
103 threadId,
104 occurredAt: now,
105 payload: {
106 createdBy: "user",
107 creationSource: "web",
108 id: threadId,
109 projectId: ProjectId.make(`project:${suffix}`),
110 title: "Rolled-back recovery candidate",
111 providerInstanceId,
112 modelSelection,
113 runtimeMode: "full-access",
114 interactionMode: "default",
115 branch: null,
116 worktreePath: null,
117 activeProviderThreadId: null,
118 lineage: {
119 parentThreadId: null,
120 relationshipToParent: null,
121 rootThreadId: threadId,
122 },
123 forkedFrom: null,
124 createdAt: now,
125 updatedAt: now,
126 archivedAt: null,
127 settledOverride: null,
128 settledAt: null,
129 lastVisitedAt: null,
130 deletedAt: null,
131 },
132 });
133 yield* projectionStore.apply({
134 id: EventId.make(`event:${suffix}:run-created`),
135 type: "run.created",
136 threadId,
137 runId,
138 nodeId: rootNodeId,
139 driver,
140 providerInstanceId,
141 occurredAt: now,
142 payload: run,
143 });
144 yield* projectionStore.apply({
145 id: EventId.make(`event:${suffix}:item-running`),
146 type: "turn-item.updated",
147 threadId,
148 runId,
149 nodeId: rootNodeId,
150 driver,
151 occurredAt: now,
152 payload: {
153 id: TurnItemId.make(`item:${suffix}:rolled-back`),
154 threadId,
155 runId,
156 nodeId: rootNodeId,
157 providerThreadId: null,
158 providerTurnId: null,
159 nativeItemRef: null,
160 parentItemId: null,
161 ordinal: 1,
162 status: "running",
163 title: "abandoned command",
164 startedAt: now,
165 completedAt: null,
166 updatedAt: now,
167 type: "command_execution",
168 input: "sleep 60",
169 },
170 });
171 yield* projectionStore.apply({
172 id: EventId.make(`event:${suffix}:run-rolled-back`),
173 type: "run.updated",
174 threadId,
175 runId,
176 nodeId: rootNodeId,
177 driver,
178 occurredAt: now,
179 payload: { ...run, status: "rolled_back", completedAt: now },
180 });
181
182 return threadId;
183});
184
185const addOrphanedRecoveryCandidate = Effect.fn("addOrphanedRecoveryCandidate")(function* (
186 suffix: string,
187) {
188 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
189 const now = yield* DateTime.now;
190 const threadId = ThreadId.make(`thread:${suffix}:orphaned`);
191 const runId = RunId.make(`run:${suffix}:missing`);
192 const rootNodeId = NodeId.make(`node:${suffix}:orphaned`);
193
194 yield* projectionStore.apply({
195 id: EventId.make(`event:${suffix}:thread-created`),
196 type: "thread.created",
197 threadId,
198 occurredAt: now,
199 payload: {
200 createdBy: "user",
201 creationSource: "web",
202 id: threadId,
203 projectId: ProjectId.make(`project:${suffix}`),
204 title: "Orphaned recovery candidate",
205 providerInstanceId,
206 modelSelection,
207 runtimeMode: "full-access",
208 interactionMode: "default",
209 branch: null,
210 worktreePath: null,
211 activeProviderThreadId: null,
212 lineage: {
213 parentThreadId: null,
214 relationshipToParent: null,
215 rootThreadId: threadId,
216 },
217 forkedFrom: null,
218 createdAt: now,
219 updatedAt: now,
220 archivedAt: null,
221 settledOverride: null,
222 settledAt: null,
223 lastVisitedAt: null,
224 deletedAt: null,
225 },
226 });
227 yield* projectionStore.apply({
228 id: EventId.make(`event:${suffix}:item-running`),
229 type: "turn-item.updated",
230 threadId,
231 runId,
232 nodeId: rootNodeId,
233 driver,
234 occurredAt: now,
235 payload: {
236 id: TurnItemId.make(`item:${suffix}:orphaned`),
237 threadId,
238 runId,
239 nodeId: rootNodeId,
240 providerThreadId: null,
241 providerTurnId: null,
242 nativeItemRef: null,
243 parentItemId: null,
244 ordinal: 1,
245 status: "running",
246 title: "orphaned command",
247 startedAt: now,
248 completedAt: null,
249 updatedAt: now,
250 type: "command_execution",
251 input: "sleep 60",
252 },
253 });
254
255 return threadId;
256});
257
258it("includes imported runless history when selecting fork context through a run", () => {
259 const firstRunId = RunId.make("run:projection-imported-fork:1");
260 const secondRunId = RunId.make("run:projection-imported-fork:2");
261 const runOrdinalById = new Map([
262 [firstRunId, 1],
263 [secondRunId, 2],
264 ]);
265
266 assert.isTrue(
267 ProjectionStore.isTurnItemAtOrBeforeRun({
268 historyOrigin: "v1_import",
269 itemRunId: null,
270 runOrdinalById,
271 sourceRunOrdinal: 1,
272 }),
273 );
274 assert.isFalse(
275 ProjectionStore.isTurnItemAtOrBeforeRun({
276 historyOrigin: undefined,
277 itemRunId: null,
278 runOrdinalById,
279 sourceRunOrdinal: 1,
280 }),
281 );
282 assert.isTrue(
283 ProjectionStore.isTurnItemAtOrBeforeRun({
284 historyOrigin: "v1_import",
285 itemRunId: firstRunId,
286 runOrdinalById,
287 sourceRunOrdinal: 1,
288 }),
289 );
290 assert.isFalse(
291 ProjectionStore.isTurnItemAtOrBeforeRun({
292 historyOrigin: "v1_import",
293 itemRunId: secondRunId,
294 runOrdinalById,
295 sourceRunOrdinal: 1,
296 }),
297 );
298});
299
300// Recovery records the work, then a run.updated snapshot taken before it lands.
301const restartCancelledWorkSurvivesStaleRunUpdate = Effect.gen(function* () {
302 const store = yield* ProjectionStore.ProjectionStoreV2;
303 const threadId = yield* addRolledBackRecoveryCandidate("restart-cancelled-work-stale-update");
304 const run = (yield* store.getThreadProjection(threadId)).runs[0]!;
305 const now = yield* DateTime.now;
306 const work = [{ kind: "subagent" as const, label: "Background subagent test" }];
307 yield* store.apply({
308 id: EventId.make("event:restart-cancelled-work-stale-update:recorded"),
309 type: "run.background-work-cancelled",
310 threadId,
311 runId: run.id,
312 providerInstanceId,
313 occurredAt: now,
314 payload: { runId: run.id, restartCancelledBackgroundWork: work },
315 });
316 yield* store.apply({
317 id: EventId.make("event:restart-cancelled-work-stale-update:completed"),
318 type: "run.updated",
319 threadId,
320 runId: run.id,
321 providerInstanceId,
322 occurredAt: now,
323 payload: { ...run, status: "completed", completedAt: now },
324 });
325 const updated = (yield* store.getThreadProjection(threadId)).runs[0];
326 assert.equal(updated?.status, "completed");
327 assert.deepEqual(updated?.restartCancelledBackgroundWork, work);
328});
329
330const threadWorktreePathsReadActiveThreads = Effect.gen(function* () {
331 const store = yield* ProjectionStore.ProjectionStoreV2;
332 const now = yield* DateTime.now;
333 const create = (id: string, input: { worktreePath: string | null; archived?: boolean }) => {
334 const threadId = ThreadId.make(id);
335 const thread = {
336 createdBy: "user" as const,
337 creationSource: "web" as const,
338 id: threadId,
339 projectId: ProjectId.make("project:worktree-paths"),
340 title: id,
341 providerInstanceId,
342 modelSelection,
343 runtimeMode: "full-access" as const,
344 interactionMode: "default" as const,
345 branch: input.worktreePath === null ? null : id,
346 worktreePath: input.worktreePath,
347 activeProviderThreadId: null,
348 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
349 forkedFrom: null,
350 createdAt: now,
351 updatedAt: now,
352 archivedAt: input.archived === true ? now : null,
353 settledOverride: null,
354 settledAt: null,
355 lastVisitedAt: null,
356 deletedAt: null,
357 };
358 return store
359 .apply({
360 id: EventId.make(`event:${id}:created`),
361 type: "thread.created",
362 threadId,
363 occurredAt: now,
364 payload: thread,
365 })
366 .pipe(Effect.as(thread));
367 };
368 yield* create("thread:active", { worktreePath: "/worktrees/active" });
369 yield* create("thread:archived", { worktreePath: "/worktrees/archived", archived: true });
370 yield* create("thread:local", { worktreePath: null });
371 const deleted = yield* create("thread:deleted", { worktreePath: "/worktrees/deleted" });
372 yield* store.apply({
373 id: EventId.make("event:thread:deleted:deleted"),
374 type: "thread.deleted",
375 threadId: deleted.id,
376 occurredAt: now,
377 payload: { ...deleted, deletedAt: now },
378 });
379
380 assert.deepEqual(yield* store.getThreadWorktreePaths(), [
381 { threadId: ThreadId.make("thread:active"), worktreePath: "/worktrees/active" },
382 ]);
383});
384
385const commandReadsSkipSettledLeafNodes = Effect.gen(function* () {
386 const store = yield* ProjectionStore.ProjectionStoreV2;
387 const now = yield* DateTime.now;
388 const threadId = yield* addRolledBackRecoveryCandidate("settled-leaf-nodes");
389 const settledRun = (yield* store.getThreadProjection(threadId)).runs[0]!;
390 const activeRun = {
391 ...settledRun,
392 id: RunId.make("run:settled-leaf-nodes:active"),
393 ordinal: 2,
394 userMessageId: MessageId.make("message:settled-leaf-nodes:active"),
395 rootNodeId: NodeId.make("node:settled-leaf-nodes:active-root"),
396 status: "running" as const,
397 completedAt: null,
398 };
399 yield* store.apply({
400 id: EventId.make("event:settled-leaf-nodes:active-run"),
401 type: "run.created",
402 threadId,
403 runId: activeRun.id,
404 nodeId: activeRun.rootNodeId,
405 driver,
406 providerInstanceId,
407 occurredAt: now,
408 payload: activeRun,
409 });
410 const nodes = [
411 ["settled-root", settledRun, "root_turn", "completed"],
412 ["settled-tool", settledRun, "tool_call", "completed"],
413 ["settled-message", settledRun, "assistant_message", "failed"],
414 ["settled-reasoning", settledRun, "reasoning", "cancelled"],
415 ["settled-subagent", settledRun, "subagent", "completed"],
416 ["settled-run-live-tool", settledRun, "tool_call", "running"],
417 ["active-run-done-tool", activeRun, "tool_call", "completed"],
418 ] as const;
419 for (const [name, run, kind, status] of nodes) {
420 const id = NodeId.make(`node:settled-leaf-nodes:${name}`);
421 yield* store.apply({
422 id: EventId.make(`event:settled-leaf-nodes:${name}`),
423 type: "node.updated",
424 threadId,
425 runId: run.id,
426 nodeId: id,
427 driver,
428 occurredAt: now,
429 payload: {
430 id,
431 threadId,
432 runId: run.id,
433 rootNodeId: kind === "root_turn" ? id : run.rootNodeId!,
434 parentNodeId: kind === "root_turn" ? null : run.rootNodeId,
435 kind,
436 status,
437 countsForRun: kind === "root_turn",
438 providerThreadId: null,
439 providerTurnId: null,
440 nativeItemRef: null,
441 runtimeRequestId: null,
442 checkpointScopeId: null,
443 startedAt: now,
444 completedAt: status === "running" ? null : now,
445 },
446 });
447 }
448 const nodeNames = (records: { readonly nodes: ReadonlyArray<{ readonly id: string }> }) =>
449 records.nodes
450 .map((node) => node.id.replace("node:settled-leaf-nodes:", ""))
451 .filter((name) => nodes.some(([candidate]) => candidate === name))
452 .toSorted();
453
454 assert.deepEqual(
455 nodeNames(yield* store.getThreadRecords(threadId, ["nodes"], { omitSettledLeafNodes: true })),
456 ["active-run-done-tool", "settled-root", "settled-run-live-tool", "settled-subagent"],
457 );
458 assert.deepEqual(
459 nodeNames(yield* store.getThreadRecords(threadId, ["nodes"])),
460 nodes.map(([name]) => name).toSorted(),
461 );
462});
463
464it.effect("memory command reads skip finished leaf nodes of finished runs", () =>
465 commandReadsSkipSettledLeafNodes.pipe(Effect.provide(ProjectionStore.layerMemory)),
466);
467
468it.effect("memory worktree path lookup reads only active threads", () =>
469 threadWorktreePathsReadActiveThreads.pipe(Effect.provide(ProjectionStore.layerMemory)),
470);
471
472it.effect("memory projection keeps restart-cancelled work through a stale run.updated", () =>
473 restartCancelledWorkSurvivesStaleRunUpdate.pipe(Effect.provide(ProjectionStore.layerMemory)),
474);
475
476it.effect("memory recovery selection ignores unfinished items from rolled-back runs", () =>
477 Effect.gen(function* () {
478 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
479 const threadId = yield* addRolledBackRecoveryCandidate("memory-recovery-candidates");
480
481 assert.notInclude(yield* projectionStore.getRecoveryThreadIds("runtime"), threadId);
482 }).pipe(Effect.provide(ProjectionStore.layerMemory)),
483);
484
485it.effect("memory recovery selection includes unfinished items from missing runs", () =>
486 Effect.gen(function* () {
487 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
488 const threadId = yield* addOrphanedRecoveryCandidate("memory-recovery-candidates");
489
490 assert.include(yield* projectionStore.getRecoveryThreadIds("runtime"), threadId);
491 }).pipe(Effect.provide(ProjectionStore.layerMemory)),
492);
493
494const runlessNativeSubagentNeedsRecovery = Effect.gen(function* () {
495 const store = yield* ProjectionStore.ProjectionStoreV2;
496 const threadId = yield* addRolledBackRecoveryCandidate("runless-native-recovery");
497 const now = yield* DateTime.now;
498 const id = NodeId.make("node:runless-native-recovery");
499 assert.notInclude(yield* store.getRecoveryThreadIds("runtime"), threadId);
500 for (const status of ["pending", "running", "waiting", "completed"] as const) {
501 yield* store.apply({
502 id: EventId.make(`event:runless-native-recovery:${status}`),
503 type: "subagent.updated",
504 threadId,
505 nodeId: id,
506 driver,
507 providerInstanceId,
508 occurredAt: now,
509 payload: {
510 id,
511 threadId,
512 runId: null,
513 parentNodeId: NodeId.make("node:runless-native-parent"),
514 origin: "provider_native",
515 createdBy: "agent",
516 driver,
517 providerInstanceId,
518 providerThreadId: null,
519 childThreadId: null,
520 nativeTaskRef: { driver, nativeId: "runless-native", strength: "strong" },
521 prompt: "Do the work",
522 title: "Workflow member",
523 model: null,
524 status,
525 result: null,
526 startedAt: now,
527 completedAt: status === "completed" ? now : null,
528 updatedAt: now,
529 },
530 });
531 assert.equal(
532 (yield* store.getRecoveryThreadIds("runtime")).includes(threadId),
533 status !== "completed",
534 );
535 }
536});
537
538it.effect("memory recovery selects unfinished runless native subagents", () =>
539 runlessNativeSubagentNeedsRecovery.pipe(Effect.provide(ProjectionStore.layerMemory)),
540);
541
542// Project A holds an active thread, an archived thread and a fork of a run in
543// an archived project B thread, so a project-scoped snapshot still has to load
544// a fork source that is outside the project and archived.
545const assertProjectScopedShells = Effect.fn("assertProjectScopedShells")(function* (
546 suffix: string,
547) {
548 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
549 const now = yield* DateTime.now;
550 const projectId = ProjectId.make(`project:${suffix}:a`);
551 const otherProjectId = ProjectId.make(`project:${suffix}:b`);
552 const activeThreadId = ThreadId.make(`thread:${suffix}:a-active`);
553 const archivedThreadId = ThreadId.make(`thread:${suffix}:a-archived`);
554 const forkThreadId = ThreadId.make(`thread:${suffix}:a-fork`);
555 const sourceThreadId = ThreadId.make(`thread:${suffix}:b-source`);
556 const otherThreadId = ThreadId.make(`thread:${suffix}:b-other`);
557 const sourceRunId = RunId.make(`run:${suffix}:b-source`);
558 const sourceNodeId = NodeId.make(`node:${suffix}:b-source`);
559 const createThread = (input: {
560 readonly threadId: ThreadId;
561 readonly projectId: ProjectId;
562 readonly archived?: boolean;
563 readonly forkedFromRunOf?: ThreadId;
564 }) =>
565 projectionStore.apply({
566 id: EventId.make(`event:${input.threadId}:created`),
567 type: "thread.created",
568 threadId: input.threadId,
569 occurredAt: now,
570 payload: {
571 createdBy: "user",
572 creationSource: "web",
573 id: input.threadId,
574 projectId: input.projectId,
575 title: input.threadId,
576 providerInstanceId,
577 modelSelection,
578 runtimeMode: "full-access",
579 interactionMode: "default",
580 branch: null,
581 worktreePath: null,
582 activeProviderThreadId: null,
583 lineage:
584 input.forkedFromRunOf === undefined
585 ? { parentThreadId: null, relationshipToParent: null, rootThreadId: input.threadId }
586 : {
587 parentThreadId: input.forkedFromRunOf,
588 relationshipToParent: "fork",
589 rootThreadId: input.forkedFromRunOf,
590 },
591 forkedFrom:
592 input.forkedFromRunOf === undefined
593 ? null
594 : { type: "run", threadId: input.forkedFromRunOf, runId: sourceRunId },
595 createdAt: now,
596 updatedAt: now,
597 archivedAt: input.archived === true ? now : null,
598 settledOverride: null,
599 settledAt: null,
600 lastVisitedAt: null,
601 deletedAt: null,
602 },
603 });
604
605 yield* createThread({ threadId: sourceThreadId, projectId: otherProjectId, archived: true });
606 yield* createThread({ threadId: otherThreadId, projectId: otherProjectId });
607 yield* createThread({ threadId: activeThreadId, projectId });
608 yield* createThread({ threadId: archivedThreadId, projectId, archived: true });
609 yield* createThread({ threadId: forkThreadId, projectId, forkedFromRunOf: sourceThreadId });
610 yield* projectionStore.apply({
611 id: EventId.make(`event:${suffix}:source-run`),
612 type: "run.created",
613 threadId: sourceThreadId,
614 runId: sourceRunId,
615 nodeId: sourceNodeId,
616 driver,
617 occurredAt: now,
618 payload: {
619 id: sourceRunId,
620 threadId: sourceThreadId,
621 ordinal: 1,
622 providerInstanceId,
623 modelSelection,
624 providerThreadId: null,
625 userMessageId: MessageId.make(`message:${suffix}:source-user`),
626 rootNodeId: sourceNodeId,
627 activeAttemptId: null,
628 status: "completed",
629 requestedAt: now,
630 startedAt: now,
631 completedAt: now,
632 checkpointId: null,
633 contextHandoffId: null,
634 },
635 });
636 yield* projectionStore.apply({
637 id: EventId.make(`event:${suffix}:source-item`),
638 type: "turn-item.updated",
639 threadId: sourceThreadId,
640 runId: sourceRunId,
641 occurredAt: now,
642 payload: {
643 id: TurnItemId.make(`turn-item:${suffix}:source`),
644 threadId: sourceThreadId,
645 runId: sourceRunId,
646 nodeId: null,
647 providerThreadId: null,
648 providerTurnId: null,
649 nativeItemRef: null,
650 parentItemId: null,
651 ordinal: 1,
652 status: "completed",
653 title: null,
654 startedAt: now,
655 completedAt: now,
656 updatedAt: now,
657 type: "assistant_message",
658 messageId: MessageId.make(`message:${suffix}:source-assistant`),
659 text: "Source answer",
660 streaming: false,
661 },
662 });
663
664 const unscoped = yield* projectionStore.getShellSnapshot();
665 const inProject = (thread: { readonly projectId: ProjectId }) => thread.projectId === projectId;
666 const active = yield* projectionStore.getShellSnapshot({ projectId, location: "active" });
667 assert.deepEqual(active.threads, unscoped.threads.filter(inProject));
668 assert.deepEqual(active.archivedThreads, []);
669 assert.sameMembers(
670 active.threads.map((thread) => thread.id),
671 [activeThreadId, forkThreadId],
672 );
673 const fork = active.threads.find((thread) => thread.id === forkThreadId);
674 assert.isAbove(fork?.visibleItemCount ?? 0, 0);
675
676 const all = yield* projectionStore.getShellSnapshot({ projectId });
677 assert.deepEqual(all.threads, active.threads);
678 assert.deepEqual(all.archivedThreads, unscoped.archivedThreads.filter(inProject));
679 assert.sameMembers(
680 all.archivedThreads.map((thread) => thread.id),
681 [archivedThreadId],
682 );
683 return { projectId, activeThreadId, forkThreadId, otherThreadId };
684});
685
686it.effect("memory shell snapshots scope to one project", () =>
687 assertProjectScopedShells("memory-project-shell").pipe(
688 Effect.asVoid,
689 Effect.provide(ProjectionStore.layerMemory),
690 ),
691);
692
693it.layer(layerTest)("ProjectionStoreV2", (it) => {
694 it.effect(
695 "SQL recovery selects unfinished runless native subagents",
696 () => runlessNativeSubagentNeedsRecovery,
697 );
698 it.effect(
699 "worktree path lookup reads only active threads",
700 () => threadWorktreePathsReadActiveThreads,
701 );
702
703 it.effect(
704 "keeps restart-cancelled work through a stale run.updated",
705 () => restartCancelledWorkSurvivesStaleRunUpdate,
706 );
707 it.effect("records restart-cancelled work without regressing a run that completed since", () =>
708 Effect.gen(function* () {
709 const store = yield* ProjectionStore.ProjectionStoreV2;
710 const threadId = yield* addRolledBackRecoveryCandidate("restart-cancelled-work");
711 const run = (yield* store.getThreadProjection(threadId)).runs[0]!;
712 const now = yield* DateTime.now;
713 // Recovery read the run as waiting; its checkpoint completed it before the commit.
714 yield* store.apply({
715 id: EventId.make("event:restart-cancelled-work:completed"),
716 type: "run.updated",
717 threadId,
718 runId: run.id,
719 providerInstanceId,
720 occurredAt: now,
721 payload: { ...run, status: "completed", completedAt: now },
722 });
723 const work = [{ kind: "subagent" as const, label: "Background subagent test" }];
724 yield* store.apply({
725 id: EventId.make("event:restart-cancelled-work:recorded"),
726 type: "run.background-work-cancelled",
727 threadId,
728 runId: run.id,
729 providerInstanceId,
730 occurredAt: now,
731 payload: { runId: run.id, restartCancelledBackgroundWork: work },
732 });
733 const recorded = (yield* store.getThreadProjection(threadId)).runs[0];
734 assert.equal(recorded?.status, "completed");
735 assert.deepEqual(recorded?.restartCancelledBackgroundWork, work);
736 const [recovery] = (yield* store.getThreadRecords(threadId, ["runs"], {
737 runIds: [run.id],
738 })).runs;
739 assert.equal(recovery?.status, "completed");
740 assert.deepEqual(recovery?.restartCancelledBackgroundWork, work);
741 }),
742 );
743 it.effect(
744 "command reads skip finished leaf nodes of finished runs",
745 () => commandReadsSkipSettledLeafNodes,
746 );
747 it.effect("limits turn-start history to the requested runs, including an empty selection", () =>
748 Effect.gen(function* () {
749 const store = yield* ProjectionStore.ProjectionStoreV2;
750 const threadId = yield* addRolledBackRecoveryCandidate("selected-turn-start-history");
751 const runId = (yield* store.getThreadProjection(threadId)).runs[0]!.id;
752 const history = yield* store.getTurnStartHistory(threadId);
753 assert.isNotEmpty(history);
754 assert.deepEqual(yield* store.getTurnStartHistory(threadId, [runId]), history);
755 assert.deepEqual(yield* store.getTurnStartHistory(threadId, []), []);
756 assert.deepEqual(yield* store.getTurnStartHistory(threadId, [RunId.make("run:other")]), []);
757 }),
758 );
759 it.effect("preserves stored provider usage when a terminal update omits it", () =>
760 Effect.gen(function* () {
761 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
762 const now = yield* DateTime.now;
763 const threadId = ThreadId.make("thread:provider-usage-reload");
764 const providerThreadId = ProviderThreadId.make("provider-thread:provider-usage-reload");
765 const providerTurnId = ProviderTurnId.make("provider-turn:provider-usage-reload");
766 const nodeId = NodeId.make("node:provider-usage-reload");
767 const initialUsage = {
768 usedTokens: 12_000,
769 maxTokens: 200_000,
770 inputTokens: 11_000,
771 outputTokens: 1_000,
772 updatedAt: "2026-08-29T12:00:00.000Z",
773 } as const;
774 const replacementUsage = {
775 usedTokens: 18_000,
776 maxTokens: 200_000,
777 inputTokens: 16_000,
778 outputTokens: 2_000,
779 updatedAt: "2026-08-29T12:00:01.000Z",
780 } as const;
781
782 yield* projectionStore.apply({
783 id: EventId.make("event:provider-usage-reload:thread"),
784 type: "thread.created",
785 threadId,
786 occurredAt: now,
787 payload: {
788 createdBy: "user",
789 creationSource: "web",
790 id: threadId,
791 projectId: ProjectId.make("project:provider-usage-reload"),
792 title: "Provider usage reload",
793 providerInstanceId,
794 modelSelection,
795 runtimeMode: "full-access",
796 interactionMode: "default",
797 branch: null,
798 worktreePath: null,
799 activeProviderThreadId: null,
800 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
801 forkedFrom: null,
802 createdAt: now,
803 updatedAt: now,
804 archivedAt: null,
805 settledOverride: null,
806 settledAt: null,
807 lastVisitedAt: null,
808 deletedAt: null,
809 },
810 });
811
812 const providerTurn = {
813 id: providerTurnId,
814 providerThreadId,
815 nodeId,
816 runAttemptId: null,
817 nativeTurnRef: null,
818 ordinal: 1,
819 status: "running" as const,
820 startedAt: now,
821 completedAt: null,
822 };
823 yield* projectionStore.apply({
824 id: EventId.make("event:provider-usage-reload:running"),
825 type: "provider-turn.updated",
826 threadId,
827 nodeId,
828 driver,
829 occurredAt: now,
830 payload: { ...providerTurn, tokenUsage: initialUsage },
831 });
832 yield* projectionStore.apply({
833 id: EventId.make("event:provider-usage-reload:completed"),
834 type: "provider-turn.updated",
835 threadId,
836 nodeId,
837 driver,
838 occurredAt: now,
839 payload: { ...providerTurn, status: "completed", completedAt: now },
840 });
841
842 const reloaded = yield* projectionStore.getThreadProjection(threadId);
843 assert.deepEqual(reloaded.providerTurns[0]?.tokenUsage, initialUsage);
844 assert.strictEqual(reloaded.providerTurns[0]?.status, "completed");
845
846 yield* projectionStore.apply({
847 id: EventId.make("event:provider-usage-reload:replacement"),
848 type: "provider-turn.updated",
849 threadId,
850 nodeId,
851 driver,
852 occurredAt: now,
853 payload: {
854 ...providerTurn,
855 status: "completed",
856 completedAt: now,
857 tokenUsage: replacementUsage,
858 },
859 });
860
861 const replaced = yield* projectionStore.getThreadProjection(threadId);
862 assert.deepEqual(replaced.providerTurns[0]?.tokenUsage, replacementUsage);
863 }),
864 );
865
866 it.effect("pages complete user turns through SQL regardless of tool count or payload size", () =>
867 Effect.gen(function* () {
868 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
869 const sql = yield* SqlClient.SqlClient;
870 const now = yield* DateTime.now;
871 const nowIso = DateTime.formatIso(now);
872 const threadId = ThreadId.make("thread:user-turn-pages");
873 yield* projectionStore.apply({
874 id: EventId.make("event:user-turn-pages:thread"),
875 type: "thread.created",
876 threadId,
877 occurredAt: now,
878 payload: {
879 createdBy: "user",
880 creationSource: "web",
881 id: threadId,
882 projectId: ProjectId.make("project:user-turn-pages"),
883 title: "Bounded SQL history",
884 providerInstanceId,
885 modelSelection,
886 runtimeMode: "full-access",
887 interactionMode: "default",
888 branch: null,
889 worktreePath: null,
890 activeProviderThreadId: null,
891 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
892 forkedFrom: null,
893 createdAt: now,
894 updatedAt: now,
895 archivedAt: null,
896 settledOverride: null,
897 settledAt: null,
898 lastVisitedAt: null,
899 deletedAt: null,
900 },
901 });
902
903 const allIds: string[] = [];
904 let lastToolPayload = "";
905 for (let turn = 1; turn <= 45; turn += 1) {
906 const rows = Array.from({ length: 102 }, (_, offset) => {
907 const ordinal = (turn - 1) * 102 + offset + 1;
908 const id = `item:user-turn-pages:${ordinal}`;
909 allIds.push(id);
910 const base = {
911 id,
912 threadId,
913 runId: null,
914 nodeId: null,
915 providerThreadId: null,
916 providerTurnId: null,
917 nativeItemRef: null,
918 parentItemId: null,
919 ordinal,
920 status: "completed",
921 title: null,
922 startedAt: nowIso,
923 completedAt: nowIso,
924 updatedAt: nowIso,
925 };
926 const item =
927 offset < 2
928 ? {
929 ...base,
930 type: "user_message",
931 createdBy: "user",
932 creationSource: "web",
933 messageId: `message:${id}`,
934 inputIntent: offset === 0 ? "turn_start" : "steer",
935 text: `Turn ${turn}`,
936 attachments: [],
937 }
938 : {
939 ...base,
940 type: "command_execution",
941 input: "command",
942 output: "x".repeat(2048),
943 exitCode: 0,
944 };
945 const payloadJson = encodeUnknownJsonString(item);
946 if (offset >= 2) lastToolPayload = payloadJson;
947 return {
948 turn_item_id: id,
949 thread_id: threadId,
950 run_id: null,
951 node_id: null,
952 provider_thread_id: null,
953 provider_turn_id: null,
954 parent_item_id: null,
955 ordinal,
956 type: item.type,
957 status: "completed",
958 updated_at: nowIso,
959 payload_json: payloadJson,
960 };
961 });
962 yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`;
963 }
964 const { statements, tracer } = traceSqlStatements();
965 const initial = yield* Effect.acquireUseRelease(
966 Effect.sync(() => vi.spyOn(JSON, "parse")),
967 (parse) =>
968 projectionStore
969 .getThreadSnapshotWindow(threadId, { rowLimit: 77, userTurnLimit: 10 })
970 .pipe(
971 Effect.withTracer(tracer),
972 Effect.tap(() =>
973 Effect.sync(() => {
974 // Tool outputs must not be allocated again just to collect cohort IDs.
975 assert.lengthOf(
976 parse.mock.calls.filter(([input]) => input === lastToolPayload),
977 1,
978 );
979 }),
980 ),
981 ),
982 (parse) => Effect.sync(() => parse.mockRestore()),
983 );
984 // Large threads blocked the server for over a second here. The boundary
985 // must come from the user-message index rather than a scan of every tool
986 // row, and payloads must not go through a sort after they are fetched.
987 const windowStatement = statements.find((statement) => statement.includes("turn_anchors"));
988 assert.isDefined(windowStatement);
989 const windowPlan = yield* sql.unsafe<{ readonly parent: number; readonly detail: string }>(
990 `EXPLAIN QUERY PLAN ${windowStatement}`,
991 );
992 assert.match(
993 windowPlan.map((row) => row.detail).join("\n"),
994 /SEARCH item USING INDEX orchestration_v2_projection_turn_items_user_message_idx \(thread_id=\? AND ordinal<\?\)/,
995 );
996 const topLevel = windowPlan.filter((row) => row.parent === 0).map((row) => row.detail);
997 assert.include(topLevel, "SEARCH item USING INTEGER PRIMARY KEY (rowid=?)");
998 assert.notInclude(topLevel, "USE TEMP B-TREE FOR ORDER BY");
999 const nodeStatement = statements.find((statement) =>
1000 statement.includes("WITH RECURSIVE retained"),
1001 );
1002 assert.isDefined(nodeStatement);
1003 const nodePlan = yield* sql.unsafe<{ readonly detail: string }>(
1004 `EXPLAIN QUERY PLAN ${nodeStatement}`,
1005 );
1006 assert.include(
1007 nodePlan.map((row) => row.detail),
1008 "SEARCH orchestration_v2_projection_nodes USING INDEX orchestration_v2_projection_nodes_live_idx (thread_id=?)",
1009 );
1010 // Only the selected turn cohort and two lookahead anchors are decoded.
1011 assert.lengthOf(initial.projection.turnItems, 12 * 102);
1012 const bounded = buildBoundedThreadProjection({
1013 projection: initial.projection,
1014 snapshotSequence: 0,
1015 });
1016 assert.lengthOf(bounded.projection.visibleTurnItems, 10 * 102);
1017 assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, allIds[35 * 102]);
1018 const loaded = bounded.projection.visibleTurnItems.map((row) => String(row.sourceItemId));
1019 let cursor = bounded.historyCursor;
1020 for (const turns of [20, 15]) {
1021 assert.isNotNull(cursor);
1022 const anchor = decodeThreadHistoryCursor(cursor!);
1023 const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1024 rowLimit: 77,
1025 userTurnLimit: 20,
1026 anchorItemId: TurnItemId.make(anchor.si),
1027 anchorThreadId: ThreadId.make(anchor.st),
1028 });
1029 const page = selectHistoryPageFromCursor({
1030 items: snapshot.projection.visibleTurnItems,
1031 cursor: cursor!,
1032 snapshotSequence: 0,
1033 });
1034 assert.lengthOf(page.items, turns * 102);
1035 loaded.unshift(...page.items.map((row) => String(row.sourceItemId)));
1036 cursor = page.nextCursor;
1037 }
1038 assert.isNull(cursor);
1039 assert.deepEqual(loaded, allIds);
1040 }),
1041 );
1042
1043 it.effect("caps a SQL window by runs that a notification started", () =>
1044 Effect.gen(function* () {
1045 // A thread woken by PR watches: 12 user turns, then 400 runs that a
1046 // notification started, with a user turn every 50 runs. The user-turn
1047 // window reached back to the 12th-newest user turn and read almost every row.
1048 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1049 const sql = yield* SqlClient.SqlClient;
1050 const now = yield* DateTime.now;
1051 const nowIso = DateTime.formatIso(now);
1052 const threadId = ThreadId.make("thread:notification-turn-pages");
1053 yield* projectionStore.apply({
1054 id: EventId.make("event:notification-turn-pages:thread"),
1055 type: "thread.created",
1056 threadId,
1057 occurredAt: now,
1058 payload: {
1059 createdBy: "user",
1060 creationSource: "web",
1061 id: threadId,
1062 projectId: ProjectId.make("project:notification-turn-pages"),
1063 title: "Woken by notifications",
1064 providerInstanceId,
1065 modelSelection,
1066 runtimeMode: "full-access",
1067 interactionMode: "default",
1068 branch: null,
1069 worktreePath: null,
1070 activeProviderThreadId: null,
1071 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
1072 forkedFrom: null,
1073 createdAt: now,
1074 updatedAt: now,
1075 archivedAt: null,
1076 settledOverride: null,
1077 settledAt: null,
1078 lastVisitedAt: null,
1079 deletedAt: null,
1080 },
1081 });
1082
1083 const startedRuns = 412;
1084 const allIds: string[] = [];
1085 const startIds: string[] = [];
1086 let ordinal = 0;
1087 const itemRow = (runId: string, item: Record<string, unknown>) => {
1088 ordinal += 1;
1089 const id = `turn-item:notification-turn-pages:${ordinal}`;
1090 allIds.push(id);
1091 return {
1092 turn_item_id: id,
1093 thread_id: threadId,
1094 run_id: runId,
1095 node_id: null,
1096 provider_thread_id: null,
1097 provider_turn_id: null,
1098 parent_item_id: null,
1099 ordinal,
1100 type: String(item.type),
1101 status: "completed",
1102 updated_at: nowIso,
1103 payload_json: encodeUnknownJsonString({
1104 id,
1105 threadId,
1106 runId,
1107 nodeId: null,
1108 providerThreadId: null,
1109 providerTurnId: null,
1110 nativeItemRef: null,
1111 parentItemId: null,
1112 ordinal,
1113 status: "completed",
1114 title: null,
1115 startedAt: nowIso,
1116 completedAt: nowIso,
1117 updatedAt: nowIso,
1118 ...item,
1119 }),
1120 };
1121 };
1122 const command = { type: "command_execution", input: "command", output: "x", exitCode: 0 };
1123 const notification = (summary: string) => ({
1124 type: "notification",
1125 source: { kind: "monitor" },
1126 outcome: "completed",
1127 summary,
1128 });
1129 // The newest run is a wake queued behind the last one. It has no items yet,
1130 // like a run cancelled before it started.
1131 for (let run = 1; run <= startedRuns + 1; run += 1) {
1132 const runId = `run:notification-turn-pages:${run}`;
1133 const status = run > startedRuns ? "queued" : "completed";
1134 yield* sql`
1135 INSERT INTO orchestration_v2_projection_runs (
1136 run_id, thread_id, ordinal, provider, provider_thread_id, status,
1137 requested_at, completed_at, payload_json
1138 ) VALUES (
1139 ${runId}, ${threadId}, ${run}, 'codex', NULL, ${status}, ${nowIso}, ${nowIso},
1140 ${encodeUnknownJsonString({
1141 id: runId,
1142 threadId,
1143 ordinal: run,
1144 providerInstanceId,
1145 modelSelection,
1146 providerThreadId: null,
1147 userMessageId: `message:notification-turn-pages:${run}`,
1148 rootNodeId: `node:notification-turn-pages:${run}`,
1149 activeAttemptId: null,
1150 status,
1151 requestedAt: nowIso,
1152 startedAt: nowIso,
1153 completedAt: nowIso,
1154 checkpointId: null,
1155 contextHandoffId: null,
1156 })}
1157 )
1158 `;
1159 if (run > startedRuns) continue;
1160 const start =
1161 run <= 12 || run % 50 === 0
1162 ? {
1163 type: "user_message",
1164 createdBy: "user",
1165 creationSource: "web",
1166 messageId: `message:notification-turn-pages:${run}`,
1167 inputIntent: "turn_start",
1168 text: `Turn ${run}`,
1169 attachments: [],
1170 }
1171 : notification(`Wake ${run}`);
1172 // A row of the run (a handoff) can come before its first input. A wake
1173 // steered into the running turn follows it and starts no turn.
1174 const rows = [
1175 itemRow(runId, command),
1176 itemRow(runId, start),
1177 itemRow(runId, command),
1178 itemRow(runId, notification(`Steer ${run}`)),
1179 ];
1180 startIds.push(rows[1]!.turn_item_id);
1181 yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`;
1182 }
1183
1184 const { statements, tracer } = traceSqlStatements();
1185 const initial = yield* projectionStore
1186 .getThreadSnapshotWindow(threadId, { rowLimit: 77, userTurnLimit: 10 })
1187 .pipe(Effect.withTracer(tracer));
1188 // The 152 newest turns are the 152 newest started runs, 261 to 412. Only
1189 // three of them are user turns, short of the 12 the user-turn window needs.
1190 assert.deepEqual(
1191 initial.projection.turnItems.map((item) => String(item.id)),
1192 allIds.slice(allIds.indexOf(startIds[260]!)),
1193 );
1194 // Each run's first input is read by run ID, newest run first, never by a
1195 // walk over every turn item in the thread.
1196 const windowStatement = statements.find((statement) => statement.includes("run_starts"));
1197 assert.isDefined(windowStatement);
1198 const windowPlan = yield* sql.unsafe<{ readonly detail: string }>(
1199 `EXPLAIN QUERY PLAN ${windowStatement}`,
1200 );
1201 const windowPlanDetails = windowPlan.map((row) => row.detail);
1202 assert.include(
1203 windowPlanDetails,
1204 "SEARCH recent_run USING INDEX orchestration_v2_projection_runs_thread_ordinal_idx (thread_id=? AND ordinal<?)",
1205 );
1206 assert.include(
1207 windowPlanDetails,
1208 "SEARCH first_input USING INDEX orchestration_v2_projection_turn_items_run_ordinal_idx (run_id=?)",
1209 );
1210 const bounded = buildBoundedThreadProjection({
1211 projection: initial.projection,
1212 snapshotSequence: 0,
1213 });
1214 assert.strictEqual(bounded.projection.visibleTurnItems[0]?.sourceItemId, startIds[262]);
1215 const loaded = bounded.projection.visibleTurnItems.map((row) => String(row.sourceItemId));
1216 let cursor = bounded.historyCursor;
1217 for (let pages = 0; cursor !== null; pages += 1) {
1218 assert.isBelow(pages, 5);
1219 const anchor = decodeThreadHistoryCursor(cursor);
1220 const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1221 rowLimit: 77,
1222 userTurnLimit: 20,
1223 anchorItemId: TurnItemId.make(anchor.si),
1224 anchorThreadId: ThreadId.make(anchor.st),
1225 });
1226 // Older windows stay bounded too.
1227 assert.isAtMost(snapshot.projection.turnItems.length, 152 * 4 + 1);
1228 const page = selectHistoryPageFromCursor({
1229 items: snapshot.projection.visibleTurnItems,
1230 cursor,
1231 snapshotSequence: 0,
1232 });
1233 loaded.unshift(...page.items.map((row) => String(row.sourceItemId)));
1234 cursor = page.nextCursor;
1235 }
1236 assert.deepEqual(loaded, allIds);
1237 }),
1238 );
1239
1240 it.effect(
1241 "bounds completed nodes within one long run while preserving ancestry and live work",
1242 () =>
1243 Effect.gen(function* () {
1244 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1245 const now = yield* DateTime.now;
1246 const threadId = ThreadId.make("thread:bounded-node-history");
1247 yield* projectionStore.apply({
1248 id: EventId.make("event:bounded-node-history:thread"),
1249 type: "thread.created",
1250 threadId,
1251 occurredAt: now,
1252 payload: {
1253 createdBy: "user",
1254 creationSource: "web",
1255 id: threadId,
1256 projectId: ProjectId.make("project:bounded-node-history"),
1257 title: "Bounded SQL history",
1258 providerInstanceId,
1259 modelSelection,
1260 runtimeMode: "full-access",
1261 interactionMode: "default",
1262 branch: null,
1263 worktreePath: null,
1264 activeProviderThreadId: null,
1265 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
1266 forkedFrom: null,
1267 createdAt: now,
1268 updatedAt: now,
1269 archivedAt: null,
1270 settledOverride: null,
1271 settledAt: null,
1272 lastVisitedAt: null,
1273 deletedAt: null,
1274 },
1275 });
1276 const runId = RunId.make("run:bounded-node-history");
1277 const rootNodeId = NodeId.make("node:bounded-node-history:root");
1278 yield* projectionStore.apply({
1279 id: EventId.make("event:bounded-node-history:run"),
1280 type: "run.created",
1281 threadId,
1282 runId,
1283 occurredAt: now,
1284 payload: {
1285 id: runId,
1286 threadId,
1287 ordinal: 1,
1288 providerInstanceId,
1289 modelSelection,
1290 providerThreadId: null,
1291 userMessageId: MessageId.make("message:bounded-node-history"),
1292 rootNodeId,
1293 activeAttemptId: null,
1294 status: "running",
1295 requestedAt: now,
1296 startedAt: now,
1297 completedAt: null,
1298 checkpointId: null,
1299 contextHandoffId: null,
1300 },
1301 });
1302 const parentNodeId = NodeId.make("node:bounded-node-history:parent");
1303 const liveNodeId = NodeId.make("node:bounded-node-history:live");
1304 for (let index = -3; index < 1000; index++) {
1305 const id =
1306 index === -3
1307 ? rootNodeId
1308 : index === -2
1309 ? parentNodeId
1310 : index === -1
1311 ? liveNodeId
1312 : NodeId.make(`node:bounded-node-history:${index}`);
1313 yield* projectionStore.apply({
1314 id: EventId.make(`event:bounded-node-history:node:${index}`),
1315 type: "node.updated",
1316 threadId,
1317 runId,
1318 nodeId: id,
1319 driver,
1320 occurredAt: now,
1321 payload: {
1322 id,
1323 threadId,
1324 runId,
1325 rootNodeId,
1326 parentNodeId: index === -3 ? null : index === -2 ? rootNodeId : parentNodeId,
1327 kind: index === -3 ? "root_turn" : "assistant_message",
1328 status: index === -1 ? "running" : "completed",
1329 countsForRun: index === -3,
1330 providerThreadId: null,
1331 providerTurnId: null,
1332 nativeItemRef: null,
1333 runtimeRequestId: null,
1334 checkpointScopeId: null,
1335 startedAt:
1336 index === 998
1337 ? null
1338 : index === 999
1339 ? DateTime.makeUnsafe("2020-01-01T00:00:00.000Z")
1340 : now,
1341 completedAt: index === -1 ? null : now,
1342 },
1343 });
1344 if (index < 0) continue;
1345 yield* projectionStore.apply({
1346 id: EventId.make(`event:bounded-node-history:item:${index}`),
1347 type: "turn-item.updated",
1348 threadId,
1349 runId,
1350 nodeId: id,
1351 driver,
1352 occurredAt: now,
1353 payload: {
1354 id: TurnItemId.make(`item:bounded-node-history:${index}`),
1355 threadId,
1356 runId,
1357 nodeId: id,
1358 providerThreadId: null,
1359 providerTurnId: null,
1360 nativeItemRef: null,
1361 parentItemId: null,
1362 ordinal: index,
1363 status: "completed",
1364 title: null,
1365 startedAt: now,
1366 completedAt: now,
1367 updatedAt: now,
1368 type: "command_execution",
1369 input: `echo ${index}`,
1370 output: "ok",
1371 exitCode: 0,
1372 },
1373 });
1374 }
1375 const requestNodeId = NodeId.make("node:bounded-node-history:0");
1376 yield* projectionStore.apply({
1377 id: EventId.make("event:bounded-node-history:request"),
1378 type: "runtime-request.updated",
1379 threadId,
1380 runId,
1381 nodeId: requestNodeId,
1382 driver,
1383 occurredAt: now,
1384 payload: {
1385 id: RuntimeRequestId.make("request:bounded-node-history"),
1386 nodeId: requestNodeId,
1387 providerTurnId: ProviderTurnId.make("provider-turn:bounded-node-history"),
1388 nativeRequestRef: null,
1389 kind: "command",
1390 status: "pending",
1391 responseCapability: {
1392 type: "live",
1393 providerSessionId: ProviderSessionId.make("session:bounded-node-history"),
1394 },
1395 createdAt: now,
1396 resolvedAt: null,
1397 },
1398 });
1399 const { statements, tracer } = traceSqlStatements();
1400 const snapshot = yield* projectionStore
1401 .getThreadSnapshotWindow(threadId, { rowLimit: 75 })
1402 .pipe(Effect.withTracer(tracer));
1403 const nodeStatement = statements.find(
1404 (statement) =>
1405 statement.includes("payload_json") &&
1406 statement.includes("orchestration_v2_projection_nodes") &&
1407 !statement.includes("WITH RECURSIVE"),
1408 );
1409 assert.isDefined(nodeStatement);
1410 const nodePlan = yield* (yield* SqlClient.SqlClient).unsafe<{ readonly detail: string }>(
1411 `EXPLAIN QUERY PLAN ${nodeStatement}`,
1412 );
1413 const nodeLookups = nodePlan
1414 .map((row) => row.detail)
1415 .filter((detail) => detail.startsWith("SEARCH "));
1416 // Cost must follow the retained cohort, even when the thread has
1417 // thousands of completed nodes outside the requested window.
1418 assert.deepEqual(nodeLookups, [
1419 "SEARCH node USING INDEX sqlite_autoindex_orchestration_v2_projection_nodes_1 (node_id=?)",
1420 ]);
1421 assert.notInclude(
1422 nodePlan.map((row) => row.detail),
1423 "USE TEMP B-TREE FOR ORDER BY",
1424 );
1425 assert.lengthOf(snapshot.projection.visibleTurnItems, 75);
1426 assert.lengthOf(snapshot.projection.nodes, 79);
1427 const retained = new Set(snapshot.projection.nodes.map((node) => node.id));
1428 assert.isTrue(retained.has(rootNodeId));
1429 assert.isTrue(retained.has(parentNodeId));
1430 assert.isTrue(retained.has(liveNodeId));
1431 assert.isTrue(retained.has(requestNodeId));
1432 assert.lengthOf(snapshot.projection.runtimeRequests, 1);
1433 assert.isFalse(retained.has(NodeId.make("node:bounded-node-history:1")));
1434 for (const row of snapshot.projection.visibleTurnItems)
1435 assert.isTrue(retained.has(row.item.nodeId!));
1436 const older = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1437 rowLimit: 75,
1438 anchorItemId: TurnItemId.make("item:bounded-node-history:925"),
1439 });
1440 assert.lengthOf(older.projection.nodes, 79);
1441 assert.isTrue(
1442 older.projection.nodes.some(
1443 (node) => node.id === NodeId.make("node:bounded-node-history:851"),
1444 ),
1445 );
1446 const full = yield* projectionStore.getThreadSnapshot(threadId);
1447 assert.lengthOf(full.projection.nodes, 1003);
1448 // Bounded reads keep the full read's payloads and ordering, including
1449 // null start times and equal timestamps.
1450 assert.deepEqual(
1451 snapshot.projection.nodes,
1452 full.projection.nodes.filter((node) => retained.has(node.id)),
1453 );
1454 const sql = yield* SqlClient.SqlClient;
1455 const duplicateItemId = TurnItemId.make("item:bounded-node-history:duplicate");
1456 yield* sql`
1457 INSERT INTO orchestration_v2_projection_turn_items
1458 SELECT ${duplicateItemId}, thread_id, run_id, node_id, provider_thread_id,
1459 provider_turn_id, parent_item_id, ordinal + 1, type, status, updated_at,
1460 json_set(payload_json, '$.id', ${duplicateItemId}, '$.ordinal', ordinal + 1)
1461 FROM orchestration_v2_projection_turn_items
1462 WHERE turn_item_id = ${TurnItemId.make("item:bounded-node-history:999")}
1463 `;
1464 const sharedNodeSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1465 rowLimit: 76,
1466 });
1467 assert.lengthOf(sharedNodeSnapshot.projection.visibleTurnItems, 76);
1468 assert.deepEqual(sharedNodeSnapshot.projection.nodes, snapshot.projection.nodes);
1469 }),
1470 );
1471
1472 it.effect("reads a fixed SQL turn-item window for long histories and repeated clients", () =>
1473 Effect.gen(function* () {
1474 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1475 const sql = yield* SqlClient.SqlClient;
1476 const now = yield* DateTime.now;
1477 const nowIso = DateTime.formatIso(now);
1478 const threadId = ThreadId.make("thread:bounded-sql-history");
1479 yield* projectionStore.apply({
1480 id: EventId.make("event:bounded-sql-history:thread"),
1481 type: "thread.created",
1482 threadId,
1483 occurredAt: now,
1484 payload: {
1485 createdBy: "user",
1486 creationSource: "web",
1487 id: threadId,
1488 projectId: ProjectId.make("project:bounded-sql-history"),
1489 title: "Bounded SQL history",
1490 providerInstanceId,
1491 modelSelection,
1492 runtimeMode: "full-access",
1493 interactionMode: "default",
1494 branch: null,
1495 worktreePath: null,
1496 activeProviderThreadId: null,
1497 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
1498 forkedFrom: null,
1499 createdAt: now,
1500 updatedAt: now,
1501 archivedAt: null,
1502 settledOverride: null,
1503 settledAt: null,
1504 lastVisitedAt: null,
1505 deletedAt: null,
1506 },
1507 });
1508
1509 for (let ordinal = 1; ordinal <= 1_000; ordinal += 1) {
1510 const id = `turn-item:bounded-sql-history:${ordinal}`;
1511 const runId = `run:bounded-sql-history:${ordinal}`;
1512 yield* sql`
1513 INSERT INTO orchestration_v2_projection_runs (
1514 run_id, thread_id, ordinal, provider, provider_thread_id, status,
1515 requested_at, completed_at, payload_json
1516 ) VALUES (
1517 ${runId}, ${threadId}, ${ordinal}, 'codex', NULL, 'completed', ${nowIso}, ${nowIso},
1518 ${encodeUnknownJsonString({
1519 id: runId,
1520 threadId,
1521 ordinal,
1522 providerInstanceId,
1523 modelSelection,
1524 providerThreadId: null,
1525 userMessageId: `message:bounded-sql-history:${ordinal}`,
1526 rootNodeId: `node:bounded-sql-history:${ordinal}`,
1527 activeAttemptId: null,
1528 status: "completed",
1529 requestedAt: nowIso,
1530 startedAt: nowIso,
1531 completedAt: nowIso,
1532 checkpointId: null,
1533 contextHandoffId: null,
1534 })}
1535 )
1536 `;
1537 yield* sql`
1538 INSERT INTO orchestration_v2_projection_turn_items (
1539 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1540 parent_item_id, ordinal, type, status, updated_at, payload_json
1541 ) VALUES (
1542 ${id}, ${threadId}, ${runId}, NULL, NULL, NULL, NULL, ${ordinal},
1543 'command_execution', 'completed', ${nowIso},
1544 ${encodeUnknownJsonString({
1545 id,
1546 threadId,
1547 runId,
1548 nodeId: null,
1549 providerThreadId: null,
1550 providerTurnId: null,
1551 nativeItemRef: null,
1552 parentItemId: null,
1553 ordinal,
1554 status: "completed",
1555 title: `command ${ordinal}`,
1556 input: `echo ${ordinal}`,
1557 output: "ok",
1558 exitCode: 0,
1559 startedAt: nowIso,
1560 completedAt: nowIso,
1561 updatedAt: nowIso,
1562 type: "command_execution",
1563 })}
1564 )
1565 `;
1566 }
1567
1568 const snapshots = yield* Effect.all(
1569 Array.from({ length: 4 }, () =>
1570 projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 76 }),
1571 ),
1572 { concurrency: 4 },
1573 );
1574 for (const snapshot of snapshots) {
1575 assert.lengthOf(snapshot.projection.turnItems, 76);
1576 assert.lengthOf(snapshot.projection.runs, 76);
1577 assert.lengthOf(snapshot.projection.visibleTurnItems, 76);
1578 assert.strictEqual(snapshot.projection.turnItems[0]?.ordinal, 925);
1579 assert.strictEqual(snapshot.projection.turnItems.at(-1)?.ordinal, 1_000);
1580 }
1581 const retainedRequestPlan = yield* sql<{ readonly detail: string }>`
1582 EXPLAIN QUERY PLAN
1583 WITH selected AS (
1584 SELECT run_id, type
1585 FROM orchestration_v2_projection_turn_items
1586 WHERE thread_id = ${threadId}
1587 ORDER BY ordinal DESC, turn_item_id DESC
1588 LIMIT 77
1589 )
1590 SELECT request.payload_json
1591 FROM orchestration_v2_projection_turn_items AS request
1592 WHERE request.run_id IN (
1593 SELECT run_id FROM selected
1594 WHERE type = 'run_interrupt_result' AND run_id IS NOT NULL
1595 )
1596 AND request.type = 'run_interrupt_request'
1597 `;
1598 assert.isTrue(
1599 retainedRequestPlan.some((row) =>
1600 row.detail.includes("orchestration_v2_projection_turn_items_run_ordinal_idx"),
1601 ),
1602 );
1603
1604 const older = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1605 rowLimit: 76,
1606 anchorItemId: TurnItemId.make("turn-item:bounded-sql-history:925"),
1607 });
1608 assert.lengthOf(older.projection.turnItems, 76);
1609 assert.lengthOf(older.projection.runs, 76);
1610 assert.strictEqual(older.projection.turnItems[0]?.ordinal, 850);
1611 assert.strictEqual(older.projection.turnItems.at(-1)?.ordinal, 925);
1612
1613 // A single item reads back with its full output, scoped to its thread.
1614 const itemId = TurnItemId.make("turn-item:bounded-sql-history:925");
1615 const stored = yield* projectionStore.getTurnItem({ threadId, itemId });
1616 assert.strictEqual(stored?.type === "command_execution" ? stored.output : undefined, "ok");
1617 assert.isNull(
1618 yield* projectionStore.getTurnItem({ threadId: ThreadId.make("thread:other"), itemId }),
1619 );
1620
1621 const sqlPageLimit = THREAD_HISTORY_PAGE_POLICY.maxItems + 2;
1622 const initialSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1623 rowLimit: sqlPageLimit,
1624 });
1625 const initialPage = buildBoundedThreadProjection({
1626 policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
1627 projection: initialSnapshot.projection,
1628 snapshotSequence: initialSnapshot.snapshotSequence,
1629 });
1630 const loadedIds = initialPage.projection.visibleTurnItems.map((row) =>
1631 String(row.sourceItemId),
1632 );
1633 let cursor = initialPage.historyCursor;
1634 let pageCount = 1;
1635 while (cursor !== null) {
1636 const anchorItemId = TurnItemId.make(decodeThreadHistoryCursor(cursor).si);
1637 const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1638 rowLimit: sqlPageLimit,
1639 anchorItemId,
1640 });
1641 const page = selectHistoryPageFromCursor({
1642 policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
1643 items: snapshot.projection.visibleTurnItems,
1644 cursor,
1645 snapshotSequence: snapshot.snapshotSequence,
1646 });
1647 loadedIds.push(...page.items.map((row) => String(row.sourceItemId)));
1648 pageCount += 1;
1649 cursor = page.nextCursor;
1650 }
1651
1652 assert.isAtLeast(pageCount, 4);
1653 assert.lengthOf(loadedIds, 1_000);
1654 assert.strictEqual(new Set(loadedIds).size, 1_000);
1655 assert.deepEqual(
1656 loadedIds.toSorted(
1657 (left, right) => Number(left.split(":").at(-1)) - Number(right.split(":").at(-1)),
1658 ),
1659 Array.from({ length: 1_000 }, (_, index) => `turn-item:bounded-sql-history:${index + 1}`),
1660 );
1661
1662 // Physical insertion order must not change the ordinal/ID timeline order.
1663 const tiedIds = [
1664 TurnItemId.make("turn-item:bounded-sql-history:tie:b"),
1665 TurnItemId.make("turn-item:bounded-sql-history:tie:a"),
1666 ];
1667 for (const id of tiedIds) {
1668 yield* sql`
1669 INSERT INTO orchestration_v2_projection_turn_items (
1670 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1671 parent_item_id, ordinal, type, status, updated_at, payload_json
1672 )
1673 SELECT ${id}, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1674 parent_item_id, 2000, type, status, updated_at,
1675 json_set(payload_json, '$.id', ${id}, '$.ordinal', 2000)
1676 FROM orchestration_v2_projection_turn_items
1677 WHERE turn_item_id = ${itemId}
1678 `;
1679 }
1680 const tied = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 3 });
1681 assert.deepEqual(
1682 tied.projection.turnItems.map((item) => item.id),
1683 [TurnItemId.make("turn-item:bounded-sql-history:1000"), tiedIds[1], tiedIds[0]],
1684 );
1685 yield* sql`
1686 DELETE FROM orchestration_v2_projection_turn_items
1687 WHERE turn_item_id IN ${sql.in(tiedIds)}
1688 `;
1689
1690 const hiddenRunId = RunId.make("run:bounded-sql-history:hidden-suffix");
1691 yield* sql`
1692 INSERT INTO orchestration_v2_projection_runs (
1693 run_id, thread_id, ordinal, provider, provider_thread_id, status,
1694 requested_at, completed_at, payload_json
1695 ) VALUES (
1696 ${hiddenRunId}, ${threadId}, 1001, 'codex', NULL, 'rolled_back', ${nowIso}, ${nowIso},
1697 ${encodeUnknownJsonString({
1698 id: hiddenRunId,
1699 threadId,
1700 ordinal: 1001,
1701 providerInstanceId,
1702 modelSelection,
1703 providerThreadId: null,
1704 userMessageId: "message:bounded-sql-history:hidden-suffix",
1705 rootNodeId: "node:bounded-sql-history:hidden-suffix",
1706 activeAttemptId: null,
1707 status: "rolled_back",
1708 requestedAt: nowIso,
1709 startedAt: nowIso,
1710 completedAt: nowIso,
1711 checkpointId: null,
1712 contextHandoffId: null,
1713 })}
1714 )
1715 `;
1716 for (let ordinal = 1_001; ordinal <= 1_100; ordinal += 1) {
1717 const id = `turn-item:bounded-sql-history:hidden:${ordinal}`;
1718 yield* sql`
1719 INSERT INTO orchestration_v2_projection_turn_items (
1720 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1721 parent_item_id, ordinal, type, status, updated_at, payload_json
1722 ) VALUES (
1723 ${id}, ${threadId}, ${hiddenRunId}, NULL, NULL, NULL, NULL, ${ordinal},
1724 'command_execution', 'completed', ${nowIso},
1725 ${encodeUnknownJsonString({
1726 id,
1727 threadId,
1728 runId: hiddenRunId,
1729 nodeId: null,
1730 providerThreadId: null,
1731 providerTurnId: null,
1732 nativeItemRef: null,
1733 parentItemId: null,
1734 ordinal,
1735 status: "completed",
1736 title: "rolled back",
1737 input: "echo hidden",
1738 output: "hidden",
1739 exitCode: 0,
1740 startedAt: nowIso,
1741 completedAt: nowIso,
1742 updatedAt: nowIso,
1743 type: "command_execution",
1744 })}
1745 )
1746 `;
1747 }
1748
1749 const hiddenSuffixSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1750 rowLimit: sqlPageLimit,
1751 });
1752 const hiddenSuffixPage = buildBoundedThreadProjection({
1753 policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
1754 projection: hiddenSuffixSnapshot.projection,
1755 snapshotSequence: hiddenSuffixSnapshot.snapshotSequence,
1756 });
1757 assert.lengthOf(hiddenSuffixPage.projection.visibleTurnItems, 75);
1758 assert.strictEqual(hiddenSuffixPage.latestLocalTurnOrdinal, 1_100);
1759 assert.isTrue(hiddenSuffixPage.hasMoreHistory);
1760 assert.isFalse(
1761 hiddenSuffixPage.projection.visibleTurnItems.some((row) =>
1762 String(row.sourceItemId).includes(":hidden:"),
1763 ),
1764 );
1765 const hiddenSuffixCursor = hiddenSuffixPage.historyCursor;
1766 assert.isNotNull(hiddenSuffixCursor);
1767 const hiddenSuffixAnchor = TurnItemId.make(decodeThreadHistoryCursor(hiddenSuffixCursor!).si);
1768 const hiddenSuffixOlderSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1769 rowLimit: sqlPageLimit,
1770 anchorItemId: hiddenSuffixAnchor,
1771 });
1772 const hiddenSuffixOlderPage = selectHistoryPageFromCursor({
1773 policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
1774 items: hiddenSuffixOlderSnapshot.projection.visibleTurnItems,
1775 cursor: hiddenSuffixCursor!,
1776 snapshotSequence: hiddenSuffixOlderSnapshot.snapshotSequence,
1777 });
1778 assert.lengthOf(hiddenSuffixOlderPage.items, 75);
1779 assert.isTrue(hiddenSuffixOlderPage.hasMoreHistory);
1780
1781 const cancelledRunId = RunId.make("run:bounded-sql-history:cancelled-suffix");
1782 yield* sql`
1783 INSERT INTO orchestration_v2_projection_runs (
1784 run_id, thread_id, ordinal, provider, provider_thread_id, status,
1785 requested_at, completed_at, payload_json
1786 ) VALUES (
1787 ${cancelledRunId}, ${threadId}, 1002, 'codex', NULL, 'cancelled', ${nowIso}, ${nowIso},
1788 ${encodeUnknownJsonString({
1789 id: cancelledRunId,
1790 threadId,
1791 ordinal: 1002,
1792 providerInstanceId,
1793 modelSelection,
1794 providerThreadId: null,
1795 userMessageId: "message:bounded-sql-history:cancelled-suffix",
1796 rootNodeId: "node:bounded-sql-history:cancelled-suffix",
1797 activeAttemptId: null,
1798 status: "cancelled",
1799 requestedAt: nowIso,
1800 startedAt: nowIso,
1801 completedAt: nowIso,
1802 checkpointId: null,
1803 contextHandoffId: null,
1804 })}
1805 )
1806 `;
1807 for (let ordinal = 1_101; ordinal <= 1_200; ordinal += 1) {
1808 const id = `turn-item:bounded-sql-history:cancelled:${ordinal}`;
1809 yield* sql`
1810 INSERT INTO orchestration_v2_projection_turn_items (
1811 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1812 parent_item_id, ordinal, type, status, updated_at, payload_json
1813 ) VALUES (
1814 ${id}, ${threadId}, ${cancelledRunId}, NULL, NULL, NULL, NULL, ${ordinal},
1815 'user_message', 'completed', ${nowIso},
1816 ${encodeUnknownJsonString({
1817 createdBy: "user",
1818 creationSource: "web",
1819 id,
1820 threadId,
1821 runId: cancelledRunId,
1822 nodeId: null,
1823 providerThreadId: null,
1824 providerTurnId: null,
1825 nativeItemRef: null,
1826 parentItemId: null,
1827 ordinal,
1828 status: "completed",
1829 title: "cancelled queued message",
1830 startedAt: nowIso,
1831 completedAt: nowIso,
1832 updatedAt: nowIso,
1833 type: "user_message",
1834 messageId: `message:bounded-sql-history:cancelled:${ordinal}`,
1835 inputIntent: "queued_turn",
1836 text: "cancelled",
1837 attachments: [],
1838 })}
1839 )
1840 `;
1841 }
1842 const runlessQueuedId = "turn-item:bounded-sql-history:runless-queued";
1843 yield* sql`
1844 INSERT INTO orchestration_v2_projection_turn_items (
1845 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1846 parent_item_id, ordinal, type, status, updated_at, payload_json
1847 ) VALUES (
1848 ${runlessQueuedId}, ${threadId}, NULL, NULL, NULL, NULL, NULL, 1201,
1849 'user_message', 'completed', ${nowIso},
1850 ${encodeUnknownJsonString({
1851 createdBy: "user",
1852 creationSource: "web",
1853 id: runlessQueuedId,
1854 threadId,
1855 runId: null,
1856 nodeId: null,
1857 providerThreadId: null,
1858 providerTurnId: null,
1859 nativeItemRef: null,
1860 parentItemId: null,
1861 ordinal: 1201,
1862 status: "completed",
1863 title: "runless queued message",
1864 startedAt: nowIso,
1865 completedAt: nowIso,
1866 updatedAt: nowIso,
1867 type: "user_message",
1868 messageId: "message:bounded-sql-history:runless-queued",
1869 inputIntent: "queued_turn",
1870 text: "still visible",
1871 attachments: [],
1872 })}
1873 )
1874 `;
1875 const laterCancelledId = "turn-item:bounded-sql-history:cancelled:1202";
1876 yield* sql`
1877 INSERT INTO orchestration_v2_projection_turn_items (
1878 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1879 parent_item_id, ordinal, type, status, updated_at, payload_json
1880 ) VALUES (
1881 ${laterCancelledId}, ${threadId}, ${cancelledRunId}, NULL, NULL, NULL, NULL, 1202,
1882 'user_message', 'completed', ${nowIso},
1883 ${encodeUnknownJsonString({
1884 createdBy: "user",
1885 creationSource: "web",
1886 id: laterCancelledId,
1887 threadId,
1888 runId: cancelledRunId,
1889 nodeId: null,
1890 providerThreadId: null,
1891 providerTurnId: null,
1892 nativeItemRef: null,
1893 parentItemId: null,
1894 ordinal: 1202,
1895 status: "completed",
1896 title: "later cancelled queued message",
1897 startedAt: nowIso,
1898 completedAt: nowIso,
1899 updatedAt: nowIso,
1900 type: "user_message",
1901 messageId: "message:bounded-sql-history:cancelled:1202",
1902 inputIntent: "queued_turn",
1903 text: "cancelled",
1904 attachments: [],
1905 })}
1906 )
1907 `;
1908 const cancelledSuffixSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
1909 rowLimit: sqlPageLimit,
1910 });
1911 const cancelledSuffixPage = buildBoundedThreadProjection({
1912 policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
1913 projection: cancelledSuffixSnapshot.projection,
1914 snapshotSequence: cancelledSuffixSnapshot.snapshotSequence,
1915 });
1916 assert.strictEqual(cancelledSuffixPage.latestLocalTurnOrdinal, 1_202);
1917 assert.lengthOf(cancelledSuffixPage.projection.visibleTurnItems, 75);
1918 assert.isFalse(
1919 cancelledSuffixPage.projection.visibleTurnItems.some((row) =>
1920 String(row.sourceItemId).includes(":cancelled:"),
1921 ),
1922 );
1923 assert.isTrue(
1924 cancelledSuffixPage.projection.visibleTurnItems.some(
1925 (row) => row.sourceItemId === runlessQueuedId,
1926 ),
1927 );
1928
1929 const interruptRunId = RunId.make("run:bounded-sql-history:interrupt");
1930 const interruptNodeId = NodeId.make("node:bounded-sql-history:interrupt");
1931 const interruptRequestId = TurnItemId.make("turn-item:bounded-sql-history:interrupt-request");
1932 const interruptResultId = TurnItemId.make("turn-item:bounded-sql-history:interrupt-result");
1933 yield* sql`
1934 INSERT INTO orchestration_v2_projection_runs (
1935 run_id, thread_id, ordinal, provider, provider_thread_id, status,
1936 requested_at, completed_at, payload_json
1937 ) VALUES (
1938 ${interruptRunId}, ${threadId}, 1003, 'codex', 'provider-thread:interrupt', 'completed',
1939 ${nowIso}, ${nowIso},
1940 ${encodeUnknownJsonString({
1941 id: interruptRunId,
1942 threadId,
1943 ordinal: 1003,
1944 providerInstanceId,
1945 modelSelection,
1946 providerThreadId: "provider-thread:interrupt",
1947 userMessageId: "message:interrupt",
1948 rootNodeId: interruptNodeId,
1949 activeAttemptId: "attempt:bounded-sql-history:interrupt",
1950 status: "completed",
1951 requestedAt: nowIso,
1952 startedAt: nowIso,
1953 completedAt: nowIso,
1954 checkpointId: null,
1955 contextHandoffId: null,
1956 })}
1957 )
1958 `;
1959 yield* sql`
1960 INSERT INTO orchestration_v2_projection_run_attempts (
1961 attempt_id, thread_id, run_id, attempt_ordinal, root_node_id, provider,
1962 provider_instance_id, provider_thread_id, provider_turn_id, status, payload_json
1963 ) VALUES (
1964 'attempt:bounded-sql-history:interrupt', ${threadId}, ${interruptRunId}, 1,
1965 ${interruptNodeId}, 'codex', ${providerInstanceId}, 'provider-thread:interrupt', NULL,
1966 'superseded',
1967 ${encodeUnknownJsonString({
1968 id: "attempt:bounded-sql-history:interrupt",
1969 runId: interruptRunId,
1970 attemptOrdinal: 1,
1971 rootNodeId: interruptNodeId,
1972 providerInstanceId,
1973 providerThreadId: "provider-thread:interrupt",
1974 providerTurnId: null,
1975 reason: "initial",
1976 status: "superseded",
1977 startedAt: nowIso,
1978 completedAt: nowIso,
1979 })}
1980 )
1981 `;
1982 const insertInterruptItem = (input: {
1983 id: TurnItemId;
1984 ordinal: number;
1985 type: "run_interrupt_request" | "run_interrupt_result";
1986 }) => sql`
1987 INSERT INTO orchestration_v2_projection_turn_items (
1988 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
1989 parent_item_id, ordinal, type, status, updated_at, payload_json
1990 ) VALUES (
1991 ${input.id}, ${threadId}, ${interruptRunId}, ${interruptNodeId}, NULL, NULL, NULL,
1992 ${input.ordinal}, ${input.type}, 'completed', ${nowIso},
1993 ${encodeUnknownJsonString({
1994 id: input.id,
1995 threadId,
1996 runId: interruptRunId,
1997 nodeId: interruptNodeId,
1998 providerThreadId: null,
1999 providerTurnId: null,
2000 nativeItemRef: null,
2001 parentItemId: null,
2002 ordinal: input.ordinal,
2003 status: "completed",
2004 title: input.type,
2005 message: input.type === "run_interrupt_request" ? "Stopping" : "Stopped",
2006 startedAt: nowIso,
2007 completedAt: nowIso,
2008 updatedAt: nowIso,
2009 type: input.type,
2010 })}
2011 )
2012 `;
2013 yield* insertInterruptItem({
2014 id: interruptRequestId,
2015 ordinal: 1_201,
2016 type: "run_interrupt_request",
2017 });
2018 for (let ordinal = 1_202; ordinal <= 1_281; ordinal += 1) {
2019 const id = `turn-item:bounded-sql-history:interrupt-filler:${ordinal}`;
2020 yield* sql`
2021 INSERT INTO orchestration_v2_projection_turn_items (
2022 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
2023 parent_item_id, ordinal, type, status, updated_at, payload_json
2024 ) VALUES (
2025 ${id}, ${threadId}, NULL, NULL, NULL, NULL, NULL, ${ordinal},
2026 'command_execution', 'completed', ${nowIso},
2027 ${encodeUnknownJsonString({
2028 id,
2029 threadId,
2030 runId: null,
2031 nodeId: null,
2032 providerThreadId: null,
2033 providerTurnId: null,
2034 nativeItemRef: null,
2035 parentItemId: null,
2036 ordinal,
2037 status: "completed",
2038 title: "filler",
2039 input: "echo filler",
2040 output: "ok",
2041 exitCode: 0,
2042 startedAt: nowIso,
2043 completedAt: nowIso,
2044 updatedAt: nowIso,
2045 type: "command_execution",
2046 })}
2047 )
2048 `;
2049 }
2050 yield* insertInterruptItem({
2051 id: interruptResultId,
2052 ordinal: 1_282,
2053 type: "run_interrupt_result",
2054 });
2055
2056 const interruptSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
2057 rowLimit: sqlPageLimit,
2058 });
2059 assert.isTrue(
2060 interruptSnapshot.projection.turnItems.some((item) => item.id === interruptRequestId),
2061 );
2062 assert.isTrue(
2063 interruptSnapshot.projection.visibleTurnItems.some(
2064 (row) => row.sourceItemId === interruptResultId,
2065 ),
2066 );
2067
2068 yield* sql`
2069 UPDATE orchestration_v2_projection_turn_items
2070 SET payload_json = json_set(payload_json, '$.createdBy', 'agent')
2071 WHERE thread_id = ${threadId} AND type = 'user_message'
2072 `;
2073 const agentPromptId = TurnItemId.make("turn-item:bounded-sql-history:interrupt-filler:1281");
2074 yield* sql`
2075 UPDATE orchestration_v2_projection_turn_items
2076 SET type = 'user_message',
2077 payload_json = json_set(payload_json,
2078 '$.type', 'user_message', '$.inputIntent', 'turn_start',
2079 '$.createdBy', 'agent', '$.creationSource', 'provider',
2080 '$.messageId', 'message:bounded-sql-history:agent-prompt',
2081 '$.text', 'Continue the child task', '$.attachments', json('[]'))
2082 WHERE turn_item_id = ${agentPromptId}
2083 `;
2084 const agentWindow = yield* projectionStore.getThreadSnapshotWindow(threadId, {
2085 rowLimit: sqlPageLimit,
2086 userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns,
2087 });
2088 assert.lengthOf(
2089 agentWindow.projection.visibleTurnItems.filter(
2090 (row) => row.item.type !== "run_interrupt_request",
2091 ),
2092 sqlPageLimit,
2093 );
2094 assert.isTrue(
2095 agentWindow.projection.visibleTurnItems.some((row) => row.sourceItemId === agentPromptId),
2096 );
2097 }),
2098 );
2099
2100 it.effect("scopes actionable provider state to the requested bounded thread", () =>
2101 Effect.gen(function* () {
2102 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2103 const now = yield* DateTime.now;
2104 const projectId = ProjectId.make("project:bounded-provider-scope");
2105 const targetThreadId = ThreadId.make("thread:bounded-provider-scope:target");
2106 const unrelatedThreadId = ThreadId.make("thread:bounded-provider-scope:unrelated");
2107 const makeThread = (threadId: ThreadId) => ({
2108 createdBy: "user" as const,
2109 creationSource: "web" as const,
2110 id: threadId,
2111 projectId,
2112 title: String(threadId),
2113 providerInstanceId,
2114 modelSelection,
2115 runtimeMode: "full-access" as const,
2116 interactionMode: "default" as const,
2117 branch: null,
2118 worktreePath: null,
2119 activeProviderThreadId: null,
2120 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
2121 forkedFrom: null,
2122 createdAt: now,
2123 updatedAt: now,
2124 archivedAt: null,
2125 settledOverride: null,
2126 settledAt: null,
2127 lastVisitedAt: null,
2128 deletedAt: null,
2129 });
2130 for (const threadId of [targetThreadId, unrelatedThreadId]) {
2131 yield* projectionStore.apply({
2132 id: EventId.make(`event:bounded-provider-scope:thread:${threadId}`),
2133 type: "thread.created",
2134 threadId,
2135 occurredAt: now,
2136 payload: makeThread(threadId),
2137 });
2138 const sessionId = ProviderSessionId.make(`provider-session:${threadId}`);
2139 const providerThreadId = ProviderThreadId.make(`provider-thread:${threadId}`);
2140 yield* projectionStore.apply({
2141 id: EventId.make(`event:bounded-provider-scope:session:${threadId}`),
2142 type: "provider-session.attached",
2143 threadId,
2144 driver,
2145 occurredAt: now,
2146 payload: {
2147 id: sessionId,
2148 driver,
2149 providerInstanceId,
2150 status: "running",
2151 cwd: "/workspace",
2152 model: modelSelection.model,
2153 capabilities: CodexProviderCapabilitiesV2,
2154 createdAt: now,
2155 updatedAt: now,
2156 lastError: null,
2157 },
2158 });
2159 yield* projectionStore.apply({
2160 id: EventId.make(`event:bounded-provider-scope:provider-thread:${threadId}`),
2161 type: "provider-thread.updated",
2162 threadId,
2163 driver,
2164 occurredAt: now,
2165 payload: {
2166 id: providerThreadId,
2167 driver,
2168 providerInstanceId,
2169 providerSessionId: sessionId,
2170 appThreadId: threadId,
2171 ownerNodeId: null,
2172 nativeThreadRef: null,
2173 nativeConversationHeadRef: null,
2174 status: "active",
2175 firstRunOrdinal: null,
2176 lastRunOrdinal: null,
2177 handoffIds: [],
2178 forkedFrom: null,
2179 createdAt: now,
2180 updatedAt: now,
2181 },
2182 });
2183 }
2184
2185 const snapshot = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
2186 rowLimit: 10,
2187 });
2188
2189 assert.deepEqual(
2190 snapshot.projection.providerSessions.map((session) => session.id),
2191 [ProviderSessionId.make(`provider-session:${targetThreadId}`)],
2192 );
2193 assert.deepEqual(
2194 snapshot.projection.providerThreads.map((thread) => thread.id),
2195 [ProviderThreadId.make(`provider-thread:${targetThreadId}`)],
2196 );
2197 }),
2198 );
2199
2200 it.effect("projects root provider owners into the shell in first-use order", () =>
2201 Effect.gen(function* () {
2202 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2203 const now = yield* DateTime.now;
2204 const projectId = ProjectId.make("project:provider-history");
2205 const threadId = ThreadId.make("thread:provider-history");
2206 const claudeInstanceId = ProviderInstanceId.make("claude");
2207 const claudeDriver = ProviderDriverKind.make("claudeAgent");
2208 yield* projectionStore.apply({
2209 id: EventId.make("event:provider-history:thread"),
2210 type: "thread.created",
2211 threadId,
2212 occurredAt: now,
2213 payload: {
2214 createdBy: "user",
2215 creationSource: "web",
2216 id: threadId,
2217 projectId,
2218 title: "Provider history",
2219 providerInstanceId,
2220 modelSelection,
2221 runtimeMode: "full-access",
2222 interactionMode: "default",
2223 branch: null,
2224 worktreePath: null,
2225 activeProviderThreadId: null,
2226 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
2227 forkedFrom: null,
2228 createdAt: now,
2229 updatedAt: now,
2230 archivedAt: null,
2231 settledOverride: null,
2232 settledAt: null,
2233 lastVisitedAt: null,
2234 deletedAt: null,
2235 },
2236 });
2237 const providerThreads = [
2238 // The root Codex conversation, then a Claude subagent it delegated to
2239 // (owned by a node, so not a handoff), then the handoff target.
2240 { suffix: "codex", instanceId: providerInstanceId, ownerNodeId: null, seconds: 0 },
2241 {
2242 suffix: "claude-subagent",
2243 instanceId: claudeInstanceId,
2244 ownerNodeId: NodeId.make("node:provider-history"),
2245 seconds: 1,
2246 },
2247 { suffix: "claude", instanceId: claudeInstanceId, ownerNodeId: null, seconds: 2 },
2248 // A second Codex conversation after handing back: no duplicate entry.
2249 { suffix: "codex-again", instanceId: providerInstanceId, ownerNodeId: null, seconds: 3 },
2250 ] as const;
2251 for (const providerThread of providerThreads) {
2252 const createdAt = DateTime.add(now, { seconds: providerThread.seconds });
2253 yield* projectionStore.apply({
2254 id: EventId.make(`event:provider-history:${providerThread.suffix}`),
2255 type: "provider-thread.updated",
2256 threadId,
2257 driver: providerThread.instanceId === claudeInstanceId ? claudeDriver : driver,
2258 occurredAt: createdAt,
2259 payload: {
2260 id: ProviderThreadId.make(`provider-thread:provider-history:${providerThread.suffix}`),
2261 driver: providerThread.instanceId === claudeInstanceId ? claudeDriver : driver,
2262 providerInstanceId: providerThread.instanceId,
2263 providerSessionId: null,
2264 appThreadId: threadId,
2265 ownerNodeId: providerThread.ownerNodeId,
2266 nativeThreadRef: null,
2267 nativeConversationHeadRef: null,
2268 status: "idle",
2269 firstRunOrdinal: null,
2270 lastRunOrdinal: null,
2271 handoffIds: [],
2272 forkedFrom: null,
2273 createdAt,
2274 updatedAt: createdAt,
2275 },
2276 });
2277 }
2278
2279 const shell = (yield* projectionStore.getShellSnapshot()).threads.find(
2280 (thread) => thread.id === threadId,
2281 );
2282 assert.deepEqual(shell?.providerInstanceHistory, [providerInstanceId, claudeInstanceId]);
2283 }),
2284 );
2285
2286 it.effect("shows the native goal of the active provider thread on the shell", () =>
2287 Effect.gen(function* () {
2288 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2289 const now = yield* DateTime.now;
2290 const threadId = ThreadId.make("thread:provider-goal");
2291 yield* projectionStore.apply({
2292 id: EventId.make("event:provider-goal:thread"),
2293 type: "thread.created",
2294 threadId,
2295 occurredAt: now,
2296 payload: {
2297 createdBy: "user",
2298 creationSource: "web",
2299 id: threadId,
2300 projectId: ProjectId.make("project:provider-goal"),
2301 title: "Provider goal",
2302 providerInstanceId,
2303 modelSelection,
2304 runtimeMode: "full-access",
2305 interactionMode: "default",
2306 branch: null,
2307 worktreePath: null,
2308 activeProviderThreadId: null,
2309 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
2310 forkedFrom: null,
2311 createdAt: now,
2312 updatedAt: now,
2313 archivedAt: null,
2314 settledOverride: null,
2315 settledAt: null,
2316 lastVisitedAt: null,
2317 deletedAt: null,
2318 },
2319 });
2320 const applyProviderThread = (
2321 suffix: string,
2322 goal: OrchestrationV2ProviderThread["goal"],
2323 seconds: number,
2324 ) =>
2325 projectionStore.apply({
2326 id: EventId.make(`event:provider-goal:${suffix}:${seconds}`),
2327 type: "provider-thread.updated",
2328 threadId,
2329 driver,
2330 occurredAt: DateTime.add(now, { seconds }),
2331 payload: {
2332 id: ProviderThreadId.make(`provider-thread:provider-goal:${suffix}`),
2333 driver,
2334 providerInstanceId,
2335 providerSessionId: null,
2336 appThreadId: threadId,
2337 ownerNodeId: null,
2338 nativeThreadRef: null,
2339 nativeConversationHeadRef: null,
2340 status: "idle",
2341 firstRunOrdinal: null,
2342 lastRunOrdinal: null,
2343 handoffIds: [],
2344 forkedFrom: null,
2345 goal,
2346 createdAt: now,
2347 updatedAt: DateTime.add(now, { seconds }),
2348 },
2349 });
2350 const shellGoal = Effect.map(
2351 projectionStore.getShellSnapshot(),
2352 (snapshot) => snapshot.threads.find((thread) => thread.id === threadId)?.goal,
2353 );
2354 const goal = { objective: "Ship it", status: "active" as const, tokensUsed: 10 };
2355
2356 yield* applyProviderThread("first", goal, 0);
2357 assert.deepEqual(yield* shellGoal, goal);
2358 // A handoff moves the conversation; the previous provider's goal stays behind.
2359 yield* applyProviderThread("second", null, 1);
2360 assert.isNull(yield* shellGoal);
2361 }),
2362 );
2363
2364 it.effect("reads the shell snapshot first and decodes it in a separate step", () =>
2365 Effect.gen(function* () {
2366 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2367 const sql = yield* SqlClient.SqlClient;
2368 const now = yield* DateTime.now;
2369 const createThread = (threadId: ThreadId) =>
2370 projectionStore.apply({
2371 id: EventId.make(`event:${threadId}:created`),
2372 type: "thread.created",
2373 threadId,
2374 occurredAt: now,
2375 payload: {
2376 createdBy: "user",
2377 creationSource: "web",
2378 id: threadId,
2379 projectId: ProjectId.make("project:shell-read"),
2380 title: "Shell read",
2381 providerInstanceId,
2382 modelSelection,
2383 runtimeMode: "full-access",
2384 interactionMode: "default",
2385 branch: null,
2386 worktreePath: null,
2387 activeProviderThreadId: null,
2388 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
2389 forkedFrom: null,
2390 createdAt: now,
2391 updatedAt: now,
2392 archivedAt: null,
2393 settledOverride: null,
2394 settledAt: null,
2395 lastVisitedAt: null,
2396 deletedAt: null,
2397 },
2398 });
2399 const readThreadId = ThreadId.make("thread:shell-read:before");
2400 const laterThreadId = ThreadId.make("thread:shell-read:after");
2401 const shellIds = (snapshot: OrchestrationV2ThreadShellSnapshot) =>
2402 snapshot.threads
2403 .map((thread) => thread.id)
2404 .filter((id) => id === readThreadId || id === laterThreadId);
2405
2406 yield* createThread(readThreadId);
2407 const decode = yield* sql.withTransaction(projectionStore.readShellSnapshot());
2408 // Commits between the read and the decode, as another request can while
2409 // the HTTP and WebSocket loaders decode outside their transaction.
2410 yield* createThread(laterThreadId);
2411
2412 assert.deepEqual(shellIds(yield* decode), [readThreadId]);
2413 assert.sameMembers(shellIds(yield* projectionStore.getShellSnapshot()), [
2414 readThreadId,
2415 laterThreadId,
2416 ]);
2417
2418 // No decoding under the read transaction: a payload that cannot decode
2419 // fails the returned step, not the read.
2420 const [stored] = yield* sql<{ readonly payload_json: string }>`
2421 SELECT payload_json FROM orchestration_v2_projection_threads
2422 WHERE thread_id = ${laterThreadId}
2423 `;
2424 const setPayload = (payload: string) =>
2425 sql`UPDATE orchestration_v2_projection_threads SET payload_json = ${payload}
2426 WHERE thread_id = ${laterThreadId}`;
2427 yield* setPayload("{}");
2428 const failure = yield* sql
2429 .withTransaction(projectionStore.readShellSnapshot())
2430 .pipe(
2431 Effect.flatMap(Effect.flip),
2432 Effect.ensuring(Effect.orDie(setPayload(stored!.payload_json))),
2433 );
2434 assert.strictEqual(failure._tag, "ProjectionStoreReadError");
2435 }),
2436 );
2437
2438 it.effect("reads only the requested project's threads into a scoped shell snapshot", () =>
2439 Effect.gen(function* () {
2440 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2441 const sql = yield* SqlClient.SqlClient;
2442 const { projectId, activeThreadId, forkThreadId, otherThreadId } =
2443 yield* assertProjectScopedShells("sql-project-shell");
2444
2445 // A thread in another project that cannot decode fails the unscoped
2446 // snapshot, so a scoped snapshot that succeeds never read it.
2447 const [stored] = yield* sql<{ readonly payload_json: string }>`
2448 SELECT payload_json FROM orchestration_v2_projection_threads
2449 WHERE thread_id = ${otherThreadId}
2450 `;
2451 const setPayload = (payload: string) =>
2452 sql`UPDATE orchestration_v2_projection_threads SET payload_json = ${payload}
2453 WHERE thread_id = ${otherThreadId}`;
2454 yield* setPayload("{}");
2455 const [unscoped, scoped] = yield* Effect.all([
2456 projectionStore.getShellSnapshot().pipe(Effect.flip),
2457 projectionStore.getShellSnapshot({ projectId, location: "active" }),
2458 ]).pipe(Effect.ensuring(Effect.orDie(setPayload(stored!.payload_json))));
2459 assert.strictEqual(unscoped._tag, "ProjectionStoreReadError");
2460 assert.sameMembers(
2461 scoped.threads.map((thread) => thread.id),
2462 [activeThreadId, forkThreadId],
2463 );
2464 }),
2465 );
2466
2467 it.effect("does not treat visited or marked-unread state as thread activity", () =>
2468 Effect.gen(function* () {
2469 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2470 const createdAt = yield* DateTime.now;
2471 const visitedOccurredAt = DateTime.add(createdAt, { seconds: 1 });
2472 const markedUnreadOccurredAt = DateTime.add(createdAt, { seconds: 2 });
2473 const threadId = ThreadId.make("thread:projection-read-state");
2474 const projectId = ProjectId.make("project:projection-read-state");
2475 const thread = {
2476 createdBy: "user" as const,
2477 creationSource: "web" as const,
2478 id: threadId,
2479 projectId,
2480 title: "Projection read state",
2481 providerInstanceId,
2482 modelSelection,
2483 runtimeMode: "full-access" as const,
2484 interactionMode: "default" as const,
2485 branch: null,
2486 worktreePath: null,
2487 activeProviderThreadId: null,
2488 lineage: {
2489 parentThreadId: null,
2490 relationshipToParent: null,
2491 rootThreadId: threadId,
2492 },
2493 forkedFrom: null,
2494 createdAt,
2495 updatedAt: createdAt,
2496 archivedAt: null,
2497 settledOverride: null,
2498 settledAt: null,
2499 lastVisitedAt: null,
2500 deletedAt: null,
2501 };
2502
2503 yield* projectionStore.apply({
2504 id: EventId.make("event:projection-read-state:created"),
2505 type: "thread.created",
2506 threadId,
2507 occurredAt: createdAt,
2508 payload: thread,
2509 });
2510 yield* projectionStore.apply({
2511 id: EventId.make("event:projection-read-state:visited"),
2512 type: "thread.visited",
2513 threadId,
2514 occurredAt: visitedOccurredAt,
2515 payload: { ...thread, lastVisitedAt: createdAt },
2516 });
2517
2518 const visited = yield* projectionStore.getThreadProjection(threadId);
2519 assert.deepEqual(visited.thread.lastVisitedAt, createdAt);
2520 assert.deepEqual(visited.thread.updatedAt, createdAt);
2521
2522 yield* projectionStore.apply({
2523 id: EventId.make("event:projection-read-state:marked-unread"),
2524 type: "thread.marked-unread",
2525 threadId,
2526 occurredAt: markedUnreadOccurredAt,
2527 payload: thread,
2528 });
2529
2530 const markedUnread = yield* projectionStore.getThreadProjection(threadId);
2531 assert.isNull(markedUnread.thread.lastVisitedAt);
2532 assert.deepEqual(markedUnread.thread.updatedAt, createdAt);
2533 }),
2534 );
2535
2536 it.effect("dates thread activity by a run event and leaves the rest of the thread alone", () =>
2537 Effect.gen(function* () {
2538 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2539 const createdAt = yield* DateTime.now;
2540 const runOccurredAt = DateTime.add(createdAt, { seconds: 5 });
2541 const threadId = ThreadId.make("thread:projection-activity");
2542 const runId = RunId.make("run:projection-activity");
2543 const rootNodeId = NodeId.make("node:projection-activity");
2544 const thread = {
2545 createdBy: "user" as const,
2546 creationSource: "web" as const,
2547 id: threadId,
2548 projectId: ProjectId.make("project:projection-activity"),
2549 title: "Projection activity",
2550 providerInstanceId,
2551 modelSelection,
2552 runtimeMode: "full-access" as const,
2553 interactionMode: "default" as const,
2554 branch: "feature/activity",
2555 worktreePath: null,
2556 activeProviderThreadId: null,
2557 lineage: {
2558 parentThreadId: null,
2559 relationshipToParent: null,
2560 rootThreadId: threadId,
2561 },
2562 forkedFrom: null,
2563 createdAt,
2564 updatedAt: createdAt,
2565 archivedAt: null,
2566 settledOverride: null,
2567 settledAt: null,
2568 lastVisitedAt: createdAt,
2569 deletedAt: null,
2570 };
2571
2572 yield* projectionStore.apply({
2573 id: EventId.make("event:projection-activity:created"),
2574 type: "thread.created",
2575 threadId,
2576 occurredAt: createdAt,
2577 payload: thread,
2578 });
2579 const created = (yield* projectionStore.getThreadProjection(threadId)).thread;
2580
2581 const { statements, tracer } = traceSqlStatements();
2582 yield* projectionStore
2583 .apply({
2584 id: EventId.make("event:projection-activity:run-created"),
2585 type: "run.created",
2586 threadId,
2587 runId,
2588 nodeId: rootNodeId,
2589 driver,
2590 providerInstanceId,
2591 occurredAt: runOccurredAt,
2592 payload: {
2593 id: runId,
2594 threadId,
2595 ordinal: 1,
2596 providerInstanceId,
2597 modelSelection,
2598 providerThreadId: null,
2599 userMessageId: MessageId.make("message:projection-activity"),
2600 rootNodeId,
2601 activeAttemptId: null,
2602 status: "running" as const,
2603 requestedAt: runOccurredAt,
2604 startedAt: runOccurredAt,
2605 completedAt: null,
2606 checkpointId: null,
2607 contextHandoffId: null,
2608 },
2609 })
2610 .pipe(Effect.withTracer(tracer));
2611 // The thread row is patched in place, never read back to be re-encoded.
2612 assert.isEmpty(
2613 statements.filter(
2614 (statement) =>
2615 /^\s*SELECT/i.test(statement) &&
2616 statement.includes("orchestration_v2_projection_threads"),
2617 ),
2618 );
2619
2620 const active = (yield* projectionStore.getThreadProjection(threadId)).thread;
2621 assert.deepEqual(active, { ...created, updatedAt: runOccurredAt });
2622 const shell = (yield* projectionStore.getShellSnapshot()).threads.find(
2623 (candidate) => candidate.id === threadId,
2624 );
2625 assert.deepEqual(shell?.updatedAt, runOccurredAt);
2626 }),
2627 );
2628
2629 it.effect("preserves delegated completion ownership across stale run and task updates", () =>
2630 Effect.gen(function* () {
2631 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2632 const now = yield* DateTime.now;
2633 const later = DateTime.add(now, { seconds: 1 });
2634 const threadId = ThreadId.make("thread:projection-delegated-completion");
2635 const projectId = ProjectId.make("project:projection-delegated-completion");
2636 const runId = RunId.make("run:projection-delegated-completion");
2637 const rootNodeId = NodeId.make("node:projection-delegated-completion-root");
2638 const taskId = NodeId.make("node:projection-delegated-completion-task");
2639 const thread = {
2640 createdBy: "user" as const,
2641 creationSource: "web" as const,
2642 id: threadId,
2643 projectId,
2644 title: "Delegated completion projection",
2645 providerInstanceId,
2646 modelSelection,
2647 runtimeMode: "full-access" as const,
2648 interactionMode: "default" as const,
2649 branch: null,
2650 worktreePath: null,
2651 activeProviderThreadId: null,
2652 lineage: {
2653 parentThreadId: null,
2654 relationshipToParent: null,
2655 rootThreadId: threadId,
2656 },
2657 forkedFrom: null,
2658 createdAt: now,
2659 updatedAt: now,
2660 archivedAt: null,
2661 settledOverride: null,
2662 settledAt: null,
2663 lastVisitedAt: null,
2664 deletedAt: null,
2665 };
2666 const run = {
2667 id: runId,
2668 threadId,
2669 ordinal: 1,
2670 providerInstanceId,
2671 modelSelection,
2672 providerThreadId: null,
2673 userMessageId: MessageId.make("message:projection-delegated-completion"),
2674 rootNodeId,
2675 activeAttemptId: null,
2676 status: "running" as const,
2677 requestedAt: now,
2678 startedAt: now,
2679 completedAt: null,
2680 checkpointId: null,
2681 contextHandoffId: null,
2682 delegatedCompletion: {
2683 disposition: "stopped" as const,
2684 nextGeneration: 2,
2685 delivery: null,
2686 },
2687 };
2688 const task = {
2689 id: taskId,
2690 threadId,
2691 runId,
2692 parentNodeId: rootNodeId,
2693 origin: "app_owned" as const,
2694 createdBy: "agent" as const,
2695 driver,
2696 providerInstanceId,
2697 providerThreadId: null,
2698 childThreadId: null,
2699 nativeTaskRef: null,
2700 prompt: "Inspect the stop barrier.",
2701 title: null,
2702 model: null,
2703 completionWake: "always" as const,
2704 completionDelivery: {
2705 state: "disposed" as const,
2706 observedByRunId: null,
2707 },
2708 status: "running" as const,
2709 result: null,
2710 startedAt: now,
2711 completedAt: null,
2712 updatedAt: now,
2713 };
2714
2715 yield* projectionStore.apply({
2716 id: EventId.make("event:projection-delegated-completion:thread"),
2717 type: "thread.created",
2718 threadId,
2719 occurredAt: now,
2720 payload: thread,
2721 });
2722 yield* projectionStore.apply({
2723 id: EventId.make("event:projection-delegated-completion:run"),
2724 type: "run.updated",
2725 threadId,
2726 runId,
2727 nodeId: rootNodeId,
2728 providerInstanceId,
2729 occurredAt: now,
2730 payload: run,
2731 });
2732 yield* projectionStore.apply({
2733 id: EventId.make("event:projection-delegated-completion:task"),
2734 type: "subagent.updated",
2735 threadId,
2736 runId,
2737 nodeId: taskId,
2738 driver,
2739 providerInstanceId,
2740 occurredAt: now,
2741 payload: task,
2742 });
2743
2744 const { delegatedCompletion: _delegatedCompletion, ...staleRun } = run;
2745 const { completionDelivery: _completionDelivery, ...staleTask } = task;
2746 yield* projectionStore.apply({
2747 id: EventId.make("event:projection-delegated-completion:stale-run"),
2748 type: "run.updated",
2749 threadId,
2750 runId,
2751 nodeId: rootNodeId,
2752 providerInstanceId,
2753 occurredAt: later,
2754 payload: { ...staleRun, status: "interrupted", completedAt: later },
2755 });
2756 yield* projectionStore.apply({
2757 id: EventId.make("event:projection-delegated-completion:stale-task"),
2758 type: "subagent.updated",
2759 threadId,
2760 runId,
2761 nodeId: taskId,
2762 driver,
2763 providerInstanceId,
2764 occurredAt: later,
2765 payload: {
2766 ...staleTask,
2767 status: "interrupted",
2768 completedAt: later,
2769 updatedAt: later,
2770 },
2771 });
2772
2773 const projection = yield* projectionStore.getThreadProjection(threadId);
2774 assert.deepEqual(projection.runs[0]?.delegatedCompletion, run.delegatedCompletion);
2775 assert.deepEqual(projection.subagents[0]?.completionDelivery, task.completionDelivery);
2776 }),
2777 );
2778
2779 it.effect("only exposes interruptible runs through the shell activeRunId", () =>
2780 Effect.gen(function* () {
2781 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2782 const now = yield* DateTime.now;
2783 const threadId = ThreadId.make("thread:projection-shell-interruptible");
2784 const projectId = ProjectId.make("project:projection-shell-interruptible");
2785 const runId = RunId.make("run:projection-shell-interruptible");
2786 const rootNodeId = NodeId.make("node:projection-shell-interruptible");
2787 const run = {
2788 id: runId,
2789 threadId,
2790 ordinal: 1,
2791 providerInstanceId,
2792 modelSelection,
2793 providerThreadId: null,
2794 userMessageId: MessageId.make("message:projection-shell-interruptible"),
2795 rootNodeId,
2796 activeAttemptId: null,
2797 status: "running" as const,
2798 requestedAt: now,
2799 startedAt: now,
2800 completedAt: null,
2801 checkpointId: null,
2802 contextHandoffId: null,
2803 };
2804
2805 yield* projectionStore.apply({
2806 id: EventId.make("event:projection-shell-interruptible:thread"),
2807 type: "thread.created",
2808 threadId,
2809 occurredAt: now,
2810 payload: {
2811 createdBy: "user",
2812 creationSource: "web",
2813 id: threadId,
2814 projectId,
2815 title: "Interruptible shell run",
2816 providerInstanceId,
2817 modelSelection,
2818 runtimeMode: "full-access",
2819 interactionMode: "default",
2820 branch: null,
2821 worktreePath: null,
2822 activeProviderThreadId: null,
2823 lineage: {
2824 parentThreadId: null,
2825 relationshipToParent: null,
2826 rootThreadId: threadId,
2827 },
2828 forkedFrom: null,
2829 createdAt: now,
2830 updatedAt: now,
2831 archivedAt: null,
2832 settledOverride: null,
2833 settledAt: null,
2834 lastVisitedAt: null,
2835 deletedAt: null,
2836 },
2837 });
2838 yield* projectionStore.apply({
2839 id: EventId.make("event:projection-shell-interruptible:running"),
2840 type: "run.created",
2841 threadId,
2842 runId,
2843 nodeId: rootNodeId,
2844 driver,
2845 occurredAt: now,
2846 payload: run,
2847 });
2848
2849 let shell = (yield* projectionStore.getShellSnapshot()).threads.find(
2850 (thread) => thread.id === threadId,
2851 );
2852 assert.equal(shell?.status, "running");
2853 assert.equal(shell?.activeRunId, runId);
2854 assert.equal(
2855 shell?.latestRunRequestedAt && DateTime.toEpochMillis(shell.latestRunRequestedAt),
2856 DateTime.toEpochMillis(now),
2857 );
2858 assert.equal(
2859 shell?.latestRunStartedAt && DateTime.toEpochMillis(shell.latestRunStartedAt),
2860 DateTime.toEpochMillis(now),
2861 );
2862 assert.isNull(shell?.latestRunCompletedAt);
2863
2864 yield* projectionStore.apply({
2865 id: EventId.make("event:projection-shell-interruptible:waiting"),
2866 type: "run.updated",
2867 threadId,
2868 runId,
2869 nodeId: rootNodeId,
2870 driver,
2871 occurredAt: now,
2872 payload: { ...run, status: "waiting" },
2873 });
2874
2875 shell = (yield* projectionStore.getShellSnapshot()).threads.find(
2876 (thread) => thread.id === threadId,
2877 );
2878 assert.equal(shell?.status, "waiting");
2879 assert.isNull(shell?.activeRunId);
2880 const later = DateTime.add(now, { hours: 1 });
2881 for (const status of ["queued", "cancelled"] as const) {
2882 yield* projectionStore.apply({
2883 id: EventId.make(`event:clock:newer:${status}`),
2884 type: "run.updated",
2885 threadId,
2886 occurredAt: later,
2887 payload: {
2888 ...run,
2889 id: RunId.make("run:clock:newer"),
2890 ordinal: 2,
2891 status,
2892 requestedAt: later,
2893 startedAt: null,
2894 completedAt: status === "cancelled" ? later : null,
2895 },
2896 });
2897 for (const activityStatus of ["preparing", "running", "waiting", "completed"] as const) {
2898 yield* projectionStore.apply({
2899 id: EventId.make(`event:clock:${status}:${activityStatus}`),
2900 type: "run.updated",
2901 threadId,
2902 occurredAt: later,
2903 payload: {
2904 ...run,
2905 status: activityStatus,
2906 startedAt: activityStatus === "preparing" ? null : now,
2907 completedAt: activityStatus === "completed" ? later : null,
2908 },
2909 });
2910 const projection = yield* projectionStore.getThreadProjection(threadId);
2911 const sqlShell = (yield* projectionStore.getShellSnapshot()).threads.find(
2912 (row) => row.id === threadId,
2913 )!;
2914 const memoryShell = ProjectionStore.threadShellFromProjection(projection);
2915 const expected = activityStatus === "completed" ? null : DateTime.toEpochMillis(now);
2916 const timestamp = (value: DateTime.Utc | null | undefined) =>
2917 value == null ? null : DateTime.toEpochMillis(value);
2918 assert.equal(timestamp(sqlShell.activityRunStartedAt), expected);
2919 assert.equal(timestamp(memoryShell.activityRunStartedAt), expected);
2920 assert.equal(sqlShell.latestRunId, "run:clock:newer");
2921 }
2922 }
2923
2924 // A held queue waits for the user, so both shells present the run before
2925 // it rather than reporting queued work.
2926 yield* projectionStore.apply({
2927 id: EventId.make("event:clock:newer:held"),
2928 type: "run.updated",
2929 threadId,
2930 occurredAt: later,
2931 payload: {
2932 ...run,
2933 id: RunId.make("run:clock:newer"),
2934 ordinal: 2,
2935 status: "queued",
2936 queueHeld: true,
2937 requestedAt: later,
2938 startedAt: null,
2939 completedAt: null,
2940 },
2941 });
2942 const heldProjection = yield* projectionStore.getThreadProjection(threadId);
2943 const heldSqlShell = (yield* projectionStore.getShellSnapshot()).threads.find(
2944 (row) => row.id === threadId,
2945 )!;
2946 for (const heldShell of [
2947 heldSqlShell,
2948 ProjectionStore.threadShellFromProjection(heldProjection),
2949 ]) {
2950 assert.equal(heldShell.latestRunId, runId);
2951 assert.equal(heldShell.status, "completed");
2952 }
2953
2954 // With only held runs, nothing has executed: both shells read idle.
2955 yield* projectionStore.apply({
2956 id: EventId.make("event:projection-shell-interruptible:held-first"),
2957 type: "run.updated",
2958 threadId,
2959 runId,
2960 nodeId: rootNodeId,
2961 driver,
2962 occurredAt: later,
2963 payload: { ...run, status: "queued", queueHeld: true, startedAt: null, completedAt: null },
2964 });
2965 const onlyHeldProjection = yield* projectionStore.getThreadProjection(threadId);
2966 const onlyHeldSqlShell = (yield* projectionStore.getShellSnapshot()).threads.find(
2967 (row) => row.id === threadId,
2968 )!;
2969 for (const onlyHeldShell of [
2970 onlyHeldSqlShell,
2971 ProjectionStore.threadShellFromProjection(onlyHeldProjection),
2972 ]) {
2973 assert.isNull(onlyHeldShell.latestRunId);
2974 assert.equal(onlyHeldShell.status, "idle");
2975 }
2976
2977 // A wake run counts from the start of the work it continues.
2978 yield* projectionStore.apply({
2979 id: EventId.make("event:projection-shell-interruptible:wake"),
2980 type: "run.updated",
2981 threadId,
2982 runId,
2983 nodeId: rootNodeId,
2984 driver,
2985 occurredAt: later,
2986 payload: {
2987 ...run,
2988 status: "running",
2989 requestedAt: later,
2990 startedAt: later,
2991 workStartedAt: now,
2992 },
2993 });
2994 const wakeProjection = yield* projectionStore.getThreadProjection(threadId);
2995 const wakeSqlShell = (yield* projectionStore.getShellSnapshot()).threads.find(
2996 (row) => row.id === threadId,
2997 )!;
2998 for (const wakeShell of [
2999 wakeSqlShell,
3000 ProjectionStore.threadShellFromProjection(wakeProjection),
3001 ]) {
3002 assert.equal(
3003 wakeShell.activityRunStartedAt && DateTime.toEpochMillis(wakeShell.activityRunStartedAt),
3004 DateTime.toEpochMillis(now),
3005 );
3006 }
3007 }),
3008 );
3009
3010 it.effect("selects only threads with runtime state that needs recovery", () =>
3011 Effect.gen(function* () {
3012 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3013 const now = yield* DateTime.now;
3014 const settledThreadId = ThreadId.make("thread:recovery-candidates:settled");
3015 const runningThreadId = ThreadId.make("thread:recovery-candidates:running");
3016 const memberThreadId = ThreadId.make("thread:recovery-candidates:member");
3017 const childThreadId = ThreadId.make("thread:recovery-candidates:child");
3018 const rolledBackThreadId = yield* addRolledBackRecoveryCandidate("recovery-candidates");
3019 const orphanedThreadId = yield* addOrphanedRecoveryCandidate("recovery-candidates");
3020 const projectId = ProjectId.make("project:recovery-candidates");
3021 const makeThread = (threadId: ThreadId) => ({
3022 createdBy: "user" as const,
3023 creationSource: "web" as const,
3024 id: threadId,
3025 projectId,
3026 title: "Recovery candidate",
3027 providerInstanceId,
3028 modelSelection,
3029 runtimeMode: "full-access" as const,
3030 interactionMode: "default" as const,
3031 branch: null,
3032 worktreePath: null,
3033 activeProviderThreadId: null,
3034 lineage: {
3035 parentThreadId: null,
3036 relationshipToParent: null,
3037 rootThreadId: threadId,
3038 },
3039 forkedFrom: null,
3040 createdAt: now,
3041 updatedAt: now,
3042 archivedAt: null,
3043 settledOverride: null,
3044 settledAt: null,
3045 lastVisitedAt: null,
3046 deletedAt: null,
3047 });
3048
3049 for (const threadId of [settledThreadId, runningThreadId, memberThreadId, childThreadId]) {
3050 yield* projectionStore.apply({
3051 id: EventId.make(`event:recovery-candidates:${threadId}:created`),
3052 type: "thread.created",
3053 threadId,
3054 occurredAt: now,
3055 payload: makeThread(threadId),
3056 });
3057 }
3058
3059 const runId = RunId.make("run:recovery-candidates:running");
3060 const rootNodeId = NodeId.make("node:recovery-candidates:running");
3061 yield* projectionStore.apply({
3062 id: EventId.make("event:recovery-candidates:run-created"),
3063 type: "run.created",
3064 threadId: runningThreadId,
3065 runId,
3066 nodeId: rootNodeId,
3067 driver,
3068 providerInstanceId,
3069 occurredAt: now,
3070 payload: {
3071 id: runId,
3072 threadId: runningThreadId,
3073 ordinal: 1,
3074 providerInstanceId,
3075 modelSelection,
3076 providerThreadId: null,
3077 userMessageId: MessageId.make("message:recovery-candidates:running"),
3078 rootNodeId,
3079 activeAttemptId: null,
3080 status: "running",
3081 requestedAt: now,
3082 startedAt: now,
3083 completedAt: null,
3084 checkpointId: null,
3085 contextHandoffId: null,
3086 },
3087 });
3088
3089 const memberId = NodeId.make("node:recovery-candidates:member");
3090 const childRootNodeId = NodeId.make("node:recovery-candidates:child-root");
3091 const nativeTaskRef = { driver, nativeId: "workflow:agent:1", strength: "strong" as const };
3092 const memberNode = {
3093 id: memberId,
3094 threadId: memberThreadId,
3095 runId: null,
3096 parentNodeId: NodeId.make("node:recovery-candidates:coordinator-root"),
3097 rootNodeId: NodeId.make("node:recovery-candidates:coordinator-root"),
3098 kind: "subagent" as const,
3099 status: "running" as const,
3100 countsForRun: false,
3101 providerThreadId: null,
3102 providerTurnId: null,
3103 nativeItemRef: nativeTaskRef,
3104 runtimeRequestId: null,
3105 checkpointScopeId: null,
3106 startedAt: now,
3107 completedAt: null,
3108 };
3109 yield* projectionStore.apply({
3110 id: EventId.make("event:recovery-candidates:member"),
3111 type: "subagent.updated",
3112 threadId: memberThreadId,
3113 nodeId: memberId,
3114 driver,
3115 providerInstanceId,
3116 occurredAt: now,
3117 payload: {
3118 id: memberId,
3119 threadId: memberThreadId,
3120 runId: null,
3121 parentNodeId: memberNode.parentNodeId,
3122 origin: "provider_native",
3123 createdBy: "agent",
3124 driver,
3125 providerInstanceId,
3126 providerThreadId: null,
3127 childThreadId,
3128 nativeTaskRef,
3129 prompt: "Do the work",
3130 title: "Member",
3131 model: null,
3132 status: "running",
3133 result: null,
3134 startedAt: now,
3135 completedAt: null,
3136 updatedAt: now,
3137 },
3138 });
3139 for (const [threadId, node] of [
3140 [memberThreadId, memberNode],
3141 [
3142 childThreadId,
3143 {
3144 ...memberNode,
3145 id: childRootNodeId,
3146 threadId: childThreadId,
3147 parentNodeId: null,
3148 rootNodeId: childRootNodeId,
3149 kind: "root_turn" as const,
3150 },
3151 ],
3152 ] as const) {
3153 yield* projectionStore.apply({
3154 id: EventId.make(`event:recovery-candidates:node:${node.id}`),
3155 type: "node.updated",
3156 threadId,
3157 nodeId: node.id,
3158 driver,
3159 occurredAt: now,
3160 payload: node,
3161 });
3162 }
3163
3164 const recoveryThreadIds = yield* projectionStore.getRecoveryThreadIds("runtime");
3165 assert.include(recoveryThreadIds, runningThreadId);
3166 assert.include(recoveryThreadIds, orphanedThreadId);
3167 assert.include(recoveryThreadIds, memberThreadId);
3168 assert.include(recoveryThreadIds, childThreadId);
3169 assert.notInclude(recoveryThreadIds, settledThreadId);
3170 assert.notInclude(recoveryThreadIds, rolledBackThreadId);
3171 const memberRecovery = yield* projectionStore.getRuntimeRecoveryProjection(memberThreadId);
3172 const childRecovery = yield* projectionStore.getRuntimeRecoveryProjection(childThreadId);
3173 assert.deepEqual(
3174 memberRecovery.subagents.map((subagent) => subagent.id),
3175 [memberId],
3176 );
3177 assert.deepEqual(
3178 memberRecovery.nodes.map((node) => node.id),
3179 [memberId],
3180 );
3181 assert.deepEqual(
3182 childRecovery.nodes.map((node) => node.id),
3183 [childRootNodeId],
3184 );
3185 yield* projectionStore.apply({
3186 id: EventId.make("event:recovery-candidates:member-cancelled"),
3187 type: "subagent.updated",
3188 threadId: memberThreadId,
3189 nodeId: memberId,
3190 driver,
3191 providerInstanceId,
3192 occurredAt: now,
3193 payload: {
3194 ...memberRecovery.subagents[0]!,
3195 status: "cancelled",
3196 completedAt: now,
3197 updatedAt: now,
3198 },
3199 });
3200 assert.include(yield* projectionStore.getRecoveryThreadIds("runtime"), childThreadId);
3201 }),
3202 );
3203
3204 it.effect("projects only the latest failed root turn's limit into SQL and memory shells", () =>
3205 Effect.gen(function* () {
3206 const store = yield* ProjectionStore.ProjectionStoreV2;
3207 const threadId = yield* addRolledBackRecoveryCandidate("limit-shell");
3208 const otherThreadId = yield* addRolledBackRecoveryCandidate("other-limit-shell");
3209 const original = (yield* store.getThreadProjection(threadId)).runs[0]!;
3210 const now = yield* DateTime.now;
3211 const limitItem = {
3212 id: TurnItemId.make("limit-shell:error"),
3213 threadId,
3214 runId: original.id,
3215 nodeId: original.rootNodeId,
3216 providerThreadId: null,
3217 providerTurnId: null,
3218 nativeItemRef: null,
3219 parentItemId: null,
3220 ordinal: 2,
3221 status: "failed" as const,
3222 title: "Usage limit reached",
3223 startedAt: now,
3224 completedAt: now,
3225 updatedAt: now,
3226 type: "error" as const,
3227 failure: {
3228 class: "usage_limit" as const,
3229 message: "Plan limit reached.",
3230 resetAt: "2099-01-01T00:00:00.000Z",
3231 code: "usageLimitExceeded",
3232 retryable: null,
3233 },
3234 };
3235 const applyRun = (status: typeof original.status, rootNodeId = original.rootNodeId) =>
3236 store.apply({
3237 id: EventId.make(`event:limit-shell:run:${status}:${rootNodeId}`),
3238 type: "run.updated",
3239 threadId,
3240 occurredAt: now,
3241 payload: { ...original, rootNodeId, status },
3242 });
3243 const assertSummary = Effect.fnUntraced(function* (
3244 lastError: string | null,
3245 lastErrorClass: string | null,
3246 ) {
3247 const projection = yield* store.getThreadProjection(threadId);
3248 const memoryShell = ProjectionStore.threadShellFromProjection(projection);
3249 const shells = yield* store.getShellSnapshot();
3250 const sqlShell = shells.threads.find((row) => row.id === threadId)!;
3251 for (const shell of [memoryShell, sqlShell]) {
3252 assert.equal(shell.lastError, lastError);
3253 assert.equal(shell.lastErrorClass, lastErrorClass);
3254 assert.equal(
3255 shell.usageLimitResetAt,
3256 lastErrorClass === "usage_limit" ? "2099-01-01T00:00:00.000Z" : null,
3257 );
3258 }
3259 assert.isNull(shells.threads.find((row) => row.id === otherThreadId)!.lastErrorClass);
3260 const candidates = yield* store.getLimitRecoveryCandidates({
3261 now,
3262 autoResume: true,
3263 snooze: false,
3264 });
3265 const candidate = candidates.find((row) => row.id === threadId);
3266 if (lastErrorClass === "usage_limit") {
3267 assert.deepEqual(candidate, {
3268 id: sqlShell.id,
3269 status: sqlShell.status,
3270 lastErrorClass: sqlShell.lastErrorClass,
3271 usageLimitResetAt: sqlShell.usageLimitResetAt,
3272 latestRunId: sqlShell.latestRunId,
3273 latestRunCompletedAt: sqlShell.latestRunCompletedAt,
3274 updatedAt: sqlShell.updatedAt,
3275 archivedAt: sqlShell.archivedAt,
3276 settledOverride: sqlShell.settledOverride,
3277 pendingRuntimeRequest: null,
3278 limitRecovery: sqlShell.limitRecovery,
3279 snoozedUntil: sqlShell.snoozedUntil,
3280 });
3281 } else assert.isUndefined(candidate);
3282 assert.isUndefined(candidates.find((row) => row.id === otherThreadId));
3283 });
3284 yield* store.apply({
3285 id: EventId.make("event:limit-shell:error"),
3286 type: "turn-item.updated",
3287 threadId,
3288 occurredAt: now,
3289 payload: limitItem,
3290 });
3291 yield* applyRun("failed");
3292 yield* assertSummary("Plan limit reached.", "usage_limit");
3293 const queuedRunId = RunId.make("run:limit-shell:queued");
3294 yield* store.apply({
3295 id: EventId.make("event:limit-shell:queued"),
3296 type: "run.created",
3297 threadId,
3298 runId: queuedRunId,
3299 nodeId: NodeId.make("node:limit-shell:queued"),
3300 driver,
3301 providerInstanceId,
3302 occurredAt: now,
3303 payload: {
3304 ...original,
3305 id: queuedRunId,
3306 ordinal: original.ordinal + 1,
3307 rootNodeId: NodeId.make("node:limit-shell:queued"),
3308 userMessageId: MessageId.make("message:limit-shell:queued"),
3309 status: "queued",
3310 startedAt: null,
3311 completedAt: null,
3312 },
3313 });
3314 yield* assertSummary("Plan limit reached.", "usage_limit");
3315 const queuedProjection = yield* store.getThreadProjection(threadId);
3316 const queuedMemoryShell = ProjectionStore.threadShellFromProjection(queuedProjection);
3317 const queuedSqlShell = (yield* store.getShellSnapshot()).threads.find(
3318 (row) => row.id === threadId,
3319 )!;
3320 assert.equal(queuedMemoryShell.status, "failed");
3321 assert.equal(queuedMemoryShell.latestRunId, original.id);
3322 assert.equal(queuedSqlShell.status, "failed");
3323 assert.equal(queuedSqlShell.latestRunId, original.id);
3324 assert.equal(queuedProjection.runs.find((run) => run.id === queuedRunId)?.status, "queued");
3325 const cancelledRunId = RunId.make("run:limit-shell:cancelled-queued");
3326 yield* store.apply({
3327 id: EventId.make("event:limit-shell:cancelled-queued"),
3328 type: "run.created",
3329 threadId,
3330 runId: cancelledRunId,
3331 nodeId: NodeId.make("node:limit-shell:cancelled-queued"),
3332 driver,
3333 providerInstanceId,
3334 occurredAt: now,
3335 payload: {
3336 ...original,
3337 id: cancelledRunId,
3338 ordinal: original.ordinal + 2,
3339 rootNodeId: NodeId.make("node:limit-shell:cancelled-queued"),
3340 userMessageId: MessageId.make("message:limit-shell:cancelled-queued"),
3341 status: "cancelled",
3342 startedAt: null,
3343 completedAt: now,
3344 },
3345 });
3346 yield* store.apply({
3347 id: EventId.make("event:limit-shell:cancelled-queued-message"),
3348 type: "turn-item.updated",
3349 threadId,
3350 runId: cancelledRunId,
3351 nodeId: NodeId.make("node:limit-shell:cancelled-queued"),
3352 driver,
3353 occurredAt: now,
3354 payload: {
3355 createdBy: "user",
3356 creationSource: "web",
3357 id: TurnItemId.make("limit-shell:cancelled-queued-message"),
3358 threadId,
3359 runId: cancelledRunId,
3360 nodeId: NodeId.make("node:limit-shell:cancelled-queued"),
3361 providerThreadId: null,
3362 providerTurnId: null,
3363 nativeItemRef: null,
3364 parentItemId: null,
3365 ordinal: 3,
3366 status: "completed",
3367 title: null,
3368 startedAt: now,
3369 completedAt: now,
3370 updatedAt: now,
3371 type: "user_message",
3372 messageId: MessageId.make("message:limit-shell:cancelled-queued"),
3373 inputIntent: "turn_start",
3374 text: "cancelled before the provider started",
3375 attachments: [],
3376 },
3377 });
3378 yield* assertSummary("Plan limit reached.", "usage_limit");
3379 // A restart continuation ran ahead of the held queue and ended before the
3380 // resumed failed run: the failure is still the latest executed run.
3381 const aheadRunId = RunId.make("run:limit-shell:ran-ahead");
3382 yield* store.apply({
3383 id: EventId.make("event:limit-shell:ran-ahead"),
3384 type: "run.created",
3385 threadId,
3386 runId: aheadRunId,
3387 nodeId: NodeId.make("node:limit-shell:ran-ahead"),
3388 driver,
3389 providerInstanceId,
3390 occurredAt: now,
3391 payload: {
3392 ...original,
3393 id: aheadRunId,
3394 ordinal: original.ordinal + 3,
3395 rootNodeId: NodeId.make("node:limit-shell:ran-ahead"),
3396 userMessageId: MessageId.make("message:limit-shell:ran-ahead"),
3397 status: "completed",
3398 startedAt: DateTime.subtract(now, { minutes: 10 }),
3399 completedAt: DateTime.subtract(now, { minutes: 5 }),
3400 },
3401 });
3402 yield* assertSummary("Plan limit reached.", "usage_limit");
3403 const sql = yield* SqlClient.SqlClient;
3404 // The rest of this case treats the failed run as the latest run.
3405 yield* sql`DELETE FROM orchestration_v2_projection_runs WHERE run_id = ${aheadRunId}`;
3406 yield* sql`DELETE FROM orchestration_v2_projection_runs WHERE run_id = ${queuedRunId}`;
3407 yield* assertSummary("Plan limit reached.", "usage_limit");
3408 yield* sql`DELETE FROM orchestration_v2_projection_runs WHERE run_id = ${cancelledRunId}`;
3409 const [originalRow] = yield* sql<{
3410 payload_json: string;
3411 }>`SELECT payload_json FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId}`;
3412 for (const [field, value] of [
3413 ["archivedAt", DateTime.formatIso(now)],
3414 ["settledOverride", "settled"],
3415 ]) {
3416 yield* sql`UPDATE orchestration_v2_projection_threads
3417 SET payload_json = json_set(payload_json, ${`$.${field}`}, ${value})
3418 WHERE thread_id = ${threadId}`;
3419 assert.isUndefined(
3420 (yield* store.getLimitRecoveryCandidates({ now, autoResume: true, snooze: false })).find(
3421 (row) => row.id === threadId,
3422 ),
3423 );
3424 yield* sql`UPDATE orchestration_v2_projection_threads
3425 SET payload_json = ${originalRow!.payload_json} WHERE thread_id = ${threadId}`;
3426 }
3427 yield* sql`UPDATE orchestration_v2_projection_threads SET deleted_at = ${DateTime.formatIso(now)} WHERE thread_id = ${threadId}`;
3428 assert.isUndefined(
3429 (yield* store.getLimitRecoveryCandidates({ now, autoResume: true, snooze: false })).find(
3430 (row) => row.id === threadId,
3431 ),
3432 );
3433 yield* sql`UPDATE orchestration_v2_projection_threads SET deleted_at = NULL WHERE thread_id = ${threadId}`;
3434 const recoveryOptions = { now, autoResume: false, snooze: false };
3435 assert.isUndefined(
3436 (yield* store.getLimitRecoveryCandidates(recoveryOptions)).find(
3437 (row) => row.id === threadId,
3438 ),
3439 );
3440 const reset = DateTime.makeUnsafe(limitItem.failure.resetAt);
3441 const recovery = {
3442 runId: original.id,
3443 resetAt: limitItem.failure.resetAt,
3444 autoResume: true,
3445 requestId: CommandId.make("recovery:choice"),
3446 };
3447 yield* sql`UPDATE orchestration_v2_projection_threads
3448 SET payload_json = json_set(payload_json, '$.limitRecovery', json(${encodeUnknownJsonString(recovery)}))
3449 WHERE thread_id = ${threadId}`;
3450 // Armed future retries need no state decoding until they become due.
3451 assert.isUndefined(
3452 (yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, autoResume: true })).find(
3453 (row) => row.id === threadId,
3454 ),
3455 );
3456 const due = (yield* store.getLimitRecoveryCandidates({
3457 ...recoveryOptions,
3458 now: reset,
3459 })).find((row) => row.id === threadId)!;
3460 assert.deepEqual(due.limitRecovery, recovery);
3461 yield* sql`INSERT INTO orchestration_v2_projection_runtime_requests
3462 (runtime_request_id, thread_id, node_id, kind, status, created_at, payload_json)
3463 VALUES ('limit-shell:pending-request', ${threadId}, ${original.rootNodeId}, 'approval', 'pending', ${DateTime.formatIso(now)}, '{}')`;
3464 assert.isUndefined(
3465 (yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, now: reset })).find(
3466 (row) => row.id === threadId,
3467 ),
3468 );
3469 yield* sql`DELETE FROM orchestration_v2_projection_runtime_requests WHERE runtime_request_id = 'limit-shell:pending-request'`;
3470 yield* sql`UPDATE orchestration_v2_projection_threads
3471 SET payload_json = json_set(payload_json, '$.snoozedUntil', ${DateTime.formatIso(DateTime.add(reset, { minutes: 1 }))})
3472 WHERE thread_id = ${threadId}`;
3473 assert.isUndefined(
3474 (yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, now: reset })).find(
3475 (row) => row.id === threadId,
3476 ),
3477 );
3478 yield* sql`UPDATE orchestration_v2_projection_threads
3479 SET payload_json = json_set(payload_json, '$.snoozedUntil', NULL, '$.limitRecovery.autoResume', json('false'))
3480 WHERE thread_id = ${threadId}`;
3481 assert.isUndefined(
3482 (yield* store.getLimitRecoveryCandidates({
3483 ...recoveryOptions,
3484 now: reset,
3485 autoResume: true,
3486 })).find((row) => row.id === threadId),
3487 );
3488 yield* sql`UPDATE orchestration_v2_projection_threads SET payload_json = ${originalRow!.payload_json} WHERE thread_id = ${threadId}`;
3489 const session = {
3490 id: ProviderSessionId.make("session:limit-shell:shared"),
3491 driver,
3492 providerInstanceId,
3493 status: "ready" as const,
3494 cwd: "/workspace",
3495 model: modelSelection.model,
3496 capabilities: CodexProviderCapabilitiesV2,
3497 createdAt: now,
3498 updatedAt: now,
3499 lastError: null,
3500 };
3501 for (const boundThreadId of [threadId, otherThreadId]) {
3502 yield* store.apply({
3503 id: EventId.make(`event:limit-shell:bind:${boundThreadId}`),
3504 type: "provider-session.attached",
3505 threadId: boundThreadId,
3506 driver,
3507 providerInstanceId,
3508 occurredAt: now,
3509 payload: session,
3510 });
3511 }
3512 yield* assertSummary("Plan limit reached.", "usage_limit");
3513 yield* store.apply({
3514 id: EventId.make("event:limit-shell:session-failed"),
3515 type: "provider-session.updated",
3516 threadId,
3517 occurredAt: now,
3518 payload: { ...session, status: "error", lastError: "Provider process exited." },
3519 });
3520 yield* assertSummary("Provider process exited.", null);
3521 yield* store.apply({
3522 id: EventId.make("event:limit-shell:session-recovered"),
3523 type: "provider-session.updated",
3524 threadId,
3525 occurredAt: now,
3526 payload: session,
3527 });
3528 yield* assertSummary("Plan limit reached.", "usage_limit");
3529 // A failed child is visible in history but does not replace the root's reason.
3530 yield* store.apply({
3531 id: EventId.make("event:limit-shell:child-error"),
3532 type: "turn-item.updated",
3533 threadId,
3534 occurredAt: now,
3535 payload: {
3536 ...limitItem,
3537 id: TurnItemId.make("limit-shell:child-error"),
3538 nodeId: NodeId.make("child-node"),
3539 ordinal: 3,
3540 failure: { ...limitItem.failure, class: "provider_error", message: "Child failed." },
3541 },
3542 });
3543 yield* assertSummary("Plan limit reached.", "usage_limit");
3544 // A later ordinary root error replaces the limit classification.
3545 yield* store.apply({
3546 id: EventId.make("event:limit-shell:replacement"),
3547 type: "turn-item.updated",
3548 threadId,
3549 occurredAt: now,
3550 payload: {
3551 ...limitItem,
3552 id: TurnItemId.make("limit-shell:replacement"),
3553 ordinal: 4,
3554 failure: { ...limitItem.failure, class: "provider_error", message: "Provider failed." },
3555 },
3556 });
3557 yield* assertSummary("Provider failed.", "provider_error");
3558 for (const status of [
3559 "running",
3560 "completed",
3561 "interrupted",
3562 "cancelled",
3563 "rolled_back",
3564 ] as const) {
3565 yield* applyRun(status);
3566 yield* assertSummary(null, null);
3567 }
3568 // A new attempt's root cannot inherit an earlier attempt's limit.
3569 yield* applyRun("failed", NodeId.make("new-attempt-root"));
3570 yield* assertSummary(null, null);
3571 }),
3572 );
3573
3574 it.effect("projects one shared provider session into multiple thread bindings", () =>
3575 Effect.gen(function* () {
3576 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3577 const now = yield* DateTime.now;
3578 const projectId = ProjectId.make("project:projection-shared-provider-session");
3579 const firstThreadId = ThreadId.make("thread:projection-shared-provider-session:first");
3580 const secondThreadId = ThreadId.make("thread:projection-shared-provider-session:second");
3581 const providerSessionId = ProviderSessionId.make(
3582 "provider-session:projection-shared-provider-session",
3583 );
3584 const makeThread = (threadId: ThreadId) => ({
3585 createdBy: "user" as const,
3586 creationSource: "web" as const,
3587 id: threadId,
3588 projectId,
3589 title: "Shared provider session",
3590 providerInstanceId,
3591 modelSelection,
3592 runtimeMode: "full-access" as const,
3593 interactionMode: "default" as const,
3594 branch: null,
3595 worktreePath: null,
3596 activeProviderThreadId: null,
3597 lineage: {
3598 parentThreadId: null,
3599 relationshipToParent: null,
3600 rootThreadId: threadId,
3601 },
3602 forkedFrom: null,
3603 createdAt: now,
3604 updatedAt: now,
3605 archivedAt: null,
3606 settledOverride: null,
3607 settledAt: null,
3608 lastVisitedAt: null,
3609 deletedAt: null,
3610 });
3611 const session = {
3612 id: providerSessionId,
3613 driver,
3614 providerInstanceId,
3615 status: "error" as const,
3616 cwd: "/workspace",
3617 model: modelSelection.model,
3618 capabilities: CodexProviderCapabilitiesV2,
3619 createdAt: now,
3620 updatedAt: now,
3621 lastError: "provider process exited",
3622 };
3623
3624 yield* projectionStore.apply({
3625 id: EventId.make("event:projection-shared-provider-session:first-thread"),
3626 type: "thread.created",
3627 threadId: firstThreadId,
3628 occurredAt: now,
3629 payload: makeThread(firstThreadId),
3630 });
3631 yield* projectionStore.apply({
3632 id: EventId.make("event:projection-shared-provider-session:second-thread"),
3633 type: "thread.created",
3634 threadId: secondThreadId,
3635 occurredAt: now,
3636 payload: makeThread(secondThreadId),
3637 });
3638 for (const [threadId, suffix] of [
3639 [firstThreadId, "first"],
3640 [secondThreadId, "second"],
3641 ] as const) {
3642 yield* projectionStore.apply({
3643 id: EventId.make(`event:projection-shared-provider-session:${suffix}-binding`),
3644 type: "provider-session.attached",
3645 threadId,
3646 driver,
3647 providerInstanceId,
3648 occurredAt: now,
3649 payload: session,
3650 });
3651 }
3652
3653 assert.deepEqual(
3654 (yield* projectionStore.getThreadProjection(firstThreadId)).providerSessions.map(
3655 (value) => value.id,
3656 ),
3657 [providerSessionId],
3658 );
3659 assert.deepEqual(
3660 (yield* projectionStore.getThreadProjection(secondThreadId)).providerSessions.map(
3661 (value) => value.id,
3662 ),
3663 [providerSessionId],
3664 );
3665 assert.deepEqual(
3666 (yield* projectionStore.getShellSnapshot()).threads
3667 .filter((thread) => thread.id === firstThreadId || thread.id === secondThreadId)
3668 .map((thread) => ({
3669 id: thread.id,
3670 lastError: thread.lastError,
3671 })),
3672 [
3673 { id: firstThreadId, lastError: "provider process exited" },
3674 { id: secondThreadId, lastError: "provider process exited" },
3675 ],
3676 );
3677
3678 yield* projectionStore.apply({
3679 id: EventId.make("event:projection-shared-provider-session:first-detached"),
3680 type: "provider-session.detached",
3681 threadId: firstThreadId,
3682 driver,
3683 providerInstanceId,
3684 occurredAt: now,
3685 payload: { providerSessionId, detachedAt: now },
3686 });
3687
3688 assert.lengthOf(
3689 (yield* projectionStore.getThreadProjection(firstThreadId)).providerSessions,
3690 0,
3691 );
3692 assert.lengthOf(
3693 (yield* projectionStore.getThreadProjection(secondThreadId)).providerSessions,
3694 1,
3695 );
3696 }),
3697 );
3698
3699 it.effect("looks up a thread's last provider error through its own session bindings", () =>
3700 Effect.gen(function* () {
3701 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3702 const sql = yield* SqlClient.SqlClient;
3703 const now = yield* DateTime.now;
3704 const later = DateTime.add(now, { minutes: 1 });
3705 const projectId = ProjectId.make("project:projection-session-error-lookup");
3706 const attachErroredSession = Effect.fnUntraced(function* (
3707 name: string,
3708 lastError: string,
3709 updatedAt: DateTime.Utc,
3710 ) {
3711 const threadId = ThreadId.make(`thread:projection-session-error-lookup:${name}`);
3712 yield* projectionStore.apply({
3713 id: EventId.make(`event:projection-session-error-lookup:${name}:thread`),
3714 type: "thread.created",
3715 threadId,
3716 occurredAt: now,
3717 payload: {
3718 createdBy: "user",
3719 creationSource: "web",
3720 id: threadId,
3721 projectId,
3722 title: name,
3723 providerInstanceId,
3724 modelSelection,
3725 runtimeMode: "full-access",
3726 interactionMode: "default",
3727 branch: null,
3728 worktreePath: null,
3729 activeProviderThreadId: null,
3730 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
3731 forkedFrom: null,
3732 createdAt: now,
3733 updatedAt: now,
3734 archivedAt: null,
3735 settledOverride: null,
3736 settledAt: null,
3737 lastVisitedAt: null,
3738 deletedAt: null,
3739 },
3740 });
3741 yield* projectionStore.apply({
3742 id: EventId.make(`event:projection-session-error-lookup:${name}:session`),
3743 type: "provider-session.attached",
3744 threadId,
3745 driver,
3746 providerInstanceId,
3747 occurredAt: updatedAt,
3748 payload: {
3749 id: ProviderSessionId.make(`provider-session:projection-session-error-lookup:${name}`),
3750 driver,
3751 providerInstanceId,
3752 status: "error",
3753 cwd: "/workspace",
3754 model: modelSelection.model,
3755 capabilities: CodexProviderCapabilitiesV2,
3756 createdAt: updatedAt,
3757 updatedAt,
3758 lastError,
3759 },
3760 });
3761 return threadId;
3762 });
3763
3764 const failure = {
3765 class: "usage_limit" as const,
3766 message: "Plan limit reached.",
3767 resetAt: "2099-01-01T00:00:00.000Z",
3768 code: "usageLimitExceeded",
3769 retryable: null,
3770 };
3771 const threadId = yield* attachErroredSession("own", failure.message, now);
3772 // The own session repeats the root turn's limit, so the thread stays a
3773 // recovery candidate only while the lookup reads that session.
3774 const runId = RunId.make("run:projection-session-error-lookup:own");
3775 const rootNodeId = NodeId.make("node:projection-session-error-lookup:own");
3776 yield* projectionStore.apply({
3777 id: EventId.make("event:projection-session-error-lookup:own:run"),
3778 type: "run.created",
3779 threadId,
3780 runId,
3781 nodeId: rootNodeId,
3782 driver,
3783 providerInstanceId,
3784 occurredAt: now,
3785 payload: {
3786 id: runId,
3787 threadId,
3788 ordinal: 1,
3789 providerInstanceId,
3790 modelSelection,
3791 providerThreadId: null,
3792 userMessageId: MessageId.make("message:projection-session-error-lookup:own"),
3793 rootNodeId,
3794 activeAttemptId: null,
3795 status: "failed",
3796 requestedAt: now,
3797 startedAt: now,
3798 completedAt: now,
3799 checkpointId: null,
3800 contextHandoffId: null,
3801 },
3802 });
3803 yield* projectionStore.apply({
3804 id: EventId.make("event:projection-session-error-lookup:own:error"),
3805 type: "turn-item.updated",
3806 threadId,
3807 occurredAt: now,
3808 payload: {
3809 id: TurnItemId.make("projection-session-error-lookup:own:error"),
3810 threadId,
3811 runId,
3812 nodeId: rootNodeId,
3813 providerThreadId: null,
3814 providerTurnId: null,
3815 nativeItemRef: null,
3816 parentItemId: null,
3817 ordinal: 1,
3818 status: "failed",
3819 title: "Usage limit reached",
3820 startedAt: now,
3821 completedAt: now,
3822 updatedAt: now,
3823 type: "error",
3824 failure,
3825 },
3826 });
3827 // Every thread on the provider instance has its own session, and these
3828 // are newer. They must neither leak into the thread nor be walked for it.
3829 for (let index = 0; index < 24; index++) {
3830 yield* attachErroredSession(`other-${index}`, "other session failed", later);
3831 }
3832
3833 const { statements, tracer } = traceSqlStatements();
3834 const shell = yield* projectionStore.getShellSnapshot().pipe(Effect.withTracer(tracer));
3835 assert.equal(
3836 shell.threads.find((thread) => thread.id === threadId)?.lastError,
3837 failure.message,
3838 );
3839 const candidates = yield* projectionStore
3840 .getLimitRecoveryCandidates({ now, autoResume: true, snooze: false })
3841 .pipe(Effect.withTracer(tracer));
3842 // Another thread's newer error would replace the limit and drop the candidate.
3843 const candidate = candidates.find((row) => row.id === threadId);
3844 assert.isDefined(candidate);
3845 assert.include(candidate, {
3846 status: "failed",
3847 lastErrorClass: "usage_limit",
3848 usageLimitResetAt: failure.resetAt,
3849 latestRunId: runId,
3850 });
3851
3852 // One provider instance usually holds most sessions. Driving the lookup
3853 // from sessions walks all of them for every thread in the snapshot.
3854 for (const marker of ["AS blocking_failure_payload_json", "AS failure_payload_json"]) {
3855 const statement = statements.find((query) => query.includes(marker));
3856 assert.isDefined(statement, marker);
3857 const plan = (yield* sql.unsafe<{ readonly detail: string }>(
3858 `EXPLAIN QUERY PLAN ${statement}`,
3859 )).map((row) => row.detail);
3860 const bindingLookup = plan.findIndex((detail) => detail.startsWith("SEARCH binding "));
3861 const sessionLookup = plan.findIndex((detail) => detail.startsWith("SEARCH session "));
3862 assert.isAtLeast(bindingLookup, 0, marker);
3863 assert.include(
3864 plan[bindingLookup],
3865 "orchestration_v2_projection_provider_session_bindings_thread_idx (thread_id=?)",
3866 );
3867 assert.isAbove(sessionLookup, bindingLookup, marker);
3868 assert.include(plan[sessionLookup], "(provider_session_id=?)");
3869 const errorLookups = plan.filter(
3870 (detail) =>
3871 /^SEARCH (?:item|error) /.test(detail) && !detail.includes("(turn_item_id=?)"),
3872 );
3873 assert.lengthOf(errorLookups, marker.includes("blocking") ? 2 : 1, marker);
3874 for (const errorLookup of errorLookups) {
3875 assert.include(errorLookup, "turn_items_failed_error_idx");
3876 assert.include(errorLookup, "(thread_id=? AND run_id=? AND node_id=?)");
3877 }
3878 }
3879 }),
3880 );
3881
3882 it.effect(
3883 "reads checkpoint context without decoding transcript or checkpoint file payloads",
3884 () =>
3885 Effect.gen(function* () {
3886 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3887 const sql = yield* SqlClient.SqlClient;
3888 const now = yield* DateTime.now;
3889 const threadId = ThreadId.make("thread:checkpoint-context");
3890 const runId = RunId.make("run:checkpoint-context");
3891 const nodeId = NodeId.make("node:checkpoint-context");
3892 const scopeId = CheckpointScopeId.make("scope:checkpoint-context");
3893 const checkpointId = CheckpointId.make("checkpoint:checkpoint-context");
3894 const ref = CheckpointRef.make("refs/t3/checkpoint-context/1");
3895 yield* projectionStore.apply({
3896 id: EventId.make("event:checkpoint-context:thread"),
3897 type: "thread.created",
3898 threadId,
3899 occurredAt: now,
3900 payload: {
3901 createdBy: "user",
3902 creationSource: "web",
3903 id: threadId,
3904 projectId: ProjectId.make("project:checkpoint-context"),
3905 title: "Checkpoint context",
3906 providerInstanceId,
3907 modelSelection,
3908 runtimeMode: "full-access",
3909 interactionMode: "default",
3910 branch: "feature",
3911 worktreePath: "/repo/worktree",
3912 activeProviderThreadId: null,
3913 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
3914 forkedFrom: null,
3915 createdAt: now,
3916 updatedAt: now,
3917 archivedAt: null,
3918 settledOverride: null,
3919 settledAt: null,
3920 lastVisitedAt: null,
3921 deletedAt: null,
3922 },
3923 });
3924 yield* projectionStore.apply({
3925 id: EventId.make("event:checkpoint-context:run"),
3926 type: "run.created",
3927 threadId,
3928 occurredAt: now,
3929 payload: {
3930 id: runId,
3931 threadId,
3932 ordinal: 1,
3933 providerInstanceId,
3934 modelSelection,
3935 providerThreadId: null,
3936 userMessageId: MessageId.make("message:checkpoint-context"),
3937 rootNodeId: nodeId,
3938 activeAttemptId: null,
3939 status: "completed",
3940 requestedAt: now,
3941 startedAt: now,
3942 completedAt: now,
3943 checkpointId,
3944 contextHandoffId: null,
3945 },
3946 });
3947 yield* projectionStore.apply({
3948 id: EventId.make("event:checkpoint-context:scope"),
3949 type: "checkpoint-scope.created",
3950 threadId,
3951 occurredAt: now,
3952 payload: {
3953 id: scopeId,
3954 threadId,
3955 runId,
3956 nodeId,
3957 parentScopeId: null,
3958 providerThreadId: null,
3959 kind: "root_run",
3960 ordinalWithinParent: 0,
3961 advancesAppRunCount: true,
3962 cwd: "/repo/worktree",
3963 createdAt: now,
3964 },
3965 });
3966 yield* projectionStore.apply({
3967 id: EventId.make("event:checkpoint-context:checkpoint"),
3968 type: "checkpoint.captured",
3969 threadId,
3970 occurredAt: now,
3971 payload: {
3972 id: checkpointId,
3973 threadId,
3974 scopeId,
3975 runId,
3976 nodeId,
3977 parentCheckpointId: null,
3978 ordinalWithinScope: 1,
3979 appRunOrdinal: 1,
3980 ref,
3981 status: "ready",
3982 files: [],
3983 capturedAt: now,
3984 },
3985 });
3986 // Old transcript shapes must not make a metadata-only diff unreadable.
3987 yield* sql`
3988 INSERT INTO orchestration_v2_projection_turn_items (
3989 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
3990 parent_item_id, ordinal, type, status, updated_at, payload_json
3991 ) VALUES (
3992 'turn-item:checkpoint-context:obsolete', ${threadId}, ${runId}, ${nodeId}, NULL, NULL,
3993 NULL, 1, 'assistant_message', 'completed', ${DateTime.formatIso(now)},
3994 ${encodeUnknownJsonString({ obsolete: "transcript shape" })}
3995 )
3996 `;
3997 assert.strictEqual(
3998 (yield* Effect.exit(projectionStore.getThreadProjection(threadId)))._tag,
3999 "Failure",
4000 );
4001 yield* sql`
4002 UPDATE orchestration_v2_projection_checkpoints
4003 SET payload_json = json_set(payload_json, '$.files', 'obsolete file summary')
4004 WHERE checkpoint_id = ${checkpointId}
4005 `;
4006 assert.deepEqual(yield* projectionStore.getCheckpointContext(threadId), {
4007 runs: [{ id: runId, ordinal: 1, status: "completed" }],
4008 checkpointScopes: [{ id: scopeId, runId, kind: "root_run", cwd: "/repo/worktree" }],
4009 checkpoints: [{ scopeId, runId, appRunOrdinal: 1, status: "ready", ref }],
4010 });
4011 const missing = yield* projectionStore
4012 .getCheckpointContext(ThreadId.make("thread:checkpoint-context:missing"))
4013 .pipe(Effect.flip);
4014 assert.instanceOf(missing, ProjectionStore.ProjectionStoreThreadNotFoundError);
4015 }),
4016 );
4017
4018 it.effect("selects the latest waiting secret only from active runs", () =>
4019 Effect.gen(function* () {
4020 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4021 const sql = yield* SqlClient.SqlClient;
4022 const suffix = "shell-pending-secret";
4023 const threadId = yield* addRolledBackRecoveryCandidate(suffix);
4024 const runId = RunId.make(`run:${suffix}:rolled-back`);
4025 const nodeId = NodeId.make(`node:${suffix}:rolled-back`);
4026 const now = yield* DateTime.now;
4027 const addSecret = (id: string, ordinal: number, status: "waiting" | "completed") =>
4028 projectionStore.apply({
4029 id: EventId.make(`event:${id}`),
4030 type: "turn-item.updated",
4031 threadId,
4032 runId,
4033 nodeId,
4034 driver,
4035 occurredAt: now,
4036 payload: {
4037 id: TurnItemId.make(id),
4038 threadId,
4039 runId,
4040 nodeId,
4041 providerThreadId: null,
4042 providerTurnId: null,
4043 nativeItemRef: null,
4044 parentItemId: null,
4045 ordinal,
4046 status,
4047 title: null,
4048 startedAt: now,
4049 completedAt: null,
4050 updatedAt: now,
4051 type: "secret_request",
4052 label: "Test credential",
4053 reason: "Test pending input",
4054 secretStatus: status === "waiting" ? "pending" : "saved",
4055 },
4056 });
4057 yield* addSecret("secret:a", 2, "waiting");
4058 yield* addSecret("secret:b", 3, "waiting");
4059 yield* addSecret("secret:c", 4, "completed");
4060
4061 for (const status of [
4062 "preparing",
4063 "starting",
4064 "running",
4065 "waiting",
4066 "completed",
4067 "rolled_back",
4068 ]) {
4069 yield* sql`
4070 UPDATE orchestration_v2_projection_runs
4071 SET status = ${status}, payload_json = json_set(payload_json, '$.status', ${status})
4072 WHERE run_id = ${runId}
4073 `;
4074 const shell = yield* projectionStore.getShellSnapshot();
4075 const thread = shell.threads.find((candidate) => candidate.id === threadId);
4076 assert.isDefined(thread);
4077 assert.equal(
4078 thread?.pendingRuntimeRequest?.id ?? null,
4079 status === "completed" || status === "rolled_back" ? null : "secret:b",
4080 );
4081 }
4082 }),
4083 );
4084
4085 it.effect("builds shell snapshots without decoding full turn item payloads", () =>
4086 Effect.gen(function* () {
4087 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4088 const sql = yield* SqlClient.SqlClient;
4089 const now = yield* DateTime.now;
4090 const nowIso = DateTime.formatIso(now);
4091 const threadId = ThreadId.make("thread:projection-shell-stale-item");
4092 const projectId = ProjectId.make("project:projection-shell");
4093
4094 yield* projectionStore.apply({
4095 id: EventId.make("event:projection-shell-thread-created"),
4096 type: "thread.created",
4097 threadId,
4098 occurredAt: now,
4099 payload: {
4100 createdBy: "user",
4101 creationSource: "web",
4102 id: threadId,
4103 projectId,
4104 title: "Projection shell",
4105 providerInstanceId,
4106 modelSelection: modelSelection,
4107 runtimeMode: "full-access",
4108 interactionMode: "default",
4109 branch: null,
4110 worktreePath: null,
4111 activeProviderThreadId: null,
4112 lineage: {
4113 parentThreadId: null,
4114 relationshipToParent: null,
4115 rootThreadId: threadId,
4116 },
4117 forkedFrom: null,
4118 createdAt: now,
4119 updatedAt: now,
4120 archivedAt: null,
4121 settledOverride: null,
4122 settledAt: null,
4123 lastVisitedAt: null,
4124 deletedAt: null,
4125 },
4126 });
4127
4128 yield* sql`
4129 INSERT INTO orchestration_v2_projection_turn_items (
4130 turn_item_id,
4131 thread_id,
4132 run_id,
4133 node_id,
4134 provider_thread_id,
4135 provider_turn_id,
4136 parent_item_id,
4137 ordinal,
4138 type,
4139 status,
4140 updated_at,
4141 payload_json
4142 )
4143 VALUES (
4144 ${"turn-item:stale-user-message"},
4145 ${threadId},
4146 ${null},
4147 ${null},
4148 ${null},
4149 ${null},
4150 ${null},
4151 ${0},
4152 ${"user_message"},
4153 ${"completed"},
4154 ${nowIso},
4155 ${encodeUnknownJsonString({
4156 id: "turn-item:stale-user-message",
4157 threadId,
4158 runId: null,
4159 nodeId: null,
4160 providerThreadId: null,
4161 providerTurnId: null,
4162 nativeItemRef: null,
4163 parentItemId: null,
4164 ordinal: 0,
4165 status: "completed",
4166 title: null,
4167 startedAt: nowIso,
4168 completedAt: nowIso,
4169 updatedAt: nowIso,
4170 type: "user_message",
4171 messageId: "message:stale-user-message",
4172 text: "stale user message",
4173 attachments: [],
4174 })}
4175 )
4176 `;
4177
4178 const shell = yield* projectionStore.getShellSnapshot();
4179 const fullProjectionExit = yield* Effect.exit(projectionStore.getThreadProjection(threadId));
4180
4181 assert.deepEqual(
4182 shell.threads
4183 .filter((thread) => thread.id === threadId)
4184 .map((thread) => ({
4185 id: thread.id,
4186 itemCount: thread.itemCount,
4187 visibleItemCount: thread.visibleItemCount,
4188 status: thread.status,
4189 })),
4190 [
4191 {
4192 id: threadId,
4193 itemCount: 1,
4194 visibleItemCount: 1,
4195 status: "idle",
4196 },
4197 ],
4198 );
4199 assert.equal(fullProjectionExit._tag, "Failure");
4200 }),
4201 );
4202
4203 it.effect("counts imported runless history inherited by fork shells", () =>
4204 Effect.gen(function* () {
4205 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4206 const now = yield* DateTime.now;
4207 const projectId = ProjectId.make("project:projection-imported-fork-shell");
4208 const sourceThreadId = ThreadId.make("thread:projection-imported-fork-shell:source");
4209 const targetThreadId = ThreadId.make("thread:projection-imported-fork-shell:target");
4210 const sourceRunId = RunId.make("run:projection-imported-fork-shell:source");
4211 const rootNodeId = NodeId.make("node:projection-imported-fork-shell:source");
4212
4213 yield* projectionStore.apply({
4214 id: EventId.make("event:projection-imported-fork-shell:source-thread"),
4215 type: "thread.created",
4216 threadId: sourceThreadId,
4217 occurredAt: now,
4218 payload: {
4219 createdBy: "system",
4220 creationSource: "server",
4221 id: sourceThreadId,
4222 projectId,
4223 title: "Imported fork source",
4224 providerInstanceId,
4225 modelSelection,
4226 runtimeMode: "full-access",
4227 interactionMode: "default",
4228 branch: null,
4229 worktreePath: null,
4230 activeProviderThreadId: null,
4231 historyOrigin: "v1_import",
4232 lineage: {
4233 parentThreadId: null,
4234 relationshipToParent: null,
4235 rootThreadId: sourceThreadId,
4236 },
4237 forkedFrom: null,
4238 createdAt: now,
4239 updatedAt: now,
4240 archivedAt: null,
4241 settledOverride: null,
4242 settledAt: null,
4243 lastVisitedAt: null,
4244 deletedAt: null,
4245 },
4246 });
4247 yield* projectionStore.apply({
4248 id: EventId.make("event:projection-imported-fork-shell:target-thread"),
4249 type: "thread.created",
4250 threadId: targetThreadId,
4251 occurredAt: now,
4252 payload: {
4253 createdBy: "user",
4254 creationSource: "web",
4255 id: targetThreadId,
4256 projectId,
4257 title: "Imported fork target",
4258 providerInstanceId,
4259 modelSelection,
4260 runtimeMode: "full-access",
4261 interactionMode: "default",
4262 branch: null,
4263 worktreePath: null,
4264 activeProviderThreadId: null,
4265 lineage: {
4266 parentThreadId: sourceThreadId,
4267 relationshipToParent: "fork",
4268 rootThreadId: sourceThreadId,
4269 },
4270 forkedFrom: {
4271 type: "run",
4272 threadId: sourceThreadId,
4273 runId: sourceRunId,
4274 },
4275 createdAt: now,
4276 updatedAt: now,
4277 archivedAt: null,
4278 settledOverride: null,
4279 settledAt: null,
4280 lastVisitedAt: null,
4281 deletedAt: null,
4282 },
4283 });
4284 yield* projectionStore.apply({
4285 id: EventId.make("event:projection-imported-fork-shell:source-run"),
4286 type: "run.created",
4287 threadId: sourceThreadId,
4288 runId: sourceRunId,
4289 nodeId: rootNodeId,
4290 driver,
4291 occurredAt: now,
4292 payload: {
4293 id: sourceRunId,
4294 threadId: sourceThreadId,
4295 ordinal: 1,
4296 providerInstanceId,
4297 modelSelection,
4298 providerThreadId: null,
4299 userMessageId: MessageId.make("message:projection-imported-fork-shell:run"),
4300 rootNodeId,
4301 activeAttemptId: null,
4302 status: "completed",
4303 requestedAt: now,
4304 startedAt: now,
4305 completedAt: now,
4306 checkpointId: null,
4307 contextHandoffId: null,
4308 },
4309 });
4310
4311 const applyAssistantItem = (suffix: string, runId: RunId | null, ordinal: number) =>
4312 projectionStore.apply({
4313 id: EventId.make(`event:projection-imported-fork-shell:item:${suffix}`),
4314 type: "turn-item.updated",
4315 threadId: sourceThreadId,
4316 ...(runId === null ? {} : { runId }),
4317 occurredAt: now,
4318 payload: {
4319 id: TurnItemId.make(`turn-item:projection-imported-fork-shell:${suffix}`),
4320 threadId: sourceThreadId,
4321 runId,
4322 nodeId: null,
4323 providerThreadId: null,
4324 providerTurnId: null,
4325 nativeItemRef: null,
4326 parentItemId: null,
4327 ordinal,
4328 status: "completed",
4329 title: null,
4330 startedAt: now,
4331 completedAt: now,
4332 updatedAt: now,
4333 type: "assistant_message",
4334 messageId: MessageId.make(`message:projection-imported-fork-shell:${suffix}`),
4335 text: suffix,
4336 streaming: false,
4337 },
4338 });
4339
4340 yield* applyAssistantItem("imported-one", null, 1);
4341 yield* applyAssistantItem("imported-two", null, 2);
4342 yield* applyAssistantItem("native-run", sourceRunId, 3);
4343
4344 const shell = yield* projectionStore.getShellSnapshot();
4345 const targetShell = shell.threads.find((thread) => thread.id === targetThreadId);
4346 const targetProjection = yield* projectionStore.getThreadProjection(targetThreadId);
4347
4348 assert.isDefined(targetShell);
4349 assert.equal(targetShell.itemCount, 0);
4350 assert.equal(targetShell.visibleItemCount, 4);
4351 assert.equal(targetProjection.visibleTurnItems.length, 4);
4352 }),
4353 );
4354
4355 it.effect("removes rolled back runs from the active visible projection", () =>
4356 Effect.gen(function* () {
4357 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4358 const now = yield* DateTime.now;
4359 const threadId = ThreadId.make("thread:projection-rollback-prune");
4360 const projectId = ProjectId.make("project:projection-rollback-prune");
4361 const runId = RunId.make("run:projection-rollback-prune");
4362 const attemptId = RunAttemptId.make("attempt:projection-rollback-prune");
4363 const rootNodeId = NodeId.make("node:projection-rollback-prune:root");
4364 const assistantNodeId = NodeId.make("node:projection-rollback-prune:assistant");
4365 const providerThreadId = ProviderThreadId.make("provider-thread:projection-rollback-prune");
4366 const providerTurnId = ProviderTurnId.make("provider-turn:projection-rollback-prune");
4367 const userMessageId = MessageId.make("message:projection-rollback-prune:user");
4368 const assistantMessageId = MessageId.make("message:projection-rollback-prune:assistant");
4369 const userTurnItemId = TurnItemId.make("turn-item:projection-rollback-prune:user");
4370 const assistantTurnItemId = TurnItemId.make("turn-item:projection-rollback-prune:assistant");
4371 const backgroundTurnItemId = TurnItemId.make(
4372 "turn-item:projection-rollback-prune:background",
4373 );
4374
4375 yield* projectionStore.apply({
4376 id: EventId.make("event:projection-rollback-prune:thread-created"),
4377 type: "thread.created",
4378 threadId,
4379 occurredAt: now,
4380 payload: {
4381 createdBy: "user",
4382 creationSource: "web",
4383 id: threadId,
4384 projectId,
4385 title: "Projection rollback prune",
4386 providerInstanceId,
4387 modelSelection,
4388 runtimeMode: "full-access",
4389 interactionMode: "default",
4390 branch: null,
4391 worktreePath: null,
4392 activeProviderThreadId: providerThreadId,
4393 lineage: {
4394 parentThreadId: null,
4395 relationshipToParent: null,
4396 rootThreadId: threadId,
4397 },
4398 forkedFrom: null,
4399 createdAt: now,
4400 updatedAt: now,
4401 archivedAt: null,
4402 settledOverride: null,
4403 settledAt: null,
4404 lastVisitedAt: null,
4405 deletedAt: null,
4406 },
4407 });
4408 yield* projectionStore.apply({
4409 id: EventId.make("event:projection-rollback-prune:provider-thread"),
4410 type: "provider-thread.updated",
4411 threadId,
4412 driver,
4413 occurredAt: now,
4414 payload: {
4415 id: providerThreadId,
4416 driver,
4417 providerInstanceId,
4418 providerSessionId: null,
4419 appThreadId: threadId,
4420 ownerNodeId: null,
4421 nativeThreadRef: null,
4422 nativeConversationHeadRef: null,
4423 status: "active",
4424 firstRunOrdinal: 1,
4425 lastRunOrdinal: 1,
4426 handoffIds: [],
4427 forkedFrom: null,
4428 createdAt: now,
4429 updatedAt: now,
4430 },
4431 });
4432 yield* projectionStore.apply({
4433 id: EventId.make("event:projection-rollback-prune:run-created"),
4434 type: "run.created",
4435 threadId,
4436 runId,
4437 nodeId: rootNodeId,
4438 driver,
4439 occurredAt: now,
4440 payload: {
4441 id: runId,
4442 threadId,
4443 ordinal: 1,
4444 providerInstanceId,
4445 modelSelection,
4446 providerThreadId,
4447 userMessageId,
4448 rootNodeId,
4449 activeAttemptId: attemptId,
4450 status: "completed",
4451 requestedAt: now,
4452 startedAt: now,
4453 completedAt: now,
4454 checkpointId: null,
4455 contextHandoffId: null,
4456 },
4457 });
4458 yield* projectionStore.apply({
4459 id: EventId.make("event:projection-rollback-prune:attempt-created"),
4460 type: "run-attempt.created",
4461 threadId,
4462 runId,
4463 nodeId: rootNodeId,
4464 driver,
4465 occurredAt: now,
4466 payload: {
4467 id: attemptId,
4468 runId,
4469 attemptOrdinal: 1,
4470 rootNodeId,
4471 providerInstanceId,
4472 providerThreadId,
4473 providerTurnId,
4474 reason: "initial",
4475 status: "completed",
4476 startedAt: now,
4477 completedAt: now,
4478 },
4479 });
4480 yield* projectionStore.apply({
4481 id: EventId.make("event:projection-rollback-prune:root-node"),
4482 type: "node.updated",
4483 threadId,
4484 runId,
4485 nodeId: rootNodeId,
4486 driver,
4487 occurredAt: now,
4488 payload: {
4489 id: rootNodeId,
4490 threadId,
4491 runId,
4492 parentNodeId: null,
4493 rootNodeId,
4494 kind: "root_turn",
4495 status: "completed",
4496 countsForRun: true,
4497 providerThreadId,
4498 providerTurnId: null,
4499 nativeItemRef: null,
4500 runtimeRequestId: null,
4501 checkpointScopeId: null,
4502 startedAt: now,
4503 completedAt: now,
4504 },
4505 });
4506 yield* projectionStore.apply({
4507 id: EventId.make("event:projection-rollback-prune:assistant-node"),
4508 type: "node.updated",
4509 threadId,
4510 runId,
4511 nodeId: assistantNodeId,
4512 driver,
4513 occurredAt: now,
4514 payload: {
4515 id: assistantNodeId,
4516 threadId,
4517 runId,
4518 parentNodeId: rootNodeId,
4519 rootNodeId,
4520 kind: "assistant_message",
4521 status: "completed",
4522 countsForRun: false,
4523 providerThreadId,
4524 providerTurnId,
4525 nativeItemRef: null,
4526 runtimeRequestId: null,
4527 checkpointScopeId: null,
4528 startedAt: now,
4529 completedAt: now,
4530 },
4531 });
4532 yield* projectionStore.apply({
4533 id: EventId.make("event:projection-rollback-prune:provider-turn"),
4534 type: "provider-turn.updated",
4535 threadId,
4536 runId,
4537 nodeId: rootNodeId,
4538 driver,
4539 occurredAt: now,
4540 payload: {
4541 id: providerTurnId,
4542 providerThreadId,
4543 nodeId: rootNodeId,
4544 runAttemptId: attemptId,
4545 nativeTurnRef: null,
4546 ordinal: 1,
4547 status: "completed",
4548 startedAt: now,
4549 completedAt: now,
4550 },
4551 });
4552 yield* projectionStore.apply({
4553 id: EventId.make("event:projection-rollback-prune:user-message"),
4554 type: "message.updated",
4555 threadId,
4556 runId,
4557 nodeId: rootNodeId,
4558 driver,
4559 occurredAt: now,
4560 payload: {
4561 createdBy: "user",
4562 creationSource: "web",
4563 id: userMessageId,
4564 threadId,
4565 runId,
4566 nodeId: rootNodeId,
4567 role: "user",
4568 text: "rolled back user",
4569 attachments: [],
4570 streaming: false,
4571 createdAt: now,
4572 updatedAt: now,
4573 },
4574 });
4575 yield* projectionStore.apply({
4576 id: EventId.make("event:projection-rollback-prune:assistant-message"),
4577 type: "message.updated",
4578 threadId,
4579 runId,
4580 nodeId: assistantNodeId,
4581 driver,
4582 occurredAt: now,
4583 payload: {
4584 createdBy: "agent",
4585 creationSource: "provider",
4586 id: assistantMessageId,
4587 threadId,
4588 runId,
4589 nodeId: assistantNodeId,
4590 role: "assistant",
4591 text: "rolled back assistant",
4592 attachments: [],
4593 streaming: false,
4594 createdAt: now,
4595 updatedAt: now,
4596 },
4597 });
4598 yield* projectionStore.apply({
4599 id: EventId.make("event:projection-rollback-prune:user-item"),
4600 type: "turn-item.updated",
4601 threadId,
4602 runId,
4603 nodeId: rootNodeId,
4604 driver,
4605 occurredAt: now,
4606 payload: {
4607 createdBy: "user",
4608 creationSource: "web",
4609 id: userTurnItemId,
4610 threadId,
4611 runId,
4612 nodeId: rootNodeId,
4613 providerThreadId,
4614 providerTurnId: null,
4615 nativeItemRef: null,
4616 parentItemId: null,
4617 ordinal: 100,
4618 status: "completed",
4619 title: null,
4620 startedAt: now,
4621 completedAt: now,
4622 updatedAt: now,
4623 type: "user_message",
4624 messageId: userMessageId,
4625 inputIntent: "turn_start",
4626 text: "rolled back user",
4627 attachments: [],
4628 },
4629 });
4630 yield* projectionStore.apply({
4631 id: EventId.make("event:projection-rollback-prune:assistant-item"),
4632 type: "turn-item.updated",
4633 threadId,
4634 runId,
4635 nodeId: assistantNodeId,
4636 driver,
4637 occurredAt: now,
4638 payload: {
4639 id: assistantTurnItemId,
4640 threadId,
4641 runId,
4642 nodeId: assistantNodeId,
4643 providerThreadId,
4644 providerTurnId,
4645 nativeItemRef: null,
4646 parentItemId: null,
4647 ordinal: 101,
4648 status: "completed",
4649 title: null,
4650 startedAt: now,
4651 completedAt: now,
4652 updatedAt: now,
4653 type: "assistant_message",
4654 messageId: assistantMessageId,
4655 text: "rolled back assistant",
4656 streaming: false,
4657 },
4658 });
4659 yield* projectionStore.apply({
4660 id: EventId.make("event:projection-rollback-prune:background-item"),
4661 type: "turn-item.updated",
4662 threadId,
4663 runId,
4664 nodeId: rootNodeId,
4665 driver,
4666 occurredAt: now,
4667 payload: {
4668 id: backgroundTurnItemId,
4669 threadId,
4670 runId,
4671 nodeId: rootNodeId,
4672 providerThreadId,
4673 providerTurnId: null,
4674 nativeItemRef: null,
4675 parentItemId: null,
4676 ordinal: 300,
4677 status: "running",
4678 title: "rolled back background command",
4679 startedAt: now,
4680 completedAt: null,
4681 updatedAt: now,
4682 type: "command_execution",
4683 input: "sleep 60",
4684 },
4685 });
4686 yield* projectionStore.apply({
4687 id: EventId.make("event:projection-rollback-prune:run-rolled-back"),
4688 type: "run.updated",
4689 threadId,
4690 runId,
4691 nodeId: rootNodeId,
4692 driver,
4693 occurredAt: now,
4694 payload: {
4695 id: runId,
4696 threadId,
4697 ordinal: 1,
4698 providerInstanceId,
4699 modelSelection,
4700 providerThreadId,
4701 userMessageId,
4702 rootNodeId,
4703 activeAttemptId: attemptId,
4704 status: "rolled_back",
4705 requestedAt: now,
4706 startedAt: now,
4707 completedAt: now,
4708 checkpointId: null,
4709 contextHandoffId: null,
4710 },
4711 });
4712 yield* projectionStore.apply({
4713 id: EventId.make("event:projection-rollback-prune:root-rolled-back"),
4714 type: "node.updated",
4715 threadId,
4716 runId,
4717 nodeId: rootNodeId,
4718 driver,
4719 occurredAt: now,
4720 payload: {
4721 id: rootNodeId,
4722 threadId,
4723 runId,
4724 parentNodeId: null,
4725 rootNodeId,
4726 kind: "root_turn",
4727 status: "rolled_back",
4728 countsForRun: true,
4729 providerThreadId,
4730 providerTurnId: null,
4731 nativeItemRef: null,
4732 runtimeRequestId: null,
4733 checkpointScopeId: null,
4734 startedAt: now,
4735 completedAt: now,
4736 },
4737 });
4738
4739 const projection = yield* projectionStore.getThreadProjection(threadId);
4740
4741 assert.deepEqual(
4742 projection.runs.map((run) => run.status),
4743 ["rolled_back"],
4744 );
4745 assert.deepEqual(
4746 projection.nodes.map((node) => [node.id, node.status]),
4747 [
4748 [assistantNodeId, "completed"],
4749 [rootNodeId, "rolled_back"],
4750 ],
4751 );
4752 assert.lengthOf(projection.providerTurns, 1);
4753 assert.lengthOf(projection.messages, 2);
4754 assert.lengthOf(projection.turnItems, 3);
4755 assert.lengthOf(projection.visibleTurnItems, 0);
4756
4757 // A rolled-back run's background item is abandoned, not pending. The
4758 // shell must not report it as Waiting, or the sidebar shows Waiting for
4759 // work nothing will ever finish.
4760 const shell = yield* projectionStore.getShellSnapshot();
4761 const rolledBackShellThread = shell.threads.find((entry) => entry.id === threadId);
4762 assert.isDefined(rolledBackShellThread);
4763 assert.isNull(rolledBackShellThread.latestVisibleMessage);
4764 assert.deepEqual(rolledBackShellThread?.pendingBackgroundTasks ?? [], []);
4765 }),
4766 );
4767
4768 it.effect("keeps fork visible items stable after a source run is rolled back", () =>
4769 Effect.gen(function* () {
4770 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4771 const sql = yield* SqlClient.SqlClient;
4772 const now = yield* DateTime.now;
4773 const projectId = ProjectId.make("project:projection-fork-source-rollback");
4774 const sourceThreadId = ThreadId.make("thread:projection-fork-source-rollback:source");
4775 const targetThreadId = ThreadId.make("thread:projection-fork-source-rollback:target");
4776 const sourceProviderThreadId = ProviderThreadId.make(
4777 "provider-thread:projection-fork-source-rollback:source",
4778 );
4779 const targetProviderThreadId = ProviderThreadId.make(
4780 "provider-thread:projection-fork-source-rollback:target",
4781 );
4782 const sourceRun1Id = RunId.make("run:projection-fork-source-rollback:source:1");
4783 const sourceRun2Id = RunId.make("run:projection-fork-source-rollback:source:2");
4784 const sourceRun3Id = RunId.make("run:projection-fork-source-rollback:source:3");
4785 const sourceRun1NodeId = NodeId.make("node:projection-fork-source-rollback:source:1");
4786 const sourceRun2NodeId = NodeId.make("node:projection-fork-source-rollback:source:2");
4787 const sourceRun3NodeId = NodeId.make("node:projection-fork-source-rollback:source:3");
4788
4789 yield* projectionStore.apply({
4790 id: EventId.make("event:projection-fork-source-rollback:source-thread"),
4791 type: "thread.created",
4792 threadId: sourceThreadId,
4793 occurredAt: now,
4794 payload: {
4795 createdBy: "user",
4796 creationSource: "web",
4797 id: sourceThreadId,
4798 projectId,
4799 title: "Projection fork source rollback source",
4800 providerInstanceId,
4801 modelSelection,
4802 runtimeMode: "full-access",
4803 interactionMode: "default",
4804 branch: null,
4805 worktreePath: null,
4806 activeProviderThreadId: sourceProviderThreadId,
4807 lineage: {
4808 parentThreadId: null,
4809 relationshipToParent: null,
4810 rootThreadId: sourceThreadId,
4811 },
4812 forkedFrom: null,
4813 createdAt: now,
4814 updatedAt: now,
4815 archivedAt: null,
4816 settledOverride: null,
4817 settledAt: null,
4818 lastVisitedAt: null,
4819 deletedAt: null,
4820 },
4821 });
4822 yield* projectionStore.apply({
4823 id: EventId.make("event:projection-fork-source-rollback:target-thread"),
4824 type: "thread.created",
4825 threadId: targetThreadId,
4826 occurredAt: now,
4827 payload: {
4828 createdBy: "user",
4829 creationSource: "web",
4830 id: targetThreadId,
4831 projectId,
4832 title: "Projection fork source rollback target",
4833 providerInstanceId,
4834 modelSelection,
4835 runtimeMode: "full-access",
4836 interactionMode: "default",
4837 branch: null,
4838 worktreePath: null,
4839 activeProviderThreadId: targetProviderThreadId,
4840 lineage: {
4841 parentThreadId: sourceThreadId,
4842 relationshipToParent: "fork",
4843 rootThreadId: sourceThreadId,
4844 },
4845 forkedFrom: {
4846 type: "run",
4847 threadId: sourceThreadId,
4848 runId: sourceRun2Id,
4849 },
4850 createdAt: now,
4851 updatedAt: now,
4852 archivedAt: null,
4853 settledOverride: null,
4854 settledAt: null,
4855 lastVisitedAt: null,
4856 deletedAt: null,
4857 },
4858 });
4859
4860 for (const [ordinal, runId, nodeId, promptText, responseText] of [
4861 [1, sourceRun1Id, sourceRun1NodeId, "source one", "one"],
4862 [2, sourceRun2Id, sourceRun2NodeId, "source two", "two"],
4863 ] as const) {
4864 yield* projectionStore.apply({
4865 id: EventId.make(`event:projection-fork-source-rollback:run-${ordinal}`),
4866 type: "run.created",
4867 threadId: sourceThreadId,
4868 runId,
4869 nodeId,
4870 driver,
4871 occurredAt: now,
4872 payload: {
4873 id: runId,
4874 threadId: sourceThreadId,
4875 ordinal,
4876 providerInstanceId,
4877 modelSelection,
4878 providerThreadId: sourceProviderThreadId,
4879 userMessageId: MessageId.make(
4880 `message:projection-fork-source-rollback:user:${ordinal}`,
4881 ),
4882 rootNodeId: nodeId,
4883 activeAttemptId: null,
4884 status: "completed",
4885 requestedAt: now,
4886 startedAt: now,
4887 completedAt: now,
4888 checkpointId: null,
4889 contextHandoffId: null,
4890 },
4891 });
4892 yield* projectionStore.apply({
4893 id: EventId.make(`event:projection-fork-source-rollback:user-item-${ordinal}`),
4894 type: "turn-item.updated",
4895 threadId: sourceThreadId,
4896 runId,
4897 nodeId,
4898 driver,
4899 occurredAt: now,
4900 payload: {
4901 createdBy: "user",
4902 creationSource: "web",
4903 id: TurnItemId.make(`turn-item:projection-fork-source-rollback:user:${ordinal}`),
4904 threadId: sourceThreadId,
4905 runId,
4906 nodeId,
4907 providerThreadId: sourceProviderThreadId,
4908 providerTurnId: null,
4909 nativeItemRef: null,
4910 parentItemId: null,
4911 ordinal: ordinal * 100,
4912 status: "completed",
4913 title: null,
4914 startedAt: now,
4915 completedAt: now,
4916 updatedAt: now,
4917 type: "user_message",
4918 messageId: MessageId.make(`message:projection-fork-source-rollback:user:${ordinal}`),
4919 inputIntent: "turn_start",
4920 text: promptText,
4921 attachments: [],
4922 },
4923 });
4924 yield* projectionStore.apply({
4925 id: EventId.make(`event:projection-fork-source-rollback:assistant-item-${ordinal}`),
4926 type: "turn-item.updated",
4927 threadId: sourceThreadId,
4928 runId,
4929 nodeId,
4930 driver,
4931 occurredAt: now,
4932 payload: {
4933 id: TurnItemId.make(`turn-item:projection-fork-source-rollback:assistant:${ordinal}`),
4934 threadId: sourceThreadId,
4935 runId,
4936 nodeId,
4937 providerThreadId: sourceProviderThreadId,
4938 providerTurnId: null,
4939 nativeItemRef: null,
4940 parentItemId: null,
4941 ordinal: ordinal * 100 + 1,
4942 status: "completed",
4943 title: null,
4944 startedAt: now,
4945 completedAt: now,
4946 updatedAt: now,
4947 type: "assistant_message",
4948 messageId: MessageId.make(
4949 `message:projection-fork-source-rollback:assistant:${ordinal}`,
4950 ),
4951 text: responseText,
4952 streaming: false,
4953 },
4954 });
4955 }
4956
4957 const targetBeforeRollback = yield* projectionStore.getThreadProjection(targetThreadId);
4958 assert.deepEqual(
4959 targetBeforeRollback.visibleTurnItems.map((row) => row.item.type),
4960 ["user_message", "assistant_message", "user_message", "assistant_message", "fork"],
4961 );
4962
4963 yield* projectionStore.apply({
4964 id: EventId.make("event:projection-fork-source-rollback:run-2-rolled-back"),
4965 type: "run.updated",
4966 threadId: sourceThreadId,
4967 runId: sourceRun2Id,
4968 nodeId: sourceRun2NodeId,
4969 driver,
4970 occurredAt: now,
4971 payload: {
4972 id: sourceRun2Id,
4973 threadId: sourceThreadId,
4974 ordinal: 2,
4975 providerInstanceId,
4976 modelSelection,
4977 providerThreadId: sourceProviderThreadId,
4978 userMessageId: MessageId.make("message:projection-fork-source-rollback:user:2"),
4979 rootNodeId: sourceRun2NodeId,
4980 activeAttemptId: null,
4981 status: "rolled_back",
4982 requestedAt: now,
4983 startedAt: now,
4984 completedAt: now,
4985 checkpointId: null,
4986 contextHandoffId: null,
4987 },
4988 });
4989
4990 const sourceMessages = (yield* projectionStore.getThreadRecords(sourceThreadId, ["messages"]))
4991 .messages;
4992 assert.deepEqual(
4993 sourceMessages,
4994 (yield* projectionStore.getThreadProjection(sourceThreadId)).messages,
4995 );
4996 const targetAfterRollback = yield* projectionStore.getThreadProjection(targetThreadId);
4997 // Both shell reads drop the rolled-back run from the source count, while
4998 // the fork keeps the prefix it inherited.
4999 const shellSnapshot = yield* projectionStore.getShellSnapshot();
5000 for (const shell of [
5001 yield* projectionStore.getThreadShell(sourceThreadId),
5002 shellSnapshot.threads.find((thread) => thread.id === sourceThreadId),
5003 ]) {
5004 assert.equal(shell?.itemCount, 2);
5005 }
5006 for (const shell of [
5007 yield* projectionStore.getThreadShell(targetThreadId),
5008 shellSnapshot.threads.find((thread) => thread.id === targetThreadId),
5009 ]) {
5010 assert.equal(shell?.visibleItemCount, targetAfterRollback.visibleTurnItems.length);
5011 }
5012 const forwardPage = yield* projectionStore.getTimelinePage(targetThreadId, {
5013 view: "activity",
5014 limit: 2,
5015 });
5016 assert.deepEqual(forwardPage.items, targetAfterRollback.visibleTurnItems.slice(0, 2));
5017 assert.equal(forwardPage.totalItems, targetAfterRollback.visibleTurnItems.length);
5018 assert.isTrue(forwardPage.hasMore);
5019 const followingPage = yield* projectionStore.getTimelinePage(targetThreadId, {
5020 view: "activity",
5021 limit: 10,
5022 afterPosition: 1,
5023 });
5024 assert.deepEqual(followingPage.items, targetAfterRollback.visibleTurnItems.slice(2));
5025 assert.isFalse(followingPage.hasMore);
5026
5027 assert.deepEqual(
5028 targetAfterRollback.visibleTurnItems.map((row) => [
5029 row.visibility,
5030 row.item.type,
5031 row.item.type === "user_message" || row.item.type === "assistant_message"
5032 ? row.item.text
5033 : row.item.title,
5034 ]),
5035 [
5036 ["inherited", "user_message", "source one"],
5037 ["inherited", "assistant_message", "one"],
5038 ["inherited", "user_message", "source two"],
5039 ["inherited", "assistant_message", "two"],
5040 ["synthetic", "fork", "Forked from conversation"],
5041 ],
5042 );
5043
5044 yield* projectionStore.apply({
5045 id: EventId.make("event:projection-fork-source-rollback:run-3"),
5046 type: "run.updated",
5047 threadId: sourceThreadId,
5048 runId: sourceRun3Id,
5049 nodeId: sourceRun3NodeId,
5050 driver,
5051 occurredAt: now,
5052 payload: {
5053 id: sourceRun3Id,
5054 threadId: sourceThreadId,
5055 ordinal: 4,
5056 providerInstanceId,
5057 modelSelection,
5058 providerThreadId: sourceProviderThreadId,
5059 userMessageId: MessageId.make("message:projection-fork-source-rollback:user:3"),
5060 rootNodeId: sourceRun3NodeId,
5061 activeAttemptId: null,
5062 status: "completed",
5063 requestedAt: now,
5064 startedAt: now,
5065 completedAt: now,
5066 checkpointId: null,
5067 contextHandoffId: null,
5068 },
5069 });
5070 const nowIso = DateTime.formatIso(now);
5071 for (let index = 0; index < 300; index += 1) {
5072 const id = `turn-item:projection-fork-source-rollback:post-fork:${index}`;
5073 const ordinal = index < 10 ? 190 + index : 300 + index;
5074 yield* sql`
5075 INSERT INTO orchestration_v2_projection_turn_items (
5076 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
5077 parent_item_id, ordinal, type, status, updated_at, payload_json
5078 ) VALUES (
5079 ${id}, ${sourceThreadId}, ${sourceRun3Id}, ${sourceRun3NodeId},
5080 ${sourceProviderThreadId}, NULL, NULL, ${ordinal}, 'command_execution',
5081 'completed', ${nowIso}, ${encodeUnknownJsonString({
5082 id,
5083 threadId: sourceThreadId,
5084 runId: sourceRun3Id,
5085 nodeId: sourceRun3NodeId,
5086 providerThreadId: sourceProviderThreadId,
5087 providerTurnId: null,
5088 nativeItemRef: null,
5089 parentItemId: null,
5090 ordinal,
5091 status: "completed",
5092 title: "post fork",
5093 input: "echo later",
5094 output: "later",
5095 exitCode: 0,
5096 startedAt: nowIso,
5097 completedAt: nowIso,
5098 updatedAt: nowIso,
5099 type: "command_execution",
5100 })}
5101 )
5102 `;
5103 }
5104
5105 const { statements, tracer } = traceSqlStatements();
5106 const boundedFork = yield* projectionStore
5107 .getThreadSnapshotWindow(targetThreadId, { rowLimit: 2 })
5108 .pipe(Effect.withTracer(tracer));
5109 // Only the target's control state is hydrated. Fork ancestors contribute
5110 // timeline rows, runs and attempts, regardless of their own live work.
5111 assert.lengthOf(
5112 statements.filter((statement) => statement.includes("WITH RECURSIVE retained")),
5113 1,
5114 );
5115 assert.lengthOf(
5116 statements.filter((statement) => statement.includes("SELECT node.payload_json")),
5117 1,
5118 );
5119 assert.lengthOf(boundedFork.projection.turnItems, 0);
5120 assert.deepEqual(
5121 boundedFork.projection.visibleTurnItems.map((row) => [row.visibility, row.item.type]),
5122 [
5123 ["inherited", "user_message"],
5124 ["inherited", "assistant_message"],
5125 ["synthetic", "fork"],
5126 ],
5127 );
5128 const parentPage = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
5129 rowLimit: 2,
5130 anchorItemId: TurnItemId.make("turn-item:projection-fork-source-rollback:assistant:1"),
5131 });
5132 assert.deepEqual(
5133 parentPage.projection.visibleTurnItems.map((row) =>
5134 row.item.type === "user_message" || row.item.type === "assistant_message"
5135 ? row.item.text
5136 : row.item.type,
5137 ),
5138 ["source one", "one", "fork"],
5139 );
5140
5141 const emptyBoundaryRunId = RunId.make(
5142 "run:projection-fork-source-rollback:source:empty-boundary",
5143 );
5144 yield* projectionStore.apply({
5145 id: EventId.make("event:projection-fork-source-rollback:empty-boundary-run"),
5146 type: "run.updated",
5147 threadId: sourceThreadId,
5148 runId: emptyBoundaryRunId,
5149 nodeId: NodeId.make("node:projection-fork-source-rollback:source:empty-boundary"),
5150 driver,
5151 occurredAt: now,
5152 payload: {
5153 id: emptyBoundaryRunId,
5154 threadId: sourceThreadId,
5155 ordinal: 3,
5156 providerInstanceId,
5157 modelSelection,
5158 providerThreadId: sourceProviderThreadId,
5159 userMessageId: MessageId.make(
5160 "message:projection-fork-source-rollback:user:empty-boundary",
5161 ),
5162 rootNodeId: NodeId.make("node:projection-fork-source-rollback:source:empty-boundary"),
5163 activeAttemptId: null,
5164 status: "completed",
5165 requestedAt: now,
5166 startedAt: now,
5167 completedAt: now,
5168 checkpointId: null,
5169 contextHandoffId: null,
5170 },
5171 });
5172 yield* sql`
5173 UPDATE orchestration_v2_projection_threads
5174 SET payload_json = json_set(payload_json, '$.forkedFrom.runId', ${emptyBoundaryRunId})
5175 WHERE thread_id = ${targetThreadId}
5176 `;
5177 const emptyBoundaryFork = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
5178 rowLimit: 2,
5179 });
5180 assert.deepEqual(
5181 emptyBoundaryFork.projection.visibleTurnItems.map((row) =>
5182 row.item.type === "user_message" || row.item.type === "assistant_message"
5183 ? row.item.text
5184 : row.item.type,
5185 ),
5186 ["source two", "two", "fork"],
5187 );
5188 yield* sql`
5189 UPDATE orchestration_v2_projection_threads
5190 SET payload_json = json_set(payload_json, '$.forkedFrom.runId', ${sourceRun2Id})
5191 WHERE thread_id = ${targetThreadId}
5192 `;
5193
5194 const targetRunId = RunId.make("run:projection-fork-source-rollback:target:1");
5195 const targetNodeId = NodeId.make("node:projection-fork-source-rollback:target:1");
5196 yield* projectionStore.apply({
5197 id: EventId.make("event:projection-fork-source-rollback:target-run"),
5198 type: "run.updated",
5199 threadId: targetThreadId,
5200 runId: targetRunId,
5201 nodeId: targetNodeId,
5202 driver,
5203 occurredAt: now,
5204 payload: {
5205 id: targetRunId,
5206 threadId: targetThreadId,
5207 ordinal: 1,
5208 providerInstanceId,
5209 modelSelection,
5210 providerThreadId: targetProviderThreadId,
5211 userMessageId: MessageId.make("message:projection-fork-source-rollback:target:1"),
5212 rootNodeId: targetNodeId,
5213 activeAttemptId: null,
5214 status: "completed",
5215 requestedAt: now,
5216 startedAt: now,
5217 completedAt: now,
5218 checkpointId: null,
5219 contextHandoffId: null,
5220 },
5221 });
5222 for (let ordinal = 1; ordinal <= 102; ordinal += 1) {
5223 const id = `turn-item:projection-fork-source-rollback:target:${ordinal}`;
5224 yield* sql`
5225 INSERT INTO orchestration_v2_projection_turn_items (
5226 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
5227 parent_item_id, ordinal, type, status, updated_at, payload_json
5228 ) VALUES (
5229 ${id}, ${targetThreadId}, ${targetRunId}, ${targetNodeId}, ${targetProviderThreadId},
5230 NULL, NULL, ${ordinal}, 'command_execution', 'completed', ${nowIso},
5231 ${encodeUnknownJsonString({
5232 id,
5233 threadId: targetThreadId,
5234 runId: targetRunId,
5235 nodeId: targetNodeId,
5236 providerThreadId: targetProviderThreadId,
5237 providerTurnId: null,
5238 nativeItemRef: null,
5239 parentItemId: null,
5240 ordinal,
5241 status: "completed",
5242 title: `target ${ordinal}`,
5243 input: `echo target ${ordinal}`,
5244 output: "target",
5245 exitCode: 0,
5246 startedAt: nowIso,
5247 completedAt: nowIso,
5248 updatedAt: nowIso,
5249 type: "command_execution",
5250 })}
5251 )
5252 `;
5253 }
5254 const nestedThreadId = ThreadId.make("thread:projection-fork-source-rollback:nested");
5255 for (const ordinal of [
5256 ...Array.from({ length: 99 }, (_, index) => index + 1),
5257 ...Array.from({ length: 99 }, (_, index) => index + 101),
5258 ]) {
5259 const runId = ordinal < 100 ? sourceRun1Id : sourceRun2Id;
5260 const id = `turn-item:projection-fork-source-rollback:lineage:${ordinal}`;
5261 yield* sql`
5262 INSERT INTO orchestration_v2_projection_turn_items (
5263 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
5264 parent_item_id, ordinal, type, status, updated_at, payload_json
5265 ) VALUES (
5266 ${id}, ${sourceThreadId}, ${runId}, NULL, NULL, NULL, NULL, ${ordinal},
5267 'command_execution', 'completed', ${nowIso},
5268 ${encodeUnknownJsonString({
5269 id,
5270 threadId: sourceThreadId,
5271 runId,
5272 nodeId: null,
5273 providerThreadId: null,
5274 providerTurnId: null,
5275 nativeItemRef: null,
5276 parentItemId: null,
5277 ordinal,
5278 status: "completed",
5279 title: `lineage ${ordinal}`,
5280 input: "echo lineage",
5281 output: "lineage",
5282 exitCode: 0,
5283 startedAt: nowIso,
5284 completedAt: nowIso,
5285 updatedAt: nowIso,
5286 type: "command_execution",
5287 })}
5288 )
5289 `;
5290 }
5291 const inheritedSupersededNodeId = NodeId.make(
5292 "node:projection-fork-source-rollback:inherited-superseded",
5293 );
5294 yield* sql`
5295 INSERT INTO orchestration_v2_projection_run_attempts (
5296 attempt_id, thread_id, run_id, attempt_ordinal, root_node_id, provider,
5297 provider_instance_id, provider_thread_id, provider_turn_id, status, payload_json
5298 ) VALUES (
5299 'attempt:projection-fork-source-rollback:inherited-superseded',
5300 ${sourceThreadId}, ${sourceRun2Id}, 2, ${inheritedSupersededNodeId}, 'codex',
5301 ${providerInstanceId}, ${sourceProviderThreadId}, NULL, 'superseded',
5302 ${encodeUnknownJsonString({
5303 id: "attempt:projection-fork-source-rollback:inherited-superseded",
5304 runId: sourceRun2Id,
5305 attemptOrdinal: 2,
5306 rootNodeId: inheritedSupersededNodeId,
5307 providerInstanceId,
5308 providerThreadId: sourceProviderThreadId,
5309 providerTurnId: null,
5310 reason: "retry",
5311 status: "superseded",
5312 startedAt: nowIso,
5313 completedAt: nowIso,
5314 })}
5315 )
5316 `;
5317 for (let ordinal = 201; ordinal <= 300; ordinal += 1) {
5318 const id = `turn-item:projection-fork-source-rollback:hidden-interrupt:${ordinal}`;
5319 yield* sql`
5320 INSERT INTO orchestration_v2_projection_turn_items (
5321 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
5322 parent_item_id, ordinal, type, status, updated_at, payload_json
5323 ) VALUES (
5324 ${id}, ${sourceThreadId}, ${sourceRun2Id}, ${inheritedSupersededNodeId},
5325 ${sourceProviderThreadId}, NULL, NULL, ${ordinal}, 'run_interrupt_result',
5326 'completed', ${nowIso},
5327 ${encodeUnknownJsonString({
5328 id,
5329 threadId: sourceThreadId,
5330 runId: sourceRun2Id,
5331 nodeId: inheritedSupersededNodeId,
5332 providerThreadId: sourceProviderThreadId,
5333 providerTurnId: null,
5334 nativeItemRef: null,
5335 parentItemId: null,
5336 ordinal,
5337 status: "completed",
5338 title: "Stopped",
5339 message: "Stopped",
5340 startedAt: nowIso,
5341 completedAt: nowIso,
5342 updatedAt: nowIso,
5343 type: "run_interrupt_result",
5344 })}
5345 )
5346 `;
5347 }
5348 yield* projectionStore.apply({
5349 id: EventId.make("event:projection-fork-source-rollback:nested-thread"),
5350 type: "thread.created",
5351 threadId: nestedThreadId,
5352 occurredAt: now,
5353 payload: {
5354 createdBy: "user",
5355 creationSource: "web",
5356 id: nestedThreadId,
5357 projectId,
5358 title: "Nested bounded fork",
5359 providerInstanceId,
5360 modelSelection,
5361 runtimeMode: "full-access",
5362 interactionMode: "default",
5363 branch: null,
5364 worktreePath: null,
5365 activeProviderThreadId: null,
5366 lineage: {
5367 parentThreadId: targetThreadId,
5368 relationshipToParent: "fork",
5369 rootThreadId: sourceThreadId,
5370 },
5371 forkedFrom: { type: "run", threadId: targetThreadId, runId: targetRunId },
5372 createdAt: now,
5373 updatedAt: now,
5374 archivedAt: null,
5375 settledOverride: null,
5376 settledAt: null,
5377 lastVisitedAt: null,
5378 deletedAt: null,
5379 },
5380 });
5381 const boundedNested = yield* projectionStore.getThreadSnapshotWindow(nestedThreadId, {
5382 rowLimit: 2,
5383 });
5384 assert.deepEqual(
5385 boundedNested.projection.visibleTurnItems.map((row) => row.item.type),
5386 [
5387 "user_message",
5388 "assistant_message",
5389 "fork",
5390 "command_execution",
5391 "command_execution",
5392 "fork",
5393 ],
5394 );
5395
5396 const nestedInitial = buildBoundedThreadProjection({
5397 policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
5398 projection: yield* projectionStore
5399 .getThreadSnapshotWindow(nestedThreadId, {
5400 rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2,
5401 })
5402 .pipe(Effect.map((snapshot) => snapshot.projection)),
5403 snapshotSequence: 1,
5404 });
5405 const expectedNestedIds = (yield* projectionStore.getThreadProjection(
5406 nestedThreadId,
5407 )).visibleTurnItems.map((row) => String(row.sourceItemId));
5408 const nestedIds = nestedInitial.projection.visibleTurnItems.map((row) =>
5409 String(row.sourceItemId),
5410 );
5411 let nestedCursor = nestedInitial.historyCursor;
5412 let nestedPages = 1;
5413 while (nestedCursor !== null) {
5414 const decoded = decodeThreadHistoryCursor(nestedCursor);
5415 const snapshot = yield* projectionStore.getThreadSnapshotWindow(nestedThreadId, {
5416 rowLimit: THREAD_HISTORY_PAGE_POLICY.maxItems + 2,
5417 anchorItemId: TurnItemId.make(decoded.si),
5418 anchorThreadId: ThreadId.make(decoded.st),
5419 });
5420 const page = selectHistoryPageFromCursor({
5421 policy: { ...THREAD_HISTORY_PAGE_POLICY, maxUserTurns: undefined },
5422 items: snapshot.projection.visibleTurnItems,
5423 cursor: nestedCursor,
5424 snapshotSequence: snapshot.snapshotSequence,
5425 });
5426 nestedIds.unshift(...page.items.map((row) => String(row.sourceItemId)));
5427 nestedCursor = page.nextCursor;
5428 nestedPages += 1;
5429 }
5430 assert.isAtLeast(nestedPages, 3);
5431 assert.lengthOf(nestedIds, expectedNestedIds.length);
5432 assert.deepEqual(nestedIds, expectedNestedIds);
5433
5434 const nestedTurnWindow = yield* projectionStore.getThreadSnapshotWindow(nestedThreadId, {
5435 rowLimit: 77,
5436 userTurnLimit: 10,
5437 });
5438 const nestedTurnPage = buildBoundedThreadProjection({
5439 projection: nestedTurnWindow.projection,
5440 snapshotSequence: 1,
5441 });
5442 const turnPagedIds = nestedTurnPage.projection.visibleTurnItems.map((row) =>
5443 String(row.sourceItemId),
5444 );
5445 let turnCursor = nestedTurnPage.historyCursor;
5446 while (turnCursor !== null) {
5447 const anchor = decodeThreadHistoryCursor(turnCursor);
5448 const snapshot = yield* projectionStore.getThreadSnapshotWindow(nestedThreadId, {
5449 rowLimit: 77,
5450 userTurnLimit: 20,
5451 anchorItemId: TurnItemId.make(anchor.si),
5452 anchorThreadId: ThreadId.make(anchor.st),
5453 });
5454 const page = selectHistoryPageFromCursor({
5455 items: snapshot.projection.visibleTurnItems,
5456 cursor: turnCursor,
5457 snapshotSequence: 1,
5458 });
5459 turnPagedIds.unshift(...page.items.map((row) => String(row.sourceItemId)));
5460 turnCursor = page.nextCursor;
5461 }
5462 assert.deepEqual(turnPagedIds, expectedNestedIds);
5463 const fullNested = yield* projectionStore.getThreadProjection(nestedThreadId);
5464 const nestedForward = yield* projectionStore.getTimelinePage(nestedThreadId, {
5465 view: "activity",
5466 limit: expectedNestedIds.length + 1,
5467 });
5468 assert.deepEqual(nestedForward.items, fullNested.visibleTurnItems);
5469 const messagePage = yield* projectionStore.getTimelinePage(nestedThreadId, {
5470 view: "messages",
5471 limit: 2,
5472 });
5473 assert.deepEqual(
5474 messagePage.items,
5475 fullNested.visibleTurnItems
5476 .filter((row) =>
5477 ["user_message", "assistant_message", "proposed_plan"].includes(row.item.type),
5478 )
5479 .slice(0, 2),
5480 );
5481
5482 const emptyMiddleThreadId = ThreadId.make(
5483 "thread:projection-fork-source-rollback:empty-middle",
5484 );
5485 const emptyMiddleRunId = RunId.make("run:projection-fork-source-rollback:empty-middle");
5486 yield* projectionStore.apply({
5487 id: EventId.make("event:projection-fork-source-rollback:empty-middle-thread"),
5488 type: "thread.created",
5489 threadId: emptyMiddleThreadId,
5490 occurredAt: now,
5491 payload: {
5492 createdBy: "user",
5493 creationSource: "web",
5494 id: emptyMiddleThreadId,
5495 projectId,
5496 title: "Empty middle fork",
5497 providerInstanceId,
5498 modelSelection,
5499 runtimeMode: "full-access",
5500 interactionMode: "default",
5501 branch: null,
5502 worktreePath: null,
5503 activeProviderThreadId: null,
5504 lineage: {
5505 parentThreadId: sourceThreadId,
5506 relationshipToParent: "fork",
5507 rootThreadId: sourceThreadId,
5508 },
5509 forkedFrom: { type: "run", threadId: sourceThreadId, runId: sourceRun2Id },
5510 createdAt: now,
5511 updatedAt: now,
5512 archivedAt: null,
5513 settledOverride: null,
5514 settledAt: null,
5515 lastVisitedAt: null,
5516 deletedAt: null,
5517 },
5518 });
5519 yield* projectionStore.apply({
5520 id: EventId.make("event:projection-fork-source-rollback:empty-middle-run"),
5521 type: "run.updated",
5522 threadId: emptyMiddleThreadId,
5523 runId: emptyMiddleRunId,
5524 nodeId: NodeId.make("node:projection-fork-source-rollback:empty-middle"),
5525 driver,
5526 occurredAt: now,
5527 payload: {
5528 id: emptyMiddleRunId,
5529 threadId: emptyMiddleThreadId,
5530 ordinal: 1,
5531 providerInstanceId,
5532 modelSelection,
5533 providerThreadId: null,
5534 userMessageId: MessageId.make("message:projection-fork-source-rollback:empty-middle"),
5535 rootNodeId: NodeId.make("node:projection-fork-source-rollback:empty-middle"),
5536 activeAttemptId: null,
5537 status: "completed",
5538 requestedAt: now,
5539 startedAt: now,
5540 completedAt: now,
5541 checkpointId: null,
5542 contextHandoffId: null,
5543 },
5544 });
5545 const emptyLeafThreadId = ThreadId.make("thread:projection-fork-source-rollback:empty-leaf");
5546 yield* projectionStore.apply({
5547 id: EventId.make("event:projection-fork-source-rollback:empty-leaf-thread"),
5548 type: "thread.created",
5549 threadId: emptyLeafThreadId,
5550 occurredAt: now,
5551 payload: {
5552 createdBy: "user",
5553 creationSource: "web",
5554 id: emptyLeafThreadId,
5555 projectId,
5556 title: "Leaf after empty fork",
5557 providerInstanceId,
5558 modelSelection,
5559 runtimeMode: "full-access",
5560 interactionMode: "default",
5561 branch: null,
5562 worktreePath: null,
5563 activeProviderThreadId: null,
5564 lineage: {
5565 parentThreadId: emptyMiddleThreadId,
5566 relationshipToParent: "fork",
5567 rootThreadId: sourceThreadId,
5568 },
5569 forkedFrom: {
5570 type: "run",
5571 threadId: emptyMiddleThreadId,
5572 runId: emptyMiddleRunId,
5573 },
5574 createdAt: now,
5575 updatedAt: now,
5576 archivedAt: null,
5577 settledOverride: null,
5578 settledAt: null,
5579 lastVisitedAt: null,
5580 deletedAt: null,
5581 },
5582 });
5583 const emptyMiddleSnapshot = yield* projectionStore.getThreadSnapshotWindow(
5584 emptyLeafThreadId,
5585 { rowLimit: 2 },
5586 );
5587 assert.deepEqual(
5588 emptyMiddleSnapshot.projection.visibleTurnItems.map((row) => [
5589 row.sourceThreadId,
5590 row.item.type,
5591 ]),
5592 [
5593 [sourceThreadId, "user_message"],
5594 [sourceThreadId, "assistant_message"],
5595 [sourceThreadId, "fork"],
5596 [emptyMiddleThreadId, "fork"],
5597 ],
5598 );
5599
5600 yield* sql`
5601 DELETE FROM orchestration_v2_projection_turn_items
5602 WHERE run_id IN (${sourceRun1Id}, ${sourceRun2Id})
5603 `;
5604 const importedItemId = "turn-item:projection-fork-source-rollback:legacy-import";
5605 yield* sql`
5606 INSERT INTO orchestration_v2_projection_turn_items (
5607 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
5608 parent_item_id, ordinal, type, status, updated_at, payload_json
5609 ) VALUES (
5610 ${importedItemId}, ${sourceThreadId}, NULL, NULL, NULL, NULL, NULL, 50,
5611 'assistant_message', 'completed', ${nowIso}, ${encodeUnknownJsonString({
5612 id: importedItemId,
5613 threadId: sourceThreadId,
5614 runId: null,
5615 nodeId: null,
5616 providerThreadId: null,
5617 providerTurnId: null,
5618 nativeItemRef: null,
5619 parentItemId: null,
5620 ordinal: 50,
5621 status: "completed",
5622 title: null,
5623 startedAt: nowIso,
5624 completedAt: nowIso,
5625 updatedAt: nowIso,
5626 type: "assistant_message",
5627 messageId: MessageId.make("message:projection-fork-source-rollback:legacy-import"),
5628 text: "legacy import before empty fork",
5629 streaming: false,
5630 historyOrigin: "v1_import",
5631 })}
5632 )
5633 `;
5634 yield* sql`
5635 UPDATE orchestration_v2_projection_threads
5636 SET payload_json = json_set(payload_json, '$.historyOrigin', 'v1_import')
5637 WHERE thread_id = ${sourceThreadId}
5638 `;
5639 yield* sql`
5640 UPDATE orchestration_v2_projection_threads
5641 SET payload_json = json_set(payload_json, '$.forkedFrom.runId', ${emptyBoundaryRunId})
5642 WHERE thread_id = ${targetThreadId}
5643 `;
5644 const importedEmptyBoundary = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
5645 rowLimit: 2,
5646 });
5647 assert.deepEqual(
5648 importedEmptyBoundary.projection.visibleTurnItems.map((row) => [
5649 row.visibility,
5650 row.sourceThreadId,
5651 row.item.type === "assistant_message" ? row.item.text : row.item.type,
5652 ]),
5653 [
5654 ["inherited", sourceThreadId, "legacy import before empty fork"],
5655 ["synthetic", sourceThreadId, "fork"],
5656 ["local", targetThreadId, "command_execution"],
5657 ["local", targetThreadId, "command_execution"],
5658 ],
5659 );
5660 }),
5661 );
5662
5663 it.effect("a pull request watch keeps a finished thread working until it ends", () =>
5664 Effect.gen(function* () {
5665 const store = yield* ProjectionStore.ProjectionStoreV2;
5666 const threadId = ThreadId.make("thread:watched-pull-request");
5667 const runId = RunId.make("run:watched-pull-request");
5668 const at = DateTime.makeUnsafe("2026-10-05T12:00:00.000Z");
5669 const thread = {
5670 createdBy: "user" as const,
5671 creationSource: "web" as const,
5672 id: threadId,
5673 projectId: ProjectId.make("project:watched-pull-request"),
5674 title: "Babysit the PR",
5675 providerInstanceId,
5676 modelSelection,
5677 runtimeMode: "full-access" as const,
5678 interactionMode: "default" as const,
5679 branch: null,
5680 worktreePath: null,
5681 activeProviderThreadId: null,
5682 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
5683 forkedFrom: null,
5684 createdAt: at,
5685 updatedAt: at,
5686 archivedAt: null,
5687 settledOverride: null,
5688 settledAt: null,
5689 lastVisitedAt: null,
5690 deletedAt: null,
5691 };
5692 yield* store.apply({
5693 id: EventId.make("event:watched-pull-request:thread"),
5694 type: "thread.created",
5695 threadId,
5696 occurredAt: at,
5697 payload: thread,
5698 });
5699 yield* store.apply({
5700 id: EventId.make("event:watched-pull-request:run"),
5701 type: "run.created",
5702 threadId,
5703 runId,
5704 driver,
5705 providerInstanceId,
5706 occurredAt: at,
5707 payload: {
5708 id: runId,
5709 threadId,
5710 ordinal: 1,
5711 providerInstanceId,
5712 modelSelection,
5713 providerThreadId: null,
5714 userMessageId: MessageId.make("message:watched-pull-request"),
5715 rootNodeId: null,
5716 activeAttemptId: null,
5717 status: "completed",
5718 requestedAt: at,
5719 startedAt: at,
5720 completedAt: at,
5721 checkpointId: null,
5722 contextHandoffId: null,
5723 },
5724 });
5725 const link = {
5726 host: "github.com",
5727 repository: "pingdotgg/t3code",
5728 number: 7,
5729 url: "https://github.com/pingdotgg/t3code/pull/7",
5730 source: "agent" as const,
5731 linkedAt: DateTime.formatIso(at),
5732 snapshot: null,
5733 stack: null,
5734 };
5735 const syncPullRequests = (id: string, pullRequests: ReadonlyArray<ThreadPullRequestLink>) =>
5736 store.apply({
5737 id: EventId.make(`event:watched-pull-request:${id}`),
5738 type: "thread.pull-request-synced",
5739 threadId,
5740 occurredAt: at,
5741 payload: { ...thread, pullRequests },
5742 });
5743 const project = { title: "Project" };
5744 const environmentId = EnvironmentId.make("environment:watched-pull-request");
5745 const phase = Effect.gen(function* () {
5746 const shell = yield* store.getThreadShell(threadId);
5747 const listed = (yield* store.getShellSnapshot()).threads.find(
5748 (candidate) => candidate.id === threadId,
5749 );
5750 assert.deepEqual(listed?.pendingBackgroundTasks, shell?.pendingBackgroundTasks);
5751 return shell && projectThreadAwarenessV2({ environmentId, project, thread: shell })?.phase;
5752 });
5753
5754 yield* syncPullRequests("watched", [
5755 {
5756 ...link,
5757 watch: {
5758 startedAt: DateTime.formatIso(at),
5759 headSha: null,
5760 failedChecks: [],
5761 passed: false,
5762 passedChecks: [],
5763 remarksThrough: DateTime.formatIso(at),
5764 remarkIds: [],
5765 conflicting: false,
5766 wakes: 0,
5767 },
5768 },
5769 ]);
5770 assert.deepEqual((yield* store.getThreadShell(threadId))?.pendingBackgroundTasks, [
5771 {
5772 taskId: "pull-request-watch:github.com/pingdotgg/t3code#7",
5773 description: "Watching pull request #7",
5774 kind: "monitor",
5775 },
5776 ]);
5777 assert.equal(yield* phase, "running");
5778
5779 yield* syncPullRequests("unwatched", [link]);
5780 assert.deepEqual((yield* store.getThreadShell(threadId))?.pendingBackgroundTasks, []);
5781 assert.equal(yield* phase, "completed");
5782 }),
5783 );
5784
5785 it.effect("dates a completion released by held background work at the release", () =>
5786 Effect.gen(function* () {
5787 const store = yield* ProjectionStore.ProjectionStoreV2;
5788 const threadId = ThreadId.make("thread:held-completion");
5789 const runId = RunId.make("run:held-completion");
5790 const providerThreadId = ProviderThreadId.make("provider-thread:held-completion");
5791 const completedAt = DateTime.makeUnsafe("2026-09-28T12:00:00.000Z");
5792 const releasedAt = DateTime.makeUnsafe("2026-09-28T12:05:00.000Z");
5793 const providerThread = {
5794 id: providerThreadId,
5795 driver,
5796 providerInstanceId,
5797 providerSessionId: null,
5798 appThreadId: threadId,
5799 ownerNodeId: null,
5800 nativeThreadRef: null,
5801 nativeConversationHeadRef: null,
5802 status: "idle" as const,
5803 firstRunOrdinal: 1,
5804 lastRunOrdinal: 1,
5805 handoffIds: [],
5806 forkedFrom: null,
5807 pendingBackgroundTasks: [{ taskId: "watch", kind: "monitor" as const }],
5808 createdAt: completedAt,
5809 updatedAt: completedAt,
5810 };
5811 yield* store.apply({
5812 id: EventId.make("event:held-completion:thread"),
5813 type: "thread.created",
5814 threadId,
5815 occurredAt: completedAt,
5816 payload: {
5817 createdBy: "user",
5818 creationSource: "web",
5819 id: threadId,
5820 projectId: ProjectId.make("project:held-completion"),
5821 title: "Watch the build",
5822 providerInstanceId,
5823 modelSelection,
5824 runtimeMode: "full-access",
5825 interactionMode: "default",
5826 branch: null,
5827 worktreePath: null,
5828 activeProviderThreadId: null,
5829 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
5830 forkedFrom: null,
5831 createdAt: completedAt,
5832 updatedAt: completedAt,
5833 archivedAt: null,
5834 settledOverride: null,
5835 settledAt: null,
5836 lastVisitedAt: null,
5837 deletedAt: null,
5838 },
5839 });
5840 yield* store.apply({
5841 id: EventId.make("event:held-completion:run"),
5842 type: "run.created",
5843 threadId,
5844 runId,
5845 driver,
5846 providerInstanceId,
5847 occurredAt: completedAt,
5848 payload: {
5849 id: runId,
5850 threadId,
5851 ordinal: 1,
5852 providerInstanceId,
5853 modelSelection,
5854 providerThreadId,
5855 userMessageId: MessageId.make("message:held-completion"),
5856 rootNodeId: null,
5857 activeAttemptId: null,
5858 status: "completed",
5859 requestedAt: completedAt,
5860 startedAt: completedAt,
5861 completedAt,
5862 checkpointId: null,
5863 contextHandoffId: null,
5864 },
5865 });
5866 yield* store.apply({
5867 id: EventId.make("event:held-completion:monitor-running"),
5868 type: "provider-thread.updated",
5869 threadId,
5870 driver,
5871 providerInstanceId,
5872 occurredAt: completedAt,
5873 payload: providerThread,
5874 });
5875 const project = { title: "Project" };
5876 const environmentId = EnvironmentId.make("environment:held-completion");
5877 const held = yield* store.getThreadShell(threadId);
5878 assert.equal(
5879 held && projectThreadAwarenessV2({ environmentId, project, thread: held })?.phase,
5880 "running",
5881 );
5882
5883 // The monitor ends five minutes after the run: the push must not look stale.
5884 yield* store.apply({
5885 id: EventId.make("event:held-completion:monitor-ended"),
5886 type: "provider-thread.updated",
5887 threadId,
5888 driver,
5889 providerInstanceId,
5890 occurredAt: releasedAt,
5891 payload: { ...providerThread, pendingBackgroundTasks: [], updatedAt: releasedAt },
5892 });
5893 const released = yield* store.getThreadShell(threadId);
5894 const state =
5895 released && projectThreadAwarenessV2({ environmentId, project, thread: released });
5896 assert.equal(state?.phase, "completed");
5897 assert.equal(state?.updatedAt, DateTime.formatIso(releasedAt));
5898
5899 // A later provider must not hide this owner's roster from restart recovery.
5900 yield* store.apply({
5901 id: EventId.make("event:held-completion:monitor-restarted"),
5902 type: "provider-thread.updated",
5903 threadId,
5904 occurredAt: releasedAt,
5905 payload: { ...providerThread, updatedAt: releasedAt },
5906 });
5907 const newerRunId = RunId.make("run:held-completion:new-provider");
5908 yield* store.apply({
5909 id: EventId.make("event:held-completion:new-provider"),
5910 type: "run.created",
5911 threadId,
5912 runId: newerRunId,
5913 occurredAt: releasedAt,
5914 payload: {
5915 ...(yield* store.getThreadProjection(threadId)).runs[0]!,
5916 id: newerRunId,
5917 ordinal: 2,
5918 providerThreadId: ProviderThreadId.make("provider-thread:held-completion:new-provider"),
5919 requestedAt: releasedAt,
5920 startedAt: releasedAt,
5921 completedAt: releasedAt,
5922 },
5923 });
5924 assert.deepEqual(
5925 (yield* store.getRuntimeRecoveryProjection(threadId)).runs.map((run) => run.id),
5926 [runId, newerRunId],
5927 );
5928 }),
5929 );
5930});