diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index a330ff866..c8b7c850a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -252,6 +252,26 @@ jobs: echo "NEXT_PUBLIC_COVEN_OPENCODE_SCHEMA_REGISTRY_CHECKPOINT=$NEXT_PUBLIC_COVEN_OPENCODE_SCHEMA_REGISTRY_CHECKPOINT" } >> "$GITHUB_ENV" + - name: Require signed Grok compatibility registry + shell: bash + env: + NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_URL: ${{ secrets.GROK_SCHEMA_REGISTRY_URL }} + NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEY: ${{ secrets.GROK_SCHEMA_REGISTRY_PUBLIC_KEY }} + NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEYS: ${{ secrets.GROK_SCHEMA_REGISTRY_PUBLIC_KEYS }} + NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT: ${{ secrets.GROK_SCHEMA_REGISTRY_CHECKPOINT }} + run: | + node scripts/check-grok-registry-release.mjs + { + echo "NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_URL=$NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_URL" + echo "NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEY<> "$GITHUB_ENV" + # Decode the App Store Connect API key onto the runner for the custom # macOS release script. The path is exported into the env for later # steps. Runs only on the macOS leg. diff --git a/docs/grok-compatibility-registry.md b/docs/grok-compatibility-registry.md new file mode 100644 index 000000000..0965ec738 --- /dev/null +++ b/docs/grok-compatibility-registry.md @@ -0,0 +1,15 @@ +# Grok Build compatibility registry + +Grok Build's built-in profile is limited to the xAI-documented `text`, `thought`, `end`, and `error` `streaming-json` frames. It contains no tool-event aliases. A tool schema is enabled only when the exact locally resolved launcher advertises the value-bearing `--output-format streaming-json` option and a selected Ed25519-signed bundle explicitly names every envelope field and lifecycle event **and pins those aliases to exact locally probed Grok Build versions**. Help/version probes never receive credential-bearing environment variables and run no model request. + +Release configuration is public verification material, never a signing key: + +- `GROK_SCHEMA_REGISTRY_URL` — canonical credential-free HTTPS bundle URL. +- `GROK_SCHEMA_REGISTRY_PUBLIC_KEY`, or `GROK_SCHEMA_REGISTRY_PUBLIC_KEYS` — PEM Ed25519 trust anchor(s), with one to four key IDs for rotation. A bundle signed against a multi-key keyring must carry its exact `keyId`. +- `GROK_SCHEMA_REGISTRY_CHECKPOINT` — JSON `{ "sequence": number, "payloadHash": "" }` that anchors first use and rollback resistance. + +The release maps these to `NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_*`; production reads only those packaged anchors. Development may use `COVEN_GROK_SCHEMA_REGISTRY_*`, which production deliberately ignores. Registry downloads reject redirects, credentials, oversized or stalled bodies, malformed bundles, invalid signatures, checkpoint regressions, and cache-anchor rollbacks. The per-user cache is bounded, atomically replaced under a short writer lock, and keeps a bounded immutable high-water journal so a resumed stale writer cannot lower the accepted sequence. It is always reverified; an unknown or malformed selected event quarantines that schema in-process and future turns fall back to plain text. If a newer remote contract was previously accepted, its cache expires, or its anchor is missing/corrupt, Cave does not revive the older compiled parser; it waits in plain chat for a verified refresh. The compiled baseline also expires rather than parsing future output indefinitely. Do not publish a Grok tool schema until its precise stdout envelope is source-verified and captured from an approved non-production fixture. Never store a private key in this repository, app configuration, or release secrets. + +## Evidence + +Verified on 2026-07-26 from xAI's [Grok Build overview](https://docs.x.ai/build/overview), which documents headless `grok -p ... --output-format streaming-json`, and the upstream [headless-mode source documentation](https://github.com/xai-org/grok-build/blob/47348d13ec4508dcfe440e34c6d511bb02998fb2/crates/codegen/xai-grok-pager/docs/user-guide/14-headless-mode.md) at Grok Build revision [`47348d13ec4508dcfe440e34c6d511bb02998fb2`](https://github.com/xai-org/grok-build/tree/47348d13ec4508dcfe440e34c6d511bb02998fb2). Those sources establish the baseline text/thought/end/error transport only; they do **not** document tool lifecycle envelope names. No live Grok capture is stored or required. Future signed schemas need separately recorded source evidence for every added event and field before release owners publish them. diff --git a/scripts/check-grok-registry-release.mjs b/scripts/check-grok-registry-release.mjs new file mode 100644 index 000000000..f8bee3574 --- /dev/null +++ b/scripts/check-grok-registry-release.mjs @@ -0,0 +1,29 @@ +// Public verification material is embedded in a desktop release; this guard +// makes a missing trust anchor a release failure rather than an unsafe default. +import { createPublicKey } from "node:crypto"; + +const url = process.env.NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_URL; +const publicKey = process.env.NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEY; +const publicKeys = process.env.NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEYS; +const checkpoint = process.env.NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT; +const fail = (message) => { console.error(`::error::${message}`); process.exitCode = 1; }; +let keyring; +try { keyring = publicKeys ? JSON.parse(publicKeys) : publicKey ? { legacy: publicKey } : null; } catch { keyring = null; } +if (!url || !keyring || typeof keyring !== "object" || Array.isArray(keyring) || !checkpoint) { + fail("Grok compatibility registry URL, Ed25519 public key/keyring, and immutable sequence checkpoint must be configured for every desktop release."); +} else { + try { + const parsedUrl = new URL(url); + if (parsedUrl.protocol !== "https:" || parsedUrl.username || parsedUrl.password) throw new Error("registry URL must use HTTPS without credentials"); + const entries = Object.entries(keyring); + if (!entries.length || entries.length > 4) throw new Error("registry keyring must contain one to four keys"); + for (const [id, pem] of entries) { + if (!/^[A-Za-z][A-Za-z0-9_-]{0,63}$/.test(id) || typeof pem !== "string") throw new Error("registry keyring has an invalid key id"); + if (createPublicKey(pem).asymmetricKeyType !== "ed25519") throw new Error("registry key must be Ed25519"); + } + const parsedCheckpoint = JSON.parse(checkpoint); + if (!parsedCheckpoint || typeof parsedCheckpoint !== "object" || Array.isArray(parsedCheckpoint) + || Object.keys(parsedCheckpoint).length !== 2 || !Number.isSafeInteger(parsedCheckpoint.sequence) || parsedCheckpoint.sequence < 1 + || typeof parsedCheckpoint.payloadHash !== "string" || !/^[a-f0-9]{64}$/.test(parsedCheckpoint.payloadHash)) throw new Error("registry checkpoint must contain a sequence and SHA-256 payload hash"); + } catch (error) { fail(`Invalid Grok compatibility registry configuration: ${error instanceof Error ? error.message : "unknown error"}`); } +} diff --git a/scripts/check-grok-registry-release.test.mjs b/scripts/check-grok-registry-release.test.mjs new file mode 100644 index 000000000..5465ec19b --- /dev/null +++ b/scripts/check-grok-registry-release.test.mjs @@ -0,0 +1,14 @@ +import assert from "node:assert/strict"; +import { generateKeyPairSync } from "node:crypto"; +import { spawnSync } from "node:child_process"; +import { readFile } from "node:fs/promises"; + +const workflow = await readFile(new URL("../.github/workflows/release.yml", import.meta.url), "utf8"); +const guard = await readFile(new URL("./check-grok-registry-release.mjs", import.meta.url), "utf8"); +assert.match(workflow, /Require signed Grok compatibility registry[\s\S]*?NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_URL[\s\S]*?check-grok-registry-release\.mjs/); +assert.match(guard, /registry URL must use HTTPS without credentials/); +const { publicKey } = generateKeyPairSync("ed25519"); +const result = spawnSync(process.execPath, ["scripts/check-grok-registry-release.mjs"], { cwd: new URL("..", import.meta.url), encoding: "utf8", env: { ...process.env, NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_URL: "https://publisher:secret@registry.example/grok.json", NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEY: publicKey.export({ type: "spki", format: "pem" }).toString(), NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT: JSON.stringify({ sequence: 1, payloadHash: "a".repeat(64) }) } }); +assert.notEqual(result.status, 0); +assert.doesNotMatch(result.stderr, /secret/); +console.log("check-grok-registry-release.test.mjs: ok"); diff --git a/scripts/run-tests.mjs b/scripts/run-tests.mjs index cac2f44e7..a51097de5 100644 --- a/scripts/run-tests.mjs +++ b/scripts/run-tests.mjs @@ -103,6 +103,7 @@ export const SUITES = { "scripts/eslint/design-system-plugin.test.mjs", "scripts/bundle-budget.test.mjs", "scripts/check-opencode-registry-release.test.mjs", + "scripts/check-grok-registry-release.test.mjs", "src/components/open-coven-tools-update.test.ts", "src/lib/opencoven-tools-state.test.ts", "src/lib/opencoven-install-job-observer.test.ts", @@ -1013,6 +1014,7 @@ export const SUITES = { "src/lib/grok-bin.test.ts", "src/lib/grok-build.test.ts", "src/lib/runtime-availability.test.ts", + "src/lib/grok-compatibility.test.ts", "src/app/api/harnesses/route.test.ts", "src/lib/copilot-stream.test.ts", "src/lib/copilot-bin.test.ts", @@ -1034,6 +1036,7 @@ export const SUITES = { "src/app/api/chat/send/chat-send-models.test.ts", "src/app/api/chat/send/chat-send-capabilities.test.ts", "src/app/api/chat/send/route-opencode.integration.test.ts", + "src/app/api/chat/send/route-grok-compatibility.integration.test.ts", "src/app/api/chat/send/route-runtime-availability.integration.test.ts", "src/app/api/chat/send/offline-queue.test.ts", "src/app/api/chat/send/first-turn-stub.test.ts", @@ -1306,6 +1309,7 @@ const ALIAS_LOADER = new Set([ "src/app/api/chat/send/chat-send-models.test.ts", "src/app/api/chat/send/chat-send-capabilities.test.ts", "src/app/api/chat/send/route-opencode.integration.test.ts", + "src/app/api/chat/send/route-grok-compatibility.integration.test.ts", "src/app/api/chat/send/route-runtime-availability.integration.test.ts", "src/lib/familiar-workspace-sessions.test.ts", "src/lib/use-projects-scope-transition.test.ts", diff --git a/src/app/api/chat/send/harness-routing-copilot-jsonl.test.ts b/src/app/api/chat/send/harness-routing-copilot-jsonl.test.ts index f3826b915..d9b23fbb8 100644 --- a/src/app/api/chat/send/harness-routing-copilot-jsonl.test.ts +++ b/src/app/api/chat/send/harness-routing-copilot-jsonl.test.ts @@ -212,8 +212,8 @@ assert.match( ); assert.match( chatRoute, - /if \(grokDirect\) \{\s*handleGrokLine\(line, isJson\);\s*return;/, - "Grok stdout routes through the native JSONL parser, never generic stream-json parsing", + /if \(grokDirect\) \{[\s\S]*?handleGrokLine\(line, isJson \|\| [\s\S]*?line\.trimStart\(\)\)\);\s*return;/, + "Grok stdout routes through the native parser, rejecting whitespace-prefixed object and array envelopes before plain fallback persistence", ); assert.match( chatRoute, @@ -255,6 +255,26 @@ assert.match( /Grok Build chats currently run on this Cave host/, "SSH Grok must fail explicitly instead of falling back to coven run", ); +assert.match( + chatRoute, + /const grokCapabilities = grokDirect[\s\S]*?probeReadyLocalRuntimeCapability\([\s\S]*?runner: "grok",[\s\S]*?probe: \(\) => probeGrokRunCapabilities\([\s\S]*?command: localRuntimePlan!\.command,[\s\S]*?fixedArgs: localRuntimePlan!\.fixedArgs[\s\S]*?localRuntimePlan!\.env[\s\S]*?const grokCompatibility = grokCapabilities[\s\S]*?resolveGrokCompatibility\(grokCapabilities\)/, + "Grok must probe the exact preflight-approved launcher and environment before selecting a structured schema", +); +assert.match( + chatRoute, + /outputFormat: grokCompatibility\?\.mode === "structured" \? "streaming-json" : null/, + "an unverified Grok client must use plain output rather than an assumed JSON protocol", +); +assert.match( + chatRoute, + /case "tool_start":[\s\S]*?envelopeToolUse[\s\S]*?consumePendingEnvelopeProgress[\s\S]*?consumePendingEnvelopeResult/, + "selected Grok schemas must reconcile reordered progress and terminal tool results through the shared tracker", +); +assert.match( + chatRoute, + /case "unknown":[\s\S]*?quarantineGrokSchema[\s\S]*?redactedGrokEventFingerprint/, + "unknown selected-schema events must quarantine future structured launches with a redacted diagnostic", +); assert.match( chatRoute, @@ -324,8 +344,8 @@ assert.match( ); assert.match( chatRoute, - /\(openCodeDirect \|\| copilotStream\) && code !== 0[\s\S]*?is_error: true/, - "a nonzero direct Copilot process exit persists the turn as an error even without a final result frame", + /\(openCodeDirect \|\| copilotStream \|\| grokDirect\) && code !== 0[\s\S]*?is_error: true/, + "a nonzero direct Copilot or Grok process exit persists the turn as an error even without a final result frame", ); assert.match( chatRoute, diff --git a/src/app/api/chat/send/harness-routing-opencode.test.ts b/src/app/api/chat/send/harness-routing-opencode.test.ts index ca56e3e54..5ee173b0c 100644 --- a/src/app/api/chat/send/harness-routing-opencode.test.ts +++ b/src/app/api/chat/send/harness-routing-opencode.test.ts @@ -169,8 +169,8 @@ assert.match( ); assert.match( route, - /child\.on\("close", \(code\) => \{[\s\S]*?if \(\(openCodeDirect \|\| copilotStream\) && code !== 0\)[\s\S]*?is_error: true/, - "a non-zero direct OpenCode or Copilot exit cannot be treated as a successful run when no JSON error arrives", + /child\.on\("close", \(code\) => \{[\s\S]*?if \(\(openCodeDirect \|\| copilotStream \|\| grokDirect\) && code !== 0\)[\s\S]*?is_error: true/, + "a non-zero direct OpenCode, Copilot, or Grok exit cannot be treated as a successful run when no JSON error arrives", ); assert.match( route, @@ -179,8 +179,8 @@ assert.match( ); assert.match( route, - /const tailBlock = !openCodeDirect && tailSource\.length/, - "OpenCode stderr never becomes assistant-visible or persisted empty-response diagnostics", + /const tailBlock = !openCodeDirect && !grokDirect && tailSource\.length/, + "OpenCode and Grok stderr never become assistant-visible or persisted empty-response diagnostics", ); assert.match( route, @@ -219,8 +219,8 @@ assert.match( ); assert.match( route, - /persistedOpenCodeDiagnostics[\s\S]*?id === "opencode-compatibility"[\s\S]*?progress: persistedOpenCodeDiagnostics/, - "safe OpenCode compatibility diagnostics persist with the completed assistant turn", + /persistedCompatibilityDiagnostics[\s\S]*?id === "opencode-compatibility" \|\| id === "grok-compatibility"[\s\S]*?progress: persistedCompatibilityDiagnostics/, + "safe OpenCode and Grok compatibility diagnostics persist with the completed assistant turn", ); assert.match( route, diff --git a/src/app/api/chat/send/route-grok-compatibility.integration.test.ts b/src/app/api/chat/send/route-grok-compatibility.integration.test.ts new file mode 100644 index 000000000..7f948fdee --- /dev/null +++ b/src/app/api/chat/send/route-grok-compatibility.integration.test.ts @@ -0,0 +1,225 @@ +// @ts-nocheck +import assert from "node:assert/strict"; +import { generateKeyPairSync, sign } from "node:crypto"; +import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises"; +import { homedir } from "node:os"; +import path from "node:path"; + +// Exercise the real Grok route with a deterministic local launcher. This covers +// the boundary pure parser tests cannot: secret-free capability probing, +// verified JSON launch, plain fallback, and persisted native session handling. +const home = await mkdtemp(path.join(homedir(), "cave-grok-route-")); +const bin = path.join(home, "bin"); +const familiarWorkspace = path.join(home, "familiars", "opal"); +await mkdir(bin, { recursive: true }); +await mkdir(familiarWorkspace, { recursive: true }); + +const previousHome = process.env.COVEN_HOME; +const previousCaveHome = process.env.COVEN_CAVE_HOME; +const previousPath = process.env.PATH; +const previousGrokBin = process.env.GROK_BIN; +const previousGrokTestMode = process.env.GROK_TEST_MODE; +const previousXaiApiKey = process.env.XAI_API_KEY; +const previousGrokRegistryKeys = process.env.COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEYS; +const previousGrokRegistryCheckpoint = process.env.COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT; +process.env.COVEN_HOME = home; +process.env.COVEN_CAVE_HOME = path.join(home, "cave"); +process.env.PATH = `${bin}${path.delimiter}${previousPath ?? ""}`; +process.env.XAI_API_KEY = "probe-must-not-receive-this"; + +// These event names are a deterministic signed-registry fixture only; they +// make no claim about Grok Build's undocumented tool protocol. +const { privateKey, publicKey } = generateKeyPairSync("ed25519"); +process.env.COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEYS = JSON.stringify({ fixture: publicKey.export({ format: "pem", type: "spki" }).toString() }); +const { grokSchemaBundlePayloadHash, grokSchemaBundleSigningPayload } = await import("@/lib/grok-compatibility"); +const fixtureBundle = { + format: 1 as const, + runtime: "grok-build" as const, + sequence: 2, + issuedAt: "2026-07-26T00:00:00.000Z", + expiresAt: "2030-01-01T00:00:00.000Z", + keyId: "fixture", + schemas: [{ + id: "grok-build-fixture-tool-events", + priority: 1, + requires: { streamingJson: true as const, options: ["--output-format"], versions: ["1.0.0"] }, + eventTypes: { + ignored: ["thought"], text: ["text"], end: ["end"], error: ["error"], + toolStart: ["fixture_tool_start"], toolProgress: ["fixture_tool_progress"], + toolEnd: ["fixture_tool_end"], toolComplete: ["fixture_tool_complete"], + }, + fields: { + type: ["type"], text: ["data"], sessionId: ["sessionId"], message: ["message"], + usage: [], totalCostUsd: [], id: ["id"], name: ["name"], input: ["input"], + output: ["output"], state: ["state"], error: ["error"], terminalStates: [], errorStates: ["error"], + }, + launch: { outputOption: "--output-format" as const, outputValue: "streaming-json" as const }, + }], +}; +fixtureBundle.signature = { + algorithm: "ed25519" as const, + value: sign(null, Buffer.from(grokSchemaBundleSigningPayload(fixtureBundle)), privateKey).toString("base64"), +}; +const fixtureHash = grokSchemaBundlePayloadHash(fixtureBundle); +process.env.COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT = JSON.stringify({ sequence: fixtureBundle.sequence, payloadHash: fixtureHash }); +const fixtureCachePath = path.join(process.env.COVEN_CAVE_HOME, "grok-schema-bundle-v1.json"); +await mkdir(`${fixtureCachePath}.anchor.journal`, { recursive: true }); +await writeFile(fixtureCachePath, JSON.stringify({ checkedAt: Date.now(), bundle: fixtureBundle })); +await writeFile(`${fixtureCachePath}.anchor.journal/${fixtureBundle.sequence}-${fixtureHash}.json`, JSON.stringify({ sequence: fixtureBundle.sequence, payloadHash: fixtureHash })); + +const executable = process.platform === "win32" ? "grok.cmd" : "grok"; +const windowsShimTarget = path.join(bin, "grok-launcher.js"); +const windowsShimProgram = [ + "const [command] = process.argv.slice(2);", + "if (command === '--help') { if (process.env.XAI_API_KEY) process.exit(12); console.log(['plain', 'plain-structured', 'plain-structured-array', 'plain-bracketed-text'].includes(process.env.GROK_TEST_MODE ?? '') ? ' --output-format Output format: text' : ' --output-format Output format: text, streaming-json'); process.exit(0); }", + "if (command === '--version') { if (process.env.XAI_API_KEY) process.exit(12); console.log('1.0.0'); process.exit(0); }", + "if (process.env.GROK_TEST_MODE === 'plain') console.log('plain fallback reply');", + "else if (process.env.GROK_TEST_MODE === 'plain-structured') console.log(' {\\\"opaque\\\":\\\"private structured payload\\\"}');", + "else if (process.env.GROK_TEST_MODE === 'plain-structured-array') console.log('[{\\\"opaque\\\":\\\"private array payload\\\"}]');", + "else if (process.env.GROK_TEST_MODE === 'plain-bracketed-text') console.log('[a safe plain-text reply]');", + "else if (process.env.GROK_TEST_MODE === 'tool-activity') console.log(['{\"type\":\"fixture_tool_end\",\"id\":\"reordered\",\"output\":\"first terminal result\"}', '{\"type\":\"fixture_tool_progress\",\"id\":\"reordered\",\"output\":\"early progress\"}', '{\"type\":\"fixture_tool_complete\",\"id\":\"reordered\",\"name\":\"fixture_call\",\"input\":{\"safe\":true},\"output\":\"duplicate terminal result\"}', '{\"type\":\"fixture_tool_complete\",\"id\":\"terminal-error\",\"name\":\"fixture_error\",\"input\":{},\"output\":\"terminal error\",\"error\":true}', '{\"type\":\"end\",\"sessionId\":\"native_grok_session\"}'].join('\\n'));", + "else if (process.env.GROK_TEST_MODE === 'malformed') console.log('unframed private tool payload');", + "else if (process.env.GROK_TEST_MODE === 'exit-error') { console.error('private Grok stderr payload'); process.exit(3); }", + "else console.log('{\\\"type\\\":\\\"text\\\",\\\"data\\\":\\\"verified route reply\\\"}\\n{\\\"type\\\":\\\"end\\\",\\\"sessionId\\\":\\\"native_grok_session\\\"}');", +].join("\n"); +const launcher = process.platform === "win32" + ? [ + "@echo off", + "\"%~dp0\\grok-launcher.js\" %*", + ].join("\r\n") + : [ + "#!/bin/sh", + "if [ \"$1\" = \"--help\" ]; then", + " [ -z \"$XAI_API_KEY\" ] || exit 12", + " if [ \"$GROK_TEST_MODE\" = \"plain\" ] || [ \"$GROK_TEST_MODE\" = \"plain-structured\" ] || [ \"$GROK_TEST_MODE\" = \"plain-structured-array\" ] || [ \"$GROK_TEST_MODE\" = \"plain-bracketed-text\" ]; then printf '%s\\n' ' --output-format Output format: text'; else printf '%s\\n' ' --output-format Output format: text, streaming-json'; fi", + " exit 0", + "fi", + "if [ \"$1\" = \"--version\" ]; then [ -z \"$XAI_API_KEY\" ] || exit 12; printf '%s\\n' '1.0.0'; exit 0; fi", + "if [ \"$GROK_TEST_MODE\" = \"plain\" ]; then printf '%s\\n' 'plain fallback reply'; exit 0; fi", + "if [ \"$GROK_TEST_MODE\" = \"plain-structured\" ]; then printf '%s\\n' ' {\"opaque\":\"private structured payload\"}'; exit 0; fi", + "if [ \"$GROK_TEST_MODE\" = \"plain-structured-array\" ]; then printf '%s\\n' '[{\"opaque\":\"private array payload\"}]'; exit 0; fi", + "if [ \"$GROK_TEST_MODE\" = \"plain-bracketed-text\" ]; then printf '%s\\n' '[a safe plain-text reply]'; exit 0; fi", + "if [ \"$GROK_TEST_MODE\" = \"tool-activity\" ]; then printf '%s\\n' '{\"type\":\"fixture_tool_end\",\"id\":\"reordered\",\"output\":\"first terminal result\"}' '{\"type\":\"fixture_tool_progress\",\"id\":\"reordered\",\"output\":\"early progress\"}' '{\"type\":\"fixture_tool_complete\",\"id\":\"reordered\",\"name\":\"fixture_call\",\"input\":{\"safe\":true},\"output\":\"duplicate terminal result\"}' '{\"type\":\"fixture_tool_complete\",\"id\":\"terminal-error\",\"name\":\"fixture_error\",\"input\":{},\"output\":\"terminal error\",\"error\":true}' '{\"type\":\"end\",\"sessionId\":\"native_grok_session\"}'; exit 0; fi", + "if [ \"$GROK_TEST_MODE\" = \"malformed\" ]; then printf '%s\\n' 'unframed private tool payload'; exit 0; fi", + "if [ \"$GROK_TEST_MODE\" = \"exit-error\" ]; then printf '%s\\n' 'private Grok stderr payload' >&2; exit 3; fi", + "printf '%s\\n' '{\"type\":\"text\",\"data\":\"verified route reply\"}' '{\"type\":\"end\",\"sessionId\":\"native_grok_session\"}'", + ].join("\n"); +const launcherPath = path.join(bin, executable); +await writeFile(launcherPath, launcher, { mode: 0o755 }); +if (process.platform === "win32") await writeFile(windowsShimTarget, windowsShimProgram); +process.env.GROK_BIN = launcherPath; + +async function readSse(response: Response) { + assert.equal(response.status, 200, await response.clone().text()); + const body = await response.text(); + return { + body, + events: body.split("\n").filter((line) => line.startsWith("data: ")).map((line) => JSON.parse(line.slice(6))), + }; +} + +try { + const { saveConfig } = await import("@/lib/cave-config"); + const { loadConversation } = await import("@/lib/cave-conversations"); + const { createProject } = await import("@/lib/cave-projects"); + const { grantProjectToFamiliar } = await import("@/lib/project-permissions"); + const { POST } = await import("./route.ts"); + await saveConfig({ familiars: { opal: { harness: "grok" } } }); + const project = await createProject({ name: "Grok route fixture", root: familiarWorkspace }); + await grantProjectToFamiliar({ familiarId: "opal", projectId: project.id, source: "human", access: "write" }); + + const structured = await readSse(await POST(new Request("http://localhost/api/chat/send", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ familiarId: "opal", prompt: "fixture", projectRoot: familiarWorkspace }), + }))); + assert.match(structured.body, /"kind":"assistant_chunk","text":"verified route reply"/, "a source-verified selected JSON schema renders assistant text"); + const structuredDone = structured.events.findLast((event) => event.kind === "done"); + assert.equal(typeof structuredDone?.sessionId, "string"); + const conversation = await loadConversation(structuredDone.sessionId); + assert.equal(conversation?.harnessSessionId, "native_grok_session", "the route persists Grok's native resume id separately from Cave's id"); + + process.env.GROK_TEST_MODE = "tool-activity"; + const toolActivity = await readSse(await POST(new Request("http://localhost/api/chat/send", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ familiarId: "opal", prompt: "tool fixture", projectRoot: familiarWorkspace }), + }))); + const toolEvents = toolActivity.events.filter((event) => event.kind === "tool_use"); + assert.deepEqual( + toolEvents.map((event) => [event.name, event.status, event.output]), + [ + ["fixture_call", "running", undefined], + ["fixture_call", "running", "early progress"], + ["fixture_call", "ok", "first terminal result"], + ["fixture_error", "running", undefined], + ["fixture_error", "error", "terminal error"], + ], + "a selected signed fixture preserves stable tool activity across reordered, duplicate, and error terminal frames", + ); + assert.doesNotMatch(toolActivity.body, /duplicate terminal result/, "the first terminal result wins when a combined completion is retransmitted"); + const toolConversation = await loadConversation(toolActivity.events.findLast((event) => event.kind === "done")?.sessionId); + assert.deepEqual(toolConversation?.turns.at(-1)?.tools?.map((tool) => [tool.name, tool.status, tool.output]), [ + ["fixture_call", "ok", "first terminal result"], + ["fixture_error", "error", "terminal error"], + ], "selected-schema tool activity persists through the real chat route"); + + process.env.GROK_TEST_MODE = "plain"; + const plain = await readSse(await POST(new Request("http://localhost/api/chat/send", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ familiarId: "opal", prompt: "fallback", projectRoot: familiarWorkspace }), + }))); + assert.match(plain.body, /"kind":"assistant_chunk","text":"plain fallback reply\\n"/, "an unverified output format remains safe plain assistant text"); + assert.match(plain.body, /without tool activity/, "plain fallback provides an accessible compatibility diagnostic"); + const plainConversation = await loadConversation(plain.events.findLast((event) => event.kind === "done")?.sessionId); + assert.ok(plainConversation?.turns.at(-1)?.progress?.some((event) => event.id === "grok-compatibility"), "plain fallback persists its value-free compatibility diagnostic for reload"); + + process.env.GROK_TEST_MODE = "plain-bracketed-text"; + const bracketedText = await readSse(await POST(new Request("http://localhost/api/chat/send", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ familiarId: "opal", prompt: "bracketed fallback", projectRoot: familiarWorkspace }), + }))); + assert.match(bracketedText.body, /\[a safe plain-text reply\]/, "plain fallback preserves prose that merely starts with a bracket"); + + process.env.GROK_TEST_MODE = "plain-structured"; + const unverifiedStructured = await readSse(await POST(new Request("http://localhost/api/chat/send", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ familiarId: "opal", prompt: "unsafe fallback", projectRoot: familiarWorkspace }), + }))); + assert.match(unverifiedStructured.body, /unverified-structured-output/, "plain fallback reports an accessible fixed diagnostic for raw structured output"); + assert.doesNotMatch(unverifiedStructured.body, /private structured payload/, "plain fallback never persists unverified structured payload values as assistant text or diagnostics"); + + process.env.GROK_TEST_MODE = "plain-structured-array"; + const unverifiedArray = await readSse(await POST(new Request("http://localhost/api/chat/send", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ familiarId: "opal", prompt: "unsafe array fallback", projectRoot: familiarWorkspace }), + }))); + assert.match(unverifiedArray.body, /unverified-structured-output/, "plain fallback also rejects array-shaped structured output"); + assert.doesNotMatch(unverifiedArray.body, /private array payload/, "plain fallback never persists unverified array payload values"); + + process.env.GROK_TEST_MODE = "malformed"; + const malformed = await readSse(await POST(new Request("http://localhost/api/chat/send", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ familiarId: "opal", prompt: "malformed", projectRoot: familiarWorkspace }), + }))); + assert.match(malformed.body, /unframed-jsonl-event/, "unframed selected-schema output emits an accessible compatibility diagnostic"); + assert.doesNotMatch(malformed.body, /private tool payload/, "unframed structured payloads never enter assistant text or diagnostics"); + + process.env.GROK_TEST_MODE = "exit-error"; + const exited = await readSse(await POST(new Request("http://localhost/api/chat/send", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ familiarId: "opal", prompt: "exit error", projectRoot: familiarWorkspace }), + }))); + assert.equal(exited.events.findLast((event) => event.kind === "done")?.isError, true, "a non-zero Grok process cannot be persisted as a successful turn"); + assert.doesNotMatch(exited.body, /private Grok stderr payload/, "Grok stderr values never enter assistant-visible or persisted diagnostics"); +} finally { + if (previousHome === undefined) delete process.env.COVEN_HOME; else process.env.COVEN_HOME = previousHome; + if (previousCaveHome === undefined) delete process.env.COVEN_CAVE_HOME; else process.env.COVEN_CAVE_HOME = previousCaveHome; + if (previousPath === undefined) delete process.env.PATH; else process.env.PATH = previousPath; + if (previousGrokBin === undefined) delete process.env.GROK_BIN; else process.env.GROK_BIN = previousGrokBin; + if (previousGrokTestMode === undefined) delete process.env.GROK_TEST_MODE; else process.env.GROK_TEST_MODE = previousGrokTestMode; + if (previousXaiApiKey === undefined) delete process.env.XAI_API_KEY; else process.env.XAI_API_KEY = previousXaiApiKey; + if (previousGrokRegistryKeys === undefined) delete process.env.COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEYS; else process.env.COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEYS = previousGrokRegistryKeys; + if (previousGrokRegistryCheckpoint === undefined) delete process.env.COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT; else process.env.COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT = previousGrokRegistryCheckpoint; + await rm(home, { recursive: true, force: true }); +} + +console.log("route-grok-compatibility.integration.test.ts: ok"); diff --git a/src/app/api/chat/send/route.ts b/src/app/api/chat/send/route.ts index a358dcbc8..68864e9ed 100644 --- a/src/app/api/chat/send/route.ts +++ b/src/app/api/chat/send/route.ts @@ -62,6 +62,13 @@ import { parseGrokStreamEvent, } from "@/lib/grok-build"; import { grokLaunchCommand } from "@/lib/grok-bin"; +import { + parseGrokCompatibilityEvent, + probeGrokRunCapabilities, + quarantineGrokSchema, + redactedGrokEventFingerprint, + resolveGrokCompatibility, +} from "@/lib/grok-compatibility"; import { openCodeCommand, openCodeLaunch, openCodeSpawnEnv, writeOpenCodeLaunchInput } from "@/lib/opencode-bin"; import { evaluateRuntimeAvailability, @@ -1160,6 +1167,22 @@ export async function POST(req: Request) { const openCodeCompatibility = openCodeCapabilities ? await resolveOpenCodeCompatibility(openCodeCapabilities) : null; + // Probe the same ready local launcher/environment that the direct run will + // use. A missing or changed CLI remains a truthful preflight failure; it + // never turns undocumented output into tool activity. + const grokCapabilities = grokDirect + ? await probeReadyLocalRuntimeCapability({ + plan: localRuntimePlan, + runner: "grok", + probe: () => probeGrokRunCapabilities( + { command: localRuntimePlan!.command, fixedArgs: localRuntimePlan!.fixedArgs }, + localRuntimePlan!.env, + ), + }) + : null; + const grokCompatibility = grokCapabilities + ? await resolveGrokCompatibility(grokCapabilities) + : null; const hermesModelCapability = hermesDirect && hermesApi === null ? await probeReadyLocalRuntimeCapability({ @@ -1649,6 +1672,7 @@ export async function POST(req: Request) { binding.display_name, binding.role, ), + outputFormat: grokCompatibility?.mode === "structured" ? "streaming-json" : null, }); } if (openCodeDirect) { @@ -1810,7 +1834,7 @@ export async function POST(req: Request) { // Compatibility notices are deliberately value-free and must survive // transcript reloads; the live SSE buffer alone expires after two // minutes. Keep only this narrowly-scoped subset of progress rows. - const persistedOpenCodeDiagnostics: NonNullable = []; + const persistedCompatibilityDiagnostics: NonNullable = []; const pushProgress = ( id: string, label: string, @@ -1818,8 +1842,8 @@ export async function POST(req: Request) { detail?: string, durationMs?: number, ) => { - if (id === "opencode-compatibility") { - persistedOpenCodeDiagnostics.push({ + if (id === "opencode-compatibility" || id === "grok-compatibility") { + persistedCompatibilityDiagnostics.push({ id, label, status, @@ -1988,6 +2012,24 @@ export async function POST(req: Request) { let openCodeProtocolQuarantineNoticeSent = false; let openCodeStructuredProtocolQuarantined = false; let openCodeModelRejected = false; + let grokCompatibilityHealthNoticeSent = false; + let grokStructuredProtocolQuarantined = false; + let grokProtocolQuarantineNoticeSent = false; + const quarantineGrokProtocol = ( + detail: "malformed-jsonl-event" | "unframed-jsonl-event" | `unknown-event:${string}`, + ) => { + grokStructuredProtocolQuarantined = true; + quarantineGrokSchema(grokCompatibility?.schema); + recordStdoutErrorTail("Grok Build emitted a malformed structured event", true); + if (grokProtocolQuarantineNoticeSent) return; + grokProtocolQuarantineNoticeSent = true; + pushProgress( + "grok-compatibility", + "Grok Build sent an unrecognized event; future turns will use safe plain chat until a compatible schema is available", + "error", + detail, + ); + }; const quarantineOpenCodeProtocol = ( label: string, detail: "malformed-json-event" | "oversized-jsonl-event", @@ -2297,12 +2339,64 @@ export async function POST(req: Request) { }; const handleGrokLine = (line: string, isJson: boolean) => { + if (!grokCompatibilityHealthNoticeSent && grokCompatibility?.diagnostic) { + grokCompatibilityHealthNoticeSent = true; + const label = grokCompatibility.diagnostic === "streaming-json-unavailable" + ? "This Grok Build client does not advertise streaming JSON; continuing without tool activity" + : grokCompatibility.diagnostic === "built-in-schema-expired" + ? "Grok Build's built-in compatibility schema has expired; continuing without tool activity" + : grokCompatibility.diagnostic === "no-compatible-schema" + ? "This Grok Build client has no verified tool-event schema; continuing without tool activity" + : "Grok Build's structured event protocol is unavailable; continuing without tool activity"; + pushProgress("grok-compatibility", label, "error", grokCompatibility.diagnostic); + } + if (grokCompatibility?.mode === "plain") { + // Plain output is safe only while it is prose. A changed client can + // still emit an unverified JSON envelope even after capability + // probing failed; never turn that possible tool payload into a + // persisted assistant message or diagnostic. + let unverifiedStructuredOutput = false; + if (isJson) { + try { + const candidate = JSON.parse(line); + unverifiedStructuredOutput = !!candidate && typeof candidate === "object"; + } catch { + // Plain prose can begin with a brace or bracket. It has no + // parseable structured boundary, so preserve it as text. + } + } + if (unverifiedStructuredOutput) { + recordStdoutErrorTail("Grok Build emitted an unverified structured event", true); + if (!grokProtocolQuarantineNoticeSent) { + grokProtocolQuarantineNoticeSent = true; + pushProgress( + "grok-compatibility", + "Grok Build emitted unverified structured output; continuing without tool activity", + "error", + "unverified-structured-output", + ); + } + return; + } + const text = `${resolveBackspaces(stripAnsi(line))}\n`; + // Plain mode has no native session event. Once actual assistant + // prose arrives, the launched UUID is safe to retain for persistence + // and the next resume; do not manufacture it for raw envelopes. + if (!grokSessionId && grokSessionHint) grokSessionId = grokSessionHint; + if (!sessionId && grokSessionHint) announceSession(grokSessionHint); + assistantText += text; + push({ kind: "assistant_chunk", text }); + return; + } if (!isJson) { - recordStdoutErrorTail(resolveBackspaces(stripAnsi(line))); + quarantineGrokProtocol("unframed-jsonl-event"); return; } try { - const event = parseGrokStreamEvent(JSON.parse(line)); + const raw = JSON.parse(line); + const event = grokStructuredProtocolQuarantined + ? parseGrokStreamEvent(raw, grokCompatibility?.schema) + : parseGrokCompatibilityEvent(raw, grokCompatibility?.schema); // A fresh native session's id is assigned by Cave because Grok only // returns it in its final frame. Once its first response frame // confirms the process is live, retain that id for a cancelled @@ -2324,7 +2418,7 @@ export async function POST(req: Request) { // launch means its --model contract accepted the selected id. if (!confirmedModel && grokForwardModel) confirmedModel = desiredModel; result = { - is_error: event.isError, + is_error: false, usage: parseStreamJsonUsage(event.usage), costUsd: parseCostUsd(event.totalCostUsd), }; @@ -2335,7 +2429,61 @@ export async function POST(req: Request) { usage: parseStreamJsonUsage(event.usage), costUsd: parseCostUsd(event.totalCostUsd), }; - recordStdoutErrorTail(event.message); + // Provider error frames are untrusted structured payloads; keep + // their values out of transcript diagnostics. + recordStdoutErrorTail("Grok Build returned a structured error", true); + return; + case "tool_start": { + boundarySentinel?.observe(event.name, event.input); + const started = toolTracker.envelopeToolUse(event.id, event.name, formatToolInputValue(event.input), assistantText.length); + if (started) push({ kind: "tool_use", ...started }); + const progress = toolTracker.consumePendingEnvelopeProgress(event.id); + if (progress) push({ kind: "tool_use", ...progress }); + const ended = toolTracker.consumePendingEnvelopeResult(event.id); + if (ended) push({ kind: "tool_use", ...ended }); + return; + } + case "tool_progress": { + const progress = toolTracker.envelopeToolProgress(event.id, typeof event.output === "string" ? event.output : formatToolInputValue(event.output)); + if (progress) push({ kind: "tool_use", ...progress }); + return; + } + case "tool_end": { + const ended = toolTracker.envelopeToolResult(event.id, typeof event.output === "string" ? event.output : formatToolInputValue(event.output), event.isError); + if (ended) push({ kind: "tool_use", ...ended }); + return; + } + case "tool_complete": { + boundarySentinel?.observe(event.name, event.input); + const started = toolTracker.envelopeToolUse(event.id, event.name, formatToolInputValue(event.input), assistantText.length); + if (started) push({ kind: "tool_use", ...started }); + const progress = toolTracker.consumePendingEnvelopeProgress(event.id); + if (progress) push({ kind: "tool_use", ...progress }); + // A terminal frame can be flushed before the combined snapshot. + // Keep ToolCallTracker's first terminal outcome instead of + // overwriting it with a later duplicate completion. + const reorderedEnd = toolTracker.consumePendingEnvelopeResult(event.id); + if (reorderedEnd) { + push({ kind: "tool_use", ...reorderedEnd }); + return; + } + const ended = toolTracker.envelopeToolResult(event.id, typeof event.output === "string" ? event.output : formatToolInputValue(event.output), event.isError); + if (ended) push({ kind: "tool_use", ...ended }); + return; + } + case "unknown": + grokStructuredProtocolQuarantined = true; + quarantineGrokSchema(grokCompatibility?.schema); + recordStdoutErrorTail("Grok Build emitted a malformed structured event", true); + if (!grokProtocolQuarantineNoticeSent) { + grokProtocolQuarantineNoticeSent = true; + pushProgress( + "grok-compatibility", + "Grok Build sent an unrecognized event; future turns will use safe plain chat until a compatible schema is available", + "error", + `unknown-event:${redactedGrokEventFingerprint(raw)}`, + ); + } return; case "ignore": return; @@ -2343,7 +2491,7 @@ export async function POST(req: Request) { } catch { /* not valid JSON after all — fall through to the error tail */ } - recordStdoutErrorTail(resolveBackspaces(stripAnsi(line))); + quarantineGrokProtocol("malformed-jsonl-event"); }; const handleOpenCodeLine = (line: string) => { @@ -2517,7 +2665,9 @@ export async function POST(req: Request) { return; } if (grokDirect) { - handleGrokLine(line, isJson); + // Plain Grok fallback must not make a structured envelope with + // leading whitespace look like harmless assistant prose. + handleGrokLine(line, isJson || /^[{\[]/.test(line.trimStart())); return; } if (openCodeDirect) { @@ -3199,7 +3349,7 @@ export async function POST(req: Request) { // OpenCode normally emits a JSON error envelope, but older CLI // builds can exit non-zero with only stderr. Do not mistake that // failed invocation for a successful model application below. - if ((openCodeDirect || copilotStream) && code !== 0) { + if ((openCodeDirect || copilotStream || grokDirect) && code !== 0) { result = { ...result, is_error: true }; } // Copilot JSONL stderr can contain raw prompt or tool payloads on @@ -3212,6 +3362,14 @@ export async function POST(req: Request) { stdoutErrTail.push("Copilot exited before completing its response."); } } + // Grok's stderr can include provider request details or local + // paths. Its structured errors already yield a fixed diagnostic; + // an unframed non-zero exit must receive the same redaction. + if (grokDirect) { + stderrTail.length = 0; + stdoutErrTail.length = 0; + if (code !== 0) stdoutErrTail.push("Grok Build exited before completing its response."); + } pushProgress( "harness-start", `${binding.harness} exited`, @@ -3406,7 +3564,7 @@ export async function POST(req: Request) { // OpenCode stderr can include request bodies, provider diagnostics, or // local paths. It is useful for other harnesses' existing recovery // guidance, but must never become assistant-visible/persisted text. - const tailBlock = !openCodeDirect && tailSource.length + const tailBlock = !openCodeDirect && !grokDirect && tailSource.length ? `\n\n\`\`\`\n${tailSource.slice(-5).join("\n")}\n\`\`\`` : ""; const diagnostic = result.is_error @@ -3472,11 +3630,11 @@ export async function POST(req: Request) { // us to normalize provider text. Preserve its leading indentation and // trailing blank lines in the durable transcript as well as the live // stream; other harnesses retain their established trim behavior. - const assistantTextForPersistence = openCodeDirect && openCodeCompatibility?.mode === "plain" + const assistantTextForPersistence = (openCodeDirect && openCodeCompatibility?.mode === "plain") || (grokDirect && grokCompatibility?.mode === "plain") ? assistantText : assistantText.trim(); const { text: cleanedAssistantText, attachments: agentAttachments } = - openCodeDirect && openCodeCompatibility?.mode === "plain" + (openCodeDirect && openCodeCompatibility?.mode === "plain") || (grokDirect && grokCompatibility?.mode === "plain") ? { text: assistantTextForPersistence, attachments: [] } : parseAgentAttachments(assistantTextForPersistence, { allowedRoots: sshRuntime ? [] : [cwd, ...grantedProjectRoots], @@ -3610,8 +3768,8 @@ export async function POST(req: Request) { ...(result.usage ? { usage: result.usage } : {}), ...(result.costUsd !== undefined ? { costUsd: result.costUsd } : {}), ...(persistedTools ? { tools: persistedTools } : {}), - ...(persistedOpenCodeDiagnostics.length - ? { progress: persistedOpenCodeDiagnostics } + ...(persistedCompatibilityDiagnostics.length + ? { progress: persistedCompatibilityDiagnostics } : {}), parentId: userTurnId, responseMetadata, diff --git a/src/lib/grok-build.test.ts b/src/lib/grok-build.test.ts index ecdcb120f..1c91e17b5 100644 --- a/src/lib/grok-build.test.ts +++ b/src/lib/grok-build.test.ts @@ -8,6 +8,10 @@ import { parseGrokModels, parseGrokStreamEvent, } from "./grok-build.ts"; +import { + BUILTIN_GROK_SCHEMA_BUNDLE, + parseGrokCompatibilityEvent, +} from "./grok-compatibility.ts"; const catalog = parseGrokModels(`You are logged in with grok.com.\n\nDefault model: grok-4.5\n\nAvailable models:\n * grok-4.5 (default)\n * grok-code-fast-1`); assert.equal(catalog.defaultModel, "grok-4.5"); @@ -29,6 +33,7 @@ assert.deepEqual( permissionMode: "read", grantDirs: ["/work/project", ""], identityRules: "You are Nova.", + outputFormat: "streaming-json", }), [ "--no-auto-update", "--output-format", "streaming-json", @@ -50,6 +55,7 @@ assert.ok( permissionMode: "full", grantDirs: [], identityRules: "", + outputFormat: "streaming-json", }).includes("--sandbox"), "resumed chats preserve Grok's session-bound sandbox instead of failing on a changed composer mode", ); @@ -63,6 +69,7 @@ assert.deepEqual( permissionMode: "full", grantDirs: [], identityRules: "", + outputFormat: "streaming-json", }), [ "--no-auto-update", "--output-format", "streaming-json", @@ -93,6 +100,11 @@ assert.deepEqual( { kind: "error", message: "not authenticated", usage: undefined, totalCostUsd: 0 }, ); assert.deepEqual(parseGrokStreamEvent({ type: "thought", data: "hidden" }), { kind: "ignore" }); +assert.deepEqual( + parseGrokCompatibilityEvent({ type: "future_event" }, BUILTIN_GROK_SCHEMA_BUNDLE.schemas[0]), + { kind: "unknown" }, + "unknown future frames never become activity in the built-in schema", +); assert.equal(grokSandboxProfileForPermission("read"), "read"); assert.equal(grokSandboxProfileForPermission(undefined), "full"); diff --git a/src/lib/grok-build.ts b/src/lib/grok-build.ts index da31a4f77..dcf2a1141 100644 --- a/src/lib/grok-build.ts +++ b/src/lib/grok-build.ts @@ -4,6 +4,12 @@ // proposed registry adapter is not merged, while Cave needs the CLI's real // streaming/session contract to drive a local chat safely. +import { + BUILTIN_GROK_SCHEMA_BUNDLE, + parseGrokCompatibilityEvent, + type GrokEventSchema, +} from "./grok-compatibility.ts"; + export type RuntimeModelOption = { id: string; label: string }; export type GrokSandboxProfile = "full" | "read"; @@ -106,8 +112,11 @@ export function buildGrokBuildArgs(input: { permissionMode: "full" | "read"; grantDirs: string[]; identityRules: string; + /** Omit structured output entirely when no local capability/schema proves it safe. */ + outputFormat?: "streaming-json" | null; }): string[] { - const args = ["--no-auto-update", "--output-format", "streaming-json"]; + const args = ["--no-auto-update"]; + if (input.outputFormat === "streaming-json") args.push("--output-format", input.outputFormat); if (input.resumeSessionId) args.push("--resume", input.resumeSessionId); else if (input.newSessionId) args.push("--session-id", input.newSessionId); const model = bareModel(input.model); @@ -134,35 +143,12 @@ export function buildGrokBuildArgs(input: { } /** Map Grok Build's documented streaming-json JSONL frames to Cave events. */ -export function parseGrokStreamEvent(raw: unknown): GrokStreamEvent { - if (!raw || typeof raw !== "object") return { kind: "ignore" }; - const event = raw as { - type?: unknown; - data?: unknown; - sessionId?: unknown; - message?: unknown; - usage?: unknown; - total_cost_usd?: unknown; - }; - if (event.type === "text" && typeof event.data === "string") { - return { kind: "text", text: event.data }; - } - if (event.type === "end") { - return { - kind: "end", - sessionId: typeof event.sessionId === "string" ? event.sessionId : undefined, - isError: false, - usage: event.usage, - totalCostUsd: event.total_cost_usd, - }; - } - if (event.type === "error") { - return { - kind: "error", - message: typeof event.message === "string" ? event.message : "Grok Build returned an error.", - usage: event.usage, - totalCostUsd: event.total_cost_usd, - }; +export function parseGrokStreamEvent(raw: unknown, schema: GrokEventSchema = BUILTIN_GROK_SCHEMA_BUNDLE.schemas[0]): GrokStreamEvent { + const event = parseGrokCompatibilityEvent(raw, schema); + switch (event.kind) { + case "text": return event; + case "end": return { ...event, isError: false }; + case "error": return event; + default: return { kind: "ignore" }; } - return { kind: "ignore" }; } diff --git a/src/lib/grok-compatibility.test.ts b/src/lib/grok-compatibility.test.ts new file mode 100644 index 000000000..6f24fa90d --- /dev/null +++ b/src/lib/grok-compatibility.test.ts @@ -0,0 +1,173 @@ +import assert from "node:assert/strict"; +import { generateKeyPairSync, sign } from "node:crypto"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { + BUILTIN_GROK_SCHEMA_BUNDLE, + grokProbeEnvironment, + grokRunCapabilitiesFromHelp, + grokSchemaBundlePayloadHash, + grokSchemaBundleSigningPayload, + isGrokSchemaBundle, + resolveGrokCompatibility, + verifyGrokSchemaBundle, +} from "./grok-compatibility.ts"; + +const { privateKey, publicKey } = generateKeyPairSync("ed25519"); +const signed = structuredClone(BUILTIN_GROK_SCHEMA_BUNDLE); +signed.sequence = 2; +signed.keyId = "test-key"; +signed.signature = { + algorithm: "ed25519", + value: sign(null, Buffer.from(grokSchemaBundleSigningPayload(signed)), privateKey).toString("base64"), +}; +const keyring = { "test-key": publicKey.export({ type: "spki", format: "pem" }).toString() }; +assert.equal(verifyGrokSchemaBundle(signed, keyring), true, "Ed25519 verification accepts a canonical signed bundle"); +assert.equal(verifyGrokSchemaBundle({ ...signed, sequence: 3 }, keyring), false, "a signature cannot be replayed onto changed schema data"); +const ambiguous = structuredClone(signed); +ambiguous.schemas[0].eventTypes.toolStart = ["text"]; +assert.equal(isGrokSchemaBundle(ambiguous), false, "ambiguous event aliases are rejected before a signed schema can be selected"); +assert.deepEqual( + signed.schemas[0].eventTypes, + { ignored: ["thought"], text: ["text"], end: ["end"], error: ["error"], toolStart: [], toolProgress: [], toolEnd: [], toolComplete: [] }, + "the built-in schema makes no undocumented tool-event claim", +); +const { keyId: _keyId, signature: _signature, ...unsignedWithoutKeyId } = signed; +const signedWithoutKeyId = { + ...unsignedWithoutKeyId, + signature: { algorithm: "ed25519" as const, value: sign(null, Buffer.from(grokSchemaBundleSigningPayload(unsignedWithoutKeyId)), privateKey).toString("base64") }, +}; +assert.equal(verifyGrokSchemaBundle(signedWithoutKeyId, { "test-key": keyring["test-key"], next: keyring["test-key"] }), false, "rotating keyrings require an explicit signed key id"); +assert.equal( + verifyGrokSchemaBundle(signed, { + "test-key": keyring["test-key"], one: keyring["test-key"], two: keyring["test-key"], three: keyring["test-key"], four: keyring["test-key"], + }), + false, + "key rotation fails closed when more than four configured public keys are supplied", +); + +assert.deepEqual( + grokRunCapabilitiesFromHelp(" --output-format FORMAT Choose text or streaming-json\n --model MODEL\n"), + { version: null, streamingJson: true, options: ["--output-format", "--model"], valueOptions: ["--output-format", "--model"] }, + "the output value must appear in --output-format's own help stanza", +); +assert.equal(grokRunCapabilitiesFromHelp(" --output-format\n --other streaming-json\n").streamingJson, false, "unrelated help prose cannot authorize structured argv"); +assert.deepEqual(grokProbeEnvironment({ NODE_ENV: "test", PATH: "/bin", XAI_API_KEY: "secret", GROK_AUTH_TOKEN: "secret", SERVICE_PASSWORD: "secret", SESSION_AUTH: "secret", HTTPS_PROXY: "https://user:secret@proxy.example" }), { NODE_ENV: "test", PATH: "/bin" }, "capability probes inherit no credential-bearing environment variables"); + +const supported = { version: "fixture", streamingJson: true, options: ["--output-format"], valueOptions: ["--output-format"] }; +const selected = await resolveGrokCompatibility(supported, { publicKeys: keyring, fetch: async () => new Response(JSON.stringify(signed)) }); +assert.equal(selected.mode, "structured"); +assert.equal(selected.schema?.id, "grok-build-streaming-json-v1"); +const unsupported = await resolveGrokCompatibility({ ...supported, streamingJson: false }); +assert.deepEqual(unsupported.diagnostic, "streaming-json-unavailable"); +let unsafeFetchCalled = false; +const unsafeRegistry = await resolveGrokCompatibility(supported, { publicKeys: keyring, url: "http://registry.example/grok.json", fetch: async () => { unsafeFetchCalled = true; return new Response(); } }); +assert.equal(unsafeFetchCalled, false, "an unsafe registry URL is rejected before any request"); +assert.equal(unsafeRegistry.bundleSource, "built-in"); + +const highWater = structuredClone(signed); +highWater.sequence = 3; +highWater.signature = { + algorithm: "ed25519", + value: sign(null, Buffer.from(grokSchemaBundleSigningPayload(highWater)), privateKey).toString("base64"), +}; +const cacheDirectory = await mkdtemp(path.join(tmpdir(), "coven-grok-registry-")); +const cachePath = path.join(cacheDirectory, "schema.json"); +try { + const refreshed = await resolveGrokCompatibility(supported, { publicKeys: keyring, cachePath, url: "https://registry.example/grok.json", fetch: async () => new Response(JSON.stringify(highWater)) }); + assert.equal(refreshed.bundleSource, "remote"); + let freshCacheFetched = false; + const freshCache = await resolveGrokCompatibility(supported, { + publicKeys: keyring, + cachePath, + url: "https://registry.example/grok.json", + fetch: async () => { freshCacheFetched = true; return new Response(JSON.stringify(signed)); }, + }); + assert.equal(freshCache.bundleSource, "cache", "a fresh verified cache remains available without a network round trip"); + assert.equal(freshCacheFetched, false, "a fresh cache does not delay every chat on registry I/O"); + const rollback = await resolveGrokCompatibility(supported, { publicKeys: keyring, cachePath, now: () => Date.now() + 7 * 60 * 60 * 1000, url: "https://registry.example/grok.json", fetch: async () => new Response(JSON.stringify(signed)) }); + assert.equal(rollback.bundleSource, "cache", "a lower signed sequence cannot replace the local high-water contract"); + assert.equal(rollback.diagnostic, "schema-registry-refresh-rejected"); + + const stalledRefresh = await Promise.race([ + resolveGrokCompatibility(supported, { + publicKeys: keyring, + cachePath, + now: () => Date.now() + 14 * 60 * 60 * 1000, + url: "https://registry.example/grok.json", + refreshTimeoutMs: 10, + fetch: async () => new Response(new ReadableStream()), + }), + new Promise((_, reject) => setTimeout(() => reject(new Error("Grok registry body read was not bounded")), 500)), + ]); + assert.equal(stalledRefresh.bundleSource, "cache", "a registry response that never finishes falls back to the verified cache"); + assert.equal(stalledRefresh.diagnostic, "schema-registry-refresh-rejected", "the bounded body timeout is an accessible refresh failure"); + + const expiredRemote = await resolveGrokCompatibility(supported, { + publicKeys: keyring, + cachePath, + now: () => Date.parse("2031-01-01T00:00:00.000Z"), + url: "https://registry.example/grok.json", + fetch: async () => new Response(JSON.stringify(signed)), + }); + assert.equal(expiredRemote.mode, "plain", "an expired previously trusted remote contract cannot fall back to the older built-in parser"); + assert.equal(expiredRemote.diagnostic, "schema-registry-refresh-rejected"); + + // The immutable journal survives loss of the replaceable recovery sidecar, + // so a stale writer cannot erase high-water state by replacing that sidecar. + await rm(`${cachePath}.anchor`); + const missingAnchor = await resolveGrokCompatibility(supported, { + publicKeys: keyring, + cachePath, + url: "https://registry.example/grok.json", + fetch: async () => new Response(JSON.stringify(signed)), + }); + assert.equal(missingAnchor.bundleSource, "cache", "the immutable high-water journal keeps the verified cache available after sidecar loss"); + assert.equal(missingAnchor.diagnostic, undefined); + + await rm(`${cachePath}.anchor.journal`, { recursive: true, force: true }); + const lostAnchor = await resolveGrokCompatibility(supported, { + publicKeys: keyring, + cachePath, + url: "https://registry.example/grok.json", + fetch: async () => new Response(JSON.stringify(signed)), + }); + assert.equal(lostAnchor.mode, "plain", "a remote cache without any high-water anchor cannot revive the older built-in parser"); + assert.equal(lostAnchor.diagnostic, "cached-schema-unavailable"); + + // Simulate an interruption after the high-water anchor is committed but + // before the replacement cache file can be installed. The prior cache was + // selected before fetch, so this verifies the current turn drops it too. + const interruptedBundle = structuredClone(highWater); + interruptedBundle.sequence = 4; + interruptedBundle.signature = { + algorithm: "ed25519", + value: sign(null, Buffer.from(grokSchemaBundleSigningPayload(interruptedBundle)), privateKey).toString("base64"), + }; + // Restore the expected anchor before exercising an interrupted update. + await writeFile(`${cachePath}.anchor`, JSON.stringify({ sequence: highWater.sequence, payloadHash: grokSchemaBundlePayloadHash(highWater) })); + const interruptedRefresh = await resolveGrokCompatibility(supported, { + publicKeys: keyring, + cachePath, + now: () => Date.now() + 14 * 60 * 60 * 1000, + url: "https://registry.example/grok.json", + fetch: async () => { + await rm(cachePath, { force: true }); + await mkdir(cachePath); + return new Response(JSON.stringify(interruptedBundle)); + }, + }); + assert.equal(interruptedRefresh.bundleSource, "built-in", "a cache below a newly committed high-water anchor is not selected in the interrupted refresh turn"); + assert.equal(interruptedRefresh.diagnostic, "schema-registry-refresh-rejected"); +} finally { + await rm(cacheDirectory, { recursive: true, force: true }); +} + +const expiredBuiltIn = await resolveGrokCompatibility(supported, { + now: () => Date.parse("2031-01-01T00:00:00.000Z"), +}); +assert.equal(expiredBuiltIn.mode, "plain", "the shipped parser expires instead of continuing to parse future Grok output"); +assert.equal(expiredBuiltIn.diagnostic, "built-in-schema-expired"); + +console.log("grok compatibility tests passed"); diff --git a/src/lib/grok-compatibility.ts b/src/lib/grok-compatibility.ts new file mode 100644 index 000000000..a61b08c2e --- /dev/null +++ b/src/lib/grok-compatibility.ts @@ -0,0 +1,792 @@ +/** + * Grok Build's streaming protocol is deliberately treated as a compatibility + * contract, not an invitation to guess at future event names. The compiled + * baseline below is restricted to the frames documented by xAI; a signed + * registry can add a tool envelope only after a local capability probe has + * established that the installed launcher advertises that protocol. + */ +import { createHash, createPublicKey, randomBytes, verify } from "node:crypto"; +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import * as fs from "node:fs/promises"; +import { homedir } from "node:os"; +import path from "node:path"; + +export type GrokRunCapabilities = { + version: string | null; + streamingJson: boolean; + options: string[]; + valueOptions: string[]; +}; + +function optionStanza(help: string, option: string): string { + return help.match(new RegExp(`^\\s*(?:-[A-Za-z],?\\s+)?${option}\\b[^\\n]*(?:\\n(?!\\s*(?:-[A-Za-z],?\\s+)?--)[^\\n]*){0,2}`, "im"))?.[0] ?? ""; +} + +function optionTakesValue(help: string, option: string): boolean { + const stanza = optionStanza(help, option); + if (!stanza) return false; + const declaration = stanza.split(/\r?\n/, 1)[0].slice(stanza.indexOf(option) + option.length); + return /(?:^|\s)(?:<[^>]+>|\[[^\]]+\]|[A-Z][A-Z_-]*)(?:\s|$)/.test(declaration) + || /\[(?:string|number|boolean|array|count)\]/i.test(stanza); +} + +export function grokRunCapabilitiesFromHelp(help: string, version: string | null = null): GrokRunCapabilities { + const options = [...help.matchAll(/^\s*(--[a-z][a-z0-9-]*)\b/gim)].map((match) => match[1]); + const unique = [...new Set(options)]; + const valueOptions = unique.filter((option) => optionTakesValue(help, option)); + return { + version, + // The protocol must be documented in the output option's own stanza; + // a banner or another option mentioning JSON is not launch evidence. + streamingJson: unique.includes("--output-format") + && valueOptions.includes("--output-format") + && /\bstreaming-json\b/i.test(optionStanza(help, "--output-format")), + options: unique, + valueOptions, + }; +} + +/** Help/version probes never receive API keys or other secret-bearing vars. */ +export function grokProbeEnvironment(env: NodeJS.ProcessEnv): NodeJS.ProcessEnv { + const probeEnv = { ...env }; + for (const key of Object.keys(probeEnv)) { + if (/(?:key|token|secret|password|credential|auth(?:orization)?|cookie|proxy|^xai_|^grok_.*(?:auth|api))/i.test(key)) delete probeEnv[key]; + } + return probeEnv; +} + +function terminateGrokProbeTree(child: ChildProcessWithoutNullStreams): Promise { + return new Promise((resolve) => { + let settled = false; + let grace: ReturnType | undefined; + const finish = () => { + if (settled) return; + settled = true; + if (grace) clearTimeout(grace); + resolve(); + }; + grace = setTimeout(finish, 1_000); + child.once("close", () => { clearTimeout(grace); finish(); }); + try { + if (process.platform === "win32" && child.pid) { + const killer = spawn("taskkill.exe", ["/PID", String(child.pid), "/T", "/F"], { stdio: "ignore", windowsHide: true }); + killer.once("close", finish); + killer.once("error", () => { try { child.kill("SIGTERM"); } catch { /* best effort */ } }); + } else if (child.pid) { + // Probes use their own process group on POSIX so a shim cannot leave + // a helper running after a capability timeout. + process.kill(-child.pid, "SIGTERM"); + } else { + child.kill("SIGTERM"); + } + } catch { try { child.kill("SIGTERM"); } catch { /* already exited */ } } + }); +} + +/** Bounded, credential-free local contract probe. It never invokes a run. */ +export async function probeGrokRunCapabilities( + launch: { command: string; fixedArgs: string[] }, + env: NodeJS.ProcessEnv, + timeoutMs = process.platform === "win32" ? 6_000 : 2_500, +): Promise { + const probe = (args: string[]): Promise => new Promise((resolve) => { + let output = ""; let settled = false; + const finish = (value: string | null) => { if (!settled) { settled = true; resolve(value); } }; + try { + const child = spawn(launch.command, [...launch.fixedArgs, ...args], { + env: grokProbeEnvironment(env), + stdio: ["ignore", "pipe", "pipe"], + windowsHide: true, + detached: process.platform !== "win32", + }) as unknown as ChildProcessWithoutNullStreams; + let overflowed = false; + const append = (chunk: Buffer) => { + if (output.length >= 64 * 1024) { overflowed = true; return; } + output += chunk.toString().slice(0, 64 * 1024 - output.length); + if (output.length >= 64 * 1024) overflowed = true; + }; + child.stdout.on("data", append); child.stderr.on("data", append); + const timer = setTimeout(() => { void terminateGrokProbeTree(child).finally(() => finish(null)); }, timeoutMs); + child.once("error", () => { clearTimeout(timer); finish(null); }); + child.once("close", (code) => { clearTimeout(timer); finish(code === 0 && !overflowed ? output : null); }); + } catch { finish(null); } + }); + const [help, versionOutput] = await Promise.all([probe(["--help"]), probe(["--version"])]); + const version = versionOutput?.match(/\d+(?:\.\d+){1,3}(?:[-+][\w.-]+)?/)?.[0] ?? null; + return help ? grokRunCapabilitiesFromHelp(help, version) : { version, streamingJson: false, options: [], valueOptions: [] }; +} + +export type GrokRegistryKeyring = Record; +export type GrokRegistryCheckpoint = { sequence: number; payloadHash: string }; + +/** Data only: registries cannot supply selectors, paths, code, or argv. */ +export type GrokEventSchema = { + id: string; + priority?: number; + requires: { streamingJson: true; options?: string[]; versions?: string[] }; + eventTypes: { + ignored: string[]; + text: string[]; + end: string[]; + error: string[]; + toolStart: string[]; + toolProgress?: string[]; + toolEnd: string[]; + toolComplete: string[]; + }; + fields: { + type: string[]; + text: string[]; + sessionId: string[]; + message: string[]; + usage: string[]; + totalCostUsd: string[]; + id: string[]; + name: string[]; + input: string[]; + output: string[]; + state: string[]; + error: string[]; + terminalStates: string[]; + errorStates: string[]; + }; + /** Only the documented output option/value may be selected by a schema. */ + launch: { outputOption: "--output-format"; outputValue: "streaming-json" }; +}; + +export type GrokSchemaBundle = { + format: 1; + runtime: "grok-build"; + sequence: number; + issuedAt: string; + expiresAt: string; + keyId?: string; + retiredSchemaIds?: string[]; + schemas: GrokEventSchema[]; + signature?: { algorithm: "ed25519"; value: string }; +}; + +export type GrokCompatibilityDiagnostic = + | "streaming-json-unavailable" + | "built-in-schema-expired" + | "no-compatible-schema" + | "schema-quarantined" + | "schema-registry-refresh-rejected" + | "cached-schema-unavailable"; + +export type GrokCompatibility = { + mode: "structured" | "plain"; + capabilities: GrokRunCapabilities; + schema?: GrokEventSchema; + bundleSource: "built-in" | "cache" | "remote"; + diagnostic?: GrokCompatibilityDiagnostic; +}; + +export type GrokParsedEvent = + | { kind: "text"; text: string } + | { kind: "end"; sessionId?: string; usage?: unknown; totalCostUsd?: unknown } + | { kind: "error"; message: string; usage?: unknown; totalCostUsd?: unknown } + | { kind: "tool_start"; id: string; name: string; input: unknown } + | { kind: "tool_progress"; id: string; output: unknown } + | { kind: "tool_end"; id: string; output: unknown; isError: boolean } + | { kind: "tool_complete"; id: string; name: string; input: unknown; output: unknown; isError: boolean } + | { kind: "ignore" } + | { kind: "unknown" }; + +const MAX_BUNDLE_BYTES = 256 * 1024; +const MAX_CACHE_BYTES = 512 * 1024; +const MAX_SCHEMAS = 32; +const CACHE_TTL_MS = 6 * 60 * 60 * 1000; +const CACHE_LOCK_STALE_MS = 30_000; +const CACHE_LOCK_WAIT_MS = 500; +const CACHE_LOCK_POLL_MS = 20; +const MAX_ANCHOR_JOURNAL_ENTRIES = 16; +const MAX_ANCHOR_JOURNAL_SCAN = 64; +const schemaQuarantine = new Map(); + +/** The only source-verified built-in Grok frame contract as of 2026-07-26. */ +export const BUILTIN_GROK_SCHEMA_BUNDLE: GrokSchemaBundle = { + format: 1, + runtime: "grok-build", + sequence: 1, + issuedAt: "2026-07-26T00:00:00.000Z", + expiresAt: "2030-01-01T00:00:00.000Z", + schemas: [{ + id: "grok-build-streaming-json-v1", + requires: { streamingJson: true, options: ["--output-format"] }, + eventTypes: { + ignored: ["thought"], + text: ["text"], end: ["end"], error: ["error"], + toolStart: [], toolProgress: [], toolEnd: [], toolComplete: [], + }, + fields: { + type: ["type"], text: ["data"], sessionId: ["sessionId", "session_id"], + message: ["message"], usage: ["usage"], totalCostUsd: ["total_cost_usd"], + id: ["id", "tool_call_id"], name: ["name", "tool_name"], input: ["input"], + output: ["output", "result"], state: ["state", "status"], error: ["error"], + terminalStates: ["completed", "complete", "failed", "error"], errorStates: ["failed", "error"], + }, + launch: { outputOption: "--output-format", outputValue: "streaming-json" }, + }], +}; + +function isRecord(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +function stringField(value: Record, aliases: string[]): string | undefined { + for (const alias of aliases) if (typeof value[alias] === "string") return value[alias]; + return undefined; +} + +function valueField(value: Record, aliases: string[]): unknown { + for (const alias of aliases) if (alias in value) return value[alias]; + return undefined; +} + +export function redactedGrokEventFingerprint(value: unknown): string { + const safe = ["type", "data", "sessionId", "session_id", "id", "tool_call_id", "name", "tool_name", "state", "status"]; + const shape = isRecord(value) + ? Object.fromEntries(safe.filter((key) => key in value).map((key) => [key, typeof value[key]])) + : typeof value; + return createHash("sha256").update(JSON.stringify(shape)).digest("hex").slice(0, 16); +} + +/** A frame only becomes tool activity after the selected schema explicitly names it. */ +export function parseGrokCompatibilityEvent(raw: unknown, schema?: GrokEventSchema): GrokParsedEvent { + if (!schema || !isRecord(raw)) return { kind: "unknown" }; + const type = stringField(raw, schema.fields.type); + if (!type) return { kind: "unknown" }; + const fields = schema.fields; + if (schema.eventTypes.ignored.includes(type)) return { kind: "ignore" }; + if (schema.eventTypes.text.includes(type)) { + const text = stringField(raw, fields.text); + return text === undefined ? { kind: "unknown" } : { kind: "text", text }; + } + if (schema.eventTypes.end.includes(type)) return { kind: "end", sessionId: stringField(raw, fields.sessionId), usage: valueField(raw, fields.usage), totalCostUsd: valueField(raw, fields.totalCostUsd) }; + if (schema.eventTypes.error.includes(type)) return { kind: "error", message: stringField(raw, fields.message) ?? "Grok Build returned an error.", usage: valueField(raw, fields.usage), totalCostUsd: valueField(raw, fields.totalCostUsd) }; + const id = stringField(raw, fields.id); + if (!id) return { kind: "unknown" }; + if (schema.eventTypes.toolStart.includes(type)) { + const name = stringField(raw, fields.name); + return name === undefined ? { kind: "unknown" } : { kind: "tool_start", id, name, input: valueField(raw, fields.input) }; + } + if ((schema.eventTypes.toolProgress ?? []).includes(type)) return { kind: "tool_progress", id, output: valueField(raw, fields.output) }; + const errorValue = valueField(raw, fields.error); + const isError = fields.errorStates.includes(stringField(raw, fields.state) ?? "") + || errorValue === true + || (typeof errorValue === "string" && errorValue.length > 0) + || (isRecord(errorValue) && Object.keys(errorValue).length > 0); + if (schema.eventTypes.toolEnd.includes(type)) return { kind: "tool_end", id, output: valueField(raw, fields.output), isError }; + if (schema.eventTypes.toolComplete.includes(type)) { + const name = stringField(raw, fields.name); + return name === undefined ? { kind: "unknown" } : { kind: "tool_complete", id, name, input: valueField(raw, fields.input), output: valueField(raw, fields.output), isError }; + } + return { kind: "unknown" }; +} + +function stableJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`; + if (!isRecord(value)) return JSON.stringify(value); + return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${stableJson(value[key])}`).join(",")}}`; +} + +export function grokSchemaBundleSigningPayload(bundle: GrokSchemaBundle): string { + const { signature: _signature, ...unsigned } = bundle; + return stableJson(unsigned); +} + +export function grokSchemaBundlePayloadHash(bundle: GrokSchemaBundle): string { + return createHash("sha256").update(grokSchemaBundleSigningPayload(bundle)).digest("hex"); +} + +function validName(value: unknown): value is string { + return typeof value === "string" && /^[a-z][a-z0-9_-]{0,95}$/i.test(value); +} + +function validAliases(value: unknown, allowEmpty = false): value is string[] { + return Array.isArray(value) && (allowEmpty || value.length > 0) && value.length <= 8 && value.every(validName) && new Set(value).size === value.length; +} + +function validOptions(value: unknown): value is string[] { + return Array.isArray(value) && value.length <= 16 && value.every((option) => typeof option === "string" && /^--[a-z][a-z0-9-]{0,63}$/.test(option)) && new Set(value).size === value.length; +} + +/** Keep rotation configuration as strict as the signed bundle's key-id rules. */ +function validKeyring(value: unknown): value is GrokRegistryKeyring { + if (!isRecord(value)) return false; + const entries = Object.entries(value); + if (entries.length < 1 || entries.length > 4) return false; + try { + for (const [id, pem] of entries) { + if (!validName(id) || typeof pem !== "string" || createPublicKey(pem).asymmetricKeyType !== "ed25519") return false; + } + return true; + } catch { + return false; + } +} + +function validSchema(value: unknown): value is GrokEventSchema { + if (!isRecord(value) || !validName(value.id) || !isRecord(value.requires) || value.requires.streamingJson !== true || !isRecord(value.eventTypes) || !isRecord(value.fields) || !isRecord(value.launch)) return false; + if (!Object.keys(value).every((key) => ["id", "priority", "requires", "eventTypes", "fields", "launch"].includes(key))) return false; + if (!Object.keys(value.requires).every((key) => key === "streamingJson" || key === "options" || key === "versions")) return false; + if (!Object.keys(value.eventTypes).every((key) => ["ignored", "text", "end", "error", "toolStart", "toolProgress", "toolEnd", "toolComplete"].includes(key))) return false; + if (!Object.keys(value.fields).every((key) => ["type", "text", "sessionId", "message", "usage", "totalCostUsd", "id", "name", "input", "output", "state", "error", "terminalStates", "errorStates"].includes(key))) return false; + if (!Object.keys(value.launch).every((key) => key === "outputOption" || key === "outputValue")) return false; + if (value.priority !== undefined && (typeof value.priority !== "number" || !Number.isSafeInteger(value.priority) || Math.abs(value.priority) > 1_000)) return false; + if (value.requires.options !== undefined && !validOptions(value.requires.options)) return false; + if (value.requires.versions !== undefined && (!Array.isArray(value.requires.versions) || value.requires.versions.length < 1 || value.requires.versions.length > 8 || new Set(value.requires.versions).size !== value.requires.versions.length || !value.requires.versions.every((version) => typeof version === "string" && /^\d+(?:\.\d+){1,3}(?:[-+][A-Za-z0-9.-]+)?$/.test(version)))) return false; + if (value.launch.outputOption !== "--output-format" || value.launch.outputValue !== "streaming-json") return false; + const events = value.eventTypes; + const eventGroups = [events.ignored, events.text, events.end, events.error, events.toolStart, events.toolProgress ?? [], events.toolEnd, events.toolComplete]; + if (!eventGroups.every((items) => validAliases(items, true))) return false; + // A signed schema is still untrusted input: reject ambiguous event names and + // incomplete tool envelopes before selection rather than quarantining only + // after a real user's stream reaches the parser. + if (new Set(eventGroups.flat()).size !== eventGroups.flat().length) return false; + const f = value.fields; + if (![f.type, f.text, f.sessionId, f.message, f.usage, f.totalCostUsd, f.id, f.name, f.input, f.output, f.state, f.error, f.terminalStates, f.errorStates].every((items) => validAliases(items, true))) return false; + const typedEvents = events as GrokEventSchema["eventTypes"]; + const typedFields = f as GrokEventSchema["fields"]; + if (!typedFields.type.length) return false; + if (typedEvents.text.length && !typedFields.text.length) return false; + if ((typedEvents.toolStart.length || typedEvents.toolEnd.length || typedEvents.toolComplete.length || (typedEvents.toolProgress?.length ?? 0)) && !typedFields.id.length) return false; + // Tool envelopes are version-sensitive protocol claims. A publisher must + // bind them to the exact locally probed launcher revisions rather than + // allowing a signed alias to match every future Grok Build version. + if ((typedEvents.toolStart.length || typedEvents.toolEnd.length || typedEvents.toolComplete.length || (typedEvents.toolProgress?.length ?? 0)) && !value.requires.versions?.length) return false; + return !(typedEvents.toolStart.length || typedEvents.toolComplete.length) || typedFields.name.length > 0; +} + +function canonicalTime(value: unknown): number | null { + if (typeof value !== "string" || !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/.test(value)) return null; + const time = Date.parse(value); + return Number.isFinite(time) && new Date(time).toISOString() === value ? time : null; +} + +export function isGrokSchemaBundle(value: unknown, now = Date.now(), allowExpired = false): value is GrokSchemaBundle { + if (!isRecord(value) || value.format !== 1 || value.runtime !== "grok-build" || typeof value.sequence !== "number" || !Number.isSafeInteger(value.sequence) || value.sequence < 1 || !Array.isArray(value.schemas) || value.schemas.length < 1 || value.schemas.length > MAX_SCHEMAS || !value.schemas.every(validSchema)) return false; + if (!Object.keys(value).every((key) => ["format", "runtime", "sequence", "issuedAt", "expiresAt", "keyId", "retiredSchemaIds", "schemas", "signature"].includes(key))) return false; + const issued = canonicalTime(value.issuedAt); const expires = canonicalTime(value.expiresAt); + if (issued === null || expires === null || issued > now || expires <= issued || (!allowExpired && expires <= now)) return false; + if (value.keyId !== undefined && !validName(value.keyId)) return false; + if (value.retiredSchemaIds !== undefined && (!validAliases(value.retiredSchemaIds, true) || !value.retiredSchemaIds.every((id) => BUILTIN_GROK_SCHEMA_BUNDLE.schemas.some((schema) => schema.id === id)))) return false; + if (value.signature !== undefined && (!isRecord(value.signature) || !Object.keys(value.signature).every((key) => key === "algorithm" || key === "value") || value.signature.algorithm !== "ed25519" || typeof value.signature.value !== "string")) return false; + const ids = value.schemas.map((schema) => schema.id); + return new Set(ids).size === ids.length; +} + +export function verifyGrokSchemaBundle(value: unknown, publicKeys: string | GrokRegistryKeyring, now = Date.now()): value is GrokSchemaBundle { + if (!isGrokSchemaBundle(value, now) || !value.signature) return false; + const keys = typeof publicKeys === "string" ? { legacy: publicKeys } : publicKeys; + if (!validKeyring(keys)) return false; + if (Object.keys(keys).length > 1 && value.keyId === undefined) return false; + const candidates = Object.entries(keys).filter(([id, pem]) => validName(id) && typeof pem === "string" && (value.keyId === undefined || value.keyId === id)); + if (candidates.length === 0 || candidates.length > 4 || !/^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/.test(value.signature.value)) return false; + try { + const signature = Buffer.from(value.signature.value, "base64"); + return signature.toString("base64") === value.signature.value && candidates.some(([, pem]) => { + const key = createPublicKey(pem); + return key.asymmetricKeyType === "ed25519" && verify(null, Buffer.from(grokSchemaBundleSigningPayload(value)), key, signature); + }); + } catch { return false; } +} + +export function selectGrokSchema(schemas: GrokEventSchema[], capabilities: GrokRunCapabilities): GrokEventSchema | null { + if (!capabilities.streamingJson) return null; + const options = new Set(capabilities.options); + const valueOptions = new Set(capabilities.valueOptions); + const matches = schemas.filter((schema) => + (schema.requires.options?.every((option) => options.has(option)) ?? true) + && (schema.requires.versions === undefined || (capabilities.version !== null && schema.requires.versions.includes(capabilities.version))) + && options.has(schema.launch.outputOption) + && valueOptions.has(schema.launch.outputOption), + ); + if (!matches.length) return null; + const specificity = Math.max(...matches.map((schema) => schema.requires.options?.length ?? 0)); + const preferred = matches.filter((schema) => (schema.requires.options?.length ?? 0) === specificity); + const priority = Math.max(...preferred.map((schema) => schema.priority ?? 0)); + const winners = preferred.filter((schema) => (schema.priority ?? 0) === priority); + return winners.length === 1 ? winners[0] : null; +} + +export function quarantineGrokSchema(schema: GrokEventSchema | undefined): void { + if (!schema) return; + if (schemaQuarantine.size >= 64 && !schemaQuarantine.has(schema.id)) schemaQuarantine.delete(schemaQuarantine.keys().next().value!); + schemaQuarantine.set(schema.id, createHash("sha256").update(stableJson(schema)).digest("hex")); +} + +type RegistrySource = { + url?: string; + publicKeys?: GrokRegistryKeyring; + checkpoint?: GrokRegistryCheckpoint; + now?: () => number; + fetch?: typeof globalThis.fetch; + cachePath?: string; + /** Test-only override; production refreshes are bounded to five seconds. */ + refreshTimeoutMs?: number; +}; +type Cached = { bundle: GrokSchemaBundle; checkedAt: number }; +type TrustAnchor = { sequence: number; payloadHash: string }; + +// Release-time public verification material becomes the production trust +// anchor. Mutable COVEN_* values remain a development/test escape hatch only; +// a production process environment must not replace its packaged keyring or +// checkpoint after the app has been built. +const PACKAGED_GROK_REGISTRY_URL = process.env.NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_URL; +const PACKAGED_GROK_REGISTRY_PUBLIC_KEY = process.env.NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEY; +const PACKAGED_GROK_REGISTRY_PUBLIC_KEYS = process.env.NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEYS; +const PACKAGED_GROK_REGISTRY_CHECKPOINT = process.env.NEXT_PUBLIC_COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT; + +function configuredGrokRegistrySource(): RegistrySource { + const production = process.env.NODE_ENV === "production"; + const url = PACKAGED_GROK_REGISTRY_URL + ?? (production ? undefined : process.env.COVEN_GROK_SCHEMA_REGISTRY_URL); + const single = PACKAGED_GROK_REGISTRY_PUBLIC_KEY + ?? (production ? undefined : process.env.COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEY); + const rawKeyring = PACKAGED_GROK_REGISTRY_PUBLIC_KEYS + ?? (production ? undefined : process.env.COVEN_GROK_SCHEMA_REGISTRY_PUBLIC_KEYS); + const rawCheckpoint = PACKAGED_GROK_REGISTRY_CHECKPOINT + ?? (production ? undefined : process.env.COVEN_GROK_SCHEMA_REGISTRY_CHECKPOINT); + let publicKeys: GrokRegistryKeyring | undefined; + let checkpoint: GrokRegistryCheckpoint | undefined; + try { publicKeys = rawKeyring ? JSON.parse(rawKeyring) : single ? { legacy: single } : undefined; } catch { /* invalid config fails closed */ } + try { checkpoint = rawCheckpoint ? JSON.parse(rawCheckpoint) : undefined; } catch { /* invalid config fails closed */ } + try { + const parsed = url ? new URL(url) : null; + if (parsed && (parsed.protocol !== "https:" || parsed.username || parsed.password)) return {}; + } catch { return {}; } + if (!validKeyring(publicKeys) || !checkpoint || !Number.isSafeInteger(checkpoint.sequence) || checkpoint.sequence < 1 || !/^[a-f0-9]{64}$/.test(checkpoint.payloadHash)) return {}; + return { url, publicKeys, checkpoint }; +} + +function defaultCachePath(): string { + return path.join(process.env.COVEN_CAVE_HOME || path.join(process.env.COVEN_HOME || homedir(), ".coven", "cave"), "grok-schema-bundle-v1.json"); +} + +function trustAnchorPath(file: string): string { return `${file}.anchor`; } +function trustAnchorJournalPath(file: string): string { return `${trustAnchorPath(file)}.journal`; } +function trustAnchorJournalEntryPath(file: string, anchor: TrustAnchor): string { + return path.join(trustAnchorJournalPath(file), `${anchor.sequence}-${anchor.payloadHash}.json`); +} +function cacheLockPath(file: string): string { return `${file}.lock`; } + +function validTrustAnchor(value: unknown): value is TrustAnchor { + return isRecord(value) && Object.keys(value).length === 2 && typeof value.sequence === "number" && Number.isSafeInteger(value.sequence) && value.sequence >= 1 && typeof value.payloadHash === "string" && /^[a-f0-9]{64}$/.test(value.payloadHash); +} + +function isSafeRegistryUrl(value: string | undefined): value is string { + try { + const parsed = value ? new URL(value) : null; + return !!parsed && parsed.protocol === "https:" && !parsed.username && !parsed.password; + } catch { return false; } +} + +async function readBoundedRegistryBody(response: Response): Promise { + const declaredLength = response.headers.get("content-length"); + if (declaredLength && (!/^\d+$/.test(declaredLength) || Number(declaredLength) > MAX_BUNDLE_BYTES)) throw new Error("oversized registry response"); + if (!response.body) return ""; + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let total = 0; + try { + while (true) { + const part = await reader.read(); + if (part.done) break; + total += part.value.byteLength; + if (total > MAX_BUNDLE_BYTES) throw new Error("oversized registry response"); + chunks.push(part.value); + } + } finally { + // A registry server can leave a chunked response open forever. The outer + // refresh deadline races this read, and cancellation makes that deadline + // release its socket/body reader rather than leaving it pinned in Node. + await reader.cancel().catch(() => undefined); + } + const bytes = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.byteLength; } + return new TextDecoder("utf-8", { fatal: true }).decode(bytes); +} + +async function readLegacyTrustAnchor(file: string): Promise { + try { + const raw = await fs.readFile(trustAnchorPath(file), "utf8"); + if (Buffer.byteLength(raw) > 1024) return null; + const anchor = JSON.parse(raw); + return validTrustAnchor(anchor) ? anchor : null; + } catch { return null; } +} + +/** + * High-water records use immutable sequence/hash filenames. A reclaimed + * writer lease can therefore add a stale record but can never replace the + * newer record that readers use for rollback protection. + */ +async function readJournalTrustAnchors(file: string): Promise { + let directory: Awaited>; + try { + directory = await fs.opendir(trustAnchorJournalPath(file)); + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ENOENT" ? [] : null; + } + const anchors: TrustAnchor[] = []; + try { + let count = 0; + for await (const entry of directory) { + count += 1; + if (count > MAX_ANCHOR_JOURNAL_SCAN || !entry.isFile()) return null; + const match = /^(\d{1,16})-([a-f0-9]{64})\.json$/.exec(entry.name); + if (!match) return null; + const sequence = Number(match[1]); + if (!Number.isSafeInteger(sequence) || sequence < 1) return null; + anchors.push({ sequence, payloadHash: match[2] }); + } + } finally { + await directory.close().catch(() => undefined); + } + return anchors; +} + +async function readTrustAnchor(file: string): Promise { + const [legacy, journal] = await Promise.all([readLegacyTrustAnchor(file), readJournalTrustAnchors(file)]); + if (journal === null) return null; + const anchors = legacy ? [...journal, legacy] : journal; + if (!anchors.length) return null; + const highWater = Math.max(...anchors.map((anchor) => anchor.sequence)); + const highest = anchors.filter((anchor) => anchor.sequence === highWater); + return new Set(highest.map((anchor) => anchor.payloadHash)).size === 1 ? highest[0] : null; +} + +async function cacheRecordExists(file: string): Promise { + try { + return (await fs.stat(file)).isFile(); + } catch { + return false; + } +} + +function meetsTrustAnchor(bundle: GrokSchemaBundle, anchor: TrustAnchor | null): boolean { + return !anchor || bundle.sequence > anchor.sequence || (bundle.sequence === anchor.sequence && grokSchemaBundlePayloadHash(bundle) === anchor.payloadHash); +} + +async function readCache(file: string, keys: GrokRegistryKeyring, now: number, anchor: TrustAnchor | null): Promise { + try { + const raw = await fs.readFile(file, "utf8"); + if (Buffer.byteLength(raw) > MAX_CACHE_BYTES) return null; + const cached = JSON.parse(raw) as Cached; + return isRecord(cached) && Object.keys(cached).length === 2 && Number.isSafeInteger(cached.checkedAt) + && verifyGrokSchemaBundle(cached.bundle, keys, now) && meetsTrustAnchor(cached.bundle, anchor) + ? cached + : null; + } catch { return null; } +} + +async function writeJsonAtomic(file: string, value: unknown): Promise { + const temporary = `${file}.${process.pid}.${randomBytes(6).toString("hex")}.tmp`; + try { + await fs.mkdir(path.dirname(file), { recursive: true }); + await fs.writeFile(temporary, JSON.stringify(value), { mode: 0o600 }); + await fs.rename(temporary, file); + return true; + } catch { return false; /* Cache is an optimization; a failed write never broadens parsing. */ } +} + +async function writeTrustAnchor(file: string, anchor: TrustAnchor): Promise { + const entry = trustAnchorJournalEntryPath(file, anchor); + try { + await fs.mkdir(path.dirname(entry), { recursive: true }); + await fs.writeFile(entry, JSON.stringify(anchor), { flag: "wx", mode: 0o600 }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") return false; + try { + const existing = JSON.parse(await fs.readFile(entry, "utf8")); + if (!validTrustAnchor(existing) || existing.sequence !== anchor.sequence || existing.payloadHash !== anchor.payloadHash) return false; + } catch { return false; } + } + const entries = await readJournalTrustAnchors(file); + if (entries === null) return false; + const retained = new Set(entries + .sort((left, right) => right.sequence - left.sequence || right.payloadHash.localeCompare(left.payloadHash)) + .slice(0, MAX_ANCHOR_JOURNAL_ENTRIES) + .map((entryValue) => `${entryValue.sequence}-${entryValue.payloadHash}.json`)); + await Promise.all(entries + .filter((entryValue) => !retained.has(`${entryValue.sequence}-${entryValue.payloadHash}.json`)) + .map((entryValue) => fs.rm(trustAnchorJournalEntryPath(file, entryValue), { force: true }).catch(() => undefined))); + // Keep the original single-record sidecar as a fast recovery hint. Readers + // always consult the immutable journal as well, so this replace cannot lower + // the accepted high-water mark after a stale lock lease resumes. + return writeJsonAtomic(trustAnchorPath(file), anchor); +} + +type CacheLock = { + owns: () => Promise; + release: () => Promise; +}; + +async function acquireCacheLock(file: string): Promise { + const lock = cacheLockPath(file); + const deadline = Date.now() + CACHE_LOCK_WAIT_MS; + while (Date.now() < deadline) { + try { + await fs.mkdir(path.dirname(file), { recursive: true }); + const handle = await fs.open(lock, "wx", 0o600); + const token = `${process.pid}:${randomBytes(8).toString("hex")}`; + await handle.writeFile(token); + await handle.close(); + const owns = async () => { + try { return await fs.readFile(lock, "utf8") === token; } catch { return false; } + }; + return { + owns, + // A stale-lock taker may install its own lock before this writer + // resumes. Never let the former owner remove that newer lock. + release: async () => { if (await owns()) await fs.rm(lock, { force: true }); }, + }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") return null; + try { + const stat = await fs.stat(lock); + if (Date.now() - stat.mtimeMs > CACHE_LOCK_STALE_MS) await fs.rm(lock, { force: true }); + } catch { /* another writer released or replaced the lock */ } + await new Promise((resolve) => setTimeout(resolve, CACHE_LOCK_POLL_MS)); + } + } + return null; +} + +/** Returns false when another process has already committed a newer anchor. */ +async function writeCache(file: string, bundle: GrokSchemaBundle, checkedAt = Date.now()): Promise { + const lock = await acquireCacheLock(file); + if (!lock) return false; + try { + if (!await lock.owns()) return false; + if (!meetsTrustAnchor(bundle, await readTrustAnchor(file))) return false; + // Anchor first: an interruption can make the cache unavailable, never make + // an older cache appear acceptable after a newer bundle was trusted. + const anchor = { sequence: bundle.sequence, payloadHash: grokSchemaBundlePayloadHash(bundle) }; + if (!await writeTrustAnchor(file, anchor)) return false; + if (!await lock.owns()) return false; + if (!meetsTrustAnchor(bundle, await readTrustAnchor(file))) return false; + return writeJsonAtomic(file, { checkedAt, bundle }); + } finally { await lock.release(); } +} + +function trustedBundle(bundle: GrokSchemaBundle, checkpoint?: GrokRegistryCheckpoint): boolean { + if (bundle.sequence < BUILTIN_GROK_SCHEMA_BUNDLE.sequence) return false; + if (bundle.sequence === BUILTIN_GROK_SCHEMA_BUNDLE.sequence && grokSchemaBundleSigningPayload(bundle) !== grokSchemaBundleSigningPayload(BUILTIN_GROK_SCHEMA_BUNDLE)) return false; + return !checkpoint || bundle.sequence > checkpoint.sequence || (bundle.sequence === checkpoint.sequence && grokSchemaBundlePayloadHash(bundle) === checkpoint.payloadHash); +} + +/** The refresh deadline covers headers and an indefinitely open response body. */ +async function fetchGrokSchemaBundle( + url: string, + fetcher: typeof globalThis.fetch, + timeoutMs: number, +): Promise { + const controller = new AbortController(); + let response: Response | undefined; + let timer: ReturnType | undefined; + let timedOut = false; + const deadline = new Promise((_, reject) => { + timer = setTimeout(() => { + timedOut = true; + controller.abort(); + void response?.body?.cancel().catch(() => undefined); + reject(new Error("registry refresh timed out")); + }, timeoutMs); + }); + try { + return await Promise.race([ + (async () => { + response = await fetcher(url, { + signal: controller.signal, + credentials: "omit", + redirect: "error", + headers: { accept: "application/json" }, + }); + if (timedOut || !response.ok) throw new Error("untrusted registry response"); + return readBoundedRegistryBody(response); + })(), + deadline, + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +export async function resolveGrokCompatibility(capabilities: GrokRunCapabilities, source: RegistrySource = configuredGrokRegistrySource()): Promise { + if (!capabilities.streamingJson) return { mode: "plain", capabilities, bundleSource: "built-in", diagnostic: "streaming-json-unavailable" }; + const now = source.now?.() ?? Date.now(); + const keys = source.publicKeys ?? {}; + const cacheFile = source.cachePath ?? defaultCachePath(); + let bundle = BUILTIN_GROK_SCHEMA_BUNDLE; let bundleSource: GrokCompatibility["bundleSource"] = "built-in"; let diagnostic: GrokCompatibilityDiagnostic | undefined; + const anchor = Object.keys(keys).length ? await readTrustAnchor(cacheFile) : null; + // A verified remote cache without its matching high-water anchor is not a + // valid offline contract. Treat it as a trust failure rather than allowing + // a lower signed response to establish a new history after the anchor was + // deleted or corrupted. + const cacheAnchorMissing = Object.keys(keys).length > 0 && !anchor && await cacheRecordExists(cacheFile); + // Once a client has accepted a newer remote contract, the compiled parser + // is no longer an offline fallback: its aliases may describe an older wire + // format. A missing/corrupt cache anchor has the same fail-closed outcome. + const builtInAnchor = { + sequence: BUILTIN_GROK_SCHEMA_BUNDLE.sequence, + payloadHash: grokSchemaBundlePayloadHash(BUILTIN_GROK_SCHEMA_BUNDLE), + }; + const remoteSchemaRequired = cacheAnchorMissing || ( + !!anchor + && (anchor.sequence !== builtInAnchor.sequence || anchor.payloadHash !== builtInAnchor.payloadHash) + ); + const cached = Object.keys(keys).length && !cacheAnchorMissing ? await readCache(cacheFile, keys, now, anchor) : null; + if (cached && trustedBundle(cached.bundle, source.checkpoint)) { bundle = cached.bundle; bundleSource = "cache"; } + const cacheFresh = !!cached && trustedBundle(cached.bundle, source.checkpoint) + && now - cached.checkedAt >= 0 && now - cached.checkedAt < CACHE_TTL_MS; + if (cacheAnchorMissing) { + diagnostic = "cached-schema-unavailable"; + } else if (source.url && Object.keys(keys).length && !cacheFresh) { + try { + if (!isSafeRegistryUrl(source.url)) throw new Error("unsafe registry URL"); + const body = await fetchGrokSchemaBundle(source.url, source.fetch ?? fetch, source.refreshTimeoutMs ?? 5_000); + const remote = JSON.parse(body); + if (!verifyGrokSchemaBundle(remote, keys, now) || !trustedBundle(remote, source.checkpoint) || !meetsTrustAnchor(remote, anchor)) throw new Error("untrusted registry bundle"); + if (await writeCache(cacheFile, remote, now)) { + bundle = remote; bundleSource = "remote"; + } else { + const currentAnchor = await readTrustAnchor(cacheFile); + const currentCache = await readCache(cacheFile, keys, now, currentAnchor); + if (!currentCache || !trustedBundle(currentCache.bundle, source.checkpoint)) throw new Error("registry cache write rejected"); + bundle = currentCache.bundle; bundleSource = "cache"; diagnostic = "schema-registry-refresh-rejected"; + } + } catch { + // `writeCache` commits the high-water anchor before its cache record. + // If the process loses the cache write after that point, the bundle that + // was read before the refresh is now below the durable anchor and cannot + // remain selected for this turn. + const currentAnchor = await readTrustAnchor(cacheFile); + if (bundleSource !== "built-in" && !meetsTrustAnchor(bundle, currentAnchor)) { + bundle = BUILTIN_GROK_SCHEMA_BUNDLE; + bundleSource = "built-in"; + } + diagnostic = "schema-registry-refresh-rejected"; + } + } + if (remoteSchemaRequired && bundleSource === "built-in") { + return { + mode: "plain", + capabilities, + bundleSource, + diagnostic: diagnostic ?? "cached-schema-unavailable", + }; + } + if (bundleSource === "built-in" && Date.parse(BUILTIN_GROK_SCHEMA_BUNDLE.expiresAt) <= now) { + return { mode: "plain", capabilities, bundleSource, diagnostic: "built-in-schema-expired" }; + } + const remoteById = new Map(bundle.schemas.map((schema) => [schema.id, schema])); + const candidates = bundleSource === "built-in" ? bundle.schemas : [...bundle.schemas, ...BUILTIN_GROK_SCHEMA_BUNDLE.schemas.filter((schema) => !bundle.retiredSchemaIds?.includes(schema.id) && !remoteById.has(schema.id))]; + const schema = selectGrokSchema(candidates, capabilities); + if (!schema) return { mode: "plain", capabilities, bundleSource, diagnostic: diagnostic ?? "no-compatible-schema" }; + if (schemaQuarantine.get(schema.id) === createHash("sha256").update(stableJson(schema)).digest("hex")) return { mode: "plain", capabilities, bundleSource, diagnostic: "schema-quarantined" }; + return { mode: "structured", capabilities, schema, bundleSource, diagnostic }; +}