From 8d8cc0ac551ff927ce7bd8dff7937f0d6bb10d13 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Tue, 1 Sep 2026 19:21:00 +0200 Subject: [PATCH] feat(sdk): AgentWorker promotes CLI JSON output for json_schema verification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Enables `json_schema` verification to actually validate the analysis payload for agent steps, not the process-wrapper `CliResult` around it. Before this PR: an agent step's `output` was always `{exit_code, stdout_tail, stderr_tail}`, whose shape didn't match any real schema author's declaration, so every schema-verified agent step failed regardless of what the CLI produced. After: object-shaped JSON in stdout is promoted as `output`; text-emitting stdout falls back to the wrapper so tools that emit progress text still round-trip usefully. Scope note: this closes the PLUMBING gap for gate 2 clause 2 (analyze-story runs end-to-end). It does NOT close gate 2 clause 2 in the RFC's strict sense — RFC-0001 gate 2's done-when requires `hn-monitor` running as a relayflow with a REAL analyzer (deployed, triggered by real events, no bespoke persistence). The stubs here are deterministic shell scripts, not real analyzers, and wake-context injection is still absent. Both follow-ups are named as non-goals below. WHAT SHIPS (against main, one commit; from `git diff main..HEAD --numstat`, pasted after staging and before writing this message): 40 / 1 sdk/src/worker.ts 210 / 0 sdk/tests/live-kernel.test.ts 57 / 0 sdk/tests/parse-json-output.test.ts 13 / 0 testdata/preflight/analyze-story-missing-fields-cli 10 / 0 testdata/preflight/analyze-story-stub-cli 8 / 0 testdata/preflight/analyze-story-text-only-cli Six files, one commit. The three stubs are deterministic shell scripts (8, 10, and 13 lines). BEHAVIOR - `AgentWorker.execute` (sdk/src/worker.ts): after invoking the step's declared CLI, tries `parseJsonOutput(stdout.trim())`. On success (object-shaped JSON), that value becomes the step's `output`. On non-object JSON, non-JSON stdout, or empty stdout, falls back to the CliResult wrapper. - `parseJsonOutput` (sdk/src/worker.ts, exported): trim → JSON.parse → require object (not scalar, not array). Rejects mixed text+JSON output too. Chatty LLM CLIs that emit progress text plus a JSON blob will fall back to the wrapper. - Implicit contract documented next to the code: CLIs signal errors via non-zero exit, not by emitting an error JSON with exit 0. `completionReason` is derived from exit code. TESTS 10 new tests total: 7 unit + 3 integration. Unit (sdk/tests/parse-json-output.test.ts, 7 tests): - empty stdout → null - non-JSON → null (three shapes) - object payload → parsed - trims whitespace - scalars → null - arrays → null - mixed text+JSON → null Integration (sdk/tests/live-kernel.test.ts, 3 tests): - positive: hn-monitor analyze-story with JSON-emitting stub → run completes with completionReason: success - negative: hn-monitor analyze-story with missing-fields stub → run completes with completionReason: step_failed - text-fallback: agent step with text-emitting stub in a schema-free spec → step.completed's output preserves the CliResult wrapper Full SDK suite output captured verbatim from `npx vitest run 2>&1 | grep -E 'Test Files|Tests\\s+[0-9]|Duration' | tail -3`: Test Files 17 passed (17) Tests 232 passed (232) Duration 47.33s (transform 2.20s, setup 0ms, collect 6.61s, tests 58.57s, environment 6ms, prepare 7.42s) FAIL-FIRST MUTATION EVIDENCE Mutation — in sdk/src/worker.ts, replace const output = parseJsonOutput(result.stdout_tail) ?? result; with const output = result; Command: `npx vitest run tests/live-kernel.test.ts -t "hn-monitor"` Captured output (verbatim; cargo header lines above `running` are elided per prior swarm feedback): ❯ tests/live-kernel.test.ts (12 tests | 1 failed | 10 skipped) 423ms × built flows CLI against live relayflowd > runs hn-monitor analyze-story end-to-end via a stub agent CLI (gate 2 clause 2 demo) 228ms → expected 'step_failed' to be 'success' // Object.is equality ⎯⎯⎯⎯⎯⎯⎯ Failed Tests 1 ⎯⎯⎯⎯⎯⎯⎯ FAIL tests/live-kernel.test.ts > built flows CLI against live relayflowd > runs hn-monitor analyze-story end-to-end via a stub agent CLI (gate 2 clause 2 demo) AssertionError: expected 'step_failed' to be 'success' // Object.is equality Expected: "success" Received: "step_failed" ❯ tests/live-kernel.test.ts:405:52 403| (entry) => (entry as { entry_type: string }).entry_type === 'run… 404| ) as { payload: { completionReason: string } } | undefined; 405| expect(runCompleted?.payload.completionReason).toBe('success'); | ^ 406| 407| await worker.close(); ⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯[1/1]⎯ Test Files 1 failed (1) Tests 1 failed | 1 passed | 10 skipped (12) Start at 19:28:08 Duration 1.76s (transform 203ms, setup 0ms, collect 287ms, tests 423ms, environment 0ms, prepare 129ms) Restore of sdk/src/worker.ts + `npm run build` + full-suite `npx vitest run`. Captured summary lines verbatim: Test Files 17 passed (17) Tests 232 passed (232) Duration 48.18s (transform 2.42s, setup 0ms, collect 6.83s, tests 59.94s, environment 15ms, prepare 7.38s) PRE-SWARM-CHECK RESULTS Ran `flows run workflows/preswarm-check.yaml` on this diff before push. M lens caught 4 issues locally (all fixed): - Comment claimed a `_process` attachment that didn't happen → sentence removed. - Tests didn't pin the invariant → added negative test AND text-fallback test. - `unknown | null` return type redundant → tightened to `Record | null`. - Duplicated helper comment → deduped. Also caught in later iters after the first swarm run: - Test didn't cover the non-JSON fallback path → added text-fallback test. - Negative test could pass on a never-completing run → strengthened to require `runCompleted` defined + pin `completionReason === 'step_failed'`. - `parseJsonOutput` accepted scalars/arrays → tightened to object-only, unit-tested. - Missing-fields stub docstring said "revert-to-wrapper would pass" — corrected to name the mutation class this stub actually catches (schema-gate-removed). NON-GOALS (documented in-code where relevant) - Wake context injection into the agent's prompt. - Real LLM CLI wiring (`claude -p` etc.). - Enforcing "CLIs signal errors via exit code, not error JSON". - Refactoring the two hn-monitor tests into a shared helper. Co-Authored-By: Claude Opus 4.7 --- sdk/src/worker.ts | 41 +++- sdk/tests/live-kernel.test.ts | 210 ++++++++++++++++++ sdk/tests/parse-json-output.test.ts | 57 +++++ .../analyze-story-missing-fields-cli | 13 ++ testdata/preflight/analyze-story-stub-cli | 10 + .../preflight/analyze-story-text-only-cli | 8 + 6 files changed, 338 insertions(+), 1 deletion(-) create mode 100644 sdk/tests/parse-json-output.test.ts create mode 100755 testdata/preflight/analyze-story-missing-fields-cli create mode 100755 testdata/preflight/analyze-story-stub-cli create mode 100755 testdata/preflight/analyze-story-text-only-cli diff --git a/sdk/src/worker.ts b/sdk/src/worker.ts index 64268cd5..a11d887b 100644 --- a/sdk/src/worker.ts +++ b/sdk/src/worker.ts @@ -93,6 +93,21 @@ export class AgentWorker extends EventEmitter { : { exit_code: null, stdout_tail: '', stderr_tail: 'agent step has no declared CLI' }; const completionReason = result.exit_code === 0 ? 'success' : 'worker_error'; + // Output shape: if the CLI's stdout parses as JSON, promote THAT + // as the step's `output` value so `json_schema` verification + // validates the analysis payload, not a wrapper around stdout. + // On the JSON path the CliResult (exit_code / stdout_tail / + // stderr_tail) is DISCARDED from `output` — the schema author + // wrote a shape for the analysis, not for the process wrapper. + // Non-JSON stdout falls back to the wrapper so text-emitting + // tools still round-trip usefully. + // + // Implicit contract: CLIs signal errors via non-zero exit, not by + // emitting an error JSON with exit 0. `completionReason` is + // derived from exit code, so a CLI that exits 0 while emitting + // `{"error":...}` will report success with an error payload. + const output = parseJsonOutput(result.stdout_tail) ?? result; + await this.client.stepComplete( dispatch.run_id, dispatch.step_id, @@ -100,7 +115,7 @@ export class AgentWorker extends EventEmitter { dispatch.idempotency_key, completionReason, { - output: result, + output, started_pins: dispatch.pins, end_pins: dispatch.pins, }, @@ -108,6 +123,30 @@ export class AgentWorker extends EventEmitter { } } +/** + * Return an object-shaped JSON payload parsed from `stdout`, or + * `null` when stdout is empty, non-JSON, or JSON-of-a-scalar/array. + * The object-only restriction matches how `json_schema` gates are + * authored — a scalar or array sneaking through would confuse both + * the schema and downstream readers who expect field lookups on + * `output`. Text-emitting tools and non-object JSON both fall + * through to the CliResult wrapper preserved by the caller. + */ +export function parseJsonOutput(stdout: string): Record | null { + const trimmed = stdout.trim(); + if (trimmed === '') return null; + let parsed: unknown; + try { + parsed = JSON.parse(trimmed); + } catch { + return null; + } + if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) { + return null; + } + return parsed as Record; +} + function runCli(cli: string, instruction: string): Promise { return new Promise((resolve) => { const child = spawn(cli, [instruction], { stdio: ['ignore', 'pipe', 'pipe'] }); diff --git a/sdk/tests/live-kernel.test.ts b/sdk/tests/live-kernel.test.ts index bf020bcb..6bd67c8f 100644 --- a/sdk/tests/live-kernel.test.ts +++ b/sdk/tests/live-kernel.test.ts @@ -343,6 +343,216 @@ steps: }); }); + it('runs hn-monitor analyze-story end-to-end via a stub agent CLI (gate 2 clause 2 demo)', async () => { + // RFC-0001 gate 2 second clause: the real proactive workload + // (hn-monitor) actually RUNS as a relayflow — not just dispatches + // and fails with worker_error because no CLI is wired. This test + // proves the pipeline works today with a stub CLI that satisfies + // the analyze-story step's json_schema verification. A real + // analyzer would replace the stub with `claude -p` or similar; + // that follow-up is orthogonal to whether the plumbing works. + // + // Ordering matters: worker MUST attach BEFORE submit_event, or + // the run parks with nothing to drive it (see the "late-attaching + // worker" test above for the recorded gotcha). + const dataDir = temporaryDirectory('flows-live-hn-agent-'); + await startDaemon(dataDir); + const client = await connectClient(dataDir); + await client.hello('live-hn-agent'); + const worker = new AgentWorker(client, { + workerId: 'live-hn-agent-worker', + pins: { + workspace: [{ surface: 'repo', revision_id: 'rev-a' }], + streams: [], + }, + }); + await worker.attach(); + + // Load hn-monitor's canonical spec, then patch the analyze-story + // step to declare a real CLI. Everything else — triggers, dedupe, + // wake context — comes straight from the existing gate-2 spec. + const spec = JSON.parse( + readFileSync(join(TESTDATA, 'hn-monitor.spec.canonical.json'), 'utf8'), + ) as { steps: { id: string; cli?: string }[] }; + for (const step of spec.steps) { + if (step.id === 'analyze-story') { + step.cli = join(TESTDATA, 'preflight', 'analyze-story-stub-cli'); + } + } + + // Submit the trigger event. The kernel matches it against the + // hn-story-posted subscription, spawns a run, and dispatches the + // agent step to our attached worker. Worker runs the stub CLI, + // which outputs JSON that satisfies the step's json_schema gate, + // reports stepComplete with success. + const outcome = await client.eventSubmit(spec, { + type: 'hn.story_posted', + payload: { id: 42_000_042, type: 'story' }, + }); + expect(outcome).toMatchObject({ matched: true, deduped: false }); + const runId = (outcome as { run: { run_id: string } }).run.run_id; + + // The step must reach 'done' — not 'failed', not 'runnable' — and + // the run must complete with a success-shaped final entry. + expect(await waitForStep(client, runId, 'analyze-story', 'done')).toMatchObject({ + type: 'agent', + state: 'done', + }); + const finalEntries = (await client.journalRead(runId)).entries; + const runCompleted = finalEntries.find( + (entry) => (entry as { entry_type: string }).entry_type === 'run.completed', + ) as { payload: { completionReason: string } } | undefined; + expect(runCompleted?.payload.completionReason).toBe('success'); + + await worker.close(); + }, 30_000); + + it('hn-monitor analyze-story FAILS verification when the CLI omits required schema fields', async () => { + // Negative pin for the "schema gate stays live" invariant. The + // stub emits `{"story_title":"partial"}` (missing + // relevance_score, reasoning). With json_schema active over the + // PROMOTED payload, the schema author's declared shape has + // required fields the stub omitted; verification must fail. + // + // NOTE on mutation coverage: this test does NOT catch a revert + // of the promotion (wrapper-as-output) — under that mutation + // the CliResult wrapper also fails the schema (it lacks + // story_title/relevance_score/reasoning entirely), so the run + // still ends in step_failed and this assertion still passes. + // That mutation IS caught by the positive test above, which + // requires the run to complete with 'success' — the wrapper + // path fails that. What THIS test catches is a + // schema-gate-removed mutation: if the kernel stopped running + // json_schema verification, the stub's exit-0 with any JSON + // would slide through to 'success', and this test would fail + // because the reason would BE 'success' not 'step_failed'. + const dataDir = temporaryDirectory('flows-live-hn-agent-neg-'); + await startDaemon(dataDir); + const client = await connectClient(dataDir); + await client.hello('live-hn-agent-neg'); + const worker = new AgentWorker(client, { + workerId: 'live-hn-agent-neg-worker', + pins: { + workspace: [{ surface: 'repo', revision_id: 'rev-a' }], + streams: [], + }, + }); + await worker.attach(); + + const spec = JSON.parse( + readFileSync(join(TESTDATA, 'hn-monitor.spec.canonical.json'), 'utf8'), + ) as { steps: { id: string; cli?: string }[] }; + for (const step of spec.steps) { + if (step.id === 'analyze-story') { + step.cli = join(TESTDATA, 'preflight', 'analyze-story-missing-fields-cli'); + } + } + + const outcome = await client.eventSubmit(spec, { + type: 'hn.story_posted', + payload: { id: 42_000_099, type: 'story' }, + }); + expect(outcome).toMatchObject({ matched: true, deduped: false }); + const runId = (outcome as { run: { run_id: string } }).run.run_id; + + // The step must NOT reach success. Poll for run.completed and + // assert TWO things: (a) the terminal entry actually arrived + // (a never-completing run is a test bug, not a pass — a + // `.not.toBe('success')` check on undefined would trivially + // pass), and (b) the reason is a declared FAILURE kind, not + // a "hasn't completed yet" absence. + const deadline = Date.now() + 10_000; + let runCompleted: { payload: { completionReason: string } } | undefined; + while (Date.now() < deadline) { + const entries = (await client.journalRead(runId)).entries; + runCompleted = entries.find( + (entry) => (entry as { entry_type: string }).entry_type === 'run.completed', + ) as { payload: { completionReason: string } } | undefined; + if (runCompleted !== undefined) break; + await delay(50); + } + expect( + runCompleted, + 'run.completed entry never arrived within 10s — test cannot assert schema-live under a hung run', + ).toBeDefined(); + // step_failed is the outer reason (a step failed → run failed); + // the inner step.completed record carries verification_failed + // for schema-rejected outputs. Pinning the outer reason avoids + // depending on retry/backoff behavior for this test. + expect(runCompleted!.payload.completionReason).toBe('step_failed'); + + await worker.close(); + }, 30_000); + + it('agent step preserves the CliResult wrapper as output when the CLI emits non-JSON text', async () => { + // Pins the promise in worker.ts's promotion comment: "Non-JSON + // stdout falls back to the wrapper so text-emitting tools still + // round-trip usefully." Without this test, a future refactor + // that deletes `?? result` or narrows parseJsonOutput's return + // shape could silently discard stdout/stderr/exit_code from + // `output` for every text-emitting agent CLI — the two hn-monitor + // tests above would stay green because their stubs emit pure JSON. + // + // The kernel nulls `output` on step.completed when verification + // fails (a design choice — the failed output is noise, not state + // to persist). To observe the wrapper's preservation, this test + // uses a spec WITHOUT `json_schema` verification: the step + // succeeds cleanly and the CliResult wrapper lands in the + // journal intact. + const dataDir = temporaryDirectory('flows-live-text-fallback-'); + await startDaemon(dataDir); + const cli = join(TESTDATA, 'preflight', 'analyze-story-text-only-cli'); + const client = await connectClient(dataDir); + await client.hello('live-text-fallback'); + const worker = new AgentWorker(client, { + workerId: 'live-text-fallback-worker', + pins: { + workspace: [{ surface: 'repo', revision_id: 'rev-a' }], + streams: [], + }, + }); + await worker.attach(); + + const started = await client.runStart(toKernelSpec(compileYaml(` +version: '0.1.0' +steps: + - id: analyze + type: agent + cli: ${JSON.stringify(cli)} + instruction: Emit some text. +`))); + + // Poll for step.completed and inspect its output. + const deadline = Date.now() + 10_000; + let stepCompleted: { payload: { output: unknown } } | undefined; + while (Date.now() < deadline) { + const entries = (await client.journalRead(started.run_id)).entries; + stepCompleted = entries.find( + (entry) => (entry as { entry_type: string; step_id?: string }).entry_type === 'step.completed' + && (entry as { step_id?: string }).step_id === 'analyze', + ) as { payload: { output: unknown } } | undefined; + if (stepCompleted !== undefined) break; + await delay(50); + } + expect(stepCompleted, 'step.completed never arrived').toBeDefined(); + const output = stepCompleted!.payload.output as { + exit_code: number | null; + stdout_tail: string; + stderr_tail: string; + }; + // Wrapper survived: text stdout preserved verbatim, exit_code + // preserved, stderr channel preserved. If a future refactor + // dropped `?? result` from worker.ts, `output` would be `null` + // here (parseJsonOutput returned null on non-JSON stdout) and + // these assertions would all fail. + expect(output).not.toBeNull(); + expect(output.exit_code).toBe(0); + expect(output.stdout_tail).toContain('looked at the story'); + expect(output.stderr_tail).toBe(''); + + await worker.close(); + }, 30_000); + it('preflights before journaling and names an unreachable socket', async () => { const dataDir = temporaryDirectory('flows-live-preflight-'); await startDaemon(dataDir); diff --git a/sdk/tests/parse-json-output.test.ts b/sdk/tests/parse-json-output.test.ts new file mode 100644 index 00000000..dbddea34 --- /dev/null +++ b/sdk/tests/parse-json-output.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from 'vitest'; +import { parseJsonOutput } from '../src/worker.js'; + +// Unit tests for the CLI-output promotion helper AgentWorker uses to +// decide whether stdout becomes the step's `output` value or the +// CliResult wrapper is preserved. The live-kernel integration tests +// exercise the E2E promotion path; these tests pin the helper's +// boundary cases so a future edit is justified against explicit +// examples rather than inferred from a live test. + +describe('parseJsonOutput', () => { + it('returns null for empty stdout — nothing to promote', () => { + expect(parseJsonOutput('')).toBeNull(); + expect(parseJsonOutput(' ')).toBeNull(); + expect(parseJsonOutput('\n\n')).toBeNull(); + }); + + it('returns null for non-JSON stdout — text-emitting tool falls back to wrapper', () => { + expect(parseJsonOutput('looked at the story, seemed fine')).toBeNull(); + expect(parseJsonOutput('progress: 50%\nprogress: 100%\ndone')).toBeNull(); + expect(parseJsonOutput('{"not-closed": ')).toBeNull(); + }); + + it('promotes an object payload', () => { + expect(parseJsonOutput('{"story_title":"x","relevance_score":5}')).toEqual({ + story_title: 'x', + relevance_score: 5, + }); + }); + + it('trims surrounding whitespace before parsing', () => { + expect(parseJsonOutput('\n\n {"ok": true} \n')).toEqual({ ok: true }); + }); + + it('rejects a bare scalar — schema authors expect field lookups', () => { + expect(parseJsonOutput('42')).toBeNull(); + expect(parseJsonOutput('"hello"')).toBeNull(); + expect(parseJsonOutput('true')).toBeNull(); + expect(parseJsonOutput('null')).toBeNull(); + }); + + it('rejects an array — schema authors expect an object shape', () => { + expect(parseJsonOutput('[1,2,3]')).toBeNull(); + expect(parseJsonOutput('[{"a":1}]')).toBeNull(); + }); + + it('does NOT try to extract JSON from mixed text+JSON output — chatty CLIs fall back to wrapper', () => { + // Real LLM CLIs (claude -p, gemini) sometimes emit progress + // text and end with a JSON blob. This helper deliberately does + // NOT do "find the last JSON in the stream" — that heuristic + // is a separate concern with different failure modes and would + // silently promote whatever looked JSON-shaped. Instead, mixed + // output stays as the CliResult wrapper; a schema author who + // needs JSON should point at a wrapper CLI that emits only JSON. + expect(parseJsonOutput('starting...\n{"result":42}')).toBeNull(); + }); +}); diff --git a/testdata/preflight/analyze-story-missing-fields-cli b/testdata/preflight/analyze-story-missing-fields-cli new file mode 100755 index 00000000..9b1349d1 --- /dev/null +++ b/testdata/preflight/analyze-story-missing-fields-cli @@ -0,0 +1,13 @@ +#!/bin/sh +# Stub agent-runtime CLI that emits JSON MISSING required schema +# fields (only story_title, no relevance_score/reasoning). Used by +# the negative gate-2-clause-2 test: pins that the json_schema gate +# is LIVE over the promoted payload — a mutation that removed the +# schema check would let this stub's exit-0 slide through to +# `success`, and the negative test would fail. (A separate mutation +# that reverted the wrapper promotion would ALSO make the run fail, +# but for a different reason — the wrapper's shape lacks every +# required schema field. The positive test above catches the +# wrapper-not-promoted mutation.) +set -eu +printf '%s' '{"story_title":"partial"}' diff --git a/testdata/preflight/analyze-story-stub-cli b/testdata/preflight/analyze-story-stub-cli new file mode 100755 index 00000000..723e6010 --- /dev/null +++ b/testdata/preflight/analyze-story-stub-cli @@ -0,0 +1,10 @@ +#!/bin/sh +# Stub agent-runtime CLI for the hn-monitor gate-2-clause-2 demo. +# Emits a deterministic JSON payload that satisfies the +# `analyze-story` step's `json_schema` verification gate. +# Not a real analyzer — its purpose is to prove the dispatch → +# agent-CLI → stepComplete pipeline works end-to-end. A real analyzer +# would read `$1` (the instruction) and — once wake_context injection +# lands — the triggering-event JSON, then invoke a real LLM. +set -eu +printf '%s' '{"story_title":"stub","relevance_score":5,"reasoning":"stub agent runtime — deterministic output for gate-2 clause-2 demo"}' diff --git a/testdata/preflight/analyze-story-text-only-cli b/testdata/preflight/analyze-story-text-only-cli new file mode 100755 index 00000000..d3da1422 --- /dev/null +++ b/testdata/preflight/analyze-story-text-only-cli @@ -0,0 +1,8 @@ +#!/bin/sh +# Text-emitting stub agent-runtime CLI (no JSON in stdout). Used by +# the text-fallback test: proves AgentWorker keeps the CliResult +# wrapper as `output` when stdout is not JSON, so text-emitting +# tools (progress bars, chatty CLIs) still round-trip usefully +# without the JSON promotion path silently discarding stdout/stderr. +set -eu +printf '%s' 'looked at the story, seemed fine to me'