diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index a3a1addfe..38796fe33 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -11,6 +11,11 @@ ### Changed - Enriched `/advisor status` to show per-advisor status glyphs, model, spend breakdown, and quota window for every configured advisor (including disabled ones), replacing the previous single-advisor-only summary. +### Fixed + +- Fixed advisors retrying a permanently rejected request forever (e.g. `invalid_request_error: model not supported with this account`): unlike quota exhaustion — which already paused with a notice — this class notified once and silently kept re-attempting every turn, re-building heavy context in a shared daemon. The runtime now hard-stops after a permanent rejection or three consecutive backlog-drop cycles, with a visible notice; an explicit reset (`/new`, config rebuild, restart) re-enables it. `waitForCatchup` resolves immediately while halted so the primary agent is never parked on a runtime that cannot drain. +- Fixed the advisor's delta render freezing the whole process on large transcripts (one agent + one advisor was enough): rendering the transcript slice for the advisor ran synchronously on the event loop, and a post-reset replay of a multi-MB session blocked it for 600ms+ per render. Large deltas now render in size- and count-bounded chunks that yield the event loop, with tool call/result pairing preserved across chunk boundaries via a shared whole-delta result index; small per-turn deltas keep the synchronous fast path. + ## [17.0.0] - 2026-07-15 ### Breaking Changes diff --git a/packages/coding-agent/src/advisor/__tests__/advisor.test.ts b/packages/coding-agent/src/advisor/__tests__/advisor.test.ts index 9a9cafc93..01d06a9da 100644 --- a/packages/coding-agent/src/advisor/__tests__/advisor.test.ts +++ b/packages/coding-agent/src/advisor/__tests__/advisor.test.ts @@ -1797,6 +1797,320 @@ describe("advisor", () => { expect(failures).toHaveLength(2); }); + it("halts permanently on an invalid_request rejection instead of retrying forever", async () => { + // The runaway observed live: a provider that refuses the configured + // model outright ("not supported ... (code=invalid_request_error)") + // failed 351 turns/hour in a shared daemon, rebuilding heavy context + // every cycle. One drop cycle must latch the runtime off. + const promptInputs: string[] = []; + const failures: unknown[] = []; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + throw new Error( + "Codex error event: The 'gpt-5.3-codex-spark' model is not supported when using Codex with a ChatGPT account. (code=invalid_request_error)", + ); + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + const messages: AgentMessage[] = [{ role: "user", content: "aaa", timestamp: 1 } as AgentMessage]; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + notifyFailure: error => failures.push(error), + }; + const runtime = new AdvisorRuntime(agent, host, 0); + + runtime.onTurnEnd(messages); + await Bun.sleep(0); + await Bun.sleep(0); + await Bun.sleep(0); + + expect(promptInputs).toHaveLength(3); + expect(failures).toHaveLength(1); + expect(runtime.halted).toBe(true); + + // New deltas must be ignored while halted — no further prompts. + messages.push({ role: "user", content: "bbb", timestamp: 2 } as AgentMessage); + runtime.onTurnEnd(messages); + await Bun.sleep(0); + await Bun.sleep(0); + expect(promptInputs).toHaveLength(3); + + // The catch-up gate must not park the primary agent on a runtime that + // will never drain again: resolve immediately regardless of maxMs. + await runtime.waitForCatchup(60_000, 0); + + // Explicit reset (config rebuild, /new) re-enables the runtime. + runtime.reset(); + expect(runtime.halted).toBe(false); + }); + + it("halts after three transient drop cycles without an intervening success, but not across successes", async () => { + const promptInputs: string[] = []; + let shouldFail = true; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + if (shouldFail) throw new Error("socket hang up"); + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + const messages: AgentMessage[] = [{ role: "user", content: "t1", timestamp: 1 } as AgentMessage]; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + notifyFailure: () => {}, + }; + const runtime = new AdvisorRuntime(agent, host, 0); + + const runTurn = async (content: string) => { + messages.push({ role: "user", content, timestamp: messages.length + 1 } as AgentMessage); + runtime.onTurnEnd(messages); + await Bun.sleep(0); + await Bun.sleep(0); + await Bun.sleep(0); + }; + + // Two failing drop cycles, then a success: the cycle counter resets. + await runTurn("f1"); + await runTurn("f2"); + expect(runtime.halted).toBe(false); + shouldFail = false; + await runTurn("ok"); + expect(runtime.halted).toBe(false); + + // Three CONSECUTIVE drop cycles with no success latch the runtime off. + shouldFail = true; + await runTurn("f3"); + await runTurn("f4"); + expect(runtime.halted).toBe(false); + await runTurn("f5"); + expect(runtime.halted).toBe(true); + const promptsAtHalt = promptInputs.length; + await runTurn("ignored"); + expect(promptInputs).toHaveLength(promptsAtHalt); + }); + + // The live incident shape: ONE agent + ONE advisor froze the whole + // process. The advisor's delta render (formatSessionHistoryMarkdown over + // the transcript slice) ran synchronously on the event loop; a post-reset + // replay of a multi-MB transcript blocked it for 600ms+ per render. These + // tests pin the bounded-stall contract and the correctness of the + // deferred/chunked path. + describe("large-transcript responsiveness", () => { + const bigMessage = (i: number, chars = 5_000): AgentMessage => { + const text = `msg-${i} ${"x".repeat(chars)}`; + return ( + i % 2 + ? { role: "assistant", content: [{ type: "text", text }], timestamp: i } + : { role: "user", content: text, timestamp: i } + ) as AgentMessage; + }; + + const stallSampler = () => { + let max = 0; + let last = performance.now(); + const timer = setInterval(() => { + const now = performance.now(); + const drift = now - last - 5; + if (drift > max) max = drift; + last = now; + }, 5); + return { stop: () => clearInterval(timer), max: () => max }; + }; + + const waitForPrompts = async (prompts: string[], count: number, timeoutMs = 10_000): Promise => { + const deadline = Date.now() + timeoutMs; + while (prompts.length < count && Date.now() < deadline) await Bun.sleep(5); + }; + + it("keeps the event loop responsive while replaying a multi-MB transcript", async () => { + const promptInputs: string[] = []; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + // ~2000 × 5KB ≈ 10MB replay — the post-reset/first-enable shape. + const messages = Array.from({ length: 2000 }, (_, i) => bigMessage(i)); + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }; + const runtime = new AdvisorRuntime(agent, host, 0); + const sampler = stallSampler(); + runtime.onTurnEnd(messages); + await waitForPrompts(promptInputs, 1); + sampler.stop(); + expect(promptInputs).toHaveLength(1); + // Nothing dropped: first and last transcript messages both rendered. + expect(promptInputs[0]).toContain("msg-0 "); + expect(promptInputs[0]).toContain("msg-1999 "); + // Pre-chunking this replay blocked the loop ~600ms+ in one shot; + // chunked rendering keeps individual stalls bounded. Generous + // ceiling: still fails the pre-fix behavior, tolerates CI/GC noise. + expect(sampler.max()).toBeLessThan(500); + runtime.dispose(); + }, 20_000); + + it("never splits a toolCall from its non-adjacent toolResult across chunk boundaries", async () => { + const promptInputs: string[] = []; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + // 150 messages force the chunked path (count > 100). The toolCall + // sits at index 99 — exactly where the count boundary closes the + // first chunk — and its result arrives in the NEXT chunk (index + // 148), far past any adjacency window: only the shared + // whole-delta result index can pair them. + const messages: AgentMessage[] = Array.from({ length: 150 }, (_, i) => bigMessage(i, 64)); + messages[99] = { + role: "assistant", + content: [{ type: "toolCall", id: "call-split", name: "read", arguments: { path: "x" } }], + timestamp: 99, + } as unknown as AgentMessage; + messages[100] = { + role: "custom", + customType: "hook", + content: "interleaved", + timestamp: 100, + } as AgentMessage; + messages[148] = { + role: "toolResult", + toolCallId: "call-split", + content: [{ type: "text", text: "result-body" }], + timestamp: 148, + } as AgentMessage; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }; + const runtime = new AdvisorRuntime(agent, host, 0); + runtime.onTurnEnd(messages); + await waitForPrompts(promptInputs, 1); + expect(promptInputs).toHaveLength(1); + expect(promptInputs[0]).toContain("read("); + // The call+result pair stayed in one chunk: rendered as completed, + // never as a spurious in-flight call. + expect(promptInputs[0]).toContain("⇒ ok"); + expect(promptInputs[0]).not.toContain("⇒ pending"); + runtime.dispose(); + }, 20_000); + + it("defers a single turn carrying a multi-MB payload instead of formatting it inline", async () => { + const promptInputs: string[] = []; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + const messages: AgentMessage[] = [{ role: "user", content: "before", timestamp: 1 } as AgentMessage]; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }; + const runtime = new AdvisorRuntime(agent, host, 0); + runtime.onTurnEnd(messages); + await waitForPrompts(promptInputs, 1); + expect(promptInputs).toHaveLength(1); + + // One turn, one message, multi-MB body (an edit-diff-sized payload): + // the count gate alone would format it synchronously; the size gate + // must route it through the deferred renderer — and still deliver. + messages.push({ + role: "assistant", + content: [{ type: "text", text: `huge ${"y".repeat(3_000_000)}` }], + timestamp: 2, + } as AgentMessage); + runtime.onTurnEnd(messages); + // Synchronous fast path would have pushed the prompt already; + // the deferred path leaves the queue empty at this tick. + expect(promptInputs).toHaveLength(1); + await waitForPrompts(promptInputs, 2); + expect(promptInputs).toHaveLength(2); + expect(promptInputs[1]).toContain("huge "); + runtime.dispose(); + }, 20_000); + + it("replays the full transcript after a reset lands mid-chunked-render", async () => { + const promptInputs: string[] = []; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + const messages = Array.from({ length: 400 }, (_, i) => bigMessage(i)); + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }; + const runtime = new AdvisorRuntime(agent, host, 0); + runtime.onTurnEnd(messages); + // Land the reset inside the render's first chunk yield. + await Bun.sleep(0); + runtime.reset(); + runtime.onTurnEnd(messages); + await waitForPrompts(promptInputs, 1); + // The aborted pre-reset render must not have advanced the cursor: + // the post-reset replay carries the whole transcript. + const replay = promptInputs.find(input => input.includes("msg-0 ") && input.includes("msg-399 ")); + expect(replay).toBeDefined(); + runtime.dispose(); + }, 20_000); + + it("delivers interleaved turns in order without loss while a chunked render is in flight", async () => { + const promptInputs: string[] = []; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + }, + abort: () => {}, + reset: () => {}, + state: { messages: [] }, + }; + const messages = Array.from({ length: 300 }, (_, i) => bigMessage(i)); + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }; + const runtime = new AdvisorRuntime(agent, host, 0); + runtime.onTurnEnd(messages); + // Second turn arrives while the first render is still on the chain. + messages.push({ role: "user", content: "late-arrival tail", timestamp: 300 } as AgentMessage); + runtime.onTurnEnd(messages); + const deadline = Date.now() + 10_000; + while (Date.now() < deadline && !promptInputs.join("\n").includes("late-arrival tail")) await Bun.sleep(5); + const combined = promptInputs.join("\n"); + // Every message exactly once, ordering preserved. + expect(combined).toContain("msg-0 "); + expect(combined).toContain("msg-299 "); + expect(combined.indexOf("msg-299 ")).toBeGreaterThan(combined.indexOf("msg-0 ")); + expect(combined.indexOf("late-arrival tail")).toBeGreaterThan(combined.indexOf("msg-299 ")); + expect(combined.match(/msg-150 /g)).toHaveLength(1); + expect(combined.match(/late-arrival tail/g)).toHaveLength(1); + runtime.dispose(); + }, 20_000); + }); + it("treats a clean prompt resolution with state.error as a failed turn (real Agent contract)", async () => { // `Agent.#runLoop` catches provider/stream failures internally — it resolves // `prompt()` cleanly and stores the message on `state.error` (e.g. the diff --git a/packages/coding-agent/src/advisor/runtime.ts b/packages/coding-agent/src/advisor/runtime.ts index f5165fc78..83a0c1308 100644 --- a/packages/coding-agent/src/advisor/runtime.ts +++ b/packages/coding-agent/src/advisor/runtime.ts @@ -1,6 +1,6 @@ import type { AgentMessage } from "@oh-my-pi/pi-agent-core"; import { estimateTokens } from "@oh-my-pi/pi-agent-core/compaction"; -import type { AssistantMessage, ImageContent, TextContent } from "@oh-my-pi/pi-ai"; +import type { AssistantMessage, ImageContent, TextContent, ToolResultMessage } from "@oh-my-pi/pi-ai"; import * as AIError from "@oh-my-pi/pi-ai/error"; import { logger } from "@oh-my-pi/pi-utils"; import { obfuscateToolArguments, type SecretObfuscator } from "../secrets/obfuscator"; @@ -68,6 +68,18 @@ export interface AdvisorRuntimeHost { notifyQuotaExhausted?(): void; } +/** + * A request rejection that no amount of retrying can fix for this advisor + * configuration: the provider refuses the model/request shape outright (e.g. + * "The 'gpt-5.3-codex-spark' model is not supported when using Codex with a + * ChatGPT account", code=invalid_request_error). Distinct from quota errors, + * which pause via the dedicated quota path and auto-resume on reset. + */ +function isPermanentAdvisorError(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + return /invalid_request_error|model[_ ]not[_ ]found|is not supported when|does not exist/i.test(message); +} + const ADVISOR_QUARANTINE_PREFIX = "Advisor response quarantined"; /** Signals that an advisor response was discarded before it could become model-visible context. */ @@ -183,6 +195,58 @@ export function buildAdvisorQuarantineSourceText(currentInput: string, messages: */ const MAX_COALESCE_ROUNDS = 3; +/** Messages formatted per event-loop slice in {@link AdvisorRuntime.#renderDeltaChunked}. */ +const RENDER_CHUNK_MESSAGES = 100; + +const ADVISOR_RENDER_OPTIONS = { + includeThinking: true, + includeToolIntent: true, + watchedRoles: true, + expandPrimaryContext: true, + expandEditDiffs: true, +} as const; + +/** Char budget for the synchronous fast-path render in {@link AdvisorRuntime.onTurnEnd}. */ +const FAST_RENDER_MAX_CHARS = 256 * 1024; + +/** + * Cheap early-exit probe: the message's aggregate string payload, capped at + * `cap`. Walks content (text/thinking blocks, tool arguments/results, string + * details like edit diffs) without serializing anything, so a single multi-MB + * message costs O(fields), not O(bytes). + */ +function estimateMessageChars(message: AgentMessage, cap: number): number { + let total = 0; + const add = (value: unknown, depth: number): boolean => { + if (total > cap) return true; + if (typeof value === "string") { + total += value.length; + return total > cap; + } + if (!value || typeof value !== "object" || depth >= 4) return false; + if (Array.isArray(value)) { + for (const item of value) if (add(item, depth + 1)) return true; + return false; + } + for (const key of Object.keys(value)) { + if (add((value as Record)[key], depth + 1)) return true; + } + return false; + }; + add(message, 0); + return total; +} + +/** Early-exit: does the delta from `from` exceed `cap` aggregate chars? */ +function deltaExceedsSize(all: readonly AgentMessage[], from: number, cap: number): boolean { + let total = 0; + for (let i = from; i < all.length; i++) { + total += estimateMessageChars(all[i]!, cap - total + 1); + if (total > cap) return true; + } + return false; +} + interface PendingDelta { text: string; turns: number; @@ -205,11 +269,26 @@ export class AdvisorRuntime { * marker so the advisor isn't re-fed the full ~1k-token rules each turn. * Cleared on every re-prime/seed and when a failed batch is dropped. */ #seenContext = new Map(); + /** Serializes deferred delta renders so `#lastCount`/`#seenContext` + * mutations stay ordered across queued turns. */ + #renderChain: Promise = Promise.resolve(); + /** Chunked renders queued or running on the chain; gates the sync fast path. */ + #renderBusy = 0; #pending: PendingDelta[] = []; #busy = false; #backlog = 0; #consecutiveFailures = 0; #failureNotified = false; + /** Completed 3-failure backlog-drop cycles since the last success/reset. */ + #droppedBacklogs = 0; + /** + * Hard stop after repeated drop cycles or a permanent request rejection + * (e.g. "model not supported"): without it the advisor re-attempts on every + * new delta forever, and in a shared daemon that unbounded churn burns CPU + * and starves every hosted session's event loop. Cleared only by an + * explicit {@link reset} (config rebuild, /new, session restart). + */ + #halted = false; #latestMessages?: AgentMessage[]; #waiters: CatchupWaiter[] = []; /** Bumped by every external {@link reset}/{@link dispose}. A drain iteration @@ -241,6 +320,10 @@ export class AdvisorRuntime { get failureNotified(): boolean { return this.#failureNotified; } + /** True after the runtime hard-stopped on repeated or permanent failures. */ + get halted(): boolean { + return this.#halted; + } /** * True when `#pending` is non-empty while the drain loop is busy — i.e., newer @@ -265,21 +348,82 @@ export class AdvisorRuntime { * the delta and forwarded to the reprime path so it is never silently dropped. */ onTurnEnd(messages?: AgentMessage[], opts?: { willContinue?: boolean }): void { - if (this.disposed || this.#quotaExhausted) return; - const all = messages ?? this.host.snapshotMessages(); + if (this.disposed || this.#quotaExhausted || this.#halted) return; + // Snapshot: the primary keeps appending to the live transcript array + // while a deferred render waits its turn on the chain. + const all = [...(messages ?? this.host.snapshotMessages())]; this.#latestMessages = all; const wip = opts?.willContinue ?? false; - const render = this.#renderDelta(all, wip); - if (render) { - this.#pending.push({ text: render, turns: 1, wip }); - this.#backlog++; - this.#notifyWaiters(); - void this.#drain(); + // Fast path: a small delta with no render in flight formats in one + // bounded synchronous call — the common per-turn shape. Large deltas + // (post-reset replay of a multi-MB transcript, or a single turn + // carrying a multi-MB edit diff) defer to the chunked renderer; + // formatted in one synchronous call those block the event loop for + // hundreds of milliseconds, freezing EVERY session hosted by a shared + // daemon. + if ( + this.#renderBusy === 0 && + all.length - this.#lastCount <= RENDER_CHUNK_MESSAGES && + !deltaExceedsSize(all, this.#lastCount, FAST_RENDER_MAX_CHARS) + ) { + const render = this.#renderDelta(all, wip); + if (render) { + this.#pending.push({ text: render, turns: 1, wip }); + this.#backlog++; + this.#notifyWaiters(); + void this.#drain(); + } + return; } + // Backlog is accounted eagerly so waitForCatchup sees the queued turn + // immediately even though rendering is deferred. + this.#backlog++; + const epoch = this.#epoch; + void this.#enqueueRender(async () => { + // A reset/dispose that landed before this queued render runs has + // already rewound the cursor — rendering now would advance it again + // and silently swallow the pre-reset replay. + if (this.disposed || this.#epoch !== epoch) { + this.#backlog = Math.max(0, this.#backlog - 1); + return; + } + let render: string | null = null; + try { + render = await this.#renderDeltaChunked(all, wip, epoch); + } catch (err) { + logger.warn("advisor delta render failed", { err: String(err) }); + } + if (this.disposed || this.#epoch !== epoch) return; + if (render) { + this.#pending.push({ text: render, turns: 1, wip }); + this.#notifyWaiters(); + void this.#drain(); + } else { + this.#backlog = Math.max(0, this.#backlog - 1); + this.#notifyWaiters(); + } + }); + } + + /** + * Serialize every chunked render — deferred turn renders AND the + * maintainContext reprime — on one chain so `#lastCount`/`#seenContext` + * mutations never interleave across chunk yields. `#renderBusy` gates the + * synchronous fast path in {@link onTurnEnd} while anything is queued or + * running here. + */ + #enqueueRender(task: () => Promise): Promise { + this.#renderBusy++; + const result = this.#renderChain.then(task); + const settle = (): void => { + this.#renderBusy--; + }; + this.#renderChain = result.then(settle, settle); + return result; } waitForCatchup(maxMs: number, threshold: number, signal?: AbortSignal): Promise { - if (this.disposed || signal?.aborted || this.#backlog < threshold || this.#quotaExhausted) + if (this.disposed || signal?.aborted || this.#backlog < threshold || this.#quotaExhausted || this.#halted) return Promise.resolve(); const { promise, resolve } = Promise.withResolvers(); let waiter!: CatchupWaiter; @@ -342,6 +486,8 @@ export class AdvisorRuntime { reset(): void { this.#epoch++; this.#quotaExhausted = false; + this.#halted = false; + this.#droppedBacklogs = 0; this.#resetAdvisorContext(true, true); } @@ -351,17 +497,42 @@ export class AdvisorRuntime { * advisor (which would be expensive and likely stale). */ seedTo(count: number): void { + this.#epoch++; this.#lastCount = count; this.#pending = []; this.#backlog = 0; this.#consecutiveFailures = 0; + this.#droppedBacklogs = 0; this.#failureNotified = false; this.#seenContext.clear(); this.#wakeAllWaiters(); } - #renderDelta(messages?: AgentMessage[], wip = false): string | null { - const all = messages ?? this.#latestMessages ?? this.host.snapshotMessages(); + /** + * Account one completed 3-failure backlog-drop cycle. Repeated cycles (or a + * single permanent request rejection, e.g. "model not supported") hard-stop + * the runtime: in a shared daemon, unbounded advisor churn re-builds heavy + * context on every new delta and starves every hosted session's event loop. + * Only an explicit {@link reset} (config rebuild, /new, restart) resumes. + */ + #noteDroppedBacklog(error: unknown): void { + this.#droppedBacklogs++; + if (this.#droppedBacklogs < 3 && !isPermanentAdvisorError(error)) return; + this.#halted = true; + this.#pending = []; + this.#wakeAllWaiters(); + logger.warn("advisor halted after repeated failures; use /advisor or reload config to re-enable", { + droppedBacklogs: this.#droppedBacklogs, + err: String(error), + }); + } + + /** + * Advance the cursor and produce the dedup'd/obfuscated delta since + * `#lastCount`, or null when empty. Synchronous: callers on the render + * chain must not interleave (see {@link #enqueueRender}). + */ + #composeDelta(all: AgentMessage[]): AgentMessage[] | null { if (all.length < this.#lastCount) { this.#lastCount = all.length; this.#seenContext.clear(); @@ -374,19 +545,90 @@ export class AdvisorRuntime { this.#lastCount = all.length; if (delta.length === 0) return null; const obfuscator = this.host.obfuscator; - const formattedDelta = obfuscator?.hasSecrets() ? obfuscateAdvisorDelta(obfuscator, delta) : delta; - const md = formatSessionHistoryMarkdown(formattedDelta, { - includeThinking: true, - includeToolIntent: true, - watchedRoles: true, - expandPrimaryContext: true, - expandEditDiffs: true, - }); + return obfuscator?.hasSecrets() ? obfuscateAdvisorDelta(obfuscator, delta) : delta; + } + + #finishRender(md: string, wip: boolean): string | null { if (!md.trim()) return null; const heading = wip ? "### Session update [in progress — more steps follow]" : "### Session update"; return `${heading}\n\n${md}`; } + /** Bounded synchronous render for small deltas (the common per-turn shape). */ + #renderDelta(messages?: AgentMessage[], wip = false): string | null { + const all = messages ?? this.#latestMessages ?? this.host.snapshotMessages(); + const formattedDelta = this.#composeDelta(all); + if (!formattedDelta) return null; + return this.#finishRender(formatSessionHistoryMarkdown(formattedDelta, ADVISOR_RENDER_OPTIONS), wip); + } + + /** + * Render the transcript delta since `#lastCount` as advisor markdown, + * yielding the event loop between message chunks. A post-reset replay + * formats the ENTIRE transcript; done synchronously that blocks the loop + * for hundreds of milliseconds per ~10MB of transcript, freezing every + * session hosted by a shared daemon. Chunk boundaries never start on a + * toolResult so a tool call and its result always format together. + * Returns null when the delta is empty or `epoch` was invalidated during + * a yield. + */ + async #renderDeltaChunked( + messages: AgentMessage[] | undefined, + wip: boolean, + epoch: number, + ): Promise { + const all = messages ?? this.#latestMessages ?? this.host.snapshotMessages(); + const formattedDelta = this.#composeDelta(all); + if (!formattedDelta) return null; + let md: string; + if ( + formattedDelta.length <= RENDER_CHUNK_MESSAGES && + !deltaExceedsSize(formattedDelta, 0, FAST_RENDER_MAX_CHARS) + ) { + md = formatSessionHistoryMarkdown(formattedDelta, ADVISOR_RENDER_OPTIONS); + } else { + // Chunks are bounded by BOTH message count and estimated payload + // size, so a handful of huge messages (multi-MB edit diffs) never + // collapses into one long synchronous format call. A single + // oversized message is irreducible — it forms its own chunk. + // Call/result pairing survives chunk boundaries: every chunk shares + // one whole-delta result index and consumed-id set, so a toolCall + // renders "⇒ ok" even when its toolResult lands chunks later and the + // result is never re-rendered as an orphan. + const toolResultIndex = new Map(); + for (const message of formattedDelta) { + if (message.role === "toolResult") toolResultIndex.set(message.toolCallId, message); + } + const chunkOptions = { + ...ADVISOR_RENDER_OPTIONS, + toolResultIndex, + consumedToolCallIds: new Set(), + }; + const parts: string[] = []; + let start = 0; + let count = 0; + let chars = 0; + for (let end = 0; end < formattedDelta.length; ) { + chars += estimateMessageChars(formattedDelta[end]!, FAST_RENDER_MAX_CHARS + 1); + count++; + end++; + const flush = + end === formattedDelta.length || count >= RENDER_CHUNK_MESSAGES || chars > FAST_RENDER_MAX_CHARS; + if (!flush) continue; + if (start > 0) { + await Bun.sleep(0); + if (this.disposed || this.#epoch !== epoch) return null; + } + parts.push(formatSessionHistoryMarkdown(formattedDelta.slice(start, end), chunkOptions)); + start = end; + count = 0; + chars = 0; + } + md = parts.filter(part => part.trim()).join("\n"); + } + return this.#finishRender(md, wip); + } + /** * Collapse a re-injected primary-context prompt (plan/goal mode rules, the * approved plan) to a short marker when its body is byte-identical to the @@ -494,7 +736,13 @@ export class AdvisorRuntime { turns += lateItems.reduce((sum, b) => sum + b.turns, 0); if (lateItems.length > 0) wip = lateItems.at(-1)!.wip; this.#resetAdvisorContext(false, false); - return { batch: this.#renderDelta(this.#latestMessages, wip), finalTurns: turns, wip }; + const reprimed = await this.#enqueueRender(async () => + this.disposed || this.#epoch !== epoch + ? null + : this.#renderDeltaChunked(this.#latestMessages, wip, epoch), + ); + if (this.#epoch !== epoch) return null; + return { batch: reprimed, finalTurns: turns, wip }; } } @@ -562,6 +810,7 @@ export class AdvisorRuntime { success = true; this.#consecutiveFailures = 0; this.#failureNotified = false; + this.#droppedBacklogs = 0; } catch (err) { // reset()/dispose() aborts the in-flight prompt; treat it as a // reset, not a transient failure — drop the stale batch. @@ -595,6 +844,7 @@ export class AdvisorRuntime { success = true; this.#consecutiveFailures = 0; this.#failureNotified = false; + this.#droppedBacklogs = 0; } catch (retryErr) { this.#rollbackFailedTurn(retrySnapshot); if (this.#epoch !== epoch) continue; @@ -647,6 +897,7 @@ export class AdvisorRuntime { } this.#consecutiveFailures = 0; this.#seenContext.clear(); + this.#noteDroppedBacklog(retryErr); success = true; } else { this.#pending.unshift({ text: batch, turns: finalTurns, wip }); @@ -703,6 +954,7 @@ export class AdvisorRuntime { // prompts instead of marking them "unchanged" against content the // advisor never received. this.#seenContext.clear(); + this.#noteDroppedBacklog(err); success = true; } else { this.#pending.unshift({ text: batch, turns: finalTurns, wip }); diff --git a/packages/coding-agent/src/session/session-history-format.ts b/packages/coding-agent/src/session/session-history-format.ts index de4d07041..a88145d3e 100644 --- a/packages/coding-agent/src/session/session-history-format.ts +++ b/packages/coding-agent/src/session/session-history-format.ts @@ -46,6 +46,15 @@ export interface HistoryFormatOptions { * this so it sees what changed without re-reading the file. */ expandEditDiffs?: boolean; + /** + * Chunked rendering support: a caller formatting one logical transcript in + * several calls (the advisor's chunked delta render) passes a result index + * built over the WHOLE delta plus one shared consumed-id set, so a toolCall + * finds its toolResult across chunk boundaries and the result is never + * re-rendered as an orphan in a later chunk. + */ + toolResultIndex?: ReadonlyMap; + consumedToolCallIds?: Set; } /** Max length of the primary-arg summary inside `→ tool(...)` lines. */ @@ -273,13 +282,19 @@ export function formatSessionHistoryMarkdown(messages: unknown[], opts?: History } // Index tool results by call id so each toolCall collapses to one line. - const resultsByCallId = new Map(); - for (const msg of typed) { - if (msg.role === "toolResult") { - resultsByCallId.set(msg.toolCallId, msg); + // Chunked callers supply a whole-delta index + shared consumed set so + // call/result pairs resolve across chunk boundaries. + let resultsByCallId = opts?.toolResultIndex; + if (!resultsByCallId) { + const local = new Map(); + for (const msg of typed) { + if (msg.role === "toolResult") { + local.set(msg.toolCallId, msg); + } } + resultsByCallId = local; } - const consumed = new Set(); + const consumed = opts?.consumedToolCallIds ?? new Set(); // In watched mode, consecutive same-role messages collapse under one label // (the watched agent emits one assistant message per tool call, so otherwise // every call repeats `**agent**:`). Cleared whenever a