drive: cloud run 88aaba35

ops/NEXT.md

@@ -1,80 +1,87 @@
-# Work package — gate 3: validate ops/NEXT.md as a checked artifact
+# NEXT — work package for this tick
 
-**Scope from this run's target:** Make ops/NEXT.md a checked artifact instead of free prose. CODE task, SDK-side.
+**Scope:** Build a minimal agent worker in the SDK. CODE task, SDK-side.
 
-## The problem, from evidence
+This run is pinned to **gate 3** and must not work on any other gate.
 
-Every run writes `ops/NEXT.md`. Reviewers have raised findings against it on FOUR separate PRs (#19, #35, #40, #48), always the same two shapes:
-
-  - it asserts a test result without carrying the command or its output ("all merged and tested", "three tests pass")
-  - it cites a file that is not in the delivered tree (`ops/TARGET.md`)
+## Objective
 
-Those are cheap findings that cost a review round trip each time, and they recur because nothing checks the file. It is prose, so anything can be written in it, including claims that are not true.
+Promote the throwaway worker the tests already build into a real SDK component
+that can execute agent steps by running their declared CLI as a subprocess.
 
-## Objective
+## Context
 
-An SDK function that validates a NEXT.md work package and refuses it with a typed reason, in the same style as `validateWorkPackage` in `sdk/src/backlog-picker.ts` — read that first and match its shape.
+Nothing in this repo can execute an agent step. Searching for `workerAttach` /
+`step.complete` finds only TESTS (`sdk/tests/live-kernel.test.ts`,
+`journal-client.test.ts`, `journal-client-loopback.ts`) and the protocol
+definitions. `sdk/src/cli/run.ts` only OBSERVES worker leases and waits for one
+that never arrives.
 
-At minimum it must catch the two observed shapes:
-  - a claim of passing tests with no captured command output near it
-  - a reference to a repo path that does not exist
+The kernel's dispatch, lease and claim machinery is real and tested. The worker
+side of the protocol is simply unimplemented, and that is what blocks gate 2
+("a workload RUNS as a relayflow" — today a run can only be shown CREATED) and
+gate 3 ("every claim/lease/retry served by the kernel").
 
-`sdk/src/work-package-consumer.ts` already takes an injected `pathExists` for exactly this kind of check — reuse that pattern rather than calling the filesystem directly, and note WHY: it is what makes the check testable.
+`sdk/tests/live-kernel.test.ts` around the `live-manual-agent` case (line 288)
+shows the whole shape: connect, `hello`, `workerAttach` with pins, receive
+`step.dispatch`, act, complete. The protocol is already proven there.
 
 ## Files in scope
 
-- `sdk/src/work-package-validator.ts` — new file, the validator function
-- `sdk/src/index.ts` — export the validator
-- `sdk/tests/work-package-validator.test.ts` — new file, comprehensive tests
-- `sdk/src/failure-kinds.ts` — add typed refusal reasons if needed
+- `sdk/src/worker.ts` — new file, the worker implementation
+- `sdk/src/index.ts` — export the worker
+- `sdk/tests/live-kernel.test.ts` OR a new test file — add a test that runs a
+  real flow with an agent step end to end against a live `relayflowd`, with
+  this worker attached, and asserts the step reaches `done`.
 
 ## Definition of done
 
 ALL of the following must hold:
 
-1. **The validator exists in sdk/src, exported from sdk/src/index.ts**
+1. The worker in `sdk/src/worker.ts`, exported from `sdk/src/index.ts`
 
-2. **Typed refusal reasons, not booleans and not thrown strings**
+2. A test that runs a real flow with an agent step end to end against a live
+   `relayflowd`, with this worker attached, and asserts the step reaches
+   `done`. `sdk/tests/live-kernel.test.ts` already starts a daemon — follow
+   that pattern.
 
-3. **Run it against the ops/NEXT.md files from PRs #19 and #35 — both must be REFUSED, and quote the reasons.** If it accepts them it has not caught the real defect.
+3. **The worker must attach BEFORE the run starts.** A run that finds no worker
+   parks, and attaching afterwards does not re-drive it — `run.resume` is what
+   picks a parked run back up. That contract is pinned in the live-kernel
+   suite; do not fight it.
 
-4. **A well-formed NEXT.md must still be ACCEPTED.** Include one in the tests.
+4. The worker must:
+   - attach for `agent` steps with the pins it holds
+   - on `step.dispatch`, run the step's declared `cli` as a subprocess
+   - report the result back through the existing protocol (`step.complete`, and
+     the failure path when the CLI exits nonzero)
+   - nothing speculative: no retries of its own, no scheduling, no LLM calls.
+     The kernel owns retry and lease policy — do not reimplement it.
 
-5. **`cd sdk && npm test` green:**
-   ```
-   cd sdk && npm test
-   ```
-   All tests must pass. Paste the literal command and output showing pass/fail counts.
+5. `cd sdk && npm test` must be green. Run it and paste the literal command and
+   output tail showing test counts.
 
-6. **`cd kernel && sh ../ops/cargo.sh test` green:**
-   ```
-   cd kernel && sh ../ops/cargo.sh test
-   ```
-   All tests must pass. Paste the literal command and output tail.
+6. `cd kernel && sh ../ops/cargo.sh test` must be green. Run it and paste the
+   literal command and output tail showing test counts.
 
-7. **The picker must not regress.** Measure against MAIN ON THE SAME BACKLOG:
-   ```
-   node -e 'const fs=require("node:fs");
-     const sdk=require("./sdk/dist/backlog-picker.js");
-     const t=fs.readFileSync("ops/BACKLOG.md","utf8");
-     const e=[...t.matchAll(/^- \*\*(.+?)\*\*\s*(.*(?:\n  .*)*)/gm)]
-       .map(m=>({title:m[1],body:m[2].replace(/\s+/g," ").trim()}));
-     let ok=0; for(const x of e)
-       if(sdk.validateWorkPackage(sdk.packageFromEntry(x)).accepted) ok++;
-     console.log("TOTAL="+e.length+" ACTIONABLE="+ok)'
-   ```
-   Baseline on current code: `TOTAL=31 ACTIONABLE=19`
-   After changes: must still show `ACTIONABLE=19` or higher.
+7. EVERY new test confirmed to FAIL against current code, with the literal
+   failing output quoted in the summary.
 
-8. **EVERY new test confirmed to FAIL against current code, with the literal failing output quoted in your summary**
-
-9. **As your LAST action, run `git status --porcelain` and paste it**
+8. As your LAST action, run `git status --porcelain` and paste it.
 
 ## Explicitly OUT of scope
 
+- LLM steps — not in the gate 3 scope
+- Retry logic in the worker — the kernel owns retry policy
+- Scheduling or lease management — the kernel owns lease policy
+- Optimizations, abstractions, or speculative features
+- Changes to the kernel
+- Changes to existing tests (except adding new test cases)
 - Work on any gate other than gate 3
-- Changing the format of ops/NEXT.md beyond validation
-- Refactoring existing validators beyond what's needed for consistency
-- Performance optimization
-- Validating BACKLOG.md entries
-- Any work in kernel/ beyond running the test suite
+
+## If blocked
+
+If gate 3 is genuinely unreachable from the current state, write
+ops/NEEDS_HUMAN.md saying exactly why and still end with ASSESS_DONE. Do not
+silently substitute different work: a run that reports progress on the wrong
+gate is worse than one that reports it is blocked.

sdk/src/index.ts

@@ -112,6 +112,7 @@ export type {
 export { JOURNAL_WRITE_FAILED, PROTOCOL_VERSION } from './protocol.js';
 
 export { JournalClient, type JournalClientOptions } from './journal-client.js';
+export { AgentWorker, type AgentWorkerOptions } from './worker.js';
 
 export {
   validateWorkPackage,

sdk/src/worker.ts

@@ -0,0 +1,91 @@
+import { spawn } from 'node:child_process';
+import { EventEmitter } from 'node:events';
+import type { JournalClient } from './journal-client.js';
+import type { Pins, StepDispatchEvent } from './protocol.js';
+import type { KernelAgentStep } from './spec.js';
+
+export interface AgentWorkerOptions {
+  workerId: string;
+  pins: Pins;
+}
+
+interface CliResult {
+  exit_code: number | null;
+  stdout_tail: string;
+  stderr_tail: string;
+}
+
+/** Executes dispatched agent steps using their declared CLI. */
+export class AgentWorker extends EventEmitter {
+  private attached = false;
+
+  constructor(
+    private readonly client: JournalClient,
+    private readonly options: AgentWorkerOptions,
+  ) {
+    super();
+  }
+
+  async attach(): Promise<void> {
+    if (this.attached) throw new Error('agent worker: already attached');
+    this.client.on('step.dispatch', this.onDispatch);
+    try {
+      await this.client.workerAttach(this.options.workerId, ['agent'], this.options.pins);
+      this.attached = true;
+    } catch (error) {
+      this.client.off('step.dispatch', this.onDispatch);
+      throw error;
+    }
+  }
+
+  close(): void {
+    this.client.off('step.dispatch', this.onDispatch);
+    this.attached = false;
+  }
+
+  private readonly onDispatch = (dispatch: StepDispatchEvent): void => {
+    if (dispatch.step_type !== 'agent') return;
+    void this.execute(dispatch).catch((error: unknown) => this.emit('error', error));
+  };
+
+  private async execute(dispatch: StepDispatchEvent): Promise<void> {
+    const spec = dispatch.spec as Partial<KernelAgentStep>;
+    const result = typeof spec.cli === 'string' && typeof spec.instruction === 'string'
+      ? await runCli(spec.cli, spec.instruction)
+      : { exit_code: null, stdout_tail: '', stderr_tail: 'agent step has no declared CLI' };
+    const completionReason = result.exit_code === 0 ? 'success' : 'worker_error';
+
+    await this.client.stepComplete(
+      dispatch.run_id,
+      dispatch.step_id,
+      dispatch.attempt,
+      dispatch.idempotency_key,
+      completionReason,
+      {
+        output: result,
+        started_pins: dispatch.pins,
+        end_pins: dispatch.pins,
+      },
+    );
+  }
+}
+
+function runCli(cli: string, instruction: string): Promise<CliResult> {
+  return new Promise((resolve) => {
+    const child = spawn(cli, [instruction], { stdio: ['ignore', 'pipe', 'pipe'] });
+    const stdout: Buffer[] = [];
+    const stderr: Buffer[] = [];
+    child.stdout.on('data', (chunk: Buffer) => stdout.push(chunk));
+    child.stderr.on('data', (chunk: Buffer) => stderr.push(chunk));
+    child.once('error', (error) => resolve({
+      exit_code: null,
+      stdout_tail: Buffer.concat(stdout).toString('utf8'),
+      stderr_tail: error.message,
+    }));
+    child.once('close', (code) => resolve({
+      exit_code: code,
+      stdout_tail: Buffer.concat(stdout).toString('utf8'),
+      stderr_tail: Buffer.concat(stderr).toString('utf8'),
+    }));
+  });
+}

sdk/tests/live-kernel.test.ts

@@ -1,5 +1,6 @@
 import {
   accessSync,
+  chmodSync,
   constants,
   existsSync,
   lstatSync,
@@ -17,6 +18,7 @@ import { afterEach, beforeAll, describe, expect, it } from 'vitest';
 import { compileYaml, toKernelSpec } from '../src/compile.js';
 import { JournalClient } from '../src/journal-client.js';
 import type { StepDispatchEvent } from '../src/protocol.js';
+import { AgentWorker } from '../src/worker.js';
 
 const ROOT = join(dirname(fileURLToPath(import.meta.url)), '..', '..');
 const SDK = join(ROOT, 'sdk');
@@ -201,6 +203,41 @@ steps:
     expect(completed.stderr).not.toContain('protocol_error');
   });
 
+  it('runs an agent CLI end to end through the SDK worker', async () => {
+    const directory = temporaryDirectory('flows-live-agent-worker-');
+    const dataDir = join(directory, 'data');
+    const cli = join(directory, 'agent-cli');
+    writeFileSync(cli, '#!/bin/sh\nprintf \'handled: %s\' "$1"\n');
+    chmodSync(cli, 0o755);
+    await startDaemon(dataDir);
+
+    const client = await connectClient(dataDir);
+    await client.hello('live-sdk-agent-worker');
+    const worker = new AgentWorker(client, {
+      workerId: 'live-sdk-agent-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: execute
+    type: agent
+    cli: ${JSON.stringify(cli)}
+    instruction: Perform the declared work.
+`)));
+
+    expect(await waitForStep(client, started.run_id, 'execute', 'done')).toMatchObject({
+      type: 'agent',
+      state: 'done',
+    });
+    worker.close();
+  });
+
   it('can always get a parked run to a late-attaching worker', async () => {
     // The contract that cost the most time to establish, so it is pinned here.
     //
@@ -598,6 +635,21 @@ async function waitForActiveRun(dataDir: string, marker: string): Promise<string
   throw new Error(`run did not reach the marked in-flight step within 5000ms: ${marker}`);
 }
 
+async function waitForStep(
+  client: JournalClient,
+  runId: string,
+  stepId: string,
+  state: string,
+): Promise<unknown> {
+  const deadline = Date.now() + 5_000;
+  while (Date.now() < deadline) {
+    const step = (await client.runGet(runId)).steps[stepId];
+    if (step?.state === state) return step;
+    await delay(20);
+  }
+  throw new Error(`step ${stepId} did not reach ${state} within 5000ms`);
+}
+
 function runArtifacts(dataDir: string): string[] {
   return readdirSync(dataDir).filter((name) => name !== 'relayflowd.sock').sort();
 }