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