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
1 change: 1 addition & 0 deletions reference/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -205,6 +205,7 @@ When navigating this codebase, these are the load-bearing entry points:
- **Update path** — `src/update.ts` is deliberately self-contained and conservative; it installs via tarball extraction (not `npm install`) to avoid lifecycle-script surprises on user machines. It is load-bearing infrastructure: changes require a no-op validation release first (§5).
- **Agent plugins** — `src/agent/*.ts` are thin host-specific adapters; shared logic lives in `src/agent/shared.ts`. Each host has a `-native.ts` bootstrap gate variant.
- **Persistence** — `src/persist.ts` wraps the acp-kernel `StateStore`; `src/session.ts` holds the in-memory session pool. Session identity rules: `src/session-id.ts`.
- **会话派生缓存** — `Session.transient` 持有插件工具视图与诊断请求副本;随会话淘汰释放,也纳入会话对象图的字节计量。它不写入持久化记录或上游协议;诊断截段独立复制 UTF-16 内容,避免保留完整请求底层字符串(#2920)。
- **Wire fidelity** — `src/stream-*.ts` files handle per-protocol streaming transformation. The invariant: never alter upstream protocol shape beyond intended injection (§7.3).
- **E2E regression** — `tests/e2e/` contains real-client suites (codex, opencode, pi, dsh) plus hermetic lanes (registry, advisory-rollback, release-canary, opencode WS, image billing); each carries its own skip gate — read the test header for the exact env var.
- **Compression kernel** — `kernel/` (in-repo acp-kernel, #2092). Human-gated boundary; see `AGENTS.md` "Kernel Boundary".
Expand Down
1 change: 1 addition & 0 deletions scripts/run-ci-tests.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ const testsDir = join(root, "tests");
// spawning suites MUST be added here after reviewing their concurrency
// assumptions.
const LIVE_FILES = [
"tests/session-transient-memory.test.ts",
"tests/auto-restart.test.ts",
"tests/build-info.test.ts",
"tests/cache-theorem.test.ts",
Expand Down
34 changes: 18 additions & 16 deletions src/cache-ledger.ts
Original file line number Diff line number Diff line change
Expand Up @@ -330,7 +330,7 @@ const BODY_FULL_CAP = 8 * 1024 * 1024;
* FULL body (the capped head cannot distinguish two payloads sharing 512 KiB);
* `full` keeps the whole payload while it fits BODY_FULL_CAP so divergent
* pairs can report their first-diverging byte offset. */
interface SeamSlot {
export interface SeamSlot {
str: string;
msgs: number | null;
digest: string;
Expand All @@ -340,8 +340,6 @@ interface SeamSlot {
* re-notes the same wireBody without headers in hand). */
keyFp?: string;
}
const seamLastSent = new WeakMap<Session, SeamSlot>();
const seamLastSettled = new WeakMap<Session, SeamSlot>();
const lastClientAbort = new WeakMap<Session, number>();

interface ContextObservation {
Expand Down Expand Up @@ -408,21 +406,25 @@ export function credentialFingerprint(headers: Record<string, string> | undefine
* the outbound credential fingerprint (#2131) — likewise inherited from the
* prior slot on a byte-identical re-note without a fresh capture. */
export function noteForwardedBody(session: Session, body: string, msgs?: number | null, keyFp?: string): void {
const str = body.length > SEAM_BODY_CAP ? body.slice(0, SEAM_BODY_CAP) : body;
// V8 的 substring 可保留整个请求;UTF-16 拷贝独立分配且保留孤立 surrogate。
const str = Buffer.from(body.slice(0, SEAM_BODY_CAP), "utf16le").toString("utf16le");
// #2131: hash the FULL body — the capped head alone cannot tell two payloads
// sharing a 512 KiB prefix apart. Re-noting a byte-identical payload (the
// non-streaming settle path) reuses the prior slot's count/full instead of
// re-deriving them.
const digest = createHash("sha256").update(body, "utf8").digest("hex");
const prior = seamLastSent.get(session);
const transient = session.transient ??= {};
const prior = transient.sentBody;
const samePayload = prior !== undefined && prior.digest === digest;
seamLastSent.set(session, {
transient.sentBody = {
str,
msgs: msgs ?? (samePayload ? prior.msgs : null),
digest,
full: body.length <= BODY_FULL_CAP ? body : samePayload ? prior.full : null,
full: body.length <= SEAM_BODY_CAP ? str
: body.length <= BODY_FULL_CAP ? Buffer.from(body, "utf16le").toString("utf16le")
: samePayload ? prior.full : null,
keyFp: keyFp ?? (samePayload ? prior.keyFp : undefined),
});
};
// prepare may reuse a measured baseline; this new payload is not measured yet.
if (session.stats.contextTokensSource !== undefined) recordContextObservation(session, session.stats.contextTokens, "estimate");
else delete session.metadata?.publicContextObservation;
Expand Down Expand Up @@ -555,8 +557,8 @@ function detectSeam(session: Session, led: CacheLedger): void {
// Substantive unexplained residual only: a big ttlRepay slice of a big bill.
if (!(line.tr > 8192 && line.tr > 0.3 * line.input)) return;
const agg = led.agg;
const cur = seamLastSent.get(session);
const prev = seamLastSettled.get(session);
const cur = session.transient?.sentBody;
const prev = session.transient?.settledBody;
let f: ReturnType<typeof seamLcp> | undefined;
if (cur !== undefined && prev !== undefined) {
f = seamLcp(prev.str, cur.str);
Expand Down Expand Up @@ -1145,7 +1147,7 @@ export function settleUsageReport(
// #2131: the credential fingerprint captured at the send chokepoint rides
// the seam slot — read it BEFORE recordCacheSample so the line it pushes
// already carries the identity (kp/kw) without a second slot lookup.
const curKeyFp = seamLastSent.get(session)?.keyFp;
const curKeyFp = session.transient?.sentBody?.keyFp;
// #2350: the host's system-prompt fingerprint as noted for THIS request by
// fold-reconcile's wire sites (they precede this settle under the session
// lock). Sparse: lanes that never noted one (title-gen / compaction-trigger
Expand All @@ -1165,12 +1167,12 @@ export function settleUsageReport(
// settled one), so "prefix unchanged between calls" is stored data, not an
// interpretation. Digest equality decides; the LCP offset localizes where
// a divergence started when both full bodies were retained.
const curSlot = seamLastSent.get(session);
const curSlot = session.transient?.sentBody;
if (curSlot !== undefined) {
const curLine = led.lines[led.lines.length - 1];
if (curLine !== undefined) {
curLine.bd = curSlot.digest;
const prevSlot = seamLastSettled.get(session);
const prevSlot = session.transient?.settledBody;
if (prevSlot !== undefined) {
if (curSlot.digest === prevSlot.digest) {
curLine.be = 1;
Expand Down Expand Up @@ -1198,9 +1200,9 @@ export function settleUsageReport(
// actually sent (noteForwardedBody), then keep it as the next pair's
// baseline. Lanes without body capture still get the aggregate flag.
detectSeam(session, led);
const seamBody = seamLastSent.get(session);
if (seamBody !== undefined) seamLastSettled.set(session, seamBody);
seamLastSent.delete(session);
const seamBody = session.transient?.sentBody;
if (seamBody !== undefined) (session.transient ??= {}).settledBody = seamBody;
delete session.transient?.sentBody;
}

/** [#1279] Price profile stamped by the last request (server.ts runPrepare).
Expand Down
29 changes: 13 additions & 16 deletions src/plugin.ts
Original file line number Diff line number Diff line change
Expand Up @@ -223,12 +223,10 @@ export function pluginHeadersMatchModel(headers: Record<string, string | string[
}

type ConversationEntry = { sessionId: string; lastSeen: number };
type RememberedMessages = { processed: CoreMessage[]; original: CoreMessage[]; nudge?: NudgeDecision };

const MAX_PLUGIN_CONVERSATIONS = 1024;

const conversations = new Map<string, ConversationEntry>();
const remembered = new Map<string, RememberedMessages>();
// #2077: which content-store object each session's stored-refs list was last
// evaluated against. Store updates are immutable (storeOriginal returns a new
// object), so a changed object identity is the exact signal that the refs
Expand Down Expand Up @@ -360,21 +358,19 @@ export function rememberPluginMessages(sessionId: string, processed: CoreMessage
// round; the next resending turn self-heals. Fresh sessions and larger or
// overlapping views always write.
const incoming = processed.length > 0 ? processed : original;
const previous = remembered.get(sessionId);
const session = peekSession(sessionId);
if (!session) return;
const transient = session.transient ??= {};
const previous = transient.pluginMessages;
if (previous) {
const previousView = previous.processed.length > 0 ? previous.processed : previous.original;
if (previousView.length > incoming.length) {
const knownIds = new Set(previousView.map((m) => m.id));
if (!incoming.some((m) => knownIds.has(m.id))) return;
}
}
const staleSessionIds = new Set(
[...remembered.keys()].filter((id) => id === sessionId || !peekSession(id)),
);
for (const id of staleSessionIds) remembered.delete(id);
remembered.set(sessionId, { processed, original, nudge });
const session = peekSession(sessionId);
if (session && typeof session.metadata.pluginAgent === "string" && original.length > 0) {
transient.pluginMessages = { processed, original, nudge };
if (typeof session.metadata.pluginAgent === "string" && original.length > 0) {
// #2077: copy-on-write snapshot maintenance. A raw id is a SHA-256 of
// the message identity plus a deterministic within-conversation
// duplicate-cluster index (acp-kernel wire/message-id.ts), but ids
Expand Down Expand Up @@ -1276,7 +1272,7 @@ export async function handlePluginFork(payload: string, res: ServerResponse): Pr
// re-bind plugin mode on the child's next request, which the map
// binding already does without dragging the derived-parent path in.
registeredIds.delete(request.childConversationId);
remembered.set(child.id, { processed: structuredClone(child.pluginSnapshot), original: structuredClone(child.pluginSnapshot) });
(child.transient ??= {}).pluginMessages = { processed: structuredClone(child.pluginSnapshot), original: structuredClone(child.pluginSnapshot) };
// #2077: the parent snapshot is now an external contract — this child
// was cut from it, and later forks/replays depend on its continuity.
// The receipt lives on the CHILD only, so without this sticky flag a
Expand Down Expand Up @@ -1415,7 +1411,7 @@ export function handlePluginStatus(conversationId: string, res: import("node:htt
const contextTokens = observation?.tokens ?? null;
let sessionRevision: string | null = null;
try { sessionRevision = sessionRevisionOf(session); } catch {}
const mem = remembered.get(session.id);
const mem = session.transient?.pluginMessages;
const modelContextLimit = typeof limit === "number" && limit > 0 ? limit : 0;
// #387: the remembered nudge is a prepare-time snapshot. A compress tool
// executed after the last model request mutates state without re-running
Expand Down Expand Up @@ -1728,7 +1724,7 @@ export async function handlePluginTool(
// request rewrites remembered atomically under this same lock
// (rememberPluginMessages), so a racing tool call sees a consistent
// state instead of a stale/empty window.
const mem = remembered.get(session.id);
const mem = session.transient?.pluginMessages;
const messages = mem ? (mem.processed.length > 0 ? mem.processed : mem.original) : (session.pluginSnapshot ?? []);
const before = currentContextObservation(session);
const creditBefore = session.stats.compressCreditTokens ?? 0;
Expand Down Expand Up @@ -4136,14 +4132,15 @@ function recordJsonToolWitnesses(sessionId: string, json: Record<string, unknown

export function _resetPluginStateForTest(): void {
conversations.clear();
remembered.clear();
for (const session of listSessions()) delete session.transient?.pluginMessages;
pendingRegisters.length = 0;
registeredIds.clear();
pluginRuntimeTable.clear();
pluginRuntimeByConversation.clear();
warnedNoModelRequests.clear();
}

export function _rememberedForTest(): Map<string, RememberedMessages> {
return remembered;
export function _rememberedForTest(): ReadonlyMap<string, NonNullable<NonNullable<Session["transient"]>["pluginMessages"]>> {
return new Map(listSessions().flatMap((session) => session.transient?.pluginMessages
? [[session.id, session.transient.pluginMessages] as const] : []));
}
9 changes: 8 additions & 1 deletion src/session.ts
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import { createInitialState, defaultConfig, highestUsedIndex, indexToRef, resetImageFullState, type CompressionState, type Config, type CoreMessage, type MessageContentStore } from "acp-kernel";
import { createInitialState, defaultConfig, highestUsedIndex, indexToRef, resetImageFullState, type CompressionState, type Config, type CoreMessage, type MessageContentStore, type NudgeDecision } from "acp-kernel";
import { createHash } from "node:crypto";
import { log as loggerLog } from "./logger.js";
import { getStore } from "./persist.js";
import { maxSessions as knobMaxSessions } from "./knobs.js";
import { durableMessageToolPaths, withDurableProtection } from "./durable-message-guards.js";
import type { AnonymousProvenance } from "./anon-provenance.js";
import type { WireProtocol } from "./util.js";
import type { SeamSlot } from "./cache-ledger.js";

export type BlockView = { text: string; count: number };

Expand Down Expand Up @@ -60,6 +61,12 @@ export type PendingRetrieval = {
};

export type Session = {
/** 派生缓存随会话释放,供驻留字节计量遍历;不进入持久化投影或 wire。 */
transient?: {
pluginMessages?: { processed: CoreMessage[]; original: CoreMessage[]; nudge?: NudgeDecision };
sentBody?: SeamSlot;
settledBody?: SeamSlot;
};
id: string;
/** Identity / descriptive metadata. Populated on first request. */
meta: {
Expand Down
1 change: 1 addition & 0 deletions tests/issue1307-side-snapshot.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ const MODEL = "claude-sonnet-4-5";

test("rememberPluginMessages: a zero-overlap smaller view never evicts the snapshot (#1307)", () => {
_resetPluginStateForTest();
for (const id of ["iso-a", "iso-b", "iso-c"]) getSession(id);
const msg = (id: string, text: string) => ({ id, role: "user", contentType: "text", text }) as const;
const mainView = Array.from({ length: 8 }, (_, i) => msg(`raw-${i}`, `main ${i}`));
// Fresh session: a 1-message view writes (no previous snapshot to protect).
Expand Down
4 changes: 2 additions & 2 deletions tests/issue2029-status-baseline-provenance.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ import type { CoreMessage } from "acp-kernel";
import { createCore, createInitialState, assignRefs, emptyRefMap, defaultConfig } from "acp-kernel";
import { zeroStats, _resetSessionsForTest, getSession, statusInputBaseline, type Session } from "../src/session.ts";
import { handleAcpStatus } from "../src/acp-status.ts";
import { handlePluginStatus, _resetPluginStateForTest, _rememberedForTest } from "../src/plugin.ts";
import { handlePluginStatus, _resetPluginStateForTest, rememberPluginMessages } from "../src/plugin.ts";

const WINDOW = 150000;

Expand Down Expand Up @@ -140,7 +140,7 @@ test("#2029: plugin status nudge + panel feed the provenance-aware baseline", ()
const messages = twelveMessages();
const refs = assignRefs(messages, { existing: emptyRefMap(), nextIndex: 0 });
session.state.messageRefs = refs.map;
_rememberedForTest().set(session.id, { processed: messages, original: messages });
rememberPluginMessages(session.id, messages, messages);

const deps = { core: createCore(), config: defaultConfig(WINDOW), log: () => {} };
const seen = captureTokenCounts(deps.core);
Expand Down
Loading
Loading