diff --git a/bun.lock b/bun.lock index f0319f4ea..eaf91dd7e 100644 --- a/bun.lock +++ b/bun.lock @@ -250,6 +250,7 @@ "@types/bun": "catalog:", "@types/react": "catalog:", "@types/react-dom": "catalog:", + "linkedom": "catalog:", "postcss": "catalog:", }, }, diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 093d109e6..9382fe22c 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -394,6 +394,7 @@ - Fixed the eval tool's status-event tree truncating from the bottom: the newest `log()` progress lines were hidden behind an `… N more` marker while the oldest stayed visible; the tree now shows a tail window behind an `… N earlier` marker, and the expanded view widens to the viewport instead of a fixed 10 events - Fixed the `//!world=main` directive being silently ignored for string expressions passed to raw Puppeteer evaluation APIs - Fixed tab reuse issues where hung navigation or unhandled modals would cause initialization to stall and trigger a force-kill +- Fixed advisor context maintenance undercounting the provider context: the compaction decision now anchors on the advisor's provider-reported context usage (cached input + generated output) floored by a full local estimate that includes the advisor system prompt and tool schemas, rejects stale provider usage retained across advisor compaction, and recovers a provider overflow by clearing only the advisor's own context at the current primary cursor — retrying the bounded failing batch once against a fresh context without replaying old primary history and keeping later updates eligible ([#5282](https://github.com/can1357/oh-my-pi/issues/5282)) - Improved search reliability for Perplexity provider by forcing retrieval for all queries - Fixed JS eval cells losing top-level `function` and `var` declarations across cells when the defining cell contained top-level `await` — the async wrapper scoped them to the cell's IIFE instead of publishing them to the worker global diff --git a/packages/coding-agent/src/advisor/__tests__/advisor.test.ts b/packages/coding-agent/src/advisor/__tests__/advisor.test.ts index ad6cc6ef9..7cb593504 100644 --- a/packages/coding-agent/src/advisor/__tests__/advisor.test.ts +++ b/packages/coding-agent/src/advisor/__tests__/advisor.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from "bun:test"; import type { AgentMessage, AgentTelemetryConfig } from "@oh-my-pi/pi-agent-core"; import type { AssistantMessage } from "@oh-my-pi/pi-ai"; +import * as AIError from "@oh-my-pi/pi-ai/error"; import type { TUI } from "@oh-my-pi/pi-tui"; import { type } from "arktype"; import type { ModelRegistry } from "../../config/model-registry"; @@ -1599,7 +1600,7 @@ describe("advisor", () => { expect(promptInputs[1]).toContain("summary-bbb"); }); - it("triggers a re-prime and full replay when maintainContext returns true", async () => { + it("clears advisor context without replaying primary history when maintenance requests recovery", async () => { const promptInputs: string[] = []; const { promise: firstPromptDone, resolve: finishFirst } = Promise.withResolvers(); const { promise: secondPromptDone, resolve: finishSecond } = Promise.withResolvers(); @@ -1619,36 +1620,327 @@ describe("advisor", () => { state: { messages: [] }, }; const messages: AgentMessage[] = [{ role: "user", content: "aaa", timestamp: 1 } as AgentMessage]; - let shouldRePrime = false; + let shouldResetContext = false; const host: AdvisorRuntimeHost = { snapshotMessages: () => messages, enqueueAdvice: () => {}, maintainContext: async tokens => { expect(tokens).toBeGreaterThan(0); - return shouldRePrime; + return shouldResetContext; }, }; const runtime = new AdvisorRuntime(agent, host); - // First turn: normal incremental prompt. + runtime.onTurnEnd(messages); await firstPromptDone; expect(promptInputs).toHaveLength(1); expect(promptInputs[0]).toContain("aaa"); expect(resetCount).toBe(0); - // Second turn: maintainContext returns true → re-prime. - shouldRePrime = true; + shouldResetContext = true; messages.push({ role: "user", content: "bbb", timestamp: 2 } as AgentMessage); runtime.onTurnEnd(messages); await secondPromptDone; - // Full replay includes both aaa and bbb. + expect(promptInputs).toHaveLength(2); - expect(promptInputs[1]).toContain("aaa"); expect(promptInputs[1]).toContain("bbb"); + expect(promptInputs[1]).not.toContain("aaa"); expect(resetCount).toBe(1); }); + + it("preserves updates queued while async maintenance resets the advisor context", async () => { + const promptInputs: string[] = []; + let resetCount = 0; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + }, + abort: () => {}, + reset: () => { + resetCount++; + }, + state: { messages: [] }, + }; + const maintenanceStarted = Promise.withResolvers(); + const maintenanceFinished = Promise.withResolvers(); + let maintenanceCalls = 0; + const messages: AgentMessage[] = [{ role: "user", content: "bbb", timestamp: 1 } as AgentMessage]; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + maintainContext: async () => { + maintenanceCalls++; + if (maintenanceCalls !== 1) return false; + maintenanceStarted.resolve(); + return await maintenanceFinished.promise; + }, + }; + const runtime = new AdvisorRuntime(agent, host); + + runtime.onTurnEnd(messages); + await maintenanceStarted.promise; + messages.push({ role: "user", content: "ccc", timestamp: 2 } as AgentMessage); + runtime.onTurnEnd(messages); + maintenanceFinished.resolve(true); + await runtime.waitForCatchup(1000, 1); + + expect(promptInputs).toHaveLength(2); + expect(promptInputs[0]).toContain("bbb"); + expect(promptInputs[0]).not.toContain("ccc"); + expect(promptInputs[1]).toContain("ccc"); + expect(promptInputs[1]).not.toContain("bbb"); + expect(resetCount).toBe(1); + }); + + it("re-expands active primary context when maintenance clears advisor history", async () => { + const promptInputs: string[] = []; + const agent = makeAgent(promptInputs); + const planRule = + "Plan mode is active. You MUST remain read-only except for the approved plan file at local://PLAN.md."; + const messages: AgentMessage[] = [ + { role: "user", content: "aaa", timestamp: 1 } as AgentMessage, + { + role: "custom", + customType: "plan-mode-context", + content: planRule, + display: false, + timestamp: 2, + } as AgentMessage, + ]; + let shouldResetContext = false; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + maintainContext: async () => shouldResetContext, + }; + const runtime = new AdvisorRuntime(agent, host); + + runtime.onTurnEnd(messages); + await runtime.waitForCatchup(1000, 1); + expect(promptInputs[0]).toContain(planRule); + + shouldResetContext = true; + messages.push({ role: "user", content: "bbb", timestamp: 3 } as AgentMessage); + messages.push({ + role: "custom", + customType: "plan-mode-context", + content: planRule, + display: false, + timestamp: 4, + } as AgentMessage); + runtime.onTurnEnd(messages); + await runtime.waitForCatchup(1000, 1); + + expect(promptInputs).toHaveLength(2); + expect(promptInputs[1]).toContain("bbb"); + expect(promptInputs[1]).not.toContain("aaa"); + expect(promptInputs[1]).toContain(planRule); + expect(promptInputs[1]).not.toContain("unchanged — still in effect"); + }); + + it("recovers a provider overflow at the current cursor without replaying primary history", async () => { + const overflowMessage = "context_length_exceeded: Your input exceeds the context window of this model."; + const promptInputs: string[] = []; + const state: { messages: AgentMessage[]; error?: string } = { + messages: [{ role: "user", content: "existing advisor context", timestamp: 1 } as AgentMessage], + }; + let promptCalls = 0; + let resetCount = 0; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + promptCalls++; + state.error = promptCalls === 1 ? overflowMessage : undefined; + }, + abort: () => {}, + reset: () => { + resetCount++; + state.messages.length = 0; + state.error = undefined; + }, + state, + }; + const messages: AgentMessage[] = [ + { role: "user", content: "ancient-primary-one", timestamp: 1 } as AgentMessage, + { + role: "assistant", + content: [{ type: "text", text: "ancient-primary-two" }], + timestamp: 2, + } as AgentMessage, + ]; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }; + const runtime = new AdvisorRuntime(agent, host, 0); + runtime.seedTo(messages.length); + + messages.push({ role: "user", content: "overflowing-current-update", timestamp: 3 } as AgentMessage); + runtime.onTurnEnd(messages); + await runtime.waitForCatchup(1000, 1); + + expect(promptInputs).toHaveLength(2); + for (const input of promptInputs) { + expect(input).toContain("overflowing-current-update"); + expect(input).not.toContain("ancient-primary-one"); + expect(input).not.toContain("ancient-primary-two"); + } + expect(resetCount).toBe(1); + + messages.push({ role: "user", content: "post-recovery-update", timestamp: 4 } as AgentMessage); + runtime.onTurnEnd(messages); + await runtime.waitForCatchup(1000, 1); + + expect(promptInputs).toHaveLength(3); + expect(promptInputs[2]).toContain("post-recovery-update"); + expect(promptInputs[2]).not.toContain("overflowing-current-update"); + expect(promptInputs[2]).not.toContain("ancient-primary-one"); + expect(promptInputs[2]).not.toContain("ancient-primary-two"); + expect(resetCount).toBe(1); + }); + + it("classifies structured overflow metadata before rolling back the failed turn", async () => { + const promptInputs: string[] = []; + const state: { messages: AgentMessage[]; error?: string } = { + messages: [{ role: "user", content: "existing advisor context", timestamp: 1 } as AgentMessage], + }; + let promptCalls = 0; + let resetCount = 0; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + promptCalls++; + if (promptCalls !== 1) { + state.error = undefined; + return; + } + state.messages.push({ role: "user", content: input, timestamp: 2 } as AgentMessage); + const failure: AssistantMessage = { + role: "assistant", + content: [], + api: "openai-responses", + provider: "openai", + model: "structured-overflow-model", + usage: { + input: 1, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 1, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "error", + errorMessage: "opaque provider rejection", + errorStatus: 400, + errorId: AIError.create(AIError.Flag.ContextOverflow), + timestamp: 3, + }; + state.messages.push(failure); + state.error = "opaque provider rejection"; + }, + abort: () => {}, + reset: () => { + resetCount++; + state.messages.length = 0; + state.error = undefined; + }, + rollbackTo: count => { + state.messages.length = Math.min(count, state.messages.length); + state.error = undefined; + }, + state, + }; + const messages: AgentMessage[] = [{ role: "user", content: "ancient-primary", timestamp: 1 } as AgentMessage]; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + }; + const runtime = new AdvisorRuntime(agent, host, 0); + runtime.seedTo(messages.length); + + messages.push({ role: "user", content: "structured-current-update", timestamp: 2 } as AgentMessage); + runtime.onTurnEnd(messages); + await runtime.waitForCatchup(1000, 1); + + expect(promptInputs).toHaveLength(2); + for (const input of promptInputs) { + expect(input).toContain("structured-current-update"); + expect(input).not.toContain("ancient-primary"); + } + expect(resetCount).toBe(1); + }); + + it("drops only a double-overflowing batch and continues queued and later updates", async () => { + const overflowMessage = "context_length_exceeded: Your input exceeds the context window of this model."; + const promptInputs: string[] = []; + const failures: unknown[] = []; + const secondAttemptStarted = Promise.withResolvers(); + const finishSecondAttempt = Promise.withResolvers(); + const state: { messages: AgentMessage[]; error?: string } = { + messages: [{ role: "user", content: "existing advisor context", timestamp: 1 } as AgentMessage], + }; + let failingAttempts = 0; + const agent: AdvisorAgent = { + prompt: async input => { + promptInputs.push(input); + if (!input.includes("first-overflow")) { + state.error = undefined; + return; + } + failingAttempts++; + if (failingAttempts === 2) { + secondAttemptStarted.resolve(); + await finishSecondAttempt.promise; + } + state.error = overflowMessage; + }, + abort: () => {}, + reset: () => { + state.messages.length = 0; + state.error = undefined; + }, + state, + }; + const messages: AgentMessage[] = [{ role: "user", content: "ancient-history", timestamp: 1 } as AgentMessage]; + const host: AdvisorRuntimeHost = { + snapshotMessages: () => messages, + enqueueAdvice: () => {}, + notifyFailure: error => failures.push(error), + }; + const runtime = new AdvisorRuntime(agent, host, 0); + runtime.seedTo(messages.length); + + messages.push({ role: "user", content: "first-overflow", timestamp: 2 } as AgentMessage); + runtime.onTurnEnd(messages); + await secondAttemptStarted.promise; + + messages.push({ role: "user", content: "queued-small-update", timestamp: 3 } as AgentMessage); + runtime.onTurnEnd(messages); + finishSecondAttempt.resolve(); + await runtime.waitForCatchup(1000, 1); + + expect(failingAttempts).toBe(2); + expect(promptInputs).toHaveLength(3); + for (const input of promptInputs.slice(0, 2)) { + expect(input).toContain("first-overflow"); + expect(input).not.toContain("ancient-history"); + } + expect(promptInputs[2]).toContain("queued-small-update"); + expect(promptInputs[2]).not.toContain("first-overflow"); + expect(promptInputs[2]).not.toContain("ancient-history"); + expect(failures).toHaveLength(1); + expect(runtime.backlog).toBe(0); + + messages.push({ role: "user", content: "later-small-update", timestamp: 4 } as AgentMessage); + runtime.onTurnEnd(messages); + await runtime.waitForCatchup(1000, 1); + + expect(promptInputs).toHaveLength(4); + expect(promptInputs[3]).toContain("later-small-update"); + expect(promptInputs[3]).not.toContain("first-overflow"); + }); it("tracks backlog and blocks until caught up", async () => { const promptInputs: string[] = []; const { promise: promptStarted, resolve: startPrompt } = Promise.withResolvers(); diff --git a/packages/coding-agent/src/advisor/runtime.ts b/packages/coding-agent/src/advisor/runtime.ts index ec2155773..a3adf015e 100644 --- a/packages/coding-agent/src/advisor/runtime.ts +++ b/packages/coding-agent/src/advisor/runtime.ts @@ -1,6 +1,7 @@ import type { AgentMessage } from "@oh-my-pi/pi-agent-core"; import { estimateTokens } from "@oh-my-pi/pi-agent-core/compaction"; import type { AssistantMessage, ImageContent, TextContent } from "@oh-my-pi/pi-ai"; +import * as AIError from "@oh-my-pi/pi-ai/error"; import { logger } from "@oh-my-pi/pi-utils"; import { obfuscateToolArguments, type SecretObfuscator } from "../secrets/obfuscator"; import { formatSessionHistoryMarkdown, PRIMARY_CONTEXT_CUSTOM_TYPES } from "../session/session-history-format"; @@ -35,10 +36,10 @@ export interface AdvisorRuntimeHost { * Pre-prompt context maintenance for the advisor's own append-only context. * Promotes the advisor model to a larger sibling when its context nears the * window (mirroring the primary's promote-first policy) and resolves `true` - * when the advisor should re-prime — reset and replay the current - * primary-bounded transcript — because promotion did not free enough room. - * Optional: hosts that omit it get no maintenance (context only shrinks when - * the primary's next compaction triggers {@link AdvisorRuntime.reset}). + * when the advisor must clear its own context before sending the current + * incremental update. The cursor stays at the current primary position: this + * recovery path must never replay the full primary transcript. + * Optional: hosts that omit it get no proactive maintenance. */ maintainContext?(incomingTokens: number): Promise; /** @@ -184,9 +185,12 @@ const MAX_COALESCE_ROUNDS = 3; interface PendingDelta { text: string; + rawMessages: AgentMessage[]; + renderRevision: number; turns: number; /** Whether the primary was mid-turn (willContinue:true) when this delta was rendered. */ wip: boolean; + overflowRecovery?: boolean; } interface CatchupWaiter { @@ -224,6 +228,8 @@ export class AdvisorRuntime { * marker so the advisor isn't re-fed the full ~1k-token rules each turn. * Cleared on every re-prime/seed and when a failed batch is dropped. */ #seenContext = new Map(); + /** Incremented whenever the advisor loses context so queued raw deltas are re-rendered against fresh dedupe state. */ + #renderRevision = 0; #pending: PendingDelta[] = []; #busy = false; #backlog = 0; @@ -275,9 +281,9 @@ export class AdvisorRuntime { const all = messages ?? this.host.snapshotMessages(); this.#latestMessages = all; const wip = opts?.willContinue ?? false; - const render = this.#renderDelta(all, wip); - if (render) { - this.#pending.push({ text: render, turns: 1, wip }); + const rendered = this.#renderDelta(all, wip); + if (rendered) { + this.#pending.push({ ...rendered, turns: 1 }); this.#backlog++; this.#notifyWaiters(); void this.#drain(); @@ -317,19 +323,15 @@ export class AdvisorRuntime { } catch {} } - #resetAdvisorContext(clearBacklog: boolean, wakeWaiters: boolean): void { - this.#lastCount = 0; - this.#deliveredPrefix = []; - this.#pending = []; + #clearSeenContext(): void { + this.#seenContext.clear(); + this.#renderRevision++; + } + + #clearAdvisorContextAtCurrentCursor(): void { this.#consecutiveFailures = 0; this.#failureNotified = false; - this.#seenContext.clear(); - if (clearBacklog) { - this.#backlog = 0; - } - if (wakeWaiters) { - this.#wakeAllWaiters(); - } + this.#clearSeenContext(); try { this.agent.reset(); } catch {} @@ -338,6 +340,19 @@ export class AdvisorRuntime { } catch {} } + #resetAdvisorContext(clearBacklog: boolean, wakeWaiters: boolean): void { + this.#lastCount = 0; + this.#deliveredPrefix = []; + this.#pending = []; + this.#clearAdvisorContextAtCurrentCursor(); + if (clearBacklog) { + this.#backlog = 0; + } + if (wakeWaiters) { + this.#wakeAllWaiters(); + } + } + /** * Re-prime the advisor after a history rewrite (compaction, session * switch/resume, branch). Clears the advisor's own (non-persisted) context @@ -366,11 +381,30 @@ export class AdvisorRuntime { this.#backlog = 0; this.#consecutiveFailures = 0; this.#failureNotified = false; - this.#seenContext.clear(); + this.#clearSeenContext(); this.#wakeAllWaiters(); } - #renderDelta(messages?: AgentMessage[], wip = false): string | null { + #formatRawDelta(rawMessages: AgentMessage[], wip = false): string | null { + const delta = rawMessages + .filter(message => !(message.role === "custom" && message.customType === "advisor")) + .map(message => this.#dedupContextMessage(message)); + if (delta.length === 0) return null; + const obfuscator = this.host.obfuscator; + const formattedDelta = obfuscator?.hasSecrets() ? obfuscateAdvisorDelta(obfuscator, delta) : delta; + const md = formatSessionHistoryMarkdown(formattedDelta, { + includeThinking: true, + includeToolIntent: true, + watchedRoles: true, + expandPrimaryContext: true, + expandEditDiffs: true, + }); + if (!md.trim()) return null; + const heading = wip ? "### Session update [in progress — more steps follow]" : "### Session update"; + return `${heading}\n\n${md}`; + } + + #renderDelta(messages?: AgentMessage[], wip = false): Omit | null { const all = messages ?? this.#latestMessages ?? this.host.snapshotMessages(); let prefixChanged = all.length < this.#lastCount; for (let i = 0; !prefixChanged && i < this.#lastCount; i++) { @@ -396,29 +430,15 @@ export class AdvisorRuntime { this.#epoch++; this.#resetAdvisorContext(true, true); } - const delta = all - .slice(this.#lastCount) - .filter(m => !(m.role === "custom" && m.customType === "advisor")) - .map(m => this.#dedupContextMessage(m)); + const rawMessages = all.slice(this.#lastCount); for (let i = this.#lastCount; i < all.length; i++) { const message = all[i]; if (message === undefined) continue; this.#deliveredPrefix.push({ message, fingerprint: fingerprintMessage(message) }); } this.#lastCount = all.length; - if (delta.length === 0) return null; - const obfuscator = this.host.obfuscator; - const formattedDelta = obfuscator?.hasSecrets() ? obfuscateAdvisorDelta(obfuscator, delta) : delta; - const md = formatSessionHistoryMarkdown(formattedDelta, { - includeThinking: true, - includeToolIntent: true, - watchedRoles: true, - expandPrimaryContext: true, - expandEditDiffs: true, - }); - if (!md.trim()) return null; - const heading = wip ? "### Session update [in progress — more steps follow]" : "### Session update"; - return `${heading}\n\n${md}`; + const text = this.#formatRawDelta(rawMessages, wip); + return text ? { text, rawMessages, renderRevision: this.#renderRevision, wip } : null; } /** @@ -479,40 +499,50 @@ export class AdvisorRuntime { } /** - * Collect all currently pending deltas into one batch, running - * `maintainContext` for correct token budgeting. Loops until the pending - * queue is stable (no new deltas arrived during a maintenance check) or a - * reprime is triggered. Every `await` inside the loop has an epoch guard so - * a reset/dispose mid-await cannot leak a stale batch into the post-reset - * conversation. + * Collect the popped deltas into one batch, running `maintainContext` for + * correct token budgeting. Loops until the pending queue is stable (no new + * deltas arrived during a maintenance check) or the round cap is reached. + * Every `await` inside the loop has an epoch guard so a reset/dispose + * mid-await cannot leak a stale batch into the post-reset conversation. + * + * When maintenance requests recovery, only the advisor Agent/log is reset + * (at the current primary cursor) and the already-collected raw batch is + * re-rendered — older, already-delivered primary transcript is never + * replayed. * * The coalescing loop is capped at {@link MAX_COALESCE_ROUNDS} iterations so * a pathologically fast primary combined with a slow `maintainContext` cannot * stall dispatch indefinitely — any items still in `#pending` after the cap - * are left for the next drain iteration. + * are left for the next drain iteration. Overflow-recovery batches skip + * coalescing entirely: they retry exactly the bounded batch that overflowed. * * Returns `null` when the epoch was invalidated — caller should `continue`. - * Returns `{ batch: null, finalTurns }` when there is nothing to render but - * backlog still needs to be decremented. */ async #collectAndMaintainBatch( epoch: number, - ): Promise<{ batch: string | null; finalTurns: number; wip: boolean } | null> { - const initial = this.#pending.splice(0); + initial: PendingDelta[], + recoveringOverflow: boolean, + ): Promise<{ + batch: string | null; + rawMessages: AgentMessage[]; + finalTurns: number; + wip: boolean; + resetContext: boolean; + } | null> { let batchText = initial.map(b => b.text).join("\n\n"); + let rawMessages = initial.flatMap(b => b.rawMessages); let turns = initial.reduce((sum, b) => sum + b.turns, 0); - // Track WIP state of the most recent delta — forwarded to the reprime - // #renderDelta so a willContinue:true turn keeps its [in progress] heading - // even when the full transcript is replayed from scratch. Also returned to - // #drain so the retry-requeue path preserves it on failed turns. + // Track WIP state of the most recent delta — forwarded to the re-render + // so a willContinue:true turn keeps its [in progress] heading. Also + // returned to #drain so the retry-requeue path preserves it on failed turns. let wip = initial.at(-1)?.wip ?? false; for (let round = 0; round < MAX_COALESCE_ROUNDS; round++) { if (this.host.maintainContext) { const incomingTokens = estimateTokens({ role: "user", content: batchText, timestamp: Date.now() }); - let shouldReprime = false; + let shouldResetContext = false; try { - shouldReprime = await this.host.maintainContext(incomingTokens); + shouldResetContext = await this.host.maintainContext(incomingTokens); } catch (err) { logger.debug("advisor context maintenance failed", { err: String(err) }); } @@ -520,18 +550,35 @@ export class AdvisorRuntime { // invalidates this batch. if (this.#epoch !== epoch) return null; - if (shouldReprime) { - // Tally deltas that arrived during this await before #resetAdvisorContext - // wipes #pending, so finalTurns stays accurate for backlog accounting. - // Also capture the latest WIP state before the queue is cleared. - const lateItems = this.#pending.splice(0); - turns += lateItems.reduce((sum, b) => sum + b.turns, 0); - if (lateItems.length > 0) wip = lateItems.at(-1)!.wip; - this.#resetAdvisorContext(false, false); - return { batch: this.#renderDelta(this.#latestMessages, wip), finalTurns: turns, wip }; + if (shouldResetContext) { + // Once coalescing has begun (round > 0), deltas that arrived during + // this await are part of the coalescing window: tally them so + // finalTurns stays accurate for backlog accounting and their raw + // messages join the bounded re-render. On the initial round the + // popped batch stays bounded exactly as dispatched — later arrivals + // remain queued and ship as their own subsequent batch. + if (round > 0) { + const lateItems = this.#pending.splice(0); + turns += lateItems.reduce((sum, b) => sum + b.turns, 0); + if (lateItems.length > 0) { + wip = lateItems.at(-1)!.wip; + rawMessages = rawMessages.concat(lateItems.flatMap(b => b.rawMessages)); + } + } + // Reset only the advisor Agent/log. The primary cursor, backlog, + // waiters, latest snapshot, and epoch stay untouched. Re-render only + // this already-popped raw batch so active plan/reference bodies are + // restored without replaying any older primary transcript. + this.#clearAdvisorContextAtCurrentCursor(); + const rerendered = this.#formatRawDelta(rawMessages, wip); + return { batch: rerendered ?? (batchText || null), rawMessages, finalTurns: turns, wip, resetContext: true }; } } + // Overflow-recovery batches retry exactly the bounded batch that + // overflowed; pending updates stay queued behind them. + if (recoveringOverflow) break; + // On the final round stop here — any late arrivals would ship without // a subsequent maintainContext budget check. Leave them in #pending for // the next drain iteration where they will be properly budgeted. @@ -543,11 +590,31 @@ export class AdvisorRuntime { const late = this.#pending.splice(0); if (late.length === 0) break; batchText = [batchText, ...late.map(b => b.text)].join("\n\n"); + rawMessages = rawMessages.concat(late.flatMap(b => b.rawMessages)); turns += late.reduce((sum, b) => sum + b.turns, 0); wip = late.at(-1)!.wip; } - return { batch: batchText || null, finalTurns: turns, wip }; + return { batch: batchText || null, rawMessages, finalTurns: turns, wip, resetContext: false }; + } + + #terminalAssistantFailure(snapshot: number): AssistantMessage | undefined { + const messages = this.agent.state.messages; + for (let i = messages.length - 1; i >= snapshot; i--) { + const message = messages[i]; + if (message.role === "assistant" && message.stopReason === "error") return message; + } + return undefined; + } + + #notifyFailureOnce(error: unknown): void { + if (this.#failureNotified) return; + this.#failureNotified = true; + try { + this.host.notifyFailure?.(error); + } catch (notifyErr) { + logger.warn("advisor failure notification failed", { err: String(notifyErr) }); + } } async #drain(): Promise { @@ -555,13 +622,28 @@ export class AdvisorRuntime { this.#busy = true; try { while (!this.disposed && this.#pending.length) { + let popped: PendingDelta[]; + if (this.#pending[0]?.overflowRecovery) { + const recovery = this.#pending.shift(); + if (!recovery) continue; + popped = [recovery]; + } else { + popped = this.#pending.splice(0); + } const epoch = this.#epoch; - const result = await this.#collectAndMaintainBatch(epoch); + for (const delta of popped) { + if (delta.renderRevision === this.#renderRevision) continue; + const refreshed = this.#formatRawDelta(delta.rawMessages, delta.wip); + if (refreshed) delta.text = refreshed; + delta.renderRevision = this.#renderRevision; + } + const recoveringOverflow = popped.some(delta => delta.overflowRecovery === true); + const result = await this.#collectAndMaintainBatch(epoch, popped, recoveringOverflow); // Epoch was invalidated during batch collection; restart the loop. if (result === null) continue; - const { batch, finalTurns, wip } = result; + const { batch, rawMessages, finalTurns, wip, resetContext } = result; if (this.disposed || batch === null) { this.#backlog = Math.max(0, this.#backlog - finalTurns); @@ -576,6 +658,7 @@ export class AdvisorRuntime { // batch on top of stale turns and the dropped-after-3 path would leak // orphan failures into the next successful run's context. const messageSnapshot = this.agent.state.messages.length; + const contextWasFresh = resetContext || recoveringOverflow || messageSnapshot === 0; try { // Reset the host's per-update advisor state (one-advise-per-update // gate) before each model cycle so the new batch starts fresh. @@ -608,6 +691,11 @@ export class AdvisorRuntime { // reset, not a transient failure — drop the stale batch. if (this.#epoch !== epoch) continue; const failedMessages = this.agent.state.messages.slice(messageSnapshot); + const terminalFailure = this.#terminalAssistantFailure(messageSnapshot); + const contextOverflow = + (terminalFailure !== undefined && + AIError.is(AIError.classifyMessage(terminalFailure), AIError.Flag.ContextOverflow)) || + AIError.is(AIError.classify(err), AIError.Flag.ContextOverflow); this.#rollbackFailedTurn(messageSnapshot); logger.debug("advisor turn failed", { err: String(err) }); let recovered = false; @@ -629,29 +717,61 @@ export class AdvisorRuntime { if (recovered) { this.#consecutiveFailures = 0; this.#failureNotified = false; - this.#pending.unshift({ text: batch, turns: finalTurns, wip }); + this.#pending.unshift({ + text: batch, + rawMessages, + renderRevision: this.#renderRevision, + turns: finalTurns, + wip, + overflowRecovery: recoveringOverflow || undefined, + }); continue; } - this.#consecutiveFailures++; - if (this.#consecutiveFailures >= 3) { - logger.warn("advisor failed consecutively 3 times; dropping backlog to prevent stall"); - if (!this.#failureNotified) { - this.#failureNotified = true; - try { - this.host.notifyFailure?.(err); - } catch (notifyErr) { - logger.warn("advisor failure notification failed", { err: String(notifyErr) }); - } + if (contextOverflow) { + this.#clearAdvisorContextAtCurrentCursor(); + if (contextWasFresh) { + // The bounded update cannot fit even with no advisor history. Drop + // only this batch after its one fresh-context retry; pending and later + // deltas remain eligible so one oversized update cannot disable the advisor. + logger.warn("advisor update overflowed a fresh context; dropping bounded batch"); + this.#notifyFailureOnce(err); + success = true; + } else { + // Retry once against the fresh advisor context, using only the same + // bounded raw batch. Pending updates remain queued behind it. + const recoveryBatch = this.#formatRawDelta(rawMessages, wip) ?? batch; + this.#pending.unshift({ + text: recoveryBatch, + rawMessages, + renderRevision: this.#renderRevision, + turns: finalTurns, + wip, + overflowRecovery: true, + }); + logger.debug("advisor context overflow recovered at current primary cursor") } - this.#consecutiveFailures = 0; - // Drop the seen-context so the next turn re-expands primary-context - // prompts instead of marking them "unchanged" against content the - // advisor never received. - this.#seenContext.clear(); - success = true; + } else { - this.#pending.unshift({ text: batch, turns: finalTurns, wip }); - await Bun.sleep(this.retryDelayMs); + this.#consecutiveFailures++; + if (this.#consecutiveFailures >= 3) { + logger.warn("advisor failed consecutively 3 times; dropping backlog to prevent stall"); + this.#notifyFailureOnce(err); + this.#consecutiveFailures = 0; + // The dropped batch may carry primary-context we never delivered; drop + // the seen-state too so queued raw deltas re-expand before delivery. + this.#clearSeenContext(); + success = true; + } else { + this.#pending.unshift({ + text: batch, + rawMessages, + renderRevision: this.#renderRevision, + turns: finalTurns, + wip, + overflowRecovery: recoveringOverflow || undefined, + }); + await Bun.sleep(this.retryDelayMs); + } } } diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 51715d4a9..9403303de 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -254,7 +254,11 @@ import { containsOrchestrate, ORCHESTRATE_NOTICE } from "../modes/orchestrate"; import { theme } from "../modes/theme/theme"; import { parseTurnBudget } from "../modes/turn-budget"; import { containsUltrathink, ULTRATHINK_NOTICE } from "../modes/ultrathink"; -import { computeNonMessageBreakdown, computeNonMessageTokens } from "../modes/utils/context-usage"; +import { + computeNonMessageBreakdown, + computeNonMessageTokens, + estimateToolSchemaTokens, +} from "../modes/utils/context-usage"; import { containsWorkflow, renderWorkflowNotice } from "../modes/workflow"; import { resolveApprovedPlan } from "../plan-mode/approved-plan"; import { createPlanReadMatcher } from "../plan-mode/plan-protection"; @@ -1234,6 +1238,13 @@ interface ActiveAdvisor { signature: string; } +/** Runtime-only advisor compaction metadata. It never enters the model-facing summary text. */ +interface AdvisorCompactionSummaryMessage extends CompactionSummaryMessage { + firstKeptEntryId?: string; + /** First message index eligible to anchor provider usage after this compaction. */ + advisorUsageAnchorStartIndex?: number; +} + /** Resolved advisor config ready to instantiate as an {@link ActiveAdvisor}. */ interface AdvisorRuntimeDescriptor { config: AdvisorConfig; @@ -3571,10 +3582,23 @@ export class AgentSession { if (contextWindow <= 0) return false; const messages = agent.state.messages; - let contextTokens = incomingTokens; + const estimateOptions = { excludeEncryptedReasoning: true } as const; + let storedConversationTokens = 0; for (const message of messages) { - contextTokens += estimateTokens(message); + storedConversationTokens += estimateTokens(message, estimateOptions); } + // Provider usage (including cache reads and generated output) is the + // trustworthy anchor for accumulated context. Add only the trailing incoming + // delta to that arm. Floor it by a full local estimate — fixed advisor system + // prompt, tool schemas, stored messages, and incoming delta — so provider + // under-reporting or payload transforms cannot suppress maintenance. + const providerContextTokens = this.#estimateAdvisorContextTokens(messages) + incomingTokens; + const localContextTokens = + countTokens(agent.state.systemPrompt) + + estimateToolSchemaTokens(agent.state.tools) + + storedConversationTokens + + incomingTokens; + const contextTokens = compactionContextTokens(providerContextTokens, localContextTokens); if (!shouldCompact(contextTokens, contextWindow, compactionSettings)) { return false; @@ -3598,6 +3622,7 @@ export class AgentSession { const timestamp = String(message.timestamp || Date.now()); if (message.role === "compactionSummary") { + const advisorSummary = message as AdvisorCompactionSummaryMessage; return { type: "compaction", id, @@ -3605,9 +3630,7 @@ export class AgentSession { timestamp, summary: message.summary, shortSummary: message.shortSummary, - firstKeptEntryId: - (message as CompactionSummaryMessage & { firstKeptEntryId?: string }).firstKeptEntryId || - `msg-${i + 1}`, + firstKeptEntryId: advisorSummary.firstKeptEntryId || `msg-${i + 1}`, tokensBefore: message.tokensBefore, } satisfies CompactionEntry; } @@ -3701,11 +3724,15 @@ export class AgentSession { const firstKeptEntryId = compactResult.firstKeptEntryId; const tokensBefore = compactResult.tokensBefore; - // Rebuild messages with the compaction summary + // The retained messages still carry provider usage from before this + // compaction. Record their exact array boundary on the in-memory summary so + // only assistants appended afterward can become the next usage anchor. + const advisorUsageAnchorStartIndex = preparation.recentMessages.length + 1; const summaryMessage = { ...createCompactionSummaryMessage(summary, tokensBefore, new Date().toISOString(), shortSummary), firstKeptEntryId, - } as CompactionSummaryMessage & { firstKeptEntryId?: string }; + advisorUsageAnchorStartIndex, + } satisfies AdvisorCompactionSummaryMessage; agent.replaceMessages([summaryMessage, ...preparation.recentMessages]); return false; @@ -17588,37 +17615,51 @@ this.#titleGenerationAbortController.abort(); } /** - * Estimate the advisor's current context tokens. When the advisor has a - * recent non-aborted assistant message with usage, use that prompt's token - * count and add a trailing estimate for messages after it. Otherwise estimate - * every message. + * Estimate the advisor's current context tokens. A successful provider usage + * after the latest advisor compaction is ground truth for the prompt plus its + * generated output; only messages after that anchor are estimated. Usage from + * retained pre-compaction messages is stale and must not immediately retrigger + * maintenance on the newly compacted context. */ #estimateAdvisorContextTokens(messages: AgentMessage[]): number { - let lastUsageIndex: number | null = null; - let lastUsage: AssistantMessage["usage"] | undefined; + let usageAnchorStartIndex = 0; for (let i = messages.length - 1; i >= 0; i--) { - const msg = messages[i]; - if (msg.role === "assistant") { - const assistantMsg = msg as AssistantMessage; - if (assistantMsg.stopReason !== "aborted" && assistantMsg.stopReason !== "error" && assistantMsg.usage) { - lastUsage = assistantMsg.usage; - lastUsageIndex = i; - break; - } + const message = messages[i]; + if (message.role !== "compactionSummary") continue; + const advisorSummary = message as AdvisorCompactionSummaryMessage; + // Advisor summaries created before this runtime-only boundary existed have + // no trustworthy way to distinguish retained from newly appended messages. + // Conservatively ignore every current assistant until the next compaction. + usageAnchorStartIndex = advisorSummary.advisorUsageAnchorStartIndex ?? messages.length; + break; + } + + let lastUsageIndex: number | undefined; + let lastUsage: AssistantMessage["usage"] | undefined; + for (let i = messages.length - 1; i >= usageAnchorStartIndex; i--) { + const message = messages[i]; + if (message.role !== "assistant") continue; + const assistant = message as AssistantMessage; + if (assistant.stopReason !== "aborted" && assistant.stopReason !== "error" && assistant.usage) { + lastUsage = assistant.usage; + lastUsageIndex = i; + break; } } - if (!lastUsage || lastUsageIndex === null) { + + const estimateOptions = { excludeEncryptedReasoning: true } as const; + if (!lastUsage || lastUsageIndex === undefined) { let estimated = 0; for (const message of messages) { - estimated += estimateTokens(message); + estimated += estimateTokens(message, estimateOptions); } return estimated; } let trailingTokens = 0; for (let i = lastUsageIndex + 1; i < messages.length; i++) { - trailingTokens += estimateTokens(messages[i]); + trailingTokens += estimateTokens(messages[i], estimateOptions); } - return calculatePromptTokens(lastUsage) + trailingTokens; + return calculateContextTokens(lastUsage) + trailingTokens; } /** diff --git a/packages/coding-agent/test/advisor-context-maintenance.test.ts b/packages/coding-agent/test/advisor-context-maintenance.test.ts new file mode 100644 index 000000000..c121501ec --- /dev/null +++ b/packages/coding-agent/test/advisor-context-maintenance.test.ts @@ -0,0 +1,199 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test"; +import { Agent, type AgentMessage, type CompactionSummaryMessage, countTokens } from "@oh-my-pi/pi-agent-core"; +import { calculateContextTokens, estimateTokens, resolveThresholdTokens } from "@oh-my-pi/pi-agent-core/compaction"; +import type { AssistantMessage } from "@oh-my-pi/pi-ai"; +import { createMockModel, type MockModel } from "@oh-my-pi/pi-ai/providers/mock"; +import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry"; +import { Settings } from "@oh-my-pi/pi-coding-agent/config/settings"; +import { estimateToolSchemaTokens } from "@oh-my-pi/pi-coding-agent/modes/utils/context-usage"; +import { AgentSession } from "@oh-my-pi/pi-coding-agent/session/agent-session"; +import { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage"; +import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager"; +import { TempDir } from "@oh-my-pi/pi-utils"; + +const CONTEXT_WINDOW = 372_000; +const CACHE_READ_TOKENS = 371_200; +const INPUT_TOKENS = 200; +const OUTPUT_TOKENS = 150; + +interface MaintenanceHarness { + advisor: Agent; + advisorMock: MockModel; + settings: Settings; +} + +interface AdvisorCompactionSummaryFixture extends CompactionSummaryMessage { + advisorUsageAnchorStartIndex?: number; +} + +describe("AgentSession advisor context maintenance", () => { + let tempDir: TempDir; + let authStorage: AuthStorage; + let session: AgentSession; + + beforeEach(async () => { + tempDir = TempDir.createSync("@pi-advisor-context-maintenance-"); + authStorage = await AuthStorage.create(tempDir.join("auth.db")); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + await session?.dispose(); + authStorage.close(); + await tempDir.remove(); + }); + + function createHarness(): MaintenanceHarness { + const primaryMock = createMockModel({ + provider: "anthropic", + responses: [{ content: ["primary complete"] }], + }); + const advisorMock = createMockModel({ + provider: "anthropic", + contextWindow: CONTEXT_WINDOW, + responses: [{ content: ["advisor reviewed current update"] }], + }); + const modelRegistry = new ModelRegistry(authStorage, tempDir.join("models.yml")); + const settings = Settings.isolated({ + "advisor.syncBacklog": "1", + "compaction.enabled": true, + "compaction.strategy": "context-full", + "contextPromotion.enabled": false, + }); + const agent = new Agent({ + getApiKey: () => "test-key", + initialState: { model: primaryMock, systemPrompt: [], tools: [] }, + streamFn: primaryMock.stream, + }); + session = new AgentSession({ + agent, + sessionManager: SessionManager.inMemory(), + settings, + modelRegistry, + advisorTools: [], + advisorStreamFn: advisorMock.stream, + }); + settings.setModelRole("advisor", "anthropic/claude-sonnet-4-5"); + expect(session.setAdvisorEnabled(true)).toBe(true); + const advisor = session.getAdvisorAgent(); + if (!advisor) throw new Error("Expected advisor agent to be active"); + advisor.setModel(advisorMock); + + // Keep maintenance on the no-summary recovery branch without blocking the + // primary prompt's own credential preflight. + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async model => + model === primaryMock ? "test-key" : undefined, + ); + return { advisor, advisorMock, settings }; + } + + function usageAnchor(advisorMock: MockModel, timestamp: number): AssistantMessage { + return { + role: "assistant", + content: [{ type: "text", text: "prior advisor output" }], + api: advisorMock.api, + provider: advisorMock.provider, + model: advisorMock.id, + usage: { + input: INPUT_TOKENS, + output: OUTPUT_TOKENS, + cacheRead: CACHE_READ_TOKENS, + cacheWrite: 0, + totalTokens: CACHE_READ_TOKENS + INPUT_TOKENS + OUTPUT_TOKENS, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp, + }; + } + + function compactionSummary(timestamp: number): AdvisorCompactionSummaryFixture { + return { + role: "compactionSummary", + summary: "bounded advisor summary", + tokensBefore: CACHE_READ_TOKENS + INPUT_TOKENS + OUTPUT_TOKENS, + timestamp, + // `[summary, retained]` is the compacted array; index 2 is the first + // position eligible for a newly appended provider-usage anchor. + advisorUsageAnchorStartIndex: 2, + }; + } + + it("maintains a 371,200-token cached advisor context before the 372,000-token window", async () => { + const { advisor, advisorMock, settings } = createHarness(); + const anchor = usageAnchor(advisorMock, Date.now() - 1_000); + advisor.state.messages.push(anchor); + + await session.prompt("small current update"); + + expect(advisorMock.calls).toHaveLength(1); + const advisorCall = advisorMock.calls[0]; + const update = advisorCall.context.messages.find(message => message.role === "user"); + if (!update) throw new Error("Expected the advisor's incremental update"); + const threshold = resolveThresholdTokens(CONTEXT_WINDOW, settings.getGroup("compaction")); + const providerAndUpdateTokens = calculateContextTokens(anchor.usage) + estimateTokens(update as AgentMessage); + expect(calculateContextTokens(anchor.usage)).toBe(CACHE_READ_TOKENS + INPUT_TOKENS + OUTPUT_TOKENS); + expect(providerAndUpdateTokens).toBeGreaterThan(threshold); + + // Provider usage triggers maintenance, but recovery sends only the bounded + // current update into the reset advisor context. + expect(JSON.stringify(advisorCall.context.messages)).toContain("small current update"); + expect(JSON.stringify(advisor.state.messages)).not.toContain("prior advisor output"); + }); + + it("includes advisor system prompt and tool schemas in the local maintenance floor", async () => { + const { advisor, advisorMock, settings } = createHarness(); + const seed: AgentMessage = { role: "user", content: "small stored advisor message", timestamp: 1 }; + advisor.state.messages.push(seed); + const storedTokens = estimateTokens(seed, { excludeEncryptedReasoning: true }); + const fixedPrefixTokens = countTokens(advisor.state.systemPrompt) + estimateToolSchemaTokens(advisor.state.tools); + const threshold = storedTokens + Math.floor(fixedPrefixTokens / 2); + settings.set("compaction.thresholdTokens", threshold); + + await session.prompt("tiny local-floor update"); + + const advisorCall = advisorMock.calls[0]; + const update = advisorCall.context.messages.find(message => message.role === "user"); + if (!update) throw new Error("Expected the advisor's incremental update"); + const messagesOnlyTokens = storedTokens + estimateTokens(update as AgentMessage); + expect(messagesOnlyTokens).toBeLessThan(threshold); + expect(messagesOnlyTokens + fixedPrefixTokens).toBeGreaterThan(threshold); + expect(JSON.stringify(advisor.state.messages)).not.toContain("small stored advisor message"); + }); + + it("ignores retained provider usage that predates the latest advisor compaction", async () => { + const { advisor, advisorMock } = createHarness(); + const compactedAt = Date.now(); + const summary = compactionSummary(compactedAt); + const retained = usageAnchor(advisorMock, compactedAt); + retained.content = [{ type: "text", text: "retained pre-compaction output" }]; + advisor.state.messages.push(summary, retained); + + await session.prompt("post-compaction update"); + + expect(advisorMock.calls).toHaveLength(1); + const sentContext = JSON.stringify(advisorMock.calls[0].context.messages); + expect(sentContext).toContain("retained pre-compaction output"); + expect(sentContext).toContain("post-compaction update"); + }); + + it("accepts equal-timestamp usage appended after the explicit compaction boundary", async () => { + const { advisor, advisorMock } = createHarness(); + const compactedAt = Date.now(); + const summary = compactionSummary(compactedAt); + const retained = usageAnchor(advisorMock, compactedAt); + retained.content = [{ type: "text", text: "retained pre-compaction output" }]; + const fresh = usageAnchor(advisorMock, compactedAt); + fresh.content = [{ type: "text", text: "fresh post-compaction output" }]; + advisor.state.messages.push(summary, retained, fresh); + + await session.prompt("equal-timestamp post-compaction update"); + + expect(advisorMock.calls).toHaveLength(1); + const sentContext = JSON.stringify(advisorMock.calls[0].context.messages); + expect(sentContext).toContain("equal-timestamp post-compaction update"); + expect(sentContext).not.toContain("retained pre-compaction output"); + expect(sentContext).not.toContain("fresh post-compaction output"); + }); +}); diff --git a/packages/stats/CHANGELOG.md b/packages/stats/CHANGELOG.md index 7cf491436..d9c35128b 100644 --- a/packages/stats/CHANGELOG.md +++ b/packages/stats/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Recent Errors now honors the selected dashboard time range before returning the newest 50 failures ([#5282](https://github.com/can1357/oh-my-pi/issues/5282)) + ## [16.4.7] - 2026-07-12 ### Fixed diff --git a/packages/stats/package.json b/packages/stats/package.json index 26482c5cb..a5f093772 100644 --- a/packages/stats/package.json +++ b/packages/stats/package.json @@ -55,6 +55,7 @@ "@types/bun": "catalog:", "@types/react": "catalog:", "@types/react-dom": "catalog:", + "linkedom": "catalog:", "postcss": "catalog:" }, "engines": { diff --git a/packages/stats/src/aggregator.ts b/packages/stats/src/aggregator.ts index 553033314..fb37e9305 100644 --- a/packages/stats/src/aggregator.ts +++ b/packages/stats/src/aggregator.ts @@ -444,9 +444,10 @@ export async function getRecentRequests(limit?: number): Promise return dbGetRecentRequests(limit); } -export async function getRecentErrors(limit?: number): Promise { +export async function getRecentErrors(range?: string | null, limit?: number): Promise { await initDb(); - return dbGetRecentErrors(limit); + const { cutoff } = getTimeRangeConfig(range); + return dbGetRecentErrors(limit, cutoff); } export async function getRequestDetails(id: number): Promise { diff --git a/packages/stats/src/client/api.ts b/packages/stats/src/client/api.ts index eef020b20..bf7985073 100644 --- a/packages/stats/src/client/api.ts +++ b/packages/stats/src/client/api.ts @@ -59,8 +59,14 @@ export async function getRecentRequests(limit = 50, signal?: AbortSignal): Promi return fetchJson(`${API_BASE}/stats/recent?limit=${limit}`, { signal }); } -export async function getRecentErrors(limit = 50, signal?: AbortSignal): Promise { - return fetchJson(`${API_BASE}/stats/errors?limit=${limit}`, { signal }); +export async function getRecentErrors( + range: TimeRange = "24h", + limit = 50, + signal?: AbortSignal, +): Promise { + return fetchJson(`${API_BASE}/stats/errors?range=${encodeURIComponent(range)}&limit=${limit}`, { + signal, + }); } export async function getRequestDetails(id: number, signal?: AbortSignal): Promise { diff --git a/packages/stats/src/client/routes/ErrorsRoute.tsx b/packages/stats/src/client/routes/ErrorsRoute.tsx index eb469bcca..273e531af 100644 --- a/packages/stats/src/client/routes/ErrorsRoute.tsx +++ b/packages/stats/src/client/routes/ErrorsRoute.tsx @@ -12,12 +12,12 @@ export interface ErrorsRouteProps { onRequestClick: (id: number) => void; } -export function ErrorsRoute({ active, refreshTrigger, onRequestClick }: ErrorsRouteProps) { +export function ErrorsRoute({ active, range, refreshTrigger, onRequestClick }: ErrorsRouteProps) { const { data: recentErrors, error, loading, - } = useResource(["recent-errors-dense", refreshTrigger], signal => getRecentErrors(50, signal), { + } = useResource(["recent-errors-dense", range, refreshTrigger], signal => getRecentErrors(range, 50, signal), { pollMs: 30000, enabled: active, }); diff --git a/packages/stats/src/db.ts b/packages/stats/src/db.ts index e913989a3..7f24d1e72 100644 --- a/packages/stats/src/db.ts +++ b/packages/stats/src/db.ts @@ -832,15 +832,18 @@ export function getRecentRequests(limit = 100): MessageStats[] { return (stmt.all(limit) as any[]).map(rowToMessageStats); } -export function getRecentErrors(limit = 100): MessageStats[] { +export function getRecentErrors(limit = 100, cutoff?: number | null): MessageStats[] { if (!db) return []; + const hasCutoff = cutoff !== undefined && cutoff !== null; const stmt = db.prepare(` - SELECT * FROM messages + SELECT * FROM messages WHERE stop_reason = 'error' - ORDER BY timestamp DESC + ${hasCutoff ? "AND timestamp >= ?" : ""} + ORDER BY timestamp DESC LIMIT ? `); - return (stmt.all(limit) as any[]).map(rowToMessageStats); + const rows = hasCutoff ? stmt.all(cutoff, limit) : stmt.all(limit); + return rows.map(rowToMessageStats); } export function getMessageById(id: number): MessageStats | null { diff --git a/packages/stats/src/server.ts b/packages/stats/src/server.ts index 17fc9fc40..607de3f88 100644 --- a/packages/stats/src/server.ts +++ b/packages/stats/src/server.ts @@ -184,7 +184,7 @@ const ensureClientBuild = async () => { /** * Handle API requests. */ -async function handleApi(req: Request): Promise { +export async function handleApi(req: Request): Promise { const url = new URL(req.url); const path = url.pathname; @@ -229,7 +229,7 @@ async function handleApi(req: Request): Promise { if (path === "/api/stats/errors") { const limit = url.searchParams.get("limit"); - const stats = await getRecentErrors(limit ? parseInt(limit, 10) : undefined); + const stats = await getRecentErrors(range, limit ? parseInt(limit, 10) : undefined); return Response.json(stats); } diff --git a/packages/stats/test/errors-range.test.ts b/packages/stats/test/errors-range.test.ts new file mode 100644 index 000000000..ce341c166 --- /dev/null +++ b/packages/stats/test/errors-range.test.ts @@ -0,0 +1,79 @@ +import { describe, expect, it } from "bun:test"; +import { initDb, insertMessageStats } from "../src/db"; +import { handleApi } from "../src/server"; +import type { MessageStats } from "../src/types"; +import { installStatsTestIsolation } from "./helpers/temp-agent"; + +const HOUR_MS = 60 * 60 * 1000; + +installStatsTestIsolation("@pi-stats-errors-range-"); + +function makeError(timestamp: number, entryId: string): MessageStats { + return { + sessionFile: "/tmp/errors-range-session.jsonl", + entryId, + folder: "/tmp/project", + model: "gpt-5.4", + provider: "openai-codex", + api: "openai-codex-responses", + timestamp, + duration: 1000, + ttft: 100, + stopReason: "error", + errorMessage: `failure ${entryId}`, + usage: { + input: 1000, + output: 500, + cacheRead: 200, + cacheWrite: 0, + totalTokens: 1700, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, + }, + agentType: "main", + }; +} + +async function readMessages(response: Response): Promise { + expect(response.status).toBe(200); + return response.json() as Promise; +} + +describe("Recent Errors range", () => { + it("filters by the mapped range before returning the newest 50 errors", async () => { + await initDb(); + const now = Date.now(); + const recentErrors = Array.from({ length: 50 }, (_, index) => makeError(now - index * 1000, `recent-${index}`)); + const oldError = makeError(now - 48 * HOUR_MS, "outside-24h"); + insertMessageStats([...recentErrors, oldError]); + + const dayErrors = await readMessages( + await handleApi(new Request("http://stats.test/api/stats/errors?range=24h&limit=50")), + ); + expect(dayErrors).toHaveLength(50); + expect(dayErrors.map(error => error.entryId)).toEqual(recentErrors.map(error => error.entryId)); + expect(dayErrors.some(error => error.entryId === oldError.entryId)).toBe(false); + + const allErrors = await readMessages( + await handleApi(new Request("http://stats.test/api/stats/errors?range=all&limit=51")), + ); + expect(allErrors).toHaveLength(51); + expect(allErrors.at(-1)?.entryId).toBe(oldError.entryId); + + const defaultErrors = await readMessages( + await handleApi(new Request("http://stats.test/api/stats/errors?limit=51")), + ); + expect(defaultErrors).toHaveLength(50); + expect(defaultErrors.some(error => error.entryId === oldError.entryId)).toBe(false); + + const fallbackErrors = await readMessages( + await handleApi(new Request("http://stats.test/api/stats/errors?range=unknown&limit=51")), + ); + expect(fallbackErrors.map(error => error.entryId)).toEqual(defaultErrors.map(error => error.entryId)); + }); +}); diff --git a/packages/stats/test/errors-route-range.test.tsx b/packages/stats/test/errors-route-range.test.tsx new file mode 100644 index 000000000..05d168f22 --- /dev/null +++ b/packages/stats/test/errors-route-range.test.tsx @@ -0,0 +1,81 @@ +import { afterEach, describe, expect, it, vi } from "bun:test"; +import { parseHTML } from "linkedom"; +import { act } from "react"; +import { createRoot, type Root } from "react-dom/client"; +import { ErrorsRoute } from "../src/client/routes/ErrorsRoute"; + +type FetchInput = string | URL | Request; +type FetchInit = RequestInit | BunFetchRequestInit; + +const originalGlobals = new Map(); +let root: Root | null = null; + +function installGlobal(name: string, value: unknown): void { + originalGlobals.set(name, Object.getOwnPropertyDescriptor(globalThis, name)); + Object.defineProperty(globalThis, name, { configurable: true, value, writable: true }); +} + +function restoreGlobals(): void { + for (const [name, descriptor] of originalGlobals) { + if (descriptor) { + Object.defineProperty(globalThis, name, descriptor); + } else { + Reflect.deleteProperty(globalThis, name); + } + } + originalGlobals.clear(); +} + +afterEach(async () => { + const activeRoot = root; + if (activeRoot) { + await act(async () => { + activeRoot.unmount(); + }); + root = null; + } + vi.restoreAllMocks(); + restoreGlobals(); +}); + +describe("ErrorsRoute range", () => { + it("requests the selected range again when the range changes", async () => { + const domWindow = parseHTML('
').window; + installGlobal("window", domWindow); + installGlobal("document", domWindow.document); + installGlobal("navigator", domWindow.navigator); + installGlobal("Node", domWindow.Node); + installGlobal("Element", domWindow.Element); + installGlobal("HTMLElement", domWindow.HTMLElement); + installGlobal("HTMLIFrameElement", domWindow.HTMLIFrameElement); + installGlobal("SVGElement", domWindow.SVGElement); + installGlobal("IS_REACT_ACT_ENVIRONMENT", true); + + const requestedUrls: string[] = []; + const fetchStub = Object.assign( + async (input: FetchInput, _init?: FetchInit) => { + requestedUrls.push(input instanceof Request ? input.url : input.toString()); + return Response.json([]); + }, + { preconnect: globalThis.fetch.preconnect }, + ); + vi.spyOn(globalThis, "fetch").mockImplementation(fetchStub); + + const container = domWindow.document.getElementById("root"); + if (!container) throw new Error("Expected test root"); + root = createRoot(container as unknown as Element); + + await act(async () => { + root?.render( {}} />); + }); + expect(requestedUrls).toEqual(["/api/stats/errors?range=24h&limit=50"]); + + await act(async () => { + root?.render( {}} />); + }); + expect(requestedUrls).toEqual([ + "/api/stats/errors?range=24h&limit=50", + "/api/stats/errors?range=7d&limit=50", + ]); + }); +}); diff --git a/packages/stats/tsconfig.client.json b/packages/stats/tsconfig.client.json index 2a4594ea1..7b97643c6 100644 --- a/packages/stats/tsconfig.client.json +++ b/packages/stats/tsconfig.client.json @@ -1,7 +1,8 @@ { "extends": "../tsconfig.workspace.json", "include": [ - "src/client" + "src/client", + "test/errors-route-range.test.tsx" ], "compilerOptions": { "jsx": "react-jsx",