From da1ab44193d86bb7756c13f917d152d5f15beffe Mon Sep 17 00:00:00 2001 From: ework-agent Date: Thu, 10 Sep 2026 07:31:03 +0800 Subject: [PATCH] fix: tolerate typo'd render-tag names in echo strip; warn on degenerate end_turn turns Follow-up to #644/#646: a model-emitted render-tag echo with a typo'd tag name (acpi/acip/apic-style variants of acp) bypassed every exact-spelling pattern in the tag-echo filter and passed through as the turn's entire visible text - a 43-char degenerate end_turn turn with zero tool calls that stalls the agent mid-orchestration until manually nudged (#673). - tag-echo-filter: all patterns now match acplike names (permutations of acp plus one inserted a/c/p/i char), still requiring a name boundary so legit text (acpi/acpi.h includes, caption, app, ACPI_DEVICE) is untouched; swallow-until-close scans loose close variants; add lifetime stats() for turn-level accounting - adapters (anthropic/openai/responses): warn [degenerate-turn] when a terminal turn ends with zero visible text and zero tool calls (thinking may be non-empty); responses RENDER_TAG_RE loosened to match - plugin passthrough pipes (openai chat / anthropic / responses): same degenerate-turn warn with per-field visible-text accounting (thinking and reasoning fields never count as visible) - tests: split-parity typo cases, gate engagement, safe-word survival, stats accounting, warning matrix, adapter e2e, pipe-level warn/no-warn --- src/degenerate-turn.ts | 37 +++++ src/loop/adapter-anthropic.ts | 20 ++- src/loop/adapter-openai.ts | 19 +++ src/loop/adapter-responses.ts | 20 ++- src/loop/tag-echo-filter.ts | 148 +++++++++++++----- src/plugin.ts | 93 ++++++++++- .../plugin-passthrough-tag-strip-chat.test.ts | 107 +++++++++++++ tests/tag-echo.test.ts | 97 ++++++++++++ 8 files changed, 493 insertions(+), 48 deletions(-) create mode 100644 src/degenerate-turn.ts diff --git a/src/degenerate-turn.ts b/src/degenerate-turn.ts new file mode 100644 index 00000000..ad7f0fd2 --- /dev/null +++ b/src/degenerate-turn.ts @@ -0,0 +1,37 @@ +import type { TagEchoFilterStats } from "./loop/tag-echo-filter.js"; + +// #673: detection of degenerate terminal turns — upstream finishes normally +// (end_turn / stop / completed) but the client receives zero visible text and +// zero tool calls. Observed cause: the turn's only text was a render-tag echo +// the stripper missed (typo'd tag name), so the agent receives an empty turn +// mid-orchestration and stalls until manually nudged. The condition is +// deliberately recall-first: a legitimately empty text-only terminal turn is +// rare and equally confusing to an agentic client, so it warns too. Callers +// gate on their wire's own state (reason seen, tool calls emitted, thinking +// presence) and feed the filter's lifetime text accounting. +export interface TurnOutcome { + /** Wire-native reason observed (stop_reason / finish_reason / status), if any. */ + reason: string | undefined; + /** The wire's normal-completion reason ("end_turn" / "stop" / "completed"). */ + terminalReason: string; + toolCalls: number; + /** Lifetime visible-text accounting (summed across text fields/blocks). */ + text: TagEchoFilterStats; + sawThinking: boolean; + /** Wire label for logs, e.g. "anthropic" or "plugin-passthrough-openai". */ + wire: string; +} + +export function degenerateTurnWarning(o: TurnOutcome): string | null { + if ((o.reason ?? o.terminalReason) !== o.terminalReason) return null; + if (o.toolCalls > 0) return null; + if (o.text.outputChars > 0) return null; + const bits: string[] = []; + if (o.sawThinking) bits.push("thinking present"); + if (o.text.dropped && o.text.inputChars > 0) bits.push(`${o.text.inputChars} chars of emitted text stripped as render-tag echo`); + else bits.push("no visible text emitted"); + return ( + `[degenerate-turn] ${o.wire}: turn ended ${o.terminalReason} with zero visible text and zero tool calls (${bits.join("; ")}) ` + + `— the agent receives an empty turn and may stall until nudged (#673)` + ); +} diff --git a/src/loop/adapter-anthropic.ts b/src/loop/adapter-anthropic.ts index 0ac26cea..dd13f09a 100644 --- a/src/loop/adapter-anthropic.ts +++ b/src/loop/adapter-anthropic.ts @@ -2,6 +2,7 @@ import type { CoreMessage } from "acp-kernel"; import { coreToAnthropic, extractSystem, buildSystem, type AnthropicRequestBody } from "acp-kernel/wire"; import { buildVisibilityMarker } from "../compress-loop.js"; import { createTagEchoFilter } from "./tag-echo-filter.js"; +import { degenerateTurnWarning } from "../degenerate-turn.js"; import { log as loggerLog } from "../logger.js"; import type { CompressLoopAdapter, @@ -222,6 +223,17 @@ export function createAnthropicAdapter(requestBody: Record, ori loggerLog("warn", `[tag-echo] stripped model-emitted render tag: ${snippet.slice(0, 80).replace(/\n/g, " ")}`); }); let lastTextIndex: number | null = null; + let sawThinking = false; + let toolCallsEmitted = 0; + let degenerateWarned = false; + const maybeWarnDegenerate = (reason: string | undefined) => { + if (degenerateWarned) return; + const msg = degenerateTurnWarning({ reason, terminalReason: "end_turn", toolCalls: toolCallsEmitted, text: tagFilter.stats(), sawThinking, wire: "anthropic" }); + if (msg) { + degenerateWarned = true; + loggerLog("warn", msg); + } + }; for await (const eventStr of iterSseEvents(upstream)) { const parsed = parseAnthropicSse(eventStr); @@ -267,7 +279,10 @@ export function createAnthropicAdapter(requestBody: Record, ori const id = typeof block.id === "string" ? block.id : `toolu_${upstreamIndex}`; pending.set(upstreamIndex, { id, name, json: "" }); } else { - if (block.type === "thinking" || block.type === "redacted_thinking") thinkingIndexes.add(upstreamIndex); + if (block.type === "thinking" || block.type === "redacted_thinking") { + thinkingIndexes.add(upstreamIndex); + sawThinking = true; + } const ci = clientIndex++; indexMap.set(upstreamIndex, ci); openBlocks.push(ci); @@ -312,6 +327,7 @@ export function createAnthropicAdapter(requestBody: Record, ori const tb = pending.get(upstreamIndex); if (tb) { pending.delete(upstreamIndex); + toolCallsEmitted++; yield { kind: "tool_call", name: tb.name, @@ -376,6 +392,7 @@ export function createAnthropicAdapter(requestBody: Record, ori cachedTokens: roundCached, } as ParsedStreamEvent; } + maybeWarnDegenerate(stopReason); yield { kind: "done", finishReason: stopReason } as ParsedStreamEvent; } else if (type === "message_stop") { if (lastTextIndex !== null) { @@ -394,6 +411,7 @@ export function createAnthropicAdapter(requestBody: Record, ori cachedTokens: roundCached, } as ParsedStreamEvent; } + maybeWarnDegenerate(stopReason); yield { kind: "done", finishReason: stopReason ?? "end_turn" } as ParsedStreamEvent; } else if (round === 1) { yield { kind: "meta", chunk: rawBuf, firstRoundOnly: true } as ParsedStreamEvent; diff --git a/src/loop/adapter-openai.ts b/src/loop/adapter-openai.ts index f0aef231..8f5f60da 100644 --- a/src/loop/adapter-openai.ts +++ b/src/loop/adapter-openai.ts @@ -2,6 +2,7 @@ import type { CoreMessage } from "acp-kernel"; import { coreToOpenai, injectOpenaiSystem } from "acp-kernel/wire"; import { buildVisibilityMarker } from "../compress-loop.js"; import { createTagEchoFilter } from "./tag-echo-filter.js"; +import { degenerateTurnWarning } from "../degenerate-turn.js"; import { log as loggerLog } from "../logger.js"; import { systemToUser } from "../util.js"; @@ -229,6 +230,17 @@ export function createOpenaiAdapter(requestBody: Record, client yield { kind: "text", delta: tail, raw: buildContent(tail) } as ParsedStreamEvent; } }; + let sawReasoning = false; + let toolCallsEmitted = 0; + let degenerateWarned = false; + const maybeWarnDegenerate = (reason: string | undefined) => { + if (degenerateWarned) return; + const msg = degenerateTurnWarning({ reason, terminalReason: "stop", toolCalls: toolCallsEmitted, text: tagFilter.stats(), sawThinking: sawReasoning, wire: "openai" }); + if (msg) { + degenerateWarned = true; + loggerLog("warn", msg); + } + }; // Raw tool_call chunks in arrival order. Backends (SGLang/vLLM) // stream a tool name across MULTIPLE deltas — the first fragment // carries the name, continuation fragments carry empty names. @@ -241,6 +253,7 @@ export function createOpenaiAdapter(requestBody: Record, client const flushPendingAsStructured = function* (): Generator { for (const [, tc] of pending) { if (tc.name.length > 0 || tc.id.length > 0) { + toolCallsEmitted++; yield { kind: "tool_call", name: tc.name, @@ -269,6 +282,7 @@ export function createOpenaiAdapter(requestBody: Record, client } for (const [idx, tc] of pending) { if (realIndexes.has(idx) && (tc.name.length > 0 || tc.id.length > 0)) { + toolCallsEmitted++; yield { kind: "tool_call", name: tc.name, @@ -288,6 +302,7 @@ export function createOpenaiAdapter(requestBody: Record, client } for (const [idx, tc] of pending) { if (!realIndexes.has(idx) && tc.name.length > 0) { + toolCallsEmitted++; yield { kind: "tool_call", name: tc.name, @@ -312,6 +327,7 @@ export function createOpenaiAdapter(requestBody: Record, client if (sawRealToolCall) { yield { kind: "meta", chunk: Buffer.from(eventStr + "\n\n", "utf8") } as ParsedStreamEvent; } + maybeWarnDegenerate("stop"); yield { kind: "done", finishReason: "stop", ...(sawRealToolCall ? { suppressCompletion: true } : {}) } as ParsedStreamEvent; continue; } @@ -370,9 +386,11 @@ export function createOpenaiAdapter(requestBody: Record, client // to the text/reasoning branches (which would re-emit the same // bytes after the finish reason). yield { kind: "meta", chunk } as ParsedStreamEvent; + maybeWarnDegenerate(finishReason); yield { kind: "done", finishReason, suppressCompletion: true } as ParsedStreamEvent; continue; } else { + maybeWarnDegenerate(finishReason); yield { kind: "done", finishReason: hadToolCalls && finishReason === "stop" ? "tool_calls" : finishReason, @@ -383,6 +401,7 @@ export function createOpenaiAdapter(requestBody: Record, client if (!delta) continue; if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) { + sawReasoning = true; yield { kind: "reasoning", delta: delta.reasoning_content, raw: finishReason ? stripFinishReasonChunk(rawBuf) : rawBuf } as ParsedStreamEvent; } diff --git a/src/loop/adapter-responses.ts b/src/loop/adapter-responses.ts index 03a89592..a6e4e3df 100644 --- a/src/loop/adapter-responses.ts +++ b/src/loop/adapter-responses.ts @@ -2,7 +2,8 @@ import type { CoreMessage } from "acp-kernel"; import { coreToResponses, injectResponsesDeveloperMessage, patchResponsesInput, type ResponseInputItem, type ResponsesProjection } from "acp-kernel/wire"; import { buildVisibilityMarker } from "../compress-loop.js"; import { hashId } from "../util.js"; -import { createTagEchoFilter, stripResponsesText, containsRenderTagText } from "./tag-echo-filter.js"; +import { createTagEchoFilter, stripResponsesText, containsRenderTagText, ACP_NAME_ALT } from "./tag-echo-filter.js"; +import { degenerateTurnWarning } from "../degenerate-turn.js"; import { log as loggerLog } from "../logger.js"; import { ACP_TEXT_OPEN, ACP_TEXT_CLOSE, ACP_STATUS_OPEN, ACP_STATUS_CLOSE, ACP_SEARCH_OPEN, ACP_SEARCH_CLOSE, ACP_DECOMPRESS_OPEN, ACP_DECOMPRESS_CLOSE, COMPRESS_TOOL_NAME, PROXY_TOOL_NAMES } from "../compress-tool.js"; import type { BiliMessage } from "acp-kernel/wire"; @@ -117,7 +118,7 @@ export function dropWhitespaceResponsesMessages(input: unknown): number { return dropped; } -const RENDER_TAG_RE = /\x3cacp\s[^>]*\x3e[^<]*\x3c\/acp\x3e|\x3cacp\s[^>]*\/\x3e/g; +const RENDER_TAG_RE = new RegExp("\x3c" + ACP_NAME_ALT + "\\s[^\x3e]*\x3e[^\x3c]*\x3c\\/" + ACP_NAME_ALT + "\x3e|\x3c" + ACP_NAME_ALT + "\\s[^\x3e]*\\/\x3e", "g"); function stripRenderTags(text: string): string { return text.replace(RENDER_TAG_RE, ""); @@ -316,6 +317,17 @@ export function createResponsesAdapter(textProtocol?: boolean, projection?: Resp yield { kind: "text", delta: tail, ...(raw ? { raw } : {}) } as ParsedStreamEvent; } }; + let sawReasoning = false; + let toolCallsEmitted = 0; + let degenerateWarned = false; + const maybeWarnDegenerate = (reason: string | undefined) => { + if (degenerateWarned) return; + const msg = degenerateTurnWarning({ reason, terminalReason: "completed", toolCalls: toolCallsEmitted, text: tagFilter.stats(), sawThinking: sawReasoning, wire: "responses" }); + if (msg) { + degenerateWarned = true; + loggerLog("warn", msg); + } + }; for await (const eventStr of iterSseEvents(upstream)) { const explicitType = extractEventType(eventStr); const dataLine = extractDataLine(eventStr); @@ -332,6 +344,7 @@ export function createResponsesAdapter(textProtocol?: boolean, projection?: Resp // back to it when no event: line is present (explicit line still wins). const type = explicitType ?? (typeof obj.type === "string" ? obj.type : null); if (!type) continue; + if (type.startsWith("response.reasoning")) sawReasoning = true; const rawBuf = Buffer.from(eventStr + "\n\n", "utf8"); if (round === 1 && typeof obj.output_index === "number") { outputIndex = Math.max(outputIndex, obj.output_index + 1); @@ -422,6 +435,7 @@ export function createResponsesAdapter(textProtocol?: boolean, projection?: Resp if (fc) { if (typeof item.arguments === "string" && item.arguments) fc.arguments = item.arguments; pending.delete(itemId); + toolCallsEmitted++; yield { kind: "tool_call", name: fc.name, @@ -430,6 +444,7 @@ export function createResponsesAdapter(textProtocol?: boolean, projection?: Resp } as ParsedStreamEvent; } else { yield { kind: "meta", chunk: rawBuf, firstRoundOnly: false } as ParsedStreamEvent; + toolCallsEmitted++; yield { kind: "tool_call", name: typeof item.name === "string" ? item.name : "", @@ -468,6 +483,7 @@ export function createResponsesAdapter(textProtocol?: boolean, projection?: Resp outputTokens: typeof respUsage?.output_tokens === "number" ? respUsage.output_tokens : undefined, cachedTokens: typeof pd?.cached_tokens === "number" ? pd.cached_tokens : undefined, } as ParsedStreamEvent; + maybeWarnDegenerate("completed"); yield { kind: "done", finishReason: "completed" } as ParsedStreamEvent; } else if (type === "response.incomplete") { yield* flushFilter(); diff --git a/src/loop/tag-echo-filter.ts b/src/loop/tag-echo-filter.ts index 3743838d..a91e48d4 100644 --- a/src/loop/tag-echo-filter.ts +++ b/src/loop/tag-echo-filter.ts @@ -3,39 +3,90 @@ // imitate them in visible output ("tag echo"), the client replays the echoed // tags on later turns, and the imitation amplifies into unbounded repetition. // Stripping render tags from outgoing text breaks the loop at the source. -// ONLY the render form (`\x3cacp attrs…>` / exact `\x3c/acp>`) is stripped — the -// underscore-namespaced text-protocol triggers (`\x3cacp_compress>` etc.) and -// ordinary prose containing `<` pass through untouched. +// ONLY the render form (\x3c attrs…\x3e, \x3c\x3e) is stripped — +// the underscore-namespaced text-protocol triggers (\x3cacp_compress\x3e etc.) and +// ordinary prose containing \x3c pass through untouched. +// #673: models typo the 3-letter name when imitating (observed: acip/acpi, +// including mixed correct-open + typo'd-close), so matches a bounded +// mutation set instead of the exact spelling: the three core letters in any +// order, plus at most ONE extra letter drawn from that set or an inserted i. +// Every match still requires the name to be followed by \s or > (attrs or +// close), so real words that merely contain the letters (acpi/acpi.h includes, +// caption, app, uppercase ACPI) never match; a false positive costs at most +// the same bounded caps as before (swallow ≤ SWALLOW_CAP, hold ≤ HOLD_LIMIT/TAG_OPEN_CAP). +function buildAcplikeName(): string { + const cores = ["acp", "apc", "cap", "cpa", "pac", "pca"]; + const names = new Set(cores); + for (const c of cores) { + for (const ch of ["a", "c", "p", "i"]) { + for (let pos = 0; pos <= c.length; pos++) names.add(c.slice(0, pos) + ch + c.slice(pos)); + } + } + return [...names].sort((a, b) => b.length - a.length).join("|"); +} + +/** Longest-first alternation of every tolerated render-tag name (#673). */ +export const ACP_NAME_ALT = `(?:${buildAcplikeName()})`; +const NAME = ACP_NAME_ALT; // Opening-tag attrs are bounded: a render tag opening is short (tokens + type, -// < 50 chars). An unbounded \x3cacp …\x3e match would swallow a long prose span -// that merely starts with \x3cacp and contains a \x3e somewhere later. -const PAIRED = /\x3cacp\s[^<>]{0,256}>([^<>]{0,64})<\/acp>/; -const LONE_OPEN = /\x3cacp(?:\s[^<>]{0,256})?>/; -const LONE_CLOSE = /<\/acp(?=[\s>])[^<>]{0,32}>/; +// \x3c 50 chars). An unbounded \x3c …\x3e match would swallow a long prose span +// that merely starts with a tag head and contains a \x3e somewhere later. +const PAIRED = new RegExp("\x3c" + NAME + "\\s[^<>]{0,256}>([^<>]{0,64})\x3c\\/" + NAME + ">"); +const LONE_OPEN = new RegExp("\x3c" + NAME + "(?:\\s[^<>]{0,256})?>"); +const LONE_CLOSE = new RegExp("\x3c\\/" + NAME + "(?=[\\s>])[^<>]{0,32}>"); // A suffix of the buffer that could still grow into a render tag: either an -// unterminated `\x3cacp …` opening (attrs so far, no `>` yet) or a short -// ambiguous prefix like `<`, `]*|\x3c\/acp(?:\s[^<>]{0,32})?|<\/?a?c?p?)$/; -// An unterminated render-tag opening at the end of a string: `` — a truncated imitation, never prose (triggers use `]*$/; -// A truncated render-tag CLOSE at the end of a string: `` or `` — -// a truncated imitation close, never prose. Mirrors TRUNC_OPEN on the close side. -const TRUNC_CLOSE = /\x3c\/acp(?:\s[^<>]{0,32})?$/; -const CLOSE_TAG = "\x3c/acp"; +// unterminated \x3c … opening (attrs so far, no \x3e yet) or a short +// ambiguous prefix like \x3c, \x3ca, \x3c/ac, \x3cacip, … +const PARTIAL_TAIL = new RegExp("(\x3c" + NAME + "\\s[^<>]*|\x3c\\/" + NAME + "(?:\\s[^<>]{0,32})?|\x3c\\/?[acip]*)$"); +// An unterminated render-tag opening at the end of a string: \x3c plus +// attrs, no \x3e — a truncated imitation, never prose (triggers use \x3cacp_). +const TRUNC_OPEN = new RegExp("\x3c" + NAME + "\\s[^<>]*$"); +// A truncated render-tag CLOSE at the end of a string: \x3c/ optionally +// plus truncated attrs — a truncated imitation close, never prose. Mirrors +// TRUNC_OPEN on the close side. +const TRUNC_CLOSE = new RegExp("\x3c\\/" + NAME + "(?:\\s[^<>]{0,32})?$"); +const DEFINITE_TAIL = new RegExp("^\x3c" + NAME + "\\s|^\x3c\\/" + NAME); +const OPEN_WITH_ATTRS = new RegExp("^\x3c" + NAME + "\\s"); +const CLOSE_HEAD = "\x3c/"; +const CLOSE_NAME_ANCHORED = new RegExp("^" + NAME); const HOLD_LIMIT = 128; // Hold cap for a definite unterminated opening tail — far beyond any real tag // opening; beyond this the tail is dropped instead of held or passed through. const TAG_OPEN_CAP = 4096; const SWALLOW_CAP = 80; +/** Exclusive end index (past the terminating \x3e) of the first loose close + * tag in s, or -1. #673: the close name may be a typo variant; termination + * still requires the strict \x3e right after the name — malformed closes are + * LONE_CLOSE's job, not the swallow terminator's. */ +function looseCloseEnd(s: string): number { + let idx = s.indexOf(CLOSE_HEAD); + while (idx >= 0) { + const m = CLOSE_NAME_ANCHORED.exec(s.slice(idx + 2)); + if (m && s[idx + 2 + m[0].length] === ">") return idx + 2 + m[0].length + 1; + idx = s.indexOf(CLOSE_HEAD, idx + 1); + } + return -1; +} + +export interface TagEchoFilterStats { + /** Raw chars pushed over the filter's lifetime (before stripping). */ + inputChars: number; + /** Clean chars emitted (push outputs + flush output). */ + outputChars: number; + /** Whether anything was dropped as an imitation. */ + dropped: boolean; +} + export interface TagEchoFilter { push(delta: string): string; flush(): string; dropped(): boolean; /** True while the filter holds a partial-tag tail that a later push may complete. */ pending(): boolean; + /** Lifetime accounting — feeds degenerate-turn detection (#673). Deliberately not reset by intermediate flushes. */ + stats(): TagEchoFilterStats; } export function stripAcpTags(text: string): string { @@ -49,9 +100,9 @@ export function stripAcpTags(text: string): string { // Cheap pre-check on a raw wire string (SSE event or JSON body): does it // contain anything that looks like a render tag (literal or JSON-escaped -// `\u003c` form)? Callers use this to skip re-serializing chunks that need +// \u003c form)? Callers use this to skip re-serializing chunks that need // no stripping, preserving byte-identical passthrough. -const RENDER_TAG_DETECT = /\x3c\/?acp(?=[\s>])|\\u003c\/?acp(?=[\s>\\])/; +const RENDER_TAG_DETECT = new RegExp("\x3c\\/?" + NAME + "(?=[\\s>])|\\\\u003c\\/?" + NAME + "(?=[\\s>\\\\])"); export function containsRenderTagText(s: string): boolean { return RENDER_TAG_DETECT.test(s); } @@ -61,8 +112,8 @@ export function containsRenderTagText(s: string): boolean { // RENDER_TAG_DETECT. Per-chunk gates must also engage when the chunk contains // or ends with the head of a render tag, so the streaming state machine can // stitch it back together. Pure-prose chunks still skip the machine -// (byte-identical passthrough); only chunks with a `<`-head tail ("<", " void): TagEcho let swallowed = ""; let droppedAny = false; let notified = false; + let inputChars = 0; + let outputChars = 0; const drop = (snippet: string) => { droppedAny = true; if (onDrop && !notified) { @@ -96,10 +149,9 @@ export function createTagEchoFilter(onDrop?: (snippet: string) => void): TagEcho for (;;) { if (swallowUntilClose) { const combined = swallowed + buf; - const closeIdx = combined.indexOf(CLOSE_TAG); - if (closeIdx >= 0 && combined[closeIdx + CLOSE_TAG.length] === ">") { - const end = closeIdx + CLOSE_TAG.length + 1; - drop(swallowed + combined.slice(0, end)); + const end = looseCloseEnd(combined); + if (end >= 0) { + drop(combined.slice(0, end)); swallowed = ""; swallowUntilClose = false; buf = combined.slice(end); @@ -124,11 +176,11 @@ export function createTagEchoFilter(onDrop?: (snippet: string) => void): TagEcho if (!m) { const t = PARTIAL_TAIL.exec(buf); if (t) { - // A definite \x3cacp opening is never prose — hold it far + // A definite \x3c opening is never prose — hold it far // past HOLD_LIMIT (drop it past TAG_OPEN_CAP); a short // ambiguous prefix stays on the small hold cap so prose // is never delayed or lost. - const definite = /^\x3cacp\s/.test(t[0]) || /^\x3c\/acp/.test(t[0]); + const definite = DEFINITE_TAIL.test(t[0]); const cap = definite ? TAG_OPEN_CAP : HOLD_LIMIT; if (t[0].length <= cap) { held = t[0]; @@ -150,7 +202,7 @@ export function createTagEchoFilter(onDrop?: (snippet: string) => void): TagEcho // A PAIRED match is by definition a complete open+content+close // span — only an attrs-bearing LONE_OPEN leaves the stream // mid-tag and needs to swallow until its close arrives. - if (m === o && /^\x3cacp\s/.test(m[0])) { + if (m === o && OPEN_WITH_ATTRS.test(m[0])) { swallowUntilClose = true; swallowed = ""; } @@ -159,9 +211,12 @@ export function createTagEchoFilter(onDrop?: (snippet: string) => void): TagEcho }; return { push(delta: string): string { + inputChars += delta.length; const chunk = held + delta; held = ""; - return process(chunk); + const r = process(chunk); + outputChars += r.length; + return r; }, flush(): string { const rest = swallowed + held; @@ -169,23 +224,29 @@ export function createTagEchoFilter(onDrop?: (snippet: string) => void): TagEcho swallowed = ""; held = ""; swallowUntilClose = false; + let result: string; if (wasSwallowing) { // Stream ended inside an unclosed render tag: the held content // is tag content (a ref), not prose. if (rest.length > 0) drop(rest); - return ""; - } - const t = new RegExp(TRUNC_OPEN.source).exec(rest); - if (t) { - drop(t[0]); - return rest.slice(0, t.index); - } - const tc = new RegExp(TRUNC_CLOSE.source).exec(rest); - if (tc) { - drop(tc[0]); - return rest.slice(0, tc.index); + result = ""; + } else { + const t = new RegExp(TRUNC_OPEN.source).exec(rest); + if (t) { + drop(t[0]); + result = rest.slice(0, t.index); + } else { + const tc = new RegExp(TRUNC_CLOSE.source).exec(rest); + if (tc) { + drop(tc[0]); + result = rest.slice(0, tc.index); + } else { + result = rest; + } + } } - return rest; + outputChars += result.length; + return result; }, dropped(): boolean { return droppedAny; @@ -193,6 +254,9 @@ export function createTagEchoFilter(onDrop?: (snippet: string) => void): TagEcho pending(): boolean { return held.length > 0 || swallowUntilClose; }, + stats(): TagEchoFilterStats { + return { inputChars, outputChars, dropped: droppedAny }; + }, }; } diff --git a/src/plugin.ts b/src/plugin.ts index 8a62ba62..cfd5d4ca 100644 --- a/src/plugin.ts +++ b/src/plugin.ts @@ -10,6 +10,7 @@ import { executeProxyTool } from "./loop/core.js"; import { normalizeSseLineEndings } from "./sse-util.js"; import { containsRenderTagText, createTagEchoFilter, mayStartRenderTag, stripAnthropicText, stripOpenaiChatText, stripResponsesText, type TagEchoFilter } from "./loop/tag-echo-filter.js"; import { log as loggerLog } from "./logger.js"; +import { degenerateTurnWarning } from "./degenerate-turn.js"; import { noteWeakOverflow } from "./weak-overflow.js"; import { warnCacheCollapse } from "./cache-warn.js"; import { backfillHostUsage, promptInputTotal, type WireProtocol } from "./util.js"; @@ -696,11 +697,19 @@ export async function pipePluginChatWithStrip( } return `data: ${JSON.stringify({ ...lastChunkMeta, object: "chat.completion.chunk", choices: [{ index, delta: { [field]: tail } }] })}\n\n`; }; + // #673: turn-level observability for degenerate terminal turns. + let sawToolUse = false; + let sawThinking = false; + let visibleTextChars = 0; + let finalFinishReason: string | undefined; const flushTails = (): string => { let out = ""; for (const s of streams.values()) { const tail = s.filter.flush(); - if (tail.length > 0) out += syntheticTail(s.field, s.index, tail); + if (tail.length > 0) { + out += syntheticTail(s.field, s.index, tail); + if (s.field === "content" || s.field === "text") visibleTextChars += tail.length; + } } return out; }; @@ -730,6 +739,28 @@ export async function pipePluginChatWithStrip( reason: "plugin chat passthrough stream ended without a completion event", }); }; + const maybeWarnDegenerate = () => { + if (!sawTerminal || res.destroyed || res.writableEnded) return; + let inputChars = 0; + let dropped = false; + for (const s of streams.values()) { + const st = s.filter.stats(); + inputChars += st.inputChars; + dropped = dropped || st.dropped; + } + const msg = degenerateTurnWarning({ + reason: finalFinishReason, + terminalReason: protocol === "anthropic" ? "end_turn" : "stop", + toolCalls: sawToolUse ? 1 : 0, + text: { inputChars, outputChars: visibleTextChars, dropped }, + sawThinking, + wire: `plugin-passthrough-${protocol}`, + }); + if (msg) { + loggerLog("warn", msg); + log?.(msg); + } + }; const pushField = (field: string, index: number, text: string): [string, boolean] => { const s = filterFor(field, index); const clean = s.filter.push(text); @@ -749,21 +780,28 @@ export async function pipePluginChatWithStrip( let hadText = false; for (let ci = 0; ci < choices.length; ci++) { const ch = choices[ci] as Record | null; + if (ch && typeof ch["finish_reason"] === "string") finalFinishReason = ch["finish_reason"] as string; const d = ch?.["delta"]; if (!d || typeof d !== "object") continue; const dd = d as Record; + if (dd["tool_calls"] !== undefined) sawToolUse = true; for (const field of ["content", "reasoning_content", "reasoning"]) { const v = dd[field]; if (typeof v !== "string") continue; hadText = true; + if (field !== "content" && v.length > 0) sawThinking = true; if (!mayStartRenderTag(v) && !anyPending()) { if (v.length > 0) keptText = true; + if (field === "content") visibleTextChars += v.length; continue; } const index = typeof ch?.["index"] === "number" ? ch["index"] : ci; const [clean, changed] = pushField(field, index, v); if (clean.length === 0) droppedText = true; - else keptText = true; + else { + keptText = true; + if (field === "content") visibleTextChars += clean.length; + } if (changed) { if (!rebuilt) { rebuilt = { ...ev, choices: choices.map((c) => ({ ...(c as Record), delta: { ...((c as Record)["delta"] as Record) } })) }; @@ -786,6 +824,13 @@ export async function pipePluginChatWithStrip( return rawEvent + "\n\n"; }; const processAnthropic = (ev: Record, rawEvent: string): string => { + if (ev["type"] === "content_block_start") { + const cb = ev["content_block"] as Record | undefined; + const bt = cb && typeof cb === "object" ? cb["type"] : undefined; + if (bt === "tool_use") sawToolUse = true; + else if (bt === "thinking" || bt === "redacted_thinking") sawThinking = true; + return anyPending() ? flushTails() + rawEvent + "\n\n" : rawEvent + "\n\n"; + } if (ev["type"] !== "content_block_delta") { return anyPending() ? flushTails() + rawEvent + "\n\n" : rawEvent + "\n\n"; } @@ -796,11 +841,17 @@ export async function pipePluginChatWithStrip( return rawEvent + "\n\n"; } const raw = d[field] as string; + if (field === "thinking" && raw.length > 0) sawThinking = true; if (!mayStartRenderTag(raw) && !anyPending()) { + if (field === "text" && raw.length > 0) visibleTextChars += raw.length; return rawEvent + "\n\n"; } const [clean, changed] = pushField(field, index, raw); - if (!changed) return rawEvent + "\n\n"; + if (!changed) { + if (field === "text" && raw.length > 0) visibleTextChars += raw.length; + return rawEvent + "\n\n"; + } + if (field === "text" && clean.length > 0) visibleTextChars += clean.length; if (clean.length === 0 && Object.keys(d ?? {}).length <= 2) return ""; return rebuildEvent(rawEvent, { ...ev, delta: { ...d, [field]: clean } }); }; @@ -831,6 +882,10 @@ export async function pipePluginChatWithStrip( continue; } if (ev["type"] === "message_stop") sawTerminal = true; + if (ev["type"] === "message_delta") { + const d = ev["delta"] as Record | undefined; + if (d && typeof d["stop_reason"] === "string") finalFinishReason = d["stop_reason"] as string; + } const sample = usageFromSseEvent(ev); if (sample) mergeUsageSample(acc, sample); // #408: backfill the input-side usage so the host anchors on @@ -867,6 +922,7 @@ export async function pipePluginChatWithStrip( // stream completes, and those must already see this usage. settleUsage(); maybeNoteTruncated(); + maybeWarnDegenerate(); } catch (e) { settleUsage(); maybeNoteTruncated(); @@ -925,6 +981,10 @@ export async function pipePluginResponsesWithStrip( loggerLog("warn", `[tag-echo] stripped model-emitted render tag (plugin passthrough): ${snippet.slice(0, 80).replace(/\n/g, " ")}`); log?.(`[tag-echo] stripped model-emitted render tag from plugin passthrough text`); }); + // #673: turn-level observability for degenerate terminal turns. + let sawFunctionCall = false; + let sawReasoning = false; + let responseStatus: string | undefined; const write = (s: string): Promise => { if (!res.write(Buffer.from(s, "utf8"))) { return new Promise((r) => res.once("drain", () => r())); @@ -950,6 +1010,22 @@ export async function pipePluginResponsesWithStrip( reason: "plugin responses passthrough stream ended without a completion event", }); }; + const maybeWarnDegenerate = () => { + if (!sawTerminal || res.destroyed || res.writableEnded) return; + const st = tagFilter.stats(); + const msg = degenerateTurnWarning({ + reason: responseStatus, + terminalReason: "completed", + toolCalls: sawFunctionCall ? 1 : 0, + text: st, + sawThinking: sawReasoning, + wire: "plugin-passthrough-responses", + }); + if (msg) { + loggerLog("warn", msg); + log?.(msg); + } + }; let lastDeltaMeta: { item_id?: unknown; output_index?: unknown } | null = null; const flushTail = (after: string) => { const tail = tagFilter.flush(); @@ -987,6 +1063,16 @@ export async function pipePluginResponsesWithStrip( const sample = usageFromSseEvent(ev); if (sample) mergeUsageSample(acc, sample); const type = ev["type"]; + if (typeof type === "string") { + if (type.startsWith("response.reasoning")) sawReasoning = true; + if (type === "response.output_item.added" || type === "response.output_item.done") { + const item = ev["item"] as Record | undefined; + const it = item?.["type"]; + if (it === "function_call" || it === "custom_tool_call") sawFunctionCall = true; + } + const resp = ev["response"] as Record | undefined; + if (resp && typeof resp["status"] === "string") responseStatus = resp["status"] as string; + } if ( type === "response.output_text.done" || type === "response.content_part.done" || @@ -1046,6 +1132,7 @@ export async function pipePluginResponsesWithStrip( const rest = flushTail(""); if (rest.length > 0) await write(rest); } + maybeWarnDegenerate(); settleUsage(); maybeNoteTruncated(); } catch (e) { diff --git a/tests/plugin-passthrough-tag-strip-chat.test.ts b/tests/plugin-passthrough-tag-strip-chat.test.ts index 07079efd..33a3bc17 100644 --- a/tests/plugin-passthrough-tag-strip-chat.test.ts +++ b/tests/plugin-passthrough-tag-strip-chat.test.ts @@ -1,6 +1,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; import { pipePluginChatWithStrip, pipePluginJson } from "../src/plugin.ts"; +import { setLogCapture } from "../src/logger.ts"; import type { Session } from "../src/session.ts"; function makeSession(): Session { @@ -305,3 +306,109 @@ test("plugin JSON passthrough stays byte-identical for tag-free bodies", async ( await pipePluginJson(streamOf([body]), res as unknown as import("node:http").ServerResponse, session, "openai"); assert.equal(out.join(""), body, "tag-free chat body byte-identical"); }); + +test("plugin passthrough strips typo'd acplike tags from anthropic text (#673)", async () => { + const out: string[] = []; + const res = makeRes(out); + const echo = "\x3cacpi tokens=\"36\" type=\"text\"\x3em00473\x3c/acpi\x3e"; + const parts: string[] = []; + for (let i = 0; i < echo.length; i += 5) parts.push(echo.slice(i, i + 5)); + const events: string[] = [ + `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "msg_1", usage: { input_tokens: 10 } } })}\n\n`, + `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } })}\n\n`, + ...parts.map((p) => `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: p } })}\n\n`), + `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 0 })}\n\n`, + `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 } })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, + ]; + await pipePluginChatWithStrip(streamOf(events), res as unknown as import("node:http").ServerResponse, "anthropic", makeSession()); + const text = out.join(""); + assert.ok(!text.includes("acpi"), "typo'd tag must not leak to the client"); + assert.ok(!text.includes("m00473"), "tag ref must not leak to the client"); +}); + +test("plugin passthrough warns on degenerate typo-tag-only anthropic turn (#673)", async () => { + const logs: string[] = []; + setLogCapture((_level, msg) => { logs.push(msg); }); + try { + const out: string[] = []; + const res = makeRes(out); + const echo = "\x3cacpi tokens=\"36\" type=\"text\"\x3em00473\x3c/acpi\x3e"; + const parts: string[] = []; + for (let i = 0; i < echo.length; i += 5) parts.push(echo.slice(i, i + 5)); + const events: string[] = [ + `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "msg_1", usage: { input_tokens: 10 } } })}\n\n`, + `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "thinking", thinking: "" } })}\n\n`, + `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "plan the next step" } })}\n\n`, + `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 0 })}\n\n`, + `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 1, content_block: { type: "text", text: "" } })}\n\n`, + ...parts.map((p) => `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 1, delta: { type: "text_delta", text: p } })}\n\n`), + `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 1 })}\n\n`, + `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 } })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, + ]; + await pipePluginChatWithStrip(streamOf(events), res as unknown as import("node:http").ServerResponse, "anthropic", makeSession()); + assert.ok(logs.some((l) => l.includes("[degenerate-turn]")), `expected degenerate-turn warn, got: ${logs.join(" | ")}`); + } finally { + setLogCapture(null); + } +}); + +test("plugin passthrough does not warn on a clean anthropic turn (#673)", async () => { + const logs: string[] = []; + setLogCapture((_level, msg) => { logs.push(msg); }); + try { + const out: string[] = []; + const res = makeRes(out); + const events: string[] = [ + `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "msg_1", usage: { input_tokens: 10 } } })}\n\n`, + `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } })}\n\n`, + `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "All done." } })}\n\n`, + `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 0 })}\n\n`, + `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 } })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, + ]; + await pipePluginChatWithStrip(streamOf(events), res as unknown as import("node:http").ServerResponse, "anthropic", makeSession()); + assert.ok(out.join("").includes("All done."), "clean text survives"); + assert.ok(!logs.some((l) => l.includes("[degenerate-turn]")), "no warn on clean turn"); + } finally { + setLogCapture(null); + } +}); + +test("plugin passthrough does not warn when openai turn ends with tool_calls (#673)", async () => { + const logs: string[] = []; + setLogCapture((_level, msg) => { logs.push(msg); }); + try { + const out: string[] = []; + const res = makeRes(out); + const events: string[] = [ + chatChunk({ role: "assistant" }), + chatChunk({ tool_calls: [{ index: 0, id: "call_1", type: "function", function: { name: "f", arguments: "{}" } }] }), + `data: ${JSON.stringify({ id: "chatcmpl-1", object: "chat.completion.chunk", created: 1, model: "qwen", choices: [{ index: 0, delta: {}, finish_reason: "stop" }] })}\n\n`, + DONE, + ]; + await pipePluginChatWithStrip(streamOf(events), res as unknown as import("node:http").ServerResponse, "openai", makeSession()); + assert.ok(!logs.some((l) => l.includes("[degenerate-turn]")), "tool call present: not degenerate"); + } finally { + setLogCapture(null); + } +}); + +test("plugin passthrough warns on degenerate zero-text openai stop turn (#673)", async () => { + const logs: string[] = []; + setLogCapture((_level, msg) => { logs.push(msg); }); + try { + const out: string[] = []; + const res = makeRes(out); + const events: string[] = [ + chatChunk({ role: "assistant" }), + `data: ${JSON.stringify({ id: "chatcmpl-1", object: "chat.completion.chunk", created: 1, model: "qwen", choices: [{ index: 0, delta: {}, finish_reason: "stop" }] })}\n\n`, + DONE, + ]; + await pipePluginChatWithStrip(streamOf(events), res as unknown as import("node:http").ServerResponse, "openai", makeSession()); + assert.ok(logs.some((l) => l.includes("[degenerate-turn]")), `expected degenerate-turn warn, got: ${logs.join(" | ")}`); + } finally { + setLogCapture(null); + } +}); diff --git a/tests/tag-echo.test.ts b/tests/tag-echo.test.ts index 0712eead..74f4d2f7 100644 --- a/tests/tag-echo.test.ts +++ b/tests/tag-echo.test.ts @@ -9,6 +9,8 @@ import { rewriteJsonResponse } from "../src/stream.ts"; import { rewriteOpenaiJsonResponse } from "../src/stream-openai.ts"; import { rewriteResponsesJsonResponse } from "../src/stream-responses.ts"; import { buildCompressSystemPrompt } from "../src/compress-tool.ts"; +import { setLogCapture } from "../src/logger.ts"; +import { degenerateTurnWarning } from "../src/degenerate-turn.ts"; const TAG = (ref: string, tokens = 177) => `\x3cacp tokens="${tokens}" type="text">${ref}\x3c/acp>`; const LT = "\x3c"; @@ -159,6 +161,10 @@ test("streaming filter matches stripAcpTags for every split position", () => { `${TAG("m1")}${TAG("m2")}`, `first ${TAG("m1")} mid prose ${TAG("m2")} last`, `好的 ${TAG("m00155")}${TAG("m00155", 44)}${TAG("m00156", 33)} 另外 5 < 6 成立${TAG("m00157")}完毕`, + `typo ${LT}acpi tokens="36" type="text"\x3em00473${LT}/acpi\x3e tail`, + `mixed ${LT}acp tokens="36" type="text"\x3em00473${LT}/acip\x3e tail`, + `rev ${LT}apic tokens="9" type="text"\x3em001${LT}/acp\x3e tail`, + `safe #include ${LT}acpi/acpi.h\x3e and ${LT}caption\x3ex${LT}/caption\x3e ${LT}app id="1"\x3erun${LT}/app\x3e`, ]; for (const full of cases) { const expected = stripAcpTags(full); @@ -511,3 +517,94 @@ test("mayStartRenderTag engages on complete tags and tag-head tails, not prose", assert.equal(mayStartRenderTag("x\x3caction y"), false); assert.equal(mayStartRenderTag("\x3cdiv>"), false); }); + +test("stripAcpTags removes typo'd acplike render tags (#673)", () => { + assert.equal(stripAcpTags(`${LT}acpi tokens="36" type="text"\x3em00473${LT}/acpi\x3e`), ""); + assert.equal(stripAcpTags(`before ${LT}acp tokens="2" type="text"\x3em00473${LT}/acip\x3e after`), "before after"); + for (const name of ["acpi", "acip", "apic", "cap", "cpa", "pac", "pca"]) { + assert.equal(stripAcpTags(`${LT}${name} tokens="1" type="text"\x3em001${LT}/${name}\x3e`), "", name); + } +}); + +test("typo'd openers engage the streaming gate (#673)", () => { + for (const s of [`${LT}acip `, `${LT}acpi`, `${LT}/acip`]) { + assert.equal(mayStartRenderTag(s), true, s); + assert.equal(containsRenderTagText(s + "\x3e"), true, s); + } +}); + +test("legit angle-bracket text survives the loosened filter (#673)", () => { + const safe = [ + "#include \\x3cacpi/acpi.h\\x3e", + "\\x3ccaption\\x3ehi\\x3c/caption\\x3e", + "\\x3capp id=\"1\"\\x3erun\\x3c/app\\x3e", + "\\x3cACPI_DEVICE\\x3e", + "a \\x3c b and b \\x3e c", + "\\x3cacp_compress\\x3ex\\x3c/acp_compress\\x3e", + ]; + for (const s of safe) { + assert.equal(stripAcpTags(s), s); + for (let split = 0; split <= s.length; split++) { + const f = createTagEchoFilter(); + const out = f.push(s.slice(0, split)) + f.push(s.slice(split)) + f.flush(); + assert.equal(out, s, `split=${split} full=${JSON.stringify(s)}`); + } + } +}); + +test("filter stats() accumulates lifetime input/output/dropped (#673)", () => { + const first = `hello ${TAG("m1")}`; + const f = createTagEchoFilter(); + f.push(first); + f.flush(); + f.push("world"); + const st = f.stats(); + assert.equal(st.inputChars, first.length + 5); + assert.equal(st.outputChars, "hello world".length); + assert.equal(st.dropped, true); +}); + +test("degenerateTurnWarning fires only on terminal zero-text zero-tool turns (#673)", () => { + const base = { + reason: "end_turn" as string | undefined, + terminalReason: "end_turn", + toolCalls: 0, + text: { inputChars: 63, outputChars: 0, dropped: true }, + sawThinking: true, + wire: "anthropic", + }; + const hit = degenerateTurnWarning(base); + assert.match(hit ?? "", /\[degenerate-turn\] anthropic: turn ended end_turn/); + assert.match(hit ?? "", /thinking present/); + assert.match(hit ?? "", /stripped as render-tag echo/); + assert.equal(degenerateTurnWarning({ ...base, reason: "tool_use" }), null); + assert.equal(degenerateTurnWarning({ ...base, toolCalls: 1 }), null); + assert.equal(degenerateTurnWarning({ ...base, text: { inputChars: 5, outputChars: 3, dropped: false } }), null); + assert.match(degenerateTurnWarning({ ...base, sawThinking: false, text: { inputChars: 0, outputChars: 0, dropped: false } }) ?? "", /no visible text emitted/); +}); + +test("anthropic adapter warns on degenerate typo-tag-only turn (#673)", async () => { + const logs: string[] = []; + setLogCapture((_level, msg) => { logs.push(msg); }); + try { + const echo = `${LT}acpi tokens="36" type="text"\x3em00473${LT}/acpi\x3e`; + const parts: string[] = []; + for (let i = 0; i < echo.length; i += 7) parts.push(echo.slice(i, i + 7)); + const sseParts: string[] = [ + `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "msg_1", usage: { input_tokens: 100 } } })}\n\n`, + `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "thinking", thinking: "" } })}\n\n`, + `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "thinking_delta", thinking: "next step: run the build" } })}\n\n`, + `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 0 })}\n\n`, + `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 1, content_block: { type: "text", text: "" } })}\n\n`, + ...parts.map((p) => `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 1, delta: { type: "text_delta", text: p } })}\n\n`), + `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 1 })}\n\n`, + `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 5 } })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, + ]; + const out = await drain(sseFromStrings(sseParts), createAnthropicAdapter({ model: "test" })); + assert.ok(!out.includes("acpi"), "typo'd tag must not leak to the client"); + assert.ok(logs.some((l) => l.includes("[degenerate-turn]")), `expected degenerate-turn warn, got: ${logs.join(" | ")}`); + } finally { + setLogCapture(null); + } +});