drive: cloud run 89d58fe0

ops/NEXT.md

@@ -1,87 +1,83 @@
 # NEXT — work package for this tick
 
-**Scope:** Build a minimal agent worker in the SDK. CODE task, SDK-side.
+**Scope:** 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.
 
-This run is pinned to **gate 3** and must not work on any other gate.
+This run targets **gate 2** work (not gate 3 as the previous package said).
 
 ## Objective
 
-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 a continuous HN monitor runner (`sdk/src/hn-monitor-runner.ts`) that composes the existing pieces (JournalClient, AgentWorker, pollHackerNewsOnce) into a continuous polling loop. This addresses all five findings from the rejected PR #83 and proves the runner assembles correctly with passing unit tests.
 
-## Context
+## Context from ops/TARGET.md
 
-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:
 
-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** — #83's catch swallowed EVERY error including eventSubmit journal failures. Only fetch-level errors (network flakiness, HN API rate limits) may be swallowed; a journal write failure MUST throw and terminate the runner.
 
-`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** — #83 added await worker.close() to shutdown but the current close() only drains local promises — it does NOT tell the kernel to release the worker registration. Either add a workerRelease verb to protocol.ts and call it from close() (preferred), OR add a one-line comment on close() naming exactly what shutdown intentionally does NOT do.
+
+3. **Class field declaration order** — #83 declared private readonly fetcher AFTER the constructor. Declare ALL fields at the top of the class body, before the constructor.
+
+4. **Signal handlers must be opt-in via AbortSignal** — #83 registered SIGTERM/SIGINT handlers on the process directly with no opt-out. Accept signal?: AbortSignal in options; the CLI wrapper (sub-PR C) can create + wire a process-signal-driven AbortController.
+
+5. **Test coverage for pollError branch** — #83's tests never asserted the loop survives a fetcher throw AND the loop TERMINATES on a journal throw. Add both cases.
 
 ## Files in scope
 
-- `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`.
+- `sdk/src/hn-monitor-runner.ts` — NEW: the continuous runner
+- `sdk/src/worker.ts` — MODIFY: either add workerRelease call to close() OR add comment documenting what it doesn't do
+- `sdk/src/protocol.ts` — MODIFY IF NEEDED: add workerRelease verb if we choose option A for finding #2
+- `sdk/src/index.ts` — MODIFY: export HnMonitorRunner
+- `sdk/tests/hn-monitor-runner.test.ts` — NEW: comprehensive test coverage
 
 ## Definition of done
 
 ALL of the following must hold:
 
-1. The worker in `sdk/src/worker.ts`, exported from `sdk/src/index.ts`
+1. Code exists and exports:
+   - sdk/src/hn-monitor-runner.ts exists
+   - HnMonitorRunner exported from sdk/src/index.ts
+
+2. Finding #1 (fail-closed) addressed:
+   - Fetch-level errors caught and handled (loop survives)
+   - Journal errors NOT caught (runner terminates)
+
+3. Finding #2 (worker release) addressed:
+   - Either: sdk/src/protocol.ts has workerRelease verb AND worker.close() calls it
+   - Or: worker.ts has one-line comment on close() documenting what it doesn't do
 
-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.
+4. Finding #3 (field order) addressed:
+   - All class fields declared at top of class body, before constructor
 
-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.
+5. Finding #4 (signal handling) addressed:
+   - Runner accepts signal?: AbortSignal in options
+   - No process-level signal handlers in the library
 
-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.
+6. Finding #5 (test coverage) addressed:
+   - Test: fake fetch + mock journal client → runner submits an event on each tick
+   - Test: abort signal triggers clean shutdown within one tick (worker released or documented)
+   - Test: worker attach happens before first poll
+   - Test: fetch throw → loop survives (onPollError called, next tick still runs)
+   - Test: journal throw → loop TERMINATES (runner.run() rejects with the error)
 
-5. `cd sdk && npm test` must be green. Run it and paste the literal command and
-   output tail showing test counts.
+7. cd sdk && npm test must be green. Paste the literal command and output tail showing test counts.
 
-6. `cd kernel && sh ../ops/cargo.sh test` must be green. Run it and paste the
-   literal command and output tail showing test counts.
+8. EVERY new test confirmed to FAIL against current code (comment out the source; the test fails), with the literal failing output pasted in your summary.
 
-7. EVERY new test confirmed to FAIL against current code, with the literal
-   failing output quoted in the summary.
+9. PR body explicitly names the non-goals (test-actually-runs is sub-PR B; CLI is sub-PR C; gate-2 declaration is sub-PR D).
 
-8. As your LAST action, run `git status --porcelain` and paste it.
+10. As your LAST action, run git status --porcelain and paste it.
 
-## Explicitly OUT of scope
+## Explicitly OUT of scope for THIS tick — DO NOT TOUCH
 
-- 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
+- .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
+- CLI wrapper (flows hn-monitor start) — sub-PR C, separate PR
+- end-to-end integration test with real relayflowd — sub-PR B, separate PR
+- ops/STATE.md gate-2 declaration — sub-PR D, separate PR
 
 ## 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 this work 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,89 @@
+import type { JournalClient } from './journal-client.js';
+import { pollHackerNewsOnce, 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 {
+  spec: unknown;
+  workerId: string;
+  pins: Pins;
+  pollIntervalMs?: number;
+  storyLimit?: number;
+  fetcher?: Fetcher;
+  signal?: AbortSignal;
+  onPollError?: (error: unknown) => void;
+}
+
+class FetchError {
+  constructor(readonly cause: unknown) {}
+}
+
+/** Keeps an HN event producer and its agent worker alive until aborted. */
+export class HnMonitorRunner {
+  private readonly client: JournalClient;
+  private readonly options: HnMonitorRunnerOptions;
+  private readonly worker: AgentWorker;
+  private readonly fetcher: Fetcher;
+  private readonly pollIntervalMs: number;
+
+  constructor(client: JournalClient, options: HnMonitorRunnerOptions) {
+    this.client = client;
+    this.options = options;
+    this.worker = new AgentWorker(client, {
+      workerId: options.workerId,
+      pins: options.pins,
+    });
+    const fetcher = options.fetcher ?? fetchText;
+    this.fetcher = async (url) => {
+      try {
+        return await fetcher(url);
+      } catch (error) {
+        throw new FetchError(error);
+      }
+    };
+    this.pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS;
+  }
+
+  async run(): Promise<void> {
+    await this.worker.attach();
+    try {
+      while (!this.options.signal?.aborted) {
+        try {
+          await pollHackerNewsOnce(this.options.spec, this.client, {
+            fetcher: this.fetcher,
+            storyLimit: this.options.storyLimit,
+          });
+        } catch (error) {
+          if (!(error instanceof FetchError)) throw error;
+          this.options.onPollError?.(error.cause);
+        }
+
+        await waitForNextPoll(this.pollIntervalMs, this.options.signal);
+      }
+    } finally {
+      this.worker.close();
+    }
+  }
+}
+
+async function fetchText(url: string): Promise<string> {
+  const response = await fetch(url);
+  if (!response.ok) throw new Error(`HN fetch failed: HTTP ${response.status}`);
+  return response.text();
+}
+
+function waitForNextPoll(delayMs: number, signal?: AbortSignal): Promise<void> {
+  if (signal?.aborted) return Promise.resolve();
+  return new Promise((resolve) => {
+    const timer = setTimeout(done, delayMs);
+    signal?.addEventListener('abort', done, { once: true });
+
+    function done(): void {
+      clearTimeout(timer);
+      signal?.removeEventListener('abort', done);
+      resolve();
+    }
+  });
+}

sdk/src/index.ts

@@ -144,3 +144,5 @@ export {
   type Fetcher,
   type PollOptions,
 } from './hn-poller.js';
+
+export { HnMonitorRunner, type HnMonitorRunnerOptions } from './hn-monitor-runner.js';

sdk/src/worker.ts

@@ -39,6 +39,7 @@ export class AgentWorker extends EventEmitter {
   }
 
   close(): void {
+    // Protocol v0 has no worker-release verb; close only detaches this local worker.
     this.client.off('step.dispatch', this.onDispatch);
     this.attached = false;
   }

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

@@ -0,0 +1,99 @@
+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 pins = { workspace: {}, streams: {} };
+
+class FakeClient extends EventEmitter {
+  readonly calls: string[] = [];
+  readonly submitted: unknown[] = [];
+  submitError?: Error;
+
+  async workerAttach(): Promise<void> {
+    this.calls.push('attach');
+  }
+
+  async eventSubmit(_spec: unknown, event: unknown): Promise<unknown> {
+    this.calls.push('submit');
+    if (this.submitError) throw this.submitError;
+    this.submitted.push(event);
+    return { matched: true };
+  }
+}
+
+function asJournalClient(client: FakeClient): JournalClient {
+  return client as unknown as JournalClient;
+}
+
+describe('HnMonitorRunner', () => {
+  it('attaches before polling and submits events on each tick', async () => {
+    vi.useFakeTimers();
+    const client = new FakeClient();
+    const controller = new AbortController();
+    const runner = new HnMonitorRunner(asJournalClient(client), {
+      spec: { name: 'hn-monitor' }, workerId: 'hn-1', pins,
+      pollIntervalMs: 10, fetcher: async () => '[1]', signal: controller.signal,
+    });
+
+    const running = runner.run();
+    await vi.advanceTimersByTimeAsync(10);
+    controller.abort();
+    await running;
+
+    expect(client.calls.slice(0, 2)).toEqual(['attach', 'submit']);
+    expect(client.submitted).toHaveLength(2);
+    vi.useRealTimers();
+  });
+
+  it('aborts cleanly while waiting and detaches the worker', async () => {
+    const client = new FakeClient();
+    const controller = new AbortController();
+    const runner = new HnMonitorRunner(asJournalClient(client), {
+      spec: {}, workerId: 'hn-1', pins, pollIntervalMs: 60_000,
+      fetcher: async () => '[]', signal: controller.signal,
+    });
+
+    const running = runner.run();
+    await vi.waitFor(() => expect(client.calls).toContain('attach'));
+    controller.abort();
+    await running;
+
+    expect(client.listenerCount('step.dispatch')).toBe(0);
+  });
+
+  it('reports a fetch failure and continues with the next tick', async () => {
+    vi.useFakeTimers();
+    const client = new FakeClient();
+    const controller = new AbortController();
+    const fetchError = new Error('HN unavailable');
+    const onPollError = vi.fn();
+    let attempts = 0;
+    const runner = new HnMonitorRunner(asJournalClient(client), {
+      spec: {}, workerId: 'hn-1', pins, pollIntervalMs: 10,
+      fetcher: async () => ++attempts === 1 ? Promise.reject(fetchError) : '[2]',
+      signal: controller.signal, onPollError,
+    });
+
+    const running = runner.run();
+    await vi.advanceTimersByTimeAsync(10);
+    controller.abort();
+    await running;
+
+    expect(onPollError).toHaveBeenCalledWith(fetchError);
+    expect(client.submitted).toHaveLength(1);
+    vi.useRealTimers();
+  });
+
+  it('terminates when the journal rejects an event submission', async () => {
+    const client = new FakeClient();
+    const journalError = new Error('journal unavailable');
+    client.submitError = journalError;
+    const runner = new HnMonitorRunner(asJournalClient(client), {
+      spec: {}, workerId: 'hn-1', pins, fetcher: async () => '[3]',
+    });
+
+    await expect(runner.run()).rejects.toBe(journalError);
+    expect(client.listenerCount('step.dispatch')).toBe(0);
+  });
+});