From ae1946aa0effeba87184c0404cadc0126b2364f3 Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Sun, 12 Jul 2026 16:08:54 -0400 Subject: [PATCH 1/2] Fix chat recovery and bound Codex usage scans --- .../services/chat/agentChatService.test.ts | 326 +++++++++++++++++- .../main/services/chat/agentChatService.ts | 189 ++++++++-- .../usage/ledgers/localUsageLedgers.ts | 283 ++++++++++++--- .../usage/usageTrackingService.test.ts | 284 +++++++++++++++ .../services/usage/usageTrackingService.ts | 7 + docs/features/chat/transcript-and-turns.md | 37 +- .../onboarding-and-settings/README.md | 8 + .../onboarding-and-settings/usage-tracking.md | 42 +++ 8 files changed, 1092 insertions(+), 84 deletions(-) diff --git a/apps/desktop/src/main/services/chat/agentChatService.test.ts b/apps/desktop/src/main/services/chat/agentChatService.test.ts index 359042a26..61165d385 100644 --- a/apps/desktop/src/main/services/chat/agentChatService.test.ts +++ b/apps/desktop/src/main/services/chat/agentChatService.test.ts @@ -10134,11 +10134,17 @@ describe("createAgentChatService", () => { // process. deriveBackgroundItems / subagentSnapshotsFromEvents read these. const orphanTail: AgentChatEventEnvelope[] = [ { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 1, event: { + type: "user_message", text: "Work interrupted by restart", turnId: "turn-old", + } as any }, + { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 2, event: { + type: "status", turnStatus: "started", turnId: "turn-old", + } as any }, + { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 3, event: { type: "scheduled_work_update", id: "background:bg-restart", kind: "background_task", status: "running", origin: "background_task", title: "npm run serve", summary: "shell", sourceTaskId: "bg-restart", turnId: "turn-old", } as any }, - { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 2, event: { + { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 4, event: { type: "subagent_started", taskId: "sub-restart", agentId: "sub-restart", agentType: "Explore", parentToolUseId: "toolu_sub_r", description: "look", turnId: "turn-old", } as any }, @@ -10176,6 +10182,151 @@ describe("createAgentChatService", () => { && (e.event as any).message.startsWith("Reconciled after restart:")); expect(notices).toHaveLength(1); expect((notices[0]!.event as any).message).toBe("Reconciled after restart: 1 background task stopped"); + + expect(events2.filter((e) => + e.event.type === "status" + && e.event.turnId === "turn-old" + && e.event.turnStatus === "interrupted" + )).toHaveLength(1); + expect(events2.filter((e) => + e.event.type === "done" + && e.event.turnId === "turn-old" + && e.event.status === "interrupted" + )).toHaveLength(1); + const turnScopedEvents = events2.filter((event) => event.event.turnId); + expect(turnScopedEvents.at(-1)?.event).toMatchObject({ + type: "done", + turnId: "turn-old", + status: "interrupted", + }); + + await service2.interrupt({ sessionId: session.id }); + expect(events2.filter((e) => + e.event.type === "status" + && e.event.turnId === "turn-old" + && e.event.turnStatus === "interrupted" + )).toHaveLength(1); + expect(events2.filter((e) => + e.event.type === "done" + && e.event.turnId === "turn-old" + && e.event.status === "interrupted" + )).toHaveLength(1); + }); + + it("reconciles before an SDK id exists and emits only a missing terminal half", async () => { + const sessionId = "claude-restart-before-sdk-init"; + const events: AgentChatEventEnvelope[] = []; + const { service, sessionService } = createService({ + onEvent: (event: AgentChatEventEnvelope) => events.push(event), + }); + sessionService.create({ + sessionId, + laneId: "lane-1", + toolType: "claude-chat", + title: "Restart before SDK init", + startedAt: "2026-07-12T12:00:00.000Z", + }); + writePersistedChatState(sessionId, { + version: 2, + sessionId, + laneId: "lane-1", + provider: "claude", + model: "sonnet", + updatedAt: "2026-07-12T12:00:00.000Z", + }); + const chatTranscriptDir = path.join(tmpRoot, ".ade", "transcripts", "chat"); + fs.mkdirSync(chatTranscriptDir, { recursive: true }); + fs.writeFileSync(path.join(chatTranscriptDir, `${sessionId}.jsonl`), "{}\n", "utf8"); + vi.mocked(parseAgentChatTranscript).mockReturnValue([ + { + sessionId, + timestamp: "2026-07-12T12:00:00.000Z", + sequence: 1, + event: { type: "status", turnStatus: "started", turnId: "partial-terminal-turn" }, + }, + { + sessionId, + timestamp: "2026-07-12T12:00:01.000Z", + sequence: 2, + event: { type: "status", turnStatus: "interrupted", turnId: "partial-terminal-turn" }, + }, + ] as AgentChatEventEnvelope[]); + + await service.resumeSession({ sessionId }); + + expect(events.filter((event) => + event.event.type === "status" + && event.event.turnId === "partial-terminal-turn" + )).toHaveLength(0); + expect(events.filter((event) => + event.event.type === "done" + && event.event.turnId === "partial-terminal-turn" + && event.event.status === "interrupted" + )).toHaveLength(1); + expect(events.at(-1)?.event).toMatchObject({ + type: "done", + turnId: "partial-terminal-turn", + status: "interrupted", + }); + }); + + it("does not reconcile an ancient incomplete turn when the latest parent turn completed", async () => { + const sessionId = "claude-restart-latest-complete"; + const events: AgentChatEventEnvelope[] = []; + const { service, sessionService } = createService({ + onEvent: (event: AgentChatEventEnvelope) => events.push(event), + }); + sessionService.create({ + sessionId, + laneId: "lane-1", + toolType: "claude-chat", + title: "Latest turn complete", + startedAt: "2026-07-12T12:00:00.000Z", + }); + writePersistedChatState(sessionId, { + version: 2, + sessionId, + laneId: "lane-1", + provider: "claude", + model: "sonnet", + updatedAt: "2026-07-12T12:00:00.000Z", + }); + const chatTranscriptDir = path.join(tmpRoot, ".ade", "transcripts", "chat"); + fs.mkdirSync(chatTranscriptDir, { recursive: true }); + fs.writeFileSync(path.join(chatTranscriptDir, `${sessionId}.jsonl`), "{}\n", "utf8"); + vi.mocked(parseAgentChatTranscript).mockReturnValue([ + { sessionId, timestamp: "2026-07-12T12:00:00.000Z", sequence: 1, event: { + type: "user_message", text: "Old turn", turnId: "old-open-turn", + } }, + { sessionId, timestamp: "2026-07-12T12:00:00.100Z", sequence: 2, event: { + type: "status", turnStatus: "started", turnId: "old-open-turn", + } }, + { sessionId, timestamp: "2026-07-12T12:01:00.000Z", sequence: 3, event: { + type: "user_message", text: "Latest turn", turnId: "latest-complete-turn", + } }, + { sessionId, timestamp: "2026-07-12T12:01:00.100Z", sequence: 4, event: { + type: "status", turnStatus: "started", turnId: "latest-complete-turn", + } }, + { sessionId, timestamp: "2026-07-12T12:01:01.000Z", sequence: 5, event: { + type: "status", turnStatus: "completed", turnId: "latest-complete-turn", + } }, + { sessionId, timestamp: "2026-07-12T12:01:01.100Z", sequence: 6, event: { + type: "done", status: "completed", turnId: "latest-complete-turn", + } }, + ] as AgentChatEventEnvelope[]); + + await service.resumeSession({ sessionId }); + + expect(events.filter((event) => + event.event.type === "status" && event.event.turnId === "old-open-turn" + )).toHaveLength(0); + expect(events.filter((event) => + event.event.type === "done" && event.event.turnId === "old-open-turn" + )).toHaveLength(0); + expect(events.some((event) => + event.event.type === "system_notice" + && event.event.message.startsWith("Reconciled after restart:") + )).toBe(false); }); }); @@ -25167,6 +25318,128 @@ describe("createAgentChatService", () => { )).toHaveLength(1); }); + it("bounds hung Claude interrupt and subagent stop calls below the desktop action timeout", async () => { + try { + const events: AgentChatEventEnvelope[] = []; + let streamCall = 0; + let warmupComplete = false; + let releaseTurn!: () => void; + const turnGate = new Promise((resolve) => { releaseTurn = resolve; }); + const neverSettles = new Promise(() => {}); + const stopTask = vi.fn(() => neverSettles); + const queryInterrupt = vi.fn(() => neverSettles); + const stream = vi.fn(() => (async function* () { + streamCall += 1; + if (streamCall === 1) { + yield { type: "system", subtype: "init", session_id: "sdk-bounded-interrupt", slash_commands: [] }; + warmupComplete = true; + yield { type: "result", usage: { input_tokens: 1, output_tokens: 1 } }; + return; + } + yield { type: "system", subtype: "task_started", task_id: "hung-task-1", description: "Hung task one" }; + yield { type: "system", subtype: "task_started", task_id: "hung-task-2", description: "Hung task two" }; + await turnGate; + })()); + vi.mocked(claudeSdkCreateSessionCompat).mockReturnValue({ + send: vi.fn().mockResolvedValue(undefined), + stream, + close: vi.fn(), + sessionId: "sdk-bounded-interrupt", + setPermissionMode: vi.fn().mockResolvedValue(undefined), + stopTask, + interrupt: queryInterrupt, + } as any); + + const { service } = createService({ + onEvent: (event: AgentChatEventEnvelope) => events.push(event), + }); + const session = await service.createSession({ laneId: "lane-1", provider: "claude", model: "sonnet" }); + await vi.waitFor(() => { expect(warmupComplete).toBe(true); }); + + const sendPromise = service.sendMessage({ sessionId: session.id, text: "Start hung subagents" }); + await waitForEvent(events, (event): event is AgentChatEventEnvelope => + event.event.type === "subagent_started" && event.event.taskId === "hung-task-2"); + + vi.useFakeTimers(); + let interruptSettled = false; + const interruptPromise = service.interrupt({ sessionId: session.id }).then(() => { + interruptSettled = true; + }); + await vi.advanceTimersByTimeAsync(1_999); + expect(interruptSettled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + expect(stopTask).toHaveBeenCalledTimes(2); + expect(queryInterrupt).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(2_499); + expect(interruptSettled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await expect(interruptPromise).resolves.toBeUndefined(); + + expect(events.filter((event) => + event.event.type === "status" && event.event.turnStatus === "interrupted" + )).toHaveLength(1); + expect(events.filter((event) => + event.event.type === "done" && event.event.status === "interrupted" + )).toHaveLength(1); + expect(events.filter((event) => + event.event.type === "subagent_result" && event.event.status === "stopped" + )).toHaveLength(2); + + releaseTurn(); + await expect(sendPromise).resolves.toBeUndefined(); + } finally { + vi.useRealTimers(); + } + }); + + it("terminalizes an orphaned Claude transcript turn when Stop sees an idle runtime", async () => { + const events: AgentChatEventEnvelope[] = []; + let warmupComplete = false; + const stream = vi.fn(() => (async function* () { + yield { type: "system", subtype: "init", session_id: "sdk-idle-stop", slash_commands: [] }; + warmupComplete = true; + yield { type: "result", usage: { input_tokens: 1, output_tokens: 1 } }; + })()); + vi.mocked(claudeSdkCreateSessionCompat).mockReturnValue({ + send: vi.fn().mockResolvedValue(undefined), + stream, + close: vi.fn(), + sessionId: "sdk-idle-stop", + setPermissionMode: vi.fn().mockResolvedValue(undefined), + } as any); + + const { service } = createService({ + onEvent: (event: AgentChatEventEnvelope) => events.push(event), + }); + const session = await service.createSession({ laneId: "lane-1", provider: "claude", model: "sonnet" }); + await vi.waitFor(() => { expect(warmupComplete).toBe(true); }); + + vi.mocked(parseAgentChatTranscript).mockReturnValue([{ + sessionId: session.id, + timestamp: new Date().toISOString(), + sequence: 1, + event: { type: "user_message", text: "Crash before started status", turnId: "orphaned-after-restart" }, + } as AgentChatEventEnvelope]); + + await service.interrupt({ sessionId: session.id }); + + expect(events.filter((event) => + event.event.type === "status" + && event.event.turnId === "orphaned-after-restart" + && event.event.turnStatus === "interrupted" + )).toHaveLength(1); + expect(events.filter((event) => + event.event.type === "done" + && event.event.turnId === "orphaned-after-restart" + && event.event.status === "interrupted" + )).toHaveLength(1); + expect(events.at(-1)?.event).toMatchObject({ + type: "done", + turnId: "orphaned-after-restart", + status: "interrupted", + }); + }); + it("resumes through a fresh SDK session after interrupt so stale stream text is not replayed", async () => { const events: AgentChatEventEnvelope[] = []; let primaryStreamCall = 0; @@ -25961,6 +26234,57 @@ describe("createAgentChatService", () => { await activeTurn; }); + it("returns an idle Claude steer after dispatch acceptance while the provider turn keeps running", async () => { + const events: AgentChatEventEnvelope[] = []; + let streamCall = 0; + let warmupComplete = false; + let finishTurn!: () => void; + const turnGate = new Promise((resolve) => { finishTurn = resolve; }); + const send = vi.fn().mockResolvedValue(undefined); + const stream = vi.fn(() => (async function* () { + streamCall += 1; + if (streamCall === 1) { + yield { type: "system", subtype: "init", session_id: "sdk-idle-steer", slash_commands: [] }; + warmupComplete = true; + yield { type: "result", usage: { input_tokens: 1, output_tokens: 1 } }; + return; + } + yield { + type: "assistant", + message: { content: [{ type: "text", text: "Working after acceptance" }], usage: { input_tokens: 1, output_tokens: 1 } }, + }; + await turnGate; + yield { type: "result", usage: { input_tokens: 1, output_tokens: 1 } }; + })()); + vi.mocked(claudeSdkCreateSessionCompat).mockReturnValue({ + send, + stream, + close: vi.fn(), + sessionId: "sdk-idle-steer", + setPermissionMode: vi.fn().mockResolvedValue(undefined), + } as any); + + const { service } = createService({ onEvent: (event: AgentChatEventEnvelope) => events.push(event) }); + const session = await service.createSession({ laneId: "lane-1", provider: "claude", model: "sonnet" }); + await vi.waitFor(() => { expect(warmupComplete).toBe(true); }); + + await expect(service.steer({ + sessionId: session.id, + text: "Treat this stale steer as a normal turn", + dispatchMode: "inline", + })).resolves.toMatchObject({ queued: false, steerId: expect.any(String) }); + + expect(events.some((event) => + event.event.type === "user_message" + && event.event.text === "Treat this stale steer as a normal turn" + )).toBe(true); + expect(events.some((event) => event.event.type === "done")).toBe(false); + + finishTurn(); + await waitForEvent(events, (event): event is AgentChatEventEnvelope => + event.event.type === "done" && event.event.status === "completed"); + }); + it("dispatchSteer mode:'interrupt' uses Claude priority-now without tearing down the query", async () => { const events: AgentChatEventEnvelope[] = []; const send = vi.fn().mockResolvedValue(undefined); diff --git a/apps/desktop/src/main/services/chat/agentChatService.ts b/apps/desktop/src/main/services/chat/agentChatService.ts index 9d3c3480b..ad2ccc544 100644 --- a/apps/desktop/src/main/services/chat/agentChatService.ts +++ b/apps/desktop/src/main/services/chat/agentChatService.ts @@ -119,6 +119,7 @@ import type { createProcessService } from "../processes/processService"; import { runGit } from "../git/git"; import { CLAUDE_RUNTIME_AUTH_ERROR, isClaudeRuntimeAuthError } from "../ai/claudeRuntimeProbe"; import { resolveCodexExecutable } from "../ai/codexExecutable"; +import { withTimeout } from "../ai/utils"; import { fileSizeOrZero, hasNullByte, @@ -2196,6 +2197,7 @@ const MAX_INJECTED_PROJECT_COMMANDS = 20; const CURSOR_SDK_AGENT_PROTOCOL_VERSION = 2; const CLAUDE_WARMUP_WAIT_TIMEOUT_MS = 20_000; const CLAUDE_STOP_TASK_TIMEOUT_MS = 2_000; +const CLAUDE_INTERRUPT_REQUEST_TIMEOUT_MS = 2_500; const DEFAULT_CODEX_DESCRIPTOR = getDefaultModelDescriptor("codex"); const DEFAULT_CLAUDE_DESCRIPTOR = getDefaultModelDescriptor("claude"); @@ -7427,6 +7429,16 @@ export function createAgentChatService(args: { }; }; + const awaitClaudeControlCall = async ( + label: string, + timeoutMs: number, + operation: () => T | PromiseLike, + ): Promise => withTimeout( + Promise.resolve().then(operation), + timeoutMs, + `${label} timed out after ${timeoutMs}ms`, + ); + const readTranscriptConversationEntries = (managed: ManagedChatSession): string[] => { try { return readTranscriptEnvelopes(managed) @@ -8157,6 +8169,52 @@ export function createAgentChatService(args: { return turnActive; }; + type UnsettledParentTurn = { + turnId: string; + terminalStatus: "completed" | "interrupted" | "failed" | null; + doneStatus: "completed" | "interrupted" | "failed" | null; + }; + + const findLatestUnsettledParentTurn = ( + entries: AgentChatEventEnvelope[], + ): UnsettledParentTurn | null => { + let latest: UnsettledParentTurn | null = null; + for (const entry of entries) { + if (isCodexSubagentTranscriptEnvelope(entry)) continue; + const event = entry.event; + + if (event.type === "user_message" && !event.steerId) { + const turnId = event.turnId?.trim(); + if (turnId) { + latest = { turnId, terminalStatus: null, doneStatus: null }; + } + continue; + } + + if (event.type === "status" && event.turnStatus === "started") { + const turnId = event.turnId?.trim(); + if (!turnId) continue; + if (latest?.turnId !== turnId) { + latest = { turnId, terminalStatus: null, doneStatus: null }; + } + continue; + } + + if (!latest || (event.type !== "status" && event.type !== "done")) continue; + if (event.turnId?.trim() !== latest.turnId) continue; + if (event.type === "status") { + if (event.turnStatus === "started") continue; + latest.terminalStatus = event.turnStatus; + } else { + latest.doneStatus = event.status; + } + } + + return latest && (latest.terminalStatus == null || latest.doneStatus == null) + ? latest + : null; + }; + const normalizeEventStatus = (status: string | undefined): string => { if (status === "failed") return "failed"; if (status === "completed") return "completed"; @@ -19344,14 +19402,14 @@ export function createAgentChatService(args: { if (activeSubagents.length === 0) return; const control = getClaudeQueryControl(runtime.query); - for (const subagent of activeSubagents) { - if (!runtime.activeSubagents.has(subagent.taskId)) continue; + await Promise.all(activeSubagents.map(async (subagent) => { + if (!runtime.activeSubagents.has(subagent.taskId)) return; // Ambient (skip_transcript) and non-agent task runs never surfaced as // subagent rows, so they must not emit a stopped subagent_result here — // just drop the tracking entry. if (subagent.skipTranscript || subagent.nonAgentTaskRun) { runtime.activeSubagents.delete(subagent.taskId); - continue; + return; } // A background shell entry with no real subagent agentType must not emit a // subagent_result — closeOpenClaudeBackgroundTasks already settled it. @@ -19362,33 +19420,22 @@ export function createAgentChatService(args: { description: subagent.description, })) { runtime.activeSubagents.delete(subagent.taskId); - continue; + return; } runtime.activeSubagents.delete(subagent.taskId); if (typeof control.stopTask === "function") { - let timeoutHandle: ReturnType | null = null; try { - const stopTaskPromise = Promise.resolve(control.stopTask(subagent.taskId)); - stopTaskPromise.catch(() => { - // The awaited race below handles timely rejections. This catch only - // prevents an unhandled rejection if the SDK rejects after our timeout. - }); - await Promise.race([ - stopTaskPromise, - new Promise((_, reject) => { - timeoutHandle = setTimeout(() => { - reject(new Error(`Timed out stopping Claude task after ${CLAUDE_STOP_TASK_TIMEOUT_MS}ms`)); - }, CLAUDE_STOP_TASK_TIMEOUT_MS); - }), - ]); + await awaitClaudeControlCall( + `Stopping Claude task '${subagent.taskId}'`, + CLAUDE_STOP_TASK_TIMEOUT_MS, + () => control.stopTask!(subagent.taskId), + ); } catch (error) { logger.warn("agent_chat.claude_stop_task_failed", { sessionId: managed.session.id, taskId: subagent.taskId, error: error instanceof Error ? error.message : String(error), }); - } finally { - if (timeoutHandle) clearTimeout(timeoutHandle); } } emitClaudeSubagentResult(managed, runtime, { @@ -19401,7 +19448,7 @@ export function createAgentChatService(args: { finalSummary: summary, turnId, }); - } + })); }; type CodexCollabAgentState = { @@ -23590,6 +23637,49 @@ export function createAgentChatService(args: { // process — background_task rows stuck "running", subagent snapshots still // open. Nothing will ever settle them, so sweep them to a terminal state and // announce it once. Pure event replay; safe to run before the first turn. + const findUnsettledClaudeParentTurn = ( + managed: ManagedChatSession, + transcriptEvents = readFullTranscriptEnvelopesForSessionId(managed.session.id), + ): UnsettledParentTurn | null => { + const recentEvents = eventHistoryBySession.get(managed.session.id) ?? []; + return findLatestUnsettledParentTurn(mergeEnvelopeStreams(transcriptEvents, recentEvents)); + }; + + const terminalizeUnsettledClaudeParentTurn = ( + managed: ManagedChatSession, + reason: "restart" | "idle_interrupt", + candidate?: UnsettledParentTurn | null, + ): string | null => { + const unsettled = candidate === undefined ? findUnsettledClaudeParentTurn(managed) : candidate; + if (!unsettled) return null; + + const status = unsettled.terminalStatus ?? unsettled.doneStatus ?? "interrupted"; + if (!unsettled.terminalStatus) { + emitChatEvent(managed, { + type: "status", + turnStatus: status, + turnId: unsettled.turnId, + }); + } + if (!unsettled.doneStatus) { + emitChatEvent(managed, { + type: "done", + turnId: unsettled.turnId, + status, + ...resolveClaudeTurnModelPayload(managed.session, []), + }); + } + markSessionIdleWithFreshCache(managed); + persistChatState(managed); + logger.info("agent_chat.claude_orphan_turn_terminalized", { + sessionId: managed.session.id, + turnId: unsettled.turnId, + status, + reason, + }); + return unsettled.turnId; + }; + const reconcileClaudeSessionAfterRestart = ( managed: ManagedChatSession, runtime: ClaudeRuntime, @@ -23598,6 +23688,8 @@ export function createAgentChatService(args: { const envelopes = readFullTranscriptEnvelopesForSessionId(managed.session.id); if (envelopes.length === 0) return; + const orphanParentTurn = findUnsettledClaudeParentTurn(managed, envelopes); + const orphanBackground = deriveBackgroundItems(envelopes).filter( (snapshot) => snapshot.status === "scheduled" || snapshot.status === "running", ); @@ -23607,7 +23699,7 @@ export function createAgentChatService(args: { && snapshot.background !== true, ); - if (orphanBackground.length === 0 && orphanSubagents.length === 0) return; + if (orphanBackground.length === 0 && orphanSubagents.length === 0 && !orphanParentTurn) return; const restartTurnId = `claude-restart-reconcile-${randomUUID()}`; @@ -23650,8 +23742,13 @@ export function createAgentChatService(args: { }); } + // Keep the parent terminal pair last. Renderer turn state is derived in + // event order, so no later reconciliation row may revive the stopped turn. + const orphanTurnId = terminalizeUnsettledClaudeParentTurn(managed, "restart", orphanParentTurn); + logger.info("agent_chat.claude_restart_reconciled", { sessionId: managed.session.id, + orphanTurnId, backgroundTasksStopped: orphanBackground.length, subagentsStopped: orphanSubagents.length, }); @@ -23730,13 +23827,11 @@ export function createAgentChatService(args: { managed.runtime = runtime; managed.runtimeInvalidated = false; - // This runtime re-binds a persisted SDK session (host restart / attach) when - // an sdkSessionId was recovered from persisted state. In that case any - // non-terminal background/subagent rows in the transcript are orphans from - // the previous process — reconcile them to a terminal state once. - if (sdkSessionId) { - reconcileClaudeSessionAfterRestart(managed, runtime); - } + // A newly created runtime may be rebinding after a host restart even when + // the prior process crashed before it persisted an SDK session id. Sweep + // any non-terminal parent/background/subagent transcript rows once; a brand + // new chat has no rows, so this is a no-op there. + reconcileClaudeSessionAfterRestart(managed, runtime); return runtime; }; @@ -30546,7 +30641,17 @@ export function createAgentChatService(args: { ? { steerId, queued: true } : { steerId, queued: false, reason: "queue_full" }; } - await executePreparedSendMessage(preparedSteer); + await sendMessage({ + sessionId, + text: trimmed, + displayText: displayText ?? trimmed, + attachments, + contextAttachments, + metadata, + reasoningEffort, + executionMode, + interactionMode, + }, { awaitDispatch: true }); return { steerId, queued: false }; } await executePreparedSendMessage(preparedSteer); @@ -31048,6 +31153,9 @@ export function createAgentChatService(args: { const runtime = ensureClaudeSessionRuntime(managed); // Idempotency guard: skip if already interrupted (e.g. rapid cancel clicks) if (runtime.interrupted) return; + if (!runtime.busy && !runtime.activeTurnId) { + terminalizeUnsettledClaudeParentTurn(managed, "idle_interrupt"); + } logger.info("agent_chat.turn_interrupt_requested", { sessionId, provider: "claude", @@ -31060,7 +31168,11 @@ export function createAgentChatService(args: { if (!claudeControl.interrupt) { throw new Error("Claude interrupt is unavailable; the replacement was not sent."); } - await claudeControl.interrupt(); + await awaitClaudeControlCall( + "Claude interrupt", + CLAUDE_INTERRUPT_REQUEST_TIMEOUT_MS, + () => claudeControl.interrupt!(), + ); } // Set interrupted before touching the runtime so the streaming loop can // break cleanly while the underlying SDK stream is aborted below. @@ -31083,7 +31195,20 @@ export function createAgentChatService(args: { runtime.queryGeneration += 1; runtime.queryStartPromise = null; if (!internalOptions.requireClaudeProviderInterrupt) { - try { await claudeControl.interrupt?.(); } catch { /* ignore */ } + try { + if (claudeControl.interrupt) { + await awaitClaudeControlCall( + "Claude interrupt", + CLAUDE_INTERRUPT_REQUEST_TIMEOUT_MS, + () => claudeControl.interrupt!(), + ); + } + } catch (error) { + logger.warn("agent_chat.claude_interrupt_failed", { + sessionId, + error: error instanceof Error ? error.message : String(error), + }); + } } try { runtime.query?.close(); } catch { /* ignore */ } // close() only ends stream iteration — it does not guarantee the SDK diff --git a/apps/desktop/src/main/services/usage/ledgers/localUsageLedgers.ts b/apps/desktop/src/main/services/usage/ledgers/localUsageLedgers.ts index cae9a9505..36869d749 100644 --- a/apps/desktop/src/main/services/usage/ledgers/localUsageLedgers.ts +++ b/apps/desktop/src/main/services/usage/ledgers/localUsageLedgers.ts @@ -3,15 +3,19 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { createRequire } from "node:module"; -import { createInterface } from "node:readline"; import type { SqlValue } from "../../state/kvDb"; import { isRecord, safeJsonParse } from "../../shared/utils"; const LOCAL_COST_SCAN_MAX_FILES = 5_000; const LOCAL_COST_SCAN_MAX_FILE_BYTES = 768 * 1024 * 1024; +const LOCAL_JSONL_MAX_LINE_BYTES = 16 * 1024 * 1024; const LOCAL_COST_SCAN_MAX_ENTRIES = 1_000_000; const LOCAL_COST_SCAN_ALL_DAYS = 3650; +const CODEX_COST_SCAN_MAX_FILE_BYTES = 256 * 1024 * 1024; +const CODEX_COST_SCAN_MAX_TOTAL_BYTES = 2 * 1024 * 1024 * 1024; +const CODEX_COST_SCAN_MAX_ENTRIES = 250_000; const LOCAL_SQLITE_SCAN_MAX_ROWS = 250_000; +const LOCAL_SQLITE_LOOKUP_BATCH_SIZE = 500; const LOCAL_CURSOR_SQLITE_RECENT_ROWS = 250_000; const CURSOR_CHARS_PER_TOKEN = 4; @@ -28,6 +32,12 @@ type RecentFileCandidate = { path: string; mtimeMs: number }; const requireForUsageSqlite = createRequire(path.join(process.cwd(), "ade-runtime.cjs")); let usageSqliteConstructor: UsageSqliteConstructor | null | undefined; +let codexLogScanInFlight: Promise | null = null; + +type CodexLogScanOptions = { + maxJsonlLineBytes?: number; + maxEntries?: number; +}; function toFiniteNumber(value: unknown): number { const numberValue = Number(value ?? 0); @@ -46,6 +56,8 @@ function normalizeUsageLabel(value: unknown, fallback: string): string { export interface TokenEntry { messageId: string; model: string; + /** Aggregate usage that belongs only in the all-time headline, without fabricated day attribution. */ + lifetimeOnly?: boolean; originator?: string; projectPath?: string; projectKey?: string; @@ -362,20 +374,58 @@ export async function scanClaudeLogs(projectDirsOverride?: string[]): Promise { +export function scanCodexLogs( + options: CodexLogScanOptions = {}, +): Promise { + // Production callers share one machine-history pass. Without this guard, + // two project runtimes opening Stats together can each retain a full Codex + // entry set and multiply both CPU and peak memory. Test-only custom limits + // bypass the shared promise so fixtures remain isolated. + if (options.maxJsonlLineBytes !== undefined || options.maxEntries !== undefined) { + return scanCodexLogsOnce(options); + } + if (codexLogScanInFlight) return codexLogScanInFlight; + + let current!: Promise; + current = scanCodexLogsOnce(options).finally(() => { + if (codexLogScanInFlight === current) codexLogScanInFlight = null; + }); + codexLogScanInFlight = current; + return current; +} + +async function scanCodexLogsOnce( + options: CodexLogScanOptions, +): Promise { const entries: TokenEntry[] = []; const seen = new Set(); const seenForkReplayKeys = new Set(); + const maxEntries = options.maxEntries !== undefined && Number.isFinite(options.maxEntries) + ? Math.max(1, Math.floor(options.maxEntries)) + : CODEX_COST_SCAN_MAX_ENTRIES; const codexHome = process.env.CODEX_HOME || path.join(os.homedir(), ".codex"); + // Leave one slot for an authoritative SQLite lifetime remainder. JSONL is + // the detailed view, but the Codex thread index owns the lifetime headline. + const maxJsonlEntries = newestCodexStateDatabase(codexHome) + ? Math.max(0, maxEntries - 1) + : maxEntries; + if (maxJsonlEntries === 0) { + return reconcileCodexStateTotals(entries, codexHome, maxEntries); + } const sessionsDir = path.join(codexHome, "sessions"); const archivedSessionsDir = path.join(codexHome, "archived_sessions"); const sessionRoots = [sessionsDir, archivedSessionsDir].filter((root) => fs.existsSync(root)); + const maxBytesPerRoot = Math.max( + 1, + Math.floor(CODEX_COST_SCAN_MAX_TOTAL_BYTES / Math.max(1, sessionRoots.length)), + ); const jsonlFiles = (await Promise.all(sessionRoots.map((root) => findJsonlFiles( root, LOCAL_COST_SCAN_ALL_DAYS, { maxFiles: LOCAL_COST_SCAN_MAX_FILES, - maxFileBytes: LOCAL_COST_SCAN_MAX_FILE_BYTES, + maxFileBytes: CODEX_COST_SCAN_MAX_FILE_BYTES, + maxTotalBytes: maxBytesPerRoot, }, )))).flat(); @@ -388,7 +438,7 @@ export async function scanCodexLogs(): Promise { let forkedFromId = ""; let previousTotals: { input: number; cached: number; output: number; reasoning: number; total: number } | null = null; - for await (const line of readJsonlLines(filePath)) { + for await (const line of readJsonlLines(filePath, options.maxJsonlLineBytes)) { const trimmed = line.trim(); if (!trimmed) continue; const mayContainUsage = @@ -508,8 +558,8 @@ export async function scanCodexLogs(): Promise { cacheWriteTokens: 0, timestamp, }); - if (entries.length >= LOCAL_COST_SCAN_MAX_ENTRIES) { - return reconcileCodexStateTotals(entries, codexHome); + if (entries.length >= maxJsonlEntries) { + return reconcileCodexStateTotals(entries, codexHome, maxEntries); } continue; } @@ -561,8 +611,8 @@ export async function scanCodexLogs(): Promise { timestamp: typeof record.timestamp === "number" ? record.timestamp : typeof record.timestamp === "string" ? new Date(record.timestamp).getTime() : Date.now(), }); - if (entries.length >= LOCAL_COST_SCAN_MAX_ENTRIES) { - return reconcileCodexStateTotals(entries, codexHome); + if (entries.length >= maxJsonlEntries) { + return reconcileCodexStateTotals(entries, codexHome, maxEntries); } } } catch { @@ -570,7 +620,7 @@ export async function scanCodexLogs(): Promise { } } - return reconcileCodexStateTotals(entries, codexHome); + return reconcileCodexStateTotals(entries, codexHome, maxEntries); } type CodexStateThreadRow = { @@ -584,6 +634,17 @@ type CodexStateThreadRow = { updated_at?: unknown; }; +type CodexStateAggregateRow = { + total_tokens?: unknown; +}; + +function tokenEntryTotal(entry: TokenEntry): number { + return toNonNegativeInt(entry.inputTokens) + + toNonNegativeInt(entry.outputTokens) + + toNonNegativeInt(entry.cachedTokens) + + toNonNegativeInt(entry.cacheWriteTokens); +} + function newestCodexStateDatabase(codexHome: string): string | null { try { const candidates = fs.readdirSync(codexHome) @@ -602,51 +663,104 @@ function newestCodexStateDatabase(codexHome: string): string | null { /** * Codex Desktop's thread index owns the lifetime total shown in its profile. * JSONL reconstruction keeps the detailed model/day/cost split, but can miss - * cumulative context retained by the thread index. Add only the positive - * per-thread remainder, distributed across the observed token mix and priced - * at zero so the exact lifetime count does not fabricate cost. + * cumulative context retained by the thread index. Add bounded per-thread + * remainders where possible, then preserve the authoritative lifetime headline + * with one unattributed all-time remainder. Reconciliation is priced at zero so + * exact token totals never fabricate cost or recent-day activity. */ -function reconcileCodexStateTotals(entries: TokenEntry[], codexHome: string): TokenEntry[] { +function reconcileCodexStateTotals( + entries: TokenEntry[], + codexHome: string, + maxEntries: number, +): TokenEntry[] { + const remainingCapacity = Math.max(0, maxEntries - entries.length); const dbPath = newestCodexStateDatabase(codexHome); - if (!dbPath) return entries; + if (!dbPath || remainingCapacity === 0) return entries; const db = openReadonlyUsageDatabase(dbPath); if (!db) return entries; + const observedByThread = new Map(); + for (const entry of entries) { + const separator = entry.messageId.indexOf(":"); + if (separator <= 0) continue; + const threadId = entry.messageId.slice(0, separator); + const observed = observedByThread.get(threadId) ?? { input: 0, output: 0, cached: 0, cacheWrite: 0 }; + observed.input += toNonNegativeInt(entry.inputTokens); + observed.output += toNonNegativeInt(entry.outputTokens); + observed.cached += toNonNegativeInt(entry.cachedTokens); + observed.cacheWrite += toNonNegativeInt(entry.cacheWriteTokens); + observedByThread.set(threadId, observed); + } + let rows: CodexStateThreadRow[] = []; + let stateLifetimeTotal = 0; + let observedStateOverlap = 0; try { - rows = usageSqliteAll(db, ` - select id, tokens_used, model, cwd, source, thread_source, created_at, updated_at + const aggregate = usageSqliteAll(db, ` + select coalesce(sum(tokens_used), 0) as total_tokens from threads where tokens_used > 0 - `); + `)[0]; + stateLifetimeTotal = toNonNegativeInt(aggregate?.total_tokens); + + // JSONL can contain archived threads no longer present in the state index. + // Look up only the bounded set of observed IDs, in batches, so the final + // summary represents JSONL UNION state rather than max(JSONL, state). + const observedThreadIds = Array.from(observedByThread.keys()); + for (let offset = 0; offset < observedThreadIds.length; offset += LOCAL_SQLITE_LOOKUP_BATCH_SIZE) { + const batch = observedThreadIds.slice(offset, offset + LOCAL_SQLITE_LOOKUP_BATCH_SIZE); + const placeholders = batch.map(() => "?").join(", "); + const stateRows = usageSqliteAll(db, ` + select id, tokens_used + from threads + where id in (${placeholders}) + `, batch); + for (const stateRow of stateRows) { + const threadId = typeof stateRow.id === "string" ? stateRow.id : ""; + const observed = observedByThread.get(threadId); + if (!observed) continue; + const observedTotal = observed.input + observed.output + observed.cached + observed.cacheWrite; + observedStateOverlap += Math.min(toNonNegativeInt(stateRow.tokens_used), observedTotal); + } + } } catch { - try { + stateLifetimeTotal = 0; + observedStateOverlap = 0; + } + + const observedLifetimeTotal = entries.reduce((total, entry) => total + tokenEntryTotal(entry), 0); + const targetLifetimeTotal = observedLifetimeTotal + + Math.max(0, stateLifetimeTotal - observedStateOverlap); + const reserveLifetimeSummary = targetLifetimeTotal > observedLifetimeTotal ? 1 : 0; + const detailCapacity = Math.max(0, remainingCapacity - reserveLifetimeSummary); + const rowLimit = Math.min(LOCAL_SQLITE_SCAN_MAX_ROWS, detailCapacity); + try { + if (rowLimit > 0) { rows = usageSqliteAll(db, ` - select id, tokens_used, created_at, updated_at + select id, tokens_used, model, cwd, source, thread_source, created_at, updated_at from threads where tokens_used > 0 - `); + order by coalesce(updated_at, created_at, 0) desc + limit ? + `, [rowLimit]); + } + } catch { + try { + if (rowLimit > 0) { + rows = usageSqliteAll(db, ` + select id, tokens_used, created_at, updated_at + from threads + where tokens_used > 0 + order by coalesce(updated_at, created_at, 0) desc + limit ? + `, [rowLimit]); + } } catch { rows = []; } } finally { db.close(); } - if (rows.length === 0) return entries; - - const observedByThread = new Map(); - for (const entry of entries) { - const separator = entry.messageId.indexOf(":"); - if (separator <= 0) continue; - const threadId = entry.messageId.slice(0, separator); - const observed = observedByThread.get(threadId) ?? { input: 0, output: 0, cached: 0, cacheWrite: 0 }; - observed.input += toNonNegativeInt(entry.inputTokens); - observed.output += toNonNegativeInt(entry.outputTokens); - observed.cached += toNonNegativeInt(entry.cachedTokens); - observed.cacheWrite += toNonNegativeInt(entry.cacheWriteTokens); - observedByThread.set(threadId, observed); - } - for (const row of rows) { const threadId = typeof row.id === "string" ? row.id.trim() : ""; const stateTotal = toNonNegativeInt(row.tokens_used); @@ -683,6 +797,28 @@ function reconcileCodexStateTotals(entries: TokenEntry[], codexHome: string): To costOverrideUsd: 0, timestamp: timestampMsFromUnixish(row.updated_at ?? row.created_at), }); + if (entries.length >= maxEntries - reserveLifetimeSummary) break; + } + + const reconciledLifetimeTotal = entries.reduce((total, entry) => total + tokenEntryTotal(entry), 0); + const lifetimeRemainder = Math.max(0, targetLifetimeTotal - reconciledLifetimeTotal); + if (lifetimeRemainder > 0 && entries.length < maxEntries) { + entries.push({ + messageId: "codex-state:lifetime-total-remainder", + model: "codex", + lifetimeOnly: true, + originator: "Codex Desktop", + estimation: "distribution", + inputTokens: lifetimeRemainder, + billableInputTokens: 0, + outputTokens: 0, + billableOutputTokens: 0, + cachedTokens: 0, + billableCachedTokens: 0, + cacheWriteTokens: 0, + costOverrideUsd: 0, + timestamp: 0, + }); } return entries; @@ -1499,12 +1635,15 @@ export async function findRecentFiles( dir: string, maxAgeDays: number, suffixes: string[], - options: { maxFiles?: number; maxFileBytes?: number } = {}, + options: { maxFiles?: number; maxFileBytes?: number; maxTotalBytes?: number } = {}, ): Promise { const cutoff = Date.now() - maxAgeDays * 24 * 60 * 60 * 1000; const maxFiles = Math.max(1, Math.floor(options.maxFiles ?? LOCAL_COST_SCAN_MAX_FILES)); const maxFileBytes = Math.max(1, Math.floor(options.maxFileBytes ?? LOCAL_COST_SCAN_MAX_FILE_BYTES)); - const files: Array<{ path: string; mtimeMs: number }> = []; + const maxTotalBytes = options.maxTotalBytes !== undefined && Number.isFinite(options.maxTotalBytes) + ? Math.max(1, Math.floor(options.maxTotalBytes)) + : Number.POSITIVE_INFINITY; + const files: Array<{ path: string; mtimeMs: number; size: number }> = []; async function walk(current: string, depth: number) { if (depth > 6) return; // Prevent deep traversal @@ -1520,7 +1659,7 @@ export async function findRecentFiles( fileStatPromises.push( fs.promises.stat(fullPath).then((stat) => { if (stat.mtimeMs >= cutoff && stat.size <= maxFileBytes) { - files.push({ path: fullPath, mtimeMs: stat.mtimeMs }); + files.push({ path: fullPath, mtimeMs: stat.mtimeMs, size: stat.size }); } }).catch(() => { // Skip files we can't stat @@ -1535,16 +1674,21 @@ export async function findRecentFiles( } await walk(dir, 0); - return files - .sort((a, b) => b.mtimeMs - a.mtimeMs) - .slice(0, maxFiles) - .map((file) => file.path); + const selected: string[] = []; + let selectedBytes = 0; + for (const file of files.sort((a, b) => b.mtimeMs - a.mtimeMs)) { + if (selected.length >= maxFiles) break; + if (selectedBytes + file.size > maxTotalBytes) continue; + selected.push(file.path); + selectedBytes += file.size; + } + return selected; } export async function findJsonlFiles( dir: string, maxAgeDays: number, - options: { maxFiles?: number; maxFileBytes?: number } = {}, + options: { maxFiles?: number; maxFileBytes?: number; maxTotalBytes?: number } = {}, ): Promise { return findRecentFiles(dir, maxAgeDays, [".jsonl"], options); } @@ -1619,15 +1763,56 @@ async function findClaudeJsonlFilesInProjectDirs(projectDirs: string[], maxAgeDa return newestCandidatePaths(Array.from(files.values())); } -async function* readJsonlLines(filePath: string): AsyncGenerator { - const stream = fs.createReadStream(filePath, { encoding: "utf8" }); - const lines = createInterface({ input: stream, crlfDelay: Infinity }); +async function* readJsonlLines( + filePath: string, + maxLineBytes = LOCAL_JSONL_MAX_LINE_BYTES, +): AsyncGenerator { + const normalizedMaxLineBytes = Number.isFinite(maxLineBytes) + ? Math.max(1, Math.floor(maxLineBytes)) + : LOCAL_JSONL_MAX_LINE_BYTES; + const stream = fs.createReadStream(filePath); + let lineChunks: Buffer[] = []; + let lineBytes = 0; + let discardingOversizedLine = false; + + const takeLine = (): string => { + const line = lineChunks.length === 1 + ? lineChunks[0]! + : Buffer.concat(lineChunks, lineBytes); + const content = line[line.length - 1] === 0x0d ? line.subarray(0, -1) : line; + return content.toString("utf8"); + }; + try { - for await (const line of lines) { - yield line; + for await (const chunk of stream) { + let start = 0; + while (start < chunk.length) { + const newline = chunk.indexOf(0x0a, start); + const end = newline >= 0 ? newline : chunk.length; + + if (!discardingOversizedLine) { + const segment = chunk.subarray(start, end); + if (lineBytes + segment.length <= normalizedMaxLineBytes) { + lineChunks.push(segment); + lineBytes += segment.length; + } else { + lineChunks = []; + lineBytes = 0; + discardingOversizedLine = true; + } + } + + if (newline < 0) break; + if (!discardingOversizedLine) yield takeLine(); + lineChunks = []; + lineBytes = 0; + discardingOversizedLine = false; + start = newline + 1; + } } + + if (!discardingOversizedLine && lineBytes > 0) yield takeLine(); } finally { - lines.close(); stream.destroy(); } } diff --git a/apps/desktop/src/main/services/usage/usageTrackingService.test.ts b/apps/desktop/src/main/services/usage/usageTrackingService.test.ts index d396285cb..670f2a7fc 100644 --- a/apps/desktop/src/main/services/usage/usageTrackingService.test.ts +++ b/apps/desktop/src/main/services/usage/usageTrackingService.test.ts @@ -41,6 +41,7 @@ import { createUsageTrackingService, _testing } from "./usageTrackingService"; const { aggregateCosts, + bucketDaily7d, localDayKey, makeDailySkeleton, dateIntersectsRange, @@ -72,6 +73,7 @@ const { scanDroidLogs, scanCopilotLogs, scanGeminiLogs, + findRecentFiles, } = _testing; // ── Helpers ────────────────────────────────────────────────────── @@ -326,6 +328,53 @@ describe("aggregateCosts", () => { expect(result.todayCostUsd).toBe(0); }); + it("keeps lifetime-only reconciliation out of recent and daily buckets", () => { + const now = Date.now(); + const result = aggregateCosts([ + { + messageId: "current:1", + model: "gpt-5.5", + inputTokens: 10, + outputTokens: 0, + cachedTokens: 0, + timestamp: now, + }, + { + messageId: "codex-state:lifetime-total-remainder", + model: "codex", + lifetimeOnly: true as const, + inputTokens: 50, + outputTokens: 0, + cachedTokens: 0, + timestamp: 0, + costOverrideUsd: 0, + }, + ], "codex"); + + expect(result.tokenBreakdownByPreset?.all?.codex?.input).toBe(50); + expect(result.tokenBreakdownByPreset?.today?.codex).toBeUndefined(); + expect(Object.values(result.dailyTokensByPreset?.all ?? {}).reduce((sum, value) => sum + value, 0)).toBe(10); + expect(bucketDaily7d([ + { + messageId: "current:1", + model: "gpt-5.5", + inputTokens: 10, + outputTokens: 0, + cachedTokens: 0, + timestamp: now, + }, + { + messageId: "codex-state:lifetime-total-remainder", + model: "codex", + lifetimeOnly: true, + inputTokens: 50, + outputTokens: 0, + cachedTokens: 0, + timestamp: 0, + }, + ], now).reduce((sum, value) => sum + value, 0)).toBe(10); + }); + it("separates today cost from 30d cost", () => { const now = Date.now(); const yesterdayMs = now - 25 * 60 * 60 * 1000; // 25h ago @@ -2440,6 +2489,129 @@ describe("scanClaudeLogs (via aggregateCosts)", () => { }); describe("scanCodexLogs", () => { + it("skips an oversized record and processes the following token record", async () => { + const tmpDir = makeTmpDir(); + const originalCodexHome = process.env.CODEX_HOME; + try { + process.env.CODEX_HOME = tmpDir; + const sessionDir = path.join(tmpDir, "sessions", "2026", "07", "12"); + fs.mkdirSync(sessionDir, { recursive: true }); + fs.writeFileSync( + path.join(sessionDir, "rollout-test.jsonl"), + [ + JSON.stringify({ + timestamp: "2026-07-12T12:00:00.000Z", + type: "session_meta", + payload: { id: "session-1", originator: "codex_cli_rs", model: "gpt-5.5" }, + }), + JSON.stringify({ + timestamp: "2026-07-12T12:00:01.000Z", + type: "response_item", + payload: { type: "function_call_output", output: "x".repeat(2_048) }, + }), + JSON.stringify({ + timestamp: "2026-07-12T12:00:02.000Z", + type: "event_msg", + payload: { + type: "token_count", + info: { + total_token_usage: { input_tokens: 12, output_tokens: 3, total_tokens: 15 }, + last_token_usage: { input_tokens: 12, output_tokens: 3, total_tokens: 15 }, + }, + }, + }), + "", + ].join("\n"), + ); + + const entries = await scanCodexLogs({ maxJsonlLineBytes: 1_024 }); + + expect(entries).toHaveLength(1); + expect(entries[0]).toMatchObject({ + messageId: "session-1:2026-07-12T12:00:02.000Z:15", + model: "gpt-5.5", + inputTokens: 12, + outputTokens: 3, + }); + } finally { + if (originalCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = originalCodexHome; + fs.rmSync(tmpDir, { recursive: true, force: true }); + } + }); + + it("coalesces concurrent production history scans", async () => { + const tmpDir = makeTmpDir(); + const originalCodexHome = process.env.CODEX_HOME; + try { + process.env.CODEX_HOME = tmpDir; + const sessionDir = path.join(tmpDir, "sessions", "2026", "07", "12"); + fs.mkdirSync(sessionDir, { recursive: true }); + fs.writeFileSync( + path.join(sessionDir, "rollout-test.jsonl"), + [ + JSON.stringify({ + timestamp: "2026-07-12T12:00:00.000Z", + type: "session_meta", + payload: { id: "session-coalesced", originator: "codex_cli_rs", model: "gpt-5.5" }, + }), + JSON.stringify({ + timestamp: "2026-07-12T12:00:01.000Z", + type: "event_msg", + payload: { + type: "token_count", + info: { + total_token_usage: { input_tokens: 4, output_tokens: 1, total_tokens: 5 }, + last_token_usage: { input_tokens: 4, output_tokens: 1, total_tokens: 5 }, + }, + }, + }), + "", + ].join("\n"), + ); + + const first = scanCodexLogs(); + const second = scanCodexLogs(); + + expect(second).toBe(first); + const [firstEntries, secondEntries] = await Promise.all([first, second]); + expect(secondEntries).toBe(firstEntries); + expect(firstEntries).toHaveLength(1); + } finally { + if (originalCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = originalCodexHome; + fs.rmSync(tmpDir, { recursive: true, force: true }); + } + }); + + it("selects newest ledger files within per-file and aggregate byte budgets", async () => { + const tmpDir = makeTmpDir(); + try { + const writeCandidate = (name: string, bytes: number, ageSeconds: number) => { + const filePath = path.join(tmpDir, name); + fs.writeFileSync(filePath, Buffer.alloc(bytes, 0x78)); + const modifiedAt = new Date(Date.now() - ageSeconds * 1_000); + fs.utimesSync(filePath, modifiedAt, modifiedAt); + return filePath; + }; + writeCandidate("old.jsonl", 700, 30); + const middle = writeCandidate("middle.jsonl", 700, 20); + const newest = writeCandidate("newest.jsonl", 700, 10); + writeCandidate("too-large.jsonl", 1_500, 1); + + const selected = await findRecentFiles(tmpDir, 3650, [".jsonl"], { + maxFiles: 10, + maxFileBytes: 1_000, + maxTotalBytes: 1_400, + }); + + expect(selected).toEqual([newest, middle]); + expect(selected).not.toContain(path.join(tmpDir, "old.jsonl")); + expect(selected).not.toContain(path.join(tmpDir, "too-large.jsonl")); + } finally { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } + }); it("parses modern token_count events from Codex session logs", async () => { const tmpDir = makeTmpDir(); const originalCodexHome = process.env.CODEX_HOME; @@ -2582,6 +2754,118 @@ describe("scanCodexLogs", () => { } }); + it("preserves the exact Codex lifetime total within the remaining entry budget", async () => { + const tmpDir = makeTmpDir(); + const originalCodexHome = process.env.CODEX_HOME; + const { DatabaseSync } = requireForTest("node:sqlite") as { DatabaseSync: new (dbPath: string) => any }; + try { + process.env.CODEX_HOME = tmpDir; + const db = new DatabaseSync(path.join(tmpDir, "state_5.sqlite")); + db.exec(` + create table threads ( + id text primary key, + tokens_used integer not null, + model text, + cwd text, + source text, + thread_source text, + created_at integer, + updated_at integer + ); + insert into threads values + ('oldest', 10, 'gpt-5.5', '/repo', 'Codex Desktop', 'Codex Desktop', 100, 100), + ('middle', 20, 'gpt-5.5', '/repo', 'Codex Desktop', 'Codex Desktop', 200, 200), + ('newest', 30, 'gpt-5.5', '/repo', 'Codex Desktop', 'Codex Desktop', 300, 300); + `); + db.close(); + + const entries = await scanCodexLogs({ maxEntries: 2 }); + + expect(entries).toHaveLength(2); + expect(entries.map((entry) => entry.messageId)).toEqual([ + "newest:state-total-remainder", + "codex-state:lifetime-total-remainder", + ]); + expect(entries.reduce( + (total, entry) => total + entry.inputTokens + entry.outputTokens + entry.cachedTokens, + 0, + )).toBe(60); + expect(entries[1]).toMatchObject({ lifetimeOnly: true, inputTokens: 30, costOverrideUsd: 0 }); + } finally { + if (originalCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = originalCodexHome; + fs.rmSync(tmpDir, { recursive: true, force: true }); + } + }); + + it("unions disjoint JSONL and SQLite threads within the bounded lifetime summary", async () => { + const tmpDir = makeTmpDir(); + const originalCodexHome = process.env.CODEX_HOME; + const { DatabaseSync } = requireForTest("node:sqlite") as { DatabaseSync: new (dbPath: string) => any }; + try { + process.env.CODEX_HOME = tmpDir; + const sessionDir = path.join(tmpDir, "sessions", "2026", "07", "12"); + fs.mkdirSync(sessionDir, { recursive: true }); + fs.writeFileSync( + path.join(sessionDir, "json-only.jsonl"), + [ + JSON.stringify({ + timestamp: "2026-07-12T12:00:00.000Z", + type: "session_meta", + payload: { id: "json-only", originator: "codex_cli_rs", model: "gpt-5.5" }, + }), + JSON.stringify({ + timestamp: "2026-07-12T12:00:01.000Z", + type: "event_msg", + payload: { + type: "token_count", + info: { + total_token_usage: { input_tokens: 80, output_tokens: 0, total_tokens: 80 }, + last_token_usage: { input_tokens: 80, output_tokens: 0, total_tokens: 80 }, + }, + }, + }), + "", + ].join("\n"), + ); + const db = new DatabaseSync(path.join(tmpDir, "state_5.sqlite")); + db.exec(` + create table threads ( + id text primary key, + tokens_used integer not null, + model text, + cwd text, + source text, + thread_source text, + created_at integer, + updated_at integer + ); + insert into threads values ( + 'state-only', 100, 'gpt-5.5', '/repo', 'Codex Desktop', + 'Codex Desktop', 100, 100 + ); + `); + db.close(); + + const entries = await scanCodexLogs({ maxEntries: 2 }); + const total = entries.reduce((sum, entry) => ( + sum + entry.inputTokens + entry.outputTokens + entry.cachedTokens + (entry.cacheWriteTokens ?? 0) + ), 0); + + expect(entries).toHaveLength(2); + expect(entries.map((entry) => entry.messageId)).toEqual([ + "json-only:2026-07-12T12:00:01.000Z:80", + "codex-state:lifetime-total-remainder", + ]); + expect(total).toBe(180); + expect(entries[1]).toMatchObject({ lifetimeOnly: true, inputTokens: 100, costOverrideUsd: 0 }); + } finally { + if (originalCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = originalCodexHome; + fs.rmSync(tmpDir, { recursive: true, force: true }); + } + }); + it("includes Codex archived session ledgers in lifetime usage", async () => { const tmpDir = makeTmpDir(); const originalCodexHome = process.env.CODEX_HOME; diff --git a/apps/desktop/src/main/services/usage/usageTrackingService.ts b/apps/desktop/src/main/services/usage/usageTrackingService.ts index 562a7230a..63c37022d 100644 --- a/apps/desktop/src/main/services/usage/usageTrackingService.ts +++ b/apps/desktop/src/main/services/usage/usageTrackingService.ts @@ -1112,6 +1112,7 @@ function bucketDaily7d(entries: TokenEntry[], nowMs: number): number[] { bucketByDay.set(localDayKey(day), index); } for (const entry of entries) { + if (entry.lifetimeOnly) continue; if (entry.timestamp > nowMs) continue; const bucketIndex = bucketByDay.get(localDayKey(entry.timestamp)); if (bucketIndex == null) continue; @@ -1217,6 +1218,12 @@ function aggregateCosts( for (const entry of entries) { const cost = calculateTokenEntryCost(entry); + if (entry.lifetimeOnly) { + const allTime = accumulators.all; + allTime.costUsd += cost; + addTokenBreakdownEntry(allTime.tokenBreakdown, entry); + continue; + } for (const preset of ADE_USAGE_RANGE_PRESETS) { const startMs = starts[preset]; if (startMs != null && entry.timestamp < startMs) continue; diff --git a/docs/features/chat/transcript-and-turns.md b/docs/features/chat/transcript-and-turns.md index e16b117b4..45c5ec9ae 100644 --- a/docs/features/chat/transcript-and-turns.md +++ b/docs/features/chat/transcript-and-turns.md @@ -325,6 +325,38 @@ Persisted-history consumers see the stored preview on replay. socket path, OpenCode runtime ids) is rehydrated so the next turn can use the same session instead of creating a new one. +### Claude restart and Stop recovery + +Every parent turn must finish with both a terminal `status` and a matching +`done` event. A process crash can occur after the user message or +`status: "started"` has been persisted but before that pair is written. When a +Claude runtime is created, `agentChatService` therefore checks the latest +non-steer parent turn even when the previous process never persisted an SDK +session id. It fills in only the missing member of the terminal pair, preserves +an already-written terminal status, marks the session idle, and persists the +repair. A newer complete parent turn makes an older incomplete turn irrelevant; +restart recovery never rewrites historical turns. + +Restart reconciliation first closes orphaned background and subagent rows, then +appends the parent terminal pair last. This ordering is deliberate: renderer +turn state is derived in event order, so a cleanup row must not make a repaired +turn look active again. Pressing Stop on an already-idle Claude runtime runs the +same parent-turn repair, which lets a stale red Stop state settle without +requiring a live Claude process. Repeated reconciliation and repeated Stop calls +are idempotent because an already-complete `status` + `done` pair is no longer an +unsettled turn. + +Live Claude control calls are bounded independently of the desktop action +timeout. Provider `interrupt()` gets 2.5 seconds; active `stopTask()` calls get +2 seconds each and run concurrently. During ordinary Stop, a hung SDK control +channel is logged and local interruption cleanup continues rather than holding +the action bridge until its 30-second request timeout; an interrupt-and-replace +request that requires provider acknowledgement fails within the control-call +bound instead of sending the replacement ambiguously. Likewise, a steer sent to +an idle or stale Claude session waits only for input-dispatch acceptance; the +provider turn keeps streaming asynchronously instead of making the steer action +wait for the full answer. + Codex adapters deduplicate repeated lifecycle notifications before converting them to envelope events. Terminal app-server failures use a bounded semantic key (turn id + message + detail + error identity) shared by @@ -349,5 +381,6 @@ regain duplicate visible failures after restart. - **Turn diff emission depends on lane context.** If a session is disassociated from a lane, `turn_diff_summary` will not emit. Do not rely on it for non-lane surfaces. - - +- **Claude parent terminal events are an ordered pair.** Restart and idle-Stop + repair must leave the parent `status` + `done` pair after any orphan cleanup. + Emitting later lifecycle rows can resurrect a stopped renderer state. diff --git a/docs/features/onboarding-and-settings/README.md b/docs/features/onboarding-and-settings/README.md index 1b6e18d98..ab41ddbc5 100644 --- a/docs/features/onboarding-and-settings/README.md +++ b/docs/features/onboarding-and-settings/README.md @@ -302,6 +302,14 @@ Renderer — settings: awaiting expensive scans, exposes freshness metadata (`fresh` / `refreshing`), and coalesces stale provider/GitHub revalidation in the background (`refreshStatsInBackground`, single-flight per range + source). +- `apps/desktop/src/main/services/usage/ledgers/localUsageLedgers.ts` — + read-only provider-history adapters. The Codex path selects recent JSONL + files within per-file and aggregate byte budgets, discards oversized physical + records with a bounded byte-stream reader, caps detailed entries, and shares + one production scan across callers. It reconciles the available JSONL history + with the newest Codex `state_*.sqlite` thread index under bounded row and + lookup budgets; a zero-cost all-time-only remainder preserves the exact union + token headline without fabricating day, project, or cost attribution. - `apps/desktop/src/main/services/usage/usageStatsStore.ts` — aggregates the project database and owns the low-volume `usage_events` ledger. Only successful, meaningful user mutations are recorded; read/poll IPC is diff --git a/docs/features/onboarding-and-settings/usage-tracking.md b/docs/features/onboarding-and-settings/usage-tracking.md index 8ca89f354..4423d96d4 100644 --- a/docs/features/onboarding-and-settings/usage-tracking.md +++ b/docs/features/onboarding-and-settings/usage-tracking.md @@ -41,6 +41,48 @@ quota HTTP, CLI fallback, and history phases, including provider, trigger, duration, outcome, and error kind. These entries identify network/auth latency without logging credentials or quota payloads. +## Bounded local ledger scanning + +Activity reads provider-owned history in place. ADE does not copy, rewrite, or +delete Codex session JSONL under `~/.codex`; Codex remains the owner of chat +history and retention. A single JSONL record can nevertheless be enormous when +it contains embedded command or tool output. That is valid JSONL, but treating +the whole physical line as one JavaScript string can exhaust the runtime before +the parser has a chance to ignore the irrelevant payload. + +The Codex history reader is therefore a bounded byte-stream pipeline: + +- candidate session and archived-session files are considered newest first, + with at most 5,000 files per root, 256 MiB per file, and 2 GiB distributed + across both roots; +- physical JSONL lines are accumulated only up to 16 MiB. An oversized record + is discarded incrementally until its newline, then scanning resumes at the + next record instead of retaining or parsing the giant line; +- detailed history stops at 250,000 token entries; and +- concurrent production callers share one in-flight Codex scan, so two open + projects cannot duplicate the same CPU work and retained entry set. + +These limits bound ADE's work; they do not truncate the source files. The +tradeoff is intentionally visible in the data model: an extreme old record or +history beyond the detail budget may be absent from per-day, per-model, +per-project, and estimated-cost attribution. + +The all-time token headline has a separate reconciliation path. ADE opens the +newest Codex `state_*.sqlite` read-only, computes the state index's authoritative +thread total, point-looks up the bounded set of JSONL thread ids to avoid double +counting, and treats the result as the union of JSONL history and the current +state index. Per-thread remainders are added newest first within the remaining +entry budget. If detail capacity is exhausted, one zero-cost `lifetimeOnly` +remainder preserves the exact union total without inventing a timestamp, +project, ADE-originated share, or recent-day activity. Daily charts skip that +entry, while the all-time token breakdown includes it. + +SQLite reconciliation is bounded too: observed-thread lookups use batches of +500, detailed state rows are capped at 250,000 and by remaining entry capacity, +and the production scanner remains single-flight. This keeps Activity useful on +large Codex histories without putting live Limits refreshes or the ADE runtime +behind an unbounded disk/memory pass. + ## Claude credential hygiene (refresh storms) `~/.claude/.credentials.json` can be a stale leftover while the live login sits From bcb3854a0dd419f27261927b4b17631d76463b04 Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Sun, 12 Jul 2026 16:56:12 -0400 Subject: [PATCH 2/2] =?UTF-8?q?ship:=20iteration=201=20=E2=80=94=20address?= =?UTF-8?q?=20chat=20lifecycle=20review?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../services/chat/agentChatService.test.ts | 180 +++++++++++++++++- .../main/services/chat/agentChatService.ts | 128 +++++++++---- .../components/chat/AgentChatPane.test.tsx | 43 +++++ .../components/chat/AgentChatPane.tsx | 9 +- .../renderer/components/chat/chatTurnState.ts | 10 +- 5 files changed, 324 insertions(+), 46 deletions(-) diff --git a/apps/desktop/src/main/services/chat/agentChatService.test.ts b/apps/desktop/src/main/services/chat/agentChatService.test.ts index 61165d385..36a7ffdb4 100644 --- a/apps/desktop/src/main/services/chat/agentChatService.test.ts +++ b/apps/desktop/src/main/services/chat/agentChatService.test.ts @@ -10135,16 +10135,14 @@ describe("createAgentChatService", () => { const orphanTail: AgentChatEventEnvelope[] = [ { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 1, event: { type: "user_message", text: "Work interrupted by restart", turnId: "turn-old", + messageId: "idle-steer-parent-message", steerId: "idle-steer-before-restart", deliveryState: "delivered", } as any }, { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 2, event: { - type: "status", turnStatus: "started", turnId: "turn-old", - } as any }, - { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 3, event: { type: "scheduled_work_update", id: "background:bg-restart", kind: "background_task", status: "running", origin: "background_task", title: "npm run serve", summary: "shell", sourceTaskId: "bg-restart", turnId: "turn-old", } as any }, - { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 4, event: { + { sessionId: session.id, timestamp: new Date().toISOString(), sequence: 3, event: { type: "subagent_started", taskId: "sub-restart", agentId: "sub-restart", agentType: "Explore", parentToolUseId: "toolu_sub_r", description: "look", turnId: "turn-old", } as any }, @@ -26268,16 +26266,31 @@ describe("createAgentChatService", () => { const session = await service.createSession({ laneId: "lane-1", provider: "claude", model: "sonnet" }); await vi.waitFor(() => { expect(warmupComplete).toBe(true); }); - await expect(service.steer({ + const result = await service.steer({ sessionId: session.id, text: "Treat this stale steer as a normal turn", dispatchMode: "inline", - })).resolves.toMatchObject({ queued: false, steerId: expect.any(String) }); + reasoningEffort: "high", + executionMode: "subagents", + interactionMode: "plan", + }); + expect(result).toMatchObject({ queued: false, steerId: expect.any(String) }); - expect(events.some((event) => + const delivered = events.find((event) => event.event.type === "user_message" && event.event.text === "Treat this stale steer as a normal turn" - )).toBe(true); + ); + expect(delivered?.event).toMatchObject({ + type: "user_message", + steerId: result.steerId, + deliveryState: "delivered", + turnId: expect.any(String), + }); + await expect(service.getSessionSummary(session.id)).resolves.toMatchObject({ + reasoningEffort: "high", + executionMode: "subagents", + interactionMode: "plan", + }); expect(events.some((event) => event.event.type === "done")).toBe(false); finishTurn(); @@ -26285,6 +26298,157 @@ describe("createAgentChatService", () => { event.event.type === "done" && event.event.status === "completed"); }); + it("emits one interrupted terminal pair when a Claude model switches mid-turn", async () => { + const events: AgentChatEventEnvelope[] = []; + let streamCall = 0; + let warmupComplete = false; + let oldStreamFinished = false; + let releaseActiveTurn!: () => void; + const activeTurnGate = new Promise((resolve) => { releaseActiveTurn = resolve; }); + let releaseReplacementTurn!: () => void; + const replacementTurnGate = new Promise((resolve) => { releaseReplacementTurn = resolve; }); + let replacementTurnStreaming = false; + const send = vi.fn().mockResolvedValue(undefined); + const close = vi.fn(); + const stream = vi.fn(() => (async function* () { + streamCall += 1; + if (streamCall === 1) { + yield { type: "system", subtype: "init", session_id: "sdk-model-switch", slash_commands: [] }; + warmupComplete = true; + yield { type: "result", usage: { input_tokens: 1, output_tokens: 1 } }; + return; + } + if (streamCall === 2) { + yield { + type: "assistant", + message: { content: [{ type: "text", text: "Switch me while I am running" }], usage: { input_tokens: 1, output_tokens: 1 } }, + }; + await activeTurnGate; + oldStreamFinished = true; + return; + } + if (streamCall === 3) { + yield { type: "system", subtype: "init", session_id: "sdk-model-switch-next", slash_commands: [] }; + yield { type: "result", usage: { input_tokens: 1, output_tokens: 1 } }; + return; + } + replacementTurnStreaming = true; + yield { + type: "assistant", + message: { content: [{ type: "text", text: "Replacement turn is active" }], usage: { input_tokens: 1, output_tokens: 1 } }, + }; + await replacementTurnGate; + yield { type: "result", usage: { input_tokens: 1, output_tokens: 1 } }; + })()); + const mockSession = { + send, + stream, + close, + sessionId: "sdk-model-switch", + setPermissionMode: vi.fn().mockResolvedValue(undefined), + }; + vi.mocked(claudeSdkCreateSessionCompat).mockReturnValue(mockSession as any); + vi.mocked(claudeSdkResumeSessionCompat).mockReturnValue(mockSession as any); + + const { service } = createService({ onEvent: (event: AgentChatEventEnvelope) => events.push(event) }); + const session = await service.createSession({ + laneId: "lane-1", + provider: "claude", + model: "claude-opus-4-8", + modelId: "anthropic/claude-opus-4-8", + }); + await vi.waitFor(() => { expect(warmupComplete).toBe(true); }); + + await service.sendMessage({ + sessionId: session.id, + text: "Keep working while I switch models", + }, { awaitDispatch: true }); + const started = await waitForEvent(events, (event): event is AgentChatEventEnvelope => + event.event.type === "status" && event.event.turnStatus === "started"); + await waitForEvent(events, (event): event is AgentChatEventEnvelope => + event.event.type === "text" && event.event.turnId === started.event.turnId); + const queued = await service.steer({ + sessionId: session.id, + text: "Do this after the old model finishes", + }); + expect(queued).toMatchObject({ queued: true, steerId: expect.any(String) }); + + // Force the restart reconciler's view of the transcript to lag behind the + // live stream, matching the race where model-switch teardown creates the + // replacement runtime before the old stream emits its terminal pair. + vi.mocked(parseAgentChatTranscript).mockReturnValue([ + { + sessionId: session.id, + timestamp: new Date().toISOString(), + sequence: 1, + event: { type: "user_message", text: "Keep working while I switch models", turnId: started.event.turnId } as any, + }, + { + sessionId: session.id, + timestamp: new Date().toISOString(), + sequence: 2, + event: { type: "status", turnStatus: "started", turnId: started.event.turnId } as any, + }, + ]); + + await service.updateSession({ + sessionId: session.id, + modelId: "anthropic/claude-sonnet-5", + }); + const interruptedDone = await waitForEvent(events, (event): event is AgentChatEventEnvelope => + event.event.type === "done" + && event.event.turnId === started.event.turnId + && event.event.status === "interrupted"); + await waitForEvent(events, (event): event is AgentChatEventEnvelope => + event.event.type === "system_notice" + && event.event.steerId === queued.steerId + && event.event.message.includes("cancelled")); + + await vi.waitFor(() => { expect(streamCall).toBeGreaterThanOrEqual(3); }); + await service.sendMessage({ + sessionId: session.id, + text: "Replacement turn after model switch", + }, { awaitDispatch: true }); + const replacementStarted = await waitForEvent(events, (event): event is AgentChatEventEnvelope => + event.event.type === "status" + && event.event.turnStatus === "started" + && event.event.turnId !== started.event.turnId); + await vi.waitFor(() => { expect(replacementTurnStreaming).toBe(true); }); + + releaseActiveTurn(); + await vi.waitFor(() => { expect(oldStreamFinished).toBe(true); }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(events.filter((event) => + event.event.type === "status" + && event.event.turnId === started.event.turnId + && event.event.turnStatus === "interrupted" + )).toHaveLength(1); + const doneEvents = events.filter((event) => + event.event.type === "done" + && event.event.turnId === started.event.turnId + && event.event.status === "interrupted" + ); + expect(doneEvents).toHaveLength(1); + expect(interruptedDone.event).toMatchObject({ + model: "claude-opus-4-8", + modelId: "anthropic/claude-opus-4-8", + }); + await expect(service.getSessionSummary(session.id)).resolves.toMatchObject({ status: "active" }); + expect(close).toHaveBeenCalled(); + expect(events.filter((event) => + event.event.type === "user_message" + && event.event.steerId === queued.steerId + && event.event.deliveryState === "delivered" + )).toHaveLength(0); + + releaseReplacementTurn(); + await waitForEvent(events, (event): event is AgentChatEventEnvelope => + event.event.type === "done" + && event.event.turnId === replacementStarted.event.turnId + && event.event.status === "completed"); + }); + it("dispatchSteer mode:'interrupt' uses Claude priority-now without tearing down the query", async () => { const events: AgentChatEventEnvelope[] = []; const send = vi.fn().mockResolvedValue(undefined); diff --git a/apps/desktop/src/main/services/chat/agentChatService.ts b/apps/desktop/src/main/services/chat/agentChatService.ts index ad2ccc544..db762d0d5 100644 --- a/apps/desktop/src/main/services/chat/agentChatService.ts +++ b/apps/desktop/src/main/services/chat/agentChatService.ts @@ -2164,6 +2164,7 @@ type PreparedSendMessage = { forceClaudeUserMessage?: boolean; onDispatched?: () => void; onBackendDispatched?: () => void; + steerId?: string; turnId?: string; optimisticCursorTurnStart?: boolean; optimisticDroidTurnStart?: boolean; @@ -8183,7 +8184,14 @@ export function createAgentChatService(args: { if (isCodexSubagentTranscriptEnvelope(entry)) continue; const event = entry.event; - if (event.type === "user_message" && !event.steerId) { + // Active/queued steers have no messageId and belong to an already-running + // parent turn, so they must not replace its anchor. A fresh idle-steer + // turn receives runClaudeTurn's durable messageId and remains recoverable + // even though it also carries the caller-facing steerId. + if ( + event.type === "user_message" + && (!event.steerId || Boolean(event.messageId?.trim())) + ) { const turnId = event.turnId?.trim(); if (turnId) { latest = { turnId, terminalStatus: null, doneStatus: null }; @@ -12534,39 +12542,69 @@ export function createAgentChatService(args: { managed.runtime = null; } if (managed.runtime?.kind === "claude") { + const runtime = managed.runtime; + const modelSwitchTurnId = openCodeReason === "model_switch" && runtime.busy + ? runtime.activeTurnId + : null; + if (modelSwitchTurnId && !runtime.interruptEventsEmitted) { + // A replacement runtime may start before this abandoned stream + // unwinds. Own the old turn's terminal pair synchronously so restart + // reconciliation and detached finalization cannot both emit it. + runtime.interruptEventsEmitted = true; + emitChatEvent(managed, { + type: "status", + turnStatus: "interrupted", + turnId: modelSwitchTurnId, + }); + void emitTurnDiffSummaryIfChanged(managed, modelSwitchTurnId); + emitChatEvent(managed, { + type: "done", + turnId: modelSwitchTurnId, + status: "interrupted", + ...resolveClaudeTurnModelPayload(managed.session, []), + }); + markSessionIdleWithFreshCache(managed); + } + if (openCodeReason === "model_switch") { + // The old runtime owns these staged rows. A model switch cannot safely + // deliver them through the replacement query, so settle their UI state + // before discarding the queue. + cancelQueuedSteers(managed, runtime, "interrupted"); + persistChatState(managed); + } // Mark interrupted so the streaming catch block takes the graceful path - managed.runtime.interrupted = true; + runtime.interrupted = true; if (preserveProviderResumeState) persistChatState(managed); - cancelClaudeWarmup(managed, managed.runtime, "teardown"); - try { managed.runtime.query?.close(); } catch { /* ignore */ } + cancelClaudeWarmup(managed, runtime, "teardown"); + try { runtime.query?.close(); } catch { /* ignore */ } // Every teardown abandons this query's control channel. Enforce process // ownership even for idle eviction so an ended iterator cannot leave a // detached Claude worker (or its children) behind. claudeSubprocessReaper.reapForSession(managed.session.id, openCodeReason); - managed.runtime.inputPump?.close(); - try { managed.runtime.warmQuery?.close(); } catch { /* ignore */ } - settleClaudeInitialInputDispatch(managed.runtime, new Error("Claude runtime was closed before the turn input was dispatched.")); - resetClaudeProcessBackgroundLevel(managed.runtime); - managed.runtime.query = null; - managed.runtime.inputPump = null; - managed.runtime.warmQuery = null; - managed.runtime.warmupDone = null; + runtime.inputPump?.close(); + try { runtime.warmQuery?.close(); } catch { /* ignore */ } + settleClaudeInitialInputDispatch(runtime, new Error("Claude runtime was closed before the turn input was dispatched.")); + resetClaudeProcessBackgroundLevel(runtime); + runtime.query = null; + runtime.inputPump = null; + runtime.warmQuery = null; + runtime.warmupDone = null; // Query is already null, so settle every visible background/native task // without trying provider stopTask on the dead control channel. void stopActiveClaudeSubagents( managed, - managed.runtime, - managed.runtime.activeTurnId ?? undefined, + runtime, + runtime.activeTurnId ?? undefined, "The Claude session ended before this task reported completion.", ); - managed.runtime.emittedSubagentStartIds.clear(); - managed.runtime.taskToolInputByToolUseId.clear(); - managed.runtime.workflowAgentsByTask.clear(); - managed.runtime.dispatchingSteerIds.clear(); - for (const pending of managed.runtime.approvals.values()) { + runtime.emittedSubagentStartIds.clear(); + runtime.taskToolInputByToolUseId.clear(); + runtime.workflowAgentsByTask.clear(); + runtime.dispatchingSteerIds.clear(); + for (const pending of runtime.approvals.values()) { pending.resolve({ decision: "cancel" }); } - managed.runtime.approvals.clear(); + runtime.approvals.clear(); managed.runtime = null; } if (managed.runtime?.kind === "opencode") { @@ -12989,6 +13027,7 @@ export function createAgentChatService(args: { metadata?: AgentChatEventMetadata | null | undefined; turnId?: string; messageId?: string; + steerId?: string; laneDirectiveKey?: string | null; onDispatched?: () => void; }, @@ -13004,6 +13043,7 @@ export function createAgentChatService(args: { ...(args.metadata ? { metadata: args.metadata } : {}), ...(args.turnId ? { turnId: args.turnId } : {}), ...(args.messageId ? { messageId: args.messageId } : {}), + ...(args.steerId ? { steerId: args.steerId, deliveryState: "delivered" as const } : {}), }); args.onDispatched?.(); }; @@ -14891,6 +14931,7 @@ export function createAgentChatService(args: { laneDirectiveKey?: string | null; providerSlashCommand?: boolean; forceClaudeUserMessage?: boolean; + steerId?: string; onDispatched?: () => void; onBackendDispatched?: () => void; }, @@ -14948,6 +14989,7 @@ export function createAgentChatService(args: { metadata: args.metadata, turnId, messageId: userMessageId, + steerId: args.steerId, laneDirectiveKey: args.laneDirectiveKey, onDispatched: args.onDispatched, }); @@ -16864,13 +16906,16 @@ export function createAgentChatService(args: { flushOpenClaudeToolUses(runtime.interrupted ? "interrupted" : "completed"); flushClaudeStructuredActivities(runtime.interrupted ? "interrupted" : "completed"); // Note: query is NOT closed here — it stays alive for the next turn. + const runtimeStillCurrent = managed.runtime === runtime; runtime.busy = false; runtime.activeTurnId = null; - markSessionIdleWithFreshCache(managed); - reportProviderRuntimeReady("claude"); + if (runtimeStillCurrent) { + markSessionIdleWithFreshCache(managed); + reportProviderRuntimeReady("claude"); + } // Flush deferred session reset from mid-turn reasoning effort change - if (runtime.pendingSessionReset) { + if (runtimeStillCurrent && runtime.pendingSessionReset) { const clearSdkSessionId = runtime.pendingSessionResetClearSdkSessionId === true; runtime.pendingSessionReset = false; runtime.pendingSessionResetClearSdkSessionId = false; @@ -16906,7 +16951,7 @@ export function createAgentChatService(args: { // Adopt the SDK-generated session title (once) when the turn settles and // the chat still carries the provider-default name. Fire-and-forget. - maybeAdoptClaudeSdkSessionTitle(managed, runtime); + if (runtimeStillCurrent) maybeAdoptClaudeSdkSessionTitle(managed, runtime); const endSha = await computeHeadShaBestEffort(resolveManagedExecutionLaneId(managed)).catch(() => null); if (endSha) { @@ -16916,11 +16961,13 @@ export function createAgentChatService(args: { persistChatState(managed); // Process queued steers (skip if session was disposed during execution) - if (runtime.pendingSteers.length) { - const delivered = await deliverNextQueuedSteer(managed, runtime); - if (!delivered) startClaudeIdleReader(managed, runtime, "turn_completed"); - } else { - startClaudeIdleReader(managed, runtime, "turn_completed"); + if (managed.runtime === runtime) { + if (runtime.pendingSteers.length) { + const delivered = await deliverNextQueuedSteer(managed, runtime); + if (!delivered) startClaudeIdleReader(managed, runtime, "turn_completed"); + } else { + startClaudeIdleReader(managed, runtime, "turn_completed"); + } } } catch (error) { const failedBeforeBackendDispatch = Boolean(onBackendDispatched); @@ -16970,7 +17017,7 @@ export function createAgentChatService(args: { void emitTurnDiffSummaryIfChanged(managed, turnId); if (runtime.interrupted) { - markSessionIdleWithFreshCache(managed); + if (managed.runtime === runtime) markSessionIdleWithFreshCache(managed); if (!runtime.interruptEventsEmitted) { emitChatEvent(managed, { type: "status", turnStatus: "interrupted", turnId }); emitChatEvent(managed, { @@ -17001,7 +17048,7 @@ export function createAgentChatService(args: { } else if (isAbortRelatedError(effectiveError)) { // System-triggered abort (dispose/teardown) that wasn't flagged as interrupted. // Treat as interruption to avoid surfacing raw SDK messages like "aborted by user". - markSessionIdleWithFreshCache(managed); + if (managed.runtime === runtime) markSessionIdleWithFreshCache(managed); if (!runtime.interruptEventsEmitted) { emitChatEvent(managed, { type: "status", turnStatus: "interrupted", turnId }); emitChatEvent(managed, { @@ -17012,7 +17059,7 @@ export function createAgentChatService(args: { }); } } else { - markSessionIdleWithFreshCache(managed); + if (managed.runtime === runtime) markSessionIdleWithFreshCache(managed); const isAuthFailure = isClaudeRuntimeAuthError(effectiveError); let errorMessage = isAuthFailure ? CLAUDE_RUNTIME_AUTH_ERROR @@ -29655,6 +29702,7 @@ export function createAgentChatService(args: { laneDirectiveKey, providerSlashCommand, forceClaudeUserMessage, + steerId, onDispatched, onBackendDispatched, turnId, @@ -29959,6 +30007,7 @@ export function createAgentChatService(args: { laneDirectiveKey, providerSlashCommand, forceClaudeUserMessage, + steerId, onDispatched, onBackendDispatched, }); @@ -30044,6 +30093,7 @@ export function createAgentChatService(args: { attachments: prepared.attachments, ...(prepared.contextAttachments.length ? { contextAttachments: prepared.contextAttachments } : {}), ...(prepared.metadata ? { metadata: prepared.metadata } : {}), + ...(prepared.steerId ? { steerId: prepared.steerId, deliveryState: "delivered" as const } : {}), turnId, }); markSessionIdleWithFreshCache(managed); @@ -30083,6 +30133,7 @@ export function createAgentChatService(args: { awaitDispatch?: boolean; awaitBackendDispatch?: boolean; onBackendDispatched?: () => void; + preparedMessage?: PreparedSendMessage; routeActiveToSteer: true; }, ): Promise; @@ -30092,6 +30143,7 @@ export function createAgentChatService(args: { awaitDispatch?: boolean; awaitBackendDispatch?: boolean; onBackendDispatched?: () => void; + preparedMessage?: PreparedSendMessage; routeActiveToSteer?: false; }, ): Promise; @@ -30101,6 +30153,7 @@ export function createAgentChatService(args: { awaitDispatch?: boolean; awaitBackendDispatch?: boolean; onBackendDispatched?: () => void; + preparedMessage?: PreparedSendMessage; routeActiveToSteer?: boolean; }, ): Promise { @@ -30125,7 +30178,7 @@ export function createAgentChatService(args: { interactionMode: args.interactionMode, }); } - const prepared = prepareSendMessage(args); + const prepared = options?.preparedMessage ?? prepareSendMessage(args); if (!prepared) return; prepared.managed.lastActivityTimestamp = Date.now(); let rejectDispatch: ((error: Error) => void) | null = null; @@ -30602,6 +30655,9 @@ export function createAgentChatService(args: { attachments, contextAttachments, metadata, + reasoningEffort, + executionMode, + interactionMode, }); if (!preparedSteer) { return { steerId, queued: false }; @@ -30641,6 +30697,7 @@ export function createAgentChatService(args: { ? { steerId, queued: true } : { steerId, queued: false, reason: "queue_full" }; } + preparedSteer.steerId = steerId; await sendMessage({ sessionId, text: trimmed, @@ -30651,7 +30708,10 @@ export function createAgentChatService(args: { reasoningEffort, executionMode, interactionMode, - }, { awaitDispatch: true }); + }, { + awaitDispatch: true, + preparedMessage: preparedSteer, + }); return { steerId, queued: false }; } await executePreparedSendMessage(preparedSteer); diff --git a/apps/desktop/src/renderer/components/chat/AgentChatPane.test.tsx b/apps/desktop/src/renderer/components/chat/AgentChatPane.test.tsx index b50f4266c..3a924ad06 100644 --- a/apps/desktop/src/renderer/components/chat/AgentChatPane.test.tsx +++ b/apps/desktop/src/renderer/components/chat/AgentChatPane.test.tsx @@ -45,6 +45,7 @@ import { type AgentChatSessionCreatedOptions, } from "./AgentChatPane"; import { CHAT_AUTH_RECOVERED_EVENT, CHAT_AUTH_RETRY_REJECTED_EVENT, CHAT_RETRY_AUTH_TURN_EVENT } from "./AgentCliAuthCard"; +import { findUserMessageForTurn, isParentUserMessage } from "./chatTurnState"; vi.mock("../terminals/TerminalView", () => { const ReactMod = require("react") as typeof React; @@ -7380,6 +7381,48 @@ describe("AgentChatPane submit recovery", () => { // Pure function unit tests (consolidated from AgentChatPane.test.ts) // --------------------------------------------------------------------------- +describe("correlated parent turn messages", () => { + it("keeps a fresh idle-steer parent retryable without promoting child steers", () => { + const parent = { + type: "user_message" as const, + text: "Retry this parent turn", + steerId: "idle-steer-correlation", + messageId: "durable-parent-message", + deliveryState: "delivered" as const, + turnId: "turn-parent", + }; + const childSteer = { + type: "user_message" as const, + text: "Adjust the active turn", + steerId: "active-steer-correlation", + deliveryState: "delivered" as const, + turnId: "turn-parent", + }; + const events: AgentChatEventEnvelope[] = [ + { + sessionId: "session-parent", + timestamp: "2026-07-12T12:00:00.000Z", + sequence: 1, + event: parent, + }, + { + sessionId: "session-parent", + timestamp: "2026-07-12T12:00:01.000Z", + sequence: 2, + event: childSteer, + }, + ]; + + expect(isParentUserMessage(parent)).toBe(true); + expect(isParentUserMessage(childSteer)).toBe(false); + expect(findUserMessageForTurn(events, "turn-parent")).toMatchObject({ + text: "Retry this parent turn", + steerId: "idle-steer-correlation", + messageId: "durable-parent-message", + }); + }); +}); + describe("resolveNextSelectedSessionId", () => { function buildMinimalSession(sessionId: string): AgentChatSessionSummary { return { diff --git a/apps/desktop/src/renderer/components/chat/AgentChatPane.tsx b/apps/desktop/src/renderer/components/chat/AgentChatPane.tsx index cfd195e1e..435115838 100644 --- a/apps/desktop/src/renderer/components/chat/AgentChatPane.tsx +++ b/apps/desktop/src/renderer/components/chat/AgentChatPane.tsx @@ -135,7 +135,7 @@ import { deriveChatSubagentSnapshots, deriveScheduledWorkSnapshots, deriveTodoIt import { deriveMissionSnapshot } from "./chatMission"; import { MissionControlPanel } from "./MissionControlPanel"; import { derivePendingInputRequests, type DerivedPendingInput } from "./pendingInput"; -import { findUserMessageForTurn, resolveTurnActive } from "./chatTurnState"; +import { findUserMessageForTurn, isParentUserMessage, resolveTurnActive } from "./chatTurnState"; import { ModelPicker } from "../shared/ModelPicker/ModelPicker"; import { ReasoningEffortPicker } from "../shared/ModelPicker/ReasoningEffortPicker"; import { ConfirmDialog, useConfirmDialog } from "../shared/InlineDialogs"; @@ -6820,7 +6820,12 @@ export function AgentChatPane({ if (!failedTurnId) { for (let index = events.length - 1; index >= 0; index -= 1) { const evt = events[index]?.event; - if (evt?.type === "user_message" && !evt.steerId && typeof evt.text === "string" && evt.text.trim().length > 0) { + if ( + evt != null + && isParentUserMessage(evt) + && typeof evt.text === "string" + && evt.text.trim().length > 0 + ) { userEvent = evt; break; } diff --git a/apps/desktop/src/renderer/components/chat/chatTurnState.ts b/apps/desktop/src/renderer/components/chat/chatTurnState.ts index 09c4148a7..80e4a8d48 100644 --- a/apps/desktop/src/renderer/components/chat/chatTurnState.ts +++ b/apps/desktop/src/renderer/components/chat/chatTurnState.ts @@ -8,6 +8,13 @@ function chatEventEndsTurn(event: AgentChatEventEnvelope["event"]): boolean { return event.type === "done" || (event.type === "status" && event.turnStatus !== "started"); } +export function isParentUserMessage( + event: AgentChatEventEnvelope["event"], +): event is Extract { + return event.type === "user_message" + && (!event.steerId || Boolean(event.messageId?.trim())); +} + export function findUserMessageForTurn( events: AgentChatEventEnvelope[], turnId: string, @@ -25,8 +32,7 @@ export function findUserMessageForTurn( for (let index = turnAnchor; index >= 0; index -= 1) { const event = events[index]!.event; if ( - event.type === "user_message" - && !event.steerId + isParentUserMessage(event) && typeof event.text === "string" && event.text.trim().length > 0 && (!event.turnId || event.turnId === turnId)