feat(sdk): AgentWorker promotes CLI JSON output for json_schema verification

sdk/src/worker.ts

@@ -93,21 +93,60 @@ 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,
       dispatch.attempt,
       dispatch.idempotency_key,
       completionReason,
       {
-        output: result,
+        output,
         started_pins: dispatch.pins,
         end_pins: dispatch.pins,
       },
     );
   }
 }
 
+/**
+ * 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<string, unknown> | 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<string, unknown>;
+}
+
 function runCli(cli: string, instruction: string): Promise<CliResult> {
   return new Promise((resolve) => {
     const child = spawn(cli, [instruction], { stdio: ['ignore', 'pipe', 'pipe'] });

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);

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();
+  });
+});

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"}'

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"}'

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'