CLI: make the stream renderer generic with per-endpoint event strategies

talagluck · talagluck · commit a0d58b0c6440 · 2026-07-17T14:53:28.000+02:00
Addresses review feedback: the pretty renderer no longer sniffs event shapes in
one monolith. createStreamRenderer now owns only the generic mechanics (NDJSON/
YAML serialization, the sink + mid-line tracking, end-of-stream flush) and
delegates per-event rendering to an injectable StreamEventRenderer. The three
schemas get dedicated strategies (answer / question / agent-response) plus a
generic fallback; the generator selects one per endpoint from the SSE response's
event schema (STREAM_RENDERERS), so a new/changed schema means a new strategy at
the call site rather than editing the core.
diff --git a/packages/cli/src/output.test.ts b/packages/cli/src/output.test.ts
@@ -5,6 +5,10 @@ import {
     CLI_TIMEOUT_MS,
     coerceArrayQueryParam,
     coerceBodyFlag,
+    createAgentResponseStreamRenderer,
+    createAnswerStreamRenderer,
+    createGenericStreamRenderer,
+    createQuestionStreamRenderer,
     createRequestTimeout,
     createStreamRenderer,
     documentToText,
@@ -335,7 +339,7 @@ describe('createStreamRenderer', () => {
 
     it('pretty renders a recommended-question stream as bullets', () => {
         const out = captureStdout(() => {
-            const r = createStreamRenderer({ pretty: true });
+            const r = createStreamRenderer({ pretty: true }, createQuestionStreamRenderer());
             r.write({ question: 'How do I start?' });
             r.write({ question: 'What is X?' });
             r.end();
@@ -345,7 +349,7 @@ describe('createStreamRenderer', () => {
 
     it('pretty prints answer text incrementally, then sources and follow-ups', () => {
         const out = captureStdout(() => {
-            const r = createStreamRenderer({ pretty: true });
+            const r = createStreamRenderer({ pretty: true }, createAnswerStreamRenderer());
             // Each event carries the answer so far; only the new suffix prints.
             r.write({
                 type: 'answer',
@@ -368,7 +372,7 @@ describe('createStreamRenderer', () => {
 
     it('pretty renders a document-form answer (--format document) as text', () => {
         const out = captureStdout(() => {
-            const r = createStreamRenderer({ pretty: true });
+            const r = createStreamRenderer({ pretty: true }, createAnswerStreamRenderer());
             r.write({
                 type: 'answer',
                 answer: {
@@ -403,7 +407,7 @@ describe('createStreamRenderer', () => {
 
     it('pretty renders agent-response events as concise status lines', () => {
         const out = captureStdout(() => {
-            const r = createStreamRenderer({ pretty: true });
+            const r = createStreamRenderer({ pretty: true }, createAgentResponseStreamRenderer());
             r.write({ type: 'response_start', messageId: 'm1' });
             r.write({ type: 'response_document', operation: 'insert', blocks: [{}, {}] });
             r.write({ type: 'response_finish', messageId: 'm1', response: {} });
@@ -413,6 +417,32 @@ describe('createStreamRenderer', () => {
             '[response_start]\n[response_document] insert — 2 block(s)\n[response_finish]\n',
         );
     });
+
+    it('generic renderer (the default) streams strings inline and blocks objects', () => {
+        const withDefault = captureStdout(() => {
+            const r = createStreamRenderer({ pretty: true }); // no strategy → generic
+            r.write('hello ');
+            r.write('world');
+            r.end();
+        });
+        expect(withDefault).toBe('hello world\n');
+
+        const explicit = captureStdout(() => {
+            const r = createStreamRenderer({ pretty: true }, createGenericStreamRenderer());
+            r.write({ a: 1 });
+            r.end();
+        });
+        expect(explicit).toBe('a: 1\n');
+    });
+
+    it('machine formats ignore the event renderer (stay schema-agnostic)', () => {
+        const out = captureStdout(() => {
+            const r = createStreamRenderer({ json: true }, createAnswerStreamRenderer());
+            r.write({ anything: true });
+            r.end();
+        });
+        expect(out).toBe('{"anything":true}\n');
+    });
 });
 
 describe('documentToText', () => {
diff --git a/packages/cli/src/output.ts b/packages/cli/src/output.ts
@@ -377,12 +377,37 @@ export interface StreamRenderer {
     end(): void;
 }
 
+// A line-oriented output target handed to StreamEventRenderers. `out` writes
+// inline (and may leave the cursor mid-line, e.g. incremental answer text);
+// `line` writes a full line. The core tracks the mid-line state so `end` can
+// flush a trailing newline before any end-of-stream content.
+export interface StreamSink {
+    out(text: string): void;
+    line(text: string): void;
+}
+
+// Strategy for rendering ONE family of events in pretty mode. The core renderer
+// owns all the generic mechanics (machine-format serialization, the sink and its
+// mid-line tracking, calling `end`); a StreamEventRenderer only has to know how to
+// turn its own event schema into text. New/changed event schemas mean a new
+// strategy passed at the call site — the core never changes. See generate-commands.ts,
+// which selects the strategy per endpoint from the response's event-stream schema.
+export interface StreamEventRenderer {
+    write(event: unknown, sink: StreamSink): void;
+    // Flush end-of-stream content (e.g. citations). Optional; defaults to a no-op.
+    end?(sink: StreamSink): void;
+}
+
 // Machine formats stream one record per event so a consumer can parse the output
 // incrementally: NDJSON for --json (one compact object per line), a stream of
-// YAML documents for --yaml (each introduced by the `---` marker). Pretty mode
-// renders progressively (see createPrettyStreamRenderer). Format resolution — and
-// the piped-defaults-to-YAML behaviour — is shared with printResult.
-export function createStreamRenderer(options: OutputOptions): StreamRenderer {
+// YAML documents for --yaml (each introduced by the `---` marker) — both fully
+// schema-agnostic. Pretty mode delegates per-event rendering to `eventRenderer`
+// (a generic fallback when the caller doesn't supply one). Format resolution —
+// and the piped-defaults-to-YAML behaviour — is shared with printResult.
+export function createStreamRenderer(
+    options: OutputOptions,
+    eventRenderer: StreamEventRenderer = createGenericStreamRenderer(),
+): StreamRenderer {
     const format = resolveFormat(options);
     if (format === 'json') {
         return {
@@ -397,89 +422,110 @@ export function createStreamRenderer(options: OutputOptions): StreamRenderer {
             end: () => {},
         };
     }
-    return createPrettyStreamRenderer();
-}
-
-// Human-readable streaming. The three event shapes these endpoints emit are each
-// rendered on the fly:
-//   - recommended-question events (`{ question }`)   → one bullet per question
-//   - answer snapshots (`{ type: 'answer', answer }`) → the answer text, printed
-//     incrementally as the snapshot grows, then its sources as citations at end
-//   - agent-response events (`{ type: 'response_*' }`) → a concise status line
-// Unknown shapes fall back to a compact block so nothing is silently dropped.
-function createPrettyStreamRenderer(): StreamRenderer {
-    let printedAnswerLen = 0; // chars of the (string) answer already written
+
     let midLine = false; // last write left the cursor mid-line
-    let sources: unknown[] | null = null;
-    let followups: unknown[] | null = null;
+    const sink: StreamSink = {
+        out(text) {
+            if (text.length === 0) return;
+            process.stdout.write(text);
+            midLine = !text.endsWith('\n');
+        },
+        line(text) {
+            process.stdout.write(`${text}\n`);
+            midLine = false;
+        },
+    };
+    return {
+        write: (event) => eventRenderer.write(event, sink),
+        end: () => {
+            if (midLine) {
+                process.stdout.write('\n');
+                midLine = false;
+            }
+            eventRenderer.end?.(sink);
+        },
+    };
+}
 
-    const out = (s: string) => {
-        if (s.length === 0) return;
-        process.stdout.write(s);
-        midLine = !s.endsWith('\n');
+// Fallback strategy: no schema knowledge. Strings stream inline; objects/scalars
+// print as a compact block. Used for any streaming endpoint without a dedicated
+// renderer, so a new SSE endpoint is never silently blank.
+export function createGenericStreamRenderer(): StreamEventRenderer {
+    return {
+        write(event, sink) {
+            if (typeof event === 'string') sink.out(event);
+            else if (isPlainObject(event)) sink.line(formatPretty(event));
+            else sink.line(formatScalar(event));
+        },
     };
-    const line = (s: string) => {
-        process.stdout.write(`${s}\n`);
-        midLine = false;
+}
+
+// `SearchAIRecommendedQuestionStream`: one `{ question }` per event → a bullet.
+export function createQuestionStreamRenderer(): StreamEventRenderer {
+    return {
+        write(event, sink) {
+            if (isPlainObject(event) && typeof event.question === 'string') {
+                sink.line(`• ${event.question}`);
+            }
+        },
     };
+}
 
+// `SearchAIAnswerStream`: progressive `{ type: 'answer', answer }` snapshots. Each
+// event carries the answer so far, so only the new suffix is printed; the latest
+// snapshot's sources/follow-ups render as citations once the stream ends.
+export function createAnswerStreamRenderer(): StreamEventRenderer {
+    let printedAnswerLen = 0;
+    let sources: unknown[] | null = null;
+    let followups: unknown[] | null = null;
     return {
-        write(event: unknown): void {
-            if (typeof event === 'string') {
-                out(event);
+        write(event, sink) {
+            if (!isPlainObject(event) || event.type !== 'answer' || !isPlainObject(event.answer)) {
                 return;
             }
-            if (!isPlainObject(event)) {
-                line(formatScalar(event));
-                return;
+            const answer = event.answer;
+            if (Array.isArray(answer.sources)) sources = answer.sources;
+            if (Array.isArray(answer.followupQuestions)) followups = answer.followupQuestions;
+            const text = streamAnswerText(answer.answer);
+            if (text !== undefined) {
+                if (text.length >= printedAnswerLen) sink.out(text.slice(printedAnswerLen));
+                else sink.out(`\n${text}`); // shrank unexpectedly: reprint whole
+                printedAnswerLen = text.length;
             }
-            // Recommended-question stream: one `{ question }` per event.
-            if (typeof event.question === 'string' && event.type === undefined) {
-                line(`• ${event.question}`);
+        },
+        end(sink) {
+            if (sources && sources.length > 0) {
+                sink.line('');
+                sink.line('Sources:');
+                for (const s of sources) sink.line(`  - ${formatStreamSource(s)}`);
+            }
+            if (followups && followups.length > 0) {
+                sink.line('');
+                sink.line('Follow-up questions:');
+                for (const q of followups) sink.line(`  - ${formatScalar(q)}`);
+            }
+        },
+    };
+}
+
+// `AIStreamResponse`: a status line per event, except the JSON-object chunks,
+// which are raw text to concatenate.
+export function createAgentResponseStreamRenderer(): StreamEventRenderer {
+    return {
+        write(event, sink) {
+            if (!isPlainObject(event)) {
+                sink.line(formatScalar(event));
                 return;
             }
-            // Answer stream: progressive `{ type: 'answer', answer }` snapshots.
-            // Each event carries the answer so far, so we print only the new
-            // suffix; sources/followups from the latest snapshot render at end.
-            if (event.type === 'answer' && isPlainObject(event.answer)) {
-                const answer = event.answer;
-                if (Array.isArray(answer.sources)) sources = answer.sources;
-                if (Array.isArray(answer.followupQuestions)) followups = answer.followupQuestions;
-                const text = streamAnswerText(answer.answer);
-                if (text !== undefined) {
-                    if (text.length >= printedAnswerLen) out(text.slice(printedAnswerLen));
-                    else out(`\n${text}`); // shrank unexpectedly: reprint whole
-                    printedAnswerLen = text.length;
-                }
+            if (typeof event.jsonChunk === 'string') {
+                sink.out(event.jsonChunk);
                 return;
             }
-            // Agent-response stream (AIStreamResponse): a status line per event,
-            // except the JSON-object chunks which are raw text to concatenate.
             if (typeof event.type === 'string') {
-                if (typeof event.jsonChunk === 'string') {
-                    out(event.jsonChunk);
-                    return;
-                }
-                line(streamStatusLine(event));
+                sink.line(streamStatusLine(event));
                 return;
             }
-            line(formatPretty(event));
-        },
-        end(): void {
-            if (midLine) {
-                process.stdout.write('\n');
-                midLine = false;
-            }
-            if (sources && sources.length > 0) {
-                line('');
-                line('Sources:');
-                for (const s of sources) line(`  - ${formatStreamSource(s)}`);
-            }
-            if (followups && followups.length > 0) {
-                line('');
-                line('Follow-up questions:');
-                for (const q of followups) line(`  - ${formatScalar(q)}`);
-            }
+            sink.line(formatPretty(event));
         },
     };
 }
diff --git a/scripts/generate-commands.ts b/scripts/generate-commands.ts
@@ -288,6 +288,20 @@ function isStreamingOperation(operation: Operation): boolean {
     return false;
 }
 
+// The schema name of an SSE operation's event (e.g. `SearchAIAnswerStream`), read
+// from the `text/event-stream` response's schema `$ref`. Drives per-endpoint
+// renderer selection in emitStreamingCommand. Undefined if not resolvable.
+function streamEventSchemaName(operation: Operation): string | undefined {
+    for (const response of Object.values(operation.responses ?? {})) {
+        const stream = (
+            response.content as Record<string, { schema?: SchemaObject }> | undefined
+        )?.['text/event-stream'];
+        const ref = stream?.schema?.$ref ?? stream?.schema?.items?.$ref;
+        if (typeof ref === 'string') return ref.match(/\/([^/]+)$/)?.[1];
+    }
+    return undefined;
+}
+
 // ─── Path parsing ───────────────────────────────────────────────────────────--
 
 interface Seg {
@@ -557,6 +571,9 @@ interface Route {
     bodyFlags: BodyFlag[] | null;
     hasBody: boolean;
     isStreaming: boolean; // SSE endpoint → emitted via the streaming path
+    // Schema name of the SSE event (from the text/event-stream response), used to
+    // pick the pretty StreamEventRenderer. Undefined for non-streaming routes.
+    streamEventSchema?: string;
     naming: Naming;
 }
 
@@ -628,6 +645,7 @@ function buildRoutes(spec: OpenAPISpec): Route[] {
                 bodyFlags,
                 hasBody,
                 isStreaming: isStreamingOperation(operation),
+                streamEventSchema: streamEventSchemaName(operation),
                 naming: computeNaming(
                     upper,
                     apiPath,
@@ -1043,6 +1061,17 @@ function emitSimpleCommand(
     return lines.join('\n');
 }
 
+// Maps an SSE event schema name to the output-module StreamEventRenderer factory
+// that knows how to render it in pretty mode. An unmapped schema falls back to the
+// generic renderer (createStreamRenderer's default) — so a new streaming endpoint
+// is never silently blank, it just gets a generic rendering until a strategy is
+// added here. This is the one place endpoint↔renderer wiring lives.
+const STREAM_RENDERERS: Record<string, string> = {
+    SearchAIAnswerStream: 'createAnswerStreamRenderer',
+    SearchAIRecommendedQuestionStream: 'createQuestionStreamRenderer',
+    AIStreamResponse: 'createAgentResponseStreamRenderer',
+};
+
 // SSE endpoint: iterate the event stream through a StreamRenderer instead of the
 // buffered printResult path. A partial answer stays on screen if the stream errors
 // mid-flight (renderer.end() flushes first, then the error prints); SIGINT flushes
@@ -1069,7 +1098,13 @@ function emitStreamingCommand(
     if (formatQp) {
         lines.push(`${I}        if (options.format === undefined) query['format'] = 'markdown';`);
     }
-    lines.push(`${I}        const stream = createStreamRenderer(options);`);
+    // Pick the pretty renderer for this endpoint's event schema; unmapped schemas
+    // fall through to createStreamRenderer's generic default.
+    const rendererFactory = route.streamEventSchema
+        ? STREAM_RENDERERS[route.streamEventSchema]
+        : undefined;
+    const rendererArg = rendererFactory ? `, ${rendererFactory}()` : '';
+    lines.push(`${I}        const stream = createStreamRenderer(options${rendererArg});`);
     lines.push(`${I}        const timeout = createRequestTimeout();`);
     lines.push(`${I}        const onSigint = () => {`);
     lines.push(`${I}            timeout?.clear();`);
@@ -1245,7 +1280,7 @@ function emitFile(tree: GroupNode, completions: Record<string, string>): string
     lines.push(`// Output formatting lives in ./output (a real, unit-tested module) rather than`);
     lines.push(`// being inlined here, so the rendering logic has a single source of truth.`);
     lines.push(
-        `import { printResult, coerceBodyFlag, coerceArrayQueryParam, explainApiError, normalizeChangesMarkdown, createStreamRenderer, createRequestTimeout, explainTimeout } from './output';`,
+        `import { printResult, coerceBodyFlag, coerceArrayQueryParam, explainApiError, normalizeChangesMarkdown, createStreamRenderer, createAnswerStreamRenderer, createQuestionStreamRenderer, createAgentResponseStreamRenderer, createRequestTimeout, explainTimeout } from './output';`,
     );
     lines.push(``);
     lines.push(`export function registerGeneratedCommands(program: Command): void {`);
