diff --git a/ops/NEXT.md b/ops/NEXT.md index 649c80cc..0bac37d5 100644 --- a/ops/NEXT.md +++ b/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. diff --git a/sdk/src/hn-monitor-runner.ts b/sdk/src/hn-monitor-runner.ts new file mode 100644 index 00000000..9bfa99d2 --- /dev/null +++ b/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 { + 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 { + 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 { + 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(); + } + }); +} diff --git a/sdk/src/index.ts b/sdk/src/index.ts index 59875f54..cbdae06b 100644 --- a/sdk/src/index.ts +++ b/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'; diff --git a/sdk/src/worker.ts b/sdk/src/worker.ts index 0cfc5849..266f09ad 100644 --- a/sdk/src/worker.ts +++ b/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; } diff --git a/sdk/tests/hn-monitor-runner.test.ts b/sdk/tests/hn-monitor-runner.test.ts new file mode 100644 index 00000000..b15910e5 --- /dev/null +++ b/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 { + this.calls.push('attach'); + } + + async eventSubmit(_spec: unknown, event: unknown): Promise { + 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); + }); +});