From 8c3986db58d45c0940b53fee6fe0fda13629a7cd Mon Sep 17 00:00:00 2001 From: tornado404 Date: Wed, 12 Aug 2026 10:18:12 +0800 Subject: [PATCH] feat: add --stream-json streaming NDJSON output for headless mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bundled ZCode runtime (vendor/zcode.cjs) already emits live session events internally via runtime.subscribeEvents, but the headless --prompt --json path ignores them and prints a single summary object at turn end. Integrations like Multica therefore see no tool calls, thinking, or incremental text while the agent works. This adds a --stream-json flag that surfaces the live event stream as qwen-compatible NDJSON on stdout (one JSON object per line): system → assistant/thinking → assistant/text → tool_use → user/tool_result → ... → result The implementation is a post-sync patch (scripts/patch-runtime-stream.ts) that injects an onEvent callback into runPrompt's submitPrompt call and maps runtime events (model_streaming, tool_call_*, turn_complete) to the qwen schema {type, message:{content:[...]}}. The patch anchors on stable substrings in the minified bundle, fails loudly on upstream changes, and is idempotent. It runs automatically as part of sync/sync:local/sync:locked. Verification: scripts/test-stream.ts runs a prompt end-to-end and asserts the NDJSON shape (system/assistant/result events, non-zero usage). Launcher change: --stream-json added to runtimeBooleanOptions so the flag is forwarded to the runtime instead of being flagged invalid. --- .gitignore | 1 + package.json | 8 +- scripts/patch-runtime-stream.ts | 243 ++++++++++++++++++++++++++++++++ scripts/test-stream.ts | 111 +++++++++++++++ src/launcher.ts | 1 + 5 files changed, 361 insertions(+), 3 deletions(-) create mode 100644 scripts/patch-runtime-stream.ts create mode 100644 scripts/test-stream.ts diff --git a/.gitignore b/.gitignore index f9b6f26..b1dcb0f 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ bin/zcode.js .zcode/ *.tgz .DS_Store +package-lock.json diff --git a/package.json b/package.json index ccb07b9..98f4ac3 100644 --- a/package.json +++ b/package.json @@ -44,10 +44,12 @@ "bench:tui-memory": "bun --expose-gc scripts/bench-tui-memory.ts", "bench:tui-stream": "bun scripts/bench-tui-stream.ts", "dev": "bun run sync:local && ZCODE_NODE=node bun bin/zcode.ts", - "sync": "bun run build && bun scripts/sync-runtime.ts", - "sync:locked": "bun run build && bun scripts/sync-runtime.ts --lock zcode-runtime.lock.json", - "sync:local": "bun run build && bun scripts/sync-runtime.ts --app /Applications/ZCode.app", + "patch:stream": "bun scripts/patch-runtime-stream.ts", + "sync": "bun run build && bun scripts/sync-runtime.ts && bun scripts/patch-runtime-stream.ts", + "sync:locked": "bun run build && bun scripts/sync-runtime.ts --lock zcode-runtime.lock.json && bun scripts/patch-runtime-stream.ts", + "sync:local": "bun run build && bun scripts/sync-runtime.ts --app /Applications/ZCode.app && bun scripts/patch-runtime-stream.ts", "check": "bun run build && bun scripts/check-runtime.ts", + "check:stream": "bun scripts/test-stream.ts", "check:oauth-callback": "bun scripts/smoke-oauth-callback.ts", "check:tui": "bun scripts/smoke-tui.ts && bun scripts/smoke-tui-features.ts && bun scripts/smoke-tui-pressure.ts && bun scripts/smoke-tui-widths.ts", "test": "bun test", diff --git a/scripts/patch-runtime-stream.ts b/scripts/patch-runtime-stream.ts new file mode 100644 index 0000000..f809f75 --- /dev/null +++ b/scripts/patch-runtime-stream.ts @@ -0,0 +1,243 @@ +#!/usr/bin/env bun +/** + * patch-runtime-stream.ts — post-sync patch for vendor/zcode.cjs. + * + * Adds a `--stream-json` flag to the bundled ZCode runtime's headless + * `--prompt` mode. When set, the runtime streams qwen-compatible NDJSON events + * to stdout (one JSON object per line) as the agent works, instead of printing + * a single JSON summary at the end. This lets integrations like Multica + * observe tool calls, thinking deltas, and assistant text in real time. + * + * The patch is purely string-based (no AST): the vendor bundle is minified, + * so we anchor on stable, distinctive substrings. If an upstream runtime + * upgrade changes those anchors, the patch fails loudly instead of silently + * corrupting the bundle. + * + * Run automatically after sync-runtime.ts via `bun run sync`. Idempotent: + * re-running on an already-patched bundle is a no-op. + */ +import { readFileSync, writeFileSync } from "node:fs"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; + +const root = join(dirname(fileURLToPath(import.meta.url)), ".."); +const runtimePath = join(root, "vendor", "zcode.cjs"); + +/** Idempotency marker: the injected helper function name. Present = already patched. */ +const PATCH_MARKER = "__zcodeStreamEmit"; + +/** + * The onEvent callback injected into runPrompt's submitPrompt call. + * + * It receives raw runtime session events ({type, payload, sessionId, ...}) and + * writes qwen-compatible NDJSON lines to stdout. The mapping mirrors + * Multica's qwen backend schema so the Multica zcode adapter can reuse qwen's + * streaming parser with minimal changes: + * + * runtime event → NDJSON line + * ───────────────────────────────────────────────────────────── + * model_streaming text_delta → {type:"assistant", message:{content:[{type:"text", text:delta}]}} + * model_streaming reasoning → {type:"assistant", message:{content:[{type:"thinking", thinking:delta}]}} + * tool_call_scheduled → {type:"assistant", message:{content:[{type:"tool_use", id, name, input}]}} + * tool_call_result → {type:"user", message:{content:[{type:"tool_result", tool_use_id, content}]}} + * tool_call_error → {type:"user", message:{content:[{type:"tool_result", tool_use_id, content, is_error:true}]}} + * turn_started → {type:"system", session_id, subtype:"init"} + * turn_complete → {type:"result", subtype, is_error, result, usage, session_id} + * turn_failed → {type:"result", subtype:"error", is_error:true, error:{message}, session_id} + * + * Unknown event types are skipped (forward-compatible). + */ +const ON_EVENT_HELPER = ` +var __zcodeStreamEmit = function(ev) { + try { + var p = ev.payload || {}; + var sid = ev.sessionId || ""; + var line = null; + var t = ev.type; + if (t === "turn_started") { + line = { type: "system", subtype: "init", session_id: sid }; + } else if (t === "model_streaming") { + var kind = p.kind; + var delta = p.delta || ""; + if (kind === "text_delta" && delta) { + line = { type: "assistant", session_id: sid, message: { content: [{ type: "text", text: delta }] } }; + } else if (kind === "reasoning_delta" && delta) { + line = { type: "assistant", session_id: sid, message: { content: [{ type: "thinking", thinking: delta }] } }; + } + } else if (t === "tool_call_scheduled") { + var input = p.input; + if (typeof input === "string") { try { input = JSON.parse(input); } catch (e) {} } + line = { type: "assistant", session_id: sid, message: { content: [{ type: "tool_use", id: p.toolCallId, name: p.toolName, input: input || {} }] } }; + } else if (t === "tool_call_result" || t === "tool_call_error") { + var content = ""; + var r = p.result; + if (typeof r === "string") { content = r; } + else if (r && typeof r === "object") { + content = r.content || r.output || r.stdout || ""; + if (typeof content !== "string") { try { content = JSON.stringify(content); } catch (e) { content = String(content); } } + } + var block = { type: "tool_result", tool_use_id: p.toolCallId, content: content }; + if (t === "tool_call_error") { block.is_error = true; } + line = { type: "user", session_id: sid, message: { content: [block] } }; + } else if (t === "turn_complete") { + var rt = p.resultType || "success"; + var isErr = rt !== "success" && rt !== "cancelled"; + var usage = p.usage || {}; + line = { + type: "result", + subtype: isErr ? "error" : "success", + session_id: sid, + is_error: isErr, + result: p.response || "", + usage: { + input_tokens: usage.inputTokens || 0, + output_tokens: usage.outputTokens || 0, + cache_read_input_tokens: usage.cacheReadTokens || 0 + } + }; + } else if (t === "turn_failed") { + var errMsg = (p.error && p.error.message) ? p.error.message : "turn failed"; + line = { type: "result", subtype: "error_during_execution", session_id: sid, is_error: true, error: { message: errMsg } }; + } + if (line) { process.stdout.write(JSON.stringify(line) + "\\n"); } + } catch (e) { /* never let event emission break the turn */ } +}; +`; + +interface PatchSpec { + /** Unique, stable anchor string present in the unpatched bundle. */ + anchor: string; + /** Replacement string. */ + replacement: string; + /** Human-readable description for error messages. */ + description: string; +} + +/** + * Patch 1 — argv parser: add `stream-json` boolean option. + * + * The runtime's global argv parser (parseArgs, strict mode) rejects unknown + * flags. We add `"stream-json":{type:"boolean"}` right after the existing + * `json:{type:"boolean"}` entry so `--stream-json` is accepted. The parsed + * value surfaces as `o["stream-json"]` (values.json), which runPrompt reads + * via the closure-captured options object. + */ +const ARGV_PATCH: Omit = { + anchor: "json:{type:\"boolean\"},\"no-color\":{type:\"boolean\"}", + description: "argv parser json option block" +}; + +/** + * Patch 2 — runPrompt: inject onEvent + emit the stream-json branch. + * + * Anchor on the exact submitPrompt call inside runPrompt (function `wqt`). + * The call passes `{abortSignal:y.signal}`; we extend it with `onEvent` and + * gate on `o["stream-json"]` to decide whether to stream. When streaming, the + * final summary print is suppressed (the terminal `result` NDJSON line already + * carries response + usage). + * + * The replacement preserves the original code path when --stream-json is NOT + * set, so existing `--prompt --json` behavior is unchanged. + */ +const SUBMIT_PROMPT_ANCHOR = + "submitPrompt(r.length>0?{text:p,attachments:r.map(J=>({type:Vga(J),path:J}))}:p,{abortSignal:y.signal})"; + +const SUBMIT_PROMPT_REPLACEMENT = + 'submitPrompt(r.length>0?{text:p,attachments:r.map(J=>({type:Vga(J),path:J}))}:p,{abortSignal:y.signal,onEvent:o.streamJson?__zcodeStreamEmit:void 0})'; + +/** + * Patch 3 — suppress the final summary print in stream-json mode. + * + * After submitPrompt returns, runPrompt does `o.json ? : `. + * In stream-json mode we must NOT print either (the stream already ended with a + * `result` line). Replace the ternary head to short-circuit when stream-json. + * + * Note: runPrompt's `o` is built by the `lva` options builder (patch 4 below), + * which exposes the parsed argv as `o.streamJson` (camelCase). We branch on + * `o.streamJson` here and in the submitPrompt call. + */ +const SUMMARY_ANCHOR = + "return m=W.traceId??m,o.json?(e.stdout.write(Sl({sessionId:f.sessionId,traceId:m"; + +const SUMMARY_REPLACEMENT = + 'return m=W.traceId??m,o.streamJson?0:o.json?(e.stdout.write(Sl({sessionId:f.sessionId,traceId:m'; + +/** + * Patch 4 — lva options builder: map `stream-json` argv value to `o.streamJson`. + * + * `lva` is the function that constructs the options object passed to runPrompt + * (as `o`). It explicitly copies selected argv fields (json, verbose, locale, + * ...) into camelCase properties. We add `streamJson` so runPrompt can read + * `o.streamJson` to decide whether to stream. + * + * Anchor: append `streamJson:e["stream-json"]===!0` after the `verbose` mapping, + * which is a stable, distinctive suffix of the lva object literal. + */ +const LVA_ANCHOR = "noColor:e[\"no-color\"]===!0,verbose:e.verbose===!0"; +const LVA_REPLACEMENT = + "noColor:e[\"no-color\"]===!0,streamJson:e[\"stream-json\"]===!0,verbose:e.verbose===!0"; + +function fail(message: string): never { + console.error(`patch-runtime-stream: ${message}`); + process.exit(1); +} + +function applyPatch(src: string, anchor: string, replacement: string, description: string): string { + const count = src.split(anchor).length - 1; + if (count === 0) { + fail(`anchor not found for ${description}. The upstream runtime may have changed; inspect vendor/zcode.cjs and update the patch.`); + } + if (count > 1) { + fail(`anchor matched ${count} times for ${description} (expected exactly 1). The anchor is no longer unique; make it more specific.`); + } + return src.replace(anchor, replacement); +} + +function main(): void { + let src: string; + try { + src = readFileSync(runtimePath, "utf8"); + } catch { + fail(`cannot read ${runtimePath}. Run \`bun run sync\` first to extract the vendor runtime.`); + } + + // Idempotent: a re-run on an already-patched bundle is a no-op. + if (src.includes("__zcodeStreamEmit")) { + console.log("patch-runtime-stream: already applied, skipping."); + return; + } + + // Patch 1: argv parser — add stream-json option. + src = applyPatch( + src, + ARGV_PATCH.anchor, + 'json:{type:"boolean"},"stream-json":{type:"boolean"},"no-color":{type:"boolean"}', + ARGV_PATCH.description + ); + + // Patch 2: runPrompt submitPrompt call — inject onEvent callback. + src = applyPatch(src, SUBMIT_PROMPT_ANCHOR, SUBMIT_PROMPT_REPLACEMENT, "runPrompt submitPrompt call"); + + // Patch 3: suppress summary print in stream-json mode. + src = applyPatch(src, SUMMARY_ANCHOR, SUMMARY_REPLACEMENT, "runPrompt summary print branch"); + + // Patch 4: lva options builder — map stream-json argv to o.streamJson. + src = applyPatch(src, LVA_ANCHOR, LVA_REPLACEMENT, "lva options builder stream-json mapping"); + + // Inject the onEvent helper at the bundle's global scope. The bundle starts + // with `#!/usr/bin/env node\n"use strict";` — insert the helper right + // after `"use strict";` so it lives at module top level (a legal statement + // boundary) and is defined before runPrompt ever calls it. + const useStrictAnchor = '"use strict";'; + const useStrictCount = src.split(useStrictAnchor).length - 1; + if (useStrictCount < 1) { + fail('cannot find "use strict"; at bundle head to insert helper.'); + } + // Only replace the FIRST occurrence (the bundle's top-level directive). + src = src.replace(useStrictAnchor, useStrictAnchor + ON_EVENT_HELPER); + + writeFileSync(runtimePath, src); + console.log("patch-runtime-stream: applied streaming NDJSON support to vendor/zcode.cjs."); +} + +main(); diff --git a/scripts/test-stream.ts b/scripts/test-stream.ts new file mode 100644 index 0000000..6a296fe --- /dev/null +++ b/scripts/test-stream.ts @@ -0,0 +1,111 @@ +#!/usr/bin/env bun +/** + * test-stream.ts — verify the patched runtime emits valid streaming NDJSON. + * + * Runs `zcode --prompt --stream-json` against the patched vendor runtime + * and asserts the output is well-formed NDJSON that matches the qwen-compatible + * schema Multica's streaming adapter expects: + * + * - every stdout line is valid JSON + * - the stream contains at least one event + * - the final event is type:"result" + * - a "result" event carries is_error:false on success and non-empty usage + * + * Exits non-zero on failure. Intended for `bun run check:stream` and CI. + */ +import { spawn } from "node:child_process"; +import { readFileSync } from "node:fs"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; + +const root = join(dirname(fileURLToPath(import.meta.url)), ".."); +const runtimePath = join(root, "vendor", "zcode.cjs"); +const node = process.env.ZCODE_NODE?.trim() || process.execPath; + +interface StreamEvent { + type: string; + subtype?: string; + session_id?: string; + is_error?: boolean; + result?: string; + usage?: Record; + message?: { content?: Array> }; +} + +function fail(message: string): never { + console.error(`test-stream: FAIL — ${message}`); + process.exit(1); +} + +async function runStream(prompt: string): Promise { + const events: StreamEvent[] = []; + return await new Promise((resolve, reject) => { + const child = spawn(node, [runtimePath, "--prompt", prompt, "--stream-json", "--cwd", "/tmp"], { + stdio: ["ignore", "pipe", "pipe"] + }); + let stderr = ""; + const stdoutChunks: Buffer[] = []; + child.stdout.on("data", (chunk: Buffer) => stdoutChunks.push(chunk)); + child.stderr.on("data", (chunk: Buffer) => { stderr += chunk.toString(); }); + child.on("error", reject); + child.on("exit", (code) => { + const stdout = Buffer.concat(stdoutChunks).toString(); + if (code !== 0) { + reject(new Error(`runtime exited with ${code}\nstderr: ${stderr.slice(-1000)}`)); + return; + } + for (const line of stdout.split("\n")) { + const trimmed = line.trim(); + if (!trimmed) continue; + try { + events.push(JSON.parse(trimmed) as StreamEvent); + } catch (error) { + reject(new Error(`non-JSON stdout line: ${trimmed.slice(0, 100)}`)); + return; + } + } + resolve(events); + }); + }); +} + +async function main(): Promise { + try { + readFileSync(runtimePath, "utf8"); + } catch { + fail(`cannot read ${runtimePath}. Run \`bun run sync:local\` first.`); + } + + console.log("test-stream: running patched runtime with --stream-json..."); + const events = await runStream("reply with exactly the word hello"); + console.log(`test-stream: received ${events.length} events`); + + if (events.length === 0) { + fail("no events emitted — the patch may not have applied. Run `bun run patch:stream`."); + } + + const last = events[events.length - 1]!; + if (last.type !== "result") { + fail(`last event is ${last.type}, expected "result". Events: ${events.map((e) => e.type).join(", ")}`); + } + + if (last.is_error) { + fail(`result event reports is_error:true — result=${last.result ?? "(empty)"}`); + } + + const usage = last.usage ?? {}; + if ((usage.input_tokens ?? 0) === 0 && (usage.output_tokens ?? 0) === 0) { + fail(`result usage is all-zero: ${JSON.stringify(usage)}`); + } + + const types = new Set(events.map((e) => e.type)); + if (!types.has("assistant") && !types.has("result")) { + fail(`expected at least an assistant or result event, got: ${[...types].join(", ")}`); + } + + console.log("test-stream: PASS"); + console.log(` events: ${events.map((e) => e.type).join(" → ")}`); + console.log(` usage: in=${usage.input_tokens} out=${usage.output_tokens} cache_read=${usage.cache_read_input_tokens}`); +} + +main().catch((error) => fail(error instanceof Error ? error.message : String(error))); diff --git a/src/launcher.ts b/src/launcher.ts index 2d9d35d..1ec46e6 100644 --- a/src/launcher.ts +++ b/src/launcher.ts @@ -40,6 +40,7 @@ const runtimeBooleanOptions = new Set([ "--no-browser", "--no-color", "--stdio", + "--stream-json", "--target-replace", "--verbose" ]);