diff --git a/src/agents/agent-hooks/compaction-instructions.test.ts b/src/agents/agent-hooks/compaction-instructions.test.ts
index 67ab15bc003a..75c5545c70b6 100644
--- a/src/agents/agent-hooks/compaction-instructions.test.ts
+++ b/src/agents/agent-hooks/compaction-instructions.test.ts
@@ -1,9 +1,6 @@
/** Tests compaction instruction defaults, precedence, and split-turn composition. */
import { describe, expect, it } from "vitest";
-import {
- resolveCompactionInstructions,
- composeSplitTurnInstructions,
-} from "./compaction-instructions.js";
+import { resolveCompactionInstructions } from "./compaction-instructions.js";
const DEFAULT_COMPACTION_INSTRUCTIONS = resolveCompactionInstructions(undefined, undefined);
@@ -188,56 +185,3 @@ describe("resolveCompactionInstructions", () => {
});
});
});
-
-describe("composeSplitTurnInstructions", () => {
- it("joins turn prefix, separator, and resolved instructions with double newlines", () => {
- const result = composeSplitTurnInstructions("Turn prefix here", "Resolved instructions here");
- expect(result).toBe(
- "Turn prefix here\n\nAdditional requirements:\n\nResolved instructions here",
- );
- });
-
- it("output contains the turn prefix verbatim", () => {
- const prefix = "Summarize the last 5 messages.";
- const result = composeSplitTurnInstructions(prefix, "Keep it short.");
- expect(result).toContain(prefix);
- });
-
- it("output contains the resolved instructions verbatim", () => {
- const instructions = "Write in Korean. Preserve persona.";
- const result = composeSplitTurnInstructions("prefix", instructions);
- expect(result).toContain(instructions);
- });
-
- it("output contains 'Additional requirements:' separator", () => {
- const result = composeSplitTurnInstructions("a", "b");
- expect(result).toContain("Additional requirements:");
- });
-
- it("KNOWN_EDGE: empty turnPrefix produces leading blank line", () => {
- const result = composeSplitTurnInstructions("", "instructions");
- expect(result).toBe("\n\nAdditional requirements:\n\ninstructions");
- expect(result.startsWith("\n")).toBe(true);
- });
-
- it("KNOWN_EDGE: empty resolvedInstructions produces trailing blank area", () => {
- const result = composeSplitTurnInstructions("prefix", "");
- expect(result).toBe("prefix\n\nAdditional requirements:\n\n");
- expect(result.endsWith("\n\n")).toBe(true);
- });
-
- it("does not deduplicate if instructions already contain 'Additional requirements:'", () => {
- const instructions = "Additional requirements: keep it short.";
- const result = composeSplitTurnInstructions("prefix", instructions);
- const count = (result.match(/Additional requirements:/g) || []).length;
- expect(count).toBe(2);
- });
-
- it("preserves multiline content in both inputs", () => {
- const prefix = "Line 1\nLine 2";
- const instructions = "Rule A\nRule B\nRule C";
- const result = composeSplitTurnInstructions(prefix, instructions);
- expect(result).toContain("Line 1\nLine 2");
- expect(result).toContain("Rule A\nRule B\nRule C");
- });
-});
diff --git a/src/agents/agent-hooks/compaction-instructions.ts b/src/agents/agent-hooks/compaction-instructions.ts
index 6a1fc736e747..135c951f6e56 100644
--- a/src/agents/agent-hooks/compaction-instructions.ts
+++ b/src/agents/agent-hooks/compaction-instructions.ts
@@ -4,6 +4,7 @@
* Provides default language-preservation instructions and a precedence-based
* resolver for customInstructions used during context compaction summaries.
*/
+import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
/**
* Default instructions injected into every safeguard-mode compaction summary.
@@ -22,22 +23,6 @@ const DEFAULT_COMPACTION_INSTRUCTIONS =
*/
const MAX_INSTRUCTION_LENGTH = 800;
-function truncateUnicodeSafe(s: string, maxCodePoints: number): string {
- const chars = Array.from(s);
- if (chars.length <= maxCodePoints) {
- return s;
- }
- return chars.slice(0, maxCodePoints).join("");
-}
-
-function normalize(s: string | undefined): string | undefined {
- if (s == null) {
- return undefined;
- }
- const trimmed = s.trim();
- return trimmed.length > 0 ? trimmed : undefined;
-}
-
/**
* Resolve compaction instructions with precedence:
* event (SDK) → runtime (config) → DEFAULT constant.
@@ -50,19 +35,8 @@ export function resolveCompactionInstructions(
runtimeInstructions: string | undefined,
): string {
const resolved =
- normalize(eventInstructions) ??
- normalize(runtimeInstructions) ??
+ normalizeOptionalString(eventInstructions) ??
+ normalizeOptionalString(runtimeInstructions) ??
DEFAULT_COMPACTION_INSTRUCTIONS;
- return truncateUnicodeSafe(resolved, MAX_INSTRUCTION_LENGTH);
-}
-
-/**
- * Compose split-turn instructions by combining the SDK's turn-prefix
- * instructions with the resolved compaction instructions.
- */
-export function composeSplitTurnInstructions(
- turnPrefixInstructions: string,
- resolvedInstructions: string,
-): string {
- return [turnPrefixInstructions, "Additional requirements:", resolvedInstructions].join("\n\n");
+ return Array.from(resolved).slice(0, MAX_INSTRUCTION_LENGTH).join("");
}
diff --git a/src/agents/agent-hooks/compaction-safeguard-quality.ts b/src/agents/agent-hooks/compaction-safeguard-quality.ts
index 0d99dd74910d..3c9d14706b1e 100644
--- a/src/agents/agent-hooks/compaction-safeguard-quality.ts
+++ b/src/agents/agent-hooks/compaction-safeguard-quality.ts
@@ -94,10 +94,7 @@ function hasRequiredSummarySections(summary: string): boolean {
}
/** Return a structured fallback summary when model output is missing/invalid. */
-export function buildStructuredFallbackSummary(
- previousSummary: string | undefined,
- _summarizationInstructions?: CompactionSummarizationInstructions,
-): string {
+export function buildStructuredFallbackSummary(previousSummary: string | undefined): string {
const trimmedPreviousSummary = previousSummary?.trim() ?? "";
if (trimmedPreviousSummary && hasRequiredSummarySections(trimmedPreviousSummary)) {
return trimmedPreviousSummary;
@@ -153,12 +150,10 @@ export function extractOpaqueIdentifiers(text: string): string[] {
text.match(
/([A-Fa-f0-9]{8,}|https?:\/\/\S+|\/[\w.-]{2,}(?:\/[\w.-]+)+|[A-Za-z]:\\[\w\\.-]+|[A-Za-z0-9._-]+\.[A-Za-z0-9._/-]+:\d{1,5}|\b\d{6,}\b)/g,
) ?? [];
- return Array.from(
- new Set(
- matches
- .map((value) => normalizeOpaqueIdentifier(sanitizeExtractedIdentifier(value)))
- .filter((value) => value.length >= 4),
- ),
+ return uniqueStrings(
+ matches
+ .map((value) => normalizeOpaqueIdentifier(sanitizeExtractedIdentifier(value)))
+ .filter((value) => value.length >= 4),
).slice(0, MAX_EXTRACTED_IDENTIFIERS);
}
diff --git a/src/agents/agent-hooks/compaction-safeguard-runtime.ts b/src/agents/agent-hooks/compaction-safeguard-runtime.ts
index 13e393c9d008..539206ee9ed1 100644
--- a/src/agents/agent-hooks/compaction-safeguard-runtime.ts
+++ b/src/agents/agent-hooks/compaction-safeguard-runtime.ts
@@ -49,14 +49,9 @@ export function setCompactionSafeguardCancelReason(
const current = getCompactionSafeguardRuntime(sessionManager);
const trimmed = reason?.trim();
- if (!current) {
- if (!trimmed) {
- return;
- }
- setCompactionSafeguardRuntime(sessionManager, { cancelReason: trimmed });
+ if (!current && !trimmed) {
return;
}
-
const next = { ...current };
if (trimmed) {
next.cancelReason = trimmed;
diff --git a/src/agents/agent-hooks/compaction-safeguard.test.ts b/src/agents/agent-hooks/compaction-safeguard.test.ts
index c018126cca3f..ef3e50f25ad8 100644
--- a/src/agents/agent-hooks/compaction-safeguard.test.ts
+++ b/src/agents/agent-hooks/compaction-safeguard.test.ts
@@ -1439,9 +1439,7 @@ describe("compaction-safeguard recent-turn preservation", () => {
});
it("does not force policy-off marker in fallback exact identifiers section", () => {
- const summary = buildStructuredFallbackSummary(undefined, {
- identifierPolicy: "off",
- });
+ const summary = buildStructuredFallbackSummary(undefined);
expect(summary).toContain("## Exact identifiers");
expect(summary).toContain("None captured.");
expect(summary).not.toContain("N/A (identifier policy off).");
@@ -2143,6 +2141,9 @@ describe("compaction-safeguard recent-turn preservation", () => {
expect(summary).toContain("latest ask status");
expect(summary).toContain("latest assistant reply");
expect(mockSummarizeInStages).toHaveBeenCalledTimes(3);
+ expect(requireRecord(mockCallArg(mockSummarizeInStages, 1)).customInstructions).toContain(
+ "Additional requirements:",
+ );
});
it("keeps required headings when all turns are preserved and history is carried forward", async () => {
@@ -2845,70 +2846,86 @@ describe("compaction-safeguard double-compaction guard", () => {
).toBe(true);
});
- it("does not replay inter-session sessions_send branch turns as fallback history", async () => {
- mockSummarizeInStages.mockReset();
- mockSummarizeInStages.mockResolvedValue(summaryResult("branch summary"));
+ it.each([
+ { assistantText: "bee reply", completed: true },
+ { assistantText: " \t\n ", completed: false },
+ ])(
+ "drops delegated branch turns only after meaningful terminal output ($completed)",
+ async ({ assistantText, completed }) => {
+ mockSummarizeInStages.mockReset();
+ mockSummarizeInStages.mockResolvedValue(summaryResult("branch summary"));
- const now = Date.now();
- const sessionManager = {
- ...stubSessionManager(),
- getBranch: () => [
- {
- type: "message",
- id: "user-1",
- parentId: null,
- timestamp: new Date(now).toISOString(),
- message: {
- role: "user",
- content: "say bee",
- provenance: {
- kind: "inter_session",
- sourceSessionKey: "agent:pm",
- sourceTool: "sessions_send",
+ const now = Date.now();
+ const sessionManager = {
+ ...stubSessionManager(),
+ getBranch: () => [
+ {
+ type: "message",
+ id: "user-1",
+ parentId: null,
+ timestamp: new Date(now).toISOString(),
+ message: {
+ role: "user",
+ content: "say bee",
+ provenance: {
+ kind: "inter_session",
+ sourceSessionKey: "agent:pm",
+ sourceTool: "sessions_send",
+ },
+ timestamp: now,
},
- timestamp: now,
},
- },
- {
- type: "message",
- id: "assistant-1",
- parentId: "user-1",
- timestamp: new Date(now + 1).toISOString(),
- message: {
- role: "assistant",
- content: [{ type: "text", text: "bee reply" }],
- timestamp: now + 1,
+ {
+ type: "message",
+ id: "assistant-1",
+ parentId: "user-1",
+ timestamp: new Date(now + 1).toISOString(),
+ message: {
+ role: "assistant",
+ content: [{ type: "text", text: assistantText }],
+ timestamp: now + 1,
+ },
},
+ ],
+ } as ExtensionContext["sessionManager"];
+ const model = createAnthropicModelFixture();
+ setCompactionSafeguardRuntime(sessionManager, { model, recentTurnsPreserve: 0 });
+
+ const mockEvent = {
+ preparation: {
+ messagesToSummarize: [] as AgentMessage[],
+ turnPrefixMessages: [] as AgentMessage[],
+ firstKeptEntryId: "entry-7",
+ tokensBefore: 38085,
+ fileOps: { read: [], edited: [], written: [] },
+ settings: { reserveTokens: 4000 },
+ isSplitTurn: true,
},
- ],
- } as ExtensionContext["sessionManager"];
- const model = createAnthropicModelFixture();
- setCompactionSafeguardRuntime(sessionManager, { model, recentTurnsPreserve: 0 });
+ customInstructions: "",
+ signal: new AbortController().signal,
+ };
+ const { result, getApiKeyAndHeadersMock } = await runCompactionScenario({
+ sessionManager,
+ event: mockEvent,
+ apiKey: "dummy",
+ });
- const mockEvent = {
- preparation: {
- messagesToSummarize: [] as AgentMessage[],
- turnPrefixMessages: [] as AgentMessage[],
- firstKeptEntryId: "entry-7",
- tokensBefore: 38085,
- fileOps: { read: [], edited: [], written: [] },
- settings: { reserveTokens: 4000 },
- isSplitTurn: true,
- },
- customInstructions: "",
- signal: new AbortController().signal,
- };
- const { result, getApiKeyAndHeadersMock } = await runCompactionScenario({
- sessionManager,
- event: mockEvent,
- apiKey: "dummy",
- });
-
- const compaction = expectCompactionResult(result);
- expect(compaction.summary).toContain("No prior history.");
- expect(mockSummarizeInStages).not.toHaveBeenCalled();
- expect(getApiKeyAndHeadersMock).not.toHaveBeenCalled();
- });
+ const compaction = expectCompactionResult(result);
+ if (completed) {
+ expect(compaction.summary).toContain("No prior history.");
+ expect(mockSummarizeInStages).not.toHaveBeenCalled();
+ expect(getApiKeyAndHeadersMock).not.toHaveBeenCalled();
+ return;
+ }
+ expect(compaction.summary).toContain("branch summary");
+ expect(mockSummarizeInStages).toHaveBeenCalledTimes(1);
+ expect(getApiKeyAndHeadersMock).toHaveBeenCalledTimes(1);
+ const summarizeCall = requireRecord(mockCallArg(mockSummarizeInStages));
+ expect(
+ requireArray(summarizeCall.messages).map((message) => requireRecord(message).role),
+ ).toEqual(["user", "assistant"]);
+ },
+ );
it.each([
{ toolName: "read", expectedRoles: ["user", "assistant", "toolResult"] },
diff --git a/src/agents/agent-hooks/compaction-safeguard.ts b/src/agents/agent-hooks/compaction-safeguard.ts
index 3eb63ff3cd5f..636e1a1ff647 100644
--- a/src/agents/agent-hooks/compaction-safeguard.ts
+++ b/src/agents/agent-hooks/compaction-safeguard.ts
@@ -2,6 +2,7 @@
import fs from "node:fs";
import path from "node:path";
+import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { sliceUtf16Safe, truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import { extractSections } from "../../auto-reply/reply/post-compaction-context.js";
import { isAbortError } from "../../infra/abort-signal.js";
@@ -14,10 +15,8 @@ import {
} from "../../plugins/compaction-provider.js";
import { normalizeInputProvenance } from "../../sessions/input-provenance.js";
import { normalizeAcceptedSessionSpawnResult } from "../accepted-session-spawn.js";
-import {
- buildHistoryPrunePlanWithWorker,
- computeAdaptiveChunkRatioWithWorker,
-} from "../compaction-planning-worker.js";
+import { computeAdaptiveChunkRatioWithWorker } from "../compaction-planning-worker.js";
+import { buildHistoryPrunePlan } from "../compaction-planning.js";
import {
hasMeaningfulConversationContent,
isRealConversationMessage,
@@ -44,10 +43,7 @@ import {
MAX_WORKSPACE_BOOTSTRAP_FILE_BYTES,
readWorkspaceBootstrapFile,
} from "../workspace-bootstrap-read.js";
-import {
- composeSplitTurnInstructions,
- resolveCompactionInstructions,
-} from "./compaction-instructions.js";
+import { resolveCompactionInstructions } from "./compaction-instructions.js";
import {
appendSummarySection,
auditSummaryQuality,
@@ -87,19 +83,6 @@ const compactionSafeguardDeps = {
summarizeInStages,
};
-function buildPreviousSummaryMessage(previousSummary: string): AgentMessage {
- return {
- role: "user",
- content: [
- {
- type: "text",
- text: `\n${PREVIOUS_SUMMARY_REDISTILL_PREFIX}\n\n${previousSummary.trim()}\n`,
- },
- ],
- timestamp: 0,
- } as AgentMessage;
-}
-
function prependPreviousSummaryForRedistill(params: {
messages: AgentMessage[];
previousSummary?: string;
@@ -108,27 +91,27 @@ function prependPreviousSummaryForRedistill(params: {
if (!previousSummary) {
return params.messages;
}
- return [buildPreviousSummaryMessage(previousSummary), ...params.messages];
+ return [
+ {
+ role: "user",
+ content: [
+ {
+ type: "text",
+ text: `\n${PREVIOUS_SUMMARY_REDISTILL_PREFIX}\n\n${previousSummary}\n`,
+ },
+ ],
+ timestamp: 0,
+ } as AgentMessage,
+ ...params.messages,
+ ];
}
-type SessionBranchEntry = {
- type?: unknown;
- message?: unknown;
- customType?: unknown;
- content?: unknown;
- display?: unknown;
- details?: unknown;
- timestamp?: unknown;
- summary?: unknown;
- fromId?: unknown;
-};
-
function coerceTimestamp(value: unknown): number {
const timestamp = typeof value === "string" ? Date.parse(value) : value;
return typeof timestamp === "number" && Number.isFinite(timestamp) ? timestamp : 0;
}
-function sessionBranchEntryToMessage(entry: SessionBranchEntry): unknown {
+function sessionBranchEntryToMessage(entry: Record): unknown {
if (entry.type === "message" && entry.message && typeof entry.message === "object") {
return entry.message;
}
@@ -154,17 +137,13 @@ function sessionBranchEntryToMessage(entry: SessionBranchEntry): unknown {
}
function collectSessionBranchMessages(sessionManager: unknown): AgentMessage[] {
- const getBranch = (sessionManager as { getBranch?: unknown })?.getBranch;
- if (typeof getBranch !== "function") {
- return [];
- }
try {
- const entries: unknown = getBranch.call(sessionManager);
+ const entries: unknown = (sessionManager as { getBranch?: () => unknown })?.getBranch?.();
return Array.isArray(entries)
? entries.flatMap((entry) => {
const message =
entry && typeof entry === "object"
- ? sessionBranchEntryToMessage(entry as SessionBranchEntry)
+ ? sessionBranchEntryToMessage(entry as Record)
: undefined;
return message ? [message as AgentMessage] : [];
})
@@ -174,42 +153,27 @@ function collectSessionBranchMessages(sessionManager: unknown): AgentMessage[] {
}
}
-function isReplayUnsafeInterSessionInput(message: AgentMessage): boolean {
- if ((message as { role?: unknown }).role !== "user") {
- return false;
- }
- const provenance = normalizeInputProvenance((message as { provenance?: unknown }).provenance);
- return provenance?.kind === "inter_session" && provenance.sourceTool === "sessions_send";
-}
-
function isSessionsSendToolName(value: unknown): boolean {
- if (typeof value !== "string") {
- return false;
- }
return (
- value
- .trim()
- .toLowerCase()
+ normalizeOptionalString(value)
+ ?.toLowerCase()
.replace(/^(?:functions?|tools?)[./_-]/, "") === "sessions_send"
);
}
function sanitizeSourceSessionSends(messages: AgentMessage[]): AgentMessage[] {
- const sendCallIds = new Set();
+ const sendCallIds = new Set(
+ messages.flatMap((message) =>
+ message.role === "assistant"
+ ? extractToolCallsFromAssistant(message)
+ .filter((call) => isSessionsSendToolName(call.name))
+ .map((call) => call.id.trim())
+ .filter(Boolean)
+ : [],
+ ),
+ );
const resultTextByCallId = new Map();
- for (const message of messages) {
- if (message.role !== "assistant") {
- continue;
- }
- for (const call of extractToolCallsFromAssistant(message)) {
- const callId = call.id.trim();
- if (callId && isSessionsSendToolName(call.name)) {
- sendCallIds.add(callId);
- }
- }
- }
-
for (const message of messages) {
if (message.role !== "toolResult") {
continue;
@@ -249,14 +213,10 @@ function sanitizeSourceSessionSends(messages: AgentMessage[]): AgentMessage[] {
const resultText = callId ? resultTextByCallId.get(callId) : undefined;
const resolved = Boolean(callId && resultTextByCallId.has(callId));
const requestText = JSON.stringify({ callId: callId || undefined, args: record.arguments });
+ const resultSuffix = resolved ? `\nResult: ${resultText || "[empty]"}` : "";
return {
type: "text",
- text:
- resolved && resultText
- ? `sessions_send result received; delivery call omitted from replay.\nRequest: ${requestText}\nResult: ${resultText}`
- : resolved
- ? `sessions_send result received; delivery call omitted from replay.\nRequest: ${requestText}\nResult: [empty]`
- : `sessions_send result missing; delivery call omitted from replay.\nRequest: ${requestText}`,
+ text: `sessions_send result ${resolved ? "received" : "missing"}; delivery call omitted from replay.\nRequest: ${requestText}${resultSuffix}`,
};
});
return replaced ? [{ ...message, content } as AgentMessage] : [message];
@@ -296,6 +256,10 @@ function filterReplayUnsafeSessionBranchMessages(messages: AgentMessage[]): Agen
return typeof type === "string" && TOOL_CALL_BLOCK_TYPES.has(type);
}));
const activeInput = sanitizedMessages[turnStart - 1];
+ const activeInputProvenance =
+ activeInput?.role === "user"
+ ? normalizeInputProvenance((activeInput as { provenance?: unknown }).provenance)
+ : undefined;
// A completed sessions_send target run is already delivered to its caller.
// Require terminal text so compaction after tool output can still recover unfinished work.
@@ -303,8 +267,8 @@ function filterReplayUnsafeSessionBranchMessages(messages: AgentMessage[]): Agen
endsWithTerminalAssistantText &&
turnStart < sanitizedMessages.length &&
turnStart > 0 &&
- activeInput !== undefined &&
- isReplayUnsafeInterSessionInput(activeInput)
+ activeInputProvenance?.kind === "inter_session" &&
+ activeInputProvenance.sourceTool === "sessions_send"
) {
return sanitizedMessages.slice(0, turnStart - 1);
}
@@ -348,16 +312,13 @@ function assembleSuffix(parts: {
fileOpsSummary?: string;
workspaceContext?: string;
}): string {
- let suffix = Object.values(parts).reduce(
+ const suffix = Object.values(parts).reduce(
(summary, section) => appendSummarySection(summary, section ?? ""),
"",
);
// Ensure leading separator so suffix does not merge with body (e.g. when body
// ends without newline: "...## Exact identifiers## Tool Failures").
- if (suffix && !/^\s/.test(suffix)) {
- suffix = `\n\n${suffix}`;
- }
- return suffix;
+ return suffix && !/^\s/.test(suffix) ? `\n\n${suffix}` : suffix;
}
type ToolFailure = {
@@ -447,51 +408,42 @@ function buildCompactionSummaryHeaders(params: {
};
}
-function clampNonNegativeInt(value: unknown, fallback: number): number {
+function clampNonNegativeInt(
+ value: unknown,
+ fallback: number,
+ max = Number.POSITIVE_INFINITY,
+): number {
const normalized = typeof value === "number" && Number.isFinite(value) ? value : fallback;
- return Math.max(0, Math.floor(normalized));
+ return Math.min(max, Math.max(0, Math.floor(normalized)));
}
function resolveRecentTurnsPreserve(value: unknown): number {
- return Math.min(
- MAX_RECENT_TURNS_PRESERVE,
- clampNonNegativeInt(value, DEFAULT_RECENT_TURNS_PRESERVE),
- );
+ return clampNonNegativeInt(value, DEFAULT_RECENT_TURNS_PRESERVE, MAX_RECENT_TURNS_PRESERVE);
}
function resolveQualityGuardMaxRetries(value: unknown): number {
- return Math.min(
+ return clampNonNegativeInt(
+ value,
+ DEFAULT_QUALITY_GUARD_MAX_RETRIES,
MAX_QUALITY_GUARD_MAX_RETRIES,
- clampNonNegativeInt(value, DEFAULT_QUALITY_GUARD_MAX_RETRIES),
);
}
-function normalizeFailureText(text: string): string {
- return text.replace(/\s+/g, " ").trim();
-}
-
-function truncateFailureText(text: string, maxChars: number): string {
- if (text.length <= maxChars) {
- return text;
- }
- return `${truncateUtf16Safe(text, Math.max(0, maxChars - 3))}...`;
-}
-
function formatToolFailureMeta(details: unknown): string | undefined {
if (!details || typeof details !== "object") {
return undefined;
}
const record = details as Record;
- const status = typeof record.status === "string" ? record.status : undefined;
- const exitCode =
- typeof record.exitCode === "number" && Number.isFinite(record.exitCode)
- ? record.exitCode
- : undefined;
- const parts = [
- status ? `status=${status}` : "",
- exitCode !== undefined ? `exitCode=${exitCode}` : "",
- ];
- return parts.filter(Boolean).join(" ") || undefined;
+ return (
+ [
+ typeof record.status === "string" && record.status ? `status=${record.status}` : "",
+ typeof record.exitCode === "number" && Number.isFinite(record.exitCode)
+ ? `exitCode=${record.exitCode}`
+ : "",
+ ]
+ .filter(Boolean)
+ .join(" ") || undefined
+ );
}
function collectToolFailures(messages: AgentMessage[]): ToolFailure[] {
@@ -530,13 +482,14 @@ function collectToolFailures(messages: AgentMessage[]): ToolFailure[] {
typeof toolResult.toolName === "string" && toolResult.toolName.trim()
? toolResult.toolName
: "tool";
- const rawText = collectTextContentBlocks(toolResult.content).join("\n");
const meta = formatToolFailureMeta(toolResult.details);
- const normalized = normalizeFailureText(rawText);
- const summary = truncateFailureText(
- normalized || (meta ? "failed" : "failed (no output)"),
- MAX_TOOL_FAILURE_CHARS,
- );
+ const failureText =
+ collectTextContentBlocks(toolResult.content).join("\n").replace(/\s+/g, " ").trim() ||
+ (meta ? "failed" : "failed (no output)");
+ const summary =
+ failureText.length > MAX_TOOL_FAILURE_CHARS
+ ? `${truncateUtf16Safe(failureText, MAX_TOOL_FAILURE_CHARS - 3)}...`
+ : failureText;
failures.push({ toolCallId, toolName, summary, meta });
}
@@ -671,10 +624,7 @@ function extractMessageText(message: AgentMessage): string {
}
function formatNonTextPlaceholder(content: unknown): string | null {
- if (content === null || content === undefined) {
- return null;
- }
- if (typeof content === "string") {
+ if (content == null || typeof content === "string") {
return null;
}
if (!Array.isArray(content)) {
@@ -692,22 +642,21 @@ function formatNonTextPlaceholder(content: unknown): string | null {
}
typeCounts.set(type, (typeCounts.get(type) ?? 0) + 1);
}
- if (typeCounts.size === 0) {
- return null;
- }
- const parts = [...typeCounts.entries()].map(([type, count]) =>
- count > 1 ? `${type} x${count}` : type,
- );
- return `[non-text content: ${parts.join(", ")}]`;
+ return typeCounts.size > 0
+ ? `[non-text content: ${Array.from(typeCounts, ([type, count]) =>
+ count > 1 ? `${type} x${count}` : type,
+ ).join(", ")}]`
+ : null;
}
function splitPreservedRecentTurns(params: {
messages: AgentMessage[];
recentTurnsPreserve: number;
}): { summarizableMessages: AgentMessage[]; preservedMessages: AgentMessage[] } {
- const preserveTurns = Math.min(
+ const preserveTurns = clampNonNegativeInt(
+ params.recentTurnsPreserve,
+ 0,
MAX_RECENT_TURNS_PRESERVE,
- clampNonNegativeInt(params.recentTurnsPreserve, 0),
);
if (preserveTurns <= 0) {
return { summarizableMessages: params.messages, preservedMessages: [] };
@@ -722,18 +671,13 @@ function splitPreservedRecentTurns(params: {
const userIndexes = conversationIndexes.filter(
(index) => params.messages[index]?.role === "user",
);
- const preservedIndexSet = new Set();
- if (userIndexes.length >= preserveTurns) {
- const boundaryStartIndex = userIndexes[userIndexes.length - preserveTurns] ?? -1;
- for (const index of conversationIndexes) {
- if (index >= boundaryStartIndex) {
- preservedIndexSet.add(index);
- }
- }
- } else {
- for (const userIndex of userIndexes) {
- preservedIndexSet.add(userIndex);
- }
+ const boundaryStartIndex = userIndexes.at(-preserveTurns);
+ const preservedIndexSet = new Set(
+ boundaryStartIndex === undefined
+ ? userIndexes
+ : conversationIndexes.filter((index) => index >= boundaryStartIndex),
+ );
+ if (boundaryStartIndex === undefined) {
for (const index of conversationIndexes.toReversed()) {
preservedIndexSet.add(index);
if (preservedIndexSet.size >= preserveTurns * 2) {
@@ -791,12 +735,12 @@ function formatContextMessages(messages: AgentMessage[]): string[] {
} else {
return null;
}
- const text = extractMessageText(message);
- const nonTextPlaceholder = formatNonTextPlaceholder(
- (message as { content?: unknown }).content,
- );
- const rendered =
- text && nonTextPlaceholder ? `${text}\n${nonTextPlaceholder}` : text || nonTextPlaceholder;
+ const rendered = [
+ extractMessageText(message),
+ formatNonTextPlaceholder((message as { content?: unknown }).content),
+ ]
+ .filter(Boolean)
+ .join("\n");
if (!rendered) {
return null;
}
@@ -1107,14 +1051,13 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
let droppedSummary: string | undefined;
if (tokensBefore !== undefined) {
- const prunePlan = await buildHistoryPrunePlanWithWorker({
+ const prunePlan = buildHistoryPrunePlan({
messagesToSummarize,
turnPrefixMessages,
tokensBefore,
contextWindowTokens,
maxHistoryShare,
parts: 2,
- signal,
});
const { newContentTokens, maxHistoryTokens, pruned } = prunePlan;
@@ -1171,18 +1114,15 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
});
messagesToSummarize = summaryTargetMessages;
const preservedTurnsSectionLocal = formatPreservedTurnsSection(preservedRecentMessages);
- const latestUserAsk = extractLatestUserAsk([...messagesToSummarize, ...turnPrefixMessages]);
- const identifierSeedText = [...messagesToSummarize, ...turnPrefixMessages]
- .slice(-10)
- .map((message) => extractMessageText(message))
- .filter(Boolean)
- .join("\n");
- const identifiers = extractOpaqueIdentifiers(identifierSeedText);
+ const allMessages = [...messagesToSummarize, ...turnPrefixMessages];
+ const latestUserAsk = extractLatestUserAsk(allMessages);
+ const identifiers = extractOpaqueIdentifiers(
+ allMessages.slice(-10).map(extractMessageText).filter(Boolean).join("\n"),
+ );
// Use adaptive chunk ratio based on message sizes, reserving headroom for
// the summarization prompt, system prompt, previous summary, and reasoning budget
// that generateSummary adds on top of the serialized conversation chunk.
- const allMessages = [...messagesToSummarize, ...turnPrefixMessages];
const adaptiveRatio = await computeAdaptiveChunkRatioWithWorker({
messages: allMessages,
contextWindow: contextWindowTokens,
@@ -1196,7 +1136,6 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
// incorporates context from pruned messages instead of losing it entirely.
const effectivePreviousSummary = droppedSummary ?? preparation.previousSummary;
- let summary = "";
let lastHistorySummary = "";
let lastSplitTurnSection = "";
let currentInstructions = structuredInstructions;
@@ -1218,7 +1157,7 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
customInstructions: currentInstructions,
previousSummary: effectivePreviousSummary,
})
- : buildStructuredFallbackSummary(effectivePreviousSummary, summarizationInstructions);
+ : buildStructuredFallbackSummary(effectivePreviousSummary);
summaryWithoutPreservedTurns = historySummary;
if (preparation.isSplitTurn && turnPrefixMessages.length > 0) {
@@ -1226,10 +1165,7 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
...llmSummaryParams,
messages: turnPrefixMessages,
maxChunkTokens,
- customInstructions: composeSplitTurnInstructions(
- TURN_PREFIX_INSTRUCTIONS,
- currentInstructions,
- ),
+ customInstructions: `${TURN_PREFIX_INSTRUCTIONS}\n\nAdditional requirements:\n\n${currentInstructions}`,
previousSummary: undefined,
});
splitTurnSectionLocal = `**Turn Context (split turn):**\n\n${prefixSummary}`;
@@ -1247,7 +1183,6 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
`Compaction safeguard: quality retry failed on attempt ${attempt + 1}; ` +
`keeping last successful summary: ${formatErrorMessage(attemptError)}`,
);
- summary = lastSuccessfulSummary;
break;
}
throw attemptError;
@@ -1260,7 +1195,6 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
messagesToSummarize.length > 0 ||
(preparation.isSplitTurn && turnPrefixMessages.length > 0);
if (!qualityGuardEnabled || !canRegenerate) {
- summary = summaryWithPreservedTurns;
break;
}
const quality = auditSummaryQuality({
@@ -1269,7 +1203,6 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
latestAsk: latestUserAsk,
identifierPolicy,
});
- summary = summaryWithPreservedTurns;
if (quality.ok || attempt >= totalAttempts - 1) {
break;
}
@@ -1288,7 +1221,7 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
}
// Cap history before suffixes so diagnostics and workspace rules survive.
- return await finalizeSummary(lastHistorySummary || summary, {
+ return await finalizeSummary(lastHistorySummary || lastSuccessfulSummary || "", {
splitTurnSection: lastSplitTurnSection,
preservedTurnsSection: preservedTurnsSectionLocal,
});
diff --git a/src/agents/compaction-planning-projection.ts b/src/agents/compaction-planning-projection.ts
index 6825379cd912..2583ab6a93f6 100644
--- a/src/agents/compaction-planning-projection.ts
+++ b/src/agents/compaction-planning-projection.ts
@@ -126,28 +126,21 @@ function jsonLengthWithin(
return length;
}
-function projectToolArguments(
- value: unknown,
- budget: ProjectionBudget,
-): { value: Record; omittedChars: number; changed: boolean } {
+function projectToolArguments(value: unknown, budget: ProjectionBudget): number | undefined {
const length = jsonLengthWithin(value, budget.remainingChars);
if (length !== undefined) {
budget.remainingChars -= length;
- return { value: {}, omittedChars: 0, changed: false };
+ return undefined;
}
budget.remainingChars = 0;
- return {
- value: {},
- // Unmeasurable arguments must force an oversized plan, never understate token pressure.
- omittedChars:
- jsonLengthWithin(value, MAX_ARGUMENT_ESTIMATE_CHARS) ?? UNMEASURABLE_ARGUMENT_OMITTED_CHARS,
- changed: true,
- };
+ // Unmeasurable arguments must force an oversized plan, never understate token pressure.
+ return (
+ jsonLengthWithin(value, MAX_ARGUMENT_ESTIMATE_CHARS) ?? UNMEASURABLE_ARGUMENT_OMITTED_CHARS
+ );
}
function projectContentBlock(
block: unknown,
- projectTextContent: boolean,
budget: ProjectionBudget,
): { block: unknown; omittedChars: number; changed: boolean } {
if (!block || typeof block !== "object") {
@@ -162,9 +155,6 @@ function projectContentBlock(
changed: true,
};
}
- if (!projectTextContent) {
- return { block, omittedChars: 0, changed: false };
- }
const hasText = typeof record.text === "string" && record.text.length > 0;
const textIsModelVisible =
type === "text" || ((type === "toolResult" || type === "tool_result") && hasText);
@@ -172,50 +162,41 @@ function projectContentBlock(
(type === "toolResult" || type === "tool_result") &&
!hasText &&
typeof record.content === "string";
- const projectedText = typeof record.text === "string" ? projectText(record.text, budget) : null;
- const projectedContent =
- typeof record.content === "string" ? projectText(record.content, budget) : null;
- const projectedThinking =
- type === "thinking" && typeof record.thinking === "string"
- ? projectText(record.thinking, budget)
- : null;
- const projectedArguments =
- type === "toolCall" ? projectToolArguments(record.arguments, budget) : undefined;
- // Tool-call IDs are provider-generated and bounded in practice. Planning never uses them as
- // durable keys, so malformed giant IDs do not justify an ID-remapping protocol here.
- const hasPlanningIrrelevantSignature =
- "textSignature" in record || "thinkingSignature" in record || "thoughtSignature" in record;
- if (
- !projectedText &&
- !projectedContent &&
- !projectedThinking &&
- !projectedArguments?.changed &&
- !hasPlanningIrrelevantSignature
- ) {
- return { block, omittedChars: 0, changed: false };
- }
- const next = { ...record };
+ let next: Record | undefined;
let omittedChars = 0;
- if (projectedText) {
- next.text = projectedText.text;
- omittedChars += textIsModelVisible ? projectedText.omittedChars : 0;
+ for (const field of ["text", "content", "thinking"] as const) {
+ if (field === "thinking" && type !== "thinking") {
+ continue;
+ }
+ const value = record[field];
+ const projected = typeof value === "string" ? projectText(value, budget) : null;
+ if (!projected) {
+ continue;
+ }
+ next ??= { ...record };
+ next[field] = projected.text;
+ const modelVisible =
+ field === "thinking" || (field === "text" ? textIsModelVisible : contentIsModelVisible);
+ omittedChars += modelVisible ? projected.omittedChars : 0;
}
- if (projectedContent) {
- next.content = projectedContent.text;
- omittedChars += contentIsModelVisible ? projectedContent.omittedChars : 0;
+ if (type === "toolCall") {
+ const omittedArguments = projectToolArguments(record.arguments, budget);
+ if (omittedArguments !== undefined) {
+ next ??= { ...record };
+ next.arguments = {};
+ omittedChars += omittedArguments;
+ }
}
- if (projectedThinking) {
- next.thinking = projectedThinking.text;
- omittedChars += projectedThinking.omittedChars;
+ // Signatures never contribute model-visible compaction text and can dwarf the planning payload.
+ for (const signature of ["textSignature", "thinkingSignature", "thoughtSignature"]) {
+ if (signature in record) {
+ next ??= { ...record };
+ delete next[signature];
+ }
}
- if (projectedArguments?.changed) {
- next.arguments = projectedArguments.value;
- omittedChars += projectedArguments.omittedChars;
- }
- delete next.textSignature;
- delete next.thinkingSignature;
- delete next.thoughtSignature;
- return { block: next, omittedChars, changed: true };
+ return next
+ ? { block: next, omittedChars, changed: true }
+ : { block, omittedChars, changed: false };
}
function projectStringFields(
@@ -245,43 +226,24 @@ function projectStringFields(
}
function projectMessage(message: AgentMessage, budget: ProjectionBudget): AgentMessage {
- const source = (() => {
- switch (message.role) {
- case "assistant":
- return {
- role: message.role,
- content: message.content,
- stopReason: message.stopReason,
- timestamp: message.timestamp,
- } as AgentMessage;
- case "bashExecution": {
- const { fullOutputPath: _, ...rest } = message;
- return rest as AgentMessage;
- }
- case "compactionSummary": {
- const { details: _, ...rest } = message;
- return rest as AgentMessage;
- }
- case "custom": {
- const { details: _, ...rest } = message;
- return rest as AgentMessage;
- }
- default:
- return message;
- }
- })();
- const currentOmittedChars = readCompactionPlanningOmittedChars(source);
+ let source = message;
+ if (message.role === "assistant") {
+ source = {
+ role: message.role,
+ content: message.content,
+ stopReason: message.stopReason,
+ timestamp: message.timestamp,
+ } as AgentMessage;
+ } else if (message.role === "bashExecution") {
+ const { fullOutputPath: _, ...rest } = message;
+ source = rest as AgentMessage;
+ } else if (message.role === "compactionSummary" || message.role === "custom") {
+ const { details: _, ...rest } = message;
+ source = rest as AgentMessage;
+ }
const content = (source as { content?: unknown }).content;
if (typeof content === "string") {
- const projected = projectText(content, budget);
- if (!projected) {
- return source;
- }
- return {
- ...(source as unknown as Record),
- content: projected.text,
- [OMITTED_CHARS_FIELD]: currentOmittedChars + projected.omittedChars,
- } as unknown as AgentMessage;
+ return projectStringFields(source, ["content"], budget);
}
if (!Array.isArray(content)) {
switch (source.role) {
@@ -298,7 +260,7 @@ function projectMessage(message: AgentMessage, budget: ProjectionBudget): AgentM
let omittedChars = 0;
let changed = false;
const projectedContent = content.map((block) => {
- const projected = projectContentBlock(block, true, budget);
+ const projected = projectContentBlock(block, budget);
omittedChars += projected.omittedChars;
changed ||= projected.changed;
return projected.block;
@@ -309,7 +271,7 @@ function projectMessage(message: AgentMessage, budget: ProjectionBudget): AgentM
return {
...(source as unknown as Record),
content: projectedContent,
- [OMITTED_CHARS_FIELD]: currentOmittedChars + omittedChars,
+ [OMITTED_CHARS_FIELD]: readCompactionPlanningOmittedChars(source) + omittedChars,
} as unknown as AgentMessage;
}
diff --git a/src/agents/compaction-planning-worker.ts b/src/agents/compaction-planning-worker.ts
index b80b28ae5ed0..ed139bcffc55 100644
--- a/src/agents/compaction-planning-worker.ts
+++ b/src/agents/compaction-planning-worker.ts
@@ -8,14 +8,12 @@ import { Worker } from "node:worker_threads";
import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion";
import { toErrorObject } from "../infra/errors.js";
import {
- buildHistoryPrunePlan,
buildOversizedFallbackPlan,
buildStageSplitPlan,
buildSummaryChunks,
computeAdaptiveChunkRatio,
projectCompactionMessagesForPlanning,
sanitizeCompactionMessages,
- type HistoryPrunePlan,
type OversizedFallbackPlan,
type StageSplitPlan,
} from "./compaction-planning.js";
@@ -60,13 +58,13 @@ function runCompactionPlanningWorker(params: {
timeoutMs?: number;
workerUrl?: URL;
}): Promise {
- if (params.signal?.aborted) {
- return Promise.reject(
- toErrorObject(
- params.signal.reason ?? new Error("compaction planning aborted"),
- "Non-Error rejection",
- ),
+ const abortError = () =>
+ toErrorObject(
+ params.signal?.reason ?? new Error("compaction planning aborted"),
+ "Non-Error rejection",
);
+ if (params.signal?.aborted) {
+ return Promise.reject(abortError());
}
const workerUrl = params.workerUrl ?? resolveCompactionPlanningWorkerUrl();
@@ -91,30 +89,11 @@ function runCompactionPlanningWorker(params: {
return new Promise((resolve, reject) => {
let settled = false;
const timeout = setTimeout(
- () => {
- settle(
- () =>
- reject(
- new CompactionPlanningWorkerError("compaction planning worker timed out", "timeout"),
- ),
- true,
- );
- },
+ () =>
+ fail(new CompactionPlanningWorkerError("compaction planning worker timed out", "timeout")),
resolveTimerTimeoutMs(params.timeoutMs, COMPACTION_PLANNING_WORKER_TIMEOUT_MS),
);
-
- const abort = () => {
- settle(
- () =>
- reject(
- toErrorObject(
- params.signal?.reason ?? new Error("compaction planning aborted"),
- "Non-Error rejection",
- ),
- ),
- true,
- );
- };
+ const abort = () => fail(abortError());
const settle = (finish: () => void, terminate: boolean) => {
if (settled) {
@@ -129,6 +108,7 @@ function runCompactionPlanningWorker(params: {
}
finish();
};
+ const fail = (error: Error, terminate = true) => settle(() => reject(error), terminate);
params.signal?.addEventListener("abort", abort, { once: true });
@@ -143,20 +123,17 @@ function runCompactionPlanningWorker(params: {
});
worker.once("error", (error) => {
const message = error instanceof Error ? error.message : String(error);
- settle(() => reject(new CompactionPlanningWorkerError(message, "unavailable")), true);
+ fail(new CompactionPlanningWorkerError(message, "unavailable"));
});
worker.once("exit", (code) => {
if (code === 0) {
return;
}
- settle(
- () =>
- reject(
- new CompactionPlanningWorkerError(
- `compaction planning worker exited with code ${code}`,
- "unavailable",
- ),
- ),
+ fail(
+ new CompactionPlanningWorkerError(
+ `compaction planning worker exited with code ${code}`,
+ "unavailable",
+ ),
false,
);
});
@@ -222,14 +199,11 @@ export async function buildSummaryChunksWithWorker(params: {
maxChunkTokens: number;
signal?: AbortSignal;
}): Promise {
+ const { signal, ...planningInput } = params;
return runCompactionPlan({
- input: {
- kind: "summaryChunks",
- messages: params.messages,
- maxChunkTokens: params.maxChunkTokens,
- },
- signal: params.signal,
- fallback: (messages) => buildSummaryChunks({ messages, maxChunkTokens: params.maxChunkTokens }),
+ input: { kind: "summaryChunks", ...planningInput },
+ signal,
+ fallback: (messages) => buildSummaryChunks({ ...planningInput, messages }),
restore: (value, messages) =>
value.chunkIndexes.map((indexes) => restoreIndexedMessages(messages, indexes)),
});
@@ -241,15 +215,11 @@ export async function buildOversizedFallbackPlanWithWorker(params: {
contextWindow: number;
signal?: AbortSignal;
}): Promise {
+ const { signal, ...planningInput } = params;
return runCompactionPlan({
- input: {
- kind: "oversizedFallback",
- messages: params.messages,
- contextWindow: params.contextWindow,
- },
- signal: params.signal,
- fallback: (messages) =>
- buildOversizedFallbackPlan({ messages, contextWindow: params.contextWindow }),
+ input: { kind: "oversizedFallback", ...planningInput },
+ signal,
+ fallback: (messages) => buildOversizedFallbackPlan({ ...planningInput, messages }),
restore: (value, messages) => ({
smallMessages: restoreIndexedMessages(messages, value.smallMessageIndexes),
oversizedNotes: value.oversizedNotes,
@@ -265,22 +235,11 @@ export async function buildStageSplitPlanWithWorker(params: {
minMessagesForSplit?: number;
signal?: AbortSignal;
}): Promise {
+ const { signal, ...planningInput } = params;
return runCompactionPlan({
- input: {
- kind: "stageSplit",
- messages: params.messages,
- maxChunkTokens: params.maxChunkTokens,
- parts: params.parts,
- minMessagesForSplit: params.minMessagesForSplit,
- },
- signal: params.signal,
- fallback: (messages) =>
- buildStageSplitPlan({
- messages,
- maxChunkTokens: params.maxChunkTokens,
- parts: params.parts,
- minMessagesForSplit: params.minMessagesForSplit,
- }),
+ input: { kind: "stageSplit", ...planningInput },
+ signal,
+ fallback: (messages) => buildStageSplitPlan({ ...planningInput, messages }),
restore: (value, messages) =>
value.mode === "split"
? {
@@ -291,38 +250,17 @@ export async function buildStageSplitPlanWithWorker(params: {
});
}
-/**
- * Builds a history-pruning plan on the owner thread.
- *
- * Pruning repairs tool-result pairs and returns exact retained/dropped messages,
- * so a bounded selection projection cannot reconstruct every result faithfully.
- */
-export async function buildHistoryPrunePlanWithWorker(params: {
- messagesToSummarize: AgentMessage[];
- turnPrefixMessages: AgentMessage[];
- tokensBefore: number;
- contextWindowTokens: number;
- maxHistoryShare: number;
- parts?: number;
- signal?: AbortSignal;
-}): Promise {
- return buildHistoryPrunePlan(params);
-}
-
/** Computes the adaptive compaction chunk ratio with worker fallback. */
export async function computeAdaptiveChunkRatioWithWorker(params: {
messages: AgentMessage[];
contextWindow: number;
signal?: AbortSignal;
}): Promise {
+ const { signal, ...planningInput } = params;
return runCompactionPlan({
- input: {
- kind: "adaptiveChunkRatio",
- messages: params.messages,
- contextWindow: params.contextWindow,
- },
- signal: params.signal,
- fallback: () => computeAdaptiveChunkRatio(params.messages, params.contextWindow),
+ input: { kind: "adaptiveChunkRatio", ...planningInput },
+ signal,
+ fallback: () => computeAdaptiveChunkRatio(planningInput.messages, planningInput.contextWindow),
restore: (value) => value.ratio,
});
}
@@ -330,7 +268,6 @@ export async function computeAdaptiveChunkRatioWithWorker(params: {
const compactionPlanningWorkerTesting = {
resolveCompactionPlanningWorkerUrl,
runCompactionPlanningWorker,
- CompactionPlanningWorkerError,
};
if (process.env.VITEST || process.env.NODE_ENV === "test") {
diff --git a/src/agents/compaction-planning.ts b/src/agents/compaction-planning.ts
index a2c80d188788..99f43f1c43ea 100644
--- a/src/agents/compaction-planning.ts
+++ b/src/agents/compaction-planning.ts
@@ -44,7 +44,7 @@ export type OversizedFallbackPlan = {
};
/** Token accounting and optional prune result for preserving context-window headroom. */
-export type HistoryPrunePlan = {
+type HistoryPrunePlan = {
summarizableTokens: number;
newContentTokens: number;
maxHistoryTokens: number;
@@ -80,11 +80,7 @@ export function sanitizeCompactionMessages(messages: AgentMessage[]): AgentMessa
}
function estimateCompactionPlanningTokens(message: AgentMessage): number {
- const omittedChars = readCompactionPlanningOmittedChars(message);
- if (omittedChars === 0) {
- return estimateTokens(message);
- }
- return estimateTokens(message) + Math.ceil(omittedChars / 4);
+ return estimateTokens(message) + Math.ceil(readCompactionPlanningOmittedChars(message) / 4);
}
/** Builds a bounded planning projection that preserves token pressure accounting. */
@@ -115,23 +111,9 @@ function groupCompactionMessages(
let currentTokens = 0;
let pendingToolCallIds = new Set();
- const finishCurrentGroup = () => {
- if (current.length === 0) {
- return;
- }
- groups.push({ messages: current, tokens: currentTokens });
- current = [];
- currentTokens = 0;
- };
-
for (const [index, message] of messages.entries()) {
- const messageTokens = perMessageTokens.at(index);
- if (messageTokens === undefined) {
- throw new Error("Compaction token estimates are out of sync with messages");
- }
-
current.push(message);
- currentTokens += messageTokens;
+ currentTokens += perMessageTokens[index]!;
if (message.role === "assistant") {
const stopReason = (message as { stopReason?: unknown }).stopReason;
@@ -152,41 +134,34 @@ function groupCompactionMessages(
// A displaced user turn still belongs to an unfinished call/result batch;
// splitting it would make one of the resulting provider transcripts invalid.
if (pendingToolCallIds.size === 0) {
- finishCurrentGroup();
+ groups.push({ messages: current, tokens: currentTokens });
+ current = [];
+ currentTokens = 0;
}
}
- finishCurrentGroup();
+ if (current.length > 0) {
+ groups.push({ messages: current, tokens: currentTokens });
+ }
return groups;
}
-/** Splits messages into roughly equal token-share chunks without separating active tool pairs. */
-function splitMessagesByTokenShare(
+/** Chunks atomic tool-call groups without splitting a provider-visible call/result pair. */
+function chunkCompactionMessageGroups(
messages: AgentMessage[],
- parts = DEFAULT_PARTS,
+ maxTokens: number,
+ perMessageTokens: number[],
+ maxChunks = Number.POSITIVE_INFINITY,
): AgentMessage[][] {
- if (messages.length === 0) {
- return [];
- }
- const normalizedParts = normalizeCompactionParts(parts, messages.length);
- if (normalizedParts <= 1) {
- return [messages];
- }
-
- // Sanitize the full array once and reuse per-message token counts; avoids the
- // per-message [msg] wrap-and-clone that previously ran on every iteration.
- const perMessageTokens = estimatePerMessageTokens(messages);
- const totalTokens = perMessageTokens.reduce((sum, tokens) => sum + tokens, 0);
- const targetTokens = totalTokens / normalizedParts;
const chunks: AgentMessage[][] = [];
let current: AgentMessage[] = [];
let currentTokens = 0;
for (const group of groupCompactionMessages(messages, perMessageTokens)) {
if (
- chunks.length < normalizedParts - 1 &&
current.length > 0 &&
- currentTokens + group.tokens > targetTokens
+ chunks.length < maxChunks - 1 &&
+ currentTokens + group.tokens > maxTokens
) {
chunks.push(current);
current = [];
@@ -204,47 +179,26 @@ function splitMessagesByTokenShare(
return chunks;
}
-/** Chunks messages by a max-token budget while applying the shared estimator safety margin. */
-function chunkMessagesByMaxTokens(messages: AgentMessage[], maxTokens: number): AgentMessage[][] {
+/** Splits messages into roughly equal token-share chunks without separating active tool pairs. */
+function splitMessagesByTokenShare(
+ messages: AgentMessage[],
+ parts = DEFAULT_PARTS,
+): AgentMessage[][] {
if (messages.length === 0) {
return [];
}
-
- // Apply safety margin to compensate for estimateTokens() underestimation
- // (chars/4 heuristic misses multi-byte chars, special tokens, code tokens, etc.)
- const effectiveMax = Math.max(1, Math.floor(maxTokens / SAFETY_MARGIN));
-
- // Sanitize the full array once and reuse per-message token counts; avoids the
- // per-message [msg] wrap-and-clone that previously ran on every iteration.
+ const normalizedParts = normalizeCompactionParts(parts, messages.length);
+ if (normalizedParts <= 1) {
+ return [messages];
+ }
const perMessageTokens = estimatePerMessageTokens(messages);
- const chunks: AgentMessage[][] = [];
- let currentChunk: AgentMessage[] = [];
- let currentTokens = 0;
-
- for (const group of groupCompactionMessages(messages, perMessageTokens)) {
- if (currentChunk.length > 0 && currentTokens + group.tokens > effectiveMax) {
- chunks.push(currentChunk);
- currentChunk = [];
- currentTokens = 0;
- }
-
- currentChunk.push(...group.messages);
- currentTokens += group.tokens;
-
- if (group.tokens > effectiveMax) {
- // A tool batch is indivisible even above the heuristic budget; the
- // oversized-summary fallback can handle it without orphaning its result.
- chunks.push(currentChunk);
- currentChunk = [];
- currentTokens = 0;
- }
- }
-
- if (currentChunk.length > 0) {
- chunks.push(currentChunk);
- }
-
- return chunks;
+ const totalTokens = perMessageTokens.reduce((sum, tokens) => sum + tokens, 0);
+ return chunkCompactionMessageGroups(
+ messages,
+ totalTokens / normalizedParts,
+ perMessageTokens,
+ normalizedParts,
+ );
}
/**
@@ -256,12 +210,8 @@ export function computeAdaptiveChunkRatio(messages: AgentMessage[], contextWindo
return BASE_CHUNK_RATIO;
}
- const totalTokens = estimateMessagesTokens(messages);
- const avgTokens = totalTokens / messages.length;
-
- // Apply safety margin to account for estimation inaccuracy
- const safeAvgTokens = avgTokens * SAFETY_MARGIN;
- const avgRatio = safeAvgTokens / contextWindow;
+ const avgRatio =
+ ((estimateMessagesTokens(messages) / messages.length) * SAFETY_MARGIN) / contextWindow;
// If average message is > 10% of context, reduce chunk ratio
if (avgRatio > 0.1) {
@@ -285,7 +235,13 @@ export function buildSummaryChunks(params: {
}): AgentMessage[][] {
// SECURITY: never feed toolResult.details or runtime-context transcript entries into summarization prompts.
const safeMessages = sanitizeCompactionMessages(params.messages);
- return chunkMessagesByMaxTokens(safeMessages, params.maxChunkTokens);
+ // The estimator can undercount Unicode/code tokens; indivisible tool batches may exceed this cap.
+ const effectiveMax = Math.max(1, Math.floor(params.maxChunkTokens / SAFETY_MARGIN));
+ return chunkCompactionMessageGroups(
+ safeMessages,
+ effectiveMax,
+ estimatePerMessageTokens(safeMessages),
+ );
}
/** Separates messages too large to summarize and emits compact placeholder notes for them. */
@@ -355,9 +311,8 @@ export function buildStageSplitPlan(params: {
function pruneHistoryForContextShare(params: {
messages: AgentMessage[];
maxContextTokens: number;
- maxHistoryShare?: number;
+ maxHistoryShare: number;
parts?: number;
- mode?: "share" | "handoff";
}): {
messages: AgentMessage[];
droppedMessagesList: AgentMessage[];
@@ -367,10 +322,7 @@ function pruneHistoryForContextShare(params: {
keptTokens: number;
budgetTokens: number;
} {
- const isHandoff = params.mode === "handoff";
- const defaultShare = isHandoff ? 0.2 : 0.5; // Stricter budget for handoff snapshots
- const maxHistoryShare = params.maxHistoryShare ?? defaultShare;
- const budgetTokens = Math.max(1, Math.floor(params.maxContextTokens * maxHistoryShare));
+ const budgetTokens = Math.max(1, Math.floor(params.maxContextTokens * params.maxHistoryShare));
let keptMessages = params.messages;
const allDroppedMessages: AgentMessage[] = [];
let droppedChunks = 0;
@@ -384,31 +336,15 @@ function pruneHistoryForContextShare(params: {
if (chunks.length <= 1) {
break;
}
- const [dropped, ...rest] = chunks;
- if (!dropped) {
- break;
- }
- const flatRest = rest.flat();
-
- // After dropping a chunk, repair tool_use/tool_result pairing to handle
- // orphaned tool_results (whose tool_use was in the dropped chunk).
- // repairToolUseResultPairing drops orphaned tool_results, preventing
- // "unexpected tool_use_id" errors from Anthropic's API.
- const repairReport = repairToolUseResultPairing(flatRest);
- const repairedKept = repairReport.messages;
-
- // Track orphaned tool_results as dropped (they were in kept but their tool_use was dropped)
- const orphanedCount = repairReport.droppedOrphanCount;
+ const dropped = chunks[0]!;
+ // Dropping a call owner also drops orphaned results; providers reject replay without the pair.
+ const repairReport = repairToolUseResultPairing(chunks.slice(1).flat());
droppedChunks += 1;
- droppedMessages += dropped.length + orphanedCount;
+ droppedMessages += dropped.length + repairReport.droppedOrphanCount;
droppedTokens += estimateMessagesTokens(dropped);
- // Note: We don't have the actual orphaned messages to add to droppedMessagesList
- // since repairToolUseResultPairing doesn't return them. This is acceptable since
- // the dropped messages are used for summarization, and orphaned tool_results
- // without their tool_use context aren't useful for summarization anyway.
allDroppedMessages.push(...dropped);
- keptMessages = repairedKept;
+ keptMessages = repairReport.messages;
}
return {
@@ -440,23 +376,16 @@ export function buildHistoryPrunePlan(params: {
params.contextWindowTokens * params.maxHistoryShare * SAFETY_MARGIN,
);
- if (newContentTokens <= maxHistoryTokens) {
- return {
- summarizableTokens,
- newContentTokens,
- maxHistoryTokens,
- };
- }
-
- return {
- summarizableTokens,
- newContentTokens,
- maxHistoryTokens,
- pruned: pruneHistoryForContextShare({
- messages: params.messagesToSummarize,
- maxContextTokens: params.contextWindowTokens,
- maxHistoryShare: params.maxHistoryShare,
- parts: params.parts,
- }),
- };
+ const plan = { summarizableTokens, newContentTokens, maxHistoryTokens };
+ return newContentTokens <= maxHistoryTokens
+ ? plan
+ : {
+ ...plan,
+ pruned: pruneHistoryForContextShare({
+ messages: params.messagesToSummarize,
+ maxContextTokens: params.contextWindowTokens,
+ maxHistoryShare: params.maxHistoryShare,
+ parts: params.parts,
+ }),
+ };
}
diff --git a/src/agents/compaction-planning.worker.ts b/src/agents/compaction-planning.worker.ts
index e3867d5cc0d0..ed4d9606fc67 100644
--- a/src/agents/compaction-planning.worker.ts
+++ b/src/agents/compaction-planning.worker.ts
@@ -12,23 +12,9 @@ import type { AgentMessage } from "./runtime/index.js";
/** Serializable request accepted by the compaction planning worker. */
export type CompactionPlanningWorkerInput =
- | {
- kind: "summaryChunks";
- messages: AgentMessage[];
- maxChunkTokens: number;
- }
- | {
- kind: "oversizedFallback";
- messages: AgentMessage[];
- contextWindow: number;
- }
- | {
- kind: "stageSplit";
- messages: AgentMessage[];
- maxChunkTokens: number;
- parts?: number;
- minMessagesForSplit?: number;
- }
+ | ({ kind: "summaryChunks" } & Parameters[0])
+ | ({ kind: "oversizedFallback" } & Parameters[0])
+ | ({ kind: "stageSplit" } & Parameters[0])
| {
kind: "adaptiveChunkRatio";
messages: AgentMessage[];
@@ -46,15 +32,7 @@ export type CompactionPlanningWorkerValue =
smallMessageIndexes: number[];
oversizedNotes: string[];
}
- | {
- kind: "stageSplit";
- mode: "single";
- }
- | {
- kind: "stageSplit";
- mode: "split";
- chunkIndexes: number[][];
- }
+ | ({ kind: "stageSplit" } & ({ mode: "single" } | { mode: "split"; chunkIndexes: number[][] }))
| {
kind: "adaptiveChunkRatio";
ratio: number;
@@ -142,19 +120,13 @@ function planCompactionWorkerInput(
/** Run one compaction planning request and return a serializable result. */
export function runCompactionPlanningWorkerInput(input: unknown): CompactionPlanningWorkerResult {
if (!isWorkerInput(input)) {
- return {
- status: "failed",
- error: "invalid compaction planning worker input",
- };
+ return { status: "failed", error: "invalid compaction planning worker input" };
}
try {
return { status: "ok", value: planCompactionWorkerInput(input) };
} catch (error) {
- return {
- status: "failed",
- error: error instanceof Error ? error.message : String(error),
- };
+ return { status: "failed", error: error instanceof Error ? error.message : String(error) };
}
}
diff --git a/src/agents/compaction-real-conversation.ts b/src/agents/compaction-real-conversation.ts
index edadf01cb876..c2e9cd162f89 100644
--- a/src/agents/compaction-real-conversation.ts
+++ b/src/agents/compaction-real-conversation.ts
@@ -17,17 +17,11 @@ const NON_CONVERSATION_BLOCK_TYPES = new Set([
function hasMeaningfulText(text: string): boolean {
const trimmed = text.trim();
- if (!trimmed) {
- return false;
- }
- if (isSilentReplyText(trimmed)) {
+ if (!trimmed || isSilentReplyText(trimmed)) {
return false;
}
const heartbeat = stripHeartbeatToken(trimmed, { mode: "message" });
- if (heartbeat.didStrip) {
- return heartbeat.text.trim().length > 0;
- }
- return true;
+ return !heartbeat.didStrip || heartbeat.text.trim().length > 0;
}
function isSummaryRole(role: unknown): boolean {
@@ -73,22 +67,15 @@ function hasMeaningfulMessageContent(content: unknown): boolean {
if (!block || typeof block !== "object") {
continue;
}
- const type = (block as { type?: unknown }).type;
- if (type !== "text") {
+ const { type, text } = block as { type?: unknown; text?: unknown };
+ if (type === "text") {
+ if (typeof text === "string" && hasMeaningfulText(text)) {
+ return true;
+ }
+ } else if (typeof type !== "string" || !NON_CONVERSATION_BLOCK_TYPES.has(type)) {
// Tool-call metadata and internal reasoning blocks do not make a
// heartbeat-only transcript count as real conversation.
- if (typeof type === "string" && NON_CONVERSATION_BLOCK_TYPES.has(type)) {
- continue;
- }
sawMeaningfulNonTextBlock = true;
- continue;
- }
- const text = (block as { text?: unknown }).text;
- if (typeof text !== "string") {
- continue;
- }
- if (hasMeaningfulText(text)) {
- return true;
}
}
return sawMeaningfulNonTextBlock;
@@ -123,10 +110,7 @@ export function isRealConversationMessage(
const start = Math.max(0, index - TOOL_RESULT_REAL_CONVERSATION_LOOKBACK);
for (let i = index - 1; i >= start; i -= 1) {
const candidate = messages[i];
- if (!candidate) {
- continue;
- }
- if (isToolResultConversationAnchor(candidate)) {
+ if (candidate && isToolResultConversationAnchor(candidate)) {
return true;
}
}
diff --git a/src/agents/compaction-usage.ts b/src/agents/compaction-usage.ts
index 22584ba9d3e1..e4993ce034f2 100644
--- a/src/agents/compaction-usage.ts
+++ b/src/agents/compaction-usage.ts
@@ -6,16 +6,8 @@ import type { AgentMessage } from "./runtime/index.js";
import { makeZeroUsageSnapshot } from "./usage.js";
function parseCompactionUsageTimestamp(value: unknown): number | null {
- if (typeof value === "number" && Number.isFinite(value)) {
- return value;
- }
- if (typeof value === "string") {
- const parsed = Date.parse(value);
- if (Number.isFinite(parsed)) {
- return parsed;
- }
- }
- return null;
+ const timestamp = typeof value === "string" ? Date.parse(value) : value;
+ return typeof timestamp === "number" && Number.isFinite(timestamp) ? timestamp : null;
}
export function stripStaleAssistantUsageBeforeLatestCompaction(
@@ -25,56 +17,46 @@ export function stripStaleAssistantUsageBeforeLatestCompaction entry?.role === "compactionSummary",
+ );
const hasCompactionSummary = latestCompactionSummaryIndex !== -1;
if (!hasCompactionSummary && options.whenMissingCompactionSummary !== "zeroAssistantUsage") {
return messages;
}
- const out = options.mutate ? messages : [...messages];
- let touched = false;
- for (let i = 0; i < out.length; i += 1) {
- const candidate = out[i] as
+ const latestCompactionTimestamp = parseCompactionUsageTimestamp(
+ (messages[latestCompactionSummaryIndex] as { timestamp?: unknown } | undefined)?.timestamp,
+ );
+ let out = messages;
+ for (let i = 0; i < messages.length; i += 1) {
+ const candidate = messages[i] as
| (AgentMessage & { usage?: unknown; timestamp?: unknown })
| undefined;
- if (!candidate || candidate.role !== "assistant") {
- continue;
- }
- if (!candidate.usage || typeof candidate.usage !== "object") {
+ if (
+ candidate?.role !== "assistant" ||
+ !candidate.usage ||
+ typeof candidate.usage !== "object"
+ ) {
continue;
}
const messageTimestamp = parseCompactionUsageTimestamp(candidate.timestamp);
- const compactionTimestamp = latestCompactionTimestamp;
- const hasTimestampBoundary =
- hasCompactionSummary && compactionTimestamp !== null && messageTimestamp !== null;
- const staleByMissingSummary = !hasCompactionSummary;
- const staleByTimestamp = hasTimestampBoundary && messageTimestamp <= compactionTimestamp;
- const staleByLegacyOrdering =
- hasCompactionSummary && !hasTimestampBoundary && i < latestCompactionSummaryIndex;
- if (!staleByMissingSummary && !staleByTimestamp && !staleByLegacyOrdering) {
+ const stale =
+ !hasCompactionSummary ||
+ (latestCompactionTimestamp !== null && messageTimestamp !== null
+ ? messageTimestamp <= latestCompactionTimestamp
+ : i < latestCompactionSummaryIndex);
+ if (!stale) {
continue;
}
// Session runtime expects assistant usage to stay structurally valid during
// accounting. Keep stale snapshots present, but zeroed after compaction.
- const candidateRecord = candidate as unknown as Record;
- out[i] = {
- ...candidateRecord,
- usage: makeZeroUsageSnapshot(),
- } as unknown as TMessage;
- touched = true;
+ if (out === messages && !options.mutate) {
+ out = [...messages];
+ }
+ out[i] = { ...candidate, usage: makeZeroUsageSnapshot() } as TMessage;
}
- return touched ? out : messages;
+ return out;
}
diff --git a/src/agents/compaction.ts b/src/agents/compaction.ts
index 270a711b4390..47d3e30456fb 100644
--- a/src/agents/compaction.ts
+++ b/src/agents/compaction.ts
@@ -26,7 +26,7 @@ import { DEFAULT_CONTEXT_TOKENS } from "./defaults.js";
import { isTimeoutError } from "./failover-error.js";
import type { AgentMessage, StreamFn, ThinkingLevel } from "./runtime/index.js";
import type { ExtensionContext } from "./sessions/index.js";
-import { generateSummary as agentGenerateSummary } from "./sessions/index.js";
+import { generateSummary } from "./sessions/index.js";
export {
BASE_CHUNK_RATIO,
@@ -71,18 +71,31 @@ export type CompactionSummarizationInstructions = {
identifierInstructions?: string;
};
+type CompactionSummaryParams = {
+ messages: AgentMessage[];
+ model: NonNullable;
+ apiKey: string;
+ headers?: Record;
+ signal: AbortSignal;
+ reserveTokens: number;
+ maxChunkTokens: number;
+ contextWindow: number;
+ customInstructions?: string;
+ summarizationInstructions?: CompactionSummarizationInstructions;
+ previousSummary?: string;
+ thinkingLevel?: ThinkingLevel;
+ streamFn?: StreamFn;
+};
+
function resolveIdentifierPreservationInstructions(
instructions?: CompactionSummarizationInstructions,
): string | undefined {
- const policy = instructions?.identifierPolicy ?? "strict";
- if (policy === "off") {
+ if (instructions?.identifierPolicy === "off") {
return undefined;
}
- if (policy === "custom") {
- const custom = instructions?.identifierInstructions?.trim();
- return custom && custom.length > 0 ? custom : IDENTIFIER_PRESERVATION_INSTRUCTIONS;
- }
- return IDENTIFIER_PRESERVATION_INSTRUCTIONS;
+ return instructions?.identifierPolicy === "custom"
+ ? instructions.identifierInstructions?.trim() || IDENTIFIER_PRESERVATION_INSTRUCTIONS
+ : IDENTIFIER_PRESERVATION_INSTRUCTIONS;
}
/** Combines identifier-preservation and caller-provided compaction instructions. */
@@ -92,32 +105,15 @@ function buildCompactionSummarizationInstructions(
): string | undefined {
const custom = customInstructions?.trim();
const identifierPreservation = resolveIdentifierPreservationInstructions(instructions);
- if (!identifierPreservation && !custom) {
- return undefined;
- }
if (!custom) {
return identifierPreservation;
}
- if (!identifierPreservation) {
- return `Additional focus:\n${custom}`;
- }
- return `${identifierPreservation}\n\nAdditional focus:\n${custom}`;
+ return identifierPreservation
+ ? `${identifierPreservation}\n\nAdditional focus:\n${custom}`
+ : `Additional focus:\n${custom}`;
}
-async function summarizeChunks(params: {
- messages: AgentMessage[];
- model: NonNullable;
- apiKey: string;
- headers?: Record;
- signal: AbortSignal;
- reserveTokens: number;
- maxChunkTokens: number;
- customInstructions?: string;
- summarizationInstructions?: CompactionSummarizationInstructions;
- previousSummary?: string;
- thinkingLevel?: ThinkingLevel;
- streamFn?: StreamFn;
-}): Promise {
+async function summarizeChunks(params: CompactionSummaryParams): Promise {
if (params.messages.length === 0) {
return params.previousSummary ?? DEFAULT_SUMMARY_FALLBACK;
}
@@ -132,8 +128,7 @@ async function summarizeChunks(params: {
params.customInstructions,
params.summarizationInstructions,
);
- let hasGeneratedChunk = false;
- for (const chunk of chunks) {
+ for (const [completedChunks, chunk] of chunks.entries()) {
try {
summary = await retryAsync(
() =>
@@ -158,43 +153,26 @@ async function summarizeChunks(params: {
// Backoff must honor caller cancellation; otherwise an abort during
// the sleep would stall compaction until the full delay elapses.
sleep: (ms) => sleepWithAbort(ms, params.signal),
- shouldRetry: (err) => {
- // Stop retrying when the caller explicitly cancelled.
- if (params.signal.aborted) {
- return false;
- }
- // Preserve existing non-retry policy for real network/transport
- // timeouts (e.g. "fetch failed", ETIMEDOUT) that are not AbortErrors.
- if (!isAbortError(err) && isTimeoutError(err)) {
- return false;
- }
- // Provider-side AbortErrors with signal not yet aborted are
- // transient disconnects — retrying is correct.
- return true;
- },
+ // Caller aborts and transport timeouts are terminal; provider-side
+ // AbortErrors without caller cancellation remain retryable.
+ shouldRetry: (err) =>
+ !params.signal.aborted && (isAbortError(err) || !isTimeoutError(err)),
},
);
- hasGeneratedChunk = true;
} catch (err) {
- // Propagate only when the caller explicitly cancelled. Provider-side
- // AbortErrors (signal not aborted) fall through to partial/fallback paths.
- if (params.signal.aborted) {
- throw err;
- }
- // Real non-abort transport timeouts still propagate immediately.
- if (!isAbortError(err) && isTimeoutError(err)) {
- throw err;
- }
- // No chunk has succeeded yet — rethrow so summarizeWithFallback
- // can run its existing "Context contained N messages" fallback.
- if (!hasGeneratedChunk) {
+ // Caller aborts, transport timeouts, and failures before any completed
+ // chunk cannot produce a recoverable partial summary.
+ if (
+ params.signal.aborted ||
+ (!isAbortError(err) && isTimeoutError(err)) ||
+ completedChunks === 0
+ ) {
throw err;
}
// At least one chunk succeeded — throw with the partial summary
// attached so summarizeWithFallback can try the oversized-message
// retry first and only fall back to the partial summary if that
// also fails.
- const completedChunks = chunks.indexOf(chunk);
log.warn("chunk summarization failed after retries; partial summary available", {
err,
completedChunks,
@@ -210,65 +188,22 @@ async function summarizeChunks(params: {
return summary ?? DEFAULT_SUMMARY_FALLBACK;
}
-function generateSummary(
- currentMessages: AgentMessage[],
- model: NonNullable,
- reserveTokens: number,
- apiKey: string,
- headers: Record | undefined,
- signal: AbortSignal,
- customInstructions?: string,
- previousSummary?: string,
- thinkingLevel?: ThinkingLevel,
- streamFn?: StreamFn,
-): Promise {
- return agentGenerateSummary(
- currentMessages,
- model,
- reserveTokens,
- apiKey,
- headers,
- signal,
- customInstructions,
- previousSummary,
- thinkingLevel,
- streamFn,
- );
-}
-
/**
* Summarize with progressive fallback for handling oversized messages.
* If full summarization fails, tries partial summarization excluding oversized messages.
*/
-async function summarizeWithFallbackResult(params: {
- messages: AgentMessage[];
- model: NonNullable;
- apiKey: string;
- headers?: Record;
- signal: AbortSignal;
- reserveTokens: number;
- maxChunkTokens: number;
- contextWindow: number;
- customInstructions?: string;
- summarizationInstructions?: CompactionSummarizationInstructions;
- previousSummary?: string;
- thinkingLevel?: ThinkingLevel;
- streamFn?: StreamFn;
-}): Promise {
+async function summarizeWithFallback(params: CompactionSummaryParams): Promise {
const { messages, contextWindow } = params;
if (messages.length === 0) {
- return {
- kind: "summary",
- text: params.previousSummary ?? DEFAULT_SUMMARY_FALLBACK,
- };
+ return params.previousSummary ?? DEFAULT_SUMMARY_FALLBACK;
}
// Try full summarization first
let partialSummaryFallback: string | undefined;
let lastError: unknown;
try {
- return { kind: "summary", text: await summarizeChunks(params) };
+ return await summarizeChunks(params);
} catch (err) {
lastError = err;
if (params.signal.aborted) {
@@ -284,6 +219,7 @@ async function summarizeWithFallbackResult(params: {
contextWindow,
signal: params.signal,
});
+ const oversizedSuffix = oversizedNotes.length > 0 ? `\n\n${oversizedNotes.join("\n")}` : "";
// When nothing was oversized, `smallMessages` is the same transcript as the full attempt.
// Re-summarizing it would duplicate the same failing API work (and duplicate warn logs).
@@ -293,8 +229,7 @@ async function summarizeWithFallbackResult(params: {
...params,
messages: smallMessages,
});
- const notes = oversizedNotes.length > 0 ? `\n\n${oversizedNotes.join("\n")}` : "";
- return { kind: "summary", text: partialSummary + notes };
+ return partialSummary + oversizedSuffix;
} catch (partialError) {
lastError = partialError;
if (params.signal.aborted) {
@@ -306,15 +241,14 @@ async function summarizeWithFallbackResult(params: {
// so the model knows large content was filtered.
const retryPartial = (lastError as PartialSummaryError).partialSummary;
if (retryPartial) {
- const notes = oversizedNotes.length > 0 ? `\n\n${oversizedNotes.join("\n")}` : "";
- partialSummaryFallback = retryPartial + notes;
+ partialSummaryFallback = retryPartial + oversizedSuffix;
}
}
}
// Final fallback: use best available partial summary, otherwise throw error
if (partialSummaryFallback) {
- return { kind: "summary", text: partialSummaryFallback };
+ return partialSummaryFallback;
}
// All summarization attempts failed — throw error so caller knows compaction
@@ -328,16 +262,10 @@ async function summarizeWithFallbackResult(params: {
);
}
-async function summarizeWithFallback(
- params: Parameters[0],
-): Promise {
- return (await summarizeWithFallbackResult(params)).text;
-}
-
/** Extracts a compact timestamp range from a chunk of messages for merge metadata. */
function extractChunkTimeRange(chunk: AgentMessage[]): string {
- let earliest: number | undefined;
- let latest: number | undefined;
+ let earliest = Number.POSITIVE_INFINITY;
+ let latest = 0;
for (const message of chunk) {
const timestamp = message.timestamp;
if (
@@ -347,10 +275,10 @@ function extractChunkTimeRange(chunk: AgentMessage[]): string {
) {
continue;
}
- earliest = earliest === undefined ? timestamp : Math.min(earliest, timestamp);
- latest = latest === undefined ? timestamp : Math.max(latest, timestamp);
+ earliest = Math.min(earliest, timestamp);
+ latest = Math.max(latest, timestamp);
}
- if (earliest === undefined || latest === undefined) {
+ if (!Number.isFinite(earliest)) {
return "";
}
const format = (timestamp: number) =>
@@ -360,29 +288,15 @@ function extractChunkTimeRange(chunk: AgentMessage[]): string {
}
/** Summarizes history in multiple stages when a single pass would be too large. */
-export async function summarizeInStages(params: {
- messages: AgentMessage[];
- model: NonNullable;
- apiKey: string;
- headers?: Record;
- signal: AbortSignal;
- reserveTokens: number;
- maxChunkTokens: number;
- contextWindow: number;
- customInstructions?: string;
- summarizationInstructions?: CompactionSummarizationInstructions;
- previousSummary?: string;
- parts?: number;
- minMessagesForSplit?: number;
- thinkingLevel?: ThinkingLevel;
- streamFn?: StreamFn;
-}): Promise {
+export async function summarizeInStages(
+ params: CompactionSummaryParams & {
+ parts?: number;
+ minMessagesForSplit?: number;
+ },
+): Promise {
const { messages } = params;
if (messages.length === 0) {
- return {
- kind: "summary",
- text: params.previousSummary ?? DEFAULT_SUMMARY_FALLBACK,
- };
+ return { kind: "summary", text: await summarizeWithFallback(params) };
}
const plan = await buildStageSplitPlanWithWorker({
@@ -394,18 +308,18 @@ export async function summarizeInStages(params: {
});
if (plan.mode === "single") {
- return summarizeWithFallbackResult(params);
+ return { kind: "summary", text: await summarizeWithFallback(params) };
}
const partialSummaries: string[] = [];
for (const [index, chunk] of plan.chunks.entries()) {
try {
- const result = await summarizeWithFallbackResult({
+ const summary = await summarizeWithFallback({
...params,
messages: chunk,
previousSummary: undefined,
});
- partialSummaries.push(result.text);
+ partialSummaries.push(summary);
} catch (err) {
// A chunk summarization failed — fail the whole stages compaction.
// This prevents silent infinite retry loops where compaction reports
@@ -461,12 +375,14 @@ export async function summarizeInStages(params: {
? `${MERGE_SUMMARIES_INSTRUCTIONS}\n\n${custom}`
: MERGE_SUMMARIES_INSTRUCTIONS;
- const mergedResult = await summarizeWithFallbackResult({
- ...params,
- messages: summaryMessages,
- customInstructions: mergeInstructions,
- });
- return mergedResult;
+ return {
+ kind: "summary",
+ text: await summarizeWithFallback({
+ ...params,
+ messages: summaryMessages,
+ customInstructions: mergeInstructions,
+ }),
+ };
}
/** Resolves a positive context-window token count from model metadata. */
diff --git a/src/agents/embedded-agent-runner/run/preemptive-compaction.ts b/src/agents/embedded-agent-runner/run/preemptive-compaction.ts
index ce7250efa0b7..c02b4b979dea 100644
--- a/src/agents/embedded-agent-runner/run/preemptive-compaction.ts
+++ b/src/agents/embedded-agent-runner/run/preemptive-compaction.ts
@@ -50,18 +50,28 @@ export type LlmBoundaryTokenPressure = {
renderedChars?: number;
};
-function estimateStringTokenPressure(text: string, charsPerToken = ESTIMATED_CHARS_PER_TOKEN) {
- return Math.ceil(estimateStringChars(text) / charsPerToken);
+type TokenPressureMode = "general" | "tool-result";
+
+function estimateStringTokenPressure(
+ text: string,
+ charsPerToken = ESTIMATED_CHARS_PER_TOKEN,
+ mode: TokenPressureMode = "general",
+) {
+ const estimatedTokens = Math.ceil(estimateStringChars(text) / charsPerToken);
+ return mode === "tool-result"
+ ? Math.max(Math.ceil(text.length / TOOL_RESULT_CHARS_PER_TOKEN), estimatedTokens)
+ : estimatedTokens;
}
function estimateJsonPayloadTokenPressure(
value: unknown,
charsPerToken = JSON_PAYLOAD_CHARS_PER_TOKEN,
+ mode: TokenPressureMode = "general",
): number {
try {
const serialized = JSON.stringify(value);
return typeof serialized === "string"
- ? Math.ceil(estimateStringChars(serialized) / charsPerToken)
+ ? estimateStringTokenPressure(serialized, charsPerToken, mode)
: 1;
} catch {
return 256;
@@ -89,75 +99,26 @@ function estimateIdentifierTokenPressure(
function estimateContentBlockTokenPressure(
block: unknown,
charsPerToken = ESTIMATED_CHARS_PER_TOKEN,
+ mode: TokenPressureMode = "general",
): number {
if (typeof block === "string") {
- return estimateStringTokenPressure(block, charsPerToken);
+ return estimateStringTokenPressure(block, charsPerToken, mode);
}
if (!isRecord(block)) {
- return estimateJsonPayloadTokenPressure(block, charsPerToken);
+ return estimateJsonPayloadTokenPressure(block, charsPerToken, mode);
}
const type = block.type;
- if (type === "text" && typeof block.text === "string") {
- return CONTENT_BLOCK_OVERHEAD_TOKENS + estimateStringTokenPressure(block.text, charsPerToken);
- }
- if (type === "thinking" && typeof block.thinking === "string") {
- return (
- CONTENT_BLOCK_OVERHEAD_TOKENS + estimateStringTokenPressure(block.thinking, charsPerToken)
- );
+ const text = type === "text" ? block.text : type === "thinking" ? block.thinking : undefined;
+ if (typeof text === "string") {
+ return CONTENT_BLOCK_OVERHEAD_TOKENS + estimateStringTokenPressure(text, charsPerToken, mode);
}
if (type === "image") {
return IMAGE_BLOCK_TOKENS;
}
- return CONTENT_BLOCK_OVERHEAD_TOKENS + estimateJsonPayloadTokenPressure(block, charsPerToken);
-}
-
-function estimateToolResultStringTokenPressure(text: string): number {
- const conservativeToolResultEstimate = Math.ceil(text.length / TOOL_RESULT_CHARS_PER_TOKEN);
- const cjkAwareEstimate = estimateStringTokenPressure(text);
- return Math.max(conservativeToolResultEstimate, cjkAwareEstimate);
-}
-
-function estimateToolResultJsonTokenPressure(value: unknown): number {
- try {
- const serialized = JSON.stringify(value);
- return typeof serialized === "string" ? estimateToolResultStringTokenPressure(serialized) : 1;
- } catch {
- return 256;
- }
-}
-
-function estimateToolResultBlockTokenPressure(block: unknown): number {
- if (typeof block === "string") {
- return estimateToolResultStringTokenPressure(block);
- }
- if (!isRecord(block)) {
- return estimateToolResultJsonTokenPressure(block);
- }
-
- if (block.type === "text" && typeof block.text === "string") {
- return CONTENT_BLOCK_OVERHEAD_TOKENS + estimateToolResultStringTokenPressure(block.text);
- }
- if (block.type === "thinking" && typeof block.thinking === "string") {
- return CONTENT_BLOCK_OVERHEAD_TOKENS + estimateToolResultStringTokenPressure(block.thinking);
- }
- if (block.type === "image") {
- return IMAGE_BLOCK_TOKENS;
- }
- return CONTENT_BLOCK_OVERHEAD_TOKENS + estimateToolResultJsonTokenPressure(block);
-}
-
-function estimateToolResultContentTokenPressure(content: unknown): number {
- if (typeof content === "string") {
- return estimateToolResultStringTokenPressure(content);
- }
- if (Array.isArray(content)) {
- return content.reduce((sum, block) => sum + estimateToolResultBlockTokenPressure(block), 0);
- }
- if (content !== undefined) {
- return estimateToolResultJsonTokenPressure(content);
- }
- return 0;
+ return (
+ CONTENT_BLOCK_OVERHEAD_TOKENS + estimateJsonPayloadTokenPressure(block, charsPerToken, mode)
+ );
}
function estimateAssistantToolCallTokenPressure(block: Record): number {
@@ -169,30 +130,36 @@ function estimateAssistantToolCallTokenPressure(block: Record):
);
}
-function estimateContentTokenPressure(content: unknown): number {
+function estimateContentTokenPressure(
+ content: unknown,
+ mode: TokenPressureMode = "general",
+): number {
if (typeof content === "string") {
- return estimateStringTokenPressure(content);
+ return estimateStringTokenPressure(content, ESTIMATED_CHARS_PER_TOKEN, mode);
}
if (Array.isArray(content)) {
- return content.reduce((sum, block) => sum + estimateContentBlockTokenPressure(block), 0);
+ return content.reduce(
+ (sum, block) =>
+ sum + estimateContentBlockTokenPressure(block, ESTIMATED_CHARS_PER_TOKEN, mode),
+ 0,
+ );
}
if (content !== undefined) {
- return estimateJsonPayloadTokenPressure(content);
+ return estimateJsonPayloadTokenPressure(
+ content,
+ mode === "tool-result" ? ESTIMATED_CHARS_PER_TOKEN : JSON_PAYLOAD_CHARS_PER_TOKEN,
+ mode,
+ );
}
return 0;
}
-function isToolResultMessage(message: AgentMessage): boolean {
- const record = message as unknown as { role?: unknown; type?: unknown };
- return record.role === "toolResult" || record.role === "tool" || record.type === "toolResult";
-}
-
function estimateMessageTokenPressure(message: AgentMessage): number {
const record = message as unknown as Record;
let tokens = MESSAGE_BOUNDARY_OVERHEAD_TOKENS;
- if (isToolResultMessage(message)) {
- tokens += estimateToolResultContentTokenPressure(record.content);
+ if (record.role === "toolResult" || record.role === "tool" || record.type === "toolResult") {
+ tokens += estimateContentTokenPressure(record.content, "tool-result");
tokens += estimateIdentifierTokenPressure(record.toolName ?? record.tool_name);
return tokens;
}
@@ -207,18 +174,13 @@ function estimateMessageTokenPressure(message: AgentMessage): number {
return tokens;
}
- if (record.role === "branchSummary") {
+ if (record.role === "branchSummary" || record.role === "compactionSummary") {
const summary = typeof record.summary === "string" ? record.summary : "";
- tokens += estimateStringTokenPressure(BRANCH_SUMMARY_PREFIX + summary + BRANCH_SUMMARY_SUFFIX);
- return tokens;
- }
-
- if (record.role === "compactionSummary") {
- const summary = typeof record.summary === "string" ? record.summary : "";
- tokens += estimateStringTokenPressure(
- COMPACTION_SUMMARY_PREFIX + summary + COMPACTION_SUMMARY_SUFFIX,
- );
- return tokens;
+ const [prefix, suffix] =
+ record.role === "branchSummary"
+ ? [BRANCH_SUMMARY_PREFIX, BRANCH_SUMMARY_SUFFIX]
+ : [COMPACTION_SUMMARY_PREFIX, COMPACTION_SUMMARY_SUFFIX];
+ return tokens + estimateStringTokenPressure(prefix + summary + suffix);
}
if (record.role === "assistant") {
@@ -255,6 +217,16 @@ function estimateMessageTokenPressure(message: AgentMessage): number {
* optional system prompt, and current prompt text. The result intentionally
* includes a safety margin because this path runs before provider tokenization.
*/
+function estimateRenderedPromptTokens(params: { systemPrompt?: string; prompt: string }): number {
+ const systemTokens =
+ typeof params.systemPrompt === "string" && params.systemPrompt.trim().length > 0
+ ? MESSAGE_BOUNDARY_OVERHEAD_TOKENS + estimateStringTokenPressure(params.systemPrompt)
+ : 0;
+ return (
+ systemTokens + MESSAGE_BOUNDARY_OVERHEAD_TOKENS + estimateStringTokenPressure(params.prompt)
+ );
+}
+
export function estimateLlmBoundaryTokenPressure(params: {
messages: AgentMessage[];
systemPrompt?: string;
@@ -264,13 +236,10 @@ export function estimateLlmBoundaryTokenPressure(params: {
(sum, message) => sum + estimateMessageTokenPressure(message),
0,
);
- const systemTokens =
- typeof params.systemPrompt === "string" && params.systemPrompt.trim().length > 0
- ? MESSAGE_BOUNDARY_OVERHEAD_TOKENS + estimateStringTokenPressure(params.systemPrompt)
- : 0;
- const promptTokens =
- MESSAGE_BOUNDARY_OVERHEAD_TOKENS + estimateStringTokenPressure(params.prompt);
- return Math.max(0, Math.ceil((historyTokens + systemTokens + promptTokens) * SAFETY_MARGIN));
+ return Math.max(
+ 0,
+ Math.ceil((historyTokens + estimateRenderedPromptTokens(params)) * SAFETY_MARGIN),
+ );
}
/** Estimates only the rendered prompt/system portion when history has already been accounted for. */
@@ -278,13 +247,7 @@ export function estimateRenderedLlmBoundaryTokenPressure(params: {
systemPrompt?: string;
prompt: string;
}): number {
- const systemTokens =
- typeof params.systemPrompt === "string" && params.systemPrompt.trim().length > 0
- ? MESSAGE_BOUNDARY_OVERHEAD_TOKENS + estimateStringTokenPressure(params.systemPrompt)
- : 0;
- const promptTokens =
- MESSAGE_BOUNDARY_OVERHEAD_TOKENS + estimateStringTokenPressure(params.prompt);
- return Math.max(0, Math.ceil((systemTokens + promptTokens) * SAFETY_MARGIN));
+ return Math.max(0, Math.ceil(estimateRenderedPromptTokens(params) * SAFETY_MARGIN));
}
function normalizeLlmBoundaryTokenPressure(
diff --git a/src/agents/embedded-agent-runner/tool-result-truncation.ts b/src/agents/embedded-agent-runner/tool-result-truncation.ts
index 004d040169b8..18bd25cc5f97 100644
--- a/src/agents/embedded-agent-runner/tool-result-truncation.ts
+++ b/src/agents/embedded-agent-runner/tool-result-truncation.ts
@@ -64,7 +64,7 @@ export function resolveCacheTtlPruningSettings(
} catch {
// Invalid durations retain the shipped five-minute default.
}
- const normalize = (value: string) => normalizeLowercaseStringOrEmpty(value);
+ const normalize = normalizeLowercaseStringOrEmpty;
const deny = compileGlobPatterns({ raw: config.tools?.deny, normalize });
const allow = compileGlobPatterns({ raw: config.tools?.allow, normalize });
return {
@@ -107,22 +107,26 @@ function cacheTtlMessageChars(message: AgentMessage): number {
}
const content = Array.isArray(message.content) ? message.content : [];
return content.reduce((chars, block) => {
+ if (!isRecord(block)) {
+ return chars;
+ }
const text = cacheTtlText(block, message.role !== "assistant");
if (text !== undefined) {
return chars + estimateStringChars(text);
}
- if (isRecord(block) && block.type === "image") {
+ if (block.type === "image") {
return chars + CACHE_TTL_IMAGE_CHARS;
}
- if (!isRecord(block) || message.role !== "assistant") {
+ if (message.role !== "assistant") {
return chars;
}
const record = block as Record;
if (record.type === "thinking" || record.type === "redacted_thinking") {
- const values = [record.thinking, record.thinkingSignature];
- if (record.type === "redacted_thinking") {
- values.push(record.data);
- }
+ const values = [
+ record.thinking,
+ record.thinkingSignature,
+ ...(record.type === "redacted_thinking" ? [record.data] : []),
+ ];
return values.reduce(
(sum, value) => sum + (typeof value === "string" ? estimateStringChars(value) : 0),
chars,
@@ -209,11 +213,12 @@ export function pruneExpiredCacheTtlToolResults(params: {
(next ??= messages.slice())[index] = projected;
}
}
- if (totalChars / charWindow < 0.5 || !settings.hardClear) {
- return next ?? messages;
- }
const output = next ?? messages;
- if (eligible.reduce((sum, index) => sum + cacheTtlMessageChars(output[index]!), 0) < 50_000) {
+ if (
+ totalChars / charWindow < 0.5 ||
+ !settings.hardClear ||
+ eligible.reduce((sum, index) => sum + cacheTtlMessageChars(output[index]!), 0) < 50_000
+ ) {
return output;
}
for (const index of eligible) {
@@ -510,18 +515,10 @@ export function truncateToolResultMessage(
const preserveSmallBlocks = smallBlockChars + largeBlockNoticeChars <= maxChars;
const preservedChars = preserveSmallBlocks ? smallBlockChars : 0;
const remainingBudget = Math.max(0, maxChars - preservedChars);
- const reducibleChars = blockTextChars.reduce(
- (sum, chars) => sum + (preserveSmallBlocks && chars > 0 && chars <= minKeepChars ? 0 : chars),
- 0,
- );
- const reducibleNoticeChars = blockTextChars.reduce(
- (sum, chars, index) =>
- sum +
- (preserveSmallBlocks && chars > 0 && chars <= minKeepChars
- ? 0
- : (blockNoticeChars[index] ?? 0)),
- 0,
- );
+ const reducibleChars = totalTextChars - preservedChars;
+ const reducibleNoticeChars = preserveSmallBlocks
+ ? largeBlockNoticeChars
+ : blockNoticeChars.reduce((sum, chars) => sum + chars, 0);
const noticeScale =
reducibleNoticeChars > 0 ? Math.min(1, remainingBudget / reducibleNoticeChars) : 0;
const distributableBudget = Math.max(0, remainingBudget - reducibleNoticeChars);
@@ -566,13 +563,7 @@ function isToolResultTextBlock(
);
}
-type ToolResultSpillDetails = {
- path: string;
- truncated: boolean;
- chars?: number;
-};
-
-function getToolResultSpillDetails(message: AgentMessage): ToolResultSpillDetails | undefined {
+function getToolResultSpillDetails(message: AgentMessage) {
const details = (message as { details?: unknown }).details;
if (!isRecord(details)) {
return undefined;
@@ -638,21 +629,6 @@ function resolveAggregateElisionMarkers(
};
}
-function formatAggregateElisionText(
- remainingTextBudget: number,
- spillMarkers: AggregateElisionMarkers | undefined,
-): string {
- if (remainingTextBudget <= 0) {
- return "";
- }
- for (const marker of [spillMarkers?.full, spillMarkers?.compact]) {
- if (marker && estimateToolResultTextChars(marker) <= remainingTextBudget) {
- return marker;
- }
- }
- return sliceToolResultTextToBudget(AGGREGATE_ELISION_MARKER, remainingTextBudget);
-}
-
/** Projects bounded tool-result history without mutating the transcript. */
export function truncateOversizedToolResultsInMessages(
messages: AgentMessage[],
@@ -667,50 +643,25 @@ export function truncateOversizedToolResultsInMessages(
aggregatePressureEngaged: boolean;
aggregateBudgetChars: number;
} {
- const maxChars = Math.max(
- 1,
- maxCharsOverride ?? calculateMaxToolResultChars(contextWindowTokens),
- );
- const aggregateBudgetChars = calculateRecoveryAggregateToolResultChars(
+ const { maxChars, aggregateBudgetChars } = resolveToolResultBudgets({
contextWindowTokens,
- maxChars,
+ maxCharsOverride,
aggregateMaxCharsOverride,
- );
- const projectionKeys = projectionState
- ? getToolResultProjectionKeys(messages, projectionState)
- : [];
- const hasFrozenProjectionBaseline = (projectionState?.frozen.size ?? 0) > 0;
- const branch = messages.map((message, index) => {
- const projectionKey = projectionKeys[index];
- const projectedMessage = projectionKey
- ? projectionState?.replacements.get(projectionKey)
- : undefined;
- if (projectionKey && projectionState && !projectionState.sourceTextByKey.has(projectionKey)) {
- projectionState.sourceTextByKey.set(projectionKey, getToolResultTextBlocks(message));
- }
- const mergedMessage = projectedMessage
- ? mergeProjectedToolResultMessage(
- message,
- projectedMessage,
- projectionState?.sourceTextByKey.get(projectionKey ?? ""),
- )
- : message;
- return {
- id: `message-${index}`,
- type: "message",
- message: mergedMessage,
- aggregateEligible:
- !projectionKey ||
- !projectionState?.frozen.has(projectionKey) ||
- (projectedMessage !== undefined && mergedMessage === message),
- // Reduce frozen history first so steering cannot make fresh output disappear.
- deferAggregateRecovery:
- projectionKey !== undefined &&
- projectionState !== undefined &&
- hasFrozenProjectionBaseline &&
- !projectionState.frozen.has(projectionKey),
- };
});
+ const sourceBranch = messages.map((message, index) => ({
+ id: `message-${index}`,
+ type: "message",
+ message,
+ }));
+ const projection = projectionState
+ ? projectToolResultBranch({
+ branch: sourceBranch,
+ projectionState,
+ recordSources: true,
+ })
+ : undefined;
+ const branch = projection?.branch ?? sourceBranch;
+ const projectionKeys = projection?.keys ?? [];
const plan = buildToolResultReplacementPlan({
branch,
maxChars,
@@ -718,7 +669,7 @@ export function truncateOversizedToolResultsInMessages(
minKeepChars: RECOVERY_MIN_KEEP_CHARS,
protectTrailingToolResults: Boolean(projectionState),
});
- const replacedBranch = applyToolResultReplacementsToBranch(branch, plan.replacements);
+ const replacedBranch = plan.branch;
if (projectionState) {
for (const [index, originalMessage] of messages.entries()) {
const projectedMessage = replacedBranch[index]?.message;
@@ -745,19 +696,26 @@ export function truncateOversizedToolResultsInMessages(
};
}
-function calculateRecoveryAggregateToolResultChars(
- contextWindowTokens: number,
- maxCharsOverride?: number,
- aggregateMaxCharsOverride?: number,
-): number {
- return Math.max(
+function resolveToolResultBudgets(params: {
+ contextWindowTokens: number;
+ maxCharsOverride?: number;
+ aggregateMaxCharsOverride?: number;
+}): { maxChars: number; aggregateBudgetChars: number } {
+ const maxChars = Math.max(
1,
- aggregateMaxCharsOverride ??
- resolveLiveToolResultAggregateMaxChars({
- contextWindowTokens,
- perResultMaxChars: maxCharsOverride ?? calculateMaxToolResultChars(contextWindowTokens),
- }),
+ params.maxCharsOverride ?? calculateMaxToolResultChars(params.contextWindowTokens),
);
+ return {
+ maxChars,
+ aggregateBudgetChars: Math.max(
+ 1,
+ params.aggregateMaxCharsOverride ??
+ resolveLiveToolResultAggregateMaxChars({
+ contextWindowTokens: params.contextWindowTokens,
+ perResultMaxChars: maxChars,
+ }),
+ ),
+ };
}
type ToolResultReductionPotential = {
@@ -849,78 +807,74 @@ function mergeProjectedToolResultMessage(
if (!Array.isArray(currentContent) || !Array.isArray(projectedContent)) {
return projectedMessage;
}
- const projectedText = projectedContent.filter(
- (block): block is { type: "text"; text: string } =>
- Boolean(block) &&
- typeof block === "object" &&
- (block as { type?: unknown }).type === "text" &&
- typeof (block as { text?: unknown }).text === "string",
+ const projectedText = projectedContent.flatMap((block) =>
+ isRecord(block) && block.type === "text" && typeof block.text === "string" ? [block.text] : [],
);
const currentText = getToolResultTextBlocks(message);
- if (sourceText && currentText.some((text, index) => text !== sourceText[index])) {
- return message;
- }
- const currentTextCount = currentContent.filter(
- (block) =>
- Boolean(block) && typeof block === "object" && (block as { type?: unknown }).type === "text",
- ).length;
- if (currentTextCount !== projectedText.length) {
+ if (
+ (sourceText && currentText.some((text, index) => text !== sourceText[index])) ||
+ currentText.length !== projectedText.length
+ ) {
return message;
}
let textIndex = 0;
const mergedContent = currentContent.map((block) => {
- if (!block || typeof block !== "object" || (block as { type?: unknown }).type !== "text") {
+ if (!isRecord(block) || block.type !== "text") {
return block;
}
- const projectedBlock = projectedText[textIndex++];
- return projectedBlock ? Object.assign({}, block, { text: projectedBlock.text }) : block;
+ return Object.assign({}, block, { text: projectedText[textIndex++] });
});
return { ...message, content: mergedContent } as AgentMessage;
}
-function seedRecoveryBranchFromFrozenProjection(params: {
+function projectToolResultBranch(params: {
branch: ToolResultBranchEntry[];
projectionState: ToolResultPromptProjectionState;
-}): ToolResultBranchEntry[] {
+ frozenOnly?: boolean;
+ recordSources?: boolean;
+}): { branch: ToolResultBranchEntry[]; keys: Array } {
const messageEntries = params.branch.filter(
(entry): entry is ToolResultBranchEntry & { message: AgentMessage } =>
entry.type === "message" && entry.message !== undefined,
);
- const projectionKeys = getToolResultProjectionKeys(
+ const keys = getToolResultProjectionKeys(
messageEntries.map((entry) => entry.message),
params.projectionState,
);
const hasFrozenProjectionBaseline = params.projectionState.frozen.size > 0;
let messageIndex = 0;
- return params.branch.map((entry) => {
- if (entry.type !== "message" || !entry.message) {
- return entry;
- }
- const projectionKey = projectionKeys[messageIndex++];
- const projectedMessage =
- projectionKey && params.projectionState.frozen.has(projectionKey)
- ? params.projectionState.replacements.get(projectionKey)
- : undefined;
- const message = projectedMessage
- ? mergeProjectedToolResultMessage(
- entry.message,
- projectedMessage,
- projectionKey ? params.projectionState.sourceTextByKey.get(projectionKey) : undefined,
- )
- : entry.message;
- return {
- ...entry,
- message,
- aggregateEligible:
- !projectionKey ||
- !params.projectionState.frozen.has(projectionKey) ||
- (projectedMessage !== undefined && message === entry.message),
- deferAggregateRecovery:
- projectionKey !== undefined &&
- hasFrozenProjectionBaseline &&
- !params.projectionState.frozen.has(projectionKey),
- };
- });
+ return {
+ keys,
+ branch: params.branch.map((entry) => {
+ if (entry.type !== "message" || !entry.message) {
+ return entry;
+ }
+ const key = keys[messageIndex++];
+ const frozen = key !== undefined && params.projectionState.frozen.has(key);
+ const projected =
+ key && (!params.frozenOnly || frozen)
+ ? params.projectionState.replacements.get(key)
+ : undefined;
+ if (key && params.recordSources && !params.projectionState.sourceTextByKey.has(key)) {
+ params.projectionState.sourceTextByKey.set(key, getToolResultTextBlocks(entry.message));
+ }
+ const message = projected
+ ? mergeProjectedToolResultMessage(
+ entry.message,
+ projected,
+ key ? params.projectionState.sourceTextByKey.get(key) : undefined,
+ )
+ : entry.message;
+ return {
+ ...entry,
+ message,
+ aggregateEligible:
+ !key || !frozen || (projected !== undefined && message === entry.message),
+ // Reduce frozen history first so steering cannot make fresh output disappear.
+ deferAggregateRecovery: key !== undefined && hasFrozenProjectionBaseline && !frozen,
+ };
+ }),
+ };
}
function getToolResultTextBlocks(message: AgentMessage): string[] {
@@ -939,26 +893,22 @@ function buildAggregateToolResultReplacements(params: {
spillSourceBranch?: ToolResultBranchEntry[];
aggregateBudgetChars: number;
minKeepChars?: number;
- protectTrailingToolResults?: boolean;
+ protectedEntryIds?: Set;
}): { replacements: ToolResultReplacement[]; pressureExceeded: boolean } {
const minKeepChars = params.minKeepChars ?? MIN_KEEP_CHARS;
- const protectedEntryIds = params.protectTrailingToolResults
- ? getTrailingToolResultEntryIds(params.branch)
- : new Set();
const candidates = params.branch
.flatMap((entry, index) => {
const message = entry.message;
return entry.type === "message" && message?.role === "toolResult"
? [
{
- index,
entryId: entry.id,
message,
spillSourceMessage: params.spillSourceBranch?.[index]?.message ?? message,
textLength: getToolResultTextBudget(message),
aggregateEligible: entry.aggregateEligible !== false,
deferredByFreshProjection: entry.deferAggregateRecovery === true,
- protectedByTrailingBatch: protectedEntryIds.has(entry.id),
+ protectedByTrailingBatch: params.protectedEntryIds?.has(entry.id) ?? false,
},
]
: [];
@@ -983,68 +933,53 @@ function buildAggregateToolResultReplacements(params: {
let remainingReduction = totalChars - params.aggregateBudgetChars;
const replacements = new Map();
- const aggregateRecoveryCandidates = candidates
- .filter((item) => !item.deferredByFreshProjection && !item.protectedByTrailingBatch)
- .toSorted((a, b) => a.index - b.index);
- const recoveryCandidates = [
- ...aggregateRecoveryCandidates.filter((item) => item.aggregateEligible),
- // Reuse frozen projections first to keep reduction shrink-only and cache-stable.
- ...aggregateRecoveryCandidates.filter((item) => !item.aggregateEligible),
- ...candidates.filter(
- (item) => item.deferredByFreshProjection && !item.protectedByTrailingBatch,
- ),
- ];
-
- // Spend aggregate reduction on older entries first so fresh tool output stays intact.
- for (const candidate of recoveryCandidates) {
- if (remainingReduction <= 0) {
- break;
- }
- const reducibleChars = Math.max(0, candidate.textLength - minTruncatedTextChars);
- if (reducibleChars <= 0) {
- continue;
- }
-
- const requestedReduction = Math.min(reducibleChars, remainingReduction);
- const targetChars = Math.max(minTruncatedTextChars, candidate.textLength - requestedReduction);
- const spillMarkers = resolveAggregateElisionMarkers(candidate.spillSourceMessage);
- const candidateSuffixFactory = spillMarkers?.truncationSuffix ?? suffixFactory;
- const candidateTargetChars = Math.max(
- targetChars,
- estimateToolResultTextChars(candidateSuffixFactory(1)),
+ // Frozen projections shrink first; stable sorting preserves the original oldest-first order.
+ const recoveryCandidates = candidates
+ .filter((candidate) => !candidate.protectedByTrailingBatch)
+ .toSorted(
+ (left, right) =>
+ Number(left.deferredByFreshProjection) - Number(right.deferredByFreshProjection) ||
+ Number(right.aggregateEligible) - Number(left.aggregateEligible),
);
- const truncatedMessage = truncateToolResultMessage(candidate.message, candidateTargetChars, {
- minKeepChars,
- suffix: candidateSuffixFactory,
- });
- const newLength = getToolResultTextBudget(truncatedMessage);
- const actualReduction = Math.max(0, candidate.textLength - newLength);
- if (actualReduction <= 0) {
- continue;
- }
- replacements.set(candidate.entryId, { entryId: candidate.entryId, message: truncatedMessage });
- remainingReduction -= actualReduction;
- }
-
- for (const candidate of recoveryCandidates) {
- if (remainingReduction <= 0) {
- break;
+ // Trim all older entries before clearing any, so fresh output and spill pointers stay recoverable.
+ for (const clear of [false, true]) {
+ for (const candidate of recoveryCandidates) {
+ if (remainingReduction <= 0) {
+ break;
+ }
+ const baseMessage = replacements.get(candidate.entryId)?.message ?? candidate.message;
+ const baseTextLength = getToolResultTextBudget(baseMessage);
+ if (!clear && baseTextLength <= minTruncatedTextChars) {
+ continue;
+ }
+ const spillMarkers = resolveAggregateElisionMarkers(candidate.spillSourceMessage);
+ let message: AgentMessage;
+ if (clear) {
+ message = clearToolResultText(
+ candidate.message,
+ Math.max(0, baseTextLength - remainingReduction),
+ spillMarkers,
+ );
+ } else {
+ const suffix = spillMarkers?.truncationSuffix ?? suffixFactory;
+ const targetChars = Math.max(
+ minTruncatedTextChars,
+ baseTextLength - remainingReduction,
+ estimateToolResultTextChars(suffix(1)),
+ );
+ message = truncateToolResultMessage(candidate.message, targetChars, {
+ minKeepChars,
+ suffix,
+ });
+ }
+ const actualReduction = Math.max(0, baseTextLength - getToolResultTextBudget(message));
+ if (actualReduction <= 0 && (!clear || !spillMarkers)) {
+ continue;
+ }
+ replacements.set(candidate.entryId, { entryId: candidate.entryId, message });
+ remainingReduction -= actualReduction;
}
- const baseMessage = replacements.get(candidate.entryId)?.message ?? candidate.message;
- const baseTextLength = getToolResultTextBudget(baseMessage);
- const spillMarkers = resolveAggregateElisionMarkers(candidate.spillSourceMessage);
- const emptyMessage = clearToolResultText(
- candidate.message,
- Math.max(0, baseTextLength - remainingReduction),
- spillMarkers,
- );
- const actualReduction = Math.max(0, baseTextLength - getToolResultTextBudget(emptyMessage));
- if (actualReduction <= 0 && !spillMarkers) {
- continue;
- }
- replacements.set(candidate.entryId, { entryId: candidate.entryId, message: emptyMessage });
- remainingReduction -= actualReduction;
}
return { replacements: [...replacements.values()], pressureExceeded: true };
@@ -1052,16 +987,14 @@ function buildAggregateToolResultReplacements(params: {
function getTrailingToolResultEntryIds(branch: ToolResultBranchEntry[]): Set {
const ids = new Set();
- let sawMessage = false;
for (let index = branch.length - 1; index >= 0; index--) {
const entry = branch[index];
if (entry?.type !== "message" || !entry.message) {
- if (!sawMessage) {
+ if (ids.size === 0) {
continue;
}
break;
}
- sawMessage = true;
if ((entry.message as { role?: string }).role !== "toolResult") {
break;
}
@@ -1094,7 +1027,12 @@ function clearToolResultText(
if (!isToolResultTextBlock(block)) {
return block;
}
- const replacementText = formatAggregateElisionText(remainingTextBudget, spillMarkers);
+ const replacementText =
+ [spillMarkers?.full, spillMarkers?.compact].find(
+ (marker): marker is string =>
+ typeof marker === "string" &&
+ estimateToolResultTextChars(marker) <= remainingTextBudget,
+ ) ?? sliceToolResultTextToBudget(AGGREGATE_ELISION_MARKER, remainingTextBudget);
remainingTextBudget = Math.max(
0,
remainingTextBudget - estimateToolResultTextChars(replacementText),
@@ -1107,82 +1045,27 @@ function clearToolResultText(
} as AgentMessage;
}
-function buildOversizedToolResultReplacements(params: {
- branch: ToolResultBranchEntry[];
- maxChars: number;
- minKeepChars?: number;
- protectedEntryIds?: Set;
-}): ToolResultReplacement[] {
- const minKeepChars = params.minKeepChars ?? MIN_KEEP_CHARS;
- const replacements: ToolResultReplacement[] = [];
-
- for (const entry of params.branch) {
- if (entry.type !== "message" || !entry.message) {
- continue;
- }
- const msg = entry.message;
- if ((msg as { role?: string }).role !== "toolResult") {
- continue;
- }
- if (getToolResultTextBudget(msg) <= params.maxChars) {
- continue;
- }
- const replacementMinKeepChars = params.protectedEntryIds?.has(entry.id)
- ? Math.max(minKeepChars, MIN_KEEP_CHARS)
- : minKeepChars;
- const spillMarkers = resolveAggregateElisionMarkers(msg);
- const suffixFactory = spillMarkers?.truncationSuffix;
- const maxChars = Math.max(
- params.maxChars,
- suffixFactory ? estimateToolResultTextChars(suffixFactory(1)) : 0,
- );
- replacements.push({
- entryId: entry.id,
- message: truncateToolResultMessage(msg, maxChars, {
- minKeepChars: replacementMinKeepChars,
- ...(suffixFactory ? { suffix: suffixFactory } : {}),
- }),
- });
- }
-
- return replacements;
-}
-
-function calculateReplacementReduction(
- branch: ToolResultBranchEntry[],
- replacements: ToolResultReplacement[],
-): number {
- if (replacements.length === 0) {
- return 0;
- }
- const branchById = new Map(branch.map((entry) => [entry.id, entry]));
- return replacements.reduce((reduction, replacement) => {
- const entry = branchById.get(replacement.entryId);
- if (!entry?.message) {
- return reduction;
- }
- return (
- reduction +
- Math.max(
- 0,
- getToolResultTextBudget(entry.message) - getToolResultTextBudget(replacement.message),
- )
- );
- }, 0);
-}
-
function applyToolResultReplacementsToBranch(
branch: ToolResultBranchEntry[],
replacements: ToolResultReplacement[],
-): ToolResultBranchEntry[] {
+): { branch: ToolResultBranchEntry[]; reducedChars: number } {
if (replacements.length === 0) {
- return branch;
+ return { branch, reducedChars: 0 };
}
const replacementsById = new Map(replacements.map(({ entryId, message }) => [entryId, message]));
- return branch.map((entry) => {
+ let reducedChars = 0;
+ const nextBranch = branch.map((entry) => {
const message = replacementsById.get(entry.id);
- return message && entry.type === "message" ? { ...entry, message } : entry;
+ if (!message || entry.type !== "message" || !entry.message) {
+ return entry;
+ }
+ reducedChars += Math.max(
+ 0,
+ getToolResultTextBudget(entry.message) - getToolResultTextBudget(message),
+ );
+ return { ...entry, message };
});
+ return { branch: nextBranch, reducedChars };
}
function buildToolResultReplacementPlan(params: {
@@ -1192,6 +1075,7 @@ function buildToolResultReplacementPlan(params: {
minKeepChars?: number;
protectTrailingToolResults?: boolean;
}): {
+ branch: ToolResultBranchEntry[];
replacements: ToolResultReplacement[];
oversizedReplacementCount: number;
aggregateReplacementCount: number;
@@ -1203,40 +1087,50 @@ function buildToolResultReplacementPlan(params: {
const protectedEntryIds = params.protectTrailingToolResults
? getTrailingToolResultEntryIds(params.branch)
: undefined;
- const oversizedReplacements = buildOversizedToolResultReplacements({
- branch: params.branch,
- maxChars: params.maxChars,
- minKeepChars,
- protectedEntryIds,
+ const oversizedReplacements = params.branch.flatMap((entry): ToolResultReplacement[] => {
+ const message = entry.message;
+ if (
+ entry.type !== "message" ||
+ message?.role !== "toolResult" ||
+ getToolResultTextBudget(message) <= params.maxChars
+ ) {
+ return [];
+ }
+ const suffix = resolveAggregateElisionMarkers(message)?.truncationSuffix;
+ const maxChars = Math.max(params.maxChars, suffix ? estimateToolResultTextChars(suffix(1)) : 0);
+ return [
+ {
+ entryId: entry.id,
+ message: truncateToolResultMessage(message, maxChars, {
+ minKeepChars: protectedEntryIds?.has(entry.id)
+ ? Math.max(minKeepChars, MIN_KEEP_CHARS)
+ : minKeepChars,
+ ...(suffix ? { suffix } : {}),
+ }),
+ },
+ ];
});
- const oversizedReducibleChars = calculateReplacementReduction(
- params.branch,
- oversizedReplacements,
- );
- const oversizedTrimmedBranch = applyToolResultReplacementsToBranch(
- params.branch,
- oversizedReplacements,
- );
+ const oversizedPhase = applyToolResultReplacementsToBranch(params.branch, oversizedReplacements);
const aggregatePlan = buildAggregateToolResultReplacements({
- branch: oversizedTrimmedBranch,
+ branch: oversizedPhase.branch,
spillSourceBranch: params.branch,
aggregateBudgetChars: params.aggregateBudgetChars,
minKeepChars,
- protectTrailingToolResults: params.protectTrailingToolResults,
+ protectedEntryIds,
});
- const aggregateReplacements = aggregatePlan.replacements;
- const aggregateReducibleChars = calculateReplacementReduction(
- oversizedTrimmedBranch,
- aggregateReplacements,
+ const aggregatePhase = applyToolResultReplacementsToBranch(
+ oversizedPhase.branch,
+ aggregatePlan.replacements,
);
return {
- replacements: [...oversizedReplacements, ...aggregateReplacements],
+ branch: aggregatePhase.branch,
+ replacements: [...oversizedReplacements, ...aggregatePlan.replacements],
oversizedReplacementCount: oversizedReplacements.length,
- aggregateReplacementCount: aggregateReplacements.length,
+ aggregateReplacementCount: aggregatePlan.replacements.length,
aggregatePressureExceeded: aggregatePlan.pressureExceeded,
- oversizedReducibleChars,
- aggregateReducibleChars,
+ oversizedReducibleChars: oversizedPhase.reducedChars,
+ aggregateReducibleChars: aggregatePhase.reducedChars,
};
}
@@ -1252,20 +1146,13 @@ function buildRecoveryToolResultReplacementPlan(params: {
aggregateBudgetChars: number;
plan: ReturnType;
} {
- const maxChars = Math.max(
- 1,
- params.maxCharsOverride ?? calculateMaxToolResultChars(params.contextWindowTokens),
- );
- const aggregateBudgetChars = calculateRecoveryAggregateToolResultChars(
- params.contextWindowTokens,
- maxChars,
- params.aggregateMaxCharsOverride,
- );
+ const { maxChars, aggregateBudgetChars } = resolveToolResultBudgets(params);
const projectedBranch = params.projectionState
- ? seedRecoveryBranchFromFrozenProjection({
+ ? projectToolResultBranch({
branch: params.branch,
projectionState: params.projectionState,
- })
+ frozenOnly: true,
+ }).branch
: params.branch;
const plan = buildToolResultReplacementPlan({
branch: projectedBranch,
@@ -1274,9 +1161,8 @@ function buildRecoveryToolResultReplacementPlan(params: {
minKeepChars: RECOVERY_MIN_KEEP_CHARS,
protectTrailingToolResults: params.protectTrailingToolResults,
});
- const finalBranch = applyToolResultReplacementsToBranch(projectedBranch, plan.replacements);
const replacements = params.branch.flatMap((entry, index) => {
- const finalEntry = finalBranch[index];
+ const finalEntry = plan.branch[index];
if (
entry.type !== "message" ||
!entry.message ||
@@ -1304,16 +1190,8 @@ export function estimateToolResultReductionPotential(params: {
maxCharsOverride?: number;
aggregateMaxCharsOverride?: number;
}): ToolResultReductionPotential {
- const { messages, contextWindowTokens } = params;
- const maxChars = Math.max(
- 1,
- params.maxCharsOverride ?? calculateMaxToolResultChars(contextWindowTokens),
- );
- const aggregateBudgetChars = calculateRecoveryAggregateToolResultChars(
- contextWindowTokens,
- maxChars,
- params.aggregateMaxCharsOverride,
- );
+ const { messages } = params;
+ const { maxChars, aggregateBudgetChars } = resolveToolResultBudgets(params);
const branch = messages.map((message, index) => ({
id: `message-${index}`,
type: "message",
diff --git a/src/agents/embedded-agent-subscribe.handlers.compaction.ts b/src/agents/embedded-agent-subscribe.handlers.compaction.ts
index d082a36f729c..a08a470bd197 100644
--- a/src/agents/embedded-agent-subscribe.handlers.compaction.ts
+++ b/src/agents/embedded-agent-subscribe.handlers.compaction.ts
@@ -44,6 +44,48 @@ function compactionLogKind(reason: CompactionReason): string {
return reason === "manual" ? "manual compaction" : "auto-compaction";
}
+function emitCompactionAgentEvent(
+ ctx: EmbeddedAgentSubscribeContext,
+ data: { phase: "start" } | { phase: "end"; willRetry: boolean; completed: boolean },
+): void {
+ const event = { stream: "compaction" as const, data };
+ emitAgentEvent({ runId: ctx.params.runId, ...event });
+ runBestEffortCallback({
+ label: "compaction agent event",
+ log: ctx.log,
+ callback: () => ctx.params.onAgentEvent?.(event),
+ });
+}
+
+function runBestEffortCompactionHook(
+ ctx: EmbeddedAgentSubscribeContext,
+ phase: "before" | "after",
+): void {
+ const hookRunner = getGlobalHookRunner();
+ const hookName = phase === "before" ? "before_compaction" : "after_compaction";
+ if (!hookRunner?.hasHooks(hookName)) {
+ return;
+ }
+ const metrics = {
+ messageCount: ctx.params.session.messages?.length ?? 0,
+ sessionFile: ctx.params.session.sessionFile,
+ };
+ const context = { sessionKey: ctx.params.sessionKey };
+ const hook =
+ phase === "before"
+ ? hookRunner.runBeforeCompaction(
+ { ...metrics, messages: ctx.params.session.messages },
+ context,
+ )
+ : hookRunner.runAfterCompaction(
+ { ...metrics, compactedCount: ctx.getCompactionCount() },
+ context,
+ );
+ void hook.catch((err: unknown) => {
+ ctx.log.warn(`${hookName} hook failed: ${String(err)}`);
+ });
+}
+
/** Handles compaction start events from an embedded agent session. */
export function handleCompactionStart(
ctx: EmbeddedAgentSubscribeContext,
@@ -60,40 +102,11 @@ export function handleCompactionStart(
reason,
consoleMessage: `embedded run ${kind} start: runId=${ctx.params.runId} reason=${reason}`,
});
- emitAgentEvent({
- runId: ctx.params.runId,
- stream: "compaction",
- data: { phase: "start" },
- });
- runBestEffortCallback({
- label: "compaction agent event",
- log: ctx.log,
- callback: () =>
- ctx.params.onAgentEvent?.({
- stream: "compaction",
- data: { phase: "start" },
- }),
- });
+ emitCompactionAgentEvent(ctx, { phase: "start" });
// Hooks are fire-and-forget so compaction state updates and liveness pauses
// cannot be delayed by plugin work.
- const hookRunner = getGlobalHookRunner();
- if (hookRunner?.hasHooks("before_compaction")) {
- void hookRunner
- .runBeforeCompaction(
- {
- messageCount: ctx.params.session.messages?.length ?? 0,
- messages: ctx.params.session.messages,
- sessionFile: ctx.params.session.sessionFile,
- },
- {
- sessionKey: ctx.params.sessionKey,
- },
- )
- .catch((err: unknown) => {
- ctx.log.warn(`before_compaction hook failed: ${String(err)}`);
- });
- }
+ runBestEffortCompactionHook(ctx, "before");
}
/** Handles compaction completion, retry, and incomplete events. */
@@ -107,7 +120,8 @@ export function handleCompactionEnd(ctx: EmbeddedAgentSubscribeContext, evt: Com
// trimming context, and the persisted count must reflect that successful trim.
const hasResult = evt.result != null;
const wasAborted = Boolean(evt.aborted);
- if (hasResult && !wasAborted) {
+ const completed = hasResult && !wasAborted;
+ if (completed) {
ctx.incrementCompactionCount();
const tokensAfter =
typeof evt.result === "object" && evt.result
@@ -130,14 +144,18 @@ export function handleCompactionEnd(ctx: EmbeddedAgentSubscribeContext, evt: Com
compactionCount: observedCompactionCount,
consoleMessage: `embedded run ${kind} complete: runId=${ctx.params.runId} reason=${reason} compactionCount=${observedCompactionCount} willRetry=${willRetry}`,
});
- void reconcileSessionStoreCompactionCountAfterSuccess({
- sessionKey: ctx.params.sessionKey,
- agentId: ctx.params.agentId,
- configStore: ctx.params.config?.session?.store,
- observedCompactionCount,
- }).catch((err: unknown) => {
- ctx.log.warn(`late compaction count reconcile failed: ${String(err)}`);
- });
+ void import("./embedded-agent-subscribe.handlers.compaction.runtime.js")
+ .then(({ default: reconcile }) =>
+ reconcile({
+ sessionKey: ctx.params.sessionKey,
+ agentId: ctx.params.agentId,
+ configStore: ctx.params.config?.session?.store,
+ observedCompactionCount,
+ }),
+ )
+ .catch((err: unknown) => {
+ ctx.log.warn(`late compaction count reconcile failed: ${String(err)}`);
+ });
}
if (willRetry) {
ctx.noteCompactionRetry();
@@ -148,9 +166,17 @@ export function handleCompactionEnd(ctx: EmbeddedAgentSubscribeContext, evt: Com
ctx.state.livenessState = "working";
}
ctx.maybeResolveCompactionWait();
- clearStaleAssistantUsageOnSessionMessages(ctx);
+ const messages = ctx.params.session.messages;
+ if (Array.isArray(messages)) {
+ // Marker-free final compaction has no fresh boundary, so stale totals
+ // must be cleared before later context counters inspect assistant usage.
+ stripStaleAssistantUsageBeforeLatestCompaction(messages, {
+ mutate: true,
+ whenMissingCompactionSummary: "zeroAssistantUsage",
+ });
+ }
}
- if (!hasResult || wasAborted) {
+ if (!completed) {
ctx.log.info(`embedded run ${kind} incomplete`, {
event: "embedded_run_compaction_end",
runId: ctx.params.runId,
@@ -161,65 +187,11 @@ export function handleCompactionEnd(ctx: EmbeddedAgentSubscribeContext, evt: Com
consoleMessage: `embedded run ${kind} incomplete: runId=${ctx.params.runId} reason=${reason} aborted=${wasAborted} willRetry=${willRetry}`,
});
}
- emitAgentEvent({
- runId: ctx.params.runId,
- stream: "compaction",
- data: { phase: "end", willRetry, completed: hasResult && !wasAborted },
- });
- runBestEffortCallback({
- label: "compaction agent event",
- log: ctx.log,
- callback: () =>
- ctx.params.onAgentEvent?.({
- stream: "compaction",
- data: { phase: "end", willRetry, completed: hasResult && !wasAborted },
- }),
- });
+ emitCompactionAgentEvent(ctx, { phase: "end", willRetry, completed });
// after_compaction runs only once the run will not retry, matching the visible
// post-compaction session state plugin authors observe.
if (!willRetry) {
- const hookRunnerEnd = getGlobalHookRunner();
- if (hookRunnerEnd?.hasHooks("after_compaction")) {
- void hookRunnerEnd
- .runAfterCompaction(
- {
- messageCount: ctx.params.session.messages?.length ?? 0,
- compactedCount: ctx.getCompactionCount(),
- sessionFile: ctx.params.session.sessionFile,
- },
- { sessionKey: ctx.params.sessionKey },
- )
- .catch((err: unknown) => {
- ctx.log.warn(`after_compaction hook failed: ${String(err)}`);
- });
- }
+ runBestEffortCompactionHook(ctx, "after");
}
}
-
-/** Lazily reconciles persisted compaction count after a successful compaction. */
-async function reconcileSessionStoreCompactionCountAfterSuccess(params: {
- sessionKey?: string;
- agentId?: string;
- configStore?: string;
- observedCompactionCount: number;
- now?: number;
-}): Promise {
- const { default: reconcile } =
- await import("./embedded-agent-subscribe.handlers.compaction.runtime.js");
- return reconcile(params);
-}
-
-function clearStaleAssistantUsageOnSessionMessages(ctx: EmbeddedAgentSubscribeContext): void {
- const messages = ctx.params.session.messages;
- if (!Array.isArray(messages)) {
- return;
- }
- // Marker-free final compaction has no fresh boundary to compare against.
- // Clear all assistant usage or stale pre-compaction totals keep driving the
- // context counter after cleanup.
- stripStaleAssistantUsageBeforeLatestCompaction(messages, {
- mutate: true,
- whenMissingCompactionSummary: "zeroAssistantUsage",
- });
-}
diff --git a/src/context-engine/context-engine.test.ts b/src/context-engine/context-engine.test.ts
index 5a1c201100b0..0d0a1627990a 100644
--- a/src/context-engine/context-engine.test.ts
+++ b/src/context-engine/context-engine.test.ts
@@ -1036,6 +1036,45 @@ describe("Invalid engine fallback", () => {
);
});
+ it("coalesces fallback initialization across concurrent lifecycle failures", async () => {
+ const defaultFactory = vi.fn(async () => new LegacyContextEngine());
+ registerContextEngineForOwner("legacy", defaultFactory, "core", {
+ allowSameOwnerRefresh: true,
+ });
+ const engineId = uniqueEngineId("concurrent-runtime-fail");
+ const assemble = vi.fn(async () => {
+ await Promise.resolve();
+ throw new Error("plugin context unavailable");
+ });
+ registerTestContextEngine(engineId, () => ({
+ info: { id: engineId, name: "Concurrent Context Engine" },
+ async ingest() {
+ return { ingested: true };
+ },
+ assemble,
+ async compact() {
+ return { ok: true, compacted: false };
+ },
+ }));
+ const engine = await resolveContextEngine(configWithSlot(engineId));
+ const messages = [makeMockMessage("user", "first"), makeMockMessage("user", "second")];
+
+ const results = await Promise.all(
+ messages.map((message, index) =>
+ engine.assemble({ sessionId: `session-${index}`, messages: [message] }),
+ ),
+ );
+
+ expect(results.map(({ messages: assembled }) => assembled)).toEqual(
+ messages.map((message) => [message]),
+ );
+ expect(assemble).toHaveBeenCalledTimes(2);
+ expect(defaultFactory).toHaveBeenCalledTimes(1);
+ expect(listContextEngineQuarantines()).toEqual([
+ expect.objectContaining({ engineId, operation: "assemble" }),
+ ]);
+ });
+
it("exposes fallback metadata on the same engine after lifecycle quarantine", async () => {
const engineId = uniqueEngineId("runtime-fail-metadata");
const assemble = vi.fn(async () => {
diff --git a/src/context-engine/delegate.ts b/src/context-engine/delegate.ts
index b805100b86e7..8ceba9b559aa 100644
--- a/src/context-engine/delegate.ts
+++ b/src/context-engine/delegate.ts
@@ -54,7 +54,8 @@ function buildCompactionResultSessionTarget(params: {
const suppliedAgentId = targetAgentId ?? requestedAgentId;
const suppliedSessionId = normalizeOptionalString(params.sessionId);
const suppliedSessionKey = targetSessionKey ?? requestedSessionKey;
- const callerAgentId = suppliedAgentId ?? parseAgentSessionKey(suppliedSessionKey)?.agentId;
+ const suppliedSessionKeyAgentId = parseAgentSessionKey(suppliedSessionKey)?.agentId;
+ const callerAgentId = suppliedAgentId ?? suppliedSessionKeyAgentId;
if (
(callerAgentId && marker && marker.agentId !== callerAgentId) ||
(targetStorePath && marker && path.resolve(marker.storePath) !== path.resolve(targetStorePath))
@@ -64,12 +65,11 @@ function buildCompactionResultSessionTarget(params: {
if (marker && suppliedSessionId && marker.sessionId !== suppliedSessionId) {
throw new Error("Context-engine successor identity is inconsistent");
}
- const candidateSessionKey = suppliedSessionKey;
const candidateEntry =
- marker && candidateSessionKey
+ marker && suppliedSessionKey
? loadSessionEntry({
agentId: marker.agentId,
- sessionKey: candidateSessionKey,
+ sessionKey: suppliedSessionKey,
storePath: marker.storePath,
})
: undefined;
@@ -86,17 +86,12 @@ function buildCompactionResultSessionTarget(params: {
)
: undefined;
const callerAuthorizedMarkerKey = Boolean(
- candidateSessionKey &&
- suppliedSessionKey &&
- candidateSessionKey === suppliedSessionKey &&
- (!candidateEntry || candidateEntry.sessionId === callerSessionId),
+ suppliedSessionKey && (!candidateEntry || candidateEntry.sessionId === callerSessionId),
);
const markerSessionKey = marker
- ? callerAuthorizedMarkerKey
- ? candidateSessionKey
- : candidateEntry?.sessionId === marker.sessionId
- ? candidateSessionKey
- : (preferredMarkerSessionKey ?? (candidateEntry ? undefined : candidateSessionKey))
+ ? callerAuthorizedMarkerKey || candidateEntry?.sessionId === marker.sessionId
+ ? suppliedSessionKey
+ : (preferredMarkerSessionKey ?? (candidateEntry ? undefined : suppliedSessionKey))
: undefined;
if (sessionFile && !marker) {
throw new Error("Legacy context-engine file successors are unsupported");
@@ -106,7 +101,7 @@ function buildCompactionResultSessionTarget(params: {
}
if (
marker &&
- candidateSessionKey &&
+ suppliedSessionKey &&
((candidateEntry &&
candidateEntry.sessionId !== marker.sessionId &&
!callerAuthorizedMarkerKey) ||
@@ -114,11 +109,7 @@ function buildCompactionResultSessionTarget(params: {
) {
throw new Error("Legacy context-engine successor session key is inconsistent");
}
- if (
- marker &&
- parseAgentSessionKey(candidateSessionKey)?.agentId &&
- parseAgentSessionKey(candidateSessionKey)?.agentId !== marker.agentId
- ) {
+ if (marker && suppliedSessionKeyAgentId && suppliedSessionKeyAgentId !== marker.agentId) {
throw new Error("Legacy context-engine successor identity is inconsistent");
}
const sessionId = marker?.sessionId ?? suppliedSessionId ?? targetSessionId ?? callerSessionId;
@@ -239,16 +230,7 @@ function renderMemorySystemPromptAddition(
params: MemoryPromptSectionParams,
prepared?: PreparedMemoryPromptSection,
): string | undefined {
- const lines = buildMemoryPromptSection(
- {
- availableTools: params.availableTools,
- citationsMode: params.citationsMode,
- agentId: params.agentId,
- agentSessionKey: params.agentSessionKey,
- sandboxed: params.sandboxed,
- },
- prepared,
- );
+ const lines = buildMemoryPromptSection(params, prepared);
if (lines.length === 0) {
return undefined;
}
@@ -277,12 +259,6 @@ export function buildMemorySystemPromptAddition(
export async function prepareMemorySystemPromptAddition(
params: MemoryPromptSectionParams,
): Promise {
- const prepared = await prepareMemoryPromptSection({
- availableTools: params.availableTools,
- citationsMode: params.citationsMode,
- agentId: params.agentId,
- agentSessionKey: params.agentSessionKey,
- sandboxed: params.sandboxed,
- });
+ const prepared = await prepareMemoryPromptSection(params);
return renderMemorySystemPromptAddition(params, prepared);
}
diff --git a/src/context-engine/legacy.ts b/src/context-engine/legacy.ts
index db8406dc6957..acb02e2b85b4 100644
--- a/src/context-engine/legacy.ts
+++ b/src/context-engine/legacy.ts
@@ -1,17 +1,7 @@
// Legacy context engine wraps pre-plugin context behavior behind the pluggable interface.
-import type { AgentMessage } from "../agents/runtime/index.js";
-import type { MemoryCitationsMode } from "../config/types.memory.js";
import { delegateCompactionToRuntime } from "./delegate.js";
import { CONTEXT_ENGINE_HOST_PARAMS } from "./registry.js";
-import type {
- ContextEngine,
- ContextEngineInfo,
- AssembleResult,
- CompactResult,
- ContextEngineRuntimeContext,
- ContextEngineSessionTarget,
- IngestResult,
-} from "./types.js";
+import type { AssembleResult, ContextEngine, ContextEngineInfo } from "./types.js";
/**
* LegacyContextEngine wraps the existing compaction behavior behind the
@@ -29,25 +19,12 @@ export class LegacyContextEngine implements ContextEngine {
acceptedHostParams: [...CONTEXT_ENGINE_HOST_PARAMS],
};
- async ingest(_params: {
- sessionId: string;
- sessionKey?: string;
- message: AgentMessage;
- isHeartbeat?: boolean;
- }): Promise {
+ async ingest(_params: Parameters[0]) {
// No-op: SessionManager handles message persistence in the legacy flow
return { ingested: false };
}
- async assemble(params: {
- sessionId: string;
- sessionKey?: string;
- messages: AgentMessage[];
- tokenBudget?: number;
- availableTools?: Set;
- citationsMode?: MemoryCitationsMode;
- model?: string;
- }): Promise {
+ async assemble(params: Parameters[0]): Promise {
// Pass-through: the existing sanitize -> validate -> limit -> repair pipeline
// in attempt.ts handles context assembly for the legacy engine.
// We just return the messages as-is with a rough token estimate.
@@ -57,33 +34,12 @@ export class LegacyContextEngine implements ContextEngine {
};
}
- async afterTurn(_params: {
- sessionId: string;
- sessionKey?: string;
- sessionFile: string;
- messages: AgentMessage[];
- prePromptMessageCount: number;
- autoCompactionSummary?: string;
- isHeartbeat?: boolean;
- tokenBudget?: number;
- runtimeContext?: ContextEngineRuntimeContext;
- }): Promise {
+ async afterTurn(_params: Parameters>[0]): Promise {
// No-op: legacy flow persists context directly in SessionManager.
}
- async compact(params: {
- sessionId: string;
- sessionKey: string;
- agentId?: string;
- sessionTarget?: ContextEngineSessionTarget;
- tokenBudget?: number;
- force?: boolean;
- currentTokenCount?: number;
- compactionTarget?: "budget" | "threshold";
- customInstructions?: string;
- runtimeContext?: ContextEngineRuntimeContext;
- }): Promise {
- return await delegateCompactionToRuntime(params);
+ compact(params: Parameters[0]) {
+ return delegateCompactionToRuntime(params);
}
async dispose(): Promise {
diff --git a/src/context-engine/quarantine-health.ts b/src/context-engine/quarantine-health.ts
index 5acd357f85d0..389956298fe4 100644
--- a/src/context-engine/quarantine-health.ts
+++ b/src/context-engine/quarantine-health.ts
@@ -14,12 +14,8 @@ type PersistedContextEngineRuntimeQuarantine = {
failedAt: Date;
};
-type PersistedContextEngineQuarantineRecord = RuntimeHealthRecordEnvelope & {
- engineId: string;
- owner?: string;
- operation: string;
- reason: string;
-};
+type PersistedContextEngineQuarantineRecord = RuntimeHealthRecordEnvelope &
+ Omit;
function isNonEmptyString(value: unknown): value is string {
return typeof value === "string" && value.trim().length > 0;
@@ -56,10 +52,6 @@ const quarantineStore = createRuntimeHealthStore) {
- return JSON.stringify([record.engineId, record.processId]);
-}
-
export function recordPersistedContextEngineQuarantine(
quarantine: PersistedContextEngineRuntimeQuarantine,
): void {
@@ -72,19 +64,19 @@ export function recordPersistedContextEngineQuarantine(
};
// The in-memory registry only records the first quarantine per engine, so
// this is called at most once per (engine, process) and overwrite is safe.
- quarantineStore.register(recordKey(record), record);
+ quarantineStore.register(JSON.stringify([record.engineId, record.processId]), record);
}
export function listPersistedContextEngineQuarantines(): PersistedContextEngineRuntimeQuarantine[] {
- return quarantineStore.list().map((record) => {
+ return quarantineStore.list().map(({ engineId, operation, reason, owner, failedAtMs }) => {
const quarantine: PersistedContextEngineRuntimeQuarantine = {
- engineId: record.engineId,
- operation: record.operation,
- reason: record.reason,
- failedAt: new Date(record.failedAtMs),
+ engineId,
+ operation,
+ reason,
+ failedAt: new Date(failedAtMs),
};
- if (record.owner) {
- quarantine.owner = record.owner;
+ if (owner) {
+ quarantine.owner = owner;
}
return quarantine;
});
diff --git a/src/context-engine/registry.ts b/src/context-engine/registry.ts
index c61438e86e60..2f8df3398803 100644
--- a/src/context-engine/registry.ts
+++ b/src/context-engine/registry.ts
@@ -21,7 +21,6 @@ import {
import type {
BootstrapResult,
ContextEngine,
- ContextEngineInfo,
ContextEngineMaintenanceResult,
IngestBatchResult,
IngestResult,
@@ -50,76 +49,130 @@ const GUARDED_CONTEXT_ENGINE_METHODS = new Set(
export const CONTEXT_ENGINE_HOST_PARAMS = new Set(
"sessionKey prompt runtimeSettings sessionTarget runtimeContext".split(" "),
);
-function wrapContextEngineWithHostParamProjection(engine: ContextEngine): ContextEngine {
- const removeAfter = getPluginCompatRecord("context-engine-legacy-host-param-default").removeAfter;
- const accepted = engine.info.acceptedHostParams;
- const engineRecord = engine as unknown as Record;
- const wrappedRecord: Record = {};
- Object.defineProperty(wrappedRecord, "info", { get: () => engine.info });
- for (const methodName of GUARDED_CONTEXT_ENGINE_METHODS) {
- const method = engineRecord[methodName];
- if (typeof method !== "function") {
- continue;
- }
- wrappedRecord[methodName] = (params: Record) => {
- // Removal(2026-08-12): undeclared engines get full params. Contract: context-engine-legacy-host-param-default.
- const useLegacyDefault =
- removeAfter !== undefined && new Date().toISOString().slice(0, 10) <= removeAfter;
- const currentAccepted = accepted ?? (useLegacyDefault ? [] : undefined);
- if (!currentAccepted) {
- return method.call(engine, params);
- }
- const projected = Object.fromEntries(
- Object.entries(params).filter(
- ([key]) => currentAccepted.includes(key) || !CONTEXT_ENGINE_HOST_PARAMS.has(key),
- ),
- );
- return method.call(engine, projected);
- };
- }
- if (engine.dispose) {
- wrappedRecord.dispose = engine.dispose.bind(engine);
- }
- return Object.create(engine, Object.getOwnPropertyDescriptors(wrappedRecord)) as ContextEngine;
-}
-
type ResolvedContextEngineMetadata = {
owner: string;
-};
-
-type RuntimeQuarantineProxyState = {
engineId: string;
- getResolvedFallbackEngine: () => ContextEngine | undefined;
};
-const RESOLVED_CONTEXT_ENGINE_METADATA = new WeakMap<
- ContextEngine,
- ResolvedContextEngineMetadata
->();
-const RUNTIME_QUARANTINE_PROXY_STATE = new WeakMap();
+const resolvedEngineMetadata = new WeakMap();
function wrapResolvedContextEngine(
engine: ContextEngine,
- metadata: {
- owner: string;
- engineId: string;
+ metadata: ResolvedContextEngineMetadata & {
defaultEngineId?: string;
factoryCtx?: ContextEngineFactoryContext;
},
): ContextEngine {
- const projected = wrapContextEngineWithHostParamProjection(engine);
- const wrapped =
+ const removeAfter = getPluginCompatRecord("context-engine-legacy-host-param-default").removeAfter;
+ const accepted = engine.info.acceptedHostParams;
+ const fallback =
metadata.defaultEngineId &&
metadata.factoryCtx &&
metadata.engineId !== metadata.defaultEngineId
- ? wrapContextEngineWithRuntimeQuarantine({
- engine: projected,
- engineId: metadata.engineId,
- owner: metadata.owner,
- defaultEngineId: metadata.defaultEngineId,
- factoryCtx: metadata.factoryCtx,
- })
- : projected;
- RESOLVED_CONTEXT_ENGINE_METADATA.set(wrapped, metadata);
+ ? { defaultEngineId: metadata.defaultEngineId, factoryCtx: metadata.factoryCtx }
+ : undefined;
+ let fallbackEnginePromise: Promise | undefined;
+ let resolvedFallbackEngine: ContextEngine | undefined;
+ const getFallbackEngine = fallback
+ ? () =>
+ (fallbackEnginePromise ??= resolveDefaultContextEngine(
+ fallback.defaultEngineId,
+ fallback.factoryCtx,
+ ).then((resolved) => {
+ resolvedFallbackEngine = resolved;
+ return resolved;
+ }))
+ : undefined;
+ const projectParams = (params: Record) => {
+ // Removal(2026-08-12): undeclared engines get full params. Contract: context-engine-legacy-host-param-default.
+ const useLegacyDefault =
+ removeAfter !== undefined && new Date().toISOString().slice(0, 10) <= removeAfter;
+ const currentAccepted = accepted ?? (useLegacyDefault ? [] : undefined);
+ return currentAccepted
+ ? Object.fromEntries(
+ Object.entries(params).filter(
+ ([key]) => currentAccepted.includes(key) || !CONTEXT_ENGINE_HOST_PARAMS.has(key),
+ ),
+ )
+ : params;
+ };
+
+ // A fresh target keeps Proxy invariants compatible with frozen engines and private getters.
+ const wrapped = new Proxy(
+ Object.create(engine, { info: { get: () => engine.info } }) as ContextEngine,
+ {
+ get(_target, property) {
+ if (property === "info") {
+ if (!fallback || !getContextEngineQuarantine(metadata.engineId)) {
+ return engine.info;
+ }
+ return (
+ resolvedFallbackEngine?.info ?? {
+ id: fallback.defaultEngineId,
+ name:
+ fallback.defaultEngineId === "legacy"
+ ? "Legacy Context Engine"
+ : `${fallback.defaultEngineId} Context Engine`,
+ }
+ );
+ }
+
+ const method = Reflect.get(engine, property, engine);
+ if (typeof method !== "function") {
+ return method;
+ }
+ if (!GUARDED_CONTEXT_ENGINE_METHODS.has(property)) {
+ return method.bind(engine);
+ }
+ if (!fallback || !getFallbackEngine) {
+ return (params: Record) => method.call(engine, projectParams(params));
+ }
+
+ const methodName = property as GuardedContextEngineMethodName;
+ return async (methodParams: Record) => {
+ const abortSignal = contextEngineAbortSignal(methodParams);
+ if (abortSignal?.aborted) {
+ const reason = abortSignal.reason;
+ throw reason instanceof Error
+ ? reason
+ : createAbortError(
+ typeof reason === "string" && reason
+ ? reason
+ : "Context engine operation aborted.",
+ );
+ }
+ const invokeFallback = () =>
+ invokeFallbackContextEngineMethod({ getFallbackEngine, methodName, methodParams });
+ if (getContextEngineQuarantine(metadata.engineId)) {
+ // Runtime failures downgrade future guarded calls for this process.
+ return await invokeFallback();
+ }
+
+ try {
+ return await method.call(engine, projectParams(methodParams));
+ } catch (error) {
+ if (isContextEngineAbortRejection(error, abortSignal)) {
+ // Abort is caller intent, not engine instability; never quarantine for it.
+ throw error;
+ }
+ recordContextEngineQuarantine({
+ engineId: metadata.engineId,
+ owner: metadata.owner,
+ operation: methodName,
+ error,
+ defaultEngineId: fallback.defaultEngineId,
+ });
+ if (methodName === "compact" || methodName === "prepareSubagentSpawn") {
+ throw error;
+ }
+ return await invokeFallback().catch(() => {
+ throw error;
+ });
+ }
+ };
+ },
+ },
+ );
+ resolvedEngineMetadata.set(wrapped, metadata);
return wrapped;
}
@@ -163,10 +216,6 @@ function requireContextEngineOwner(owner: string): string {
return normalizedOwner;
}
-function formatContextEngineError(error: unknown): string {
- return error instanceof Error ? error.message : String(error);
-}
-
function recordContextEngineQuarantine(params: {
engineId: string;
owner?: string;
@@ -183,7 +232,7 @@ function recordContextEngineQuarantine(params: {
const quarantine: ContextEngineRuntimeQuarantine = {
engineId: params.engineId,
operation: params.operation,
- reason: formatContextEngineError(params.error),
+ reason: params.error instanceof Error ? params.error.message : String(params.error),
failedAt: new Date(),
...(params.owner ? { owner: params.owner } : {}),
};
@@ -211,23 +260,13 @@ export function listContextEngineQuarantines(): ContextEngineRuntimeQuarantine[]
({ failedAt, ...quarantine }) => ({ ...quarantine, failedAt: new Date(failedAt) }),
);
const seenEngineIds = new Set(quarantines.map((entry) => entry.engineId));
- for (const entry of listPersistedContextEngineQuarantines()) {
- if (seenEngineIds.has(entry.engineId)) {
- continue;
- }
- quarantines.push(entry);
- seenEngineIds.add(entry.engineId);
- }
- return quarantines;
+ return quarantines.concat(
+ listPersistedContextEngineQuarantines().filter(({ engineId }) => !seenEngineIds.has(engineId)),
+ );
}
-function clearContextEngineRuntimeQuarantine(engineId?: string): void {
- const quarantinedEngines = contextEngineRegistryState.quarantinedEngines;
- if (engineId === undefined) {
- quarantinedEngines.clear();
- } else {
- quarantinedEngines.delete(engineId);
- }
+function clearContextEngineRuntimeQuarantine(engineId: string): void {
+ contextEngineRegistryState.quarantinedEngines.delete(engineId);
clearPersistedContextEngineQuarantineForProcess(engineId, process.pid);
}
@@ -324,7 +363,10 @@ export function clearContextEnginesForOwner(owner: string): void {
export function resolveContextEngineOwnerPluginId(
engine: ContextEngine | undefined | null,
): string | undefined {
- const owner = engine && resolveEffectiveContextEngineMetadata(engine)?.owner;
+ const metadata = engine ? resolvedEngineMetadata.get(engine) : undefined;
+ // Quarantined work belongs to its core-owned fallback, never the disabled plugin.
+ const owner =
+ metadata && !getContextEngineQuarantine(metadata.engineId) ? metadata.owner : undefined;
if (!owner?.startsWith("plugin:")) {
return undefined;
}
@@ -332,23 +374,6 @@ export function resolveContextEngineOwnerPluginId(
return pluginId || undefined;
}
-function resolveEffectiveContextEngineMetadata(
- engine: ContextEngine,
-): ResolvedContextEngineMetadata | undefined {
- const quarantineState = RUNTIME_QUARANTINE_PROXY_STATE.get(engine);
- if (quarantineState && getContextEngineQuarantine(quarantineState.engineId)) {
- // After quarantine, metadata follows the resolved fallback so plugin-scoped operations do not
- // keep attributing work to a disabled engine.
- const fallbackEngine = quarantineState.getResolvedFallbackEngine();
- return (
- (fallbackEngine ? RESOLVED_CONTEXT_ENGINE_METADATA.get(fallbackEngine) : undefined) ?? {
- owner: CORE_CONTEXT_ENGINE_OWNER,
- }
- );
- }
- return RESOLVED_CONTEXT_ENGINE_METADATA.get(engine);
-}
-
function describeResolvedContextEngineContractError(
engineId: string,
engine: unknown,
@@ -405,14 +430,10 @@ const CONTEXT_ENGINE_FALLBACK_RESULTS = {
};
function contextEngineAbortSignal(methodParams: unknown): AbortSignal | undefined {
- if (!methodParams || typeof methodParams !== "object") {
- return undefined;
- }
- const signal = (methodParams as { abortSignal?: unknown }).abortSignal;
- if (signal && typeof signal === "object" && "aborted" in signal) {
- return signal as AbortSignal;
- }
- return undefined;
+ const signal = (methodParams as { abortSignal?: unknown } | null | undefined)?.abortSignal;
+ return signal && typeof signal === "object" && "aborted" in signal
+ ? (signal as AbortSignal)
+ : undefined;
}
function isContextEngineAbortRejection(error: unknown, signal: AbortSignal | undefined): boolean {
@@ -450,93 +471,6 @@ async function invokeFallbackContextEngineMethod(params: {
return fallbackResult ? { ...fallbackResult } : undefined;
}
-function wrapContextEngineWithRuntimeQuarantine(params: {
- engine: ContextEngine;
- engineId: string;
- owner: string;
- defaultEngineId: string;
- factoryCtx: ContextEngineFactoryContext;
-}): ContextEngine {
- let fallbackEnginePromise: Promise | undefined;
- let resolvedFallbackEngine: ContextEngine | undefined;
- const getFallbackEngine = () => {
- fallbackEnginePromise ??= resolveDefaultContextEngine(
- params.defaultEngineId,
- params.factoryCtx,
- ).then((engine) => {
- resolvedFallbackEngine = engine;
- return engine;
- });
- return fallbackEnginePromise;
- };
- const fallbackInfo = (): ContextEngineInfo =>
- resolvedFallbackEngine?.info ?? {
- id: params.defaultEngineId,
- name:
- params.defaultEngineId === "legacy"
- ? "Legacy Context Engine"
- : `${params.defaultEngineId} Context Engine`,
- };
- const isQuarantined = () => Boolean(getContextEngineQuarantine(params.engineId));
-
- const proxy = new Proxy(params.engine, {
- get(target, property, receiver) {
- if (property === "info" && isQuarantined()) {
- return fallbackInfo();
- }
- const value = Reflect.get(target, property, receiver);
- if (typeof value !== "function" || !GUARDED_CONTEXT_ENGINE_METHODS.has(property)) {
- return typeof value === "function" ? value.bind(target) : value;
- }
-
- const methodName = property as GuardedContextEngineMethodName;
- return async (methodParams: unknown) => {
- const abortSignal = contextEngineAbortSignal(methodParams);
- if (abortSignal?.aborted) {
- const reason = abortSignal.reason;
- throw reason instanceof Error
- ? reason
- : createAbortError(
- typeof reason === "string" && reason ? reason : "Context engine operation aborted.",
- );
- }
- const invokeFallback = () =>
- invokeFallbackContextEngineMethod({ getFallbackEngine, methodName, methodParams });
- if (isQuarantined()) {
- // Runtime failures downgrade future guarded calls for this process.
- return await invokeFallback();
- }
-
- try {
- return await (value as (methodParams: unknown) => unknown).call(target, methodParams);
- } catch (error) {
- if (isContextEngineAbortRejection(error, abortSignal)) {
- // Abort is caller intent, not engine instability; never quarantine for it.
- throw error;
- }
- recordContextEngineQuarantine({
- engineId: params.engineId,
- owner: params.owner,
- operation: methodName,
- error,
- defaultEngineId: params.defaultEngineId,
- });
- if (methodName === "compact" || methodName === "prepareSubagentSpawn") {
- throw error;
- }
- return await invokeFallback().catch(() => {
- throw error;
- });
- }
- };
- },
- });
- RUNTIME_QUARANTINE_PROXY_STATE.set(proxy, {
- engineId: params.engineId,
- getResolvedFallbackEngine: () => resolvedFallbackEngine,
- });
- return proxy;
-}
// ---------------------------------------------------------------------------
// Resolution
// ---------------------------------------------------------------------------
diff --git a/src/context-engine/runtime-settings.ts b/src/context-engine/runtime-settings.ts
index 715ab337c153..5c2ad151f4b6 100644
--- a/src/context-engine/runtime-settings.ts
+++ b/src/context-engine/runtime-settings.ts
@@ -1,3 +1,4 @@
+import { normalizeNullableString } from "@openclaw/normalization-core/string-coerce";
import type { ContextEngineHostSupport } from "./host-compat.js";
import type {
ContextEngineRuntimeReasonCode,
@@ -7,7 +8,6 @@ import type {
} from "./types.js";
type OptionalString = string | null | undefined;
-type OptionalReason = string | null | undefined;
const RUNTIME_REASON_CODES = new Set([
"provider_timeout",
@@ -17,20 +17,19 @@ const RUNTIME_REASON_CODES = new Set([
"runtime_unavailable",
"unknown",
]);
-
-function normalizeNullableString(value: OptionalString): string | null {
- if (typeof value !== "string") {
- return null;
- }
- const trimmed = value.trim();
- return trimmed ? trimmed : null;
-}
+const RUNTIME_REASON_PATTERNS: Array<[ContextEngineRuntimeReasonCode, RegExp]> = [
+ ["provider_timeout", /timeout/iu],
+ ["rate_limited", /rate|limit|429/iu],
+ ["context_overflow", /overflow|context|pressure/iu],
+ ["runtime_unavailable", /runtime/iu],
+ ["provider_unavailable", /provider|primary|unavailable/iu],
+];
function normalizeNullableNumber(value: number | null | undefined): number | null {
return typeof value === "number" && Number.isFinite(value) ? value : null;
}
-function normalizeReasonCode(value: OptionalReason): ContextEngineRuntimeReasonCode | null {
+function normalizeReasonCode(value: OptionalString): ContextEngineRuntimeReasonCode | null {
const normalized = normalizeNullableString(value);
if (!normalized) {
return null;
@@ -39,23 +38,7 @@ function normalizeReasonCode(value: OptionalReason): ContextEngineRuntimeReasonC
return normalized as ContextEngineRuntimeReasonCode;
}
- const lower = normalized.toLowerCase();
- if (lower.includes("timeout")) {
- return "provider_timeout";
- }
- if (lower.includes("rate") || lower.includes("limit") || lower.includes("429")) {
- return "rate_limited";
- }
- if (lower.includes("overflow") || lower.includes("context") || lower.includes("pressure")) {
- return "context_overflow";
- }
- if (lower.includes("runtime")) {
- return "runtime_unavailable";
- }
- if (lower.includes("provider") || lower.includes("primary") || lower.includes("unavailable")) {
- return "provider_unavailable";
- }
- return "unknown";
+ return RUNTIME_REASON_PATTERNS.find(([, pattern]) => pattern.test(normalized))?.[0] ?? "unknown";
}
export function buildContextEngineRuntimeSettings(params: {
@@ -68,8 +51,8 @@ export function buildContextEngineRuntimeSettings(params: {
modelFamily?: OptionalString;
selectedContextEngineId?: OptionalString;
contextEngineSelectionSource?: ContextEngineSelectionSource;
- fallbackReason?: OptionalReason;
- degradedReason?: OptionalReason;
+ fallbackReason?: OptionalString;
+ degradedReason?: OptionalString;
promptTokenBudget?: number | null;
maxOutputTokens?: number | null;
contextEngineHost: ContextEngineHostSupport;