diff --git a/CHANGELOG.md b/CHANGELOG.md index f0fb354..5094f0b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ # Changelog ## Unreleased (master, since v0.1.38) +- **fix(nudge): 瞬态注入统一治理——注入预算账本（统一 #6/#7/#9/#194/#217 一族补丁的根因）** — pi 每次 LLM 调用重建出站数组，任何"瞬态"提示若不预算都会逐次重注入；#223 放大链上每张补丁各记各的账（Set 去重 / 失败调用数 / fire 预算），没有一个在数"注入次数本身"。现在 runtime 提供唯一账本 `noteInjection(turnKey, kind, budget)`：普通 nudge 每真实用户轮 1 次、emergency nudge 每轮 `MAX_EMERGENCY_NUDGES_PER_TURN=3` 次，预算数的是**注入**而非模型行为（忽略、失败、no-op、中性结果都无法延长），仅真实用户轮切换时重置。轮键 `lastUserMessageId` 现在跳过合成 user 消息（`[ACP:provider-throttle]` kick、`[acp_delegate …]` 通知不再重置预算）。删除整套 compress-outcome 记账机器（`noteCompressOutcomes`/`compressOutcomeSeen`/`compressFailTurnKey`+`compressFailCount`/`compressRetryCappedFor`/`collectCompressOutcomes`/`isCompressSuccessText`/`isCompressNoopText`/独立 `nudgeShownTurns`，约 -150 行）；emergency nudge 的中性逃生门（模型回 `compress({content: []})` → "No ranges provided." isError:false → 旧失败计数器不增 → 永不封顶、nudge 无限重注入）随之消失。封顶后 UI 一次性提示 nudge 已暂停，kernel ≥95% 紧急截断仍机械兜底；测试加入对抗性回归（20 连发交替 失败/no-op/中性/沉默 → 恰好 3 次注入） - **fix(nudge): 移除瞬态 compress 重试提示注入（closes #223，取代 #217）** — compress 失败后每次 LLM 调用重注入的 `compressRetryMessage` 瞬态 user 提示整体移除：对从不重试的模型，该提示无限追加（用户日志 ~400 次/小时、emergency pct 95→127%），即 #223 的"永远追加失败标记"。失败信息本身仍以 toolResult 形式持久留在 session 日志中（模型可见、可自我纠正，随正常压缩流程淘汰）；issue #6 的 nudge 断路器保留：每用户轮 MAX_COMPRESS_ATTEMPTS=3 次失败/no-op 后 emergency nudge 停止重注入（kernel 紧急截断仍机械兜底），UI 提示改为 "nudge paused until the next user message (emergency truncation still active)"。`noteCompressOutcomes` 返回值去掉 `retryFor` - **fix(tokens): 图片 token 计入发送视图估算 (closes #200)** — `extractText` 只投影 `type:"text"` 块，图片在 sent-view 估算中计 0 token：含图会话的 nudge/truncation/compress 仲裁系统性偏晚（只等真实 400 后 overflow-selfheal 被动触发），且 density 校准被 phantom gap 污染（provider 真实 usage 含图、估算不含 → 图片轮 dReal/dEst 爆表被 clamp 到 2.5×，纯文本轮又拉回 1.0，density 振荡且仍低估 5-10× → 过早/过晚压缩交替）。现在 `collectImageTokens` 按 `IMAGE_TOKEN_COST=1600`/张计入（仅视觉模型，`model.input` 含 `image`；非视觉模型 pi-ai 静默丢图、计 0），density 校准环自动收敛真实成本（A/B 实测收敛 ~0.98）；出站 payload 字节不变（sha256 一致），前缀缓存不受影响 (#201) - **fix(delegate): 并发多 agent 时失败必达，不再静默 (#16)** — async delegate 此前有三条失败路径完全不通知主模型(spawn error、结果持久化 error、`sendUserMessage` 注入丢失)，模型未挂在 `acp_delegate_wait` 上时失败被吞，直到收尾汇总才发现少了结果。现在：所有终止路径 best-effort 注入 `FAILED ⚠️` 通知(带错误摘录，与 sync 路径对齐，明确提示"该任务结果缺失、收尾前决定是否重派")；注入失败的 run 进入未送达集，随**下一个** delegate 通知或任何 delegate 工具结果(`acp_delegate`/`wait`/`cancel`)捎带 Recovery notice 补投；Recovery notice 的 delivered 标记改为 carrier 发送成功后才提交（发送抛错不再永久吞掉其他 run 的结果）；system prompt 补充 FAILED/Recovery 通知说明 diff --git a/src/compress-tool.ts b/src/compress-tool.ts index 9f0b579..9cf941f 100644 --- a/src/compress-tool.ts +++ b/src/compress-tool.ts @@ -97,28 +97,6 @@ function describeDiagnostics(diagnostics: CompressParseDiagnostics, content: Com return `${base}. content must be an ARRAY of {startId, endId, summary} objects.`; } -/** Panel block count ("… (~N reclaimed, B blocks)"), or -1 for non-panels. */ -function compressPanelBlocks(text: string): number { - if (!text.trimStart().startsWith("▣ ACP |")) return -1; - const m = text.match(/, (\d+) blocks?\)/); - return m ? Number(m[1]) : -1; -} - -/** Success = completed run that created >= 1 block (partial range errors - * still count: progress was made). A 0-block panel must NOT be success — - * it would reset the retry counter while the emergency nudge re-fires, - * looping no-op compressions (issue #6). */ -export function isCompressSuccessText(text: string): boolean { - return compressPanelBlocks(text) > 0; -} - -/** No-op = completed run that compressed nothing (0-block panel: every - * range skipped). Counted as a FAILED attempt by noteCompressOutcomes so - * the retry cap applies. Non-panels ("No ranges provided.") stay neutral. */ -export function isCompressNoopText(text: string): boolean { - return compressPanelBlocks(text) === 0; -} - function tier3OnlyRewrite(newBlocks: CompressionBlock[], allBlocks: CompressionBlock[]): string[] | null { if (newBlocks.length === 0) return null; const byId = new Map(allBlocks.map((b) => [b.blockId, b])); diff --git a/src/delegate-tool.ts b/src/delegate-tool.ts index c563ff3..3b8cc4a 100644 --- a/src/delegate-tool.ts +++ b/src/delegate-tool.ts @@ -480,6 +480,12 @@ export function findUndeliveredRuns(all: DelegateRun[], excludeRunId?: string): ); } +/** Prefix of every injected delegate notification user message + * (`[acp_delegate ] ...`). Shared with the injection-ledger turn + * key (SYNTHETIC_USER_PREFIXES in src/tokens.ts) so these synthetic user + * messages cannot reset per-turn injection budgets. */ +export const DELEGATE_NOTIFY_PREFIX = "[acp_delegate "; + /** Compute the recovery notice for undelivered runs WITHOUT marking them * delivered. The caller commits the marking (covered[].injected = true) only * after the carrier message is actually sent: if the send throws, the runs @@ -1141,7 +1147,7 @@ export function injectResult( const closing = failed ? "This delegate did NOT complete its task — its result is missing from your work. Read the error excerpt (and the result file if present), then decide whether to re-dispatch the task before wrapping up. This is an automated system notification, NOT a user message." : "This is an automated system notification, NOT a user message. Read the result file if you need the details, then continue your original task; do not treat this as a new user request."; - const header = `[acp_delegate ${status}] **${agent}** (runId \`${runId}\`, exit ${code ?? "?"})${timeoutNote}${remainingLine}${usageNote} ${closing}`; + const header = `${DELEGATE_NOTIFY_PREFIX}${status}] **${agent}** (runId \`${runId}\`, exit ${code ?? "?"})${timeoutNote}${remainingLine}${usageNote} ${closing}`; const { text: recoveryText, covered } = buildRecoveryNotice(Array.from(runs.values()), runId); const text = formatPayload(header, file, task, failed ? body : undefined) + (recoveryText ? `\n\n${recoveryText}` : ""); try { diff --git a/src/index.ts b/src/index.ts index 6464fdc..f7679ed 100644 --- a/src/index.ts +++ b/src/index.ts @@ -7,8 +7,8 @@ import type { import type { CoreMessage, NudgeDecision, CompressionBlock, Prompts } from "acp-kernel"; import { renderNudgeText, resolvePrompts, defaultPrompts, viableRanges } from "acp-kernel"; import { type AdapterConfig, resolveDelegate } from "./config.js"; -import { createRuntime, type AcpRuntime } from "./runtime.js"; -import { makeCompressTool, isCompressSuccessText, isCompressNoopText } from "./compress-tool.js"; +import { createRuntime, type AcpRuntime, MAX_EMERGENCY_NUDGES_PER_TURN } from "./runtime.js"; +import { makeCompressTool } from "./compress-tool.js"; import { makeDecompressTool } from "./decompress-tool.js"; import { makeSearchTool } from "./search-tool.js"; import { makeStatusTool } from "./status-tool.js"; @@ -77,9 +77,8 @@ function wireCompactionDisable(pi: ExtensionAPI): void { function wireSessionLifecycle(pi: ExtensionAPI, runtime: AcpRuntime): void { pi.on("session_start", async (_event, ctx) => { runtime.store.invalidate(); - runtime.clearNudgeTracking(); + runtime.clearInjectionLedger(); runtime.throttleFor(ctx.sessionManager.getSessionId()).reset(); - runtime.clearCompressRetryTracking(); // 新会话重置该模型的密度校准（文档 §5.3：模型/窗口切换时重新收敛） const modelId = (ctx.model as { id?: string } | undefined)?.id ?? "default"; runtime.density.resetModel(modelId); @@ -280,19 +279,6 @@ function wireContextTransform(pi: ExtensionAPI, runtime: AcpRuntime): void { const turnKey = lastUserMessageId(entries) ?? sid; - // Compress-outcome tracking feeds ONLY the nudge circuit breaker below: - // failed/no-op attempts are counted (capped at MAX_COMPRESS_ATTEMPTS per - // user turn) to stop re-injecting the nudge at a model that keeps failing - // compress. The failed toolResult itself already persists in the session - // log with the full error text — the model sees it and can self-correct — - // so NO transient retry prompt is injected (transient re-injection per - // LLM call caused the #223 infinite-append loop). Only outcomes from the - // CURRENT user turn are considered; processed BEFORE the nudge block so - // the cap suppression sees the newest outcome (a success on this fire - // must lift the cap on this same fire). - const compressOutcomes = collectCompressOutcomes(entries, turnStartIndex(entries)); - const outcome = compressOutcomes.length > 0 ? runtime.noteCompressOutcomes(turnKey, compressOutcomes) : null; - if (turn.nudge?.shouldInject) { // Two independent channels for the nudge: // 1. CONTEXT injection (always on): the nudge is appended to the @@ -302,50 +288,43 @@ function wireContextTransform(pi: ExtensionAPI, runtime: AcpRuntime): void { // 2. TERMINAL echo (debug only): when debug is on, also print the exact // text via ctx.ui.notify so the user can observe what is being // injected while debugging. The model never sees terminal output. - // Emergency nudges (usage >= 80%) bypass the per-turn dedup so the - // overflow warning always reaches the model. Other nudges inject at most - // once per turn: pi fires the context event multiple times per assistant - // reply (streaming/tool loop), and without this gate the same nudge - // would be appended on every event. + // Budget gate: pi fires the context event multiple times per assistant + // reply (streaming/tool loop), so EVERY transient injection must pass + // the runtime ledger or it would append on every event (#223). Normal + // nudges: once per turn. Emergency nudges: MAX_EMERGENCY_NUDGES_PER_TURN + // per turn — bounded by injections, not by the model's response + // behavior (ignoring, no-op or degenerate neutral compress answers + // cannot keep the loop alive); kernel emergency truncation (>= 95%) + // stays the mechanical backstop. const emergency = turn.nudge.breakdown?.emergencyOverride === 1; // Recommend only ranges the model can actually compress: a tiny // fragmented range in the list makes batched attempts fail atomically // (kernel validates the whole batch). See viableRanges in acp-kernel. turn.nudge.compressibleRanges = viableRanges(turn.nudge.compressibleRanges); - // Retry-cap circuit breaker (issue #6): emergency nudges re-inject on - // every LLM call, so a model answering each one with a failed/no-op - // compress call loops forever (each attempt adds protected tokens and - // keeps usage pinned at emergency). Once this turn burned - // MAX_COMPRESS_ATTEMPTS attempts, stop re-injecting the nudge — the - // kernel's emergency truncation still shrinks context mechanically. - const retryCapped = runtime.compressRetryCappedFor(turnKey); - const alreadyShown = retryCapped || (!emergency && runtime.nudgeShownFor(turnKey)); - if (!alreadyShown) { + const inj = runtime.noteInjection(turnKey, emergency ? "emergency" : "nudge", emergency ? MAX_EMERGENCY_NUDGES_PER_TURN : 1); + if (inj.allowed) { rebuilt.push(nudgeMessage(turn.nudge, turn.state.blocks.filter((b) => b.active), runtime.prompts)); const rendered = renderNudgeText(turn.nudge, runtime.prompts); const top = [...turn.nudge.compressibleRanges].sort((a, b) => b.tokens - a.tokens)[0]; const example = top ? `\n\nExample: compress({ content: [{ startId: "${top.startRef}", endId: "${top.endRef}", summary: "..." }] })` : ""; if (emergency) { logWarn("nudge", { sid: ctx.sessionManager.getSessionId(), event: "emergency-inject", pct: Math.round(turn.nudge.contextUsage * 100), voice: rendered.voice, compressible: turn.nudge.compressibleRanges.length }); + if (inj.exhaustedNow) { + logWarn("nudge", { sid, event: "emergency-nudge-exhausted", count: inj.count }); + if (ctx.hasUI) { + ctx.ui.notify(`[ACP] emergency nudge injected ${inj.count}× this turn with no compress response — nudge paused until the next user message (emergency truncation still active).`); + } + } } if (debugOn && ctx.hasUI) { ctx.ui.notify(`[ACP nudge → context]${emergency ? " [EMERGENCY]" : ""}\n${rendered.text}${example}`); } - if (!emergency) runtime.markNudgeShown(turnKey); debug.event("nudge-injected", { sid: ctx.sessionManager.getSessionId(), voice: rendered.voice, channels: ["context", debugOn ? "terminal" : null].filter(Boolean), emergency, turnKey, text: rendered.text + example }); } else { debug.event("nudge-suppressed", { sid: ctx.sessionManager.getSessionId(), turnKey, reason: turn.nudge.reason }); } } - if (outcome !== null && outcome.cappedNow) { - logWarn("nudge", { sid, event: "compress-retry-capped", failures: outcome.count }); - debug.event("compress-retry-capped", { sid, turnKey, failures: outcome.count }); - if (ctx.hasUI) { - ctx.ui.notify(`[ACP] compress failed ${outcome.count}× this turn — nudge paused until the next user message (emergency truncation still active).`); - } - } - // Always return the transformed array: every message needs its [mNNNNN] ref // tag applied, so there is no meaningful "no change" case to short-circuit. debug.event("context-out", { outMsgs: rebuilt.length, injected: turn.nudge?.shouldInject ?? false, emergency: turn.nudge?.breakdown?.emergencyOverride === 1 }); @@ -507,33 +486,6 @@ function collectOriginals(entries: Array<{ type: string; id: string; message?: A return map; } -// Index of the last user-role entry — the start of the current turn. -// Everything strictly AFTER this index belongs to the current turn; -1 when -// the session has no user message yet. -function turnStartIndex(entries: Array<{ type: string; message?: { role?: string } }>): number { - for (let i = entries.length - 1; i >= 0; i--) { - if (entries[i]!.message?.role === "user") return i; - } - return -1; -} - -// Compress toolResults from the CURRENT user turn only — the raw material for -// the nudge circuit breaker above. Scoping matters: feeding the whole session -// would keep an old failure counting against the current turn's budget -// forever (review finding on 7ddd2c6). -function collectCompressOutcomes(entries: Array<{ type: string; id: string; message?: AgentMessage }>, startIndex: number): Array<{ toolCallId: string; isError: boolean; success: boolean; noop: boolean; text: string }> { - const out: Array<{ toolCallId: string; isError: boolean; success: boolean; noop: boolean; text: string }> = []; - for (let i = Math.max(startIndex, -1) + 1; i < entries.length; i++) { - const entry = entries[i]!; - if (entry.type !== "message" || !entry.message) continue; - const m = entry.message as { role?: string; toolName?: string; toolCallId?: string; isError?: boolean; content?: unknown }; - if (m.role !== "toolResult" || m.toolName !== "compress" || !m.toolCallId) continue; - const text = extractText(m.content); - out.push({ toolCallId: m.toolCallId, isError: m.isError === true, success: m.isError !== true && isCompressSuccessText(text), noop: m.isError !== true && isCompressNoopText(text), text }); - } - return out; -} - function nudgeMessage(nudge: NudgeDecision, blocks: CompressionBlock[], prompts: Prompts): AgentMessage { const rendered = renderNudgeText(nudge, prompts); const lines = [rendered.text]; diff --git a/src/runtime.ts b/src/runtime.ts index 908b5bb..f44e751 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -66,21 +66,12 @@ export interface AcpRuntime { setAdapter(adapter: AdapterConfig): void; prompts: Prompts; setPrompts(prompts: Prompts): void; - markNudgeShown(turnKey: string): void; - nudgeShownFor(turnKey: string): boolean; - /** Process compress toolResults for the CURRENT user turn only (the caller - * scopes the list — see collectCompressOutcomes in src/index.ts); idempotent - * per toolCallId. Outcome classes: isError or noop (0-block panel) → - * failure (count++), success panel (>= 1 block) → reset, other non-error - * text → neutral (count unchanged). Returns the failure count and - * whether the cap was just reached. */ - noteCompressOutcomes(turnKey: string, outcomes: ReadonlyArray<{ toolCallId: string; isError: boolean; success: boolean; noop?: boolean }>): { count: number; cappedNow: boolean }; - /** True when this turn already burned MAX_COMPRESS_ATTEMPTS failed/no-op - * compress calls — used to stop re-injecting the (dedup-exempt) emergency - * nudge that would otherwise keep looping no-op compressions (issue #6). */ - compressRetryCappedFor(turnKey: string): boolean; - clearNudgeTracking(): void; - clearCompressRetryTracking(): void; + /** Single gate for every transient context injection (nudge, emergency + * nudge). Counts INJECTIONS per genuine user turn against a per-kind + * budget; denied calls do not increment. exhaustedNow is edge-triggered + * on the injection that reaches the budget (one-shot UI notice). */ + noteInjection(turnKey: string, kind: InjectionKind, budget: number): { allowed: boolean; count: number; exhaustedNow: boolean }; + clearInjectionLedger(): void; liveContextLimit(ctx: ExtensionContext): number; configFor(ctx: ExtensionContext): Config; /** Re-read ~/./acp.json + //acp.json and re-derive the adapter @@ -234,8 +225,16 @@ function pruneOrphanRefs(state: CompressionState, messages: ReturnType= 95% usage) remains the + * mechanical backstop. */ +export const MAX_EMERGENCY_NUDGES_PER_TURN = 3; + +export type InjectionKind = "nudge" | "emergency"; export function createRuntime(adapter: AdapterConfig): AcpRuntime { const density = new DensityEstimator(); @@ -251,7 +250,6 @@ export function createRuntime(adapter: AdapterConfig): AcpRuntime { let adapterRef = adapter; let lastUserConfigKey: string | undefined; let promptsRef: Prompts = defaultPrompts; - const nudgeShownTurns = new Set(); // Per-session overflow self-heal state (learned window + armed emergency). const overflowEpisodes = new Map(); function overflowFor(sid: string): OverflowEpisode { @@ -275,47 +273,32 @@ export function createRuntime(adapter: AdapterConfig): AcpRuntime { throttleEpisodes.delete(sid); } - // Compress-failure tracking (see wireContextTransform): counts FAILED/no-op - // compress calls per user turn so the nudge circuit breaker can stop - // re-injecting the nudge at a model that answers every nudge with another - // failed attempt (issue #6 emergency loop). The failed toolResult itself - // persists in the session log — no transient retry prompt is injected - // (transient re-injection per LLM call caused the #223 infinite loop). The - // caller feeds only CURRENT-turn outcomes; success resets the counter, - // neutral outcomes (non-error text that is not a success panel) leave it - // frozen so mixed failure modes cannot bypass the cap. - const compressOutcomeSeen = new Set(); - let compressFailTurnKey: string | null = null; - let compressFailCount = 0; + // Per-fire injection ledger: the ONE gate every transient context + // injection must pass through. Budgets count INJECTIONS — not failed + // calls, not distinct ids; #223 happened because a budget measured a + // different quantity than the injection it was supposed to bound. Reset + // happens ONLY on a genuine user-turn change (the caller derives the + // turn key from genuine user input, skipping synthetic machinery + // messages) or session start. Single-turn design: rollover discards the + // previous turn's counts, so memory is O(1) for the session lifetime. + let injectionTurnKey: string | null = null; + const injectionCounts = new Map(); - function noteCompressOutcomes(turnKey: string, outcomes: ReadonlyArray<{ toolCallId: string; isError: boolean; success: boolean; noop?: boolean }>): { count: number; cappedNow: boolean } { - if (compressFailTurnKey !== turnKey) { - compressFailTurnKey = turnKey; - compressFailCount = 0; - } - const prevCount = compressFailCount; - for (const o of outcomes) { - if (compressOutcomeSeen.has(o.toolCallId)) continue; - compressOutcomeSeen.add(o.toolCallId); - if (o.isError || o.noop === true) { - compressFailCount += 1; - } else if (o.success) { - compressFailCount = 0; - } - // neutral: counter untouched + function noteInjection(turnKey: string, kind: InjectionKind, budget: number): { allowed: boolean; count: number; exhaustedNow: boolean } { + if (injectionTurnKey !== turnKey) { + injectionTurnKey = turnKey; + injectionCounts.clear(); } - const cappedNow = compressFailCount >= MAX_COMPRESS_ATTEMPTS && prevCount < MAX_COMPRESS_ATTEMPTS; - return { count: compressFailCount, cappedNow }; - } - - function compressRetryCappedFor(turnKey: string): boolean { - return compressFailTurnKey === turnKey && compressFailCount >= MAX_COMPRESS_ATTEMPTS; + const prev = injectionCounts.get(kind) ?? 0; + if (prev >= budget) return { allowed: false, count: prev, exhaustedNow: false }; + const count = prev + 1; + injectionCounts.set(kind, count); + return { allowed: true, count, exhaustedNow: count >= budget }; } - function clearCompressRetryTracking(): void { - compressOutcomeSeen.clear(); - compressFailTurnKey = null; - compressFailCount = 0; + function clearInjectionLedger(): void { + injectionTurnKey = null; + injectionCounts.clear(); } async function acquireLock(sid: string): Promise<() => void> { @@ -403,4 +386,4 @@ export function createRuntime(adapter: AdapterConfig): AcpRuntime { lastActiveBlockIds.delete(sid); } - return { core, store, density, setCountModel: (m) => { countModelId = m; }, noteActiveBlocks, clearSessionTracking, get adapter() { return adapterRef; }, setAdapter: (a) => { adapterRef = a; }, get prompts() { return promptsRef; }, setPrompts: (p) => { promptsRef = p; }, markNudgeShown: (k) => { nudgeShownTurns.add(k); }, nudgeShownFor: (k) => nudgeShownTurns.has(k), clearNudgeTracking: () => { nudgeShownTurns.clear(); }, noteCompressOutcomes, compressRetryCappedFor, clearCompressRetryTracking, liveContextLimit, configFor, reloadConfig, stateFor, save, acquireLock, overflowFor, overflowDrop, throttleFor, throttleDrop };} + return { core, store, density, setCountModel: (m) => { countModelId = m; }, noteActiveBlocks, clearSessionTracking, get adapter() { return adapterRef; }, setAdapter: (a) => { adapterRef = a; }, get prompts() { return promptsRef; }, setPrompts: (p) => { promptsRef = p; }, noteInjection, clearInjectionLedger, liveContextLimit, configFor, reloadConfig, stateFor, save, acquireLock, overflowFor, overflowDrop, throttleFor, throttleDrop };} diff --git a/src/tokens.ts b/src/tokens.ts index e98c49d..5262bbd 100644 --- a/src/tokens.ts +++ b/src/tokens.ts @@ -1,6 +1,8 @@ import { defaultCountTokens, type CoreMessage } from "acp-kernel"; import type { SessionMessageEntry } from "@earendil-works/pi-coding-agent"; -import { countImageBlocks } from "./messages.js"; +import { countImageBlocks, extractText } from "./messages.js"; +import { THROTTLE_KICK_SENTINEL } from "./throttle-retry.js"; +import { DELEGATE_NOTIFY_PREFIX } from "./delegate-tool.js"; type AgentMessage = SessionMessageEntry["message"]; @@ -55,12 +57,26 @@ export function calibrateTokens(estimate: number, density: number): number { return density === 1 ? estimate : Math.round(estimate * density); } -/** Id of the last user-role entry — used as a per-turn key so a nudge prints at - * most once per turn. Returns undefined if there is no user message yet. */ -export function lastUserMessageId(entries: { id: string; message?: { role?: string } }[]): string | undefined { +// Synthetic user-message prefixes the injection-ledger turn key must skip. +// These are machinery injected via pi.sendUserMessage (throttle kicks, +// delegate notifications incl. appended recovery notices), not conversation. +// CAUTION: any NEW synthetic sendUserMessage injection site MUST add its +// prefix here too — otherwise it silently resets the per-turn injection +// budgets (see wireContextTransform in src/index.ts). +export const SYNTHETIC_USER_PREFIXES = [THROTTLE_KICK_SENTINEL, DELEGATE_NOTIFY_PREFIX] as const; + +/** Id of the last GENUINE user-role entry — used as the per-turn key for + * injection budgets. Synthetic user messages (throttle kicks, delegate + * notifications) are skipped: they are machinery, not conversation, and + * letting them rotate the turn key would reset the very budgets that bound + * runaway injection loops. Returns undefined if there is no user message. */ +export function lastUserMessageId(entries: { id: string; message?: { role?: string; content?: unknown } }[]): string | undefined { for (let i = entries.length - 1; i >= 0; i--) { const e = entries[i]!; - if (e.message?.role === "user") return e.id; + if (e.message?.role !== "user") continue; + const text = extractText(e.message.content).trimStart(); + if (SYNTHETIC_USER_PREFIXES.some((p) => text.startsWith(p))) continue; + return e.id; } return undefined; } diff --git a/tests/compress-retry.test.ts b/tests/compress-retry.test.ts index b507e93..46b07c8 100644 --- a/tests/compress-retry.test.ts +++ b/tests/compress-retry.test.ts @@ -2,28 +2,28 @@ import { test } from "node:test"; import assert from "node:assert/strict"; import { rm } from "node:fs/promises"; import { createAcpExtension } from "../src/index.js"; -import { createRuntime, MAX_COMPRESS_ATTEMPTS } from "../src/runtime.js"; -import { isCompressSuccessText, isCompressNoopText } from "../src/compress-tool.js"; - -// Compress-failure handling (session 01a00a38 post-mortem + issue #223): -// the model's ONLY compress call in a 3-hour session was rejected by pi's -// typebox validation ("content.0: must be object" — vLLM non-strict tools -// stringified the array). +import { createRuntime, MAX_EMERGENCY_NUDGES_PER_TURN } from "../src/runtime.js"; +import { lastUserMessageId } from "../src/tokens.js"; + +// Transient-injection governance (post-mortem for #223 + #6, unified ledger): +// pi rebuilds the sent context on every LLM call, so any transient injection +// re-appends per fire unless budgeted. #223 happened because the retry +// prompt's budget counted DISTINCT FAILED CALLS instead of INJECTIONS; +// #6 happened because the emergency nudge had no budget a no-op-looping +// model could reach. The ledger inverts that: budgets count INJECTIONS per +// GENUINE user turn (1 for normal nudges, MAX_EMERGENCY_NUDGES_PER_TURN for +// emergency), and NOTHING the model does (ignore, fail, no-op, neutral) can +// extend them — only a genuine user turn resets. // // Behavior under test: -// 1. compress-tool accepts a JSON-encoded string for content (root cause). -// 2. Argument errors THROW (pi only marks thrown tool errors isError:true — -// a returned error string would be isError:false: not counted + counter -// reset). -// 3. A failed compress toolResult persists in the session log (the model -// sees the error and can self-correct) but NO transient retry prompt is -// injected: per-LLM-call re-injection caused the #223 infinite-append -// loop (~400 injections/hour when the model never retries). -// 4. Failed/no-op outcomes are counted per user turn to drive the nudge -// circuit breaker (issue #6): once MAX_COMPRESS_ATTEMPTS attempts burn -// in one turn, the emergency nudge stops re-injecting. Neutral outcomes -// (non-error text that is not a success panel) neither reset nor advance -// the counter; success resets it; a new user turn gets a fresh budget. +// 1. compress-tool accepts a JSON-encoded string for content (root cause of +// session 01a00a38) and THROWS on garbage (isError:true). +// 2. Failed compress toolResults persist in the session log; NO transient +// retry prompt is ever injected (#223 regression). +// 3. Emergency nudges are bounded by INJECTIONS per turn regardless of the +// model's response mix (#6 + the neutral/no-response escape hatches). +// 4. Synthetic user messages (throttle kicks, delegate notifications) do +// NOT reset budgets; genuine user input does. function captureApi() { const handlers = new Map any)[]>(); @@ -59,10 +59,10 @@ function toolResultMsg(id: string, toolCallId: string, text: string, isError: bo // call looks like in entries — whether thrown by pi-ai validation or by // handleCompress's own argument checks). const VALIDATION_ERR = 'Validation failed for tool "compress":\n - content.0: must be object\n\nReceived arguments:\n{"content":"[{\\"topic\\":\\"x\\"}]"}'; -const SUCCESS_PANEL = "▣ ACP | 58.5K → 5.7K tokens (~52.8K reclaimed, 4 blocks)"; -const PARTIAL_PANEL = "▣ ACP | 58.5K → 30K tokens (~28.5K reclaimed, 3 blocks)\nErrors: range m00009..m00012: Summary too long"; const NOOP_PANEL = "▣ ACP | 58.5K → 58.5K tokens (~0 reclaimed, 0 blocks)\nErrors: range m00001..m00002: Requested range(s) already compressed; nothing to compress"; const NEUTRAL_TEXT = "No ranges provided."; +const KICK_TEXT = "[ACP:provider-throttle] The previous assistant response was interrupted by a provider rate limit."; +const DELEGATE_TEXT = "[acp_delegate done] ** researcher ** (runId `r1`, exit 0) result follows"; function fakeCtx(getEntries: () => any[], stateFile: string) { return { @@ -87,56 +87,50 @@ const retryMsgs = (r: any) => const ZH = "中".repeat(6000); -// ─── unit: runtime counter ────────────────────────────────────────────────── +// ─── unit: the injection ledger ───────────────────────────────────────────── -test("noteCompressOutcomes: counts, caps, resets on success, resets per turn, neutral freezes", () => { +test("noteInjection: counts injections, denies without incrementing, resets on genuine turn, exhaustion is one-shot", () => { const rt = createRuntime({}); - const fail = (id: string) => ({ toolCallId: id, isError: true, success: false }); - const success = (id: string) => ({ toolCallId: id, isError: false, success: true }); - const neutral = (id: string) => ({ toolCallId: id, isError: false, success: false }); - - let r = rt.noteCompressOutcomes("u1", [fail("t0")]); - assert.equal(r.count, 1); - assert.equal(r.cappedNow, false); - - // idempotent re-fire (same toolCallIds): count frozen - r = rt.noteCompressOutcomes("u1", [fail("t0")]); - assert.equal(r.count, 1, "no double count on re-fire"); - // neutral outcome: no reset - r = rt.noteCompressOutcomes("u1", [fail("t0"), neutral("n1")]); - assert.equal(r.count, 1, "neutral does not reset the counter"); + let r = rt.noteInjection("u1", "nudge", 1); + assert.deepEqual(r, { allowed: true, count: 1, exhaustedNow: true }, "budget 1 → allowed once, exhausted edge fires"); + r = rt.noteInjection("u1", "nudge", 1); + assert.equal(r.allowed, false, "denied after budget"); + assert.equal(r.count, 1, "denied calls do not increment"); - // a NEW failure after a neutral one: attempt 2, not 1 — neutral cannot - // bypass the cap by resetting between failures - r = rt.noteCompressOutcomes("u1", [fail("t0"), neutral("n1"), fail("t9")]); + r = rt.noteInjection("u1", "emergency", MAX_EMERGENCY_NUDGES_PER_TURN); + assert.equal(r.allowed, true); + assert.equal(r.exhaustedNow, false, "kinds are independent budgets"); + r = rt.noteInjection("u1", "emergency", MAX_EMERGENCY_NUDGES_PER_TURN); assert.equal(r.count, 2); + assert.equal(r.exhaustedNow, false); + r = rt.noteInjection("u1", "emergency", MAX_EMERGENCY_NUDGES_PER_TURN); + assert.deepEqual(r, { allowed: true, count: MAX_EMERGENCY_NUDGES_PER_TURN, exhaustedNow: true }, "third emergency reaches the cap"); + r = rt.noteInjection("u1", "emergency", MAX_EMERGENCY_NUDGES_PER_TURN); + assert.equal(r.allowed, false); + + // turn rollover resets every kind at once + r = rt.noteInjection("u2", "emergency", MAX_EMERGENCY_NUDGES_PER_TURN); + assert.deepEqual(r, { allowed: true, count: 1, exhaustedNow: false }, "new genuine turn → fresh budgets"); + // rolling back to an OLD turn key is also a rollover (fresh state) + r = rt.noteInjection("u1", "emergency", MAX_EMERGENCY_NUDGES_PER_TURN); + assert.equal(r.count, 1, "returning to a previous turn key still resets"); + + rt.clearInjectionLedger(); + r = rt.noteInjection("u1", "emergency", MAX_EMERGENCY_NUDGES_PER_TURN); + assert.equal(r.count, 1, "session_start clears the ledger"); +}); - // third distinct failure → cap, cappedNow fires once - r = rt.noteCompressOutcomes("u1", [fail("t0"), neutral("n1"), fail("t9"), fail("tc")]); - assert.equal(r.count, 3); - assert.equal(r.cappedNow, true); - r = rt.noteCompressOutcomes("u1", [fail("t0"), neutral("n1"), fail("t9"), fail("tc")]); - assert.equal(r.cappedNow, false, "cap notification is one-shot"); - assert.equal(MAX_COMPRESS_ATTEMPTS, 3); - - // success resets the counter - r = rt.noteCompressOutcomes("u1", [fail("t0"), neutral("n1"), fail("t9"), fail("tc"), success("ts")]); - assert.equal(r.count, 0); - - // a NEW failure after success counts a fresh cycle - r = rt.noteCompressOutcomes("u1", [fail("t0"), neutral("n1"), fail("t9"), fail("tc"), success("ts"), fail("td")]); - assert.equal(r.count, 1); - - // new user turn → fresh counter even without a success in between - r = rt.noteCompressOutcomes("u1", [fail("t0"), neutral("n1"), fail("t9"), fail("tc"), success("ts"), fail("td"), fail("te"), fail("tf")]); - assert.equal(r.count, 3, "back at cap"); - r = rt.noteCompressOutcomes("u2", [fail("x0")]); - assert.equal(r.count, 1); - - // a deduped stale failure must not count against a new turn - r = rt.noteCompressOutcomes("u3", [fail("x0")]); - assert.equal(r.count, 0, "stale id deduped, count stays 0 after turn change"); +test("lastUserMessageId skips synthetic machinery messages (throttle kicks, delegate notifications)", () => { + const entries = [ + userMsg("u1", "genuine question"), + userMsg("k1", KICK_TEXT), + userMsg("d1", DELEGATE_TEXT), + ]; + assert.equal(lastUserMessageId(entries as any), "u1", "synthetic users do not rotate the turn key"); + const withGenuine = [...entries, userMsg("u2", "next real question")]; + assert.equal(lastUserMessageId(withGenuine as any), "u2"); + assert.equal(lastUserMessageId([userMsg("only", KICK_TEXT)] as any), undefined, "all-synthetic → no genuine turn yet"); }); // ─── unit: normalizeRanges via the tool ───────────────────────────────────── @@ -162,7 +156,7 @@ test("compress tool accepts JSON-encoded string content (non-strict-tool provide await rm(`${stateFile}.acp.json`, { force: true }); }); -test("compress tool THROWS on garbage string content (isError:true → counted by the outcome tracker)", async () => { +test("compress tool THROWS on garbage string content (isError:true so the failure persists)", async () => { const { api, handlers } = captureApi(); createAcpExtension({ modelContextLimit: 200_000 })(api as any); const stateFile = "/tmp/pai-acp-retry-str2.session.json"; @@ -173,8 +167,7 @@ test("compress tool THROWS on garbage string content (isError:true → counted b const compressTool = api.tools.find((t: any) => t.name === "compress")!; // pi-agent-core marks only THROWN tool errors isError:true; returning the - // error string would be isError:false (not counted as a failure + counter - // reset), so the tool must reject. + // error string would be isError:false and silently success-shaped. await assert.rejects( () => compressTool.execute("tc1", { content: "not json {" }, undefined, undefined, ctx), /Invalid compress content[\s\S]*ARRAY/, @@ -208,15 +201,12 @@ test("failed compress toolResults never inject a transient retry prompt; the err const r2 = await fire(handlers, ctx); assert.equal(retryMsgs(r2).length, 0, "re-fire injects nothing"); - // second and third failures → cap burned for the nudge breaker, still no prompt entries = [...entries, toolResultMsg("e3", "call_2", VALIDATION_ERR, true)]; - const r3 = await fire(handlers, ctx); - assert.equal(retryMsgs(r3).length, 0); entries = [...entries, toolResultMsg("e4", "call_3", VALIDATION_ERR, true)]; - const r4 = await fire(handlers, ctx); - assert.equal(retryMsgs(r4).length, 0, "cap reached → still no prompt ever"); + const r3 = await fire(handlers, ctx); + assert.equal(retryMsgs(r3).length, 0, "cap reached → still no prompt ever"); - // later turns: a stale failure and a fresh one both stay silent + // later turns stay silent too entries = [...entries, userMsg("e5", "next question")]; for (let i = 0; i < 3; i++) { const r = await fire(handlers, ctx); @@ -228,84 +218,16 @@ test("failed compress toolResults never inject a transient retry prompt; the err await rm(`${stateFile}.acp.json`, { force: true }); }); -test("neutral and no-op outcomes inject nothing; only the counter state changes", async () => { - const { api, handlers } = captureApi(); - createAcpExtension({ modelContextLimit: 200_000 })(api as any); - const stateFile = "/tmp/pai-acp-retry-noop.session.json"; - await rm(`${stateFile}.acp.json`, { force: true }); - - let entries: any[] = [userMsg("e1", ZH)]; - const ctx = fakeCtx(() => entries, stateFile); - await fire(handlers, ctx); - - entries = [...entries, toolResultMsg("e2", "call_1", NOOP_PANEL, false)]; - const r1 = await fire(handlers, ctx); - assert.equal(retryMsgs(r1).length, 0, "no-op → no prompt"); - - entries = [...entries, toolResultMsg("e3", "call_2", NEUTRAL_TEXT, false)]; - const r2 = await fire(handlers, ctx); - assert.equal(retryMsgs(r2).length, 0, "neutral → no prompt"); - - entries = [...entries, toolResultMsg("e4", "call_3", NOOP_PANEL, false)]; - const r3 = await fire(handlers, ctx); - assert.equal(retryMsgs(r3).length, 0, "third no-op → capped breaker, no prompt"); - - entries = [...entries, toolResultMsg("e5", "call_4", SUCCESS_PANEL, false)]; - const r4 = await fire(handlers, ctx); - assert.equal(retryMsgs(r4).length, 0, "success → no prompt"); - await rm(`${stateFile}.acp.json`, { force: true }); -}); - -// ─── issue #6: no-op compress runs must not bypass the nudge cap ──────────── -// -// handleCompress returns a "▣ ACP | …" panel even when blocksCreated === 0 -// (every range skipped: already compressed / below min). The old -// isCompressSuccessText matched ANY panel prefix → no-op runs counted as -// success → counter reset → the (dedup-exempt) emergency nudge re-fired on -// every LLM call → unbounded emergency-nudge ↔ no-op-compress ping-pong. - -test("classification: 0-block panels are no-ops, not successes; >=1 block is success", () => { - assert.equal(isCompressSuccessText(SUCCESS_PANEL), true); - assert.equal(isCompressSuccessText(PARTIAL_PANEL), true, "partial errors with progress still count as success"); - assert.equal(isCompressSuccessText(NOOP_PANEL), false, "0-block panel must NOT be success (the issue #6 bug)"); - assert.equal(isCompressSuccessText(NEUTRAL_TEXT), false); - assert.equal(isCompressSuccessText("Validation failed"), false); - assert.equal(isCompressNoopText(NOOP_PANEL), true); - assert.equal(isCompressNoopText(SUCCESS_PANEL), false); - assert.equal(isCompressNoopText(PARTIAL_PANEL), false); - assert.equal(isCompressNoopText(NEUTRAL_TEXT), false, "non-panels stay neutral"); -}); - -test("noteCompressOutcomes: no-op panels advance the counter toward the cap", () => { - const rt = createRuntime({}); - const noop = (id: string) => ({ toolCallId: id, isError: false, success: false, noop: true }); - - let r = rt.noteCompressOutcomes("u1", [noop("t0")]); - assert.equal(r.count, 1); - - r = rt.noteCompressOutcomes("u1", [noop("t0"), noop("t1")]); - assert.equal(r.count, 2); +// ─── integration: emergency budget is bounded by injections (#6 + escapes) ── - r = rt.noteCompressOutcomes("u1", [noop("t0"), noop("t1"), noop("t2")]); - assert.equal(r.count, 3); - assert.equal(r.cappedNow, true); - assert.equal(rt.compressRetryCappedFor("u1"), true, "capped state is queryable per turn"); - assert.equal(rt.compressRetryCappedFor("u2"), false, "other turns are unaffected"); - - const success = (id: string) => ({ toolCallId: id, isError: false, success: true, noop: false }); - r = rt.noteCompressOutcomes("u1", [noop("t0"), noop("t1"), noop("t2"), success("ts")]); - assert.equal(r.count, 0, "genuine success lifts the cap"); - assert.equal(rt.compressRetryCappedFor("u1"), false); -}); - -test("emergency nudge stops re-injecting once the turn's cap is burned (issue #6 loop breaker)", async () => { +test("emergency nudge injects at most MAX_EMERGENCY_NUDGES_PER_TURN per turn, regardless of the model's response mix", async () => { const { api, handlers } = captureApi(); createAcpExtension({ modelContextLimit: 180_000 })(api as any); const stateFile = "/tmp/pai-acp-retry-emerg.session.json"; await rm(`${stateFile}.acp.json`, { force: true }); // ~270K tokens of sent view vs a 180K window → kernel goes EMERGENCY and - // the nudge re-injects on every context fire (dedup bypass). + // the nudge wants to re-inject on every context fire. const MID = "lorem ".repeat(3000); const roleMsg = (id: string, role: string, text: string) => ({ type: "message", id, parentId: null, timestamp: "", @@ -317,22 +239,37 @@ test("emergency nudge stops re-injecting once the turn's cap is burned (issue #6 const nudgeCount = (r: any) => (r?.messages ?? []).filter((m: any) => m.role === "user" && /Context limit reached/.test(JSON.stringify(m.content))).length; - const r0 = await fire(handlers, ctx); - assert.ok(nudgeCount(r0) >= 1, "emergency nudge fires on real overflow"); - assert.equal(retryMsgs(r0).length, 0); - - // model "answers" each emergency nudge with a no-op compress (stale refs) - for (let i = 1; i <= 3; i++) { - entries = [...entries, toolResultMsg(`ec${i}`, `call_${i}`, NOOP_PANEL, false)]; - await fire(handlers, ctx); + let injected = 0; + // adversarial loop: alternate every response shape the old budgets could + // not reach — hard failures, no-op panels, neutral "No ranges provided.", + // and plain silence (no compress result at all) + const responses = [ + () => toolResultMsg("ea", "call_a", VALIDATION_ERR, true), + () => toolResultMsg("eb", "call_b", NOOP_PANEL, false), + () => toolResultMsg("ec", "call_c", NEUTRAL_TEXT, false), + () => null, + ]; + for (let i = 0; i < 20; i++) { + const r = await fire(handlers, ctx); + injected += nudgeCount(r); + assert.ok(nudgeCount(r) <= 1, "at most one injection per fire"); + const mk = responses[i % responses.length]!; + const msg = mk(); + if (msg) entries = [...entries, msg]; } - const rCapped = await fire(handlers, ctx); - assert.equal(nudgeCount(rCapped), 0, "cap burned → emergency nudge no longer re-injects"); - assert.equal(retryMsgs(rCapped).length, 0, "no retry prompt either"); - - // a genuine success lifts the cap → emergency guidance resumes - entries = [...entries, toolResultMsg("ec4", "call_4", SUCCESS_PANEL, false)]; - const rRe = await fire(handlers, ctx); - assert.ok(nudgeCount(rRe) >= 1, "after a successful compress the emergency nudge may resume"); + assert.equal(injected, MAX_EMERGENCY_NUDGES_PER_TURN, `exactly the budget across 20 fires (got ${injected})`); + + // synthetic machinery messages must NOT re-arm the budget… + entries = [...entries, userMsg("ek", KICK_TEXT)]; + let r = await fire(handlers, ctx); + assert.equal(nudgeCount(r), 0, "throttle kick does not reset the emergency budget"); + entries = [...entries, userMsg("ed", DELEGATE_TEXT)]; + r = await fire(handlers, ctx); + assert.equal(nudgeCount(r), 0, "delegate notification does not reset the emergency budget"); + + // …but a genuine user message does + entries = [...entries, userMsg("e99", "actual next question")]; + r = await fire(handlers, ctx); + assert.equal(nudgeCount(r), 1, "genuine user turn → fresh emergency budget"); await rm(`${stateFile}.acp.json`, { force: true }); });