ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-26 20:35:39 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(compaction): summarize the prepared window and always restore audited identifiers (#129423)

Browse Source 
* fix(compaction): restore omitted source identifiers

* fix: log ChatGPT stream termination reasons

* fix(compaction): summarize the prepared window and always restore audited identifiers

Two safeguard defects made auto-compaction fail closed on tool-heavy sessions:

1. When the prepared window held only tool calls/results, the safeguard replaced it
   with sessionManager.getBranch() — the whole branch from root. Since #112988 made
   /new a reset entry inside the same transcript, that branch spans every reset and
   prior compaction (3,565 messages / 5.2 MB on the reporting session), so one
   compaction became dozens of sequential model calls until the 900s watchdog fired
   ("Compaction timed out"). The #78390 reason for that fallback (preparation
   omitting custom messages) is handled by core projectSessionEntryMessage now.
   Summarize only the prepared window; decide "nothing to summarize" from the
   boundary-scoped context (buildSessionContext), never the raw branch. Delete the
   fallback and its sessions_send replay sanitizers, which only existed for it.

2. #122824 moved the identifier oracle to the messages before the recent-turns split,
   so audited identifiers can live only in preserved turns that are capped at 600
   chars/message and 8K per section. The deterministic restore added in #128968 ran
   only when the artifact exceeded 16K chars; an under-budget body that omits an
   identifier still failed the audit (missing_identifiers -> guard_blocked). Run the
   retention plan whenever a strict identifier is missing from the body, and only
   emit the truncation marker when body text was actually trimmed.

* fix(compaction): fund audit sections first and cap the identifier list

The retention plan protected the model-written "## Pending user asks" and
"## Exact identifiers" sections verbatim. Each re-distill fed the previous
summary back, so the identifier list only grew: on the reporting session it was
13,568 of a 13,953-char summary, every other section an empty heading, and the
next compaction cancelled with "required quality facts exceed finalized artifact
budget" because the protected content alone passed the 16K cap.

Protect only what the audit requires (headings, bounded latest-ask context, the
audited identifiers). Fund the two audit-bearing sections first so an oversized
earlier section cannot starve them, but cap each at 25% of the content budget;
the remainder is split across decisions, TODOs, and constraints.

* fix(ai): log only structured ChatGPT stream termination facts

projectProviderError keeps provider message/body text after credential
redaction, so logging errorMessage could persist prompt- or response-derived
content. Log timing, stop classification, the error class name, and the
projection's structured errorCode/errorType instead, and cover a provider
failure whose message must not reach the transport log.

* fix(ai): classify ChatGPT stream termination locally

Provider error code/type fields are provider-controlled text, so the transport
warning logs a fixed local failure category (timeout, caller-abort,
provider-failure, transport) plus timing instead. The failed-response test
feeds hostile code/type/message values and asserts none reach the log.

* fix(compaction): recover omitted conversation from the boundary-scoped range

Replace the deleted whole-branch fallback with the bounded safety net it was
meant to be: when the prepared window holds no real conversation, project the
branch entries the preparation covered — after the last reset/compaction
boundary, before firstKeptEntryId — through the core context builder and
summarize that range if it contains conversation the preparation dropped. It
can never reach behind a boundary, so it cannot re-read the whole session.

* fix(compaction): make the audit-section cap hard even when the summary fits

Two holes let the identifier snowball survive the 25% share cap: a summary
that still fit the 16K budget skipped the retention plan entirely, so a
13.9K identifier dump was fed unchanged into the next re-distill; and the
surplus loop handed unused budget back to the protected sections whenever
the optional sections were empty — exactly the degraded shape in production.
Rebuild whenever an audit section exceeds its cap, and return surplus to the
optional sections only.

Also classify a direct non-OK HTTP reply (CodexApiError) as a provider
failure in the stream-termination diagnostic, with a direct-400 test that
keeps provider text out of the log.

---------

Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com>
This commit is contained in:
Vito Cappello
2026-08-26 09:55:53 -04:00
committed by GitHub
parent f1c60c7c4b
commit 6124d47e04
6 changed files with 646 additions and 659 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/assertion-safety-baseline.txt
+1 -1
View File
@@ -1667,7 +1667,7 @@ src/agents/agent-bundle-mcp-runtime-shared.ts 2
src/agents/agent-bundle-mcp-runtime.ts 9
src/agents/agent-command-restart-recovery.ts 3
src/agents/agent-command.ts 1
src/agents/agent-hooks/compaction-safeguard.ts 19
src/agents/agent-hooks/compaction-safeguard.ts 14
src/agents/agent-model-discovery.ts 5
src/agents/agent-project-settings-snapshot.ts 5
src/agents/agent-run-terminal-delivery.ts 1
packages/ai/src/providers/openai-chatgpt-responses-streaming.test.ts
+98 -1
View File
@@ -262,6 +262,80 @@ describe("OpenAI ChatGPT Responses inference streaming", () => {
expect(fetchMock).not.toHaveBeenCalled();
});
it("logs the caller abort reason with bounded ChatGPT transport metadata", async () => {
const logWarn = vi.fn();
configureAiTransportHost({ logWarn });
const controller = new AbortController();
class AbortedWebSocket extends EventTarget {
constructor() {
super();
queueMicrotask(() => this.dispatchEvent(new Event("open")));
}
send(): void {
controller.abort(new Error("Compaction timed out"));
}
close(): void {}
}
vi.stubGlobal("WebSocket", AbortedWebSocket);
const fetchMock = vi.fn();
vi.stubGlobal("fetch", fetchMock);
const result = await streamOpenAICodexResponses(model, context, {
apiKey: createJwt({
"https://api.openai.com/auth": { chatgpt_account_id: "acct-1" },
}),
signal: controller.signal,
}).result();
expect(result).toMatchObject({ stopReason: "aborted", errorMessage: "Request was aborted" });
expect(logWarn).toHaveBeenCalledWith(
"openai-transport",
"ChatGPT Responses stream terminated",
{
api: "openai-chatgpt-responses",
elapsedMs: expect.any(Number),
failureKind: "caller-abort",
model: "gpt-5.6-luna",
provider: "openai",
stopReason: "aborted",
transport: "auto",
},
);
expect(fetchMock).not.toHaveBeenCalled();
});
it("classifies a direct HTTP rejection as a provider failure without logging its text", async () => {
const logWarn = vi.fn();
configureAiTransportHost({ logWarn });
const fetchMock = vi.fn<typeof fetch>().mockResolvedValueOnce(
new Response(
JSON.stringify({
error: { message: "hostile prompt echo in a 400 body", code: "invalid_request" },
}),
{ status: 400, statusText: "Bad Request" },
),
);
vi.stubGlobal("fetch", fetchMock);
const result = await streamOpenAICodexResponses(model, context, {
apiKey: createJwt({
"https://api.openai.com/auth": { chatgpt_account_id: "acct-1" },
}),
transport: "sse",
}).result();
expect(result.stopReason).toBe("error");
expect(logWarn).toHaveBeenCalledTimes(1);
const logged = logWarn.mock.calls[0]?.[2];
expect(logged).toMatchObject({ stopReason: "error", failureKind: "provider-failure" });
const serialized = JSON.stringify(logged);
for (const hostile of ["hostile", "invalid_request", "400 body"]) {
expect(serialized).not.toContain(hostile);
}
});
it.each(["sse", "websocket"] as const)(
"preserves failed response identity and provider error details over %s",
async (transport) => {
@@ -270,9 +344,15 @@ describe("OpenAI ChatGPT Responses inference streaming", () => {
response: {
id: "resp_failed",
status: "failed",
error: { code: "invalid_prompt", message: "rejected" },
error: {
code: "invalid_prompt",
type: "hostile type: user prompt echoed here",
message: "rejected",
},
},
};
const logWarn = vi.fn();
configureAiTransportHost({ logWarn });
const fetchMock = vi.fn();
vi.stubGlobal("fetch", fetchMock);
@@ -323,6 +403,23 @@ describe("OpenAI ChatGPT Responses inference streaming", () => {
errorMessage: "invalid_prompt: rejected",
},
});
// Provider message text reaches the stream consumer only; the transport
// log keeps timing and classification and never the message body.
expect(logWarn).toHaveBeenCalledTimes(1);
const logged = logWarn.mock.calls[0]?.[2];
expect(logged).toEqual({
api: "openai-chatgpt-responses",
elapsedMs: expect.any(Number),
failureKind: "provider-failure",
model: "gpt-5.6-luna",
provider: "openai",
stopReason: "error",
transport,
});
const serialized = JSON.stringify(logged);
for (const hostile of ["rejected", "invalid_prompt", "hostile type", "echoed"]) {
expect(serialized).not.toContain(hostile);
}
if (transport === "websocket") {
expect(fetchMock).not.toHaveBeenCalled();
}
packages/ai/src/providers/openai-chatgpt-responses.ts
+38 -2
View File
@@ -243,6 +243,26 @@ function isRequestTimeoutError(
);
}
type StreamFailureKind = "timeout" | "caller-abort" | "provider-failure" | "transport";
/** Local-only failure category for transport diagnostics; never provider text. */
function classifyStreamFailure(
error: unknown,
signal: AbortSignal | undefined,
requestTimedOut: boolean,
): StreamFailureKind {
if (requestTimedOut) {
return "timeout";
}
if (signal?.aborted) {
return "caller-abort";
}
// CodexApiError carries a non-OK HTTP reply; both are provider decisions, not transport faults.
return error instanceof ResponsesStreamFailure || error instanceof CodexApiError
? "provider-failure"
: "transport";
}
function formatRequestTimeoutError(timeoutMs: number, cause: unknown): Error {
return new Error(`Request timed out after ${timeoutMs}ms`, {
cause: cause instanceof Error ? cause : undefined,
@@ -288,6 +308,7 @@ export const streamOpenAICodexResponses: StreamFunction<
const stream = new AssistantMessageEventStream();
void (async () => {
const startedAt = Date.now();
let requestTimeoutMs: number | undefined;
let requestTimeoutSignal: AbortSignal | undefined;
let activeSignal: AbortSignal | undefined;
@@ -621,9 +642,11 @@ export const streamOpenAICodexResponses: StreamFunction<
});
stream.end();
} catch (error) {
const normalizedError =
const requestTimedOut =
isRequestTimeoutError(error, options?.signal, requestTimeoutSignal, requestTimeoutMs) &&
requestTimeoutMs !== undefined
requestTimeoutMs !== undefined;
const normalizedError =
requestTimedOut && requestTimeoutMs !== undefined
? formatRequestTimeoutError(requestTimeoutMs, error)
: error;
for (const block of output.content) {
@@ -631,6 +654,19 @@ export const streamOpenAICodexResponses: StreamFunction<
delete (block as { partialJson?: string }).partialJson;
}
const terminal = projectProviderError(normalizedError, options?.signal);
// Log only locally-derived facts: timing and a fixed failure category. No
// projected provider field (message, body, code, type, name) is logged —
// all of them are provider-controlled text that can carry prompt- or
// response-derived content.
getAiTransportHost().logWarn("openai-transport", "ChatGPT Responses stream terminated", {
provider: model.provider,
api: model.api,
model: model.id,
transport: options?.transport || "auto",
elapsedMs: Math.max(0, Date.now() - startedAt),
stopReason: terminal.stopReason,
failureKind: classifyStreamFailure(error, options?.signal, requestTimedOut),
});
Object.assign(output, terminal);
stream.push({ type: "error", reason: terminal.stopReason, error: output });
stream.end();
src/agents/agent-hooks/compaction-safeguard-quality.ts
+110 -52
View File
@@ -20,6 +20,9 @@ const REQUIRED_SUMMARY_SECTIONS = [
"## Exact identifiers",
] as const;
const QUALITY_PROTECTED_SECTION_START = 3;
const PENDING_ASK_SECTION_INDEX = 3;
const EXACT_IDENTIFIERS_SECTION_INDEX = 4;
const MAX_PROTECTED_SECTION_CONTENT_SHARE = 0.25;
const STRICT_EXACT_IDENTIFIERS_INSTRUCTION =
"For ## Exact identifiers, preserve literal values exactly as seen (IDs, URLs, file paths, ports, hashes, dates, times).";
const POLICY_OFF_EXACT_IDENTIFIERS_INSTRUCTION =
@@ -97,7 +100,13 @@ function hasRequiredSummarySections(summary: string): boolean {
type SummaryQualityRetentionPlan = {
minimumChars: number;
render: (maxChars: number) => string | null;
/**
* True when render() must rebuild even a body that fits: a strict source
* identifier is missing, or an audit-bearing section exceeds its share cap.
*/
needsRebuild: (maxChars: number) => boolean;
/** Null when even the protected facts cannot fit `maxChars`. */
render: (maxChars: number) => { text: string; trimmed: boolean } | null;
};
function parseRequiredSummarySectionContents(summary: string): string[] | null {
@@ -120,7 +129,14 @@ function parseRequiredSummarySectionContents(summary: string): string[] | null {
return contents.map((lines) => lines.join("\n").trim());
}
/** Plan truncation that keeps audit-required headings, pending asks, and exact identifiers. */
/**
* Plan truncation that keeps the audit facts and lets everything else shrink.
* Only the headings, the bounded latest-ask context, and the audited source
* identifiers are untrimmable. Model-written section text — including the
* "## Exact identifiers" list — is optional content; protecting it verbatim let
* a re-distilled identifier dump grow past the whole artifact budget while the
* real sections were starved to empty headings.
*/
export function createSummaryQualityRetentionPlan(
summary: string,
truncatedMarker: string,
@@ -138,81 +154,123 @@ export function createSummaryQualityRetentionPlan(
}
const enforceIdentifiers = (params.identifierPolicy ?? "strict") === "strict";
const auditSummary = params.auditSummary ?? summary;
if (
enforceIdentifiers &&
params.identifiers.some((identifier) => !summaryIncludesIdentifier(auditSummary, identifier))
) {
return null;
}
if (!hasAskOverlap(auditSummary, params.latestAsk)) {
return null;
}
const pendingAsk = contents[QUALITY_PROTECTED_SECTION_START] ?? "";
const requiredAskContext = params.requiredAskContext?.trim() ?? "";
const exactIdentifiers = contents[QUALITY_PROTECTED_SECTION_START + 1] ?? "";
const missingIdentifiers = enforceIdentifiers
? params.identifiers.filter(
(identifier) => !summaryIncludesIdentifier(exactIdentifiers, identifier),
)
: [];
const protectedContents = [
[
pendingAsk,
requiredAskContext && !pendingAsk.includes(requiredAskContext) ? requiredAskContext : "",
]
.filter(Boolean)
.join("\n"),
[exactIdentifiers, ...missingIdentifiers].filter(Boolean).join("\n"),
];
const auditedIdentifiers = enforceIdentifiers ? params.identifiers : [];
const marker = truncatedMarker.trim();
const protectedBlocks = REQUIRED_SUMMARY_SECTIONS.slice(QUALITY_PROTECTED_SECTION_START).map(
(heading, index) => {
const content = protectedContents[index];
// Protected tails render after each section's optional content so the audit
// facts survive regardless of how much model text the budget keeps.
const protectedTails = REQUIRED_SUMMARY_SECTIONS.map((_, index) =>
index === PENDING_ASK_SECTION_INDEX
? requiredAskContext
: index === EXACT_IDENTIFIERS_SECTION_INDEX
? auditedIdentifiers.join("\n")
: "",
);
const bodyHasIdentifiers = auditedIdentifiers.every((identifier) =>
summaryIncludesIdentifier(summary, identifier),
);
const renderSections = (sectionContents: string[]) =>
REQUIRED_SUMMARY_SECTIONS.map((heading, index) => {
const content = sectionContents[index];
return content ? `${heading}\n${content}` : heading;
},
});
const joinSectionContent = (index: number, optional: string) => {
const tail = protectedTails[index] ?? "";
if (!tail) {
return optional;
}
if (index === PENDING_ASK_SECTION_INDEX && optional.includes(tail)) {
return optional;
}
if (index === EXACT_IDENTIFIERS_SECTION_INDEX) {
const missing = auditedIdentifiers.filter(
(identifier) => !summaryIncludesIdentifier(optional, identifier),
);
return [optional, ...missing].filter(Boolean).join("\n");
}
return [optional, tail].filter(Boolean).join("\n");
};
// Reserve every heading/content/tail separator up front so trimmed optional
// text can never push the rendered artifact past `maxChars`.
const minimumBlocks = REQUIRED_SUMMARY_SECTIONS.map(
(heading, index) => `${heading}\n\n${protectedTails[index] ?? ""}`,
);
const optionalHeadings = REQUIRED_SUMMARY_SECTIONS.slice(0, QUALITY_PROTECTED_SECTION_START);
const optionalContents = contents.slice(0, QUALITY_PROTECTED_SECTION_START);
const optionalScaffolds = optionalHeadings.map((heading, index) =>
optionalContents[index] ? `${heading}\n` : heading,
);
const minimumSummary = [...optionalScaffolds, marker, ...protectedBlocks].join("\n\n");
const minimumSummary = [
...minimumBlocks.slice(0, QUALITY_PROTECTED_SECTION_START),
marker,
...minimumBlocks.slice(QUALITY_PROTECTED_SECTION_START),
].join("\n\n");
// Audit-bearing sections (pending asks, exact identifiers) are funded first so
// a runaway earlier section cannot starve them, but each is hard-capped: an
// uncapped identifier list re-distills into the whole budget — even while the
// artifact still fits — and leaves every other section as a bare heading.
const protectedCapFor = (maxChars: number) =>
Math.floor(Math.max(0, maxChars - minimumSummary.length) * MAX_PROTECTED_SECTION_CONTENT_SHARE);
const protectedWithinCap = (maxChars: number) =>
contents
.slice(QUALITY_PROTECTED_SECTION_START)
.every((content) => content.length <= protectedCapFor(maxChars));
return {
minimumChars: minimumSummary.length,
needsRebuild: (maxChars) => !bodyHasIdentifiers || !protectedWithinCap(maxChars),
render(maxChars) {
const bodyHasRequiredAskContext = !requiredAskContext || summary.includes(requiredAskContext);
const bodyHasIdentifiers =
!enforceIdentifiers ||
params.identifiers.every((identifier) => summaryIncludesIdentifier(summary, identifier));
if (summary.length <= maxChars && bodyHasRequiredAskContext && bodyHasIdentifiers) {
return summary;
if (
summary.length <= maxChars &&
bodyHasRequiredAskContext &&
bodyHasIdentifiers &&
protectedWithinCap(maxChars)
) {
return { text: summary, trimmed: false };
}
if (maxChars < minimumSummary.length) {
return null;
}
const contentBudget = maxChars - minimumSummary.length;
const totalContentChars = optionalContents.reduce(
(total, content) => total + content.length,
const protectedCap = protectedCapFor(maxChars);
const allocations = contents.map((content, index) =>
index >= QUALITY_PROTECTED_SECTION_START ? Math.min(content.length, protectedCap) : 0,
);
const optionalBudget = Math.max(
0,
contentBudget - allocations.reduce((total, chars) => total + chars, 0),
);
const allocations = optionalContents.map((content) =>
totalContentChars > 0
? Math.floor((contentBudget * content.length) / totalContentChars)
: 0,
);
let remainder = contentBudget - allocations.reduce((total, chars) => total + chars, 0);
const optionalContents = contents.slice(0, QUALITY_PROTECTED_SECTION_START);
const optionalTotal = optionalContents.reduce((total, content) => total + content.length, 0);
for (const [index, content] of optionalContents.entries()) {
allocations[index] =
optionalTotal > 0 ? Math.floor((optionalBudget * content.length) / optionalTotal) : 0;
}
// Surplus returns to the optional sections only; the protected caps stay
// hard so short decisions cannot hand the budget back to the identifier dump.
let remainder =
optionalBudget -
allocations
.slice(0, QUALITY_PROTECTED_SECTION_START)
.reduce((total, chars) => total + chars, 0);
for (const [index, content] of optionalContents.entries()) {
const allocation = allocations[index] ?? 0;
const extra = Math.min(remainder, Math.max(0, content.length - allocation));
allocations[index] = allocation + extra;
remainder -= extra;
}
const optionalBlocks = optionalHeadings.map((heading, index) => {
const content = truncateUtf16Safe(optionalContents[index] ?? "", allocations[index] ?? 0);
return content ? `${heading}\n${content}` : heading;
});
return [...optionalBlocks, marker, ...protectedBlocks].join("\n\n");
const trimmed = contents.some((content, index) => content.length > (allocations[index] ?? 0));
const sectionContents = contents.map((content, index) =>
joinSectionContent(index, truncateUtf16Safe(content, allocations[index] ?? 0)),
);
const blocks = renderSections(sectionContents);
return {
text: [
...blocks.slice(0, QUALITY_PROTECTED_SECTION_START),
...(trimmed ? [marker] : []),
...blocks.slice(QUALITY_PROTECTED_SECTION_START),
].join("\n\n"),
trimmed,
};
},
};
}
src/agents/agent-hooks/compaction-safeguard.test.ts
+332 -426
View File
@@ -2180,6 +2180,234 @@ describe("compaction-safeguard recent-turn preservation", () => {
expect(consumeCompactionSafeguardCancelReason(sessionManager)).toBeNull();
});
it("restores source identifiers omitted by an oversized generated summary", async () => {
mockSummarizeInStages.mockReset();
const latestAsk = "preserve the pending deployment status";
const identifier = "/tmp/source-only-compaction-id.log";
const generatedSummary = [
"## Decisions",
"x".repeat(MAX_COMPACTION_SUMMARY_CHARS),
"## Open TODOs",
"None.",
"## Constraints/Rules",
"Preserve exact context.",
"## Pending user asks",
latestAsk,
"## Exact identifiers",
"None.",
].join("\n");
mockSummarizeInStages.mockResolvedValue(summaryResult(generatedSummary));
const sessionManager = stubSessionManager();
setCompactionSafeguardRuntime(sessionManager, {
model: createAnthropicModelFixture(),
recentTurnsPreserve: 0,
qualityGuardEnabled: true,
qualityGuardMaxRetries: 1,
});
const event = createCompactionEvent({
messageText: `${latestAsk} ${identifier}`,
tokensBefore: 1_500,
});
(
event.preparation as { settings?: { reserveTokens: number }; isSplitTurn?: boolean }
).settings = { reserveTokens: 4_000 };
(event.preparation as { isSplitTurn?: boolean }).isSplitTurn = false;
const { result } = await runCompactionScenario({ sessionManager, event, apiKey: "test-key" });
const summary = expectCompactionResult(result).summary;
expect(summary.length).toBeLessThanOrEqual(MAX_COMPACTION_SUMMARY_CHARS);
expect(summary).toContain(`## Exact identifiers\nNone.\n${identifier}`);
expect(mockSummarizeInStages).toHaveBeenCalledTimes(1);
expect(consumeCompactionSafeguardCancelReason(sessionManager)).toBeNull();
});
it("keeps real sections when a re-distilled identifier list outgrows the budget", async () => {
mockSummarizeInStages.mockReset();
const latestAsk = "preserve the pending deployment status";
const identifier = "/tmp/source-only-compaction-id.log";
// A model that hoards every identifier it has ever seen: the list alone is
// larger than the whole artifact budget while the real sections stay small.
const hoardedIdentifiers = Array.from(
{ length: 400 },
(_, index) => `- /home/vac/clawd/tmp/session-artifacts/run-${index}/output.log`,
).join("\n");
const generatedSummary = [
"## Decisions",
"Deployment stays paused until the backup is verified.",
"## Open TODOs",
"Verify the backup.",
"## Constraints/Rules",
"Preserve exact context.",
"## Pending user asks",
latestAsk,
"## Exact identifiers",
hoardedIdentifiers,
].join("\n");
expect(generatedSummary.length).toBeGreaterThan(MAX_COMPACTION_SUMMARY_CHARS);
mockSummarizeInStages.mockResolvedValue(summaryResult(generatedSummary));
const sessionManager = stubSessionManager();
setCompactionSafeguardRuntime(sessionManager, {
model: createAnthropicModelFixture(),
recentTurnsPreserve: 0,
qualityGuardEnabled: true,
qualityGuardMaxRetries: 1,
});
const event = createCompactionEvent({
messageText: `${latestAsk} ${identifier}`,
tokensBefore: 1_500,
});
(event.preparation as { settings?: { reserveTokens: number } }).settings = {
reserveTokens: 4_000,
};
const { result } = await runCompactionScenario({ sessionManager, event, apiKey: "test-key" });
const summary = expectCompactionResult(result).summary;
expect(summary.length).toBeLessThanOrEqual(MAX_COMPACTION_SUMMARY_CHARS);
expect(summary).toContain(
"## Decisions\nDeployment stays paused until the backup is verified.",
);
expect(summary).toContain("## Open TODOs\nVerify the backup.");
expect(summary).toContain(`## Pending user asks\n${latestAsk}`);
expect(summary).toContain(identifier);
expect(summary).toContain("run-0/output.log");
expect(summary).not.toContain("run-399/output.log");
const identifiersSection = summary.slice(summary.indexOf("## Exact identifiers"));
expect(identifiersSection.length).toBeLessThanOrEqual(
MAX_COMPACTION_SUMMARY_CHARS * 0.25 + 200,
);
expect(consumeCompactionSafeguardCancelReason(sessionManager)).toBeNull();
});
it("keeps surplus budget out of the protected sections when trimming", () => {
const identifier = "/tmp/surplus-compaction-id.log";
const latestAsk = "preserve the pending deployment status";
// The re-distilled shape seen in production: every optional section is an
// empty heading, so all surplus would otherwise flow to the identifier list.
const body = [
"## Decisions",
"## Open TODOs",
"## Constraints/Rules",
"## Pending user asks",
latestAsk,
"## Exact identifiers",
[identifier, ...Array.from({ length: 300 }, (_, i) => `- /tmp/run-${i}/out.log`)].join("\n"),
].join("\n");
const maxChars = 4_000;
expect(body.length).toBeGreaterThan(maxChars);
const finalized = budgetCompactionSummary(body, "", maxChars, {
identifiers: [identifier],
latestAsk,
identifierPolicy: "strict",
});
const structural = (finalized as { structuralSummary: string }).structuralSummary;
expect(structural.length).toBeLessThanOrEqual(maxChars);
expect(structural).toContain("## Decisions");
expect(structural).toContain(identifier);
const identifiersSection = structural.slice(structural.indexOf("## Exact identifiers"));
expect(identifiersSection.length).toBeLessThanOrEqual(maxChars * 0.25 + 100);
});
it("caps an identifier list that outgrew its share while the summary still fits", async () => {
mockSummarizeInStages.mockReset();
const latestAsk = "preserve the pending deployment status";
const identifier = "/tmp/source-only-compaction-id.log";
const hoardedIdentifiers = Array.from(
{ length: 200 },
(_, index) => `- /home/vac/clawd/tmp/session-artifacts/run-${index}/output.log`,
).join("\n");
const generatedSummary = [
"## Decisions",
"Deployment stays paused until the backup is verified.",
"## Open TODOs",
"Verify the backup.",
"## Constraints/Rules",
"Preserve exact context.",
"## Pending user asks",
latestAsk,
"## Exact identifiers",
`${identifier}\n${hoardedIdentifiers}`,
].join("\n");
expect(generatedSummary.length).toBeLessThan(MAX_COMPACTION_SUMMARY_CHARS);
mockSummarizeInStages.mockResolvedValue(summaryResult(generatedSummary));
const sessionManager = stubSessionManager();
setCompactionSafeguardRuntime(sessionManager, {
model: createAnthropicModelFixture(),
recentTurnsPreserve: 0,
qualityGuardEnabled: true,
qualityGuardMaxRetries: 1,
});
const event = createCompactionEvent({
messageText: `${latestAsk} ${identifier}`,
tokensBefore: 1_500,
});
(event.preparation as { settings?: { reserveTokens: number } }).settings = {
reserveTokens: 4_000,
};
const { result } = await runCompactionScenario({ sessionManager, event, apiKey: "test-key" });
const summary = expectCompactionResult(result).summary;
expect(summary).toContain(
"## Decisions\nDeployment stays paused until the backup is verified.",
);
expect(summary).toContain(identifier);
expect(summary).not.toContain("run-199/output.log");
const identifiersSection = summary.slice(summary.indexOf("## Exact identifiers"));
expect(identifiersSection.length).toBeLessThanOrEqual(
MAX_COMPACTION_SUMMARY_CHARS * 0.25 + 200,
);
expect(consumeCompactionSafeguardCancelReason(sessionManager)).toBeNull();
});
it("restores source identifiers omitted by a generated summary that fits the budget", async () => {
mockSummarizeInStages.mockReset();
const latestAsk = "preserve the pending deployment status";
const identifier = "/tmp/source-only-compaction-id.log";
const generatedSummary = [
"## Decisions",
"Deployment stays paused.",
"## Open TODOs",
"None.",
"## Constraints/Rules",
"Preserve exact context.",
"## Pending user asks",
latestAsk,
"## Exact identifiers",
"None.",
].join("\n");
mockSummarizeInStages.mockResolvedValue(summaryResult(generatedSummary));
const sessionManager = stubSessionManager();
setCompactionSafeguardRuntime(sessionManager, {
model: createAnthropicModelFixture(),
recentTurnsPreserve: 0,
qualityGuardEnabled: true,
qualityGuardMaxRetries: 1,
});
const event = createCompactionEvent({
messageText: `${latestAsk} ${identifier}`,
tokensBefore: 1_500,
});
(event.preparation as { settings?: { reserveTokens: number } }).settings = {
reserveTokens: 4_000,
};
const { result } = await runCompactionScenario({ sessionManager, event, apiKey: "test-key" });
const summary = expectCompactionResult(result).summary;
expect(summary).toContain(`## Exact identifiers\nNone.\n${identifier}`);
expect(summary).not.toContain(SUMMARY_TRUNCATED_MARKER.trim());
expect(mockSummarizeInStages).toHaveBeenCalledTimes(1);
expect(consumeCompactionSafeguardCancelReason(sessionManager)).toBeNull();
});
it("fails closed when audit-required tail sections cannot fit the artifact cap", async () => {
mockSummarizeInStages.mockReset();
const latestAsk = "preserve the pending deployment status";
@@ -3586,61 +3814,80 @@ describe("compaction-safeguard double-compaction guard", () => {
expect(getApiKeyAndHeadersMock).toHaveBeenCalledWith(model);
});
it("falls back to visible custom session branch entries before writing an empty boundary", async () => {
it("summarizes only the prepared tool-only window when the context anchors it", async () => {
mockSummarizeInStages.mockReset();
mockSummarizeInStages.mockResolvedValue(summaryResult("branch summary"));
mockSummarizeInStages.mockResolvedValue(summaryResult("tool window summary"));
const now = Date.now();
// History behind a reset and a compaction boundary: the old fallback re-read
// all of it through getBranch() and summarized the whole session again.
const sessionManager = {
...stubSessionManager(),
getBranch: () => [
{
type: "custom_message",
id: "custom-1",
type: "message",
id: "old-user",
parentId: null,
timestamp: new Date(now).toISOString(),
customType: "cron-request",
content: "prepare the daily report",
display: true,
message: { role: "user", content: "old request behind reset", timestamp: now },
},
{
type: "message",
id: "assistant-1",
parentId: "custom-1",
id: "old-assistant",
parentId: "old-user",
timestamp: new Date(now + 1).toISOString(),
message: {
role: "assistant",
content: [{ type: "toolCall", id: "call-1", name: "read", arguments: {} }],
content: [{ type: "text", text: "old reply behind reset" }],
timestamp: now + 1,
},
},
{
type: "message",
id: "tool-1",
parentId: "assistant-1",
type: "reset",
id: "reset-1",
parentId: "old-assistant",
timestamp: new Date(now + 2).toISOString(),
message: {
role: "toolResult",
toolCallId: "call-1",
toolName: "read",
content: [{ type: "text", text: "report source data" }],
timestamp: now + 2,
},
reason: "new",
firstKeptEntryId: "old-assistant",
},
{
type: "compaction",
id: "compaction-1",
parentId: "reset-1",
timestamp: new Date(now + 3).toISOString(),
summary: "## Decisions\nUser asked for the deploy status.",
firstKeptEntryId: "compaction-1",
tokensBefore: 90_000,
fromHook: true,
},
],
} as ExtensionContext["sessionManager"];
const model = createAnthropicModelFixture();
setCompactionSafeguardRuntime(sessionManager, { model, recentTurnsPreserve: 0 });
const toolOnlyWindow = [
{
role: "assistant",
content: [{ type: "toolCall", id: "call-1", name: "exec", arguments: {} }],
timestamp: now + 4,
},
{
role: "toolResult",
toolCallId: "call-1",
toolName: "exec",
content: [{ type: "text", text: "deploy green" }],
timestamp: now + 5,
},
] as AgentMessage[];
const mockEvent = {
preparation: {
messagesToSummarize: [] as AgentMessage[],
messagesToSummarize: toolOnlyWindow,
turnPrefixMessages: [] as AgentMessage[],
firstKeptEntryId: "entry-5",
firstKeptEntryId: "entry-6",
tokensBefore: 38085,
fileOps: { read: [], edited: [], written: [] },
settings: { reserveTokens: 4000 },
isSplitTurn: true,
isSplitTurn: false,
},
customInstructions: "",
signal: new AbortController().signal,
@@ -3652,182 +3899,74 @@ describe("compaction-safeguard double-compaction guard", () => {
});
const compaction = expectCompactionResult(result);
expect(compaction.summary).toContain("branch summary");
expect(compaction.summary).not.toContain("No prior history.");
expect(compaction.summary).toContain("tool window summary");
expect(mockSummarizeInStages).toHaveBeenCalledTimes(1);
const summarizeCall = requireRecord(mockCallArg(mockSummarizeInStages));
const messages = requireArray(summarizeCall.messages);
expect(
messages.some((message) => {
const record = requireRecord(message);
return (
record.role === "custom" &&
record.customType === "cron-request" &&
record.content === "prepare the daily report"
);
}),
).toBe(true);
expect(
messages.some((message) => {
const record = requireRecord(message);
return record.role === "toolResult" && record.toolName === "read";
}),
).toBe(true);
expect(messages.map((message) => requireRecord(message).role)).toEqual([
"assistant",
"toolResult",
]);
expect(JSON.stringify(messages)).not.toContain("behind reset");
});
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",
},
timestamp: now,
},
},
{
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,
},
customInstructions: "",
signal: new AbortController().signal,
};
const { result, getApiKeyAndHeadersMock } = await runCompactionScenario({
sessionManager,
event: mockEvent,
apiKey: "dummy",
});
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"] },
{ toolName: "functions.sessions_send", expectedRoles: ["user", "assistant"] },
])("preserves unfinished inter-session work after a $toolName result", async (scenario) => {
it("recovers real conversation the preparation omitted from its boundary-scoped range", async () => {
mockSummarizeInStages.mockReset();
mockSummarizeInStages.mockResolvedValue(summaryResult("unfinished branch summary"));
mockSummarizeInStages.mockResolvedValue(summaryResult("range summary"));
const now = Date.now();
const entry = (id: string, parentId: string | null, offset: number, message: AgentMessage) => ({
type: "message",
id,
parentId,
timestamp: new Date(now + offset).toISOString(),
message: { ...message, timestamp: now + offset },
});
const omittedUser = { role: "user", content: "verify the deploy status now" } as AgentMessage;
const toolCallAssistant = {
role: "assistant",
content: [{ type: "toolCall", id: "call-1", name: "exec", arguments: {} }],
} as AgentMessage;
const toolResult = {
role: "toolResult",
toolCallId: "call-1",
toolName: "exec",
content: [{ type: "text", text: "deploy green" }],
} as AgentMessage;
const sessionManager = {
...stubSessionManager(),
getBranch: () => [
entry("old-user", null, 0, {
role: "user",
content: "old request behind reset",
} as AgentMessage),
{
type: "message",
id: "user-1",
parentId: null,
timestamp: new Date(now).toISOString(),
message: {
role: "user",
content: "continue the delegated task",
provenance: {
kind: "inter_session",
sourceSessionKey: "agent:pm",
sourceTool: "sessions_send",
},
timestamp: now,
},
},
{
type: "message",
id: "assistant-1",
parentId: "user-1",
type: "reset",
id: "reset-1",
parentId: "old-user",
timestamp: new Date(now + 1).toISOString(),
message: {
role: "assistant",
content: [
{
type: "toolCall",
id: "call-tool",
name: scenario.toolName,
arguments: {},
},
],
timestamp: now + 1,
},
},
{
type: "message",
id: "tool-1",
parentId: "assistant-1",
timestamp: new Date(now + 2).toISOString(),
message: {
role: "toolResult",
toolCallId: "call-tool",
toolName: scenario.toolName,
content: [{ type: "text", text: "partial evidence" }],
timestamp: now + 2,
},
reason: "new",
firstKeptEntryId: "reset-1",
},
entry("omitted-user", "reset-1", 2, omittedUser),
entry("assistant-1", "omitted-user", 3, toolCallAssistant),
entry("tool-1", "assistant-1", 4, toolResult),
entry("kept-user", "tool-1", 5, { role: "user", content: "and then?" } as AgentMessage),
],
} as ExtensionContext["sessionManager"];
const model = createAnthropicModelFixture();
setCompactionSafeguardRuntime(sessionManager, { model, recentTurnsPreserve: 0 });
// The preparation covers everything before "kept-user" but lost the user turn.
const mockEvent = {
preparation: {
messagesToSummarize: [] as AgentMessage[],
messagesToSummarize: [toolCallAssistant, toolResult],
turnPrefixMessages: [] as AgentMessage[],
firstKeptEntryId: "entry-8",
firstKeptEntryId: "kept-user",
tokensBefore: 38085,
fileOps: { read: [], edited: [], written: [] },
settings: { reserveTokens: 4000 },
isSplitTurn: true,
isSplitTurn: false,
},
customInstructions: "",
signal: new AbortController().signal,
@@ -3835,259 +3974,29 @@ describe("compaction-safeguard double-compaction guard", () => {
const { result } = await runCompactionScenario({
sessionManager,
event: mockEvent,
apiKey: "dummy",
apiKey: "test-key",
});
const compaction = expectCompactionResult(result);
expect(compaction.summary).toContain("unfinished branch summary");
expect(mockSummarizeInStages).toHaveBeenCalledTimes(1);
const summarizeCall = requireRecord(mockCallArg(mockSummarizeInStages));
const messages = requireArray(summarizeCall.messages);
expect(messages.map((message) => requireRecord(message).role)).toEqual(scenario.expectedRoles);
});
it("keeps source-session sends as inert status history", async () => {
mockSummarizeInStages.mockReset();
mockSummarizeInStages.mockResolvedValue(summaryResult("completed send 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: "ask the bee session to respond",
timestamp: now,
},
},
{
type: "message",
id: "assistant-1",
parentId: "user-1",
timestamp: new Date(now + 1).toISOString(),
message: {
role: "assistant",
content: [
{
type: "toolCall",
id: "call-1",
name: "functions.sessions_send",
arguments: { sessionKey: "agent:bee", message: "say bee" },
},
{
type: "toolCall",
id: "call-2",
name: "tools/sessions_send",
arguments: { sessionKey: "agent:wasp", message: "say wasp" },
},
{
type: "toolCall",
id: "call-3",
name: "sessions_send",
arguments: { sessionKey: "agent:ant", message: "say ant" },
},
],
timestamp: now + 1,
},
},
{
type: "message",
id: "tool-1",
parentId: "assistant-1",
timestamp: new Date(now + 2).toISOString(),
message: {
role: "toolResult",
toolCallId: "call-1",
toolName: "functions.sessions_send",
content: [{ type: "text", text: '{"status":"ok","reply":"bee replied"}' }],
timestamp: now + 2,
},
},
{
type: "message",
id: "tool-3",
parentId: "tool-1",
timestamp: new Date(now + 3).toISOString(),
message: {
role: "toolResult",
toolCallId: "call-3",
toolName: "sessions_send",
content: [{ type: "text", text: '{"status":"error","error":"ant unavailable"}' }],
timestamp: now + 3,
},
},
],
} as ExtensionContext["sessionManager"];
const model = createAnthropicModelFixture();
setCompactionSafeguardRuntime(sessionManager, { model, recentTurnsPreserve: 0 });
const mockEvent = {
preparation: {
messagesToSummarize: [] as AgentMessage[],
turnPrefixMessages: [] as AgentMessage[],
firstKeptEntryId: "entry-8",
tokensBefore: 38085,
fileOps: { read: [], edited: [], written: [] },
settings: { reserveTokens: 4000 },
isSplitTurn: true,
},
customInstructions: "",
signal: new AbortController().signal,
};
const { result } = await runCompactionScenario({
sessionManager,
event: mockEvent,
apiKey: "dummy",
});
const compaction = expectCompactionResult(result);
expect(compaction.summary).toContain("completed send summary");
expect(mockSummarizeInStages).toHaveBeenCalledTimes(1);
const summarizeCall = requireRecord(mockCallArg(mockSummarizeInStages));
const messages = requireArray(summarizeCall.messages);
expect(messages.map((message) => requireRecord(message).role)).toEqual(["user", "assistant"]);
expect(JSON.stringify(messages)).toContain("sessions_send result received");
expect(JSON.stringify(messages)).toContain("sessions_send result missing");
expect(JSON.stringify(messages)).toContain("bee replied");
expect(JSON.stringify(messages)).toContain("ant unavailable");
expect(JSON.stringify(messages)).toContain("agent:wasp");
expect(JSON.stringify(messages)).toContain("say wasp");
expect(JSON.stringify(messages)).not.toContain("sessions_send completed");
expect(JSON.stringify(messages)).not.toContain("functions.sessions_send");
expect(JSON.stringify(messages)).not.toContain("tools/sessions_send");
});
it("preserves completed historical inter-session turns outside the active tail", async () => {
mockSummarizeInStages.mockReset();
mockSummarizeInStages.mockResolvedValue(summaryResult("historical 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: "historical inter-session request",
provenance: {
kind: "inter_session",
sourceSessionKey: "agent:pm",
sourceTool: "sessions_send",
},
timestamp: now,
},
},
{
type: "message",
id: "assistant-1",
parentId: "user-1",
timestamp: new Date(now + 1).toISOString(),
message: {
role: "assistant",
content: [{ type: "text", text: "historical reply" }],
timestamp: now + 1,
},
},
{
type: "message",
id: "user-2",
parentId: "assistant-1",
timestamp: new Date(now + 2).toISOString(),
message: { role: "user", content: "later user request", timestamp: now + 2 },
},
{
type: "message",
id: "assistant-2",
parentId: "user-2",
timestamp: new Date(now + 3).toISOString(),
message: {
role: "assistant",
content: [{ type: "text", text: "later reply" }],
timestamp: now + 3,
},
},
],
} as ExtensionContext["sessionManager"];
const model = createAnthropicModelFixture();
setCompactionSafeguardRuntime(sessionManager, { model, recentTurnsPreserve: 0 });
const mockEvent = {
preparation: {
messagesToSummarize: [] as AgentMessage[],
turnPrefixMessages: [] as AgentMessage[],
firstKeptEntryId: "entry-8",
tokensBefore: 38085,
fileOps: { read: [], edited: [], written: [] },
settings: { reserveTokens: 4000 },
isSplitTurn: true,
},
customInstructions: "",
signal: new AbortController().signal,
};
const { result } = await runCompactionScenario({
sessionManager,
event: mockEvent,
apiKey: "dummy",
});
const compaction = expectCompactionResult(result);
expect(compaction.summary).toContain("historical branch summary");
expect(expectCompactionResult(result).summary).toContain("range summary");
expect(mockSummarizeInStages).toHaveBeenCalledTimes(1);
const summarizeCall = requireRecord(mockCallArg(mockSummarizeInStages));
const messages = requireArray(summarizeCall.messages);
expect(messages.map((message) => requireRecord(message).role)).toEqual([
"user",
"assistant",
"user",
"assistant",
"toolResult",
]);
expect(JSON.stringify(messages)).toContain("historical inter-session request");
expect(JSON.stringify(messages)).toContain("historical reply");
const serialized = JSON.stringify(messages);
expect(serialized).toContain("verify the deploy status now");
expect(serialized).not.toContain("behind reset");
expect(serialized).not.toContain("and then?");
});
it("recovers user and assistant branch turns when compaction preparation has only tool output", async () => {
it("writes the anti-loop boundary for a tool-only window when nothing anchors it", async () => {
mockSummarizeInStages.mockReset();
mockSummarizeInStages.mockResolvedValue(summaryResult("branch summary with visible turns"));
const now = Date.now();
const sessionManager = {
...stubSessionManager(),
getBranch: () => [
{
type: "message",
id: "user-1",
parentId: null,
timestamp: new Date(now).toISOString(),
message: {
role: "user",
content: "what is the deployment status?",
timestamp: now,
},
},
{
type: "message",
id: "assistant-1",
parentId: "user-1",
timestamp: new Date(now + 1).toISOString(),
message: {
role: "assistant",
content: [{ type: "text", text: "I will check the deploy." }],
timestamp: now + 1,
},
},
],
} as ExtensionContext["sessionManager"];
const sessionManager = stubSessionManager();
const model = createAnthropicModelFixture();
setCompactionSafeguardRuntime(sessionManager, { model, recentTurnsPreserve: 0 });
setCompactionSafeguardRuntime(sessionManager, { model });
const mockEvent = {
preparation: {
@@ -4095,32 +4004,29 @@ describe("compaction-safeguard double-compaction guard", () => {
{
role: "toolResult",
toolCallId: "call-1",
toolName: "status",
content: [{ type: "text", text: "deploy green" }],
timestamp: now + 2,
toolName: "exec",
content: [{ type: "text", text: "heartbeat probe ok" }],
timestamp: Date.now(),
},
] as AgentMessage[],
turnPrefixMessages: [] as AgentMessage[],
firstKeptEntryId: "entry-6",
tokensBefore: 38085,
firstKeptEntryId: "entry-1",
tokensBefore: 1500,
fileOps: { read: [], edited: [], written: [] },
settings: { reserveTokens: 4000 },
isSplitTurn: true,
},
customInstructions: "",
signal: new AbortController().signal,
};
const { result } = await runCompactionScenario({
const { result, getApiKeyAndHeadersMock } = await runCompactionScenario({
sessionManager,
event: mockEvent,
apiKey: "test-key",
});
const compaction = expectCompactionResult(result);
expect(compaction.summary).toContain("branch summary with visible turns");
const summarizeCall = requireRecord(mockCallArg(mockSummarizeInStages));
const messages = requireArray(summarizeCall.messages);
expect(messages.map((message) => requireRecord(message).role)).toEqual(["user", "assistant"]);
expect(compaction.summary).toContain("No prior history.");
expect(mockSummarizeInStages).not.toHaveBeenCalled();
expect(getApiKeyAndHeadersMock).not.toHaveBeenCalled();
});
it("continues when messages include real conversation content", async () => {
src/agents/agent-hooks/compaction-safeguard.ts
+67 -177
View File
@@ -3,8 +3,6 @@
import fs from "node:fs";
import path from "node:path";
/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */
import { parseDateFirstTimestampMs } from "@openclaw/normalization-core/number-coercion";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { sliceUtf16Safe, truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import {
capCompactionSummary,
@@ -27,7 +25,6 @@ import {
getCompactionProvider,
type CompactionProvider,
} from "../../plugins/compaction-provider.js";
import { normalizeInputProvenance } from "../../sessions/input-provenance.js";
import { normalizeAcceptedSessionSpawnResult } from "../accepted-session-spawn.js";
import { computeAdaptiveChunkRatioWithWorker } from "../compaction-planning-worker.js";
import { buildHistoryPrunePlan } from "../compaction-planning.js";
@@ -48,7 +45,11 @@ import { collectTextContentBlocks } from "../content-blocks.js";
import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "../copilot-dynamic-headers.js";
import { isTimeoutError } from "../failover-error.js";
import { stripRuntimeContextCustomMessages } from "../internal-runtime-context.js";
import type { AgentMessage } from "../runtime/index.js";
import {
buildSessionContext as buildCoreSessionContext,
type AgentMessage,
type SessionTreeEntry as CoreSessionTreeEntry,
} from "../runtime/index.js";
import { repairToolUseResultPairing } from "../session-transcript-repair.js";
import type { ExtensionAPI, ExtensionContext } from "../sessions/index.js";
import { extractToolCallsFromAssistant, extractToolResultId } from "../tool-call-id.js";
@@ -93,7 +94,6 @@ const MAX_QUALITY_GUARD_MAX_RETRIES = 3;
const MAX_RECENT_TURN_TEXT_CHARS = 600;
const MAX_REQUIRED_ASK_CONTEXT_CHARS = 2_000;
const REQUIRED_ASK_CONTEXT_TRUNCATED_MARKER = "\n[... split-turn ask context truncated ...]\n";
const TOOL_CALL_BLOCK_TYPES = new Set(["toolCall", "toolUse", "functionCall"]);
const PREVIOUS_SUMMARY_REDISTILL_PREFIX =
"Previous compaction summary to re-distill with the current conversation. " +
"Prune stale, duplicate, or superseded details instead of preserving it verbatim.";
@@ -130,172 +130,50 @@ function prependPreviousSummaryForRedistill(params: {
];
}
function coerceTimestamp(value: unknown): number {
return parseDateFirstTimestampMs(value) ?? 0;
/**
* Messages the model currently sees: the last reset/compaction boundary's kept
* tail plus everything after it. Never the raw branch — that re-reads history
* behind every boundary and turns one compaction into dozens of model calls.
*/
function collectSessionContextMessages(sessionManager: unknown): AgentMessage[] {
return projectBranchEntries(readSessionBranch(sessionManager));
}
function sessionBranchEntryToMessage(entry: Record<string, unknown>): unknown {
if (entry.type === "message" && entry.message && typeof entry.message === "object") {
return entry.message;
/**
* The boundary-scoped range a preparation was meant to cover: everything the
* current context holds before its kept tail, minus the prior summary message
* (that is re-distilled separately). Bounded by construction — it can never
* reach behind the last reset/compaction boundary.
*/
function collectPreparationRangeMessages(
sessionManager: unknown,
firstKeptEntryId: string,
): AgentMessage[] {
const entries = readSessionBranch(sessionManager);
const firstKeptIndex = entries.findIndex((entry) => entry.id === firstKeptEntryId);
if (firstKeptIndex < 0) {
return [];
}
if (entry.type === "custom_message") {
return {
role: "custom",
customType: typeof entry.customType === "string" ? entry.customType : "custom",
content: entry.content,
display: entry.display !== false,
details: entry.details,
timestamp: coerceTimestamp(entry.timestamp),
};
}
if (entry.type === "branch_summary") {
return {
role: "branchSummary",
summary: typeof entry.summary === "string" ? entry.summary : "",
fromId: typeof entry.fromId === "string" ? entry.fromId : "root",
timestamp: coerceTimestamp(entry.timestamp),
};
}
return undefined;
return projectBranchEntries(entries.slice(0, firstKeptIndex)).filter(
(message) => message.role !== "compactionSummary",
);
}
function collectSessionBranchMessages(sessionManager: unknown): AgentMessage[] {
function readSessionBranch(sessionManager: unknown): CoreSessionTreeEntry[] {
try {
const entries: unknown = (sessionManager as { getBranch?: () => unknown })?.getBranch?.();
return Array.isArray(entries)
? entries.flatMap((entry) => {
const message =
entry && typeof entry === "object"
? sessionBranchEntryToMessage(entry as Record<string, unknown>)
: undefined;
return message ? [message as AgentMessage] : [];
})
: [];
return Array.isArray(entries) ? (entries as CoreSessionTreeEntry[]) : [];
} catch {
return [];
}
}
function isSessionsSendToolName(value: unknown): boolean {
return (
normalizeOptionalString(value)
?.toLowerCase()
.replace(/^(?:functions?|tools?)[./_-]/, "") === "sessions_send"
);
}
function sanitizeSourceSessionSends(messages: AgentMessage[]): AgentMessage[] {
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<string, string>();
for (const message of messages) {
if (message.role !== "toolResult") {
continue;
}
const callId = extractToolResultId(message);
if (!callId || !sendCallIds.has(callId)) {
continue;
}
resultTextByCallId.set(
callId,
extractMessageText(message) || formatNonTextPlaceholder(message.content) || "",
);
function projectBranchEntries(entries: CoreSessionTreeEntry[]): AgentMessage[] {
try {
return buildCoreSessionContext(entries).messages as AgentMessage[];
} catch {
return [];
}
return messages.flatMap((message) => {
if (message.role === "assistant" && Array.isArray(message.content)) {
let replaced = false;
const content = message.content.map((block) => {
if (!block || typeof block !== "object") {
return block;
}
const record = block as {
type?: unknown;
id?: unknown;
name?: unknown;
arguments?: unknown;
};
if (
typeof record.type !== "string" ||
!TOOL_CALL_BLOCK_TYPES.has(record.type) ||
!isSessionsSendToolName(record.name)
) {
return block;
}
replaced = true;
const callId = typeof record.id === "string" ? record.id.trim() : "";
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: `sessions_send result ${resolved ? "received" : "missing"}; delivery call omitted from replay.\nRequest: ${requestText}${resultSuffix}`,
};
});
return replaced ? [{ ...message, content } as AgentMessage] : [message];
}
if (message.role === "toolResult") {
const callId = extractToolResultId(message);
if ((callId && sendCallIds.has(callId)) || isSessionsSendToolName(message.toolName)) {
return [];
}
}
return [message];
});
}
function filterReplayUnsafeSessionBranchMessages(messages: AgentMessage[]): AgentMessage[] {
const sanitizedMessages = sanitizeSourceSessionSends(messages);
let turnStart = sanitizedMessages.length;
while (turnStart > 0) {
const role = (sanitizedMessages[turnStart - 1] as { role?: unknown }).role;
if (role !== "assistant" && role !== "toolResult") {
break;
}
turnStart -= 1;
}
const tailMessage = messages.at(-1);
const endsWithTerminalAssistantText =
tailMessage !== undefined &&
tailMessage.role === "assistant" &&
Boolean(extractMessageText(tailMessage).trim()) &&
(!Array.isArray(tailMessage.content) ||
!tailMessage.content.some((block) => {
if (!block || typeof block !== "object") {
return false;
}
const type = (block as { type?: unknown }).type;
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.
if (
endsWithTerminalAssistantText &&
turnStart < sanitizedMessages.length &&
turnStart > 0 &&
activeInputProvenance?.kind === "inter_session" &&
activeInputProvenance.sourceTool === "sessions_send"
) {
return sanitizedMessages.slice(0, turnStart - 1);
}
return sanitizedMessages;
}
function containsRealConversation(messages: AgentMessage[]): boolean {
@@ -621,7 +499,13 @@ function budgetCompactionSummary(
) {
const suffix = normalizeCompactionSuffix(suffixInput);
const joined = `${summaryBody}${suffix.text}`;
if (maxChars <= 0 || joined.length <= maxChars) {
// A body that fits still goes through the retention plan when it omits an
// audited identifier or lets an audit section outgrow its cap; both would
// re-distill into the next summary otherwise.
const retentionPlan = qualityRetention
? createSummaryQualityRetentionPlan(summaryBody, SUMMARY_TRUNCATED_MARKER, qualityRetention)
: null;
if (maxChars <= 0 || (joined.length <= maxChars && !retentionPlan?.needsRebuild(maxChars))) {
return {
summary: joined,
structuralSummary: summaryBody,
@@ -632,9 +516,6 @@ function budgetCompactionSummary(
};
}
const retentionPlan = qualityRetention
? createSummaryQualityRetentionPlan(summaryBody, SUMMARY_TRUNCATED_MARKER, qualityRetention)
: null;
const bodyCapacity = retentionPlan ? maxChars : summaryBody.length;
const bodyFloor = Math.min(
bodyCapacity,
@@ -643,14 +524,15 @@ function budgetCompactionSummary(
);
const suffixReservation = Math.min(suffix.text.length, maxChars);
const bodySlot = Math.min(bodyCapacity, Math.max(bodyFloor, maxChars - suffixReservation));
const cappedBody = retentionPlan?.render(bodySlot) ?? capCompactionSummary(summaryBody, bodySlot);
const rendered = retentionPlan?.render(bodySlot);
const cappedBody = rendered?.text ?? capCompactionSummary(summaryBody, bodySlot);
const suffixBudget = Math.max(0, maxChars - cappedBody.length);
const cappedSuffix = capCompactionSuffix(suffix, suffixBudget);
return {
summary: `${cappedBody}${cappedSuffix}`,
structuralSummary: cappedBody,
bodyBudget: bodySlot,
bodyTrimmed: cappedBody.length < summaryBody.length,
bodyTrimmed: rendered ? rendered.trimmed : cappedBody.length < summaryBody.length,
suffixTrimmed: cappedSuffix.length < suffix.text.length,
qualityRetentionInfeasible: retentionPlan !== null && retentionPlan.minimumChars > maxChars,
};
@@ -1044,29 +926,37 @@ export default function compactionSafeguardExtension(api: ExtensionAPI): void {
thinkingLevel,
streamFn,
} = event;
const rawTurnPrefixMessages = preparation.turnPrefixMessages ?? [];
let baseMessagesToSummarize = stripRuntimeContextCustomMessages(
preparation.messagesToSummarize,
);
let baseTurnPrefixMessages = stripRuntimeContextCustomMessages(rawTurnPrefixMessages);
let hasRealSummarizable = containsRealConversation(baseMessagesToSummarize);
let hasRealTurnPrefix = containsRealConversation(baseTurnPrefixMessages);
if (!hasRealSummarizable && !hasRealTurnPrefix) {
const branchMessages = filterReplayUnsafeSessionBranchMessages(
stripRuntimeContextCustomMessages(collectSessionBranchMessages(ctx.sessionManager)),
let baseTurnPrefixMessages = stripRuntimeContextCustomMessages(
preparation.turnPrefixMessages ?? [],
);
if (!containsRealConversation([...baseMessagesToSummarize, ...baseTurnPrefixMessages])) {
// Safety net for a preparation that dropped real conversation from the
// range it covers: summarize that boundary-scoped range instead. It is
// never the raw branch, which re-read every reset and prior compaction.
const rangeMessages = stripRuntimeContextCustomMessages(
collectPreparationRangeMessages(ctx.sessionManager, preparation.firstKeptEntryId),
);
if (containsRealConversation(branchMessages)) {
if (containsRealConversation(rangeMessages)) {
log.info(
"Compaction safeguard: using session branch messages after compaction preparation omitted real conversation content.",
"Compaction safeguard: summarizing the boundary-scoped preparation range after compaction preparation omitted real conversation content.",
);
baseMessagesToSummarize = branchMessages;
baseMessagesToSummarize = rangeMessages;
baseTurnPrefixMessages = [];
hasRealSummarizable = true;
hasRealTurnPrefix = false;
}
}
// A prepared window of pure tool traffic is still real work when the current
// context anchors it (a kept user turn or a prior compaction summary); only a
// context with no real conversation at all gets the anti-loop boundary below.
const hasRealConversation =
containsRealConversation([...baseMessagesToSummarize, ...baseTurnPrefixMessages]) ||
containsRealConversation(
stripRuntimeContextCustomMessages(collectSessionContextMessages(ctx.sessionManager)),
);
setCompactionSafeguardCancelReason(ctx.sessionManager, undefined);
if (!hasRealSummarizable && !hasRealTurnPrefix) {
if (!hasRealConversation) {
// When there are no summarizable messages AND no real turn-prefix content,
// cancelling compaction leaves context unchanged but the SDK re-triggers
// _checkCompaction after every assistant response — creating a cancel loop