drive: cloud run d7634d6a

ops/NEXT.md

@@ -1,87 +1,77 @@
 # NEXT — work package for this tick
 
-**Scope:** Build a minimal agent worker in the SDK. CODE task, SDK-side.
-
 This run is pinned to **gate 3** and must not work on any other gate.
 
-## Objective
+## Scope
 
-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.
+Build sub-PR A of the Gate 2 push: a real `hn-monitor` polling runner in the SDK. CODE task, `sdk/src/`-side. This is a scaffolding PR — proof that the workload EXECUTES end-to-end is deliberately deferred to sub-PR B (integration test). Do not conflate the two.
 
-## Context
+RFC-0001 §3 gate 2 is done when "hn-monitor runs as a relayflow in production, triggered by its real events, with zero bespoke persistence." Every primitive already exists in this repo — event triggers (PR #14, `kernel/relayflowd/tests/event_wake.rs`), the flow spec (`testdata/hn-monitor.flow.yaml`), the poller (`sdk/src/hn-poller.ts`), the agent worker (`sdk/src/worker.ts` from PR #53), a one-shot demo (`sdk/src/demo-hn-monitor.ts`) — but nothing has ever run them together as a continuous workload. This PR fixes that.
 
-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.
+Prior attempt (PR #83, closed) produced a functional runner but was rejected by the swarm on five real findings. Address them in this attempt:
 
-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").
+1. **Fail-closed on journal errors.** Only fetch-level errors (network flakiness, HN API rate limits) may be swallowed; a journal write failure MUST throw and terminate the runner. Split: `try { fetch } catch { onFetchError }` around the network call, `try { eventSubmit } catch { rethrow }` around the journal call.
 
-`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.
+2. **AgentWorker.close() must release the worker (or explicitly document it does not).** Either add a `workerRelease` verb to `sdk/src/protocol.ts` and call it from `close()` (preferred), OR add a one-line comment on `close()` naming exactly what shutdown intentionally does NOT do.
 
-## Files in scope
+3. **Class field declaration order.** Declare ALL fields at the top of the class body, before the constructor.
 
-- `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`.
+4. **Signal handlers must be opt-in via AbortSignal.** Accept `signal?: AbortSignal` in options; the CLI wrapper (sub-PR C) can create + wire a process-signal-driven AbortController.
 
-## Definition of done
+5. **Test coverage for pollError branch.** Assert the loop survives a fetcher throw AND the loop TERMINATES on a journal throw.
 
-ALL of the following must hold:
+## Objective
 
-1. The worker in `sdk/src/worker.ts`, exported from `sdk/src/index.ts`
+Add `sdk/src/hn-monitor-runner.ts` — a continuous polling runner that composes existing pieces (JournalClient, AgentWorker, pollHackerNewsOnce) into a workload that runs continuously, handles failures correctly (fail-closed on journal errors, survives fetch errors), and shuts down cleanly via AbortSignal.
 
-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.
+## Files in scope
 
-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.
+- `sdk/src/hn-monitor-runner.ts` (NEW)
+- `sdk/src/index.ts` (MODIFY: export HnMonitorRunner)
+- `sdk/src/worker.ts` (MODIFY: either add workerRelease call in close(), or document what it doesn't do)
+- `sdk/src/protocol.ts` (MODIFY: add workerRelease verb IF we choose to implement it)
+- `sdk/tests/hn-monitor-runner.test.ts` (NEW)
 
-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.
+## Definition of done
 
-5. `cd sdk && npm test` must be green. Run it and paste the literal command and
-   output tail showing test counts.
+ALL of the following must hold:
 
-6. `cd kernel && sh ../ops/cargo.sh test` must be green. Run it and paste the
-   literal command and output tail showing test counts.
+1. `sdk/src/hn-monitor-runner.ts` exists and exports `HnMonitorRunner` class
+2. `sdk/src/index.ts` exports HnMonitorRunner
+3. `sdk/src/worker.ts` — either `close()` calls `workerRelease` (add to protocol.ts if missing), OR a one-line comment names what close() intentionally does NOT do
+4. `sdk/src/protocol.ts` — if workerRelease was added, matching request/response definitions
+5. `sdk/tests/hn-monitor-runner.test.ts` covers ALL of these test cases:
+   - fake fetch + mock journal client → runner submits an event on each tick
+   - abort signal triggers clean shutdown within one tick (worker released or documented)
+   - worker attach happens before first poll
+   - **fetch throw → loop survives** (onPollError called, next tick still runs)
+   - **journal throw → loop TERMINATES** (runner.run() rejects with the error)
+6. All tests confirmed to FAIL against missing/broken source — for each new test, comment out the corresponding source code and verify the test fails, paste the literal failing output
+7. Test suite passes with literal output:
+   ```
+   cd sdk && npm test
+   ```
+   Paste the command and output showing test counts.
+8. Field declaration order correct — all class fields declared at top of class body, before constructor
+9. AbortSignal for shutdown — signal is opt-in via options parameter, not process-level SIGTERM/SIGINT handlers
+10. Fail-closed journal errors — journal errors rethrow and terminate the runner; only fetch errors are swallowed
+11. As your LAST action, run `git status --porcelain` and paste it
 
-7. EVERY new test confirmed to FAIL against current code, with the literal
-   failing output quoted in the summary.
+## Explicitly OUT of scope
 
-8. As your LAST action, run `git status --porcelain` and paste it.
+- Proving the workload actually executes end-to-end (dispatch → step complete). That is sub-PR B (integration test with real relayflowd + fake HN fetch + assert step reaches `done`). This PR ONLY proves the runner assembles and its unit tests hold.
+- CLI wrapper (`flows hn-monitor start`). That is sub-PR C.
+- ops/STATE.md gate-2 GREEN declaration. That is sub-PR D.
+- `.github/workflows/*` — no GHA changes
+- `kernel/*` — the kernel side of gate 2 already works via PR #14
+- `workflows/*.yaml` — those are for later sub-PRs
+- `ops/AUTODRIVE_BRIEF.md` — chief owns this file, not the drive loop
+- end-to-end integration test with real relayflowd — sub-PR B, separate PR
 
-## Explicitly OUT of scope
+## Gate
 
-- 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
+Gate 2 (proactive agent workload)
 
 ## 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.
+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.

sdk/src/hn-monitor-runner.ts

@@ -0,0 +1,96 @@
+import type { JournalClient } from './journal-client.js';
+import {
+  pollHackerNewsOnce,
+  type EventSink,
+  type Fetcher,
+} from './hn-poller.js';
+import type { Pins } from './protocol.js';
+import { AgentWorker } from './worker.js';
+
+const DEFAULT_POLL_INTERVAL_MS = 30_000;
+
+export interface HnMonitorRunnerOptions {
+  workerId: string;
+  pins: Pins;
+  intervalMs?: number;
+  storyLimit?: number;
+  fetcher?: Fetcher;
+  signal?: AbortSignal;
+  onPollError?: (error: unknown) => void;
+}
+
+/** Continuously feeds Hacker News events to an attached agent worker. */
+export class HnMonitorRunner {
+  private readonly client: JournalClient;
+  private readonly spec: unknown;
+  private readonly options: HnMonitorRunnerOptions;
+  private readonly worker: AgentWorker;
+  private readonly sink: EventSink;
+  private readonly intervalMs: number;
+
+  constructor(
+    client: JournalClient,
+    spec: unknown,
+    options: HnMonitorRunnerOptions,
+  ) {
+    this.client = client;
+    this.spec = spec;
+    this.options = options;
+    this.worker = new AgentWorker(client, {
+      workerId: options.workerId,
+      pins: options.pins,
+    });
+    this.sink = {
+      eventSubmit: async (spec, event) => {
+        try {
+          return await this.client.eventSubmit(spec, event);
+        } catch (cause) {
+          throw new JournalSubmissionError(cause);
+        }
+      },
+    };
+    this.intervalMs = options.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
+  }
+
+  async run(): Promise<void> {
+    await this.worker.attach();
+    try {
+      while (this.options.signal?.aborted !== true) {
+        try {
+          await pollHackerNewsOnce(this.spec, this.sink, {
+            storyLimit: this.options.storyLimit,
+            fetcher: this.options.fetcher,
+          });
+        } catch (error) {
+          if (error instanceof JournalSubmissionError) throw error.cause;
+          this.options.onPollError?.(error);
+        }
+        await delay(this.intervalMs, this.options.signal);
+      }
+    } finally {
+      this.worker.close();
+    }
+  }
+}
+
+class JournalSubmissionError extends Error {
+  constructor(readonly cause: unknown) {
+    super('journal event submission failed');
+  }
+}
+
+function delay(ms: number, signal?: AbortSignal): Promise<void> {
+  return new Promise((resolve) => {
+    const finish = (): void => {
+      signal?.removeEventListener('abort', abort);
+      resolve();
+    };
+    const timer = setTimeout(finish, ms);
+    const abort = (): void => {
+      clearTimeout(timer);
+      finish();
+    };
+    if (signal?.aborted === true) abort();
+    else signal?.addEventListener('abort', abort, { once: true });
+  });
+}

sdk/src/index.ts

@@ -113,6 +113,7 @@ 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 { HnMonitorRunner, type HnMonitorRunnerOptions } from './hn-monitor-runner.js';
 
 export {
   validateWorkPackage,

sdk/src/worker.ts

@@ -39,6 +39,7 @@ export class AgentWorker extends EventEmitter {
   }
 
   close(): void {
+    // Shutdown intentionally does not release the kernel's worker registration; protocol v0 has no release verb.
     this.client.off('step.dispatch', this.onDispatch);
     this.attached = false;
   }

sdk/tests/hn-monitor-runner.test.ts

@@ -0,0 +1,114 @@
+import { EventEmitter } from 'node:events';
+import { describe, expect, it, vi } from 'vitest';
+import { HnMonitorRunner } from '../src/hn-monitor-runner.js';
+import type { JournalClient } from '../src/journal-client.js';
+
+const SPEC = { subscriptions: [{ event: 'hn.story_posted' }] };
+const PINS = { workspace: [{ surface: 'repo', revision_id: 'rev-a' }] };
+
+class FakeJournalClient extends EventEmitter {
+  readonly calls: string[] = [];
+  eventError: Error | undefined;
+
+  async workerAttach(): Promise<{ worker_id: string }> {
+    this.calls.push('attach');
+    return { worker_id: 'hn-monitor' };
+  }
+
+  async eventSubmit(): Promise<{ matched: boolean; deduped: boolean }> {
+    this.calls.push('submit');
+    if (this.eventError !== undefined) throw this.eventError;
+    return { matched: true, deduped: false };
+  }
+}
+
+function runner(
+  client: FakeJournalClient,
+  options: Partial<ConstructorParameters<typeof HnMonitorRunner>[2]> = {},
+): HnMonitorRunner {
+  return new HnMonitorRunner(client as unknown as JournalClient, SPEC, {
+    workerId: 'hn-monitor',
+    pins: PINS,
+    intervalMs: 1,
+    ...options,
+  });
+}
+
+describe('HnMonitorRunner', () => {
+  it('submits an event on each tick', async () => {
+    const client = new FakeJournalClient();
+    const controller = new AbortController();
+    let polls = 0;
+    await runner(client, {
+      signal: controller.signal,
+      fetcher: async () => {
+        polls += 1;
+        if (polls === 2) controller.abort();
+        return `[${polls}]`;
+      },
+    }).run();
+
+    expect(client.calls.filter((call) => call === 'submit')).toHaveLength(2);
+  });
+
+  it('shuts down cleanly when aborted during the tick delay', async () => {
+    const client = new FakeJournalClient();
+    const controller = new AbortController();
+    const monitor = runner(client, {
+      signal: controller.signal,
+      intervalMs: 10_000,
+      fetcher: async () => '[1]',
+    });
+
+    const running = monitor.run();
+    await vi.waitFor(() => expect(client.calls).toContain('submit'));
+    controller.abort();
+    await expect(running).resolves.toBeUndefined();
+    expect(client.listenerCount('step.dispatch')).toBe(0);
+  });
+
+  it('attaches the worker before the first poll', async () => {
+    const client = new FakeJournalClient();
+    const controller = new AbortController();
+    await runner(client, {
+      signal: controller.signal,
+      fetcher: async () => {
+        client.calls.push('fetch');
+        controller.abort();
+        return '[]';
+      },
+    }).run();
+
+    expect(client.calls.slice(0, 2)).toEqual(['attach', 'fetch']);
+  });
+
+  it('reports a fetch error and polls again', async () => {
+    const client = new FakeJournalClient();
+    const controller = new AbortController();
+    const onPollError = vi.fn();
+    let polls = 0;
+    await runner(client, {
+      signal: controller.signal,
+      onPollError,
+      fetcher: async () => {
+        polls += 1;
+        if (polls === 1) throw new Error('temporary HN failure');
+        controller.abort();
+        return '[2]';
+      },
+    }).run();
+
+    expect(onPollError).toHaveBeenCalledOnce();
+    expect(polls).toBe(2);
+    expect(client.calls).toContain('submit');
+  });
+
+  it('terminates when the journal rejects an event', async () => {
+    const client = new FakeJournalClient();
+    const journalError = new Error('journal write failed');
+    client.eventError = journalError;
+
+    await expect(runner(client, { fetcher: async () => '[1]' }).run()).rejects.toBe(journalError);
+    expect(client.listenerCount('step.dispatch')).toBe(0);
+  });
+});