diff --git a/src/chain-checkpoint.ts b/src/chain-checkpoint.ts index 9c97795055..71a6cb23b7 100644 --- a/src/chain-checkpoint.ts +++ b/src/chain-checkpoint.ts @@ -1,5 +1,5 @@ import { createHash } from "node:crypto"; -import type { WireProtocol } from "./util.js"; +import { trailingSystemRunStart, type WireProtocol } from "./util.js"; import { normalizeGoogleUserPartOrder, type GooglePart } from "acp-kernel/wire"; // #1357/#1395/#1421 — Chain Checkpoint (first-processor-wins idempotent interop). @@ -509,6 +509,13 @@ export function insertCheckpointCarrier(parsed: unknown, wire: WireProtocol, tag return { ...body, messages: [...messages.slice(0, li), { ...m, content: merged }] }; } } + // #5432: a body ending in a client trailing system run (legal — the + // kernel heal keeps it verbatim at array end) admits NO carrier slot: + // recognition only sees the TRAILING USER run (contract above), so a + // carrier inserted before the run would never be recognized again, and + // appending after the run is exactly the 400 shape (#5432 case 2). + // Skip the stamp — relay forwards unstamped on null (best-effort). + if (trailingSystemRunStart(messages) < messages.length) return null; return { ...body, messages: [...messages, { role: "user", content: tag }] }; } if (wire === "responses") { diff --git a/src/loop/adapter-anthropic.ts b/src/loop/adapter-anthropic.ts index a2aa0ac283..adfb16e031 100644 --- a/src/loop/adapter-anthropic.ts +++ b/src/loop/adapter-anthropic.ts @@ -1,5 +1,5 @@ import type { CoreMessage } from "acp-kernel"; -import { coreToAnthropic, healAnthropicSystemPositions, type AnthropicRequestBody } from "acp-kernel/wire"; +import { coreToAnthropic, healAnthropicSystemPositions, insertBeforeTrailingSystemRun, type AnthropicRequestBody } from "acp-kernel/wire"; import { dedupeAdjacentOutputConfig } from "../dedupe-adjacent-output-config.js"; import { appendSystemText, declaredInclusiveFromUsage } from "../util.js"; import { stampAnthropicSystemCacheControl } from "./cache-control.js"; @@ -316,8 +316,10 @@ export function createAnthropicAdapter(requestBody: Record, ori // injectSystem — both must emit byte-identical system (F2 seam). const system = originalSystem !== undefined ? appendSystemText(systemPrompt, originalSystem) : systemPrompt; const stamped = cacheMarks ? stampAnthropicSystemCacheControl(system) : system; + // #5432: notes go BEFORE a trailing client system run; the #4570 + // backstop heals whatever still arrives illegal (fold orphans). const withNotes = notes && notes.length > 0 - ? healAnthropicSystemPositions([...messages, ...notes.map((text) => ({ role: "user" as const, content: text }))]) + ? healAnthropicSystemPositions(insertBeforeTrailingSystemRun(messages, notes.map((text) => ({ role: "user" as const, content: text })))) : messages; return { ...body, system: stamped, messages: withNotes }; }, diff --git a/src/server/prepare-anthropic.ts b/src/server/prepare-anthropic.ts index a76a48b3b5..f3f08e6c8a 100644 --- a/src/server/prepare-anthropic.ts +++ b/src/server/prepare-anthropic.ts @@ -489,7 +489,7 @@ export async function prepareAnthropic( const egressMessages = await applyImageRescalePass(processedMessages as BiliMessage[], { session, sessionId, log, maxDim: imageRescaleMaxDim(), toolArmed: imageRescaleToolArmed(session) }); rebuiltMessages = coreToAnthropic(egressMessages, anthropicCacheMarks ?? cacheControls); if (sysNotes.length > 0) { - rebuiltMessages = [...rebuiltMessages, ...sysNotes.map((text) => ({ role: "user" as const, content: text }))]; + rebuiltMessages = insertBeforeTrailingSystemRun(rebuiltMessages, sysNotes.map((text) => ({ role: "user" as const, content: text }))); } systemOut = injectSystem(parsed, opts, prompts, loopConfig, surface, visibilityMarkers, laneDecisions.compressSurfaceOff, laneDecisions.mcpDialect, detectGenerationBoundary(session, processedMessages, log)); @@ -523,7 +523,7 @@ export async function prepareAnthropic( const rendered = renderNudgeText(turn.nudge, prompts, surface?.nudgeSections, "anthropic"); const renderedWithPayload = rendered.text; if (rendered.text) { - rebuiltMessages = [...rebuiltMessages, { role: "user", content: withMarkerIntegrityNote(withSummaryBudgetNote(withFirstSightDrain(renderedWithPayload, turn.nudge.reason, externalSummaryEnabled(config)), externalSummaryEnabled(config)), visibilityMarkers) }]; + rebuiltMessages = insertBeforeTrailingSystemRun(rebuiltMessages, [{ role: "user", content: withMarkerIntegrityNote(withSummaryBudgetNote(withFirstSightDrain(renderedWithPayload, turn.nudge.reason, externalSummaryEnabled(config)), externalSummaryEnabled(config)), visibilityMarkers) }]); nudgeOutcome = "injected"; } else { nudgeOutcome = "skipped"; @@ -536,15 +536,17 @@ export async function prepareAnthropic( // [#5688] the side call shares the main request's upstream — // un-deduped rebuiltMessages would 400 the decision too. // [#4570] and the side array must obey the system-position rule - // on the array actually sent, which is exactly this one. - const sideBody: Record = { ...parsed, messages: healAnthropicSystemPositions([...dedupeAdjacentOutputConfig(rebuiltMessages), { role: "user", content: buildDecisionPrompt(ranges) }]), system: systemOut, tools: toolsOut, stream: false, max_tokens: decide.maxTokens }; + // on the array actually sent, which is exactly this one; #5432 + // inserts the prompt BEFORE a trailing client system run so the + // backstop below stays a no-op for legal shapes. + const sideBody: Record = { ...parsed, messages: healAnthropicSystemPositions(insertBeforeTrailingSystemRun(dedupeAdjacentOutputConfig(rebuiltMessages), [{ role: "user", content: buildDecisionPrompt(ranges) }])), system: systemOut, tools: toolsOut, stream: false, max_tokens: decide.maxTokens }; delete sideBody.prompt_cache_key; stripDecisionReasoningFields("anthropic", sideBody); const outcome = await runNudgeDecision({ req, opts, protocol: "anthropic", sideBody, session, log }); if (outcome.kind === "yes") { const span = resolveDecisionRange(outcome, ranges); if (span) { - rebuiltMessages = [...rebuiltMessages, { role: "user", content: withMarkerIntegrityNote(withSummaryBudgetNote(withFirstSightDrain(buildDirectiveText(span.startRef, span.endRef, outcome.topic), turn.nudge.reason, externalSummaryEnabled(config)), externalSummaryEnabled(config)), visibilityMarkers) }]; + rebuiltMessages = insertBeforeTrailingSystemRun(rebuiltMessages, [{ role: "user", content: withMarkerIntegrityNote(withSummaryBudgetNote(withFirstSightDrain(buildDirectiveText(span.startRef, span.endRef, outcome.topic), turn.nudge.reason, externalSummaryEnabled(config)), externalSummaryEnabled(config)), visibilityMarkers) }]); nudgeOutcome = "injected"; } else { nudgeOutcome = "skipped"; @@ -560,7 +562,7 @@ export async function prepareAnthropic( const rendered = renderNudgeText(turn.nudge, prompts, surface?.nudgeSections, "anthropic"); const renderedWithPayload = rendered.text; if (rendered.text) { - rebuiltMessages = [...rebuiltMessages, { role: "user", content: withMarkerIntegrityNote(withSummaryBudgetNote(withFirstSightDrain(renderedWithPayload, turn.nudge.reason, externalSummaryEnabled(config)), externalSummaryEnabled(config)), visibilityMarkers) }]; + rebuiltMessages = insertBeforeTrailingSystemRun(rebuiltMessages, [{ role: "user", content: withMarkerIntegrityNote(withSummaryBudgetNote(withFirstSightDrain(renderedWithPayload, turn.nudge.reason, externalSummaryEnabled(config)), externalSummaryEnabled(config)), visibilityMarkers) }]); nudgeOutcome = "injected"; } else { nudgeOutcome = "skipped"; @@ -580,16 +582,17 @@ export async function prepareAnthropic( // carrier as the nudge (never persisted, never in the client's // re-sent history, prefix-cache safe). if (turn.statusLine) { - rebuiltMessages = [...rebuiltMessages, { role: "user", content: turn.statusLine }]; + rebuiltMessages = insertBeforeTrailingSystemRun(rebuiltMessages, [{ role: "user", content: turn.statusLine }]); } // [#1343/#1457] surface any earlier undelivered retrieve as an ephemeral - // trailing user note (kept last so it never reorders cached messages). + // trailing user note (after every stable message; only a client trailing + // system run may follow it — #5432). // Snapshot WITHOUT consuming: forward() commits the ids only once // upstream accepts this request; on failure they ride the next one. const retrNotes = snapshotRetrievalNotes(session); const retrNote = renderRetrievalNotes(retrNotes); if (retrNote) { - rebuiltMessages = [...rebuiltMessages, { role: "user", content: retrNote }]; + rebuiltMessages = insertBeforeTrailingSystemRun(rebuiltMessages, [{ role: "user", content: retrNote }]); attachedRetrievalNoteIds = retrNotes.map((n) => n.id); } } catch (err) { diff --git a/src/util.ts b/src/util.ts index 0d8252895a..7755425244 100644 --- a/src/util.ts +++ b/src/util.ts @@ -659,6 +659,21 @@ export function systemToUser(messages: T[], should?: ); } +// #5432: index of the first entry of the trailing run of role:"system" entries +// (== length when nothing trails). Shared by every host path that mutates an +// ALREADY assembled anthropic body: the kernel heal keeps a client's legal +// trailing system run verbatim at array end, so any post-assembly user insert +// must land BEFORE it or the request 400s ("role 'system' must precede an +// 'assistant' message or end the array"). +export function trailingSystemRunStart(messages: readonly unknown[]): number { + let i = messages.length; + while (i > 0) { + const m = messages[i - 1]; + if (!m || typeof m !== "object" || Array.isArray(m) || (m as Record).role !== "system") break; + i--; + } + return i; +} /** #719: Some OpenAI-compatible backends (DeepSeek) reject assistant messages * whose `content` is null — they require a string content (possibly "") or diff --git a/src/wire-body.ts b/src/wire-body.ts index 3fc261d82e..b11f683613 100644 --- a/src/wire-body.ts +++ b/src/wire-body.ts @@ -1,4 +1,4 @@ -import type { WireProtocol } from "./util.js"; +import { trailingSystemRunStart, type WireProtocol } from "./util.js"; import { normalizeGoogleUserPartOrder, type GooglePart } from "acp-kernel/wire"; /** Append `text` as a trailing user turn on the wire's own request body. @@ -62,6 +62,13 @@ export function appendTrailingUserText(protocol: WireProtocol, body: string | Bu if (typeof c === "string") last.content = `${c}\n\n${text}`; else if (Array.isArray(c)) last.content = [...c, { type: "text", text }]; else last.content = text; + } else if (protocol === "anthropic") { + // #5432: the assembled body can legally END in a client trailing system + // run (the kernel heal keeps it verbatim at array end); a plain push + // would put a user after it and 400. Insert before the run instead — + // identical to the push when no system trails. OpenAI has no such rule, + // so its bytes stay exactly as before. + arr.splice(trailingSystemRunStart(arr), 0, { role: "user", content: text }); } else { arr.push({ role: "user", content: text }); } diff --git a/tests/issue5432-trailing-system-tails.test.ts b/tests/issue5432-trailing-system-tails.test.ts new file mode 100644 index 0000000000..c2ece710ac --- /dev/null +++ b/tests/issue5432-trailing-system-tails.test.ts @@ -0,0 +1,372 @@ +// #5432: Anthropic's mid-conversation system rule is TWO-sided — a non-directive +// system entry must FOLLOW a user message AND PRECEDE an assistant message or +// end the array. Four bili producers broke the PRECEDE side whenever the +// assembled body legally ends in a client trailing system run (Claude Code's +// "The user hasn't heard from you…" reminder): +// (A) the steady path (prepare-anthropic) appended ephemeral tail users +// (sysNotes/nudge/decide/statusLine/retrieval notes) AFTER coreToAnthropic's +// position heal; +// (B) the round-2 compress-loop adapter (adapter-anthropic) did the same with +// its notes; +// (C) wire-body appendTrailingUserText (fake-completion hint / degenerate-turn +// retry nudge) pushed a user after whatever ended the ALREADY assembled +// body; +// (D) chain-checkpoint insertCheckpointCarrier (opt-in egress stamping) had +// no carrier slot for a system-final body at all — the frozen contract +// (#1395/#1421) recognizes carriers only in the TRAILING USER run, and +// appending after the run was the exact 400 shape. Now returns null → +// relay forwards unstamped (best-effort). +// Pinned through all layers: the ordering helper itself, the adapter's +// buildRequest, both post-hoc body-mutation functions, and the REAL steady +// path via the hermetic server harness (client history ending in a system + an +// armed over-limit nudge). Bidirectional discipline (#3617): the +// no-trailing-system negative cases pin that the legacy tail-append behavior +// is UNCHANGED when nothing follows. +import assert from "node:assert/strict"; +import http from "node:http"; +import { once } from "node:events"; +import test from "node:test"; + +process.env.NODE_ENV = "test"; + +import { defaultConfig } from "acp-kernel"; +import { anthropicToCore, insertBeforeTrailingSystemRun, type AnthropicWireMessage } from "acp-kernel/wire"; +import { createAnthropicAdapter } from "../src/loop/index.ts"; +import { insertCheckpointCarrier } from "../src/chain-checkpoint.ts"; +import { appendTrailingUserText } from "../src/wire-body.ts"; +import { startServer } from "../src/server.ts"; +import type { ProxyOptions } from "../src/config.ts"; +import { SessionStore, _setStoreForTest } from "../src/persist.ts"; +import { _setForTest as setRegistryForTest } from "../src/registry.ts"; + +type Msg = { role: string; content: string | Array<{ type?: string; text?: string }> }; +const textOf = (m: Msg): string => + typeof m.content === "string" ? m.content : m.content.map((b) => b.text ?? "").join(""); + +test("#5432 helper: insertions land BEFORE a trailing system run, run stays verbatim at the end", () => { + const base: AnthropicWireMessage[] = [ + { role: "user", content: "q" }, + { role: "assistant", content: "a" }, + { role: "system", content: "sys1" }, + { role: "system", content: "sys2" }, + ]; + const out = insertBeforeTrailingSystemRun(base, [{ role: "user", content: "note" }]); + assert.deepEqual( + out.map((m) => `${m.role}:${String(m.content)}`), + ["user:q", "assistant:a", "user:note", "system:sys1", "system:sys2"], + ); +}); + +test("#5432 helper: no trailing system → plain tail append (legacy behavior unchanged)", () => { + const base: AnthropicWireMessage[] = [ + { role: "user", content: "q" }, + { role: "assistant", content: "a" }, + ]; + const out = insertBeforeTrailingSystemRun(base, [{ role: "user", content: "note" }]); + assert.deepEqual( + out.map((m) => `${m.role}:${String(m.content)}`), + ["user:q", "assistant:a", "user:note"], + ); +}); + +test("#5432 helper: empty additions → same array back; all-system input → additions lead", () => { + const base: AnthropicWireMessage[] = [{ role: "system", content: "s" }]; + assert.equal(insertBeforeTrailingSystemRun(base, []), base); + const out = insertBeforeTrailingSystemRun(base, [{ role: "user", content: "n" }]); + assert.deepEqual( + out.map((m) => m.role), + ["user", "system"], + ); +}); + +test("#5432 helper: sequential batches keep their relative order before the run", () => { + let msgs: AnthropicWireMessage[] = [ + { role: "user", content: "q" }, + { role: "system", content: "sys" }, + ]; + msgs = insertBeforeTrailingSystemRun(msgs, [{ role: "user", content: "first" }]); + msgs = insertBeforeTrailingSystemRun(msgs, [{ role: "user", content: "second" }]); + assert.deepEqual( + msgs.map((m) => `${m.role}:${String(m.content)}`), + ["user:q", "user:first", "user:second", "system:sys"], + ); +}); + +// --- round-2 adapter layer ------------------------------------------------- + +const ADAPTER_TRAILING_SYSTEM = "The user hasn't heard from you in a while; continue working."; + +function coreEndingInSystem(): ReturnType["msgs"] { + // #5438 landed on master: the kernel heal now enforces BOTH sides of the + // rule, so the trailing system must FOLLOW a user (or a server-tool-result + // assistant) or the kernel demotes it before the host helper runs. q2 + // makes the fixture legal-follow, matching real Claude Code requests. + return anthropicToCore({ + messages: [ + { role: "user", content: [{ type: "text", text: "q1" }] }, + { role: "assistant", content: [{ type: "text", text: "a1" }] }, + { role: "user", content: [{ type: "text", text: "q2" }] }, + { role: "system", content: [{ type: "text", text: ADAPTER_TRAILING_SYSTEM }] }, + ], + } as never).msgs; +} + +function coreNoTrailingSystem(): ReturnType["msgs"] { + return anthropicToCore({ + messages: [ + { role: "user", content: [{ type: "text", text: "q1" }] }, + { role: "assistant", content: [{ type: "text", text: "a1" }] }, + ], + } as never).msgs; +} + +test("#5432 adapter: notes insert before the trailing client system, which stays verbatim and last", () => { + const adapter = createAnthropicAdapter({ model: "claude-test", max_tokens: 10 }, undefined, ["ADAPTER-NOTE"]); + const req = adapter.buildRequest(coreEndingInSystem(), "top-system", { model: "claude-test", max_tokens: 10 }) as { messages: Msg[] }; + const msgs = req.messages; + assert.equal(msgs[msgs.length - 1]!.role, "system", "the client's trailing system must still end the array"); + assert.equal(textOf(msgs[msgs.length - 1]!), ADAPTER_TRAILING_SYSTEM, "the trailing system stays verbatim"); + assert.equal(msgs[msgs.length - 2]!.role, "user", "the note lands immediately before the trailing system"); + assert.equal(textOf(msgs[msgs.length - 2]!), "ADAPTER-NOTE"); +}); + +test("#5432 adapter negative: no trailing system → note still lands dead-last (legacy behavior)", () => { + const adapter = createAnthropicAdapter({ model: "claude-test", max_tokens: 10 }, undefined, ["ADAPTER-NOTE"]); + const req = adapter.buildRequest(coreNoTrailingSystem(), "top-system", { model: "claude-test", max_tokens: 10 }) as { messages: Msg[] }; + const msgs = req.messages; + assert.equal(msgs[msgs.length - 1]!.role, "user"); + assert.equal(textOf(msgs[msgs.length - 1]!), "ADAPTER-NOTE"); +}); + +// --- post-hoc body-mutation paths (fake-completion / degenerate retry) ----- + +test("#5432 wire-body: anthropic text lands BEFORE a trailing system run, run stays verbatim and last", () => { + const out = appendTrailingUserText("anthropic", JSON.stringify({ + model: "m", max_tokens: 10, + messages: [ + { role: "user", content: "q" }, + { role: "assistant", content: "a" }, + { role: "system", content: [{ type: "text", text: "reminder" }] }, + ], + }), "hint-text"); + assert.ok(out !== null); + const body = JSON.parse(out!) as { messages: Msg[] }; + assert.deepEqual( + body.messages.map((m) => m.role), + ["user", "assistant", "user", "system"], + ); + assert.equal(body.messages[2]!.content, "hint-text"); + assert.deepEqual(body.messages[3]!.content, [{ type: "text", text: "reminder" }], "the trailing system stays verbatim"); +}); + +test("#5432 wire-body negative: no trailing system → plain append; user-final → merge (legacy contract unchanged)", () => { + const appended = JSON.parse(appendTrailingUserText("anthropic", JSON.stringify({ + messages: [{ role: "user", content: "q" }, { role: "assistant", content: "a" }], + }), "hint-text")!) as { messages: Msg[] }; + assert.deepEqual(appended.messages.map((m) => m.role), ["user", "assistant", "user"]); + const merged = JSON.parse(appendTrailingUserText("anthropic", JSON.stringify({ + messages: [{ role: "assistant", content: "a" }, { role: "user", content: "q" }], + }), "more")!) as { messages: Msg[] }; + assert.equal(merged.messages[1]!.content, "q\n\nmore"); +}); + +test("#5432 wire-body openai pin: system-final bodies still get the plain tail append (no such rule on that wire)", () => { + const out = JSON.parse(appendTrailingUserText("openai", JSON.stringify({ + model: "m", + messages: [{ role: "user", content: "q" }, { role: "system", content: "s" }], + }), "hint-text")!) as { messages: Msg[] }; + assert.deepEqual(out.messages.map((m) => m.role), ["user", "system", "user"]); +}); + +// --- chain-checkpoint carrier (opt-in egress stamping) ---------------------- + +test("#5432 chain-checkpoint: anthropic system-final body admits no carrier slot → null (stamp skipped)", () => { + assert.equal(insertCheckpointCarrier({ + model: "m", + messages: [ + { role: "user", content: "q" }, + { role: "assistant", content: "a" }, + { role: "system", content: [{ type: "text", text: "reminder" }] }, + ], + }, "anthropic", "TAG"), null); +}); + +test("#5432 chain-checkpoint negative: assistant-final still gets the appended carrier (legacy behavior)", () => { + const out = insertCheckpointCarrier({ + model: "m", + messages: [{ role: "user", content: "q" }, { role: "assistant", content: "a" }], + }, "anthropic", "TAG")! as { messages: Msg[] }; + assert.deepEqual(out.messages.map((m) => m.role), ["user", "assistant", "user"]); + assert.equal(out.messages[2]!.content, "TAG"); +}); + +// --- real steady path (hermetic server harness) ---------------------------- + +const WINDOW = 15_000; +const MODEL = "claude-test"; +const SYSTEM = "You are a coding agent."; +// Distinctive slice of ANTHROPIC_LANE_NOTE — unique to the anthropic-lane nudge. +const NOTE = "Write the summary ONLY as the `compress` tool's `content` argument"; +// Upstream-reported usage crossing the OVER-LIMIT band (90% of WINDOW) while +// the LOCAL estimate of the fixture stays well under it — so turn 1 is clean +// and only the recorded usage arms the turn-2 nudge (mirrors #3910). +const ARMED_INPUT_TOKENS = 13_500; +// Claude Code's trailing reminder shape: the client's LAST message is a +// mid-conversation system (legal at array end pre-injection). +const TRAILING_SYSTEM_TEXT = "The user hasn't heard from you in a while; continue working on the task."; + +const notesIn = (s: string): number => s.split(NOTE).length - 1; + +function filler(seed: number, targetChars: number): string { + const para = `Paragraph ${seed}: the ledger module under review passed its integration suite across all four regions without any reported regressions. `; + const unit = Math.max(1, Math.ceil(targetChars / para.length)); + return Array.from({ length: unit }, (_, i) => para.replace(String(seed), `${seed}-${i}`)).join(""); +} + +function clientMessages(withTrailingSystem: boolean): Array<{ role: string; content: Array<{ type: string; text: string }> }> { + const out: Array<{ role: string; content: Array<{ type: string; text: string }> }> = []; + for (let i = 1; i <= 7; i++) { + out.push({ role: i % 2 === 1 ? "user" : "assistant", content: [{ type: "text", text: `Turn ${i}: working session entry. ` + filler(i, 4400) }] }); + } + if (withTrailingSystem) { + out.push({ role: "system", content: [{ type: "text", text: TRAILING_SYSTEM_TEXT }] }); + } + return out; +} + +const anthropicSse = (event: string, data: unknown): string => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; + +const anthropicReply = (): string => + [ + anthropicSse("message_start", { type: "message_start", message: { id: "msg_r", role: "assistant", usage: { input_tokens: ARMED_INPUT_TOKENS } } }), + anthropicSse("content_block_start", { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }), + anthropicSse("content_block_delta", { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "ack" } }), + anthropicSse("content_block_stop", { type: "content_block_stop", index: 0 }), + anthropicSse("message_delta", { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 3 } }), + anthropicSse("message_stop", { type: "message_stop" }), + ].join(""); + +interface Harness { + origin: string; + upstreamUrl: string; + captured: string[]; + close(): Promise; +} + +async function startHarness(): Promise { + const captured: string[] = []; + const upstream = http.createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + captured.push(Buffer.concat(chunks).toString("utf8")); + res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-cache" }); + res.end(anthropicReply()); + }); + }); + upstream.listen(0, "127.0.0.1"); + await once(upstream, "listening"); + const upstreamUrl = `http://127.0.0.1:${(upstream.address() as { port: number }).port}`; + + _setStoreForTest(new SessionStore({ enabled: false })); + setRegistryForTest({}); + const proxy = await startServer({ + port: 0, + host: "127.0.0.1", + upstream: upstreamUrl, + routes: { [upstreamUrl]: { models: { [MODEL]: { context: WINDOW } } } }, + modelContextLimit: WINDOW, + // Sub-floor fixture: pin legacy any-pending semantics (#3006) so the + // escalation override keeps firing instead of being vetoed by the + // minimum-benefit floor. + kernelConfig: { ...defaultConfig(WINDOW), nudge: { ...defaultConfig(WINDOW).nudge, minPressureBenefitTokens: 0 } }, + compress: { injectTool: true, injectNudge: true }, + promptCache: { routing: "auto" }, + sessionHeader: "x-acp-session", + log: false, + debug: false, + passthrough: false, + autoUpdate: false, + mitm: { enabled: false, domains: [] }, + } as ProxyOptions); + await once(proxy, "listening"); + const origin = `http://127.0.0.1:${(proxy.address() as { port: number }).port}`; + + return { + origin, + upstreamUrl, + captured, + close: async () => { + proxy.close(); + proxy.closeIdleConnections?.(); + await once(proxy, "close"); + upstream.close(); + upstream.closeIdleConnections?.(); + await once(upstream, "close"); + }, + }; +} + +test("#5432 steady path: nudge inserts BEFORE the client's trailing system, which stays verbatim and last", async () => { + const h = await startHarness(); + try { + const url = `${h.origin}/bili/${h.upstreamUrl}/v1/messages`; + const post = (session: string) => fetch(url, { + method: "POST", + headers: { "content-type": "application/json", "x-acp-session": session }, + body: JSON.stringify({ model: MODEL, max_tokens: 1024, stream: true, system: SYSTEM, messages: clientMessages(true) }), + }); + + const r1 = await post("issue5432-anth-a"); + assert.equal(r1.status, 200); + await r1.text(); + const r2 = await post("issue5432-anth-a"); + assert.equal(r2.status, 200); + await r2.text(); + + assert.ok(h.captured.length >= 2, `expected an arm + inject pair, got ${h.captured.length} upstream bodies`); + assert.equal(notesIn(h.captured[0]!), 0, "turn 1 carries no nudge (fresh session, local estimate under the band)"); + + const body = JSON.parse(h.captured[1]!) as { messages: Msg[] }; + assert.equal(body.messages.length, clientMessages(true).length + 1, "exactly ONE ephemeral message added"); + const last = body.messages[body.messages.length - 1]!; + assert.equal(last.role, "system", "the client's trailing system must STILL end the outbound array (the #5432 case-2 shape)"); + // Not byte-equal: the steady path renders every message text through + // the ACP tag strategy (KDD#2); the client's reminder text must survive + // intact inside that rendering (undemoted, unsplit). + assert.ok(textOf(last).includes(TRAILING_SYSTEM_TEXT), `trailing system keeps the client's text (got ${JSON.stringify(textOf(last).slice(0, 120))})`); + const secondLast = body.messages[body.messages.length - 2]!; + assert.equal(secondLast.role, "user", "the nudge lands immediately before the trailing system"); + assert.equal(notesIn(textOf(secondLast)), 1, "the lane note appears exactly once, in that preceding user message"); + } finally { + await h.close(); + } +}); + +test("#5432 steady path negative: no trailing system → nudge still lands dead-last (legacy behavior unchanged)", async () => { + const h = await startHarness(); + try { + const url = `${h.origin}/bili/${h.upstreamUrl}/v1/messages`; + const post = (session: string) => fetch(url, { + method: "POST", + headers: { "content-type": "application/json", "x-acp-session": session }, + body: JSON.stringify({ model: MODEL, max_tokens: 1024, stream: true, system: SYSTEM, messages: clientMessages(false) }), + }); + + const r1 = await post("issue5432-anth-b"); + assert.equal(r1.status, 200); + await r1.text(); + const r2 = await post("issue5432-anth-b"); + assert.equal(r2.status, 200); + await r2.text(); + + assert.ok(h.captured.length >= 2, `expected an arm + inject pair, got ${h.captured.length} upstream bodies`); + const body = JSON.parse(h.captured[1]!) as { messages: Msg[] }; + assert.equal(body.messages.length, clientMessages(false).length + 1, "exactly ONE ephemeral message added"); + const last = body.messages[body.messages.length - 1]!; + assert.equal(last.role, "user", "without a trailing system the nudge keeps landing as the final message"); + assert.equal(notesIn(textOf(last)), 1, "the lane note appears exactly once, in the trailing nudge message"); + } finally { + await h.close(); + } +});