refactor(pi-compaction): 改为压力触发的观察遮蔽与空闲阈值整段摘要

原实现对所有工具结果(含 read)在下一次请求即替换为约 4K 的首尾存根,
不看上下文压力,导致模型看不到大文件中段;后台逐条语义摘要又反复改写
历史、打碎 prompt 缓存。参照 Anthropic 上下文编辑、Manus 与 JetBrains
观察遮蔽研究重构为:
- L2:上下文达到阈值时在 turn_end 批量提交 Pi 原生 context_edit,把最近
  keepRecentTurns 轮之外的旧 toolResult 换成确定性占位符;trigger/target
  回差与 minClearTokens 保证少触发、一次清够;原始条目保留,可经
  context_tree_query 取回
- L3:空闲 agent_settled 且超过绝对阈值时调用一次 ctx.compact
- 阈值按绝对 token 计并受窗口比例约束,默认针对 1M 窗口(DeepSeek V4.1
  Flash):遮蔽 200K→100K、整段摘要 400K
- 删除逐条存根、语义 worker、soft/hard/target 比例与自行取消 Pi 压缩的门控;
  Pi 原生阈值压缩保留为最终安全网

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
云服务部-叶林立
2026-10-21 19:51:36 -07:00
co-authored by Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
parent a254d4fd68
commit a6d8416c76
21 changed files with 1174 additions and 804 deletions
+174 -76
View File
@@ -14,6 +14,12 @@ import {
type SessionEntry,
} from "@earendil-works/pi-coding-agent";
import { DEFAULT_CONFIG, loadConfig } from "./src/config.js";
import {
applyContextEdits,
planMasking,
rebuildMaskedLedger,
type MaskedLedgerEntry,
} from "./src/masking.js";
import { buildNativeCompactionTransport, isNativeCompactionModel } from "./src/native-provider.js";
import {
buildCompactionRequestBody,
@@ -32,27 +38,28 @@ import {
type NativeCompactionDetails,
type ResponseItem,
} from "./src/native-compaction.js";
import { SegmentIndex } from "./src/segment-index.js";
import { SemanticWorker } from "./src/semantic-worker.js";
import { shouldCompact } from "./src/token-accounting.js";
import { contentText, SegmentIndex } from "./src/segment-index.js";
import {
CompactionServiceRegistry,
publishCompactionService,
unpublishCompactionService,
} from "./src/service.js";
import { effectiveThreshold } from "./src/token-accounting.js";
import {
COMPACTION_CUSTOM_TYPE,
COMPACTION_VERSION,
QUERY_TOOL_NAME,
type MaskedResultRecord,
type PiCompactionConfig,
type ProjectionSnapshot,
type ToolHistoryRecord,
} from "./src/types.js";
interface SessionRuntime {
index: SegmentIndex;
config: PiCompactionConfig;
codexCache?: { fingerprint: string; modelKey: string; toolsHash: string; items: ResponseItem[] };
masked: Map<string, MaskedLedgerEntry>;
lastAutoCompactLeafId: string | null;
}
interface CachedPayload {
@@ -64,10 +71,6 @@ function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
function hash(value: string): string {
return createHash("sha256").update(value).digest("hex").slice(0, 20);
}
interface EnvironmentValue {
hash: string;
tokens: number;
@@ -81,6 +84,10 @@ interface EnvironmentCacheEntry extends EnvironmentValue {
tools: unknown[];
}
function hash(value: string): string {
return createHash("sha256").update(value).digest("hex").slice(0, 20);
}
function environment(
pi: ExtensionAPI,
ctx: ExtensionContext,
@@ -182,34 +189,46 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
const runtimes = new Map<string, SessionRuntime>();
const payloads = new Map<string, CachedPayload>();
const environments = new Map<string, EnvironmentCacheEntry>();
const worker = new SemanticWorker();
const service = new CompactionServiceRegistry();
publishCompactionService(service);
const maskedRecords = (sessionId: string, ids: string[]): MaskedResultRecord[] => {
const masked = runtimes.get(sessionId)?.masked;
if (!masked) return [];
return ids.flatMap((id) => {
const entry = masked.get(id);
return entry
? [{
toolCallId: id,
resultTimestamp: entry.resultTimestamp,
deterministicSummary: entry.placeholder,
}]
: [];
});
};
const runtimeFor = (ctx: ExtensionContext): SessionRuntime => {
const sessionId = ctx.sessionManager.getSessionId();
let runtime = runtimes.get(sessionId);
if (!runtime) {
const index = new SegmentIndex(sessionId);
index.hydrate(ctx.sessionManager.getBranch() as SessionEntry[], DEFAULT_CONFIG);
runtime = { index, config: structuredClone(DEFAULT_CONFIG) };
runtime = {
index: new SegmentIndex(sessionId),
config: structuredClone(DEFAULT_CONFIG),
masked: rebuildLedger(ctx),
lastAutoCompactLeafId: null,
};
runtimes.set(sessionId, runtime);
service.setSession(sessionId, (ids) => index.query(ids));
service.setSession(sessionId, (ids) => maskedRecords(sessionId, ids));
}
return runtime;
};
const persist = (record: unknown): void => {
pi.appendEntry(COMPACTION_CUSTOM_TYPE, record);
};
const drainSemantic = (ctx: ExtensionContext, runtime: SessionRuntime): void => {
const sessionId = ctx.sessionManager.getSessionId();
for (const result of worker.drain(sessionId)) {
const record = runtime.index.applySemantic(result.sourceKey, result.sourceHash, result.summary);
if (record) persist(record);
const ledgerStats = (runtime: SessionRuntime): { maskedResults: number; reclaimedTokens: number } => {
let reclaimedTokens = 0;
for (const entry of runtime.masked.values()) {
reclaimedTokens += Math.max(0, entry.originalTokens - entry.placeholderTokens);
}
runtime.index.setSemanticJobs(worker.count(sessionId));
return { maskedResults: runtime.masked.size, reclaimedTokens };
};
const projectMessages = (
@@ -218,16 +237,32 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
messages: readonly any[],
): ProjectionSnapshot => {
const env = environment(pi, ctx, environments);
runtime.index.setSemanticJobs(worker.count(ctx.sessionManager.getSessionId()));
const stats = ledgerStats(runtime);
const snapshot = runtime.index.project(messages, {
environmentHash: env.hash,
environmentTokens: env.tokens,
contextWindow: env.contextWindow,
maskedResults: stats.maskedResults,
reclaimedTokens: stats.reclaimedTokens,
});
service.setSnapshot(snapshot);
return snapshot;
};
const rebuildLedger = (ctx: ExtensionContext): Map<string, MaskedLedgerEntry> =>
rebuildMaskedLedger(
ctx.sessionManager.getBranch(),
ctx.sessionManager.getLeafId(),
(message) => estimateTokens(message as any),
);
const hydrateRuntime = (ctx: ExtensionContext, runtime: SessionRuntime): void => {
runtime.index.hydrate();
runtime.masked = rebuildLedger(ctx);
runtime.lastAutoCompactLeafId = null;
runtime.codexCache = undefined;
};
const branchMessages = (
ctx: ExtensionContext,
runtime: SessionRuntime,
@@ -237,9 +272,10 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
const checkpoint = findNativeCheckpoint(branch);
let messages: any[];
if (checkpoint.status === "valid") {
messages = branch
.slice(checkpoint.checkpoint.entryIndex + 1)
.flatMap((entry) => sessionEntryToContextMessages(entry));
// sessionEntryToContextMessages sees only raw entries; applyContextEdits
// additionally replays the latest context_edit per target (omissions and
// placeholder replacements included).
messages = applyContextEdits(branch.slice(checkpoint.checkpoint.entryIndex + 1));
if (checkpoint.checkpoint.details.modelKey !== (ctx.model ? modelKey(ctx.model) : "")) {
messages = [asPortableMessage(checkpoint.checkpoint.details.portableSummary), ...messages];
}
@@ -278,15 +314,22 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
publishCompactionService(service);
const sessionId = ctx.sessionManager.getSessionId();
const config = await loadConfig();
const index = new SegmentIndex(sessionId);
index.hydrate(ctx.sessionManager.getBranch() as SessionEntry[], config);
runtimes.set(sessionId, { index, config });
service.setSession(sessionId, (ids) => index.query(ids));
const runtime = runtimes.get(sessionId) ?? {
index: new SegmentIndex(sessionId),
config,
masked: new Map<string, MaskedLedgerEntry>(),
lastAutoCompactLeafId: null,
};
runtime.config = config;
runtime.index.hydrate();
runtime.masked = rebuildLedger(ctx);
runtime.lastAutoCompactLeafId = null;
runtimes.set(sessionId, runtime);
service.setSession(sessionId, (ids) => maskedRecords(sessionId, ids));
});
pi.on("session_shutdown", (_event, ctx) => {
const sessionId = ctx.sessionManager.getSessionId();
worker.cancelSession(sessionId);
runtimes.delete(sessionId);
payloads.delete(sessionId);
environments.delete(sessionId);
@@ -296,15 +339,13 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
pi.on("session_tree", (_event, ctx) => {
const runtime = runtimeFor(ctx);
runtime.index.hydrate(ctx.sessionManager.getBranch() as SessionEntry[], runtime.config);
runtime.codexCache = undefined;
hydrateRuntime(ctx, runtime);
service.clearSnapshot(ctx.sessionManager.getSessionId());
});
pi.on("session_compact", (_event, ctx) => {
const runtime = runtimeFor(ctx);
runtime.index.hydrate(ctx.sessionManager.getBranch() as SessionEntry[], runtime.config);
runtime.codexCache = undefined;
hydrateRuntime(ctx, runtime);
service.clearSnapshot(ctx.sessionManager.getSessionId());
payloads.delete(ctx.sessionManager.getSessionId());
});
@@ -319,21 +360,9 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
pi.on("turn_end", (event, ctx) => {
const runtime = runtimeFor(ctx);
drainSemantic(ctx, runtime);
const previous = runtime.index.getSnapshot();
const captured = runtime.index.captureTurn(event.message, event.toolResults, runtime.config);
for (const record of captured) persist(runtime.index.toPersisted(record));
const pressure = previous && previous.contextWindow > 0
? previous.upperBound / previous.contextWindow >= runtime.config.softRatio
: false;
const semanticCandidates = pressure
? runtime.index.allRecords()
: captured.filter((record) => record.rawText.length >= runtime.config.semantic.maxInputChars);
for (const record of semanticCandidates) {
worker.enqueue(ctx.sessionManager.getSessionId(), record, ctx, runtime.config);
}
runtime.index.setSemanticJobs(worker.count(ctx.sessionManager.getSessionId()));
if (!runtime.config.enabled) return undefined;
const previous = runtime.index.getSnapshot();
const usage = ctx.getContextUsage();
if (usage?.tokens && previous) {
const env = environment(pi, ctx, environments);
@@ -343,6 +372,50 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
env.hash,
);
}
const currentTokens = usage?.tokens
?? (event.context?.contextMessages ?? []).reduce((sum, message) => sum + estimateTokens(message as any), 0)
+ environment(pi, ctx, environments).tokens;
const plan = planMasking({
contextEntries: event.context?.contextEntries ?? [],
currentTokens,
contextWindow: ctx.model?.contextWindow ?? 0,
masking: runtime.config.masking,
estimate: (message) => estimateTokens(message as any),
});
if (!plan.drafts.length) return undefined;
for (const result of plan.masked) {
runtime.masked.set(result.toolCallId, {
placeholder: result.placeholder,
originalTokens: result.originalTokens,
placeholderTokens: result.placeholderTokens,
resultTimestamp: result.resultTimestamp,
});
}
return { entries: [...(event.entries ?? []), ...plan.drafts] };
});
pi.on("agent_settled", (_event, ctx) => {
const runtime = runtimeFor(ctx);
if (!runtime.config.enabled || ctx.isIdle?.() === false) return;
const snapshot = runtime.index.getSnapshot();
const tokens = ctx.getContextUsage()?.tokens ?? snapshot?.effectiveTokens;
if (tokens == null) return;
const contextWindow = ctx.model?.contextWindow ?? snapshot?.contextWindow ?? 0;
const threshold = effectiveThreshold(
runtime.config.summary.triggerTokens,
runtime.config.summary.triggerRatio,
contextWindow,
);
if (tokens < threshold) return;
const leafId = ctx.sessionManager.getLeafId();
if (leafId !== null && runtime.lastAutoCompactLeafId === leafId) return;
runtime.lastAutoCompactLeafId = leafId;
ctx.compact({
onError: (error) => {
if (ctx.hasUI) ctx.ui.notify(`pi-compaction auto summary failed: ${errorMessage(error)}`, "warning");
},
});
});
pi.on("context", (event, ctx) => {
@@ -358,7 +431,10 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
const projected = branchMessages(ctx, runtime, branch);
return { messages: projected.snapshot.messages as any };
}
return { messages: projectMessages(ctx, runtime, event.messages).messages as any };
// Non-checkpoint path only refreshes the shared snapshot; the message list
// is returned to Pi unchanged — masking is append-only via context_edit.
projectMessages(ctx, runtime, event.messages);
return undefined;
});
pi.on("before_provider_headers", (event, ctx) => {
@@ -387,7 +463,6 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
pi.on("session_before_compact", async (event, ctx) => {
const runtime = runtimeFor(ctx);
if (!runtime.config.enabled) return undefined;
drainSemantic(ctx, runtime);
const { snapshot, checkpoint } = branchMessages(
ctx,
runtime,
@@ -395,8 +470,6 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
event.reason === "overflow" && event.willRetry,
);
if (!shouldCompact(event.reason, snapshot, runtime.config)) return { cancel: true };
const model = ctx.model;
if (!model) return undefined;
if (!runtime.config.native.enabled || !isNativeCompactionModel(model)) {
@@ -412,7 +485,7 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
event.signal,
event.customInstructions,
undefined,
runtime.config.semantic.thinking,
runtime.config.summary.thinking,
);
return {
compaction: {
@@ -463,7 +536,7 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
replacementHistory: buildReplacementHistory(
input,
remote.compactionItem,
Math.min(64_000, Math.max(4_096, Math.floor(snapshot.contextWindow * Math.max(0.05, runtime.config.targetRatio - 0.3)))),
Math.min(64_000, Math.max(4_096, Math.floor(snapshot.contextWindow * 0.25))),
),
portableSummary: portableSummary(snapshot.messages),
sourceFingerprint: snapshot.fingerprint,
@@ -504,7 +577,7 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
event.signal,
event.preparation.customInstructions,
undefined,
runtime.config.semantic.thinking,
runtime.config.summary.thinking,
undefined,
undefined,
undefined,
@@ -531,38 +604,64 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
pi.registerTool({
name: QUERY_TOOL_NAME,
label: "Query Compacted Tool History",
description: "Recover original tool outputs by source key or tool call id after pi-compaction replaced them with bounded summaries.",
promptSnippet: "Recover original compacted tool evidence",
description: "Recover the original output of a tool result that pi-compaction cleared from context, by tool call id.",
promptSnippet: "Recover original tool output cleared from context",
promptGuidelines: [
"Use context_tree_query only when a pi-compaction summary lacks evidence needed for the current task.",
"Use context_tree_query when you see a \"[pi-compaction] Earlier\" placeholder and still need the original tool output.",
],
parameters: Type.Object({
toolCallIds: Type.Array(Type.String(), { minItems: 1, maxItems: 20 }),
}),
async execute(_id, params, _signal, _update, ctx) {
const records = service.query(ctx.sessionManager.getSessionId(), params.toolCallIds);
const blocks = records.map((record: ToolHistoryRecord) => {
const perResultBytes = Math.max(1_024, Math.floor(DEFAULT_MAX_BYTES / Math.max(1, records.length)));
const args = truncateHead(JSON.stringify(record.args), { maxLines: 20, maxBytes: 2_048 }).content;
const truncated = truncateHead(record.rawText, {
maxLines: Math.max(20, Math.floor(DEFAULT_MAX_LINES / Math.max(1, records.length))),
const branch = ctx.sessionManager.getBranch() as SessionEntry[];
const calls = new Map<string, { toolName: string; args: unknown }>();
const results = new Map<string, any>();
for (const entry of branch) {
const message = (entry as any)?.message;
if ((entry as any)?.type !== "message" || !message) continue;
if (message.role === "assistant" && Array.isArray(message.content)) {
for (const block of message.content) {
if (block?.type === "toolCall" && typeof block.id === "string") {
calls.set(block.id, {
toolName: String(block.name ?? "unknown"),
args: block.arguments ?? block.input ?? block.args ?? {},
});
}
}
} else if (message.role === "toolResult" && typeof message.toolCallId === "string") {
results.set(message.toolCallId, message);
}
}
const found: string[] = [];
const blocks: string[] = [];
for (const id of params.toolCallIds) {
const result = results.get(id);
if (!result) continue;
found.push(id);
const call = calls.get(id) ?? { toolName: String(result.toolName ?? "unknown"), args: {} };
const perResultBytes = Math.max(1_024, Math.floor(DEFAULT_MAX_BYTES / Math.max(1, params.toolCallIds.length)));
const args = truncateHead(JSON.stringify(call.args ?? {}), { maxLines: 20, maxBytes: 2_048 }).content;
const rawText = contentText(result.content);
const truncated = truncateHead(rawText, {
maxLines: Math.max(20, Math.floor(DEFAULT_MAX_LINES / Math.max(1, params.toolCallIds.length))),
maxBytes: perResultBytes,
});
return [
`## ${record.sourceKey}`,
`Tool: ${record.toolName}`,
blocks.push([
`## ${id}`,
`Tool: ${call.toolName}`,
`Args: ${args}`,
`Status: ${record.isError ? "ERROR" : "OK"}`,
`Status: ${result.isError ? "ERROR" : "OK"}`,
"",
truncated.content,
...(truncated.truncated ? [`[truncated: ${truncated.outputLines}/${truncated.totalLines} lines]`] : []),
].join("\n");
});
const missing = params.toolCallIds.filter((id: string) => !records.some((record) => record.sourceKey === id || record.toolCallId === id));
].join("\n"));
}
const missing = params.toolCallIds.filter((id: string) => !results.has(id));
for (const id of missing) blocks.push(`## ${id}\n(not found)`);
return {
content: [{ type: "text", text: blocks.join("\n\n---\n\n") }],
details: { sourceKeys: records.map((record) => record.sourceKey), missing },
details: { toolCallIds: found, missing },
};
},
});
@@ -576,12 +675,11 @@ export default function piCompactionExtension(pi: ExtensionAPI): void {
`Effective: ${metrics.effectiveTokens}/${metrics.contextWindow} (${metrics.percent.toFixed(1)}%, ${metrics.confidence})`,
`Raw: ${metrics.rawTokens}/${metrics.contextWindow}`,
`Reclaimed: ${metrics.reclaimedTokens}`,
`Segments: ${metrics.segmentCount}; semantic jobs: ${metrics.semanticJobs}; generation: ${metrics.generation}`,
`Segments: ${metrics.segmentCount}; masked tool results: ${metrics.maskedResults}; generation: ${metrics.generation}`,
].join("\n")
: "pi-compaction has no projection snapshot yet.";
if (ctx.hasUI) ctx.ui.notify(text, "info");
else process.stderr.write(`${text}\n`);
},
});
}