fix(advisor): anchored context maintenance on provider usage
- Anchored advisor compaction on provider-reported context usage (cached input + generated output) floored by a full local estimate including the advisor system prompt and tool schemas, so a near-full cached context is no longer undercounted by the per-message estimate. - Rejected stale provider usage retained across advisor compaction via a runtime-only usage-anchor boundary recorded on the summary message. - Recovered 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. - Threaded the selected dashboard range through the stats Recent Errors UI, API, and database timestamp filter before ordering and the 50-row limit. Fixes #5282
This commit is contained in:
roboomp
@@ -217,6 +217,7 @@
|
||||
"@types/bun": "catalog:",
|
||||
"@types/react": "catalog:",
|
||||
"@types/react-dom": "catalog:",
|
||||
"linkedom": "catalog:",
|
||||
"postcss": "catalog:",
|
||||
},
|
||||
},
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
|
||||
### Fixed
|
||||
|
||||
- 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
|
||||
|
||||
|
||||
@@ -1,5 +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";
|
||||
@@ -978,7 +980,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[] = [];
|
||||
let resetCount = 0;
|
||||
const agent: AdvisorAgent = {
|
||||
@@ -992,37 +994,326 @@ 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 Promise.resolve();
|
||||
expect(promptInputs).toHaveLength(1);
|
||||
expect(promptInputs[0]).toContain("aaa");
|
||||
expect(resetCount).toBe(0);
|
||||
|
||||
// Second turn: maintainContext resolves true, triggering a re-prime
|
||||
shouldRePrime = true;
|
||||
shouldResetContext = true;
|
||||
messages.push({ role: "user", content: "bbb", timestamp: 2 } as AgentMessage);
|
||||
runtime.onTurnEnd(messages);
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
|
||||
// The reset cleared history and prompted a full replay (so the batch contains 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<void>();
|
||||
const maintenanceFinished = Promise.withResolvers<boolean>();
|
||||
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<void>();
|
||||
const finishSecondAttempt = Promise.withResolvers<void>();
|
||||
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<void>();
|
||||
|
||||
@@ -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<boolean>;
|
||||
/**
|
||||
@@ -65,7 +66,10 @@ export interface AdvisorRuntimeHost {
|
||||
|
||||
interface PendingDelta {
|
||||
text: string;
|
||||
rawMessages: AgentMessage[];
|
||||
renderRevision: number;
|
||||
turns: number;
|
||||
overflowRecovery?: boolean;
|
||||
}
|
||||
|
||||
interface CatchupWaiter {
|
||||
@@ -83,6 +87,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<string, string>();
|
||||
/** 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;
|
||||
@@ -111,9 +117,9 @@ export class AdvisorRuntime {
|
||||
if (this.disposed) return;
|
||||
const all = messages ?? this.host.snapshotMessages();
|
||||
this.#latestMessages = all;
|
||||
const render = this.#renderDelta(all);
|
||||
if (render) {
|
||||
this.#pending.push({ text: render, turns: 1 });
|
||||
const rendered = this.#renderDelta(all);
|
||||
if (rendered) {
|
||||
this.#pending.push({ ...rendered, turns: 1 });
|
||||
this.#backlog++;
|
||||
this.#notifyWaiters();
|
||||
void this.#drain();
|
||||
@@ -153,18 +159,15 @@ export class AdvisorRuntime {
|
||||
} catch {}
|
||||
}
|
||||
|
||||
#resetAdvisorContext(clearBacklog: boolean, wakeWaiters: boolean): void {
|
||||
this.#lastCount = 0;
|
||||
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 {}
|
||||
@@ -173,6 +176,18 @@ export class AdvisorRuntime {
|
||||
} catch {}
|
||||
}
|
||||
|
||||
#resetAdvisorContext(clearBacklog: boolean, wakeWaiters: boolean): void {
|
||||
this.#lastCount = 0;
|
||||
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
|
||||
@@ -196,22 +211,14 @@ export class AdvisorRuntime {
|
||||
this.#backlog = 0;
|
||||
this.#consecutiveFailures = 0;
|
||||
this.#failureNotified = false;
|
||||
this.#seenContext.clear();
|
||||
this.#clearSeenContext();
|
||||
this.#wakeAllWaiters();
|
||||
}
|
||||
|
||||
#renderDelta(messages?: AgentMessage[]): string | null {
|
||||
const all = messages ?? this.#latestMessages ?? this.host.snapshotMessages();
|
||||
if (all.length < this.#lastCount) {
|
||||
this.#lastCount = all.length;
|
||||
this.#seenContext.clear();
|
||||
return null;
|
||||
}
|
||||
const delta = all
|
||||
.slice(this.#lastCount)
|
||||
.filter(m => !(m.role === "custom" && (m as { customType?: string }).customType === "advisor"))
|
||||
.map(m => this.#dedupContextMessage(m));
|
||||
this.#lastCount = all.length;
|
||||
#formatRawDelta(rawMessages: AgentMessage[]): 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;
|
||||
@@ -226,6 +233,19 @@ export class AdvisorRuntime {
|
||||
return `### Session update\n\n${md}`;
|
||||
}
|
||||
|
||||
#renderDelta(messages?: AgentMessage[]): Omit<PendingDelta, "turns" | "overflowRecovery"> | null {
|
||||
const all = messages ?? this.#latestMessages ?? this.host.snapshotMessages();
|
||||
if (all.length < this.#lastCount) {
|
||||
this.#lastCount = all.length;
|
||||
this.#clearSeenContext();
|
||||
return null;
|
||||
}
|
||||
const rawMessages = all.slice(this.#lastCount);
|
||||
this.#lastCount = all.length;
|
||||
const text = this.#formatRawDelta(rawMessages);
|
||||
return text ? { text, rawMessages, renderRevision: this.#renderRevision } : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Collapse a re-injected primary-context prompt (plan/goal mode rules, the
|
||||
* approved plan) to a short marker when its body is byte-identical to the
|
||||
@@ -236,12 +256,12 @@ export class AdvisorRuntime {
|
||||
*/
|
||||
#dedupContextMessage(msg: AgentMessage): AgentMessage {
|
||||
if (msg.role !== "custom") return msg;
|
||||
const type = (msg as { customType?: string }).customType;
|
||||
if (!type || !PRIMARY_CONTEXT_CUSTOM_TYPES.has(type)) return msg;
|
||||
const content = (msg as { content?: unknown }).content;
|
||||
const type = msg.customType;
|
||||
if (!PRIMARY_CONTEXT_CUSTOM_TYPES.has(type)) return msg;
|
||||
const content = msg.content;
|
||||
if (typeof content !== "string") return msg;
|
||||
if (this.#seenContext.get(type) === content) {
|
||||
return { ...(msg as object), content: "(unchanged — still in effect)" } as AgentMessage;
|
||||
return { ...msg, content: "(unchanged — still in effect)" };
|
||||
}
|
||||
this.#seenContext.set(type, content);
|
||||
return msg;
|
||||
@@ -284,27 +304,61 @@ export class AdvisorRuntime {
|
||||
}
|
||||
}
|
||||
|
||||
#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<void> {
|
||||
if (this.#busy) return;
|
||||
this.#busy = true;
|
||||
try {
|
||||
while (!this.disposed && this.#pending.length) {
|
||||
const popped = this.#pending.splice(0);
|
||||
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;
|
||||
for (const delta of popped) {
|
||||
if (delta.renderRevision === this.#renderRevision) continue;
|
||||
const refreshed = this.#formatRawDelta(delta.rawMessages);
|
||||
if (refreshed) delta.text = refreshed;
|
||||
delta.renderRevision = this.#renderRevision;
|
||||
}
|
||||
const rawMessages = popped.flatMap(delta => delta.rawMessages);
|
||||
// Each delta already opens with a `### Session update` heading, so
|
||||
// join with a blank line rather than a `---` rule.
|
||||
const candidateBatch = popped.map(b => b.text).join("\n\n");
|
||||
const turnsCovered = popped.reduce((sum, b) => sum + b.turns, 0);
|
||||
let batch = popped.map(delta => delta.text).join("\n\n");
|
||||
const finalTurns = popped.reduce((sum, delta) => sum + delta.turns, 0);
|
||||
const recoveringOverflow = popped.some(delta => delta.overflowRecovery === true);
|
||||
const incomingTokens = estimateTokens({
|
||||
role: "user",
|
||||
content: candidateBatch,
|
||||
content: batch,
|
||||
timestamp: Date.now(),
|
||||
});
|
||||
|
||||
let shouldReprime = false;
|
||||
let shouldResetContext = false;
|
||||
if (this.host.maintainContext) {
|
||||
try {
|
||||
shouldReprime = await this.host.maintainContext(incomingTokens);
|
||||
shouldResetContext = await this.host.maintainContext(incomingTokens);
|
||||
} catch (err) {
|
||||
logger.debug("advisor context maintenance failed", { err: String(err) });
|
||||
}
|
||||
@@ -312,20 +366,16 @@ export class AdvisorRuntime {
|
||||
// A reset/dispose during context maintenance invalidates this batch.
|
||||
if (this.#epoch !== epoch) continue;
|
||||
|
||||
let batch: string | null;
|
||||
let finalTurns: number;
|
||||
if (shouldReprime) {
|
||||
// Promotion could not fit the advisor's context — re-prime.
|
||||
const newTurns = this.#pending.reduce((sum, b) => sum + b.turns, 0);
|
||||
this.#resetAdvisorContext(false, false);
|
||||
batch = this.#renderDelta(this.#latestMessages);
|
||||
finalTurns = turnsCovered + newTurns;
|
||||
} else {
|
||||
batch = candidateBatch;
|
||||
finalTurns = turnsCovered;
|
||||
if (shouldResetContext) {
|
||||
// Reset only the advisor Agent/log. The primary cursor, queued deltas,
|
||||
// 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();
|
||||
batch = this.#formatRawDelta(rawMessages) ?? batch;
|
||||
}
|
||||
|
||||
if (this.disposed || batch === null) {
|
||||
if (this.disposed) {
|
||||
this.#backlog = Math.max(0, this.#backlog - finalTurns);
|
||||
this.#notifyWaiters();
|
||||
continue;
|
||||
@@ -338,6 +388,7 @@ export class AdvisorRuntime {
|
||||
// failed batch on top of the 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 = shouldResetContext || 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 with a
|
||||
@@ -356,11 +407,14 @@ export class AdvisorRuntime {
|
||||
this.#consecutiveFailures = 0;
|
||||
this.#failureNotified = false;
|
||||
} catch (err) {
|
||||
// reset()/dispose() aborts the in-flight prompt; the rejection is the
|
||||
// reset itself, not a transient advisor failure. Drop the stale batch
|
||||
// (reset already cleared #pending and rewound the cursor) instead of
|
||||
// requeuing it into the post-reset conversation.
|
||||
// An external reset/dispose invalidates the in-flight bounded batch;
|
||||
// never requeue it into the post-reset conversation.
|
||||
if (this.#epoch !== epoch) continue;
|
||||
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) });
|
||||
try {
|
||||
@@ -371,28 +425,50 @@ export class AdvisorRuntime {
|
||||
// The hook awaits; a reset during it invalidates this batch like the
|
||||
// prompt await above — drop it instead of requeueing stale content.
|
||||
if (this.#epoch !== epoch) continue;
|
||||
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) ?? batch;
|
||||
this.#pending.unshift({
|
||||
text: recoveryBatch,
|
||||
rawMessages,
|
||||
renderRevision: this.#renderRevision,
|
||||
turns: finalTurns,
|
||||
overflowRecovery: true,
|
||||
});
|
||||
logger.debug("advisor context overflow recovered at current primary cursor");
|
||||
}
|
||||
} else {
|
||||
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) });
|
||||
}
|
||||
}
|
||||
this.#notifyFailureOnce(err);
|
||||
this.#consecutiveFailures = 0;
|
||||
// The dropped batch may carry primary-context we never delivered; drop
|
||||
// the seen-state too so the next turn re-expands it instead of marking
|
||||
// it "unchanged" against content the advisor never received.
|
||||
this.#seenContext.clear();
|
||||
// the seen-state too so queued raw deltas re-expand before delivery.
|
||||
this.#clearSeenContext();
|
||||
success = true;
|
||||
} else {
|
||||
this.#pending.unshift({ text: batch, turns: finalTurns });
|
||||
this.#pending.unshift({
|
||||
text: batch,
|
||||
rawMessages,
|
||||
renderRevision: this.#renderRevision,
|
||||
turns: finalTurns,
|
||||
overflowRecovery: recoveringOverflow || undefined,
|
||||
});
|
||||
await Bun.sleep(this.retryDelayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (success && this.#epoch === epoch) {
|
||||
this.#backlog = Math.max(0, this.#backlog - finalTurns);
|
||||
|
||||
@@ -248,7 +248,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 { createPlanReadMatcher } from "../plan-mode/plan-protection";
|
||||
import type { PlanModeState } from "../plan-mode/state";
|
||||
@@ -1037,6 +1041,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;
|
||||
@@ -2849,10 +2860,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;
|
||||
@@ -2876,6 +2900,7 @@ export class AgentSession {
|
||||
const timestamp = String(message.timestamp || Date.now());
|
||||
|
||||
if (message.role === "compactionSummary") {
|
||||
const advisorSummary = message as AdvisorCompactionSummaryMessage;
|
||||
return {
|
||||
type: "compaction",
|
||||
id,
|
||||
@@ -2883,9 +2908,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;
|
||||
}
|
||||
@@ -2979,11 +3002,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;
|
||||
@@ -16426,37 +16453,51 @@ export class AgentSession {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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
|
||||
|
||||
@@ -55,6 +55,7 @@
|
||||
"@types/bun": "catalog:",
|
||||
"@types/react": "catalog:",
|
||||
"@types/react-dom": "catalog:",
|
||||
"linkedom": "catalog:",
|
||||
"postcss": "catalog:"
|
||||
},
|
||||
"engines": {
|
||||
|
||||
@@ -444,9 +444,10 @@ export async function getRecentRequests(limit?: number): Promise<MessageStats[]>
|
||||
return dbGetRecentRequests(limit);
|
||||
}
|
||||
|
||||
export async function getRecentErrors(limit?: number): Promise<MessageStats[]> {
|
||||
export async function getRecentErrors(range?: string | null, limit?: number): Promise<MessageStats[]> {
|
||||
await initDb();
|
||||
return dbGetRecentErrors(limit);
|
||||
const { cutoff } = getTimeRangeConfig(range);
|
||||
return dbGetRecentErrors(limit, cutoff);
|
||||
}
|
||||
|
||||
export async function getRequestDetails(id: number): Promise<RequestDetails | null> {
|
||||
|
||||
@@ -59,8 +59,14 @@ export async function getRecentRequests(limit = 50, signal?: AbortSignal): Promi
|
||||
return fetchJson<MessageStats[]>(`${API_BASE}/stats/recent?limit=${limit}`, { signal });
|
||||
}
|
||||
|
||||
export async function getRecentErrors(limit = 50, signal?: AbortSignal): Promise<MessageStats[]> {
|
||||
return fetchJson<MessageStats[]>(`${API_BASE}/stats/errors?limit=${limit}`, { signal });
|
||||
export async function getRecentErrors(
|
||||
range: TimeRange = "24h",
|
||||
limit = 50,
|
||||
signal?: AbortSignal,
|
||||
): Promise<MessageStats[]> {
|
||||
return fetchJson<MessageStats[]>(`${API_BASE}/stats/errors?range=${encodeURIComponent(range)}&limit=${limit}`, {
|
||||
signal,
|
||||
});
|
||||
}
|
||||
|
||||
export async function getRequestDetails(id: number, signal?: AbortSignal): Promise<RequestDetails> {
|
||||
|
||||
@@ -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,
|
||||
});
|
||||
|
||||
@@ -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
|
||||
WHERE stop_reason = 'error'
|
||||
${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 {
|
||||
|
||||
@@ -184,7 +184,7 @@ const ensureClientBuild = async () => {
|
||||
/**
|
||||
* Handle API requests.
|
||||
*/
|
||||
async function handleApi(req: Request): Promise<Response> {
|
||||
export async function handleApi(req: Request): Promise<Response> {
|
||||
const url = new URL(req.url);
|
||||
const path = url.pathname;
|
||||
|
||||
@@ -229,7 +229,7 @@ async function handleApi(req: Request): Promise<Response> {
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
|
||||
@@ -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<MessageStats[]> {
|
||||
expect(response.status).toBe(200);
|
||||
return response.json() as Promise<MessageStats[]>;
|
||||
}
|
||||
|
||||
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));
|
||||
});
|
||||
});
|
||||
@@ -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<string, PropertyDescriptor | undefined>();
|
||||
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('<html><body><div id="root"></div></body></html>').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(<ErrorsRoute active range="24h" refreshTrigger={0} onRequestClick={() => {}} />);
|
||||
});
|
||||
expect(requestedUrls).toEqual(["/api/stats/errors?range=24h&limit=50"]);
|
||||
|
||||
await act(async () => {
|
||||
root?.render(<ErrorsRoute active range="7d" refreshTrigger={0} onRequestClick={() => {}} />);
|
||||
});
|
||||
expect(requestedUrls).toEqual([
|
||||
"/api/stats/errors?range=24h&limit=50",
|
||||
"/api/stats/errors?range=7d&limit=50",
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -1,7 +1,8 @@
|
||||
{
|
||||
"extends": "../tsconfig.workspace.json",
|
||||
"include": [
|
||||
"src/client"
|
||||
"src/client",
|
||||
"test/errors-route-range.test.tsx"
|
||||
],
|
||||
"compilerOptions": {
|
||||
"jsx": "react-jsx",
|
||||
|
||||
Reference in New Issue
Block a user