Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/agent/pi.ts
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,10 @@ function renderAcpStatus(s: Record<string, unknown>): string {
const pct = contextLimit !== null && contextLimit > 0 ? ` (${((contextTokens / contextLimit) * 100).toFixed(1)}%)` : "";
lines.push(` context: ${fmtTok(contextTokens)}${contextLimit !== null ? ` / ${fmtTok(contextLimit)}` : ""}${pct}`);
}
const hostCredit = num(s.hostCredit);
if (hostCredit !== null && hostCredit > 0) {
lines.push(` host baseline: uncompressed (proxy backfilled +${fmtTok(hostCredit)} tok)`);
}
if (inputTokens !== null || outputTokens !== null || cachedTokens !== null) {
lines.push(` in/out/cached: ${fmtTok(inputTokens ?? 0)} / ${fmtTok(outputTokens ?? 0)} / ${fmtTok(cachedTokens ?? 0)}`);
}
Expand Down
22 changes: 20 additions & 2 deletions src/loop/adapter-anthropic.ts
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ function buildTextDeltaEvent(index: number, text: string): Buffer {
);
}

export function createAnthropicAdapter(requestBody: Record<string, unknown>, originalSystem?: AnthropicRequestBody["system"]): CompressLoopAdapter {
export function createAnthropicAdapter(requestBody: Record<string, unknown>, originalSystem?: AnthropicRequestBody["system"], hostCredit = 0): CompressLoopAdapter {
const model = (requestBody.model as string) ?? undefined;
let messageId: string | undefined;
let clientIndex = 0;
Expand Down Expand Up @@ -236,8 +236,26 @@ export function createAnthropicAdapter(requestBody: Record<string, unknown>, ori
if (typeof u.input_tokens === "number") roundInput = u.input_tokens;
if (typeof u.cache_read_input_tokens === "number") roundCached = u.cache_read_input_tokens;
if (round === 1) {
// #408: this raw message_start (post-fold input_tokens)
// reaches the host verbatim — add the prepare-time
// credit back so the host anchors on the uncompressed
// baseline.
let chunk = rawBuf;
if (hostCredit > 0 && typeof u.input_tokens === "number") {
const patched = structuredClone(data);
const pmsg = patched["message"] as Record<string, unknown> | undefined;
const pu = (pmsg?.["usage"] ?? {}) as Record<string, unknown>;
if (typeof pu.input_tokens === "number") {
pu.input_tokens += hostCredit;
const out = eventStr
.split("\n")
.map((l) => (l.startsWith("data:") ? `data: ${JSON.stringify(patched)}` : l))
.join("\n");
chunk = Buffer.from(out + "\n\n", "utf8");
}
}
messageStartForwarded = true;
yield { kind: "meta", chunk: rawBuf, firstRoundOnly: true } as ParsedStreamEvent;
yield { kind: "meta", chunk, firstRoundOnly: true } as ParsedStreamEvent;
}
} else if (type === "ping") {
yield { kind: "meta", chunk: rawBuf } as ParsedStreamEvent;
Expand Down
28 changes: 26 additions & 2 deletions src/loop/adapter-openai.ts
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,7 @@ function stripFinishReasonChunk(buf: Buffer): Buffer {
}
}

export function createOpenaiAdapter(requestBody: Record<string, unknown>, clientSystem?: string): CompressLoopAdapter {
export function createOpenaiAdapter(requestBody: Record<string, unknown>, clientSystem?: string, hostCredit = 0): CompressLoopAdapter {
const model = (requestBody.model as string) ?? "unknown";
let responseId = `chatcmpl-proxy-${Date.now()}`;
let toolIndex = 0;
Expand Down Expand Up @@ -330,11 +330,35 @@ export function createOpenaiAdapter(requestBody: Record<string, unknown>, client
cachedTokens: typeof pd?.cached_tokens === "number" ? pd.cached_tokens : undefined,
} as ParsedStreamEvent;
if (sawRealToolCall) {
// #408: this raw finish chunk (with the provider's
// post-fold usage) reaches the host verbatim — add the
// prepare-time credit back so the host anchors on the
// uncompressed baseline.
let chunk = rawBuf;
if (hostCredit > 0 && u) {
const pu = typeof u.prompt_tokens === "number" ? u.prompt_tokens : undefined;
const tu = typeof u.total_tokens === "number" ? u.total_tokens : undefined;
if (pu !== undefined || tu !== undefined) {
const patched = {
...parsed,
usage: {
...u,
...(pu !== undefined ? { prompt_tokens: pu + hostCredit } : {}),
...(tu !== undefined ? { total_tokens: tu + hostCredit } : {}),
},
};
const out = eventStr
.split("\n")
.map((l) => (l.startsWith("data:") ? `data: ${JSON.stringify(patched)}` : l))
.join("\n");
chunk = Buffer.from(out + "\n\n", "utf8");
}
}
// This verbatim chunk IS the round's authoritative completion
// (suppressCompletion); write it once and never fall through
// to the text/reasoning branches (which would re-emit the same
// bytes after the finish reason).
yield { kind: "meta", chunk: rawBuf } as ParsedStreamEvent;
yield { kind: "meta", chunk } as ParsedStreamEvent;
yield { kind: "done", finishReason, suppressCompletion: true } as ParsedStreamEvent;
continue;
} else {
Expand Down
25 changes: 14 additions & 11 deletions src/loop/core.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import { proxyDispatcher } from "../upstream-proxy.js";
import { noteWeakOverflow } from "../weak-overflow.js";
import { warnCacheCollapse } from "../cache-warn.js";
import { log as loggerLog } from "../logger.js";
import type { WireProtocol } from "../util.js";
import { promptInputTotal, type WireProtocol } from "../util.js";

export const MAX_LOOP_ROUNDS = 10;

Expand Down Expand Up @@ -200,21 +200,15 @@ function recordUsage(
const prompt = usage.inputTokens;
const cached = usage.cachedTokens;
const out = usage.outputTokens;
// Adapters emit `inputTokens` in protocol-native units: Anthropic's
// `input_tokens` is the NEW (uncached) portion only (cached reported
// separately), while OpenAI/Responses report the TOTAL (cached already
// included). Add `cached` back in ONLY when it is not already part of
// `prompt` — otherwise the cached portion is double-counted, inflating the
// context size (→ premature compression) and deflating the hit rate.
const includesCached = ctx.protocol === "openai" || ctx.protocol === "responses";
const total =
(typeof prompt === "number" ? prompt : 0) +
(!includesCached && typeof cached === "number" ? cached : 0);
const total = promptInputTotal(ctx.protocol, prompt, cached);
if (total > 0) ctx.session.stats.inputTokens += total;
// Net out this turn's compress credit: the post-compress re-request
// re-sends the unfolded history, so its usage report over-reports the
// context the NEXT request will actually carry (see stream.ts applyRanges).
ctx.session.stats.lastInputTokens = Math.max(0, total - (ctx.session.stats.compressCreditTokens ?? 0));
// #408: remember the input-side total reported to the host AFTER the
// prepare-time fold backfill (uncompressed baseline), for the /acp panel.
ctx.session.hostContextTokens = total + (ctx.session.hostCreditTokens ?? 0);
if (typeof cached === "number") {
ctx.session.stats.cachedTokens += cached;
ctx.session.stats.cacheSamples += 1;
Expand Down Expand Up @@ -395,6 +389,15 @@ export async function* runCompressLoop(
) {
recordUsage(ctx, usage, round);
}
// #408: the provider measured the FOLDED (post-compress) view; add
// the prepare-time credit back so the completion event the host
// anchors on reports the uncompressed baseline. recordUsage above
// already ran on the un-backfilled numbers (internal ledger stays
// post-fold).
const hostCredit = ctx.session.hostCreditTokens ?? 0;
if (hostCredit > 0 && typeof usage.inputTokens === "number") {
usage.inputTokens += hostCredit;
}

let resolvedText = assistantText;
let allCalls = calls;
Expand Down
5 changes: 3 additions & 2 deletions src/loop/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,9 +25,10 @@ export function pickAdapter(
responsesProjection?: ResponsesProjection,
anthropicSystem?: AnthropicRequestBody["system"],
openaiSystem?: string,
hostCredit = 0,
): CompressLoopAdapter {
if (protocol === "responses") return createResponsesAdapter(textProtocol, responsesProjection);
if (protocol === "openai") return createOpenaiAdapter(requestBody, openaiSystem);
if (protocol === "anthropic") return createAnthropicAdapter(requestBody, anthropicSystem);
if (protocol === "openai") return createOpenaiAdapter(requestBody, openaiSystem, hostCredit);
if (protocol === "anthropic") return createAnthropicAdapter(requestBody, anthropicSystem, hostCredit);
throw new Error(`[acp-loop] unknown protocol: ${protocol}`);
}
44 changes: 40 additions & 4 deletions src/persist.ts
Original file line number Diff line number Diff line change
Expand Up @@ -207,8 +207,20 @@ export class SessionStore {
async loadAll(): Promise<Map<string, Session>> {
const out = new Map<string, Session>();
if (!this.enabled) return out;
let clamped = 0;
for (const [id, envelope] of await this.store.loadAll()) {
out.set(id, buildSession(envelope.payload));
const session = buildSession(envelope.payload);
if (hasNegativePersistedTokens(envelope.payload)) {
// #408 one-time migration: the in-memory value is already
// clamped by buildSession — rewrite the stale file so the
// negative value is gone from disk.
await this.store.writeNow(id, () => buildRecord(session));
clamped++;
}
out.set(id, session);
}
if (clamped > 0) {
loggerLog("info", `[persist] one-time migration (#408): clamped negative lastInputTokens/contextTokens in ${clamped} session(s) to 0`);
}
return out;
}
Expand Down Expand Up @@ -349,7 +361,16 @@ export class SessionStore {
meta?.protocol ? this.store.loadSync(id, relPathFor(id)) : null,
];
for (const envelope of envelopes) {
if (envelope) return buildSession(envelope.payload);
if (envelope) {
const session = buildSession(envelope.payload);
if (hasNegativePersistedTokens(envelope.payload)) {
// #408: sync context — debounce the stale-file rewrite
// (buildSession already clamped the in-memory value).
this.scheduleSave(session);
loggerLog("info", `[persist] clamped negative token stats on reload for ${id} (#408)`);
}
return session;
}
}
return null;
}
Expand Down Expand Up @@ -449,10 +470,14 @@ function buildSession(parsed: PersistedSession): Session {
cachedTokens: stats.cachedTokens ?? parsed.cachedTokens ?? 0,
outputTokens: stats.outputTokens ?? parsed.outputTokens ?? 0,
cacheSamples: stats.cacheSamples ?? parsed.cacheSamples ?? 0,
lastInputTokens: stats.lastInputTokens ?? parsed.lastInputTokens ?? 0,
// #408: clamp at restore — pre-clamp versions persisted negative
// values (lastInputTokens = total − credit before the Math.max
// guard existed) which would otherwise revive after upgrade and
// feed the /acp panel + web stats as negative percentages.
lastInputTokens: Math.max(0, stats.lastInputTokens ?? parsed.lastInputTokens ?? 0),
// In-memory only — a fresh process has no pending compress fold.
compressCreditTokens: 0,
contextTokens: stats.contextTokens ?? parsed.contextTokens ?? 0,
contextTokens: Math.max(0, stats.contextTokens ?? parsed.contextTokens ?? 0),
},
metadata: parsed.metadata ?? {},
state: mergeState(parsed.state),
Expand All @@ -477,6 +502,17 @@ function isValidRecord(parsed: unknown): parsed is PersistedSession {
return typeof r.id === "string" && typeof r.state === "object" && r.state !== null && Array.isArray(r.state.blocks);
}

/** #408: true when a persisted record (grouped v2+ or flat v1) carries a
* negative lastInputTokens/contextTokens — only possible in files written by
* pre-clamp versions. buildSession clamps on read; this lets the loaders
* rewrite the stale file once so the negative value is gone from disk. */
function hasNegativePersistedTokens(parsed: PersistedSession): boolean {
const stats = parsed.stats ?? {};
const last = stats.lastInputTokens ?? parsed.lastInputTokens;
const ctx = stats.contextTokens ?? parsed.contextTokens;
return (typeof last === "number" && last < 0) || (typeof ctx === "number" && ctx < 0);
}

function defaultDir(): string {
return sessionsDir();
}
Expand Down
Loading
Loading