1import { describe, expect, it } from "@effect/vitest";
2import {
3 DEFAULT_SERVER_SETTINGS,
4 EventId,
5 MessageId,
6 type OrchestrationV2ThreadProjection,
7 NodeId,
8 ProjectId,
9 ProviderDriverKind,
10 ProviderInstanceId,
11 ProviderThreadId,
12 ProviderTurnId,
13 RunId,
14 ThreadId,
15} from "@t3tools/contracts";
16import * as DateTime from "effect/DateTime";
17import * as Deferred from "effect/Deferred";
18import * as Effect from "effect/Effect";
19import * as Layer from "effect/Layer";
20import * as Stream from "effect/Stream";
21import * as Tracer from "effect/Tracer";
22import * as TestClock from "effect/testing/TestClock";
23
24import * as ServerSettings from "../serverSettings.ts";
25import * as IdleCompactionService from "./IdleCompactionService.ts";
26import * as Orchestrator from "./Orchestrator.ts";
27import * as ProjectionStore from "./ProjectionStore.ts";
28
29import { idleCompactionWindow, type IdleCompactionRecords } from "./IdleCompaction.ts";
30
31const NOW_MS = Date.parse("2026-10-19T12:00:00.000Z");
32const MINUTE_MS = 60_000;
33const at = (offset = 0) => DateTime.makeUnsafe(NOW_MS + offset);
34const threadId = ThreadId.make("idle-thread");
35const runId = RunId.make("run-1");
36const instanceId = ProviderInstanceId.make("claudeAgent");
37const providerThreadId = ProviderThreadId.make("provider-thread");
38const modelSelection = { instanceId, model: "claude-opus-5-5" };
39
40function records(
41 usage: { readonly usedTokens?: number; readonly cacheExpiresInMs?: number } = {},
42): { -readonly [Key in keyof IdleCompactionRecords]: IdleCompactionRecords[Key] } {
43 const messageId = MessageId.make("message-1");
44 const nodeId = NodeId.make("root-node");
45 return {
46 thread: {
47 id: threadId,
48 projectId: ProjectId.make("project"),
49 title: "Idle thread",
50 createdBy: "user",
51 creationSource: "web",
52 providerInstanceId: instanceId,
53 modelSelection,
54 runtimeMode: "full-access",
55 interactionMode: "default",
56 branch: null,
57 worktreePath: null,
58 activeProviderThreadId: providerThreadId,
59 lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null },
60 forkedFrom: null,
61 createdAt: at(-MINUTE_MS),
62 updatedAt: at(),
63 archivedAt: null,
64 settledOverride: null,
65 settledAt: null,
66 lastVisitedAt: at(),
67 deletedAt: null,
68 },
69 runs: [
70 {
71 id: runId,
72 threadId,
73 ordinal: 1,
74 providerInstanceId: instanceId,
75 modelSelection,
76 providerThreadId,
77 userMessageId: messageId,
78 rootNodeId: nodeId,
79 activeAttemptId: null,
80 status: "completed",
81 requestedAt: at(-MINUTE_MS),
82 startedAt: at(-MINUTE_MS),
83 completedAt: at(),
84 checkpointId: null,
85 contextHandoffId: null,
86 },
87 ],
88 messages: [
89 {
90 id: messageId,
91 threadId,
92 runId,
93 nodeId,
94 role: "user",
95 text: "Explain this code",
96 attachments: [],
97 streaming: false,
98 createdBy: "user",
99 creationSource: "web",
100 createdAt: at(-MINUTE_MS),
101 updatedAt: at(-MINUTE_MS),
102 },
103 ],
104 providerThreads: [
105 {
106 id: providerThreadId,
107 driver: ProviderDriverKind.make("claudeAgent"),
108 providerInstanceId: instanceId,
109 providerSessionId: null,
110 appThreadId: threadId,
111 ownerNodeId: nodeId,
112 nativeThreadRef: null,
113 nativeConversationHeadRef: null,
114 status: "idle",
115 firstRunOrdinal: 1,
116 lastRunOrdinal: 1,
117 handoffIds: [],
118 forkedFrom: null,
119 pendingBackgroundTasks: [],
120 createdAt: at(-MINUTE_MS),
121 updatedAt: at(),
122 },
123 ],
124 providerTurns: [
125 {
126 id: ProviderTurnId.make("provider-turn-1"),
127 providerThreadId,
128 nodeId,
129 runAttemptId: null,
130 nativeTurnRef: null,
131 ordinal: 1,
132 status: "completed",
133 startedAt: at(-MINUTE_MS),
134 completedAt: at(),
135 tokenUsage: {
136 usedTokens: usage.usedTokens ?? 250_000,
137 ...(usage.cacheExpiresInMs === undefined
138 ? {}
139 : { promptCacheExpiresAt: DateTime.formatIso(at(usage.cacheExpiresInMs)) }),
140 updatedAt: DateTime.formatIso(at()),
141 },
142 },
143 ],
144 runtimeRequests: [],
145 turnItems: [],
146 contextTransfers: [],
147 };
148}
149
150describe("idleCompactionWindow", () => {
151 it("skips small contexts, unknown expiry, other providers, switched threads, compactions, and side chats", () => {
152 const cache = { cacheExpiresInMs: 60 * MINUTE_MS };
153 expect(idleCompactionWindow(records({ ...cache, usedTokens: 199_999 }), NOW_MS)).toBeNull();
154 expect(idleCompactionWindow(records(), NOW_MS)).toBeNull();
155 const codex = records(cache);
156 codex.providerThreads = [
157 { ...codex.providerThreads[0]!, driver: ProviderDriverKind.make("codex") },
158 ];
159 expect(idleCompactionWindow(codex, NOW_MS)).toBeNull();
160 const switched = records(cache);
161 switched.thread = { ...switched.thread, providerInstanceId: ProviderInstanceId.make("codex") };
162 expect(idleCompactionWindow(switched, NOW_MS)).toBeNull();
163 const compacted = records(cache);
164 compacted.messages = [{ ...compacted.messages[0]!, text: "/compact" }];
165 expect(idleCompactionWindow(compacted, NOW_MS)).toBeNull();
166 const steered = records(cache);
167 steered.runs = [
168 ...steered.runs,
169 {
170 ...steered.runs[0]!,
171 id: RunId.make("run-2"),
172 ordinal: 2,
173 userMessageId: MessageId.make("message-2"),
174 status: "cancelled",
175 startedAt: null,
176 },
177 ];
178 expect(idleCompactionWindow(steered, NOW_MS)?.runId).toBe(runId);
179 // A turn Claude itself reported as cancelled did run, so it stays the latest.
180 steered.runs = steered.runs.map((run) => ({ ...run, startedAt: at(-MINUTE_MS) }));
181 expect(idleCompactionWindow(steered, NOW_MS)).toBeNull();
182 const archived = records(cache);
183 archived.thread = { ...archived.thread, archivedAt: at() };
184 expect(idleCompactionWindow(archived, NOW_MS)).toBeNull();
185 const sideChat = records(cache);
186 expect(idleCompactionWindow(sideChat, NOW_MS)).not.toBeNull();
187 sideChat.thread = {
188 ...sideChat.thread,
189 lineage: { ...sideChat.thread.lineage, relationshipToParent: "fork", sideChat: true },
190 };
191 expect(idleCompactionWindow(sideChat, NOW_MS)).toBeNull();
192 });
193});
194
195describe("IdleCompactionService", () => {
196 it.effect.each([
197 { mode: "onReturn", reads: 0 },
198 { mode: "off", reads: 0 },
199 { mode: "beforeCacheExpires", reads: 1 },
200 { mode: "beforeCacheExpires", override: "off", reads: 0 },
201 { mode: "onReturn", override: "beforeCacheExpires", reads: 1 },
202 { mode: "beforeCacheExpires", projectDisabled: true, reads: 0 },
203 { mode: "beforeCacheExpires", enabled: false, reads: 0 },
204 { mode: "beforeCacheExpires", driver: "codex", instance: "codex", reads: 0 },
205 { mode: "beforeCacheExpires", driver: "codex", instance: "custom-codex", reads: 0 },
206 { mode: "beforeCacheExpires", driver: "claudeAgent", instance: "custom-claude", reads: 1 },
207 { mode: "beforeCacheExpires", unconfigured: true, reads: 1 },
208 ] as const)(
209 "loads history only for eligible idle compaction: $mode/$override/$enabled/$projectDisabled/$instance/$unconfigured",
210 (scenario) => {
211 const test = {
212 override: "default",
213 enabled: true,
214 projectDisabled: false,
215 ...scenario,
216 } as const;
217 return Effect.scoped(
218 Effect.gen(function* () {
219 yield* TestClock.setTime(NOW_MS);
220 const selectedInstance = ProviderInstanceId.make(
221 "instance" in test ? test.instance : "claudeAgent",
222 );
223 const driver = ProviderDriverKind.make("driver" in test ? test.driver : "claudeAgent");
224 const base = records({ cacheExpiresInMs: 2 * MINUTE_MS });
225 const data: OrchestrationV2ThreadProjection = {
226 ...base,
227 thread: {
228 ...base.thread,
229 providerInstanceId: selectedInstance,
230 modelSelection: { ...base.thread.modelSelection, instanceId: selectedInstance },
231 },
232 runs: base.runs.map((run) => ({
233 ...run,
234 providerInstanceId: selectedInstance,
235 modelSelection: { ...run.modelSelection, instanceId: selectedInstance },
236 })),
237 providerThreads: base.providerThreads.map((thread) => ({
238 ...thread,
239 driver,
240 providerInstanceId: selectedInstance,
241 })),
242 attempts: [],
243 nodes: [],
244 subagents: [],
245 providerSessions: [],
246 plans: [],
247 checkpointScopes: [],
248 checkpoints: [],
249 contextHandoffs: [],
250 visibleTurnItems: [],
251 updatedAt: at(),
252 };
253 let historyReads = 0;
254 let dispatches = 0;
255 const done = yield* Deferred.make<void>();
256 const tracer = Tracer.make({
257 span(options) {
258 const span = new Tracer.NativeSpan(options);
259 const end = span.end.bind(span);
260 span.end = (...args) => {
261 end(...args);
262 if (span.name === "IdleCompactionService.compactWhenDue")
263 Deferred.doneUnsafe(done, Effect.void);
264 };
265 return span;
266 },
267 });
268 const dependencies = Layer.mergeAll(
269 Layer.mock(Orchestrator.OrchestratorV2)({
270 streamDomainEvents: Stream.make({
271 id: EventId.make("idle-completed"),
272 type: "run.updated",
273 threadId,
274 runId,
275 occurredAt: at(),
276 payload: data.runs[0]!,
277 }),
278 dispatch: () =>
279 Effect.sync(() => {
280 dispatches++;
281 return { sequence: 1, storedEvents: [] };
282 }),
283 }),
284 Layer.mock(ProjectionStore.ProjectionStoreV2)({
285 getThread: () => Effect.succeed(data.thread),
286 getThreadRecords: () =>
287 Effect.sync(() => {
288 historyReads++;
289 return data;
290 }),
291 }),
292 Layer.mock(ServerSettings.ServerSettingsService)({
293 getSettings: Effect.succeed({
294 ...DEFAULT_SERVER_SETTINGS,
295 idleCompaction: test.mode,
296 providerInstances:
297 "unconfigured" in test
298 ? {}
299 : {
300 [selectedInstance]: {
301 driver,
302 enabled: test.enabled,
303 config: { idleCompaction: test.override },
304 },
305 },
306 projectSettingsOverrides: test.projectDisabled
307 ? {
308 [data.thread.projectId]: {
309 providerInstances: { [selectedInstance]: { enabled: false } },
310 },
311 }
312 : {},
313 }),
314 }),
315 );
316 yield* Effect.gen(function* () {
317 const service = yield* IdleCompactionService.make;
318 yield* service.start();
319 yield* Deferred.await(done);
320 }).pipe(Effect.provide(dependencies), Effect.withTracer(tracer));
321 expect(historyReads).toBe(test.reads);
322 expect(dispatches).toBe(test.reads);
323 }),
324 );
325 },
326 );
327
328 it.effect("keeps the latest run's compaction when an earlier run reports completed again", () =>
329 Effect.scoped(
330 Effect.gen(function* () {
331 yield* TestClock.setTime(NOW_MS);
332 const base = records({ cacheExpiresInMs: 10 * MINUTE_MS });
333 const laterRunId = RunId.make("run-2");
334 const laterMessageId = MessageId.make("message-2");
335 const earlierRun = base.runs[0]!;
336 const laterRun = {
337 ...earlierRun,
338 id: laterRunId,
339 ordinal: 2,
340 userMessageId: laterMessageId,
341 };
342 const cancelledRun = {
343 ...earlierRun,
344 id: RunId.make("run-3"),
345 ordinal: 3,
346 userMessageId: MessageId.make("message-3"),
347 status: "cancelled" as const,
348 startedAt: null,
349 };
350 const data: OrchestrationV2ThreadProjection = {
351 ...base,
352 runs: [earlierRun, laterRun, cancelledRun],
353 messages: [
354 ...base.messages,
355 { ...base.messages[0]!, id: laterMessageId, runId: laterRunId, text: "Keep going" },
356 ],
357 providerThreads: [{ ...base.providerThreads[0]!, lastRunOrdinal: 2 }],
358 attempts: [],
359 nodes: [],
360 subagents: [],
361 providerSessions: [],
362 plans: [],
363 checkpointScopes: [],
364 checkpoints: [],
365 contextHandoffs: [],
366 visibleTurnItems: [],
367 updatedAt: at(),
368 };
369 const completed = (run: typeof earlierRun) => ({
370 id: EventId.make(`completed-${run.id}`),
371 type: "run.updated" as const,
372 threadId,
373 runId: run.id,
374 occurredAt: at(),
375 payload: run,
376 });
377 const dispatched: Array<string> = [];
378 const streamDone = yield* Deferred.make<void>();
379 const dependencies = Layer.mergeAll(
380 Layer.mock(Orchestrator.OrchestratorV2)({
381 // A delegated completion on the earlier run lands after the later run completed.
382 streamDomainEvents: Stream.make(completed(laterRun), completed(earlierRun)).pipe(
383 Stream.ensuring(Deferred.succeed(streamDone, undefined)),
384 ),
385 dispatch: (command) =>
386 Effect.sync(() => {
387 dispatched.push(command.commandId);
388 return { sequence: 1, storedEvents: [] };
389 }),
390 }),
391 Layer.mock(ProjectionStore.ProjectionStoreV2)({
392 getThread: () => Effect.succeed(data.thread),
393 getThreadRecords: (_threadId, _keys, options) =>
394 Effect.succeed({
395 ...data,
396 messages: data.messages.filter(
397 (message) =>
398 options?.messageRunIds === undefined ||
399 (message.runId !== null && options.messageRunIds.includes(message.runId)),
400 ),
401 }),
402 }),
403 Layer.mock(ServerSettings.ServerSettingsService)({
404 getSettings: Effect.succeed({
405 ...DEFAULT_SERVER_SETTINGS,
406 idleCompaction: "beforeCacheExpires",
407 }),
408 }),
409 );
410 yield* Effect.gen(function* () {
411 const service = yield* IdleCompactionService.make;
412 yield* service.start();
413 yield* Deferred.await(streamDone);
414 yield* TestClock.adjust(9 * MINUTE_MS);
415 }).pipe(Effect.provide(dependencies));
416 expect(dispatched).toEqual(["server:idle-compact:idle-thread:run-2"]);
417 }),
418 ),
419 );
420});