diff --git a/packages/coding-agent/.changes/res-1272-coalesced-bookkeeping-appends.md b/packages/coding-agent/.changes/res-1272-coalesced-bookkeeping-appends.md new file mode 100644 index 00000000000..0de98f3d581 --- /dev/null +++ b/packages/coding-agent/.changes/res-1272-coalesced-bookkeeping-appends.md @@ -0,0 +1 @@ +- Fixed unbounded session-journal growth from derived bookkeeping: child usage attribution now flushes one entry per child turn instead of one per model request, and idle status sweeps no longer persist fabricated fallback verdicts, duplicate statuses, or retry failed summary generations (including paid model calls) every 25 seconds on unchanged content. diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index be375cc0481..e6b8c968bb9 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -255,7 +255,13 @@ import { transitionSessionAction, type WakePolicy, } from "./session-action-store.js"; -import type { BranchSummaryEntry, CompactionEntry, SessionContext, SessionMessageEntry } from "./session-manager.js"; +import type { + BranchSummaryEntry, + ChildUsageAttributionEntry, + CompactionEntry, + SessionContext, + SessionMessageEntry, +} from "./session-manager.js"; import { CURRENT_SESSION_VERSION, getLatestCompactionEntry, @@ -1037,6 +1043,26 @@ function waitForPromiseOrAbort( }); } +// Bounds how much accumulated child usage a parent process crash can lose. +const RLM_CHILD_USAGE_FLUSH_MAX_PENDING_MS = 60_000; + +/** Label a child completion's usage by the nearest preceding prompt that triggered it. */ +function rlmChildUsageOrigin( + messages: readonly AgentMessage[], + assistant: AssistantMessage, +): ChildUsageAttributionEntry["origin"] { + for (let index = messages.lastIndexOf(assistant) - 1; index >= 0; index--) { + const message = messages[index]; + if (message.role !== "user" && message.role !== "custom") continue; + return message.role === "custom" && isAgentSessionMessage(message) + ? message.details.id.startsWith("spawn:") + ? "spawn_task" + : "agent_message" + : "direct_user"; + } + return "direct_user"; +} + function attributeChildUsage(parentUsage: Usage, childUsage: Usage): void { const parentContextTokens = parentUsage.totalTokens || @@ -10599,6 +10625,43 @@ export class AgentSession { if (!requestedSessionName) await this._assertRlmSubagentSessionNameAvailable(sessionName); const startedAt = Date.now(); const parentAssistantForUsage = this._findLastAssistantMessage(); + // Child completions accumulate per origin and flush one durable entry per + // settle boundary (agent_end, settlement); the staleness checkpoints and + // timer bound crash loss to one window of accumulated usage. + const pendingChildUsage = new Map(); + let pendingChildUsageSince = 0; + let pendingChildUsageTimer: ReturnType | undefined; + const flushPendingChildUsageAttribution = () => { + if (pendingChildUsageTimer !== undefined) { + clearTimeout(pendingChildUsageTimer); + pendingChildUsageTimer = undefined; + } + if (pendingChildUsage.size === 0 || !parentAssistantForUsage) return; + const batches = [...pendingChildUsage.entries()]; + pendingChildUsage.clear(); + const parentEntry = this._findAssistantEntryForMessage(parentAssistantForUsage); + if (!parentEntry) return; + for (const [origin, childUsage] of batches) { + try { + this.sessionManager.appendChildUsageAttribution( + parentEntry.id, + childUsage, + parentAssistantForUsage.usage, + origin, + ); + } catch { + // Attribution is recoverable bookkeeping; a failed append must not break run settlement. + } + } + }; + const flushPendingChildUsageIfStale = () => { + if ( + pendingChildUsage.size > 0 && + Date.now() - pendingChildUsageSince >= RLM_CHILD_USAGE_FLUSH_MAX_PENDING_MS + ) { + flushPendingChildUsageAttribution(); + } + }; let runningToolCount = 0; let childSession: AgentSession | undefined; const run: RlmChildRun = { @@ -10712,34 +10775,30 @@ export class AgentSession { run.activity = { kind: "waiting" }; emitChildUpdate(); } else if (event.type === "agent_end") { + flushPendingChildUsageAttribution(); run.activity = undefined; emitChildUpdate(); } else if (event.type === "message_end" && event.message.role === "assistant") { const assistant = event.message as AssistantMessage; if (assistant.stopReason !== "error" && assistant.stopReason !== "aborted") { + // Flush before the fold: a persisted aggregate may only include + // completions whose childUsage is durable with or before it. + flushPendingChildUsageIfStale(); attributeChildUsage(parentAssistantForUsage?.usage ?? emptyUsage(), assistant.usage); if (parentAssistantForUsage) { - const parentEntry = this._findAssistantEntryForMessage(parentAssistantForUsage); - if (parentEntry) { - const messages = child.messages; - const assistantIndex = messages.lastIndexOf(assistant); - const precedingPrompt = messages - .slice(0, assistantIndex) - .reverse() - .find((message) => message.role === "user" || message.role === "custom"); - const origin = - precedingPrompt?.role === "custom" && isAgentSessionMessage(precedingPrompt) - ? precedingPrompt.details.id.startsWith("spawn:") - ? "spawn_task" - : "agent_message" - : "direct_user"; - this.sessionManager.appendChildUsageAttribution( - parentEntry.id, - assistant.usage, - parentAssistantForUsage.usage, - origin, + const origin = rlmChildUsageOrigin(child.messages, assistant); + if (pendingChildUsage.size === 0) { + pendingChildUsageSince = Date.now(); + // Wall-clock backstop for long tool runs without checkpoints. + pendingChildUsageTimer = setTimeout( + flushPendingChildUsageAttribution, + RLM_CHILD_USAGE_FLUSH_MAX_PENDING_MS, ); + pendingChildUsageTimer.unref?.(); } + const bucket = pendingChildUsage.get(origin) ?? emptyUsage(); + addAssistantUsage(bucket, assistant.usage); + pendingChildUsage.set(origin, bucket); } } const text = compactRlmText(readAssistantText(assistant)); @@ -10753,6 +10812,7 @@ export class AgentSession { emitChildUpdate(); } } else if (event.type === "tool_execution_start") { + flushPendingChildUsageIfStale(); run.toolUseCount += 1; runningToolCount += 1; run.activity = { kind: "executing", toolName: event.toolName }; @@ -10901,6 +10961,7 @@ export class AgentSession { } } } finally { + flushPendingChildUsageAttribution(); if (run.detachedDeletion) { run.deletionRunFinished = true; if (!run.settled) { diff --git a/packages/coding-agent/src/modes/daemon/daemon-session-summarizer.ts b/packages/coding-agent/src/modes/daemon/daemon-session-summarizer.ts index f2dba51f67b..5c01b741bb8 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-session-summarizer.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-session-summarizer.ts @@ -8,6 +8,10 @@ import type { ActiveSessionState } from "./active-session-state.js"; const SWEEP_INTERVAL_MS = 25_000; // Collapse a tool-use loop's rapid turn_end bursts into one summarization. const SETTLE_DEBOUNCE_MS = 2_000; +// Idle generations stop retrying (and paying) on unchanged content until the +// backoff elapses, so transient outages and late credentials still recover. +const IDLE_GENERATION_ATTEMPT_LIMIT = 3; +const IDLE_GENERATION_RETRY_BACKOFF_MS = 30 * 60_000; const SUMMARY_MODEL_PROVIDER = "prime-inference"; const SUMMARY_MODEL_ID = "qwen/qwen3-30b-a3b-instruct-2507"; @@ -207,6 +211,11 @@ export class DaemonSessionSummarizer { private readonly inFlight = new Map(); // Sessions requested while one was running; get one more pass on completion. private readonly rerunRequested = new Set(); + // Failed idle generations per session, keyed to the settled content they saw. + private readonly failedIdleGenerations = new Map< + string, + { contentKey: string; attempts: number; lastFailureAt: number } + >(); constructor( private readonly listSessions: () => readonly ActiveSessionState[], @@ -242,6 +251,7 @@ export class DaemonSessionSummarizer { controller.abort(); } this.rerunRequested.clear(); + this.failedIdleGenerations.clear(); } /** Drop any pending work for a session that is closing. */ @@ -253,6 +263,7 @@ export class DaemonSessionSummarizer { } this.inFlight.get(activeSessionId)?.abort(); this.rerunRequested.delete(activeSessionId); + this.failedIdleGenerations.delete(activeSessionId); } /** Seed in-memory status from the persisted entry when a session is added. */ @@ -306,6 +317,18 @@ export class DaemonSessionSummarizer { if (contentUnchanged && !isWorking && !owesIdleVerdict && !owesSummary) { return; } + // The leaf entry id is the branch-tip identity (appends, edits, and branch + // navigation all move it; counts and timestamps collide across siblings). + const contentKey = `${session.sessionManager.getLeafId() ?? "root"}:${messageCount}`; + const failed = this.failedIdleGenerations.get(id); + if ( + !isWorking && + failed?.contentKey === contentKey && + failed.attempts >= IDLE_GENERATION_ATTEMPT_LIMIT && + Date.now() - failed.lastFailureAt < IDLE_GENERATION_RETRY_BACKOFF_MS + ) { + return; + } // Include the in-progress message so a long streaming turn gets a live recap. const streaming = isWorking ? session.state.streamingMessage : undefined; const contextMessages = streaming ? [...messages, streaming] : messages; @@ -319,6 +342,16 @@ export class DaemonSessionSummarizer { isWorking, signal: controller.signal, }); + if (generated) { + this.failedIdleGenerations.delete(id); + } else if (!isWorking && !controller.signal.aborted) { + // The aborted check keeps a racing forget() from repopulating the map. + this.failedIdleGenerations.set(id, { + contentKey, + attempts: failed?.contentKey === contentKey ? failed.attempts + 1 : 1, + lastFailureAt: Date.now(), + }); + } // A failed classification on an idle session would spin at "working" // forever (the activity axis holds unjudged idle sessions there), so // settle it to needs_input. @@ -356,12 +389,20 @@ export class DaemonSessionSummarizer { previous?.taskState !== status.taskState || (!isWorking && previous?.basedOnMessageCount !== status.basedOnMessageCount); state.summaryState = status; - // Persist only settled idle verdicts, never mid-stream. - if (!isWorking) { - try { - session.sessionManager.appendAgentStatus(status); - } catch { - // best-effort; in-memory status still shows + // Persist only settled idle verdicts from real classifications that + // differ from the latest persisted entry: idle sweeps must not grow the journal. + if (!isWorking && generated) { + const persisted = session.sessionManager.getLatestAgentStatus(); + if ( + persisted?.summary !== status.summary || + persisted.taskState !== status.taskState || + persisted.basedOnMessageCount !== status.basedOnMessageCount + ) { + try { + session.sessionManager.appendAgentStatus(status); + } catch { + // best-effort; in-memory status still shows + } } } if (changed) { diff --git a/packages/coding-agent/test/agent-session-recursion.test.ts b/packages/coding-agent/test/agent-session-recursion.test.ts index 2dad4fcd9dc..54f929d33d0 100644 --- a/packages/coding-agent/test/agent-session-recursion.test.ts +++ b/packages/coding-agent/test/agent-session-recursion.test.ts @@ -248,6 +248,51 @@ describe("AgentSession rlm recursion", () => { return session; } + /** Session with a zero-usage parent assistant whose child answers a tool loop: usages[i] per request, tool calls until the last, then stop. */ + function createToolLoopSession(requests: number, usages: Usage[], onRequest?: (toolResultCount: number) => void) { + const tool = { + name: "echo", + description: "Echo a value", + label: "echo", + parameters: Type.Object({ value: Type.String() }), + execute: async (_toolCallId: string, params: { value: string }) => ({ + content: [{ type: "text" as const, text: params.value }], + details: {}, + }), + }; + const root = createSession({ + customTools: [tool], + streamFn: (_model, context) => { + const toolResultCount = context.messages.filter((message) => message.role === "toolResult").length; + onRequest?.(toolResultCount); + const last = toolResultCount >= requests - 1; + const stream = createAssistantMessageEventStream(); + queueMicrotask(() => { + const message = last + ? assistantMessage("done", usages[toolResultCount]) + : { + ...assistantMessage("", usages[toolResultCount]), + content: [ + { + type: "toolCall" as const, + id: `echo-${toolResultCount}`, + name: "echo", + arguments: { value: "ok" }, + }, + ], + stopReason: "toolUse" as const, + }; + stream.push({ type: "done", reason: last ? "stop" : "toolUse", message }); + }); + return stream; + }, + }); + const parentAssistant = assistantMessage("running ipython", usage(0, 0)); + root.agent.state.messages.push(parentAssistant); + root.sessionManager.appendMessage(parentAssistant); + return root; + } + function createAbortInsensitiveChild(): { child: AgentSession; completion: ReturnType>; @@ -2462,56 +2507,58 @@ describe("AgentSession rlm recursion", () => { expect(attribution.aggregateUsage.cost.total).toBe(10); }); - it("attributes every tool-loop turn in the admitted task to spawn usage", async () => { - const tool = { - name: "echo", - description: "Echo a value", - label: "echo", - parameters: Type.Object({ value: Type.String() }), - execute: async (_toolCallId: string, params: { value: string }) => ({ - content: [{ type: "text" as const, text: params.value }], - details: {}, - }), - }; - const root = createSession({ - customTools: [tool], - streamFn: (_model, context) => { - const toolResultCount = context.messages.filter((message) => message.role === "toolResult").length; - const stream = createAssistantMessageEventStream(); - queueMicrotask(() => { - const message = - toolResultCount === 0 - ? { - ...assistantMessage("", usage(1, 1)), - content: [ - { type: "toolCall" as const, id: "echo-1", name: "echo", arguments: { value: "ok" } }, - ], - stopReason: "toolUse" as const, - } - : assistantMessage("done", usage(2, 2)); - stream.push({ - type: "done", - reason: toolResultCount === 0 ? "toolUse" : "stop", - message, - }); - }); - return stream; - }, - }); - const parentAssistant = assistantMessage("running ipython", usage(0, 0)); - root.agent.state.messages.push(parentAssistant); - root.sessionManager.appendMessage(parentAssistant); + it("coalesces the admitted task's tool-loop turns into one flushed spawn-usage attribution", async () => { + const root = createToolLoopSession(2, [usage(1, 1), usage(2, 2)]); await root.runRlmChild("use a tool"); await vi.waitFor(() => { const attributions = root.sessionManager .getEntries() .filter((entry) => entry.type === "child_usage_attributed"); - expect(attributions).toHaveLength(2); - expect(attributions.map((entry) => entry.origin)).toEqual(["spawn_task", "spawn_task"]); + expect(attributions).toHaveLength(1); + expect(attributions[0]?.origin).toBe("spawn_task"); + expect(attributions[0]?.childUsage.input).toBe(3); + expect(attributions[0]?.childUsage.output).toBe(3); }); }); + it("flushes a stale pending usage batch before extending it, bounding crash loss", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + try { + // The batch from the first two completions is older than the staleness + // bound when the third lands. + const root = createToolLoopSession(3, [usage(1, 1), usage(2, 2), usage(4, 4)], (toolResultCount) => { + if (toolResultCount === 2) vi.setSystemTime(Date.now() + 61_000); + }); + + await root.runRlmChild("use a tool"); + await vi.waitFor(() => { + const attributions = root.sessionManager + .getEntries() + .filter((entry) => entry.type === "child_usage_attributed"); + expect(attributions.map((entry) => [entry.childUsage.input, entry.childUsage.output])).toEqual([ + [3, 3], + [4, 4], + ]); + expect(attributions.map((entry) => entry.origin)).toEqual(["spawn_task", "spawn_task"]); + // Each aggregate covers exactly the completions durable with or + // before it, so any prefix replays to the exact own spend. + expect(attributions.map((entry) => entry.aggregateUsage.input)).toEqual([3, 7]); + }); + + const sessionFile = root.sessionManager.getSessionFile(); + if (!sessionFile) throw new Error("parent session file was not created"); + const reloadedAttributions = SessionManager.open(sessionFile, join(tempDir, "sessions")) + .getEntries() + .filter((entry) => entry.type === "child_usage_attributed"); + const childTotal = reloadedAttributions.reduce((total, entry) => total + entry.childUsage.input, 0); + // Parent own spend is zero here, so the reloaded aggregate must equal the summed child usage. + expect(reloadedAttributions.at(-1)?.aggregateUsage.input).toBe(childTotal); + } finally { + vi.useRealTimers(); + } + }); + it("gets and persists per-chat max-depth changes without transcript messages", async () => { const root = createSession(); const originalMessages = [...root.messages]; diff --git a/packages/coding-agent/test/daemon-session-summarizer-lifecycle.test.ts b/packages/coding-agent/test/daemon-session-summarizer-lifecycle.test.ts index c14c2a8a191..ad5cdbb3bcf 100644 --- a/packages/coding-agent/test/daemon-session-summarizer-lifecycle.test.ts +++ b/packages/coding-agent/test/daemon-session-summarizer-lifecycle.test.ts @@ -26,6 +26,7 @@ function makeState( sessionManager: { appendAgentStatus: (s: unknown) => appended.push(s), getLatestAgentStatus: () => opts.persisted, + getLeafId: () => null, }, }, }, diff --git a/packages/coding-agent/test/daemon-session-summarizer.test.ts b/packages/coding-agent/test/daemon-session-summarizer.test.ts index 18cc2752a2a..659885dacd6 100644 --- a/packages/coding-agent/test/daemon-session-summarizer.test.ts +++ b/packages/coding-agent/test/daemon-session-summarizer.test.ts @@ -160,6 +160,9 @@ describe("daemon session summarizer", () => { messages: AgentMessage[]; isSessionActive: boolean; summaryState?: AgentStatus; + persistedStatus?: AgentStatus; + appendAgentStatus?: (status: AgentStatus) => void; + getLeafId?: () => string | null; }): ActiveSessionState { return { activeSessionId: "active-1", @@ -170,7 +173,11 @@ describe("daemon session summarizer", () => { messages: options.messages, modelRegistry: {}, state: { streamingMessage: undefined }, - sessionManager: { appendAgentStatus: () => {} }, + sessionManager: { + appendAgentStatus: options.appendAgentStatus ?? (() => {}), + getLatestAgentStatus: () => options.persistedStatus, + getLeafId: options.getLeafId ?? (() => null), + }, }, }, } as unknown as ActiveSessionState; @@ -201,6 +208,109 @@ describe("daemon session summarizer", () => { expect(onStatusChanged).toHaveBeenCalledOnce(); }); + function failingIdleSetup( + stateOptions: Parameters[0], + generateFn: () => Promise = async () => undefined, + ) { + const state = makeState(stateOptions); + const generate = vi.fn(generateFn); + const summarizer = new DaemonSessionSummarizer(() => [state], undefined, generate); + const internal = summarizer as unknown as { + summarize(state: ActiveSessionState): Promise; + failedIdleGenerations: Map; + }; + return { state, generate, summarizer, internal }; + } + + // Warmup pins the ceiling itself: repeated failing idle sweeps stop paying + // after three attempts and never persist the fabricated in-memory fallback. + test.each([ + { + rearm: "branch navigation moving the leaf at the same length", + trigger: (leaf: { id: string }) => { + leaf.id = "leaf-b"; + }, + }, + { + rearm: "the backoff elapsing so external failures recover", + trigger: () => vi.setSystemTime(Date.now() + 31 * 60_000), + }, + ])("the exhausted idle retry ceiling re-arms on $rearm", async ({ trigger }) => { + vi.useFakeTimers(); + try { + const appendAgentStatus = vi.fn(); + const leaf = { id: "leaf-a" }; + const { state, generate, internal } = failingIdleSetup({ + messages: [userMessage("hi")], + isSessionActive: false, + appendAgentStatus, + getLeafId: () => leaf.id, + }); + + for (let sweep = 0; sweep < 5; sweep++) { + await internal.summarize(state); + } + expect(generate).toHaveBeenCalledTimes(3); + expect(appendAgentStatus).not.toHaveBeenCalled(); + expect(state.summaryState).toEqual({ summary: "", taskState: "needs_input", basedOnMessageCount: 1 }); + + trigger(leaf); + await internal.summarize(state); + expect(generate).toHaveBeenCalledTimes(4); + expect(appendAgentStatus).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + test("a forget() during an in-flight idle generation leaves no failure record behind", async () => { + let release!: () => void; + const gate = new Promise((resolveGate) => { + release = resolveGate; + }); + const { state, summarizer, internal } = failingIdleSetup( + { messages: [userMessage("hi")], isSessionActive: false }, + async () => { + await gate; + return undefined; + }, + ); + + const pass = internal.summarize(state); + summarizer.forget("active-1"); + release(); + await pass; + + expect(internal.failedIdleGenerations.size).toBe(0); + }); + + test("an idle re-settle matching the latest persisted status appends nothing", async () => { + const appendAgentStatus = vi.fn(); + const persisted: AgentStatus = { + summary: "Awaiting review", + taskState: "needs_input", + basedOnMessageCount: 1, + }; + const state = makeState({ + messages: [userMessage("hi")], + isSessionActive: false, + persistedStatus: persisted, + appendAgentStatus, + }); + + const onStatusChanged = vi.fn(); + const summarizer = new DaemonSessionSummarizer( + () => [state], + onStatusChanged, + async () => ({ summary: "Awaiting review", taskState: "needs_input" as const }), + ); + await (summarizer as unknown as { summarize(state: ActiveSessionState): Promise }).summarize(state); + + expect(appendAgentStatus).not.toHaveBeenCalled(); + expect(onStatusChanged).toHaveBeenCalledOnce(); + expect(state.summaryState).toEqual(persisted); + }); + test("a working refresh with unchanged text stays quiet", async () => { const previous: AgentStatus = { summary: "Working on it", taskState: "needs_input", basedOnMessageCount: 2 }; const state = makeState({