From 84447ade8cf8cc009fb388a4e5bc6b0d848c325c Mon Sep 17 00:00:00 2001 From: Steve James Date: Wed, 12 Aug 2026 10:40:15 +0200 Subject: [PATCH] unify context overflow recovery --- docs/architecture/context-compaction.md | 28 +- docs/reference/configuration.md | 2 +- docs/reference/syscalls.md | 2 +- docs/reference/websocket-protocol.md | 3 + gateway/src/process/do.test.ts | 574 +++++++++++++++++++++++- gateway/src/process/do.ts | 201 +++++---- 6 files changed, 710 insertions(+), 100 deletions(-) diff --git a/docs/architecture/context-compaction.md b/docs/architecture/context-compaction.md index 66e1cd00b..21062a023 100644 --- a/docs/architecture/context-compaction.md +++ b/docs/architecture/context-compaction.md @@ -30,20 +30,29 @@ Provider usage updates the state after a response; preflight always recomputes the estimate for the next request. An unknown model context window produces unknown pressure, not an invented -limit. The provider may still reject the request; normal generation fallback and -error handling then apply. +limit. If the provider reports that the assembled request exceeds its context +window, the Process applies the same history policy regardless of the estimate. +Context overflow does not advance the main generation fallback chain. ## Overflow policy Each process has an `auto-compact` or `fail` policy, a pressure threshold, and a -`keepLast` value. The default auto-compacts at `0.9` pressure +`keepLast` value. The threshold governs proactive preflight compaction. The +default auto-compacts at `0.9` pressure while retaining the newest 80 stored messages. The policy is exposed through `proc.history.policy.get` and `proc.history.policy.set`. -- `auto-compact` generates a summary and compacts the old prefix before the - model call. -- `fail` ends the run with a visible system error and leaves the process - available for explicit compaction or reset. +- `auto-compact` generates a summary and compacts the old prefix during + preflight or after the first provider-confirmed overflow. It rebuilds the + context and retries the same active model configuration once. +- `fail` ends the run with a visible system error during preflight or after a + provider-confirmed overflow, and leaves the process available for explicit + compaction or reset. + +One generation cycle installs at most one automatic compaction. A later tool +round may compact again if newly stored results grow the next assembled context. +If the rebuilt request still overflows, or no older prefix can be archived, the +current run stops explicitly rather than looping or switching models. Explicit compaction remains available as an operation; `manual` is not an overflow policy. @@ -56,8 +65,9 @@ overflow policy. - `throughMessageId` selects a prefix through a stored message id. The caller must also provide a summary or set `generateSummary: true`. Explicit -compaction rejects an active process. Automatic compaction runs in the -owning run's preflight and stops if that run is superseded or aborted. +compaction rejects an active process. Automatic compaction runs inside the +owning run's lifecycle, from preflight or provider-overflow recovery, and stops +if that run is superseded or aborted. A successful compaction: diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index ae2b5ac21..97ff2537e 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -48,7 +48,7 @@ The AI runtime resolves per-user values first, then falls back to system default |---|---|---|---| | `config/ai/provider` | `users/{uid}/ai/provider` | `workers-ai` | Provider adapter. | | `config/ai/model` | `users/{uid}/ai/model` | `@cf/zai-org/glm-5.2` | Provider model identifier. | -| `config/ai/fallback_model_profile` | `users/{uid}/ai/fallback_model_profile` | `workers-ai-kimi-k2-6` | Saved model profile to try if the selected model fails. | +| `config/ai/fallback_model_profile` | `users/{uid}/ai/fallback_model_profile` | `workers-ai-kimi-k2-6` | Saved model profile to try after eligible generation failures. Context overflow is handled by the process history policy instead. | | `config/ai/api_key` | `users/{uid}/ai/api_key` | empty | Provider credential. Sensitive. | | `config/ai/reasoning` | `users/{uid}/ai/reasoning` | `medium` | Reasoning mode hint: `off`, `minimal`, `low`, `medium`, `high`, or `xhigh`. Unsupported values are clamped to the nearest model-supported level at generation time. | | `config/ai/max_tokens` | `users/{uid}/ai/max_tokens` | `8192` | Maximum output tokens. | diff --git a/docs/reference/syscalls.md b/docs/reference/syscalls.md index d178f6c00..ea39d6af8 100644 --- a/docs/reference/syscalls.md +++ b/docs/reference/syscalls.md @@ -353,7 +353,7 @@ Runtime behavior: | `proc.media.write` | Process DO | Streams one request body directly into process-scoped R2 storage. The body descriptor must declare its exact length so R2 receives a fixed-length stream. An internal caller may supply `mediaId` as an idempotency key: an exact repeated descriptor drains the repeated body and returns the original reference, while conflicting metadata is rejected. Returns a stable media reference for `proc.send`, including its read-only `/var/media/{uid}/{pid}/{id}` filesystem path. | | `proc.media.delete` | Process DO | Idempotently deletes one unreferenced process-scoped media object. Keys outside the target process or already referenced by process history are rejected. Used to roll back uploads that are not admitted by `proc.send`. | | `proc.history.policy.get` | Process DO | Returns the process context-overflow policy. The default is `auto-compact` at 90% pressure while retaining the newest 80 stored messages. | -| `proc.history.policy.set` | Process DO | Sets the process context-overflow policy. Supported `overflow` values are `auto-compact` and `fail`; the policy is applied during run preflight. | +| `proc.history.policy.set` | Process DO | Sets the process context-overflow policy. Supported `overflow` values are `auto-compact` and `fail`; the policy is applied during run preflight and after a provider-confirmed overflow. Provider overflow does not advance the main generation fallback chain. | | `proc.history.compact` | Process DO | Archives an old history prefix, inserts a visible system summary marker, and records a `compaction` segment. Requires a supplied or generated summary and exactly one of `keepLast` or `throughMessageId`. | | `proc.history.segment.read` | Process DO | Reads paged messages from a compacted segment without restoring them into active history. | | `proc.history.segments` | Process DO | Lists compacted segments, including archive paths and summary marker ids. | diff --git a/docs/reference/websocket-protocol.md b/docs/reference/websocket-protocol.md index 7dae96baa..2d9324ef5 100644 --- a/docs/reference/websocket-protocol.md +++ b/docs/reference/websocket-protocol.md @@ -252,6 +252,9 @@ Current role defaults from `buildSignalList()`: - `proc.run.started` - `proc.run.stream` - `proc.run.retrying` + - Carries `attempt`, `nextAttempt`, `maxAttempts`, and a sanitized `reason`. + A retry after context compaction stays on the active model and has no + `fallback` field; model fallback transitions include their source and target. - `proc.run.output` - Carries assembled assistant text/thinking and, when present, process-owned `media` references registered for the automatic final reply. diff --git a/gateway/src/process/do.test.ts b/gateway/src/process/do.test.ts index bad6c2d0e..b48b535c3 100644 --- a/gateway/src/process/do.test.ts +++ b/gateway/src/process/do.test.ts @@ -112,6 +112,36 @@ function testUsage(input = 0, output = 0) { }; } +const KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR = + '8007: {"object":"error","message":"The input (301552 tokens) is longer than the model\'s context length (262144 tokens).","type":"BadRequestError","param":null,"code":400}'; + +function kimiWorkersConfigWithFallback(pid: string, contextWindowTokens = 1_000_000) { + return { + executor: { kind: "process" as const, pid }, + profile: "task" as const, + provider: "workers-ai", + model: "@cf/moonshotai/kimi-k2.6", + apiKey: "", + reasoning: "off", + maxTokens: 100, + contextWindowTokens, + contextWindowSource: "config" as const, + maxContextBytes: 32768, + fallbacks: [{ + profileId: "overflow-backup", + profileName: "Overflow Backup", + provider: "openrouter", + model: "fallback-model", + apiKey: "fallback-key", + maxTokens: 100, + contextWindowTokens, + contextWindowSource: "config" as const, + generationTimeoutMs: 180000, + generationStreaming: "auto" as const, + }], + }; +} + async function stubGeneration( stub: DurableObjectStub, generate: (request: any) => string | Promise, @@ -2644,6 +2674,515 @@ describe("Process DO — mechanical", () => { ]); }); + it("auto-compacts and retries the same Kimi model after a thrown provider overflow", async () => { + const pid = "mech-chat-kimi-overflow-throw-compact"; + const runId = "run-chat-kimi-overflow-throw-compact"; + const stub = await initProcess(pid, ROOT_IDENTITY); + + const result = await runInDurableObject(stub, async (instance: Process) => { + const process = instance as any; + const emitted: Array<{ signal: string; payload: unknown }> = []; + const calls: Array<{ provider: string; model: string; context: string }> = []; + const timeline: string[] = []; + let summaryCalls = 0; + process.sendSignal = async (signal: string, payload: unknown) => { + emitted.push({ signal, payload }); + if (signal === "proc.run.retrying") { + timeline.push("retrying"); + } + if (signal === "proc.changed" && (payload as any).event) { + timeline.push((payload as any).event); + } + }; + process.generation = { + async generate(request: any) { + calls.push({ + provider: request.config.provider, + model: request.config.model, + context: JSON.stringify(request.context), + }); + timeline.push(`generate:${calls.length}`); + if (calls.length === 1) { + throw new Error(KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR); + } + return { + role: "assistant", + content: [{ type: "text", text: "same model after compaction" }], + api: "test", + provider: request.config.provider, + model: request.config.model, + usage: testUsage(20, 3), + stopReason: "stop", + timestamp: Date.now(), + }; + }, + async generateText(request: any) { + summaryCalls += 1; + expect(request.config).toMatchObject({ + provider: "workers-ai", + model: "@cf/moonshotai/kimi-k2.6", + }); + expect(JSON.stringify(request.context)).toContain("old Kimi context A"); + return "Kimi overflow compact summary."; + }, + }; + + process.store.appendMessage("user", "old Kimi context A"); + process.store.appendMessage("assistant", "old Kimi context B"); + process.store.appendMessage("user", "Kimi context that must stay live."); + process.store.setValue("historyPolicy", JSON.stringify({ + overflow: "auto-compact", + compactAtPressure: 0.9, + keepLast: 1, + updatedAt: Date.now(), + })); + process.currentRun = { + runId, + config: kimiWorkersConfigWithFallback(pid), + tools: [], + devices: [], + systemPrompt: "Test system prompt.", + approvalPolicy: { default: "auto", rules: [] }, + }; + + await process.runTick(runId); + return { + calls, + emitted, + messages: process.store.getMessages(), + segments: process.store.listHistorySegments(), + summaryCalls, + timeline, + }; + }); + + expect(result.calls).toHaveLength(2); + expect(result.calls.map(({ provider, model }) => ({ provider, model }))).toEqual([ + { provider: "workers-ai", model: "@cf/moonshotai/kimi-k2.6" }, + { provider: "workers-ai", model: "@cf/moonshotai/kimi-k2.6" }, + ]); + expect(result.calls[0].context).toContain("old Kimi context A"); + expect(result.calls[1].context).toContain("Kimi overflow compact summary."); + expect(result.calls[1].context).toContain("Kimi context that must stay live."); + expect(result.calls[1].context).not.toContain("old Kimi context A"); + expect(result.summaryCalls).toBe(1); + expect(result.segments).toHaveLength(1); + expect(result.messages.map((message: any) => [message.role, message.content])).toEqual([ + ["system", expect.stringContaining("Kimi overflow compact summary.")], + ["user", "Kimi context that must stay live."], + ["assistant", "same model after compaction"], + ]); + const retrying = result.emitted.filter((entry) => entry.signal === "proc.run.retrying"); + expect(retrying).toHaveLength(1); + expect(retrying[0]?.payload).toMatchObject({ + pid, + runId, + attempt: 1, + nextAttempt: 2, + maxAttempts: 2, + reason: KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR, + }); + expect(retrying[0]?.payload).not.toHaveProperty("fallback"); + expect(result.timeline).toEqual([ + "generate:1", + "history.compacted", + "history.auto_compacted", + "retrying", + "generate:2", + ]); + }); + + it("auto-compacts a returned provider overflow, retries Kimi, and records usage once", async () => { + const pid = "mech-chat-kimi-overflow-response-compact"; + const runId = "run-chat-kimi-overflow-response-compact"; + const stub = await initProcess(pid, ROOT_IDENTITY); + + const result = await runInDurableObject(stub, async (instance: Process) => { + const process = instance as any; + const emitted: Array<{ signal: string; payload: unknown }> = []; + const calls: Array<{ provider: string; model: string; context: string }> = []; + let summaryCalls = 0; + process.sendSignal = async (signal: string, payload: unknown) => { + emitted.push({ signal, payload }); + }; + process.generation = { + async generate(request: any) { + calls.push({ + provider: request.config.provider, + model: request.config.model, + context: JSON.stringify(request.context), + }); + if (calls.length === 1) { + return { + role: "assistant", + content: [], + api: "test", + provider: request.config.provider, + model: request.config.model, + usage: { + ...testUsage(301_552, 0), + cost: { + input: 0.12, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0.12, + }, + }, + stopReason: "error", + errorMessage: KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR, + timestamp: Date.now(), + }; + } + return { + role: "assistant", + content: [{ type: "text", text: "returned overflow recovered" }], + api: "test", + provider: request.config.provider, + model: request.config.model, + usage: testUsage(20, 3), + stopReason: "stop", + timestamp: Date.now(), + }; + }, + async generateText() { + summaryCalls += 1; + return "Returned overflow compact summary."; + }, + }; + + process.store.appendMessage("user", "old returned overflow context A"); + process.store.appendMessage("assistant", "old returned overflow context B"); + process.store.appendMessage("user", "Returned overflow context that must stay live."); + process.store.setValue("historyPolicy", JSON.stringify({ + overflow: "auto-compact", + compactAtPressure: 0.9, + keepLast: 1, + updatedAt: Date.now(), + })); + process.currentRun = { + runId, + config: kimiWorkersConfigWithFallback(pid), + tools: [], + devices: [], + systemPrompt: "Test system prompt.", + approvalPolicy: { default: "auto", rules: [] }, + }; + + await process.runTick(runId); + return { + calls, + emitted, + historyUsage: process.store.getHistoryUsage(), + messages: process.store.getMessages(), + segments: process.store.listHistorySegments(), + summaryCalls, + }; + }); + + expect(result.calls).toHaveLength(2); + expect(result.calls.map(({ provider, model }) => ({ provider, model }))).toEqual([ + { provider: "workers-ai", model: "@cf/moonshotai/kimi-k2.6" }, + { provider: "workers-ai", model: "@cf/moonshotai/kimi-k2.6" }, + ]); + expect(result.calls[1].context).toContain("Returned overflow compact summary."); + expect(result.calls[1].context).toContain("Returned overflow context that must stay live."); + expect(result.calls[1].context).not.toContain("old returned overflow context A"); + expect(result.summaryCalls).toBe(1); + expect(result.segments).toHaveLength(1); + expect(result.historyUsage).toMatchObject({ + inputTokens: 301_572, + outputTokens: 3, + totalTokens: 301_575, + cost: { total: 0.12, source: "model-pricing" }, + generations: 2, + }); + expect(result.messages.map((message: any) => [message.role, message.content])).toEqual([ + ["system", expect.stringContaining("Returned overflow compact summary.")], + ["user", "Returned overflow context that must stay live."], + ["assistant", "returned overflow recovered"], + ]); + const retrying = result.emitted.filter((entry) => entry.signal === "proc.run.retrying"); + expect(retrying).toHaveLength(1); + expect(retrying[0]?.payload).toMatchObject({ + pid, + runId, + attempt: 1, + nextAttempt: 2, + maxAttempts: 2, + reason: KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR, + }); + expect(retrying[0]?.payload).not.toHaveProperty("fallback"); + }); + + it("applies fail policy to provider overflow without compacting or using fallback", async () => { + const pid = "mech-chat-kimi-overflow-policy-fail"; + const runId = "run-chat-kimi-overflow-policy-fail"; + const stub = await initProcess(pid, ROOT_IDENTITY); + + const result = await runInDurableObject(stub, async (instance: Process) => { + const process = instance as any; + const emitted: Array<{ signal: string; payload: unknown }> = []; + const calls: Array<{ provider: string; model: string }> = []; + let summaryCalls = 0; + process.sendSignal = async (signal: string, payload: unknown) => { + emitted.push({ signal, payload }); + }; + process.generation = { + async generate(request: any) { + calls.push({ + provider: request.config.provider, + model: request.config.model, + }); + if (calls.length === 1) { + throw new Error(KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR); + } + return { + role: "assistant", + content: [{ type: "text", text: "fallback must not run" }], + api: "test", + provider: request.config.provider, + model: request.config.model, + stopReason: "stop", + timestamp: Date.now(), + }; + }, + async generateText() { + summaryCalls += 1; + return "summary must not run"; + }, + }; + + process.store.appendMessage("user", "old fail-policy context A"); + process.store.appendMessage("assistant", "old fail-policy context B"); + process.store.appendMessage("user", "Fail-policy context that must stay live."); + process.store.setValue("historyPolicy", JSON.stringify({ + overflow: "fail", + compactAtPressure: 0.9, + keepLast: 1, + updatedAt: Date.now(), + })); + process.currentRun = { + runId, + config: kimiWorkersConfigWithFallback(pid), + tools: [], + devices: [], + systemPrompt: "Test system prompt.", + approvalPolicy: { default: "auto", rules: [] }, + }; + + await process.runTick(runId); + return { + calls, + currentRun: process.currentRun, + emitted, + messages: process.store.getMessages(), + segments: process.store.listHistorySegments(), + summaryCalls, + }; + }); + + expect(result.calls).toEqual([ + { provider: "workers-ai", model: "@cf/moonshotai/kimi-k2.6" }, + ]); + expect(result.summaryCalls).toBe(0); + expect(result.segments).toHaveLength(0); + expect(result.currentRun).toBeNull(); + expect(result.messages.slice(0, 3).map((message: any) => message.content)).toEqual([ + "old fail-policy context A", + "old fail-policy context B", + "Fail-policy context that must stay live.", + ]); + expect(result.messages.at(-1)?.content).toContain("Context limit policy stopped this run."); + expect(result.emitted.some((entry) => entry.signal === "proc.run.retrying")).toBe(false); + expect(result.emitted).toEqual(expect.arrayContaining([ + { + signal: "proc.run.finished", + payload: expect.objectContaining({ + runId, + status: "error", + reason: "context.policy.fail", + }), + }, + ])); + }); + + it("terminates repeated provider overflow after one compaction without using fallback", async () => { + const pid = "mech-chat-kimi-overflow-repeated"; + const runId = "run-chat-kimi-overflow-repeated"; + const stub = await initProcess(pid, ROOT_IDENTITY); + + const result = await runInDurableObject(stub, async (instance: Process) => { + const process = instance as any; + const emitted: Array<{ signal: string; payload: unknown }> = []; + const calls: Array<{ provider: string; model: string }> = []; + let summaryCalls = 0; + process.sendSignal = async (signal: string, payload: unknown) => { + emitted.push({ signal, payload }); + }; + process.generation = { + async generate(request: any) { + calls.push({ + provider: request.config.provider, + model: request.config.model, + }); + throw new Error(KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR); + }, + async generateText() { + summaryCalls += 1; + return "Repeated overflow compact summary."; + }, + }; + + process.store.appendMessage("user", "old repeated-overflow context A"); + process.store.appendMessage("assistant", "old repeated-overflow context B"); + process.store.appendMessage("user", "Repeated-overflow context that must stay live."); + process.store.setValue("historyPolicy", JSON.stringify({ + overflow: "auto-compact", + compactAtPressure: 0.9, + keepLast: 1, + updatedAt: Date.now(), + })); + process.currentRun = { + runId, + config: kimiWorkersConfigWithFallback(pid), + tools: [], + devices: [], + systemPrompt: "Test system prompt.", + approvalPolicy: { default: "auto", rules: [] }, + }; + + await process.runTick(runId); + return { + calls, + currentRun: process.currentRun, + emitted, + messages: process.store.getMessages(), + segments: process.store.listHistorySegments(), + summaryCalls, + }; + }); + + expect(result.calls).toEqual([ + { provider: "workers-ai", model: "@cf/moonshotai/kimi-k2.6" }, + { provider: "workers-ai", model: "@cf/moonshotai/kimi-k2.6" }, + ]); + expect(result.summaryCalls).toBe(1); + expect(result.segments).toHaveLength(1); + expect(result.currentRun).toBeNull(); + expect(result.messages.at(-1)?.content).toContain( + "Context limit reached for workers-ai/@cf/moonshotai/kimi-k2.6.", + ); + const retrying = result.emitted.filter((entry) => entry.signal === "proc.run.retrying"); + expect(retrying).toHaveLength(1); + expect(retrying[0]?.payload).toMatchObject({ + pid, + runId, + attempt: 1, + nextAttempt: 2, + maxAttempts: 2, + reason: KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR, + }); + expect(retrying[0]?.payload).not.toHaveProperty("fallback"); + expect(result.emitted).toEqual(expect.arrayContaining([ + { + signal: "proc.run.finished", + payload: expect.objectContaining({ + runId, + status: "error", + reason: "context.provider_overflow", + }), + }, + ])); + }); + + it("terminates provider overflow when no history prefix can be compacted", async () => { + const pid = "mech-chat-kimi-overflow-empty-prefix"; + const runId = "run-chat-kimi-overflow-empty-prefix"; + const stub = await initProcess(pid, ROOT_IDENTITY); + + const result = await runInDurableObject(stub, async (instance: Process) => { + const process = instance as any; + const emitted: Array<{ signal: string; payload: unknown }> = []; + const calls: Array<{ provider: string; model: string }> = []; + let summaryCalls = 0; + process.sendSignal = async (signal: string, payload: unknown) => { + emitted.push({ signal, payload }); + }; + process.generation = { + async generate(request: any) { + calls.push({ + provider: request.config.provider, + model: request.config.model, + }); + if (calls.length === 1) { + throw new Error(KIMI_WORKERS_CONTEXT_OVERFLOW_ERROR); + } + return { + role: "assistant", + content: [{ type: "text", text: "fallback must not run" }], + api: "test", + provider: request.config.provider, + model: request.config.model, + stopReason: "stop", + timestamp: Date.now(), + }; + }, + async generateText() { + summaryCalls += 1; + return "summary must not run"; + }, + }; + + process.store.appendMessage("user", "Only live message."); + process.store.setValue("historyPolicy", JSON.stringify({ + overflow: "auto-compact", + compactAtPressure: 0.9, + keepLast: 1, + updatedAt: Date.now(), + })); + process.currentRun = { + runId, + config: kimiWorkersConfigWithFallback(pid), + tools: [], + devices: [], + systemPrompt: "Test system prompt.", + approvalPolicy: { default: "auto", rules: [] }, + }; + + await process.runTick(runId); + return { + calls, + currentRun: process.currentRun, + emitted, + messages: process.store.getMessages(), + segments: process.store.listHistorySegments(), + summaryCalls, + }; + }); + + expect(result.calls).toEqual([ + { provider: "workers-ai", model: "@cf/moonshotai/kimi-k2.6" }, + ]); + expect(result.summaryCalls).toBe(0); + expect(result.segments).toHaveLength(0); + expect(result.currentRun).toBeNull(); + expect(result.messages.at(-1)?.content).toContain( + "Context limit reached, but auto-compaction could not archive any older messages.", + ); + expect(result.emitted.some((entry) => entry.signal === "proc.run.retrying")).toBe(false); + expect(result.emitted).toEqual(expect.arrayContaining([ + { + signal: "proc.run.finished", + payload: expect.objectContaining({ + runId, + status: "error", + reason: "context.auto_compact.empty", + }), + }, + ])); + }); + it("surfaces thrown provider context overflow separately from generation errors", async () => { const pid = "mech-chat-provider-context-overflow-throw"; const stub = await initProcess(pid, ROOT_IDENTITY); @@ -2693,15 +3232,16 @@ describe("Process DO — mechanical", () => { expect(result.currentRun).toBeNull(); const systemMessage = result.messages.find((message: any) => message.role === "system"); - expect(systemMessage?.content).toContain("Context limit reached for openai/gpt-test."); - expect(systemMessage?.content).toContain("Provider message: Your input exceeds the context window of this model"); + expect(systemMessage?.content).toContain( + "Context limit reached, but auto-compaction could not archive any older messages.", + ); expect(systemMessage?.content).not.toContain("Generation failed:"); expect(result.emitted).toEqual(expect.arrayContaining([ { signal: "proc.run.finished", payload: expect.objectContaining({ status: "error", - reason: "context.provider_overflow", + reason: "context.auto_compact.empty", runId: "run-chat-provider-context-overflow-throw", }), }, @@ -2714,7 +3254,10 @@ describe("Process DO — mechanical", () => { const result = await runInDurableObject(stub, async (instance: Process) => { const process = instance as any; - process.sendSignal = async () => {}; + const emitted: Array<{ signal: string; payload: unknown }> = []; + process.sendSignal = async (signal: string, payload: unknown) => { + emitted.push({ signal, payload }); + }; process.generation = { async generate() { throw new Error("request failed", { @@ -2753,15 +3296,27 @@ describe("Process DO — mechanical", () => { await process.runTick("run-chat-provider-context-overflow-nested"); return { currentRun: process.currentRun, + emitted, messages: process.store.getMessages(), }; }); expect(result.currentRun).toBeNull(); const systemMessage = result.messages.find((message: any) => message.role === "system"); - expect(systemMessage?.content).toContain("Context limit reached for openai/gpt-test."); - expect(systemMessage?.content).toContain("Provider message: Your input exceeds the context window of this model"); + expect(systemMessage?.content).toContain( + "Context limit reached, but auto-compaction could not archive any older messages.", + ); expect(systemMessage?.content).not.toContain("Generation failed:"); + expect(result.emitted).toEqual(expect.arrayContaining([ + { + signal: "proc.run.finished", + payload: expect.objectContaining({ + status: "error", + reason: "context.auto_compact.empty", + runId: "run-chat-provider-context-overflow-nested", + }), + }, + ])); }); it("surfaces returned provider context overflow and records provider usage", async () => { @@ -2832,8 +3387,9 @@ describe("Process DO — mechanical", () => { }); const systemMessage = result.messages.find((message: any) => message.role === "system"); - expect(systemMessage?.content).toContain("Context limit reached for google/gemini-test."); - expect(systemMessage?.content).toContain("Provider message: The input token count"); + expect(systemMessage?.content).toContain( + "Context limit reached, but auto-compaction could not archive any older messages.", + ); expect(systemMessage?.content).not.toContain("Generation failed:"); expect(result.contextState).toMatchObject({ inputTokens: 1196265, @@ -2855,7 +3411,7 @@ describe("Process DO — mechanical", () => { signal: "proc.run.finished", payload: expect.objectContaining({ status: "error", - reason: "context.provider_overflow", + reason: "context.auto_compact.empty", runId: "run-chat-provider-context-overflow-response", }), }, diff --git a/gateway/src/process/do.ts b/gateway/src/process/do.ts index d75cf4a8a..1081242ee 100644 --- a/gateway/src/process/do.ts +++ b/gateway/src/process/do.ts @@ -3897,22 +3897,21 @@ export class Process extends Host { tools: tools.length > 0 ? tools : undefined, }; let autoCompactionPressure: number | null = null; - const prepareGenerationContext = async ( + let contextState!: ProcContextState; + const applyGenerationContextPolicy = async ( config: AiConfigResult, - ): Promise<"ready" | "stopped"> => { - context = await buildGenerationContext(); - const contextState = await this.updateContextState(runId, config, context); - if (this.handleRunStopped(runId)) { - return "stopped"; - } - + trigger: "preflight" | "provider-overflow", + ): Promise<"ready" | "compacted" | "stopped"> => { const policy = this.getHistoryContextPolicy(); if (autoCompactionPressure !== null) { + if (trigger === "provider-overflow") { + return "ready"; + } if (contextState.pressure !== null && contextState.pressure >= 1) { await this.finishInsufficientCompactionRun( runId, policy, - autoCompactionPressure ?? policy.compactAtPressure, + autoCompactionPressure, contextState.pressure, ); return "stopped"; @@ -3920,38 +3919,46 @@ export class Process extends Host { return "ready"; } - const contextPreflight = await this.applyHistoryContextPolicy( + const policyResult = await this.applyHistoryContextPolicy( runId, config, contextState, + trigger, ); - if (contextPreflight === "stopped") { + if (policyResult !== "compacted") { + return policyResult; + } + + autoCompactionPressure = contextState.pressure ?? policy.compactAtPressure; + if (this.handleRunStopped(runId)) { return "stopped"; } - if (contextPreflight === "compacted") { - autoCompactionPressure = contextState.pressure ?? policy.compactAtPressure; - if (this.handleRunStopped(runId)) { - return "stopped"; - } - context = await buildGenerationContext(); - const compactedState = await this.updateContextState(runId, config, context); - if (this.handleRunStopped(runId)) { - return "stopped"; - } - if ( - compactedState.pressure !== null && - compactedState.pressure >= 1 - ) { - await this.finishInsufficientCompactionRun( - runId, - policy, - autoCompactionPressure, - compactedState.pressure, - ); - return "stopped"; - } + context = await buildGenerationContext(); + contextState = await this.updateContextState(runId, config, context); + if (this.handleRunStopped(runId)) { + return "stopped"; + } + if (contextState.pressure !== null && contextState.pressure >= 1) { + await this.finishInsufficientCompactionRun( + runId, + policy, + autoCompactionPressure, + contextState.pressure, + ); + return "stopped"; } - return "ready"; + return "compacted"; + }; + const prepareGenerationContext = async ( + config: AiConfigResult, + ): Promise<"ready" | "stopped"> => { + context = await buildGenerationContext(); + contextState = await this.updateContextState(runId, config, context); + if (this.handleRunStopped(runId)) { + return "stopped"; + } + const result = await applyGenerationContextPolicy(config, "preflight"); + return result === "stopped" ? "stopped" : "ready"; }; const contextPreflight = await prepareGenerationContext(run.config!); @@ -4003,6 +4010,60 @@ export class Process extends Host { } return this.handleRunStopped(runId) ? "stopped" : "switched"; }; + const recoverProviderContextOverflow = async ( + errorMsg: string, + failedResponse?: AssistantMessage, + ): Promise<"retry" | "stopped"> => { + if (failedResponse) { + const overflowUsage = this.recordUnpersistedAssistantUsage( + failedResponse, + run.config!, + ); + contextState = await this.updateContextState( + runId, + run.config!, + context, + failedResponse.usage, + overflowUsage, + ); + if (this.handleRunStopped(runId)) { + return "stopped"; + } + } + + if (autoCompactionPressure !== null) { + await this.finishProviderContextOverflowRun( + runId, + run.config!, + errorMsg, + ); + return "stopped"; + } + + const policyResult = await applyGenerationContextPolicy( + run.config!, + "provider-overflow", + ); + if (policyResult !== "compacted") { + if (policyResult === "ready" && !this.handleRunStopped(runId)) { + await this.finishProviderContextOverflowRun( + runId, + run.config!, + errorMsg, + ); + } + return "stopped"; + } + + const retryState = await this.beginGenerationRetry({ + runId, + attempt: 1, + maxAttempts: 2, + reason: errorMsg, + cause: "provider context overflow", + }); + return retryState === "stopped" ? "stopped" : "retry"; + }; let attempt = 1; while (attempt <= MAX_RETRYABLE_GENERATION_ATTEMPTS) { try { @@ -4027,21 +4088,11 @@ export class Process extends Host { model: run.config!.model, contextWindowTokens: run.config!.contextWindowTokens, })) { - const fallbackState = await switchToFallback(errorMsg); - if (fallbackState === "stopped") { - return; - } - if (fallbackState === "switched") { - attempt = 1; + const recovery = await recoverProviderContextOverflow(errorMsg); + if (recovery === "retry") { response = null; continue; } - console.error(`[Process] LLM context overflow:`, e); - await this.finishProviderContextOverflowRun( - runId, - run.config!, - errorMsg, - ); return; } if ( @@ -4099,16 +4150,12 @@ export class Process extends Host { if (isProviderContextOverflow(response, run.config!.contextWindowTokens)) { const errorMsg = response.errorMessage ?? describeAssistantResponseFailure(response) ?? "Provider context overflow"; - const fallbackState = await switchToFallback(errorMsg, response); - if (fallbackState === "stopped") { - return; - } - if (fallbackState === "switched") { - attempt = 1; - response = null; + const recovery = await recoverProviderContextOverflow(errorMsg, response); + response = null; + if (recovery === "retry") { continue; } - break; + return; } const responseFailure = describeAssistantResponseFailure(response); @@ -4156,21 +4203,6 @@ export class Process extends Host { return; } - if (isProviderContextOverflow(response, run.config!.contextWindowTokens)) { - const overflowUsage = this.recordUnpersistedAssistantUsage(response, run.config!); - await this.updateContextState(runId, run.config!, context, response.usage, overflowUsage); - if (this.handleRunStopped(runId)) { - return; - } - const errorMsg = response.errorMessage ?? describeAssistantResponseFailure(response) ?? undefined; - await this.finishProviderContextOverflowRun( - runId, - run.config!, - errorMsg, - ); - return; - } - const responseFailure = describeAssistantResponseFailure(response); if (responseFailure) { this.recordUnpersistedAssistantUsage(response, run.config!); @@ -4583,22 +4615,28 @@ export class Process extends Host { runId: string, config: AiConfigResult, state: ProcContextState, + trigger: "preflight" | "provider-overflow" = "preflight", ): Promise<"ready" | "compacted" | "stopped"> { const pressure = state.pressure; - if (pressure === null || !Number.isFinite(pressure)) { - return "ready"; - } - const policy = this.getHistoryContextPolicy(); - if (pressure < policy.compactAtPressure) { - return "ready"; + if (trigger === "preflight") { + if (pressure === null || !Number.isFinite(pressure)) { + return "ready"; + } + if (pressure < policy.compactAtPressure) { + return "ready"; + } } if (policy.overflow === "fail") { const message = [ "Context limit policy stopped this run.", - `Policy: fail at ${Math.round(policy.compactAtPressure * 100)}% context pressure.`, - `Current estimate: ${Math.round(pressure * 100)}%.`, + trigger === "provider-overflow" + ? "The AI provider reported that the request exceeds its context window." + : `Policy: fail at ${Math.round(policy.compactAtPressure * 100)}% context pressure.`, + ...(pressure !== null && Number.isFinite(pressure) + ? [`Current estimate: ${Math.round(pressure * 100)}%.`] + : []), "Compact the history or reset the process before sending more work.", ].join("\n"); this.store.appendMessage("system", message, { runId }); @@ -4620,7 +4658,7 @@ export class Process extends Host { keepLast: policy.keepLast, }); if (selected.length === 0) { - if (pressure < 1) { + if (trigger === "preflight" && pressure !== null && pressure < 1) { return "ready"; } const message = [ @@ -4658,7 +4696,9 @@ export class Process extends Host { return "stopped"; } if (!result.ok) { - const message = `Auto-compaction failed before model call: ${result.error}`; + const message = trigger === "provider-overflow" + ? `Auto-compaction failed after provider context overflow: ${result.error}` + : `Auto-compaction failed before model call: ${result.error}`; this.store.appendMessage("system", message, { runId }); await this.emitProcChanged(["messages"], { runId, @@ -4682,7 +4722,8 @@ export class Process extends Host { pid: this.pid, provider: config.provider, model: config.model, - pressure, + ...(pressure !== null && Number.isFinite(pressure) ? { pressure } : {}), + trigger, policy, segment: result.segment, archivedMessages: result.archivedMessages,