Skip to content
Original file line number Diff line number Diff line change
@@ -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.
94 changes: 72 additions & 22 deletions packages/coding-agent/src/core/agent-session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -1037,6 +1043,27 @@ function waitForPromiseOrAbort<T>(
});
}

// Bounds how much accumulated child usage a parent process crash can lose
// before the batch is flushed durable.
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 ||
Expand Down Expand Up @@ -10599,6 +10626,41 @@ 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 attribution
// entry per settle boundary (child agent_end, run settlement) instead of
// one journal append per child assistant message. Crash durability is
// bounded, not per-message: a batch older than the staleness bound flushes
// before it grows or a tool starts, so process death loses at most that
// window of accumulated child usage.
const pendingChildUsage = new Map<ChildUsageAttributionEntry["origin"], Usage>();
let pendingChildUsageSince = 0;
const flushPendingChildUsageAttribution = () => {
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
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
) {
flushPendingChildUsageAttribution();
}
};
let runningToolCount = 0;
let childSession: AgentSession | undefined;
const run: RlmChildRun = {
Expand Down Expand Up @@ -10712,34 +10774,20 @@ 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") {
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,
);
}
flushPendingChildUsageIfStale();
const origin = rlmChildUsageOrigin(child.messages, assistant);
if (pendingChildUsage.size === 0) pendingChildUsageSince = Date.now();
const bucket = pendingChildUsage.get(origin) ?? emptyUsage();
addAssistantUsage(bucket, assistant.usage);
pendingChildUsage.set(origin, bucket);
Comment thread
cursor[bot] marked this conversation as resolved.
}
}
const text = compactRlmText(readAssistantText(assistant));
Expand All @@ -10753,6 +10801,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 };
Expand Down Expand Up @@ -10901,6 +10950,7 @@ export class AgentSession {
}
}
} finally {
flushPendingChildUsageAttribution();
if (run.detachedDeletion) {
run.deletionRunFinished = true;
if (!run.settled) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,11 @@ 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;
// An idle session whose generation keeps failing on the same settled content
// stops retrying (and paying for model calls) until new activity arrives or
// the backoff elapses, so transient outages and late credentials 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";
Expand Down Expand Up @@ -207,6 +212,11 @@ export class DaemonSessionSummarizer {
private readonly inFlight = new Map<string, AbortController>();
// Sessions requested while one was running; get one more pass on completion.
private readonly rerunRequested = new Set<string>();
// 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[],
Expand Down Expand Up @@ -242,6 +252,7 @@ export class DaemonSessionSummarizer {
controller.abort();
}
this.rerunRequested.clear();
this.failedIdleGenerations.clear();
}

/** Drop any pending work for a session that is closing. */
Expand All @@ -253,6 +264,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. */
Expand Down Expand Up @@ -306,6 +318,20 @@ export class DaemonSessionSummarizer {
if (contentUnchanged && !isWorking && !owesIdleVerdict && !owesSummary) {
return;
}
// A generation that keeps failing on identical idle content will keep
// failing: stop re-attempting until the content changes (count plus last
// timestamp, so a branch/edit back to the same length re-arms) or the
// backoff elapses for externally-caused failures.
const contentKey = `${messageCount}:${messages[messageCount - 1]?.timestamp ?? 0}`;
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;
Expand All @@ -319,6 +345,17 @@ export class DaemonSessionSummarizer {
isWorking,
signal: controller.signal,
});
if (generated) {
this.failedIdleGenerations.delete(id);
} else if (!isWorking && !controller.signal.aborted) {
// The aborted check keeps a forget() during the call from
// repopulating the map for a closed session.
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.
Expand Down Expand Up @@ -356,12 +393,21 @@ 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, never
// mid-stream, never a fabricated fallback, and never a duplicate of the
// latest persisted entry: an idle sweep 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) {
Expand Down
76 changes: 73 additions & 3 deletions packages/coding-agent/test/agent-session-recursion.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2462,7 +2462,7 @@ 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 () => {
it("coalesces the admitted task's tool-loop turns into one flushed spawn-usage attribution", async () => {
const tool = {
name: "echo",
description: "Echo a value",
Expand Down Expand Up @@ -2507,11 +2507,81 @@ describe("AgentSession rlm recursion", () => {
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 {
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;
if (toolResultCount === 2) {
// The pending batch from the first two completions is now older
// than the staleness bound when the third completion lands.
vi.setSystemTime(Date.now() + 61_000);
}
const stream = createAssistantMessageEventStream();
queueMicrotask(() => {
const message =
toolResultCount < 2
? {
...assistantMessage("", usage(toolResultCount + 1, toolResultCount + 1)),
content: [
{
type: "toolCall" as const,
id: `echo-${toolResultCount}`,
name: "echo",
arguments: { value: "ok" },
},
],
stopReason: "toolUse" as const,
}
: assistantMessage("done", usage(4, 4));
stream.push({
type: "done",
reason: toolResultCount < 2 ? "toolUse" : "stop",
message,
});
});
return stream;
},
});
const parentAssistant = assistantMessage("running ipython", usage(0, 0));
root.agent.state.messages.push(parentAssistant);
root.sessionManager.appendMessage(parentAssistant);

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"]);
});
} finally {
vi.useRealTimers();
}
});

it("gets and persists per-chat max-depth changes without transcript messages", async () => {
const root = createSession();
const originalMessages = [...root.messages];
Expand Down
Loading
Loading