Merge remote-tracking branch 'origin/main' into fix/structured-data-depth-cap

server.mjs

@@ -251,8 +251,8 @@ function parseStreamJsonLines(buffered) {
 // Reference: OLP lib/providers/anthropic.mjs anthropicStreamJsonEventToIR (commit 97e7d16).
 //
 // @param {object} event — parsed NDJSON event
-// @param {boolean} isFirstDelta — true if no content has been yielded yet
+// @param {boolean} sawTextDelta — true if a streaming content_block_delta text was already seen
-function parseStreamJsonEvent(event, isFirstDelta) {
+function parseStreamJsonEvent(event, sawTextDelta) {
   const t = event?.type;
 
   // system/* — first-event init + other system meta (api_retry etc.)
@@ -264,19 +264,23 @@ function parseStreamJsonEvent(event, isFirstDelta) {
   if (t === "stream_event") {
     const inner = event.event ?? event;
     if (inner?.type === "content_block_delta" && inner.delta?.type === "text_delta") {
-      return { text: inner.delta.text ?? "" };
+      return { text: inner.delta.text ?? "", fromDelta: true };
     }
     // Other stream_event sub-types (content_block_start, message_delta, etc.) — consumed
     return null;
   }
 
-  // assistant — aggregate message (fallback when no prior content_block_delta seen)
+  // assistant — aggregate message. claude CLI without --include-partial-messages emits NO
-  // Empirically (claude CLI without --include-partial-messages, verified v2.1.104 through v2.1.158): fast/short
+  // content_block_delta events; each assistant message arrives as its own aggregate `assistant`
-  // responses may emit ONLY the aggregate assistant event, no content_block_delta events.
+  // event. An agentic/tool-using turn has SEVERAL (preamble + one per tool round + final answer),
-  // If isFirstDelta is true, extract text here; otherwise it's a duplicate, ignore.
+  // so we must accumulate the text of EVERY such event. The prior `isFirstDelta` guard kept only
+  // the FIRST message's text and dropped the rest — silently losing the post-tool-use final answer
+  // on every tool-using turn (verified v2.1.104 through v2.1.211; see PR body capture).
+  // The only real hazard is the delta+aggregate DOUBLE-COUNT: if streaming deltas were already
+  // seen (sawTextDelta), the aggregate duplicates them — ignore it.
   // Reference: OLP commit 65f945c (assistant-aggregate fallback, fold-in).
   if (t === "assistant") {
-    if (isFirstDelta) {
+    if (!sawTextDelta) {
       const blocks = event.message?.content;
       if (Array.isArray(blocks)) {
         const text = blocks
@@ -1501,7 +1505,7 @@ async function callClaude(model, messages, conversationId, keyName, res) {
     const { proc, cliModel, conversationId: convId, t0, cleanup, handleSessionFailure, markFirstByte } = ctx;
     let lineBuffer = "";
     let assembledText = "";
-    let isFirstDelta = true;
+    let sawTextDelta = false;
     let resultEventSeen = false;
     let stderr = "";
 
@@ -1511,11 +1515,18 @@ async function callClaude(model, messages, conversationId, keyName, res) {
       const { events, remainder } = parseStreamJsonLines(lineBuffer);
       lineBuffer = remainder;
       for (const event of events) {
-        const parsed = parseStreamJsonEvent(event, isFirstDelta);
+        const parsed = parseStreamJsonEvent(event, sawTextDelta);
         if (!parsed) continue;
         if (parsed.text !== undefined) {
-          assembledText += parsed.text;
+          if (parsed.fromDelta) {
-          isFirstDelta = false;
+            assembledText += parsed.text;
+            sawTextDelta = true;
+          } else {
+            // aggregate assistant message — separate successive messages so the preamble and
+            // the post-tool-use final answer don't run together.
+            if (assembledText && !assembledText.endsWith("\n")) assembledText += "\n\n";
+            assembledText += parsed.text;
+          }
         } else if (parsed.stop) {
           resultEventSeen = true;
         } else if (parsed.error) {
@@ -1937,9 +1948,10 @@ async function callClaudeStreaming(model, messages, conversationId, res, authInf
   let stderr = "";
   let headersSent = false;
   let totalChars = 0;
+  let streamEndsWithNewline = false; // tracks whether emitted text ends in "\n" — see the separator guard below
   let cachedContent = ""; // accumulate for cache write-back
   let lineBuffer = "";
-  let isFirstDelta = true;
+  let sawTextDelta = false;
   let resultEventSeen = false;
   // Separate flag for is_error result — must NOT be conflated with resultEventSeen.
   // If errored===true the close handler must not cache the response or record success
@@ -1976,15 +1988,26 @@ async function callClaudeStreaming(model, messages, conversationId, res, authInf
     lineBuffer = remainder;
 
     for (const event of events) {
-      const parsed = parseStreamJsonEvent(event, isFirstDelta);
+      const parsed = parseStreamJsonEvent(event, sawTextDelta);
       if (!parsed) continue;
 
       if (parsed.text !== undefined) {
-        // content_block_delta text — forward as SSE delta
+        // Streamed delta, or an aggregate assistant-message text (agentic turns emit several).
-        const text = parsed.text;
+        // For an aggregate message after earlier text, prepend a separator so the preamble and
+        // the post-tool-use final answer don't run together in the forwarded stream.
+        let text = parsed.text;
+        if (parsed.fromDelta) {
+          sawTextDelta = true;
+        } else if (totalChars > 0 && !streamEndsWithNewline) {
+          // Mirror the buffered path's guard (assembledText.endsWith("\n")): only inject the
+          // blank-line separator when the already-emitted text doesn't already end in a newline,
+          // so a message ending in "\n" doesn't produce a triple newline here while the buffered
+          // path produces a single. Keeps the two assembly paths byte-identical. (PR #183 review.)
+          text = "\n\n" + text;
+        }
+        streamEndsWithNewline = text.endsWith("\n");
         totalChars += text.length;
         if (CACHE_TTL > 0) cachedContent += text;
-        isFirstDelta = false;
 
         if (!ensureHeaders()) continue;
         sendSSE(res, {

test-features.mjs

@@ -1355,7 +1355,7 @@ function parseStreamJsonLines(buffered) {
   return { events, remainder: remainder ?? "" };
 }
 
-function parseStreamJsonEvent(event, isFirstDelta) {
+function parseStreamJsonEvent(event, sawTextDelta) {
   const t = event?.type;
 
   // system/* — first-event init + other system meta (api_retry etc.)
@@ -1367,19 +1367,19 @@ function parseStreamJsonEvent(event, isFirstDelta) {
   if (t === "stream_event") {
     const inner = event.event ?? event;
     if (inner?.type === "content_block_delta" && inner.delta?.type === "text_delta") {
-      return { text: inner.delta.text ?? "" };
+      return { text: inner.delta.text ?? "", fromDelta: true };
     }
     // Other stream_event sub-types (content_block_start, message_delta, etc.) — consumed
     return null;
   }
 
-  // assistant — aggregate message (fallback when no prior content_block_delta seen)
+  // assistant — aggregate message. Without --include-partial-messages each assistant message
-  // Empirically (claude CLI without --include-partial-messages, verified v2.1.104 through v2.1.158): fast/short
+  // arrives as its own aggregate event; an agentic turn emits several (preamble + tool rounds +
-  // responses may emit ONLY the aggregate assistant event, no content_block_delta events.
+  // final answer), so accumulate EVERY one. Only guard the delta+aggregate double-count case:
-  // If isFirstDelta is true, extract text here; otherwise it's a duplicate, ignore.
+  // if streaming deltas were already seen (sawTextDelta), the aggregate duplicates them.
   // Reference: OLP commit 65f945c (assistant-aggregate fallback, fold-in).
   if (t === "assistant") {
-    if (isFirstDelta) {
+    if (!sawTextDelta) {
       const blocks = event.message?.content;
       if (Array.isArray(blocks)) {
         const text = blocks
@@ -1428,25 +1428,25 @@ test("parseStreamJsonEvent: stream_event content_block_delta yields text", () =>
     type: "stream_event",
     event: { type: "content_block_delta", delta: { type: "text_delta", text: "Hello" } }
   };
-  const result = parseStreamJsonEvent(event, true);
+  const result = parseStreamJsonEvent(event, false);
-  assert.deepEqual(result, { text: "Hello" });
+  assert.deepEqual(result, { text: "Hello", fromDelta: true });
 });
 
-test("parseStreamJsonEvent: assistant-aggregate used when isFirstDelta=true (no prior delta)", () => {
+test("parseStreamJsonEvent: assistant-aggregate used when no delta seen (sawTextDelta=false)", () => {
-  const event = {
-    type: "assistant",
-    message: { content: [{ type: "text", text: "Short answer." }] }
-  };
-  const result = parseStreamJsonEvent(event, true);
-  assert.deepEqual(result, { text: "Short answer." });
-});
-
-test("parseStreamJsonEvent: assistant-aggregate skipped when isFirstDelta=false (no double-count)", () => {
   const event = {
     type: "assistant",
     message: { content: [{ type: "text", text: "Short answer." }] }
   };
   const result = parseStreamJsonEvent(event, false);
+  assert.deepEqual(result, { text: "Short answer." });
+});
+
+test("parseStreamJsonEvent: assistant-aggregate skipped when a delta was seen (sawTextDelta=true, no double-count)", () => {
+  const event = {
+    type: "assistant",
+    message: { content: [{ type: "text", text: "Short answer." }] }
+  };
+  const result = parseStreamJsonEvent(event, true);
   assert.equal(result, null);
 });
 
@@ -1460,14 +1460,39 @@ test("parseStreamJsonEvent: stream_event + assistant → assembled without doubl
     type: "assistant",
     message: { content: [{ type: "text", text: "Streaming text." }] }
   };
-  // First event: isFirstDelta=true → yields text
+  // First event: no delta seen yet → yields text and marks fromDelta
-  const r1 = parseStreamJsonEvent(delta, true);
+  const r1 = parseStreamJsonEvent(delta, false);
-  assert.deepEqual(r1, { text: "Streaming text." });
+  assert.deepEqual(r1, { text: "Streaming text.", fromDelta: true });
-  // Second event (aggregate): isFirstDelta is now false (content already emitted) → null
+  // Second event (aggregate): a delta was seen (sawTextDelta=true) → duplicate, null
-  const r2 = parseStreamJsonEvent(agg, false);
+  const r2 = parseStreamJsonEvent(agg, true);
   assert.equal(r2, null);
 });
 
+// REGRESSION (agentic turns): without --include-partial-messages a tool-using turn emits SEVERAL
+// aggregate `assistant` events (preamble, then the final answer after tool use) and NO deltas.
+// Every one must be captured — the old first-only guard dropped the final answer.
+test("parseStreamJsonEvent: multi-message agentic turn captures preamble AND final answer", () => {
+  const preamble = {
+    type: "assistant",
+    message: { content: [
+      { type: "text", text: "I'll find the homepage repo and remove the calendar." },
+      { type: "tool_use", id: "t1", name: "Bash" },
+    ] }
+  };
+  const toolResult = { type: "user", message: { content: [{ type: "tool_result", content: "ok" }] } };
+  const finalMsg = {
+    type: "assistant",
+    message: { content: [{ type: "text", text: "Done — removed the calendar widget and pushed." }] }
+  };
+  // No deltas are ever emitted in aggregate mode, so sawTextDelta stays false throughout.
+  const r1 = parseStreamJsonEvent(preamble, false);
+  assert.deepEqual(r1, { text: "I'll find the homepage repo and remove the calendar." });
+  const r2 = parseStreamJsonEvent(toolResult, false); // user/tool_result echo — consumed
+  assert.equal(r2, null);
+  const r3 = parseStreamJsonEvent(finalMsg, false);   // <- old code returned null here (bug)
+  assert.deepEqual(r3, { text: "Done — removed the calendar widget and pushed." });
+});
+
 // (b) aggregate-only short response → assembles correctly
 test("parseStreamJsonEvent: aggregate-only multi-block response assembles all text blocks", () => {
   const event = {
@@ -1480,7 +1505,7 @@ test("parseStreamJsonEvent: aggregate-only multi-block response assembles all te
       ]
     }
   };
-  const result = parseStreamJsonEvent(event, true);
+  const result = parseStreamJsonEvent(event, false);
   assert.deepEqual(result, { text: "Part one. Part two." });
 });
 
@@ -1498,8 +1523,8 @@ test("parseStreamJsonLines: partial line carried as remainder", () => {
   assert.equal(ev2[0].type, "stream_event");
   assert.equal(rem2, "");
   // Verify the reassembled event parses through parseStreamJsonEvent correctly
-  const parsed = parseStreamJsonEvent(ev2[0], true);
+  const parsed = parseStreamJsonEvent(ev2[0], false);
-  assert.deepEqual(parsed, { text: "Hi" });
+  assert.deepEqual(parsed, { text: "Hi", fromDelta: true });
 });
 
 test("parseStreamJsonLines: empty input returns no events and empty remainder", () => {