History

refactor(server): let adapters declare context compaction (#10112)

Theo Browne committed 2 months ago

Commit 5fa35d21 parent e4e9fa911 files · +185 −36Browse files

Unified · Split

11 files
  1. apps/server/src/provider/Layers/AntigravityAdapter.ts +1 −0
  2. apps/server/src/provider/Layers/ClaudeAdapter.ts +1 −0
  3. apps/server/src/provider/Layers/CodexAdapter.test.ts +2 −1
  4. apps/server/src/provider/Layers/CodexAdapter.ts +7 −9
  5. apps/server/src/provider/Layers/CursorAdapter.ts +1 −0
  6. apps/server/src/provider/Layers/GrokAdapter.ts +1 −0
  7. apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +2 −1
  8. apps/server/src/provider/Layers/OpenCodeAdapter.ts +6 −4
  9. apps/server/src/provider/Layers/ProviderService.test.ts +127 −2
  10. apps/server/src/provider/Layers/ProviderService.ts +20 −15
  11. apps/server/src/provider/Services/ProviderAdapter.ts +17 −4
apps/server/src/provider/Layers/AntigravityAdapter.ts+1−0
@@ -1245,6 +1245,7 @@ export const makeAntigravityAdapter = Effect.fn("makeAntigravityAdapter")(functi · 1244 unmodified lines
12451245 return {
12461246 provider: PROVIDER,
12471247 capabilities: { sessionModelSwitch: "in-session", supportsConversationRollback: false },
1248+ compaction: { type: "slash-command", command: "/compact" },
12481249 startSession,
12491250 sendTurn,
12501251 interruptTurn,
apps/server/src/provider/Layers/ClaudeAdapter.ts+1−0
@@ -5075,6 +5075,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( · 5074 unmodified lines
50755075 capabilities: {
50765076 sessionModelSwitch: "in-session",
50775077 },
5078+ compaction: { type: "slash-command", command: "/compact" },
50785079 startSession,
50795080 sendTurn,
50805081 interruptTurn,
apps/server/src/provider/Layers/CodexAdapter.test.ts+2−1
@@ -353,7 +353,8 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { · 352 unmodified lines
353353 Stream.runHead,
354354 Effect.forkChild,
355355 );
356− yield* adapter.compactThread!(threadId);
356+ NodeAssert.ok(adapter.compaction?.type === "native");
357+ yield* adapter.compaction.start(threadId);
357358 yield* runtime.emit({
358359 id: asEventId("evt-compaction-item-completed"),
359360 kind: "notification",
apps/server/src/provider/Layers/CodexAdapter.ts+7−9
@@ -2504,14 +2504,12 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( · 2503 unmodified lines
25042504 ),
25052505 );
25062506
2507− const compactThread: NonNullable<CodexAdapterShape["compactThread"]> = Effect.fn("compactThread")(
2508− function* (threadId) {
2509− const session = yield* requireSession(threadId);
2510− yield* session.runtime.compactThread.pipe(
2511− Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/compact/start", cause)),
2512− );
2513− },
2514− );
2507+ const compactThread = Effect.fn("compactThread")(function* (threadId: ThreadId) {
2508+ const session = yield* requireSession(threadId);
2509+ yield* session.runtime.compactThread.pipe(
2510+ Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/compact/start", cause)),
2511+ );
2512+ });
25152513
25162514 const readThread: CodexAdapterShape["readThread"] = (threadId) =>
25172515 requireSession(threadId).pipe(
@@ -2658,7 +2656,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( · 140 unmodified lines
26582656 },
26592657 startSession,
26602658 sendTurn,
2661− compactThread,
2659+ compaction: { type: "native", start: compactThread },
26622660 interruptTurn,
26632661 readThread,
26642662 rollbackThread,
apps/server/src/provider/Layers/CursorAdapter.ts+1−0
@@ -1212,6 +1212,7 @@ export function makeCursorAdapter( · 1211 unmodified lines
12121212 return {
12131213 provider: PROVIDER,
12141214 capabilities: { sessionModelSwitch: "in-session" },
1215+ compaction: { type: "slash-command", command: "/compress" },
12151216 startSession,
12161217 sendTurn,
12171218 interruptTurn,
apps/server/src/provider/Layers/GrokAdapter.ts+1−0
@@ -2128,6 +2128,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte · 2127 unmodified lines
21282128 return {
21292129 provider: PROVIDER,
21302130 capabilities: { sessionModelSwitch: "in-session" },
2131+ compaction: { type: "slash-command", command: "/compact" },
21312132 startSession,
21322133 sendTurn,
21332134 interruptTurn,
apps/server/src/provider/Layers/OpenCodeAdapter.test.ts+2−1
@@ -1080,7 +1080,8 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { · 1079 unmodified lines
10801080 threadId,
10811081 runtimeMode: "full-access",
10821082 });
1083− yield* adapter.compactThread!(
1083+ NodeAssert.ok(adapter.compaction?.type === "native");
1084+ yield* adapter.compaction.start(
10841085 threadId,
10851086 createModelSelection(ProviderInstanceId.make("opencode"), "openai/gpt-5"),
10861087 );
apps/server/src/provider/Layers/OpenCodeAdapter.ts+6−4
@@ -4,6 +4,7 @@ import { · 3 unmodified lines
44 ProviderDriverKind,
55 ProviderInstanceId,
66 type ProviderRuntimeEvent,
7+ type ProviderSendTurnInput,
78 type ProviderSession,
89 RuntimeItemId,
910 RuntimeRequestId,
@@ -3426,9 +3427,10 @@ export function makeOpenCodeAdapter( · 3416 unmodified lines
34263427 );
34273428 });
34283429
3429− const compactThread: NonNullable<OpenCodeAdapterShape["compactThread"]> = Effect.fn(
3430− "compactThread",
3431− )(function* (threadId, requestedModelSelection) {
3430+ const compactThread = Effect.fn("compactThread")(function* (
3431+ threadId: ThreadId,
3432+ requestedModelSelection?: ProviderSendTurnInput["modelSelection"],
3433+ ) {
34323434 const context = yield* ensureSessionContext(sessions, threadId);
34333435 yield* awaitOpenCodeContextReady(context);
34343436 const modelSelection =
@@ -3843,7 +3845,7 @@ export function makeOpenCodeAdapter( · 408 unmodified lines
38433845 },
38443846 startSession,
38453847 sendTurn,
3846− compactThread,
3848+ compaction: { type: "native", start: compactThread },
38473849 interruptTurn,
38483850 respondToRequest,
38493851 respondToUserInput,
apps/server/src/provider/Layers/ProviderService.test.ts+127−2
@@ -190,7 +190,7 @@ function makeFakeCodexAdapter( · 189 unmodified lines
190190 Effect.void,
191191 );
192192
193− const compactThread = vi.fn((threadId: ThreadId) =>
193+ const compactThread = vi.fn((threadId: ThreadId): Effect.Effect<void, ProviderAdapterError> =>
194194 Effect.sync(() =>
195195 emit({
196196 type: "thread.state.changed",
@@ -278,7 +278,13 @@ function makeFakeCodexAdapter( · 81 unmodified lines
278278 },
279279 startSession,
280280 sendTurn,
281− ...(provider === CODEX_DRIVER ? { compactThread } : {}),
281+ ...(provider === CODEX_DRIVER
282+ ? { compaction: { type: "native", start: compactThread } }
283+ : provider === CURSOR_DRIVER
284+ ? { compaction: { type: "slash-command", command: "/compress" } }
285+ : provider === CLAUDE_AGENT_DRIVER
286+ ? { compaction: { type: "slash-command", command: "/compact" } }
287+ : {}),
282288 interruptTurn,
283289 respondToRequest,
284290 respondToUserInput,
@@ -981,6 +987,125 @@ it.effect("ProviderServiceLive rejects new sessions for disabled custom instance · 696 unmodified lines
981987
982988 const routing = makeProviderServiceLayer();
983989
990+const customCompactionDriver = ProviderDriverKind.make("custom-compaction-provider");
991+const nativeCompactionInstanceId = ProviderInstanceId.make("native-compaction");
992+const slashCompactionInstanceId = ProviderInstanceId.make("slash-compaction");
993+const unsupportedCompactionInstanceId = ProviderInstanceId.make("unsupported-compaction");
994+const customNativeCompaction = makeFakeCodexAdapter(customCompactionDriver);
995+const customSlashCompaction = makeFakeCodexAdapter(customCompactionDriver);
996+const unsupportedCompaction = makeFakeCodexAdapter(customCompactionDriver);
997+const declaredCompaction = makeProviderServiceLayer({
998+ registry: makeStaticInstanceRegistry([
999+ [
1000+ nativeCompactionInstanceId,
1001+ {
1002+ ...customNativeCompaction.adapter,
1003+ compaction: { type: "native", start: customNativeCompaction.compactThread },
1004+ },
1005+ ],
1006+ [
1007+ slashCompactionInstanceId,
1008+ {
1009+ ...customSlashCompaction.adapter,
1010+ compaction: { type: "slash-command", command: "/reduce-context" },
1011+ },
1012+ ],
1013+ [unsupportedCompactionInstanceId, unsupportedCompaction.adapter],
1014+ ]),
1015+});
1016+
1017+declaredCompaction.layer("ProviderService declared compaction", (it) => {
1018+ it.effect("starts declared native compaction instead of sending a prompt", () =>
1019+ Effect.gen(function* () {
1020+ const provider = yield* ProviderService.ProviderService;
1021+ const threadId = asThreadId("custom-native-compaction");
1022+ const requestId = MessageId.make("custom-native-request");
1023+ yield* provider.startSession(threadId, {
1024+ providerInstanceId: nativeCompactionInstanceId,
1025+ threadId,
1026+ runtimeMode: "full-access",
1027+ });
1028+ const compactedEventFiber = yield* provider.streamEvents.pipe(
1029+ Stream.filter(
1030+ (event) => event.threadId === threadId && event.type === "thread.state.changed",
1031+ ),
1032+ Stream.runHead,
1033+ Effect.forkChild({ startImmediately: true }),
1034+ );
1035+ yield* advanceTestClock(50);
1036+ yield* provider.compactThread(threadId, undefined, requestId);
1037+ const compacted = Option.getOrThrow(yield* Fiber.join(compactedEventFiber));
1038+ assert.equal(compacted.requestId, String(requestId));
1039+ assert.equal(customNativeCompaction.compactThread.mock.calls.length, 1);
1040+ assert.equal(customNativeCompaction.sendTurn.mock.calls.length, 0);
1041+ yield* provider.stopSession({ threadId });
1042+ }),
1043+ );
1044+
1045+ it.effect("sends the declared slash command as the compaction turn", () =>
1046+ Effect.gen(function* () {
1047+ const provider = yield* ProviderService.ProviderService;
1048+ const threadId = asThreadId("custom-slash-compaction");
1049+ const requestId = MessageId.make("custom-slash-request");
1050+ const modelSelection = createModelSelection(slashCompactionInstanceId, "custom-model");
1051+ yield* provider.startSession(threadId, {
1052+ providerInstanceId: slashCompactionInstanceId,
1053+ threadId,
1054+ runtimeMode: "full-access",
1055+ });
1056+ const compactedEventFiber = yield* provider.streamEvents.pipe(
1057+ Stream.filter(
1058+ (event) => event.threadId === threadId && event.type === "thread.state.changed",
1059+ ),
1060+ Stream.runHead,
1061+ Effect.forkChild({ startImmediately: true }),
1062+ );
1063+ const compactFiber = yield* provider
1064+ .compactThread(threadId, modelSelection, requestId)
1065+ .pipe(Effect.forkChild);
1066+ yield* advanceTestClock(50);
1067+ customSlashCompaction.emit({
1068+ type: "turn.completed",
1069+ eventId: asEventId("custom-slash-completed"),
1070+ provider: customCompactionDriver,
1071+ createdAt: "2026-01-01T00:00:01.000Z",
1072+ threadId,
1073+ turnId: asTurnId(`turn-${threadId}`),
1074+ payload: { state: "completed" },
1075+ });
1076+ yield* Fiber.join(compactFiber);
1077+ const compacted = Option.getOrThrow(yield* Fiber.join(compactedEventFiber));
1078+ assert.equal(compacted.requestId, String(requestId));
1079+ assert.equal(customSlashCompaction.compactThread.mock.calls.length, 0);
1080+ assert.equal(customSlashCompaction.sendTurn.mock.calls.length, 1);
1081+ assert.equal(customSlashCompaction.sendTurn.mock.calls[0]?.[0].input, "/reduce-context");
1082+ assert.deepEqual(
1083+ customSlashCompaction.sendTurn.mock.calls[0]?.[0].modelSelection,
1084+ modelSelection,
1085+ );
1086+ yield* provider.stopSession({ threadId });
1087+ }),
1088+ );
1089+
1090+ it.effect("rejects compaction for adapters without a declared strategy", () =>
1091+ Effect.gen(function* () {
1092+ const provider = yield* ProviderService.ProviderService;
1093+ const threadId = asThreadId("custom-unsupported-compaction");
1094+ yield* provider.startSession(threadId, {
1095+ providerInstanceId: unsupportedCompactionInstanceId,
1096+ threadId,
1097+ runtimeMode: "full-access",
1098+ });
1099+ const failure = yield* provider.compactThread(threadId).pipe(Effect.flip);
1100+ assert.instanceOf(failure, ProviderValidationError);
1101+ assert.include(failure.message, "does not support context compaction");
1102+ assert.equal(unsupportedCompaction.sendTurn.mock.calls.length, 0);
1103+ assert.equal(unsupportedCompaction.compactThread.mock.calls.length, 0);
1104+ yield* provider.stopSession({ threadId });
1105+ }),
1106+ );
1107+});
1108+
9841109 const antigravityDriver = ProviderDriverKind.make("antigravity");
9851110 const replacementAntigravity = makeFakeCodexAdapter(antigravityDriver);
9861111 const originalAntigravityInstanceId = ProviderInstanceId.make("antigravity-personal");
apps/server/src/provider/Layers/ProviderService.ts+20−15
@@ -76,6 +76,9 @@ import * as ServerSettings from "../../serverSettings.ts"; · 75 unmodified lines
7676 import * as ProjectionSnapshotQuery from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
7777 const isModelSelection = Schema.is(ModelSelection);
7878
79+/** How long a manual context compaction may run before ProviderService gives up on it. */
80+const COMPACTION_COMPLETION_TIMEOUT = "10 minutes";
81+
7982 interface PendingCompaction {
8083 readonly completion: Deferred.Deferred<string>;
8184 readonly native: boolean;
@@ -1522,18 +1525,24 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( · 1440 unmodified lines
15221525 "provider.thread_id": threadId,
15231526 });
15241527 yield* McpSessionRegistry.touchActiveMcpThread(threadId);
1525− const nativeCompaction = routed.adapter.compactThread;
1528+ const compaction = routed.adapter.compaction;
1529+ if (compaction === undefined) {
1530+ return yield* toValidationError(
1531+ "ProviderService.compactThread",
1532+ `Provider '${routed.adapter.provider}' does not support context compaction.`,
1533+ );
1534+ }
15261535 const completion = yield* Deferred.make<string>();
15271536 const pending: PendingCompaction = {
15281537 completion,
1529− native: nativeCompaction !== undefined,
1538+ native: compaction.type === "native",
15301539 providerInstanceId: routed.instanceId,
15311540 requestId,
15321541 earlyEvents: [],
15331542 compactedEventObserved: false,
15341543 expectedTurnId: undefined,
15351544 };
1536− if (nativeCompaction !== undefined && timedOutNativeCompactions.has(threadId)) {
1545+ if (compaction.type === "native" && timedOutNativeCompactions.has(threadId)) {
15371546 return yield* new ProviderAdapterRequestError({
15381547 provider: routed.adapter.provider,
15391548 method: "thread/compact",
@@ -1558,14 +1567,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( · 18 unmodified lines
15581567 pendingCompactions.delete(threadId);
15591568 }
15601569 });
1561− const nativeCompletionTimeout =
1562− routed.adapter.provider === "codex" || routed.adapter.provider === "opencode"
1563− ? "10 minutes"
1564− : "30 seconds";
15651570 const awaitNativeCompaction = (start: Effect.Effect<void, ProviderAdapterError>) =>
15661571 start.pipe(
15671572 Effect.andThen(Deferred.await(completion)),
1568− Effect.timeout(nativeCompletionTimeout),
1573+ Effect.timeout(COMPACTION_COMPLETION_TIMEOUT),
15691574 Effect.catchTag("TimeoutError", (cause) =>
15701575 Effect.sync(() => {
15711576 timedOutNativeCompactions.add(threadId);
@@ -1575,7 +1580,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( · 3 unmodified lines
15751580 new ProviderAdapterRequestError({
15761581 provider: routed.adapter.provider,
15771582 method: "thread/compact",
1578− detail: `Provider did not report completed context compaction within ${nativeCompletionTimeout}.`,
1583+ detail: `Provider did not report completed context compaction within ${COMPACTION_COMPLETION_TIMEOUT}.`,
15791584 cause,
15801585 }),
15811586 ),
@@ -1584,24 +1589,24 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( · 2 unmodified lines
15841589 ),
15851590 );
15861591 const awaitFallbackCompaction = Deferred.await(completion).pipe(
1587− Effect.timeout("10 minutes"),
1592+ Effect.timeout(COMPACTION_COMPLETION_TIMEOUT),
15881593 Effect.mapError(
15891594 (cause) =>
15901595 new ProviderAdapterRequestError({
15911596 provider: routed.adapter.provider,
15921597 method: "turn/start",
1593− detail: "Provider did not finish context compaction within 10 minutes.",
1598+ detail: `Provider did not finish context compaction within ${COMPACTION_COMPLETION_TIMEOUT}.`,
15941599 cause,
15951600 }),
15961601 ),
15971602 );
15981603 const terminal = yield* (
1599− nativeCompaction
1600− ? awaitNativeCompaction(nativeCompaction(routed.threadId, modelSelection))
1604+ compaction.type === "native"
1605+ ? awaitNativeCompaction(compaction.start(routed.threadId, modelSelection))
16011606 : Effect.gen(function* () {
16021607 const turn = yield* sendTurn({
16031608 threadId,
1604− input: routed.adapter.provider === "cursor" ? "/compress" : "/compact",
1609+ input: compaction.command,
16051610 ...(modelSelection !== undefined ? { modelSelection } : {}),
16061611 }).pipe(
16071612 Effect.onError(() =>
@@ -1621,7 +1626,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( · 13 unmodified lines
16211626 if (terminal !== "completed") {
16221627 return yield* new ProviderAdapterRequestError({
16231628 provider: routed.adapter.provider,
1624− method: nativeCompaction ? "thread/compact" : "turn/start",
1629+ method: compaction.type === "native" ? "thread/compact" : "turn/start",
16251630 detail: `Context compaction ended with ${terminal}.`,
16261631 });
16271632 }
apps/server/src/provider/Services/ProviderAdapter.ts+17−4
@@ -27,6 +27,21 @@ import type * as Stream from "effect/Stream"; · 26 unmodified lines
2727
2828 export type ProviderSessionModelSwitchMode = "in-session" | "unsupported";
2929
30+/**
31+ * How ProviderService runs manual context compaction for an adapter.
32+ * Native adapters expose a start call and must emit a compacted thread state
33+ * when they finish. Slash-command adapters get the command sent as a turn.
34+ */
35+export type ProviderCompaction<TError> =
36+ | {
37+ readonly type: "native";
38+ readonly start: (
39+ threadId: ThreadId,
40+ modelSelection?: ProviderSendTurnInput["modelSelection"],
41+ ) => Effect.Effect<void, TError>;
42+ }
43+ | { readonly type: "slash-command"; readonly command: `/${string}` };
44+
3045 export interface ProviderAdapterCapabilities {
3146 /**
3247 * Declares whether changing the model on an existing session is supported.
@@ -70,10 +85,8 @@ export interface ProviderAdapterShape<TError> { · 37 unmodified lines
7085 input: ProviderSendTurnInput,
7186 ) => Effect.Effect<ProviderTurnStartResult, TError>;
7287
73− readonly compactThread?: (
74− threadId: ThreadId,
75− modelSelection?: ProviderSendTurnInput["modelSelection"],
76− ) => Effect.Effect<void, TError>;
88+ /** Omitted when this adapter does not support manual context compaction. */
89+ readonly compaction?: ProviderCompaction<TError>;
7790
7891 /**
7992 * Interrupt an active turn.