1import { assert, it } from "@effect/vitest";
2import {
3 CommandId,
4 ContextTransferId,
5 EventId,
6 MessageId,
7 NodeId,
8 type OrchestrationV2AppThread,
9 type OrchestrationV2Run,
10 ProjectId,
11 ProviderDriverKind,
12 ProviderInstanceId,
13 ProviderSessionId,
14 ProviderThreadId,
15 RunId,
16 RuntimeRequestId,
17 ThreadId,
18 TurnItemId,
19} from "@t3tools/contracts";
20import * as DateTime from "effect/DateTime";
21import * as Effect from "effect/Effect";
22import * as Layer from "effect/Layer";
23import * as SqlClient from "effect/sql/SqlClient";
24
25import * as SqlitePersistence from "../persistence/Sqlite.ts";
26import * as ServerSettings from "../serverSettings.ts";
27import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
28import * as EffectOutbox from "./EffectOutbox.ts";
29import * as ProjectionStore from "./ProjectionStore.ts";
30import * as EventSink from "./EventSink.ts";
31import * as EventStore from "./EventStore.ts";
32import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
33import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
34import { restartContinuationRun } from "./RestartContinuation.ts";
35
36const layerTest = Layer.mergeAll(ProjectionStore.layer, EffectOutbox.layer).pipe(
37 Layer.provideMerge(SqlitePersistence.layerMemory),
38);
39const layerRecoveryTest = ProviderRuntimeRecovery.layer.pipe(
40 Layer.provideMerge(
41 EventSink.layer.pipe(
42 Layer.provideMerge(
43 Layer.mergeAll(
44 EventStore.layer,
45 ProjectionStore.layer,
46 EffectOutbox.layer,
47 IdAllocator.layer,
48 ).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
49 ),
50 ),
51 ),
52 Layer.provide(ServerSettings.layerTest()),
53);
54const providerInstanceId = ProviderInstanceId.make("codex");
55const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" };
56const driver = ProviderDriverKind.make("codex");
57
58const createThread = Effect.fn(function* (
59 name: string,
60 overrides: Partial<OrchestrationV2AppThread> = {},
61) {
62 const projections = yield* ProjectionStore.ProjectionStoreV2;
63 const now = yield* DateTime.now;
64 const threadId = ThreadId.make(`thread:recovery:${name}`);
65 const thread: OrchestrationV2AppThread = {
66 createdBy: "user",
67 creationSource: "web",
68 id: threadId,
69 projectId: ProjectId.make("project:recovery"),
70 title: name,
71 providerInstanceId,
72 modelSelection,
73 runtimeMode: "full-access",
74 interactionMode: "default",
75 branch: null,
76 worktreePath: null,
77 activeProviderThreadId: null,
78 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
79 forkedFrom: null,
80 createdAt: now,
81 updatedAt: now,
82 archivedAt: null,
83 settledOverride: null,
84 settledAt: null,
85 lastVisitedAt: null,
86 deletedAt: null,
87 ...overrides,
88 };
89 yield* projections.apply({
90 id: EventId.make(`event:${threadId}:created`),
91 type: "thread.created",
92 threadId,
93 occurredAt: now,
94 payload: thread,
95 });
96 return threadId;
97});
98
99const createRun = Effect.fn(function* (
100 threadId: ThreadId,
101 status: OrchestrationV2Run["status"],
102 overrides: Partial<OrchestrationV2Run> = {},
103) {
104 const projections = yield* ProjectionStore.ProjectionStoreV2;
105 const now = yield* DateTime.now;
106 const ordinal = overrides.ordinal ?? 1;
107 const runId = RunId.make(`run:${threadId}:${ordinal}`);
108 yield* projections.apply({
109 id: EventId.make(`event:${runId}:created`),
110 type: "run.created",
111 threadId,
112 runId,
113 occurredAt: now,
114 payload: {
115 id: runId,
116 threadId,
117 ordinal,
118 providerInstanceId,
119 modelSelection,
120 providerThreadId: null,
121 userMessageId: MessageId.make(`message:${runId}`),
122 rootNodeId: null,
123 activeAttemptId: null,
124 status,
125 requestedAt: now,
126 startedAt: now,
127 completedAt: status === "completed" ? now : null,
128 checkpointId: null,
129 contextHandoffId: null,
130 ...overrides,
131 },
132 });
133 return runId;
134});
135
136it.effect.each(["startup", "shutdown"] as const)(
137 "%s cancels queued buffered wakes and holds durable messages",
138 (trigger) =>
139 Effect.gen(function* () {
140 const projections = yield* ProjectionStore.ProjectionStoreV2;
141 const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
142 const now = yield* DateTime.now;
143 const bufferedThreads = [] as ThreadId[];
144 const durableThreads = [] as ThreadId[];
145 for (const kind of ["buffered", "user", "notification", "delegated"] as const) {
146 const threadId = yield* createThread(`wake:${trigger}:${kind}`);
147 const buffered = kind === "buffered";
148 (buffered ? bufferedThreads : durableThreads).push(threadId);
149 const messageId = MessageId.make(`message:${threadId}`);
150 const runId = yield* createRun(threadId, "queued", {
151 userMessageId: messageId,
152 startedAt: null,
153 queuePosition: 1,
154 });
155 yield* createRun(threadId, "completed", { ordinal: 2 });
156 yield* projections.apply({
157 id: EventId.make(`message:${threadId}`),
158 type: "message.updated",
159 threadId,
160 runId,
161 occurredAt: now,
162 payload: {
163 id: messageId,
164 threadId,
165 runId,
166 nodeId: null,
167 role: "user",
168 text: "Queued work",
169 attachments: [],
170 streaming: false,
171 createdAt: now,
172 updatedAt: now,
173 createdBy: kind === "user" ? "user" : "agent",
174 creationSource: buffered ? "provider" : "server",
175 ...(kind === "user" || kind === "delegated"
176 ? {}
177 : {
178 notification: {
179 source: { kind: "background_task" as const },
180 outcome: "updated" as const,
181 summary: "Background activity updated",
182 },
183 }),
184 ...(kind === "delegated"
185 ? { delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [] } }
186 : {}),
187 },
188 });
189 }
190 assert.deepEqual(
191 new Set(yield* projections.getRecoveryThreadIds("runtime")),
192 new Set([...bufferedThreads, ...durableThreads]),
193 );
194 const summary = yield* recovery.reconcile(trigger);
195 assert.equal(summary.terminalizedRuns, 1);
196 for (const threadId of bufferedThreads) {
197 const projection = yield* projections.getThreadProjection(threadId);
198 assert.equal(projection.runs[0]?.status, "cancelled");
199 assert.isNotNull(projection.runs[0]?.completedAt);
200 assert.isNull(projection.runs[0]?.queuePosition);
201 assert.equal(projection.runs[1]?.status, "completed");
202 }
203 for (const threadId of durableThreads) {
204 const projection = yield* projections.getThreadProjection(threadId);
205 assert.equal(projection.runs[0]?.status, "queued");
206 assert.isTrue(projection.runs[0]?.queueHeld);
207 }
208 assert.deepEqual(yield* projections.getRecoveryThreadIds("runtime"), []);
209 }).pipe(Effect.provide(layerRecoveryTest)),
210);
211
212it.effect("stops a session its thread detached, which kept its live status", () =>
213 Effect.gen(function* () {
214 const projections = yield* ProjectionStore.ProjectionStoreV2;
215 const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
216 const now = yield* DateTime.now;
217 const threadId = yield* createThread("detached-session");
218 const providerSessionId = ProviderSessionId.make("session:recovery:detached");
219 yield* projections.apply({
220 id: EventId.make("event:recovery:detached-session:attached"),
221 type: "provider-session.attached",
222 threadId,
223 driver,
224 providerInstanceId,
225 occurredAt: now,
226 payload: {
227 id: providerSessionId,
228 driver,
229 providerInstanceId,
230 status: "ready",
231 cwd: "/workspace",
232 model: modelSelection.model,
233 capabilities: CodexProviderCapabilitiesV2,
234 createdAt: now,
235 updatedAt: now,
236 lastError: null,
237 },
238 });
239 // The server died before the release that should follow the detach.
240 yield* projections.apply({
241 id: EventId.make("event:recovery:detached-session:detached"),
242 type: "provider-session.detached",
243 threadId,
244 driver,
245 providerInstanceId,
246 occurredAt: now,
247 payload: { providerSessionId, detachedAt: now, reason: "Thread settled." },
248 });
249 assert.deepEqual(yield* projections.getRecoveryThreadIds("runtime"), [threadId]);
250
251 const summary = yield* recovery.reconcile("startup");
252
253 assert.equal(summary.stoppedSessions, 1);
254 const sql = yield* SqlClient.SqlClient;
255 const rows = yield* sql<{ readonly status: string }>`
256 SELECT status FROM orchestration_v2_projection_provider_sessions
257 WHERE provider_session_id = ${providerSessionId}
258 `;
259 assert.deepEqual(rows, [{ status: "stopped" }]);
260 assert.deepEqual(yield* projections.getRecoveryThreadIds("runtime"), []);
261 }).pipe(Effect.provide(layerRecoveryTest)),
262);
263
264it.effect("selects unfinished recovery work without reading settled thread histories", () =>
265 Effect.gen(function* () {
266 const projections = yield* ProjectionStore.ProjectionStoreV2;
267 const outbox = yield* EffectOutbox.EffectOutboxV2;
268 const sql = yield* SqlClient.SqlClient;
269 const now = yield* DateTime.now;
270 for (let index = 0; index < 600; index += 1) {
271 yield* createRun(yield* createThread(`settled-${index}`), "completed");
272 }
273 const queued = yield* createThread("queued");
274 const archived = yield* createThread("archived", { archivedAt: now });
275 const deleted = yield* createThread("deleted", { deletedAt: now });
276 const blocked = yield* createThread("blocked");
277 yield* createRun(queued, "completed");
278 for (const threadId of [queued, archived, deleted, blocked]) {
279 yield* createRun(threadId, "queued", threadId === queued ? { ordinal: 2 } : {});
280 }
281 yield* createRun(blocked, "waiting", { ordinal: 2 });
282 const background = yield* createThread("background");
283 yield* createRun(background, "completed");
284 yield* projections.apply({
285 id: EventId.make("event:recovery:background"),
286 type: "turn-item.updated",
287 threadId: background,
288 occurredAt: now,
289 payload: {
290 id: TurnItemId.make("item:recovery:background"),
291 threadId: background,
292 runId: null,
293 nodeId: null,
294 providerThreadId: null,
295 providerTurnId: null,
296 nativeItemRef: null,
297 parentItemId: null,
298 ordinal: 1,
299 type: "dynamic_tool",
300 status: "waiting",
301 title: null,
302 startedAt: now,
303 completedAt: null,
304 updatedAt: now,
305 toolName: "background",
306 input: null,
307 output: null,
308 },
309 });
310 const outboxOnly = yield* createThread("outbox-only");
311 yield* outbox.enqueue([
312 {
313 id: "effect:recovery:outbox-only",
314 commandId: CommandId.make("command:recovery:outbox-only"),
315 threadId: outboxOnly,
316 request: { type: "provider-turn.start", runId: RunId.make("run:outbox-only") },
317 },
318 ]);
319 const requestOnly = yield* createThread("request-only");
320 yield* projections.apply({
321 id: EventId.make("event:recovery:request-only"),
322 type: "runtime-request.updated",
323 threadId: requestOnly,
324 occurredAt: now,
325 payload: {
326 id: RuntimeRequestId.make("request:recovery:pending"),
327 nodeId: NodeId.make("node:recovery:request"),
328 providerTurnId: null,
329 nativeRequestRef: null,
330 kind: "user_input",
331 status: "pending",
332 responseCapability: { type: "not_resumable", reason: "Process stopped" },
333 createdAt: now,
334 resolvedAt: null,
335 },
336 });
337 const delivery = yield* createThread("delivery", { archivedAt: now });
338 yield* createRun(delivery, "completed", {
339 delegatedCompletion: {
340 disposition: "open",
341 nextGeneration: 2,
342 delivery: { generation: 1, messageId: MessageId.make("message:delivery"), taskIds: [] },
343 },
344 });
345
346 // This historical payload cannot be decoded. Candidate discovery must not
347 // materialize it while deciding which threads have work to reconcile.
348 yield* sql`
349 UPDATE orchestration_v2_projection_runs SET payload_json = '{broken'
350 WHERE thread_id = ${ThreadId.make("thread:recovery:settled-599")}
351 `;
352 yield* sql`
353 UPDATE orchestration_v2_projection_runs SET payload_json = '{broken'
354 WHERE thread_id = ${queued} AND ordinal = 1
355 `;
356 assert.deepEqual(yield* projections.getRecoveryThreadIds("queued-runs"), [queued]);
357 assert.deepEqual(
358 new Set(yield* projections.getRecoveryThreadIds("runtime")),
359 new Set([queued, archived, blocked, background, outboxOnly, requestOnly]),
360 );
361 assert.deepEqual(yield* projections.getRecoveryThreadIds("delegated-completions"), [delivery]);
362 assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), []);
363 const recoveryState = yield* projections.getRuntimeRecoveryProjection(queued);
364 assert.deepEqual(
365 recoveryState.runs.map((run) => run.id),
366 [RunId.make(`run:${queued}:2`)],
367 );
368 assert.deepEqual(yield* projections.getUnreadableThreadIds(), [
369 queued,
370 ThreadId.make("thread:recovery:settled-599"),
371 ]);
372 }).pipe(Effect.provide(layerTest)),
373);
374
375it.effect("recovers terminal subagent results until their cross-thread transfer exists", () =>
376 Effect.gen(function* () {
377 const projections = yield* ProjectionStore.ProjectionStoreV2;
378 const sql = yield* SqlClient.SqlClient;
379 const now = yield* DateTime.now;
380 const parent = yield* createThread("subagent-parent");
381 const children: Array<ThreadId> = [];
382 for (const name of ["terminal", "archived", "deleted", "running", "held", "queued"]) {
383 const child = yield* createThread(`subagent-${name}`, {
384 lineage: { parentThreadId: parent, relationshipToParent: "subagent", rootThreadId: parent },
385 forkedFrom: { type: "node", nodeId: NodeId.make(`node:${name}`) },
386 archivedAt: name === "archived" ? now : null,
387 deletedAt: name === "deleted" ? now : null,
388 });
389 yield* createRun(child, name === "running" ? "running" : "completed");
390 // A wake queued behind the result: held by Stop or a restart, or still deliverable.
391 if (name === "held" || name === "queued") {
392 yield* createRun(child, "queued", {
393 ordinal: 2,
394 startedAt: null,
395 ...(name === "held" ? { queueHeld: true } : {}),
396 });
397 }
398 children.push(child);
399 }
400 const terminalChildren = new Set([children[0]!, children[1]!, children[4]!]);
401 assert.deepEqual(
402 new Set(yield* projections.getRecoveryThreadIds("subagent-results")),
403 terminalChildren,
404 );
405
406 const completed = children[0]!;
407 const transferId = ContextTransferId.make("transfer:recovery:subagent-result");
408 yield* projections.apply({
409 id: EventId.make("event:recovery:subagent-result"),
410 type: "context-transfer.created",
411 threadId: parent,
412 occurredAt: now,
413 payload: {
414 id: transferId,
415 type: "subagent_result",
416 sourceThreadId: completed,
417 targetThreadId: parent,
418 sourcePoint: { threadId: completed },
419 basePoint: null,
420 sourceProviderInstanceId: providerInstanceId,
421 targetProviderInstanceId: providerInstanceId,
422 targetRunId: null,
423 status: "pending",
424 resolution: null,
425 createdBy: "system",
426 error: null,
427 createdAt: now,
428 updatedAt: now,
429 consumedAt: null,
430 },
431 });
432 const archivedChild = children[1];
433 assert.isDefined(archivedChild);
434 assert.deepEqual(
435 new Set(yield* projections.getRecoveryThreadIds("subagent-results")),
436 new Set([archivedChild, children[4]!]),
437 );
438 assert.deepEqual(yield* projections.getUnreadableThreadIds(), []);
439 yield* sql`
440 UPDATE orchestration_v2_projection_context_transfers SET payload_json = '{}'
441 WHERE context_transfer_id = ${transferId}
442 `;
443 assert.deepEqual(
444 new Set(yield* projections.getUnreadableThreadIds()),
445 new Set([parent, completed]),
446 );
447 }).pipe(Effect.provide(layerTest)),
448);
449
450it.effect.each([
451 { name: "SQLite", layer: layerTest },
452 { name: "memory", layer: ProjectionStore.layerMemory },
453])("recovers only recorded follow-up results with $name projections", ({ layer }) =>
454 Effect.gen(function* () {
455 const projections = yield* ProjectionStore.ProjectionStoreV2;
456 const now = yield* DateTime.now;
457 const parent = yield* createThread("followup-parent");
458 const child = yield* createThread("followup-child", {
459 lineage: { parentThreadId: parent, relationshipToParent: "subagent", rootThreadId: parent },
460 forkedFrom: { type: "node", nodeId: NodeId.make("node:original") },
461 });
462 const originalRunId = yield* createRun(child, "completed");
463 yield* projections.apply({
464 id: EventId.make("event:original-result"),
465 type: "context-transfer.created",
466 threadId: parent,
467 occurredAt: now,
468 payload: {
469 id: ContextTransferId.make("transfer:original-result"),
470 type: "subagent_result",
471 sourceThreadId: child,
472 targetThreadId: parent,
473 sourcePoint: { threadId: child, runId: originalRunId },
474 basePoint: null,
475 sourceProviderInstanceId: providerInstanceId,
476 targetProviderInstanceId: providerInstanceId,
477 targetRunId: null,
478 status: "pending",
479 resolution: null,
480 createdBy: "system",
481 error: null,
482 createdAt: now,
483 updatedAt: now,
484 consumedAt: null,
485 },
486 });
487 // Historical untracked work does not acquire a notification on restart.
488 yield* createRun(child, "completed", { ordinal: 2 });
489 assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), []);
490 const taskId = NodeId.make("node:followup");
491 yield* createRun(child, "completed", { ordinal: 3, delegatedTaskId: taskId });
492 yield* createRun(child, "running", { ordinal: 4 });
493 const task = {
494 id: taskId,
495 threadId: parent,
496 runId: null,
497 parentNodeId: NodeId.make("node:parent"),
498 origin: "app_owned" as const,
499 createdBy: "agent" as const,
500 driver,
501 providerInstanceId,
502 providerThreadId: null,
503 childThreadId: child,
504 nativeTaskRef: null,
505 prompt: "Check the follow-up.",
506 title: null,
507 model: null,
508 status: "running" as const,
509 result: null,
510 startedAt: now,
511 completedAt: null,
512 updatedAt: now,
513 };
514 yield* projections.apply({
515 id: EventId.make("event:followup-task"),
516 type: "subagent.updated",
517 threadId: parent,
518 occurredAt: now,
519 payload: task,
520 });
521 assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), [child]);
522 yield* projections.apply({
523 id: EventId.make("event:followup-settled"),
524 type: "subagent.updated",
525 threadId: parent,
526 occurredAt: now,
527 payload: { ...task, status: "completed", result: "Follow-up result.", completedAt: now },
528 });
529 assert.deepEqual(yield* projections.getRecoveryThreadIds("subagent-results"), []);
530 }).pipe(Effect.provide(layer)),
531);
532
533it.effect("includes shared sessions and provider-owned background rosters in recovery", () =>
534 Effect.gen(function* () {
535 const projections = yield* ProjectionStore.ProjectionStoreV2;
536 const now = yield* DateTime.now;
537 const first = yield* createThread("shared-first");
538 const second = yield* createThread("shared-second", { archivedAt: now });
539 const providerSessionId = ProviderSessionId.make("session:recovery:shared");
540 for (const threadId of [first, second]) {
541 yield* projections.apply({
542 id: EventId.make(`event:${threadId}:session`),
543 type: "provider-session.attached",
544 threadId,
545 driver,
546 providerInstanceId,
547 occurredAt: now,
548 payload: {
549 id: providerSessionId,
550 driver,
551 providerInstanceId,
552 status: "ready",
553 cwd: "/workspace",
554 model: modelSelection.model,
555 capabilities: CodexProviderCapabilitiesV2,
556 createdAt: now,
557 updatedAt: now,
558 lastError: null,
559 },
560 });
561 }
562 const roster = yield* createThread("roster");
563 yield* projections.apply({
564 id: EventId.make("event:recovery:roster"),
565 type: "provider-thread.updated",
566 threadId: roster,
567 driver,
568 providerInstanceId,
569 occurredAt: now,
570 payload: {
571 id: ProviderThreadId.make("provider-thread:recovery:roster"),
572 appThreadId: roster,
573 ownerNodeId: null,
574 driver,
575 providerInstanceId,
576 providerSessionId: null,
577 nativeThreadRef: null,
578 nativeConversationHeadRef: null,
579 status: "idle",
580 firstRunOrdinal: null,
581 lastRunOrdinal: null,
582 handoffIds: [],
583 forkedFrom: null,
584 createdAt: now,
585 updatedAt: now,
586 pendingBackgroundTasks: [
587 { taskId: "background", description: "Still running", kind: "command" },
588 ],
589 },
590 });
591 const prepared = yield* createThread("prepared-continuation");
592 const preparedSessionId = ProviderSessionId.make("session:recovery:prepared");
593 const preparedProviderThreadId = ProviderThreadId.make("provider-thread:recovery:prepared");
594 yield* projections.apply({
595 id: EventId.make("event:recovery:prepared-session"),
596 type: "provider-session.attached",
597 threadId: prepared,
598 driver,
599 providerInstanceId,
600 occurredAt: now,
601 payload: {
602 id: preparedSessionId,
603 driver,
604 providerInstanceId,
605 status: "stopped",
606 cwd: "/workspace",
607 model: modelSelection.model,
608 capabilities: CodexProviderCapabilitiesV2,
609 createdAt: now,
610 updatedAt: now,
611 lastError: null,
612 },
613 });
614 yield* projections.apply({
615 id: EventId.make("event:recovery:prepared-thread"),
616 type: "provider-thread.updated",
617 threadId: prepared,
618 driver,
619 providerInstanceId,
620 occurredAt: now,
621 payload: {
622 id: preparedProviderThreadId,
623 appThreadId: prepared,
624 ownerNodeId: null,
625 driver,
626 providerInstanceId,
627 providerSessionId: preparedSessionId,
628 nativeThreadRef: { driver, nativeId: "native:prepared", strength: "strong" },
629 nativeConversationHeadRef: null,
630 status: "idle",
631 firstRunOrdinal: 1,
632 lastRunOrdinal: 1,
633 handoffIds: [],
634 forkedFrom: null,
635 createdAt: now,
636 updatedAt: now,
637 pendingBackgroundTasks: [],
638 },
639 });
640 const preparedRunId = yield* createRun(prepared, "starting", {
641 providerThreadId: preparedProviderThreadId,
642 restartContinuationOfRunId: RunId.make("run:recovery:source"),
643 });
644 assert.deepEqual(
645 new Set(yield* projections.getRecoveryThreadIds("runtime")),
646 new Set([first, second, roster, prepared]),
647 );
648 const preparedState = yield* projections.getRuntimeRecoveryProjection(prepared);
649 assert.deepEqual(
650 preparedState.providerSessions.map((session) => session.id),
651 [preparedSessionId],
652 );
653 assert.equal(restartContinuationRun(preparedState, yield* DateTime.now)?.id, preparedRunId);
654 assert.deepEqual(yield* projections.getUnreadableThreadIds(), []);
655 const sql = yield* SqlClient.SqlClient;
656 yield* sql`
657 UPDATE orchestration_v2_projection_provider_sessions SET payload_json = '{}'
658 WHERE provider_session_id = ${providerSessionId}
659 `;
660 assert.deepEqual(
661 new Set(yield* projections.getUnreadableThreadIds()),
662 new Set([first, second]),
663 );
664 }).pipe(Effect.provide(layerTest)),
665);
666
667it.effect("reads the run that owns a background roster, not a queued or resumed one", () =>
668 Effect.gen(function* () {
669 const projections = yield* ProjectionStore.ProjectionStoreV2;
670 const now = yield* DateTime.now;
671 const withRoster = Effect.fn(function* (threadId: ThreadId) {
672 yield* projections.apply({
673 id: EventId.make(`event:${threadId}:roster`),
674 type: "provider-thread.updated",
675 threadId,
676 driver,
677 providerInstanceId,
678 occurredAt: now,
679 payload: {
680 id: ProviderThreadId.make(`provider-thread:${threadId}`),
681 appThreadId: threadId,
682 ownerNodeId: null,
683 driver,
684 providerInstanceId,
685 providerSessionId: null,
686 nativeThreadRef: null,
687 nativeConversationHeadRef: null,
688 status: "idle",
689 firstRunOrdinal: null,
690 lastRunOrdinal: null,
691 handoffIds: [],
692 forkedFrom: null,
693 createdAt: now,
694 updatedAt: now,
695 pendingBackgroundTasks: [
696 { taskId: "background", description: "Still running", kind: "command" },
697 ],
698 },
699 });
700 });
701 const runIds = (threadId: ThreadId) =>
702 projections
703 .getRuntimeRecoveryProjection(threadId)
704 .pipe(Effect.map((state) => state.runs.map((run) => run.id)));
705 // Settled with background work left, then a queued follow-up.
706 const settled = yield* createThread("roster-before-queue");
707 const settledRun = yield* createRun(settled, "completed", {
708 providerThreadId: ProviderThreadId.make(`provider-thread:${settled}`),
709 });
710 const queuedRun = yield* createRun(settled, "queued", { ordinal: 2, completedAt: null });
711 yield* withRoster(settled);
712 assert.deepEqual(yield* runIds(settled), [settledRun, queuedRun]);
713 // A resumed queued run (ordinal 1) ended after a continuation (ordinal 2).
714 const resumed = yield* createThread("roster-after-resume");
715 const resumedRun = yield* createRun(resumed, "completed", {
716 providerThreadId: ProviderThreadId.make(`provider-thread:${resumed}`),
717 completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00.000Z"),
718 });
719 const continuationRun = yield* createRun(resumed, "completed", {
720 providerThreadId: ProviderThreadId.make(`provider-thread:${resumed}`),
721 ordinal: 2,
722 completedAt: DateTime.makeUnsafe("2026-10-03T10:00:00.000Z"),
723 });
724 yield* withRoster(resumed);
725 assert.deepEqual(yield* runIds(resumed), [resumedRun, continuationRun]);
726 // Without a roster, settled history is not read.
727 const quiet = yield* createThread("no-roster-before-queue");
728 yield* createRun(quiet, "completed");
729 const quietQueued = yield* createRun(quiet, "queued", { ordinal: 2, completedAt: null });
730 assert.deepEqual(yield* runIds(quiet), [quietQueued]);
731 }).pipe(Effect.provide(layerTest)),
732);
733
734it.effect("marks fork descendants unreadable when their source is missing or corrupt", () =>
735 Effect.gen(function* () {
736 const projections = yield* ProjectionStore.ProjectionStoreV2;
737 const sql = yield* SqlClient.SqlClient;
738 const source = yield* createThread("source");
739 const sourceRun = yield* createRun(source, "completed");
740 const fork = yield* createThread("fork", {
741 forkedFrom: { type: "run", threadId: source, runId: sourceRun },
742 });
743 const descendant = yield* createThread("descendant", {
744 forkedFrom: { type: "run", threadId: fork, runId: RunId.make("run:fork") },
745 });
746 assert.deepEqual(yield* projections.getUnreadableThreadIds(), []);
747 yield* sql`
748 UPDATE orchestration_v2_projection_runs SET payload_json = '{}'
749 WHERE run_id = ${sourceRun}
750 `;
751 assert.deepEqual(
752 new Set(yield* projections.getUnreadableThreadIds()),
753 new Set([source, fork, descendant]),
754 );
755 yield* sql`DELETE FROM orchestration_v2_projection_threads WHERE thread_id = ${source}`;
756 assert.deepEqual(
757 new Set(yield* projections.getUnreadableThreadIds()),
758 new Set([fork, descendant]),
759 );
760 }).pipe(Effect.provide(layerTest)),
761);