From 7c128b0f4ec8cc2c03225517af72082e32ee66c9 Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Fri, 15 May 2026 00:36:41 -0300 Subject: [PATCH] =?UTF-8?q?feat(cli):=20adicionar=20comando=20stream=20com?= =?UTF-8?q?=20inspe=C3=A7=C3=A3o=20SSE=20(Fase=202.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implementa `omniroute stream [prompt]` com suporte a: - --raw: imprime linhas SSE brutas sem parsing - --debug: timing por chunk no stderr com timestamp relativo - --save : persiste eventos em arquivo .jsonl - --output json: retorna chunks + métricas (TTFT, totalMs, tokens/s) - --responses-api: usa /v1/responses e lê campo delta - SIGINT gracioso via reader.cancel() - Métricas de TTFT e tokens/s no stderr ao final --- bin/cli/commands/registry.mjs | 2 + bin/cli/commands/stream.mjs | 166 ++++++++++++++++++++++++++++++++++ bin/cli/locales/en.json | 16 ++++ bin/cli/locales/pt-BR.json | 16 ++++ tests/unit/cli-stream.test.ts | 146 ++++++++++++++++++++++++++++++ 5 files changed, 346 insertions(+) create mode 100644 bin/cli/commands/stream.mjs create mode 100644 tests/unit/cli-stream.test.ts diff --git a/bin/cli/commands/registry.mjs b/bin/cli/commands/registry.mjs index 67f083c6ae..7b65dbf06e 100644 --- a/bin/cli/commands/registry.mjs +++ b/bin/cli/commands/registry.mjs @@ -1,4 +1,5 @@ import { registerChat } from "./chat.mjs"; +import { registerStream } from "./stream.mjs"; import { registerServe } from "./serve.mjs"; import { registerStop } from "./stop.mjs"; import { registerRestart } from "./restart.mjs"; @@ -27,6 +28,7 @@ import { registerCompletion } from "./completion.mjs"; export function registerCommands(program) { registerChat(program); + registerStream(program); registerServe(program); registerStop(program); registerRestart(program); diff --git a/bin/cli/commands/stream.mjs b/bin/cli/commands/stream.mjs new file mode 100644 index 0000000000..5ef7845603 --- /dev/null +++ b/bin/cli/commands/stream.mjs @@ -0,0 +1,166 @@ +import { appendFileSync, readFileSync } from "node:fs"; +import { apiFetch } from "../api.mjs"; +import { t } from "../i18n.mjs"; + +export function registerStream(program) { + program + .command("stream [prompt]") + .description(t("stream.description")) + .option("--file ", t("stream.file")) + .option("--stdin", t("stream.stdin")) + .option("-m, --model ", t("stream.model"), "auto") + .option("-s, --system ", t("stream.system")) + .option("--combo ", t("stream.combo")) + .option("--max-tokens ", t("stream.max_tokens"), parseInt) + .option("--responses-api", t("stream.responses_api")) + .option("--raw", t("stream.raw")) + .option("--debug", t("stream.debug")) + .option("--save ", t("stream.save")) + .action(runStreamCommand); +} + +export async function runStreamCommand(promptArg, opts, cmd) { + const globalOpts = cmd.optsWithGlobals(); + const prompt = await resolvePrompt(promptArg, opts); + + if (!prompt) { + process.stderr.write(t("stream.error.empty_prompt") + "\n"); + process.exit(2); + } + + const messages = []; + if (opts.system) messages.push({ role: "system", content: opts.system }); + messages.push({ role: "user", content: prompt }); + + const body = { + model: opts.model, + messages, + stream: true, + ...(opts.maxTokens && { max_tokens: opts.maxTokens }), + ...(opts.combo && { combo: opts.combo }), + }; + + const endpoint = opts.responsesApi ? "/v1/responses" : "/v1/chat/completions"; + + const t0 = Date.now(); + const res = await apiFetch(endpoint, { + method: "POST", + body, + acceptNotOk: true, + timeout: globalOpts.timeout, + }); + + if (!res.ok) { + const errText = await res.text().catch(() => ""); + process.stderr.write(`[error] HTTP ${res.status}: ${errText.slice(0, 200)}\n`); + process.exit(1); + } + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let firstTokenAt = null; + let totalContent = ""; + const allChunks = []; + + const sigintHandler = () => { + reader.cancel(); + process.stderr.write("\n[cancelled]\n"); + process.exit(0); + }; + process.on("SIGINT", sigintHandler); + + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + + let idx; + while ((idx = buffer.indexOf("\n")) !== -1) { + const line = buffer.slice(0, idx); + buffer = buffer.slice(idx + 1); + + if (opts.raw) { + process.stdout.write(line + "\n"); + continue; + } + + if (!line.startsWith("data:")) continue; + const payload = line.slice(5).trim(); + if (payload === "[DONE]") continue; + + let event; + try { + event = JSON.parse(payload); + } catch { + continue; + } + + if (opts.save) appendFileSync(opts.save, JSON.stringify(event) + "\n"); + if (globalOpts.output === "json") allChunks.push(event); + + if (opts.debug) { + const sinceStart = Date.now() - t0; + process.stderr.write(`[+${sinceStart}ms] ${JSON.stringify(event).slice(0, 100)}...\n`); + } + + const delta = opts.responsesApi + ? (event.delta ?? event.output_text?.delta) + : event.choices?.[0]?.delta?.content; + + if (delta) { + if (firstTokenAt === null) firstTokenAt = Date.now() - t0; + totalContent += delta; + if (globalOpts.output !== "json") process.stdout.write(delta); + } + } + } + } finally { + process.off("SIGINT", sigintHandler); + } + + const totalMs = Date.now() - t0; + const tokens = Math.ceil(totalContent.length / 4); + + if (globalOpts.output === "json") { + process.stdout.write( + JSON.stringify( + { + chunks: allChunks, + content: totalContent, + metrics: { + ttftMs: firstTokenAt, + totalMs, + approxTokens: tokens, + tokensPerSec: Math.round(tokens / (totalMs / 1000)), + }, + }, + null, + 2 + ) + "\n" + ); + } else { + if (!globalOpts.quiet) { + process.stderr.write( + `\n\n[TTFT: ${firstTokenAt}ms · Total: ${totalMs}ms · ~${tokens} tok · ~${Math.round(tokens / (totalMs / 1000))} tok/s]\n` + ); + } + process.stdout.write("\n"); + } +} + +async function resolvePrompt(arg, opts) { + if (opts.file) return readFileSync(opts.file, "utf8").trim(); + if (opts.stdin) return readStdin(); + return arg?.trim() || ""; +} + +function readStdin() { + return new Promise((resolve) => { + let buf = ""; + process.stdin.setEncoding("utf8"); + process.stdin.on("data", (c) => (buf += c)); + process.stdin.on("end", () => resolve(buf.trim())); + }); +} diff --git a/bin/cli/locales/en.json b/bin/cli/locales/en.json index e654d8acd2..e113f688a6 100644 --- a/bin/cli/locales/en.json +++ b/bin/cli/locales/en.json @@ -54,6 +54,22 @@ "noKeys": "No keys configured.", "confirmRemove": "Remove key {id}?" }, + "stream": { + "description": "Stream a chat response with SSE inspection modes", + "file": "Read prompt from file", + "stdin": "Read prompt from stdin", + "model": "Model ID (default: auto)", + "system": "System prompt", + "combo": "Force a specific combo by name", + "max_tokens": "Maximum tokens in response", + "responses_api": "Use /v1/responses instead of /v1/chat/completions", + "raw": "Print raw SSE lines as received", + "debug": "Print per-chunk timing info to stderr", + "save": "Save all SSE events to a .jsonl file", + "error": { + "empty_prompt": "Error: prompt is required (positional arg, --file, or --stdin)" + } + }, "chat": { "description": "Send a one-shot chat prompt to OmniRoute", "file": "Read prompt from file", diff --git a/bin/cli/locales/pt-BR.json b/bin/cli/locales/pt-BR.json index 9e7c2510df..4332974ee1 100644 --- a/bin/cli/locales/pt-BR.json +++ b/bin/cli/locales/pt-BR.json @@ -54,6 +54,22 @@ "noKeys": "Nenhuma chave configurada.", "confirmRemove": "Remover chave {id}?" }, + "stream": { + "description": "Transmitir resposta de chat com modos de inspeção SSE", + "file": "Ler prompt de arquivo", + "stdin": "Ler prompt da entrada padrão", + "model": "ID do modelo (padrão: auto)", + "system": "Prompt de sistema", + "combo": "Forçar um combo específico pelo nome", + "max_tokens": "Máximo de tokens na resposta", + "responses_api": "Usar /v1/responses em vez de /v1/chat/completions", + "raw": "Imprimir linhas SSE brutas como recebidas", + "debug": "Imprimir informações de timing por chunk no stderr", + "save": "Salvar todos os eventos SSE em arquivo .jsonl", + "error": { + "empty_prompt": "Erro: prompt obrigatório (argumento posicional, --file ou --stdin)" + } + }, "chat": { "description": "Enviar um prompt único ao OmniRoute", "file": "Ler prompt de arquivo", diff --git a/tests/unit/cli-stream.test.ts b/tests/unit/cli-stream.test.ts new file mode 100644 index 0000000000..a54d99ec32 --- /dev/null +++ b/tests/unit/cli-stream.test.ts @@ -0,0 +1,146 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { tmpdir } from "node:os"; +import { mkdtempSync, existsSync, readFileSync } from "node:fs"; +import { join } from "node:path"; + +function makeSseStream(lines: string[]) { + const body = lines.join("\n") + "\n"; + return new Response(body, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + +function mockStreamFetch(chunks: string[], status = 200) { + const sseLines = chunks.map((c) => `data: ${c}`); + sseLines.push("data: [DONE]"); + return () => Promise.resolve(makeSseStream(sseLines)); +} + +async function captureStdout(fn: () => Promise): Promise { + const chunks: string[] = []; + const orig = process.stdout.write.bind(process.stdout); + process.stdout.write = (chunk: string | Uint8Array) => { + chunks.push(typeof chunk === "string" ? chunk : chunk.toString()); + return true; + }; + try { + await fn(); + } finally { + process.stdout.write = orig; + } + return chunks.join(""); +} + +async function captureStderr(fn: () => Promise): Promise { + const chunks: string[] = []; + const orig = process.stderr.write.bind(process.stderr); + process.stderr.write = (chunk: string | Uint8Array) => { + chunks.push(typeof chunk === "string" ? chunk : chunk.toString()); + return true; + }; + try { + await fn(); + } finally { + process.stderr.write = orig; + } + return chunks.join(""); +} + +const DELTA1 = JSON.stringify({ choices: [{ delta: { content: "Hello" } }] }); +const DELTA2 = JSON.stringify({ choices: [{ delta: { content: " world" } }] }); + +test("runStreamCommand imprime deltas no stdout", async () => { + const origFetch = globalThis.fetch; + globalThis.fetch = mockStreamFetch([DELTA1, DELTA2]) as any; + + const { runStreamCommand } = await import("../../bin/cli/commands/stream.mjs"); + const cmd = { optsWithGlobals: () => ({ output: "text", quiet: true }) }; + const out = await captureStdout(() => runStreamCommand("hi", { model: "auto" }, cmd as any)); + + globalThis.fetch = origFetch; + assert.ok(out.includes("Hello")); + assert.ok(out.includes("world")); +}); + +test("runStreamCommand --raw imprime linhas SSE brutas", async () => { + const origFetch = globalThis.fetch; + globalThis.fetch = mockStreamFetch([DELTA1]) as any; + + const { runStreamCommand } = await import("../../bin/cli/commands/stream.mjs"); + const cmd = { optsWithGlobals: () => ({ output: "text", quiet: true }) }; + const out = await captureStdout(() => + runStreamCommand("hi", { model: "auto", raw: true }, cmd as any) + ); + + globalThis.fetch = origFetch; + assert.ok(out.includes("data:")); +}); + +test("runStreamCommand --output json retorna chunks e métricas", async () => { + const origFetch = globalThis.fetch; + globalThis.fetch = mockStreamFetch([DELTA1, DELTA2]) as any; + + const { runStreamCommand } = await import("../../bin/cli/commands/stream.mjs"); + const cmd = { optsWithGlobals: () => ({ output: "json", quiet: true }) }; + const out = await captureStdout(() => runStreamCommand("hi", { model: "auto" }, cmd as any)); + + globalThis.fetch = origFetch; + const parsed = JSON.parse(out); + assert.ok(Array.isArray(parsed.chunks)); + assert.equal(parsed.chunks.length, 2); + assert.ok(parsed.content.includes("Hello")); + assert.ok(typeof parsed.metrics.totalMs === "number"); +}); + +test("runStreamCommand --save grava eventos em arquivo", async () => { + const tmpDir = mkdtempSync(join(tmpdir(), "stream-test-")); + const savePath = join(tmpDir, "events.jsonl"); + const origFetch = globalThis.fetch; + globalThis.fetch = mockStreamFetch([DELTA1, DELTA2]) as any; + + const { runStreamCommand } = await import("../../bin/cli/commands/stream.mjs"); + const cmd = { optsWithGlobals: () => ({ output: "text", quiet: true }) }; + await captureStdout(() => runStreamCommand("hi", { model: "auto", save: savePath }, cmd as any)); + + globalThis.fetch = origFetch; + assert.ok(existsSync(savePath)); + const lines = readFileSync(savePath, "utf8").trim().split("\n"); + assert.equal(lines.length, 2); + assert.ok(JSON.parse(lines[0]).choices); +}); + +test("runStreamCommand --debug imprime timing no stderr", async () => { + const origFetch = globalThis.fetch; + globalThis.fetch = mockStreamFetch([DELTA1]) as any; + + const { runStreamCommand } = await import("../../bin/cli/commands/stream.mjs"); + const cmd = { optsWithGlobals: () => ({ output: "text", quiet: true }) }; + const err = await captureStderr(() => + captureStdout(() => runStreamCommand("hi", { model: "auto", debug: true }, cmd as any)) + ); + + globalThis.fetch = origFetch; + assert.ok(err.includes("[+")); +}); + +test("runStreamCommand usa /v1/responses com --responses-api", async () => { + const respDelta = JSON.stringify({ delta: "Hi there" }); + let capturedUrl = ""; + const origFetch = globalThis.fetch; + globalThis.fetch = ((url: string, init: any) => { + capturedUrl = url; + return Promise.resolve(makeSseStream([`data: ${respDelta}`, "data: [DONE]"])); + }) as any; + + const { runStreamCommand } = await import("../../bin/cli/commands/stream.mjs"); + const cmd = { optsWithGlobals: () => ({ output: "text", quiet: true }) }; + const out = await captureStdout(() => + runStreamCommand("hi", { model: "auto", responsesApi: true }, cmd as any) + ); + + globalThis.fetch = origFetch; + assert.ok(capturedUrl.includes("/v1/responses")); + assert.ok(out.includes("Hi there")); +});