1import {
2 CommandId,
3 IDLE_COMPACTION_TEXT,
4 MessageId,
5 resolveIdleCompaction,
6 resolveProjectProviderInstanceEnabled,
7 type RunId,
8 type ThreadId,
9} from "@t3tools/contracts";
10import * as DateTime from "effect/DateTime";
11import * as Duration from "effect/Duration";
12import * as Effect from "effect/Effect";
13import * as FiberMap from "effect/FiberMap";
14import * as Stream from "effect/Stream";
15
16import { forkParked } from "../serverActivation.ts";
17import * as ServerSettings from "../serverSettings.ts";
18import { idleCompactionWindow } from "./IdleCompaction.ts";
19import * as Orchestrator from "./Orchestrator.ts";
20import * as ProjectionStore from "./ProjectionStore.ts";
21
22/**
23 * Beta: when a run completes on a Claude instance set to compact before the
24 * cache expires, waits until shortly before its prompt cache expires and sends
25 * `/compact` as a visible T3 Code turn. Timers live in
26 * memory, so a server restart skips compactions that were still waiting.
27 */
28export const make = Effect.gen(function* () {
29 const orchestrator = yield* Orchestrator.OrchestratorV2;
30 const projections = yield* ProjectionStore.ProjectionStoreV2;
31 const settings = yield* ServerSettings.ServerSettingsService;
32
33 const compactWhenDue = Effect.fn("IdleCompactionService.compactWhenDue")(
34 function* (threadId: ThreadId, completedRunId: RunId) {
35 let runId = completedRunId;
36 // Loops after the wait or a retarget: the thread may have moved on.
37 while (true) {
38 const current = yield* settings.getSettings;
39 const thread = yield* projections.getThread(threadId);
40 const claudeInstance = Object.hasOwn(current.providerInstances, thread.providerInstanceId)
41 ? current.providerInstances[thread.providerInstanceId]?.driver === "claudeAgent"
42 : thread.providerInstanceId === "claudeAgent";
43 if (
44 !claudeInstance ||
45 resolveIdleCompaction(current, thread.providerInstanceId) !== "beforeCacheExpires" ||
46 !resolveProjectProviderInstanceEnabled(
47 current,
48 thread.projectId,
49 thread.providerInstanceId,
50 )
51 )
52 return;
53 const activeProviderThreadIds =
54 thread.activeProviderThreadId === null ? [] : [thread.activeProviderThreadId];
55 const records = yield* projections.getThreadRecords(
56 threadId,
57 [
58 "runs",
59 "messages",
60 "providerThreads",
61 "providerTurns",
62 "runtimeRequests",
63 "turnItems",
64 "contextTransfers",
65 ],
66 {
67 messageRunIds: [runId],
68 turnItemStatuses: ["pending", "running", "waiting"],
69 // The latest run, active runs that hold background work, and rolled-back runs whose
70 // items it ignores.
71 runStatuses: ["preparing", "starting", "running", "rolled_back"],
72 includeLatestRun: true,
73 // Without an active provider thread there is no session to compact.
74 providerThreadIds: activeProviderThreadIds,
75 providerTurnThreadIds: activeProviderThreadIds,
76 latestProviderTurnUsageOnly: true,
77 runtimeRequestStatuses: ["pending"],
78 contextTransferStatuses: ["pending"],
79 },
80 );
81 if (
82 resolveIdleCompaction(current, records.thread.providerInstanceId) !==
83 "beforeCacheExpires" ||
84 !resolveProjectProviderInstanceEnabled(
85 current,
86 records.thread.projectId,
87 records.thread.providerInstanceId,
88 )
89 )
90 return;
91 const latestRun = records.runs
92 .filter((run) => run.status !== "cancelled" || run.startedAt !== null)
93 .reduce<(typeof records.runs)[number] | undefined>(
94 (latest, run) => (latest === undefined || run.ordinal > latest.ordinal ? run : latest),
95 undefined,
96 );
97 // A late update for an earlier run replaced the latest run's timer, so take it over.
98 if (latestRun?.status === "completed" && latestRun.id !== runId) {
99 runId = latestRun.id;
100 continue;
101 }
102 const nowMs = DateTime.toEpochMillis(yield* DateTime.now);
103 const window = idleCompactionWindow(records, nowMs);
104 if (window?.runId !== runId) return;
105 if (nowMs < window.startMs) {
106 yield* Effect.sleep(Duration.millis(window.startMs - nowMs));
107 continue;
108 }
109 yield* orchestrator.dispatch({
110 type: "message.dispatch",
111 // One compaction per completed run, even across duplicate events.
112 commandId: CommandId.make(`server:idle-compact:${threadId}:${runId}`),
113 messageId: MessageId.make(`idle-compact:${threadId}:${runId}`),
114 threadId,
115 text: IDLE_COMPACTION_TEXT,
116 attachments: [],
117 createdBy: "agent",
118 creationSource: "server",
119 idleCompaction: "beforeCacheExpires",
120 dispatchMode: { type: "start_immediately" },
121 });
122 return;
123 }
124 },
125 (effect, threadId) =>
126 effect.pipe(
127 Effect.catch((cause) => Effect.logWarning("idle compaction skipped", { threadId, cause })),
128 ),
129 );
130
131 const start = Effect.fn("IdleCompactionService.start")(function* () {
132 // Keyed by thread: a newer completed run replaces the waiting timer.
133 const timers = yield* FiberMap.make<ThreadId>();
134 yield* forkParked(
135 Stream.runForEach(orchestrator.streamDomainEvents, (event) =>
136 event.type === "run.updated" && event.payload.status === "completed"
137 ? FiberMap.run(timers, event.threadId, compactWhenDue(event.threadId, event.payload.id))
138 : Effect.void,
139 ),
140 );
141 });
142 return { start };
143});