diff --git a/.changeset/fn-7722-grok-cli-streaming.md b/.changeset/fn-7722-grok-cli-streaming.md new file mode 100644 index 0000000000..f76b62dae2 --- /dev/null +++ b/.changeset/fn-7722-grok-cli-streaming.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": minor +--- + +summary: Grok can now run through the Grok CLI's NDJSON stream, so CLI-authenticated setups need no Fusion-visible API key. +category: feature +dev: GrokRuntimeAdapter.promptWithFallback now spawns `grok --prompt --format json`, parses the NDJSON event stream (new src/stream-parser.ts, fixture-tested), and drives onText/onThinking — replacing the FN-7715 no-op. Direct xAI OpenAI-compatible path (FN-7711/FN-7714) is unchanged and remains the default; end-to-end runtimeHint="grok" routing is a follow-up. Contract captured in docs/grok-cli-contract.md. diff --git a/docs/grok-cli-contract.md b/docs/grok-cli-contract.md new file mode 100644 index 0000000000..2739ca32d0 --- /dev/null +++ b/docs/grok-cli-contract.md @@ -0,0 +1,200 @@ +# Grok CLI Contract (FN-7722) + +Date: 2026-07-09 + + + +## Research method + +- `fn_web_fetch` against the canonical upstream repository + (https://github.com/superagent-ai/grok-cli), specifically: + - `README.md` (headless-mode overview, feature summary). + - `src/index.ts` (commander CLI argument parsing — the exact flag + spellings and headless dispatch). + - `src/headless/output.ts` (the actual NDJSON event emitter — the + authoritative schema source, not just docs prose). + - `src/headless/output.test.ts` (fixture-level confirmation of the emitted + JSONL shapes, used as ground truth for this plugin's own fixture tests). +- No live `grok` binary was invoked; no field name or flag spelling in this + document is guessed — every claim below traces to one of the four files + above. Raw captured research (queries + verbatim schema) is preserved as + this task's `research` task document (`fn_task_document_read` key + `research` on FN-7722). + +## Confirmed non-interactive invocation + +```bash +grok --prompt "" --format json +# short flags: +grok -p "" --format json +``` + +- `-p, --prompt ` — run a single prompt headlessly, then exit. +- `--format ` — headless output format, `text` (default) or `json`; + invalid values are rejected by commander's `InvalidArgumentError` + (`parseHeadlessOutputFormat`/`isHeadlessOutputFormat` in `src/index.ts`). +- Useful companion flags confirmed in the same `program.option(...)` chain: + `-d, --directory ` (cwd), `-m, --model `, `-s, --session ` + (resume a saved session, or `latest`), `-k, --api-key ` (inline key). +- `--format json` output is **newline-delimited JSON (NDJSON/JSONL)** — one + JSON object per line — not a single JSON document. This is directly + confirmed by `createHeadlessJsonlEmitter()`'s `jsonLine()` helper in + `src/headless/output.ts`, which appends `\n` after each `JSON.stringify`. + +## Verified NDJSON event schema (verbatim) + +Source: `HeadlessJsonEvent` union type in `src/headless/output.ts`. + +```ts +type HeadlessJsonEvent = + | { type: "step_start"; sessionID?: string; stepNumber: number; timestamp: number } + | { type: "text"; sessionID?: string; stepNumber: number; text: string; timestamp: number } + | { + type: "tool_use"; + sessionID?: string; + stepNumber: number; + timestamp: number; + toolCall: ToolCall; + toolResult: ToolResult; + timing?: { startedAt?: number; finishedAt?: number; durationMs?: number }; + } + | { + type: "step_finish"; + sessionID?: string; + stepNumber: number; + timestamp: number; + finishReason: string; + usage: { inputTokens?: number; outputTokens?: number; totalTokens?: number; costUsdTicks?: number }; + } + | { type: "error"; sessionID?: string; message: string; timestamp: number }; +``` + +Notes: + +- `sessionID` appears on every event type when a session id is available + (`agent.getSessionId()`); it is simply absent from the JSON object + otherwise (not `null`). +- `text` events are per-step, buffered assistant content — one `text` event + per step carrying the accumulated text for that step, flushed either right + before a tool-triggering `step_finish` or inline with a tool-less + `step_finish`. +- **No `thinking`/`reasoning` NDJSON event exists.** The underlying + `StreamChunk` union used internally does carry a `"reasoning"` chunk type, + but `createHeadlessJsonlEmitter().consumeChunk()` explicitly no-ops on it + (`case "reasoning": break;` in `src/headless/output.ts`) — reasoning + content is never surfaced through `--format json`. This is a **confirmed + absence**, not `upstream-pending-verification`: the Grok streaming adapter + therefore drives `onText` only; there is no `onThinking` signal to bridge + for this CLI path today. +- There is **no explicit terminal `done`/`result` event type**. A prompt run + can contain multiple `step_start`/`step_finish` pairs (multi-round tool + use); the authoritative "the run is over" signal is the headless process's + stdout stream ending (readline `close`) / subprocess exit, mirroring how + the Droid CLI adapter treats subprocess `close` as terminal. `error` events + (`{ type: "error", message, timestamp }`) can also appear inline without + necessarily ending the process. +- A **fatal, pre-JSON failure** (e.g. missing API key) is not a JSON line at + all: `src/index.ts`'s `requireApiKey()` writes a plain `console.error(...)` + line to stderr and calls `process.exit(1)` before any NDJSON is emitted. + Consumers must therefore also treat a non-zero exit with no JSON output as + a distinct failure mode from a well-formed `error` event. + +## Auth / readiness + +- **The `grok` CLI owns authentication end-to-end for CLI-routed execution.** + `runHeadless()` in `src/index.ts` is only reached via + `requireApiKey(config.apiKey)`, which resolves the key from (in order via + `resolveConfig`/`getApiKey()`): `-k/--api-key` flag, `GROK_API_KEY` env var, + project `.env`, or `~/.grok/user-settings.json`'s `apiKey` field. If none + resolve, the CLI itself exits 1 with an actionable error — Fusion does not + need to pass, see, or validate a key for this path to work, as long as the + operator's `grok` install already has one configured by any of those + methods. +- **Auth implication for this task:** because CLI-routed model selections let + the `grok` binary own both auth and inference, the direct-endpoint + `GROK_API_KEY` Fusion-visibility requirement established by FN-7711 + (built-in `xai`/`openai-completions` provider) and FN-7714 (hydrating + `GROK_API_KEY` from `~/.grok/user-settings.json` when the env var is unset) + becomes **unnecessary for CLI-routed selections specifically**. It remains + necessary and unchanged for the direct xAI OpenAI-compatible path, which + stays the default (see "What stays unchanged" below). +- This mirrors FN-7716's separate finding that Grok CLI *readiness* (probe/ + auth-status surfacing) does not require Fusion to see a key either — that + surface (`probe.ts`, `register-auth-routes.ts`, + `GrokCliProviderCard.tsx`) is out of scope for this task and is not + modified here. + +## Wiring gap (recorded, not closed by this task) + +`packages/engine/src/runtime-resolution.ts`'s `resolveRuntime()` only reaches +a plugin runtime adapter (like `GrokRuntimeAdapter`) when +`runtimeConfig.runtimeHint === "grok"`. A repo-wide grep at Step 0 of this +task confirmed **nothing in the product sets `runtimeHint` to `"grok"` +today** (task/agent config, settings, or otherwise) — the same wiring gap +FN-7715's stale comment already noted. This task lands the adapter +implementation and its tests, but does **not** wire an end-to-end path that +exercises it (no product code sets `runtimeHint: "grok"`, and no settings +toggle exists to prefer CLI execution over the direct endpoint). That wiring +is filed as a follow-up task (see `fn_task_create` entries linked from this +task). + +## Decision + +**Route Grok execution through the CLI: YES, as a scoped, additive +`GrokRuntimeAdapter` implementation.** + +Rationale: + +- The non-interactive contract is fully pinned to primary source code + (`src/index.ts` CLI parsing + `src/headless/output.ts` emitter + + `src/headless/output.test.ts` fixtures), not just README prose — this + clears the External Integration Evidence bar and the "testable from + fixture lines without a live binary" bar from the task mission — the + parser can be fixture-tested exactly like the Droid plugin's + `stream-parser.ts`, with no live-binary dependency in tests. +- The event schema is simple (`step_start` / `text` / `tool_use` / + `step_finish` / `error`) and text-only for this scoped adapter (no + `thinking` event exists to bridge), so the implementation stays narrow: a + resilient NDJSON line parser plus an `onText` bridge, deliberately leaving + tool-call/break-early bridging as a documented follow-up (the Droid + adapter's much larger `provider.ts` is the effort ceiling, not the target + shape). +- It is fully reversible: the adapter is only reachable via + `runtimeHint === "grok"`, which nothing sets today, so landing it carries + no behavioral change to any exercised path. + +## What stays unchanged + +- The **direct xAI OpenAI-compatible streaming path** (base URL + `https://api.x.ai/v1`, api type `openai-completions`, `GROK_API_KEY` + sourced per FN-7711/FN-7714) remains the default, exercised Grok execution + path. This task does not touch `packages/core/src/grok-provider.ts` or + `packages/engine/src/pi.ts`. +- FN-7716's probe/auth-readiness surface (`probe.ts`, + `register-auth-routes.ts`, `GrokCliProviderCard.tsx`) is untouched by this + task. +- End-to-end routing (making the product actually set + `runtimeHint === "grok"`, or adding a settings toggle to prefer the CLI + over the direct endpoint) is explicitly out of scope here and is filed as + a follow-up task. + +## Follow-ups filed from this task + +See the task's `fn_task_create` calls (linked from FN-7722) for: + +1. End-to-end routing wiring — actually setting `runtimeHint === "grok"` (or + a settings toggle preferring the CLI) so `GrokRuntimeAdapter` is + exercised in a real execution path. +2. Full tool-call/break-early bridging for `tool_use` NDJSON events, if a + future need for Grok-CLI-driven tool execution arises (out of scope for + the scoped text/no-thinking adapter landed here). diff --git a/plugins/fusion-plugin-grok-runtime/README.md b/plugins/fusion-plugin-grok-runtime/README.md index 513be89425..743e377acc 100644 --- a/plugins/fusion-plugin-grok-runtime/README.md +++ b/plugins/fusion-plugin-grok-runtime/README.md @@ -43,6 +43,34 @@ binary on PATH — Fusion never downloads or bundles the CLI itself. deduplicated. Output that happens to be JSON is tolerated defensively even though the CLI is not known to emit it. +## CLI streaming execution path (FN-7722) + +In addition to model discovery/probe, this plugin's `GrokRuntimeAdapter` can +stream a real Grok response through the CLI itself: + +```bash +grok --prompt "" --format json +``` + +- `--format json` emits newline-delimited JSON (NDJSON) — one JSON object + per line — with event types `step_start`, `text`, `tool_use`, + `step_finish`, and `error` (verified against upstream source, not just + docs prose; see `docs/grok-cli-contract.md`). +- The adapter parses that stream (`src/stream-parser.ts`) and drives + `onText` as `text` events arrive. There is no `thinking`/`reasoning` event + in the verified schema, so `onThinking` is never invoked for this path. +- **Auth implication:** because the `grok` binary resolves its own + credentials for this path (env var, project `.env`, `grok -k`, or + `~/.grok/user-settings.json`), a CLI-routed selection needs **no + Fusion-visible `GROK_API_KEY`** — unlike the direct xAI + OpenAI-compatible streaming path (`https://api.x.ai/v1`), which still + requires one. +- This adapter is only reached when an agent's + `runtimeConfig.runtimeHint === "grok"`. Nothing in the product sets that + today — routing Grok execution through the CLI end-to-end (vs. the direct + xAI endpoint, which remains the default) is tracked as a follow-up. See + `docs/grok-cli-contract.md` for the full contract and decision record. + ## Enable via Settings → Authentication 1. Install the `grok` CLI and authenticate it by any method it supports diff --git a/plugins/fusion-plugin-grok-runtime/src/__tests__/runtime-adapter.test.ts b/plugins/fusion-plugin-grok-runtime/src/__tests__/runtime-adapter.test.ts index d5e063b1f5..fda1ff36ca 100644 --- a/plugins/fusion-plugin-grok-runtime/src/__tests__/runtime-adapter.test.ts +++ b/plugins/fusion-plugin-grok-runtime/src/__tests__/runtime-adapter.test.ts @@ -1,5 +1,27 @@ -import { describe, expect, it } from "vitest"; +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { GrokRuntimeAdapter } from "../runtime-adapter.js"; +import type { GrokStreamProcess } from "../cli-stream.js"; + +/* +FNXC:GrokCli 2026-07-09-00:00: +FN-7722: replaces FN-7715's "intentional no-op" assertion. `promptWithFallback` +is now a real NDJSON streaming implementation; these tests inject a FAKE +stdout stream (no live binary, no real subprocess spawn) through the +constructor's `spawn` seam and feed verified-shape NDJSON fixture lines +(docs/grok-cli-contract.md), asserting onText fires in order and the promise +resolves on close/error. Uses fake timers for the lifecycle timeout paths +per AGENTS.md "Do Not Add Slow Tests". +*/ + +function makeFakeProc(): { proc: GrokStreamProcess; stdout: PassThrough; kill: ReturnType } { + const stdout = new PassThrough(); + const emitter = new EventEmitter(); + const kill = vi.fn(); + const proc = Object.assign(emitter, { stdout, kill }) as unknown as GrokStreamProcess; + return { proc, stdout, kill }; +} describe("GrokRuntimeAdapter", () => { it("creates a session with default model fallback", async () => { @@ -9,22 +31,109 @@ describe("GrokRuntimeAdapter", () => { expect(result.session.systemPrompt).toBe("sys"); }); - // FN-7715: promptWithFallback is an INTENTIONAL no-op — Grok streaming - // flows through the pi/xAI OpenAI-compatible path registered by FN-7711, - // not through this plugin runtime adapter (which is only reached via - // runtimeConfig.runtimeHint === "grok", which nothing in the product - // sets). This module imports no process-spawning seam (compare - // process-manager.ts's `runGrokCommand`), so this asserts the intentional - // no-op contract at the only observable boundary: it resolves without - // throwing and returns no value, taking no action. - it("promptWithFallback is an intentional no-op: resolves without throwing, returns undefined", async () => { - const adapter = new GrokRuntimeAdapter(); + it("streams onText for each text NDJSON event in order and resolves on close", async () => { + const { proc, stdout } = makeFakeProc(); + const spawn = vi.fn().mockReturnValue(proc); + const adapter = new GrokRuntimeAdapter({ spawn }); - await expect(adapter.promptWithFallback()).resolves.toBeUndefined(); + const onText = vi.fn(); + const { session } = await adapter.createSession({ onText }); + + const promise = adapter.promptWithFallback(session, "hello grok"); + + stdout.write(`${JSON.stringify({ type: "step_start", stepNumber: 1, timestamp: 1 })}\n`); + stdout.write(`${JSON.stringify({ type: "text", stepNumber: 1, text: "hel", timestamp: 2 })}\n`); + stdout.write(`${JSON.stringify({ type: "text", stepNumber: 1, text: "lo!", timestamp: 3 })}\n`); + stdout.write( + `${JSON.stringify({ type: "step_finish", stepNumber: 1, timestamp: 4, finishReason: "stop", usage: {} })}\n`, + ); + proc.emit("close", 0, null); + + await promise; + + expect(spawn).toHaveBeenCalledWith("grok", "hello grok", expect.objectContaining({})); + expect(onText.mock.calls.map((c) => c[0])).toEqual(["hel", "lo!"]); + }); + + it("skips malformed/unrecognized lines without invoking onText and without throwing", async () => { + const { proc, stdout } = makeFakeProc(); + const spawn = vi.fn().mockReturnValue(proc); + const adapter = new GrokRuntimeAdapter({ spawn }); + const onText = vi.fn(); + const { session } = await adapter.createSession({ onText }); + + const promise = adapter.promptWithFallback(session, "hi"); + + stdout.write("[SandboxDebug] booting\n"); + stdout.write("{not valid json\n"); + stdout.write(`${JSON.stringify({ type: "tool_use", stepNumber: 1, timestamp: 5, toolCall: {}, toolResult: {} })}\n`); + proc.emit("close", 0, null); + + await expect(promise).resolves.toBeUndefined(); + expect(onText).not.toHaveBeenCalled(); + }); + + it("resolves (never rejects) when the subprocess emits an error", async () => { + const { proc } = makeFakeProc(); + const spawn = vi.fn().mockReturnValue(proc); + const adapter = new GrokRuntimeAdapter({ spawn }); + const { session } = await adapter.createSession({}); + + const promise = adapter.promptWithFallback(session, "hi"); + proc.emit("error", new Error("ENOENT")); + + await expect(promise).resolves.toBeUndefined(); + }); + + it("never invokes onThinking: the verified grok-cli NDJSON schema has no thinking/reasoning event", async () => { + const { proc, stdout } = makeFakeProc(); + const spawn = vi.fn().mockReturnValue(proc); + const adapter = new GrokRuntimeAdapter({ spawn }); + const onThinking = vi.fn(); + const { session } = await adapter.createSession({ onThinking }); + + const promise = adapter.promptWithFallback(session, "hi"); + stdout.write(`${JSON.stringify({ type: "text", stepNumber: 1, text: "hi", timestamp: 1 })}\n`); + proc.emit("close", 0, null); + + await promise; + expect(onThinking).not.toHaveBeenCalled(); + }); + + describe("lifecycle timeouts (fake timers)", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it("kills the subprocess and resolves if no stdout line arrives within the cold-start ceiling", async () => { + const { proc, kill } = makeFakeProc(); + const spawn = vi.fn().mockReturnValue(proc); + const adapter = new GrokRuntimeAdapter({ spawn }); + const { session } = await adapter.createSession({}); + + const promise = adapter.promptWithFallback(session, "hi"); + await vi.advanceTimersByTimeAsync(60_000); + + await promise; + expect(kill).toHaveBeenCalledWith("SIGKILL"); + }); + }); + + it("resolves without throwing if the injected spawn function throws synchronously", async () => { + const spawn = vi.fn().mockImplementation(() => { + throw new Error("spawn ENOENT"); + }); + const adapter = new GrokRuntimeAdapter({ spawn }); + const { session } = await adapter.createSession({}); + + await expect(adapter.promptWithFallback(session, "hi")).resolves.toBeUndefined(); }); it("describeModel formats grok prefix", () => { const adapter = new GrokRuntimeAdapter(); - expect(adapter.describeModel({ model: "grok/pro" })).toBe("grok/grok/pro"); + expect(adapter.describeModel({ model: "grok/pro" } as never)).toBe("grok/grok/pro"); }); }); diff --git a/plugins/fusion-plugin-grok-runtime/src/__tests__/stream-parser.test.ts b/plugins/fusion-plugin-grok-runtime/src/__tests__/stream-parser.test.ts new file mode 100644 index 0000000000..51674cd1de --- /dev/null +++ b/plugins/fusion-plugin-grok-runtime/src/__tests__/stream-parser.test.ts @@ -0,0 +1,79 @@ +import { describe, expect, it } from "vitest"; +import { parseLine } from "../stream-parser.js"; + +/* +FNXC:GrokCli 2026-07-09-00:00: +FN-7722: fixture lines below are copied verbatim in shape from upstream +grok-cli's `src/headless/output.test.ts` (the authoritative fixture-level +confirmation of the JSONL emitter's output), not invented. See +docs/grok-cli-contract.md for the full verified schema. +*/ + +describe("parseLine (Grok CLI NDJSON)", () => { + it("parses a step_start event", () => { + const line = JSON.stringify({ type: "step_start", sessionID: "sess-1", stepNumber: 1, timestamp: 100 }); + expect(parseLine(line)).toEqual({ type: "step_start", sessionID: "sess-1", stepNumber: 1, timestamp: 100 }); + }); + + it("parses a text delta event", () => { + const line = JSON.stringify({ type: "text", sessionID: "sess-1", stepNumber: 1, text: "hello", timestamp: 150 }); + const parsed = parseLine(line); + expect(parsed).toEqual({ type: "text", sessionID: "sess-1", stepNumber: 1, text: "hello", timestamp: 150 }); + }); + + it("parses a tool_use event", () => { + const line = JSON.stringify({ + type: "tool_use", + sessionID: "sess-1", + stepNumber: 1, + timestamp: 130, + toolCall: { id: "tc-1", type: "function", function: { name: "bash", arguments: "{}" } }, + toolResult: { success: true, output: "ok" }, + timing: { startedAt: 110, finishedAt: 130, durationMs: 20 }, + }); + const parsed = parseLine(line); + expect(parsed?.type).toBe("tool_use"); + expect((parsed as { toolCall: { function: { name: string } } }).toolCall.function.name).toBe("bash"); + }); + + it("parses a terminal step_finish event", () => { + const line = JSON.stringify({ + type: "step_finish", + sessionID: "sess-1", + stepNumber: 1, + timestamp: 200, + finishReason: "stop", + usage: { inputTokens: 1, outputTokens: 2, totalTokens: 3 }, + }); + const parsed = parseLine(line); + expect(parsed).toMatchObject({ type: "step_finish", finishReason: "stop" }); + }); + + it("parses an error event", () => { + const line = JSON.stringify({ type: "error", sessionID: "err-session", message: "boom", timestamp: 1 }); + expect(parseLine(line)).toEqual({ type: "error", sessionID: "err-session", message: "boom", timestamp: 1 }); + }); + + it("skips an empty line", () => { + expect(parseLine("")).toBeNull(); + expect(parseLine(" ")).toBeNull(); + }); + + it("skips non-JSON debug output", () => { + expect(parseLine("[SandboxDebug] booting shuru vm")).toBeNull(); + }); + + it("skips malformed JSON without throwing", () => { + expect(() => parseLine("{not valid json")).not.toThrow(); + expect(parseLine("{not valid json")).toBeNull(); + }); + + it("skips a JSON object with an unrecognized/missing type", () => { + expect(parseLine(JSON.stringify({ foo: "bar" }))).toBeNull(); + expect(parseLine(JSON.stringify({ type: "some_future_event", data: 1 }))).toBeNull(); + }); + + it("skips a JSON array (not an object)", () => { + expect(parseLine(JSON.stringify([{ type: "text" }]))).toBeNull(); + }); +}); diff --git a/plugins/fusion-plugin-grok-runtime/src/cli-stream.ts b/plugins/fusion-plugin-grok-runtime/src/cli-stream.ts new file mode 100644 index 0000000000..7784b02965 --- /dev/null +++ b/plugins/fusion-plugin-grok-runtime/src/cli-stream.ts @@ -0,0 +1,52 @@ +import { spawn, type ChildProcessByStdio } from "node:child_process"; +import type { Readable } from "node:stream"; + +/* +FNXC:GrokCli 2026-07-09-00:00: +FN-7722: streaming spawn seam for `grok --prompt --format json` +(verified contract: docs/grok-cli-contract.md). `cli-spawn.ts`'s +`runGrokCommand` buffers stdout/stderr until process close, which cannot +host line-by-line NDJSON streaming for the real-time onText bridge this +adapter needs. This module is the line-streaming counterpart: same +Windows-shell handling as `cli-spawn.ts` (Windows `grok.cmd`/`grok.bat` PATH +shims need shell:true; Unix/macOS stay direct-spawned), same "no raw +detached spawn/nohup" foreground-pipe pattern. Mirrors the Droid plugin's +`spawnDroid` (process-manager.ts) shape. +*/ + +export type GrokStreamProcess = ChildProcessByStdio; + +export interface SpawnGrokStreamOptions { + cwd?: string; + signal?: AbortSignal; +} + +/** + * Spawn `grok --prompt --format json [--model ] [--directory ]` + * with piped stdio for line-by-line NDJSON consumption via readline. + * + * Does not read/buffer output itself — callers attach a `readline` interface + * to `proc.stdout` (see `runtime-adapter.ts`). + */ +export function spawnGrokStream(binary: string, prompt: string, options?: SpawnGrokStreamOptions): GrokStreamProcess { + const args: string[] = ["--prompt", prompt, "--format", "json"]; + if (options?.cwd) { + args.push("--directory", options.cwd); + } + + return spawn(binary, args, { + cwd: options?.cwd, + stdio: ["ignore", "pipe", "pipe"], + shell: process.platform === "win32", + signal: options?.signal, + }) as GrokStreamProcess; +} + +/** Force-kill a Grok CLI streaming subprocess. Best-effort; never throws. */ +export function forceKillGrokStream(proc: GrokStreamProcess): void { + try { + proc.kill("SIGKILL"); + } catch { + // best effort + } +} diff --git a/plugins/fusion-plugin-grok-runtime/src/runtime-adapter.ts b/plugins/fusion-plugin-grok-runtime/src/runtime-adapter.ts index 1bc56e7fe6..191eedc38b 100644 --- a/plugins/fusion-plugin-grok-runtime/src/runtime-adapter.ts +++ b/plugins/fusion-plugin-grok-runtime/src/runtime-adapter.ts @@ -1,39 +1,165 @@ -export class GrokRuntimeAdapter { +import { createInterface } from "node:readline"; +import { forceKillGrokStream, spawnGrokStream, type GrokStreamProcess, type SpawnGrokStreamOptions } from "./cli-stream.js"; +import { parseLine } from "./stream-parser.js"; +import type { AgentRuntime, AgentRuntimeOptions, AgentSession, AgentSessionResult, GrokSession } from "./types.js"; + +/* +FNXC:GrokCli 2026-07-09-00:00: +FN-7722: replaces the FN-7715 intentional no-op. Upstream grok-cli DOES +document and implement a non-interactive `grok --prompt --format +json` NDJSON event stream (verified against primary source, not just docs +prose: src/index.ts's CLI parsing + src/headless/output.ts's +`createHeadlessJsonlEmitter` + its fixture tests). Contract captured in +docs/grok-cli-contract.md. This adapter spawns that command via the +`cli-stream` seam, parses NDJSON via `stream-parser.parseLine`, and drives +`onText` as `text` events arrive. Scoped deliberately narrow, mirroring the +Droid plugin's parser+text-bridge pattern but NOT its full tool-call/ +break-early machinery: the verified schema has no thinking/reasoning event +(onThinking is therefore never invoked here — kept only for AgentRuntime +interface parity), and tool_use bridging is filed as a follow-up (see +docs/grok-cli-contract.md "Follow-ups"). This adapter is only reached when +an agent's `runtimeConfig.runtimeHint === "grok"`, which nothing in the +product sets today (recorded as the wiring gap in the contract doc) — this +task lands the adapter without wiring an end-to-end exercised path. +*/ + +/** + * Cold-start ceiling: if `grok --prompt --format json` produces no stdout + * line within this window, treat it as a hung/failed subprocess and resolve + * (never reject — mirrors the Droid adapter's resolve-on-error lifecycle so + * pi always gets a well-formed, if empty, result instead of an unhandled + * rejection). + */ +const FIRST_LINE_TIMEOUT_MS = 60_000; + +/** + * Inactivity safety net: kill the subprocess if no stdout line arrives for + * this long after the first line. Generous ceiling mirroring the Droid + * adapter's rationale — the caller (Fusion's stuck-task detection / abort + * signal) is the authoritative "this session is stuck" source; this is a + * last-resort guard for a catastrophically hung `grok` process. + */ +const INACTIVITY_TIMEOUT_MS = 30 * 60_000; + +export interface GrokRuntimeAdapterOptions { + /** Binary name/path to invoke. Defaults to "grok" (PATH resolution). */ + binary?: string; + /** Injectable spawn seam for tests — defaults to the real `spawnGrokStream`. */ + spawn?: (binary: string, prompt: string, options?: SpawnGrokStreamOptions) => GrokStreamProcess; +} + +export class GrokRuntimeAdapter implements AgentRuntime { readonly id = "grok"; readonly name = "Grok Runtime"; + private readonly binary: string; + private readonly spawnFn: (binary: string, prompt: string, options?: SpawnGrokStreamOptions) => GrokStreamProcess; - async createSession(options: { defaultModelId?: string; systemPrompt?: string }) { - return { - session: { - model: options.defaultModelId ?? "grok/default", - systemPrompt: options.systemPrompt, - messages: [], + constructor(options?: GrokRuntimeAdapterOptions) { + this.binary = options?.binary ?? "grok"; + this.spawnFn = options?.spawn ?? spawnGrokStream; + } + + async createSession(options: { defaultModelId?: string; systemPrompt?: string; onText?: (text: string) => void; onThinking?: (text: string) => void } = {}): Promise { + const model = options.defaultModelId ?? "grok/default"; + const session: GrokSession = { + model, + systemPrompt: options.systemPrompt, + messages: [], + sessionId: undefined, + lastModelDescription: `grok/${model}`, + callbacks: { + onText: options.onText, + onThinking: options.onThinking, }, - sessionFile: undefined, }; + return { session, sessionFile: undefined }; } - /* - FNXC:GrokCli 2026-07-09-00:00: - FN-7715: this is an INTENTIONAL no-op, not unfinished work. FN-7711 already - routes `grok-cli/*` model selections through the standard pi/openai-completions - streaming path against https://api.x.ai/v1 — that is the real, exercised Grok - streaming path. The `grok` binary (see provider.ts / process-manager.ts) is - wired for discovery/probe only (`grok models`, `grok --version`); there is no - documented non-interactive prompt/stream subcommand to invoke here, and - inventing one would violate the external-integration-evidence policy. This - adapter's `promptWithFallback` is only reached when an agent's - `runtimeConfig.runtimeHint === "grok"`, which nothing in the product sets - today. Mirrors the identical intentional stub in the sibling Cursor plugin - (`fusion-plugin-cursor-runtime/src/runtime-adapter.ts`, TODO(FN-3396)). If a - stable non-interactive `grok` CLI streaming contract is confirmed upstream in - the future, a follow-up task can revisit this. - */ - async promptWithFallback(): Promise { - return; + async promptWithFallback(session: AgentSession, prompt: string, options?: AgentRuntimeOptions): Promise { + const grokSession = session as GrokSession; + const cwd = options?.cwd; + const signal = options?.signal; + + return new Promise((resolve) => { + let proc: GrokStreamProcess; + try { + proc = this.spawnFn(this.binary, prompt, { cwd, signal }); + } catch { + // Spawn threw synchronously (e.g. binary not found without shell + // resolution) — resolve, never reject, matching the CLI-adapter + // contract of always producing a well-formed (if empty) result. + resolve(); + return; + } + + let settled = false; + let firstLineReceived = false; + let firstLineTimer: NodeJS.Timeout | undefined; + let inactivityTimer: NodeJS.Timeout | undefined; + + const finish = () => { + if (settled) return; + settled = true; + if (firstLineTimer) clearTimeout(firstLineTimer); + if (inactivityTimer) clearTimeout(inactivityTimer); + resolve(); + }; + + const resetInactivityTimer = () => { + if (inactivityTimer) clearTimeout(inactivityTimer); + inactivityTimer = setTimeout(() => { + forceKillGrokStream(proc); + finish(); + }, INACTIVITY_TIMEOUT_MS); + }; + + firstLineTimer = setTimeout(() => { + if (firstLineReceived) return; + forceKillGrokStream(proc); + finish(); + }, FIRST_LINE_TIMEOUT_MS); + + const rl = createInterface({ input: proc.stdout, crlfDelay: Infinity, terminal: false }); + + rl.on("line", (line: string) => { + if (!firstLineReceived) { + firstLineReceived = true; + if (firstLineTimer) clearTimeout(firstLineTimer); + } + resetInactivityTimer(); + + const event = parseLine(line); + if (!event) return; + + if (event.type === "text") { + grokSession.callbacks.onText?.(event.text); + } + // step_start / tool_use / step_finish / error: intentionally not + // bridged by this scoped adapter (text-only). tool_use bridging is + // a follow-up (docs/grok-cli-contract.md). + }); + + proc.on("error", () => { + finish(); + }); + + proc.on("close", () => { + try { + rl.close(); + } catch { + // already closed + } + finish(); + }); + + rl.on("close", () => { + finish(); + }); + }); } - describeModel(session: { model?: string }) { - return `grok/${session.model ?? "default"}`; + describeModel(session: AgentSession): string { + const grokSession = session as GrokSession; + return grokSession.lastModelDescription || `grok/${grokSession.model ?? "default"}`; } } diff --git a/plugins/fusion-plugin-grok-runtime/src/stream-parser.ts b/plugins/fusion-plugin-grok-runtime/src/stream-parser.ts new file mode 100644 index 0000000000..5a5fcc29e4 --- /dev/null +++ b/plugins/fusion-plugin-grok-runtime/src/stream-parser.ts @@ -0,0 +1,54 @@ +import type { GrokNdjsonEvent } from "./types.js"; + +/* +FNXC:GrokCli 2026-07-09-00:00: +FN-7722: `grok --prompt --format json` emits newline-delimited JSON +(one JSON object per line) per the verified upstream contract captured in +docs/grok-cli-contract.md (source: src/headless/output.ts's +`createHeadlessJsonlEmitter` / `HeadlessJsonEvent`). This parser mirrors the +Droid plugin's `stream-parser.ts` shape and resilience contract: it never +throws. Debug noise, empty lines, and malformed/unrecognized JSON all return +null so the streaming pipeline can safely skip them and continue. +*/ + +const KNOWN_EVENT_TYPES = new Set(["step_start", "text", "tool_use", "step_finish", "error"]); + +/** + * Parse a single NDJSON line from `grok --prompt --format json` stdout into a + * typed event, or null when the line should be skipped (empty, non-JSON + * debug noise, malformed JSON, or a JSON object whose `type` isn't one of + * the five verified event types). + */ +export function parseLine(line: string): GrokNdjsonEvent | null { + const trimmed = line.trim(); + + // Skip empty lines + if (!trimmed) { + return null; + } + + // Skip non-JSON lines (e.g. any stray debug/log output not part of the JSONL stream) + if (!trimmed.startsWith("{")) { + return null; + } + + let parsed: unknown; + try { + parsed = JSON.parse(trimmed); + } catch { + console.error("Failed to parse Grok CLI NDJSON line:", trimmed); + return null; + } + + // Validate that the parsed result is a non-null object (not array, not primitive) + if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) { + return null; + } + + const candidate = parsed as { type?: unknown }; + if (typeof candidate.type !== "string" || !KNOWN_EVENT_TYPES.has(candidate.type)) { + return null; + } + + return parsed as GrokNdjsonEvent; +} diff --git a/plugins/fusion-plugin-grok-runtime/src/types.ts b/plugins/fusion-plugin-grok-runtime/src/types.ts index a2e13f53a9..f8963cbf66 100644 --- a/plugins/fusion-plugin-grok-runtime/src/types.ts +++ b/plugins/fusion-plugin-grok-runtime/src/types.ts @@ -1,3 +1,123 @@ +/* +FNXC:GrokCli 2026-07-09-00:00: +FN-7722: additive types for the real (non-no-op) `GrokRuntimeAdapter` +streaming implementation. `GrokNdjsonEvent` mirrors the VERIFIED +`HeadlessJsonEvent` union from upstream grok-cli's `src/headless/output.ts` +(captured in docs/grok-cli-contract.md) — there is deliberately no +thinking/reasoning event type here because upstream's JSONL emitter never +surfaces one (confirmed absence, not an omission). `GrokSession` / +`GrokCallbacks` / `AgentRuntime*` mirror the Droid plugin's `types.ts` shape +so this adapter satisfies the same plugin runtime contract +(`packages/engine/src/runtime-resolution.ts`'s `resolveRuntime`). Additive +only — does not collide with FN-7716's `GrokBinaryStatus` fields below. +*/ + +export interface GrokToolCallLike { + id?: string; + type?: string; + function?: { name?: string; arguments?: string }; + [key: string]: unknown; +} + +export interface GrokToolResultLike { + success?: boolean; + output?: string; + [key: string]: unknown; +} + +export interface GrokStepStartEvent { + type: "step_start"; + sessionID?: string; + stepNumber: number; + timestamp: number; +} + +export interface GrokTextEvent { + type: "text"; + sessionID?: string; + stepNumber: number; + text: string; + timestamp: number; +} + +export interface GrokToolUseEvent { + type: "tool_use"; + sessionID?: string; + stepNumber: number; + timestamp: number; + toolCall: GrokToolCallLike; + toolResult: GrokToolResultLike; + timing?: { startedAt?: number; finishedAt?: number; durationMs?: number }; +} + +export interface GrokStepFinishEvent { + type: "step_finish"; + sessionID?: string; + stepNumber: number; + timestamp: number; + finishReason: string; + usage: { inputTokens?: number; outputTokens?: number; totalTokens?: number; costUsdTicks?: number }; +} + +export interface GrokErrorEvent { + type: "error"; + sessionID?: string; + message: string; + timestamp: number; +} + +export type GrokNdjsonEvent = + | GrokStepStartEvent + | GrokTextEvent + | GrokToolUseEvent + | GrokStepFinishEvent + | GrokErrorEvent; + +export interface GrokCallbacks { + onText?: (text: string) => void; + /** + * FNXC:GrokCli 2026-07-09-00:00: kept for AgentRuntime interface parity + * with the Droid/Cursor plugins, but never invoked by this adapter — + * upstream grok-cli's `--format json` stream has no thinking/reasoning + * event to bridge (see docs/grok-cli-contract.md). + */ + onThinking?: (text: string) => void; +} + +export interface GrokSession { + model: string; + systemPrompt?: string; + messages: unknown[]; + sessionId?: string; + lastModelDescription: string; + callbacks: GrokCallbacks; +} + +export type AgentSession = GrokSession; + +export interface AgentRuntimeOptions { + cwd?: string; + systemPrompt?: string; + defaultModelId?: string; + onText?: (text: string) => void; + onThinking?: (text: string) => void; + signal?: AbortSignal; +} + +export interface AgentSessionResult { + session: AgentSession; + sessionFile?: string; +} + +export interface AgentRuntime { + id: string; + name: string; + createSession(options: AgentRuntimeOptions): Promise; + promptWithFallback(session: AgentSession, prompt: string, options?: unknown): Promise; + describeModel(session: AgentSession): string; + dispose?(session: AgentSession): Promise; +} + export interface GrokBinaryStatus { available: boolean; /**