1import { assert, it } from "@effect/vitest";
2import {
3 EventId,
4 MessageId,
5 NodeId,
6 type OrchestrationV2AppThread,
7 type OrchestrationV2Run,
8 ProjectId,
9 ProviderDriverKind,
10 ProviderInstanceId,
11 ProviderThreadId,
12 RunId,
13 RuntimeRequestId,
14 ThreadId,
15 TurnItemId,
16} from "@t3tools/contracts";
17import * as DateTime from "effect/DateTime";
18import * as Effect from "effect/Effect";
19import * as Layer from "effect/Layer";
20import * as Schema from "effect/Schema";
21import * as SqlClient from "effect/sql/SqlClient";
22import * as Statement from "effect/sql/Statement";
23
24import * as SqlitePersistence from "../persistence/Sqlite.ts";
25import * as ProjectionStore from "./ProjectionStore.ts";
26import { isAutoSettlementCandidate, resolveAutoSettlementAt } from "./ThreadSettlementService.ts";
27
28const layerSql = ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory));
29const now = DateTime.makeUnsafe("2026-09-04T12:00:00Z");
30const old = DateTime.subtract(now, { days: 10 });
31const providerInstanceId = ProviderInstanceId.make("codex");
32const modelSelection = { instanceId: providerInstanceId, model: "test-model" };
33const encodeNumbers = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema.Number)));
34
35const createThread = Effect.fn(function* (
36 name: string,
37 overrides: Partial<OrchestrationV2AppThread> = {},
38) {
39 const store = yield* ProjectionStore.ProjectionStoreV2;
40 const threadId = ThreadId.make(`thread:settlement:${name}`);
41 const thread: OrchestrationV2AppThread = {
42 createdBy: "user",
43 creationSource: "web",
44 id: threadId,
45 projectId: ProjectId.make("project:settlement"),
46 title: name,
47 providerInstanceId,
48 modelSelection,
49 runtimeMode: "full-access",
50 interactionMode: "default",
51 branch: "feature",
52 worktreePath: null,
53 activeProviderThreadId: null,
54 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
55 forkedFrom: null,
56 createdAt: old,
57 updatedAt: old,
58 archivedAt: null,
59 settledOverride: null,
60 settledAt: null,
61 lastVisitedAt: null,
62 deletedAt: null,
63 ...overrides,
64 };
65 yield* store.apply({
66 id: EventId.make(`event:${threadId}`),
67 type: "thread.created",
68 threadId,
69 occurredAt: old,
70 payload: thread,
71 });
72 return threadId;
73});
74
75const createRun = Effect.fn(function* (
76 threadId: ThreadId,
77 status: OrchestrationV2Run["status"] = "completed",
78 ordinal = 1,
79) {
80 const store = yield* ProjectionStore.ProjectionStoreV2;
81 const runId = RunId.make(`run:${threadId}:${ordinal}`);
82 yield* store.apply({
83 id: EventId.make(`event:${runId}`),
84 type: "run.created",
85 threadId,
86 occurredAt: old,
87 payload: {
88 id: runId,
89 threadId,
90 ordinal,
91 providerInstanceId,
92 modelSelection,
93 providerThreadId: null,
94 userMessageId: MessageId.make(`message:${runId}`),
95 rootNodeId: null,
96 activeAttemptId: null,
97 status,
98 requestedAt: old,
99 startedAt: status === "queued" ? null : old,
100 completedAt: ["completed", "rolled_back", "failed"].includes(status) ? old : null,
101 checkpointId: null,
102 contextHandoffId: null,
103 },
104 });
105 return runId;
106});
107
108const createItem = Effect.fn(function* (
109 threadId: ThreadId,
110 runId: RunId,
111 status: "idle" | "running" | "completed",
112 persistent = false,
113) {
114 const store = yield* ProjectionStore.ProjectionStoreV2;
115 yield* store.apply({
116 id: EventId.make(`event:item:${runId}`),
117 type: "turn-item.updated",
118 threadId,
119 occurredAt: old,
120 payload: {
121 id: TurnItemId.make(`item:${runId}`),
122 threadId,
123 runId,
124 nodeId: null,
125 providerThreadId: null,
126 providerTurnId: null,
127 nativeItemRef: null,
128 parentItemId: null,
129 ordinal: 1,
130 type: "dynamic_tool",
131 status,
132 title: null,
133 startedAt: old,
134 completedAt: status === "completed" ? old : null,
135 updatedAt: old,
136 toolName: "background",
137 input: persistent ? { persistent: true } : null,
138 output: null,
139 },
140 });
141});
142
143it.effect.each([
144 ["sql", layerSql],
145 ["memory", ProjectionStore.layerMemory],
146] as const)(
147 "%s: discovers settlement work with the same activity and background semantics as the shell",
148 ([, testLayer]) =>
149 Effect.gen(function* () {
150 const store = yield* ProjectionStore.ProjectionStoreV2;
151 const idle = yield* createThread("idle");
152 const completed = yield* createThread("completed");
153 yield* createRun(completed);
154 for (const [name, overrides] of [
155 ["archived", { archivedAt: old }],
156 ["deleted", { deletedAt: old }],
157 ["settled", { settledOverride: "settled" }],
158 ["unsettled", { settledOverride: "active" }],
159 ["pinned", { pinnedAt: old }],
160 ["auto-settle-disabled", { autoSettleDisabledAt: old }],
161 ] satisfies ReadonlyArray<readonly [string, Partial<OrchestrationV2AppThread>]>) {
162 yield* createRun(yield* createThread(name, overrides));
163 }
164 for (const status of ["preparing", "starting", "running", "waiting"] as const) {
165 yield* createRun(yield* createThread(status), status);
166 }
167 const queued = yield* createThread("queued");
168 yield* createRun(queued, "queued");
169 const blocked = yield* createThread("blocked");
170 yield* store.apply({
171 id: EventId.make("event:settlement:request"),
172 type: "runtime-request.updated",
173 threadId: blocked,
174 occurredAt: old,
175 payload: {
176 id: RuntimeRequestId.make("request:settlement"),
177 nodeId: NodeId.make("node:settlement"),
178 providerTurnId: null,
179 nativeRequestRef: null,
180 kind: "user_input",
181 status: "pending",
182 responseCapability: { type: "not_resumable", reason: "Process stopped" },
183 createdAt: old,
184 resolvedAt: null,
185 },
186 });
187 const recent = yield* createThread("recent-message");
188 yield* createRun(recent);
189 yield* store.apply({
190 id: EventId.make("event:settlement:message"),
191 type: "message.updated",
192 threadId: recent,
193 occurredAt: now,
194 payload: {
195 createdBy: "user",
196 creationSource: "web",
197 id: MessageId.make("message:settlement:recent"),
198 threadId: recent,
199 runId: null,
200 nodeId: null,
201 role: "user",
202 text: "Next turn",
203 attachments: [],
204 streaming: false,
205 createdAt: now,
206 updatedAt: now,
207 },
208 });
209 yield* createRun(
210 yield* createThread("snoozed", {
211 snoozedUntil: DateTime.add(now, { days: 1 }),
212 snoozedAt: now,
213 }),
214 );
215 const woke = yield* createThread("woke", {
216 snoozedUntil: DateTime.add(now, { days: 1 }),
217 snoozedAt: DateTime.subtract(old, { days: 1 }),
218 });
219 yield* createRun(woke);
220 const background = yield* createThread("idle-background");
221 yield* createItem(background, yield* createRun(background), "idle");
222 const persistent = yield* createThread("persistent-monitor");
223 yield* createItem(persistent, yield* createRun(persistent), "running", true);
224 const rolledBack = yield* createThread("rolled-back-background");
225 yield* createItem(rolledBack, yield* createRun(rolledBack, "rolled_back"), "running");
226 yield* createRun(rolledBack, "completed", 2);
227 const roster = yield* createThread("provider-roster");
228 yield* createRun(roster);
229 yield* store.apply({
230 id: EventId.make("event:settlement:roster"),
231 type: "provider-thread.updated",
232 threadId: roster,
233 occurredAt: old,
234 payload: {
235 id: ProviderThreadId.make("provider-thread:settlement"),
236 appThreadId: roster,
237 ownerNodeId: null,
238 driver: ProviderDriverKind.make("codex"),
239 providerInstanceId,
240 providerSessionId: null,
241 nativeThreadRef: null,
242 nativeConversationHeadRef: null,
243 status: "idle",
244 firstRunOrdinal: null,
245 lastRunOrdinal: null,
246 handoffIds: [],
247 forkedFrom: null,
248 createdAt: old,
249 updatedAt: old,
250 pendingBackgroundTasks: [{ taskId: "running-task", kind: "subagent" }],
251 },
252 });
253 const candidates = yield* store.getSettlementCandidates();
254 const shell = yield* store.getShellSnapshot({ location: "active" });
255 const eligible = candidates.filter((thread) =>
256 isAutoSettlementCandidate(thread, DateTime.toEpochMillis(now)),
257 );
258 assert.deepEqual(
259 new Set(eligible.map((thread) => thread.id)),
260 new Set([idle, completed, queued, woke, background, persistent, rolledBack]),
261 );
262 assert.deepEqual(
263 new Set(eligible.map((thread) => thread.id)),
264 new Set(
265 shell.threads
266 .filter((thread) => isAutoSettlementCandidate(thread, DateTime.toEpochMillis(now)))
267 .map((thread) => thread.id),
268 ),
269 );
270 for (const candidate of candidates) {
271 const expected = shell.threads.find((thread) => thread.id === candidate.id)!;
272 assert.deepEqual(candidate.pendingBackgroundTasks, expected.pendingBackgroundTasks);
273 const settings = {
274 pullRequest: null,
275 nowMs: DateTime.toEpochMillis(now),
276 autoSettleAfterDays: 7,
277 autoSettleOnMerge: false,
278 };
279 assert.deepEqual(
280 resolveAutoSettlementAt({ ...settings, thread: candidate }),
281 resolveAutoSettlementAt({
282 ...settings,
283 thread: {
284 ...expected,
285 latestUserAuthoredMessageAt: candidate.latestUserAuthoredMessageAt,
286 },
287 }),
288 );
289 }
290 assert.equal(
291 DateTime.formatIso((yield* store.getThread(completed)).createdAt),
292 DateTime.formatIso(old),
293 );
294 assert.equal(
295 (yield* store.getThread(ThreadId.make("missing")).pipe(Effect.flip))._tag,
296 "ProjectionStoreThreadNotFoundError",
297 );
298 }).pipe(Effect.provide(testLayer)),
299);
300
301it.effect.each([
302 ["sql", layerSql],
303 ["memory", ProjectionStore.layerMemory],
304] as const)("%s: the user-authored message time ignores agent notifications", ([, testLayer]) =>
305 Effect.gen(function* () {
306 const store = yield* ProjectionStore.ProjectionStoreV2;
307 const threadId = yield* createThread("agent-woken");
308 yield* createRun(threadId);
309 const message = (
310 id: string,
311 createdBy: "user" | "agent",
312 creationSource: "web" | "provider",
313 at: DateTime.Utc,
314 ) =>
315 store.apply({
316 id: EventId.make(`event:settlement:${id}`),
317 type: "message.updated",
318 threadId,
319 occurredAt: at,
320 payload: {
321 createdBy,
322 creationSource,
323 id: MessageId.make(`message:settlement:${id}`),
324 threadId,
325 runId: null,
326 nodeId: null,
327 role: "user",
328 text: id,
329 attachments: [],
330 streaming: false,
331 createdAt: at,
332 updatedAt: at,
333 },
334 });
335 const written = DateTime.subtract(now, { hours: 2 });
336 yield* message("written", "user", "web", written);
337 yield* message("notification", "agent", "provider", now);
338
339 const [candidate] = yield* store.getSettlementCandidates(threadId);
340 assert.isDefined(candidate);
341 assert.equal(DateTime.formatIso(candidate.latestUserMessageAt!), DateTime.formatIso(now));
342 assert.equal(
343 DateTime.formatIso(candidate.latestUserAuthoredMessageAt!),
344 DateTime.formatIso(written),
345 );
346 // The shell carries the same stamp, which orders the Working section.
347 const shellThread = (yield* store.getShellSnapshot()).threads.find(
348 (thread) => thread.id === threadId,
349 );
350 assert.equal(
351 DateTime.formatIso(shellThread!.latestUserAuthoredMessageAt!),
352 DateTime.formatIso(written),
353 );
354 }).pipe(Effect.provide(testLayer)),
355);
356
357const pullRequestLink = (number: number) => ({
358 host: "github.com",
359 repository: "owner/repository",
360 number,
361 url: `https://github.com/owner/repository/pull/${number}`,
362 source: "manual" as const,
363 linkedAt: DateTime.formatIso(old),
364 snapshot: null,
365 stack: null,
366});
367
368it.effect.each([
369 ["sql", layerSql],
370 ["memory", ProjectionStore.layerMemory],
371] as const)(
372 "%s: lists only active threads with pull request links, oldest first",
373 ([, testLayer]) =>
374 Effect.gen(function* () {
375 const store = yield* ProjectionStore.ProjectionStoreV2;
376 yield* createThread("no-links");
377 yield* createThread("empty-links", { pullRequests: [] });
378 yield* createThread("archived-link", {
379 archivedAt: old,
380 pullRequests: [pullRequestLink(1)],
381 });
382 const settled = yield* createThread("settled-link", {
383 settledOverride: "settled",
384 settledAt: old,
385 updatedAt: DateTime.subtract(now, { days: 12 }),
386 pullRequests: [pullRequestLink(2)],
387 });
388 const open = yield* createThread("open-link", {
389 pullRequests: [pullRequestLink(3), pullRequestLink(4)],
390 });
391
392 const threads = yield* store.getThreadsWithPullRequests();
393 assert.deepEqual(
394 threads.map((thread) => [thread.id, thread.settledOverride, thread.pullRequests?.length]),
395 [
396 [settled, "settled", 1],
397 [open, null, 2],
398 ],
399 );
400 assert.deepEqual(
401 (yield* store.getThreadsWithPullRequests(open)).map((thread) => thread.id),
402 [open],
403 );
404 }).pipe(Effect.provide(testLayer)),
405);
406
407it.effect.each([
408 ["sql", layerSql],
409 ["memory", ProjectionStore.layerMemory],
410] as const)("%s: an unsettled-only shell read skips settled threads", ([, testLayer]) =>
411 Effect.gen(function* () {
412 const store = yield* ProjectionStore.ProjectionStoreV2;
413 const open = yield* createThread("unsettled-open");
414 const reopened = yield* createThread("unsettled-reopened", { settledOverride: "active" });
415 yield* createThread("unsettled-manual", { settledOverride: "settled", settledAt: old });
416 yield* createThread("unsettled-auto", { settledAt: old });
417 yield* createThread("unsettled-archived", { archivedAt: old });
418
419 const shell = yield* store.getShellSnapshot({ location: "active", unsettledOnly: true });
420 assert.deepEqual(new Set(shell.threads.map((thread) => thread.id)), new Set([open, reopened]));
421 assert.equal(shell.archivedThreads.length, 0);
422 const all = yield* store.getShellSnapshot({ location: "active" });
423 assert.equal(all.threads.length, 4);
424 }).pipe(Effect.provide(testLayer)),
425);
426
427it.effect(
428 "reads settlement candidates and thread metadata without loading historical or archived payloads",
429 () =>
430 Effect.gen(function* () {
431 const store = yield* ProjectionStore.ProjectionStoreV2;
432 const sql = yield* SqlClient.SqlClient;
433 const source = yield* createThread("historical-source", { archivedAt: old });
434 const sourceRun = yield* createRun(source);
435 const threadId = yield* createThread("fork", {
436 forkedFrom: { type: "run", threadId: source, runId: sourceRun },
437 });
438 const runId = yield* createRun(threadId);
439 yield* createItem(threadId, runId, "completed");
440 yield* sql`UPDATE orchestration_v2_projection_turn_items SET payload_json = '{broken' WHERE thread_id = ${threadId}`;
441 yield* sql`UPDATE orchestration_v2_projection_runs SET payload_json = '{broken' WHERE thread_id = ${source}`;
442 yield* sql`INSERT INTO orchestration_v2_projection_turn_items
443 SELECT 'history:' || value, i.thread_id, i.run_id, i.node_id, i.provider_thread_id, i.provider_turn_id,
444 i.parent_item_id, i.ordinal + value, i.type, i.status, i.updated_at, i.payload_json
445 FROM orchestration_v2_projection_turn_items i, json_each(${encodeNumbers(Array.from({ length: 2000 }, (_, i) => i + 1))})
446 WHERE i.thread_id = ${threadId}`;
447 const queries: Array<readonly [string, ReadonlyArray<unknown>]> = [];
448 const record: Statement.Transformer = (statement) =>
449 Effect.sync(() => {
450 queries.push(statement.compile());
451 return statement;
452 });
453 const candidates = yield* store
454 .getSettlementCandidates()
455 .pipe(Effect.provideService(Statement.CurrentTransformer, record));
456 assert.deepEqual(
457 candidates.map((thread) => thread.id),
458 [threadId],
459 );
460 assert.isTrue(queries.every(([query]) => !query.includes("COUNT(")));
461 const pendingQuery = queries.find(([query]) =>
462 query.includes("FROM orchestration_v2_projection_turn_items"),
463 );
464 assert.isDefined(pendingQuery);
465 const metadata = yield* store.getThread(threadId);
466 assert.equal(metadata.id, threadId);
467 assert.equal((yield* store.getThreadProjection(threadId).pipe(Effect.exit))._tag, "Failure");
468 const plan = yield* sql.unsafe<{ readonly detail: string }>(
469 `EXPLAIN QUERY PLAN ${pendingQuery![0]}`,
470 pendingQuery![1],
471 );
472 assert.isTrue(plan.some((row) => row.detail.includes("turn_items_shell_pending_idx")));
473 const countsPlan = yield* sql<{ readonly detail: string }>`EXPLAIN QUERY PLAN
474 SELECT thread_id, run_id, COUNT(*) FROM orchestration_v2_projection_turn_items
475 WHERE thread_id = ${threadId} AND run_id IS NOT NULL GROUP BY thread_id, run_id`;
476 assert.isTrue(
477 countsPlan.some((row) =>
478 row.detail.includes(
479 "COVERING INDEX orchestration_v2_projection_turn_items_thread_run_idx",
480 ),
481 ),
482 );
483 assert.isFalse(countsPlan.some((row) => row.detail.includes("TEMP B-TREE")));
484 const messagePlan = yield* sql<{ readonly detail: string }>`EXPLAIN QUERY PLAN
485 SELECT updated_at FROM orchestration_v2_projection_messages
486 WHERE thread_id = ${threadId} AND role = 'user' ORDER BY updated_at DESC, message_id DESC LIMIT 1`;
487 assert.isTrue(messagePlan.some((row) => row.detail.includes("messages_latest_user_idx")));
488 assert.isFalse(messagePlan.some((row) => row.detail.includes("TEMP B-TREE")));
489 }).pipe(Effect.provide(layerSql)),
490);
491
492it.effect("shell failure lookups stay on the thread's own turn items", () =>
493 Effect.gen(function* () {
494 const store = yield* ProjectionStore.ProjectionStoreV2;
495 const sql = yield* SqlClient.SqlClient;
496 const failed = yield* createThread("failed-latest");
497 yield* createRun(failed, "failed");
498 const queries: Array<readonly [string, ReadonlyArray<unknown>]> = [];
499 const record: Statement.Transformer = (statement) =>
500 Effect.sync(() => {
501 queries.push(statement.compile());
502 return statement;
503 });
504 const shell = yield* store
505 .getShellSnapshot()
506 .pipe(Effect.provideService(Statement.CurrentTransformer, record));
507 assert.deepEqual(
508 shell.threads.map((thread) => thread.id),
509 [failed],
510 );
511 const shellQuery = queries.find(([query]) =>
512 query.includes("AS blocking_failure_payload_json"),
513 );
514 assert.isDefined(shellQuery);
515 const plan = yield* sql.unsafe<{ readonly detail: string }>(
516 `EXPLAIN QUERY PLAN ${shellQuery![0]}`,
517 shellQuery![1],
518 );
519 // A failed run's root node is often null, and every runless item shares that
520 // node_id, so a node_ordinal lookup walks the whole history once per thread.
521 const itemLookups = plan.filter((row) => row.detail.startsWith("SEARCH item "));
522 assert.lengthOf(itemLookups, 2);
523 assert.isTrue(itemLookups.every((row) => row.detail.includes("turn_items_failed_error_idx")));
524 // The pending secret request lookup is bounded the same way.
525 const secretLookups = plan.filter((row) => row.detail.startsWith("SEARCH secret "));
526 assert.lengthOf(secretLookups, 1);
527 assert.include(secretLookups[0]!.detail, "turn_items_thread_run_idx");
528 }).pipe(Effect.provide(layerSql)),
529);