FN-7722: route Grok execution through CLI with NDJSON streaming
Switches the Grok runtime plugin to execute prompts via the Grok CLI, parsing its NDJSON stream output instead of the previous invocation path. - Add cli-stream.ts to spawn and stream the Grok CLI process - Add stream-parser.ts to parse NDJSON CLI output into runtime events - Rework runtime-adapter.ts to route execution through CLI streaming - Extend types.ts with CLI stream/NDJSON event types - Add docs/grok-cli-contract.md documenting the CLI streaming contract - Add/update tests for stream-parser and runtime-adapter - Update plugin README with CLI streaming details - Add changeset for the Grok CLI streaming change Files changed: .changeset/fn-7722-grok-cli-streaming.md | 7 + docs/grok-cli-contract.md | 200 +++++++++++++++++++++ plugins/fusion-plugin-grok-runtime/README.md | 28 +++ .../src/__tests__/runtime-adapter.test.ts | 135 ++++++++++++-- .../src/__tests__/stream-parser.test.ts | 79 ++++++++ .../fusion-plugin-grok-runtime/src/cli-stream.ts | 52 ++++++ .../src/runtime-adapter.ts | 182 ++++++++++++++++--- .../src/stream-parser.ts | 54 ++++++ plugins/fusion-plugin-grok-runtime/src/types.ts | 120 +++++++++++++ 9 files changed, 816 insertions(+), 41 deletions(-) Fusion-Task-Id: FN-7722 Fusion-Task-Lineage: c5f33e9d-0032-432b-88b5-4ad8d786d67e Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-7722-grok-cli-streaming.md
Normal file
7
.changeset/fn-7722-grok-cli-streaming.md
Normal file
@@ -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.
|
||||
200
docs/grok-cli-contract.md
Normal file
200
docs/grok-cli-contract.md
Normal file
@@ -0,0 +1,200 @@
|
||||
# Grok CLI Contract (FN-7722)
|
||||
|
||||
Date: 2026-07-09
|
||||
|
||||
<!--
|
||||
FNXC:GrokCli 2026-07-09-00:00:
|
||||
FN-7715 shipped GrokRuntimeAdapter.promptWithFallback as an intentional no-op,
|
||||
justified by an FNXC comment asserting "no documented non-interactive
|
||||
prompt/stream subcommand" for the `grok` CLI. FN-7722 (this doc) corrects that
|
||||
assumption: upstream grok-cli DOES document and implement a non-interactive
|
||||
`grok --prompt <text> --format json` NDJSON event stream
|
||||
(src/headless/output.ts's `createHeadlessJsonlEmitter`), and this task lands a
|
||||
real streaming GrokRuntimeAdapter against that verified contract. See
|
||||
"Decision" below.
|
||||
-->
|
||||
|
||||
## 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 "<text>" --format json
|
||||
# short flags:
|
||||
grok -p "<text>" --format json
|
||||
```
|
||||
|
||||
- `-p, --prompt <prompt>` — run a single prompt headlessly, then exit.
|
||||
- `--format <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 <dir>` (cwd), `-m, --model <model>`, `-s, --session <id>`
|
||||
(resume a saved session, or `latest`), `-k, --api-key <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).
|
||||
@@ -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 "<text>" --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
|
||||
|
||||
@@ -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<typeof vi.fn> } {
|
||||
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");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
52
plugins/fusion-plugin-grok-runtime/src/cli-stream.ts
Normal file
52
plugins/fusion-plugin-grok-runtime/src/cli-stream.ts
Normal file
@@ -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 <text> --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<null, Readable, Readable>;
|
||||
|
||||
export interface SpawnGrokStreamOptions {
|
||||
cwd?: string;
|
||||
signal?: AbortSignal;
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawn `grok --prompt <prompt> --format json [--model <model>] [--directory <cwd>]`
|
||||
* 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
|
||||
}
|
||||
}
|
||||
@@ -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 <text> --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<AgentSessionResult> {
|
||||
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<void> {
|
||||
return;
|
||||
async promptWithFallback(session: AgentSession, prompt: string, options?: AgentRuntimeOptions): Promise<void> {
|
||||
const grokSession = session as GrokSession;
|
||||
const cwd = options?.cwd;
|
||||
const signal = options?.signal;
|
||||
|
||||
return new Promise<void>((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"}`;
|
||||
}
|
||||
}
|
||||
|
||||
54
plugins/fusion-plugin-grok-runtime/src/stream-parser.ts
Normal file
54
plugins/fusion-plugin-grok-runtime/src/stream-parser.ts
Normal file
@@ -0,0 +1,54 @@
|
||||
import type { GrokNdjsonEvent } from "./types.js";
|
||||
|
||||
/*
|
||||
FNXC:GrokCli 2026-07-09-00:00:
|
||||
FN-7722: `grok --prompt <text> --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;
|
||||
}
|
||||
@@ -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<AgentSessionResult>;
|
||||
promptWithFallback(session: AgentSession, prompt: string, options?: unknown): Promise<void>;
|
||||
describeModel(session: AgentSession): string;
|
||||
dispose?(session: AgentSession): Promise<void>;
|
||||
}
|
||||
|
||||
export interface GrokBinaryStatus {
|
||||
available: boolean;
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user