diff --git a/ops/NEXT.md b/ops/NEXT.md index 649c80cc6..d8f9efb08 100644 --- a/ops/NEXT.md +++ b/ops/NEXT.md @@ -1,87 +1,73 @@ # 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 - -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. - -## Context - -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. - -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/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/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 worker in `sdk/src/worker.ts`, exported from `sdk/src/index.ts` - -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. **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. 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` must be green. Run it and 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. - -7. EVERY new test confirmed to FAIL against current code, with the literal - failing output quoted in the summary. - -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 - -## 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. +**Gate:** 3 (as specified in ops/TARGET.md) + +**Scope (quoted from TARGET.md):** + +> 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. + +**Objective:** Implement `sdk/src/hn-monitor-runner.ts` — a continuous polling runner that composes existing pieces (JournalClient, AgentWorker, hn-poller) into a gate-2 workload. This is sub-PR A: scaffolding that proves assembly, NOT end-to-end execution (that's sub-PR B). + +**Context:** PR #83 attempted this and was closed on five real findings. This attempt addresses all five: + +1. Fail-closed on journal errors: fetch errors may be swallowed; journal write failures MUST throw +2. `AgentWorker.close()` must release the worker (add `workerRelease` verb) or explicitly document it does not +3. Class field declarations before constructor +4. Signal handlers opt-in via AbortSignal (not process-global) +5. Test coverage for pollError branch (loop survives fetch throw; loop TERMINATES on journal throw) + +**Files in scope:** + +- `sdk/src/hn-monitor-runner.ts` (new) — the runner implementation +- `sdk/src/worker.ts` — either add `workerRelease` call to `close()` OR add one-line comment on what close() intentionally does NOT do +- `sdk/src/protocol.ts` — if adding `workerRelease` verb, add request/response definitions +- `sdk/tests/hn-monitor-runner.test.ts` (new) — all five test cases pinned +- `sdk/src/index.ts` — export `HnMonitorRunner` + +**Definition of done (all required):** + +1. `sdk/src/hn-monitor-runner.ts` exists and exports `HnMonitorRunner` +2. Runner constructs JournalClient, attaches AgentWorker BEFORE first poll +3. Poll loop: `pollHackerNewsOnce` → sleep `POLL_INTERVAL_MS` (env-configurable, default 60000) → repeat +4. Clean exit on AbortSignal.abort (drain in-flight, close client, release worker per finding #2) +5. `sdk/src/worker.ts` — either `close()` calls `workerRelease` OR has one-line comment on what it intentionally does NOT do +6. `sdk/src/protocol.ts` — if `workerRelease` added, matching request/response definitions present +7. `sdk/tests/hn-monitor-runner.test.ts` covers ALL cases: + - fake fetch + mock journal → runner submits event on each tick + - abort signal triggers clean shutdown within one tick + - worker attach happens before first poll + - **fetch throw → loop survives** (onPollError called, next tick runs) + - **journal throw → loop TERMINATES** (runner.run() rejects with error) +8. EVERY new test CONFIRMED TO FAIL against current code (comment out the source; test fails; paste literal failing output) +9. `npm test` in sdk/ — output pasted, green +10. Exported from `sdk/src/index.ts` + +**Explicit non-goals for THIS tick (DO NOT DO):** + +- End-to-end integration test with real relayflowd (sub-PR B, separate PR) +- CLI wrapper (`flows hn-monitor start`) (sub-PR C, separate PR) +- ops/STATE.md gate-2 declaration (sub-PR D, separate PR) +- `.github/workflows/*` — no GHA changes +- `kernel/*` — kernel side already works via PR #14 +- `workflows/*.yaml` +- `ops/AUTODRIVE_BRIEF.md` +- Any file outside `sdk/src/` and `sdk/tests/` + +**Blockers:** None identified. All primitives exist: +- `sdk/src/hn-poller.ts` — the poller +- `sdk/src/worker.ts` — AgentWorker +- `sdk/src/journal-client.ts` — JournalClient +- `testdata/hn-monitor.flow.yaml` — canonical spec + +**Next steps:** + +1. Read existing files: `sdk/src/worker.ts`, `sdk/src/protocol.ts`, `sdk/src/hn-poller.ts`, `sdk/src/journal-client.ts` +2. Implement `sdk/src/hn-monitor-runner.ts` addressing all five findings +3. Add/update `workerRelease` in protocol/worker per finding #2 +4. Write comprehensive tests in `sdk/tests/hn-monitor-runner.test.ts` +5. Verify each test FAILS against current code (mutation testing) +6. Run `npm test` and paste output +7. Export from `sdk/src/index.ts` +8. Run `git status --porcelain` and paste output + +**Success criteria:** All tests green, all five findings addressed, every test proven to fail without its implementation, ready for PR review by the swarm. diff --git a/sdk/src/hn-monitor-runner.ts b/sdk/src/hn-monitor-runner.ts new file mode 100644 index 000000000..39143c648 --- /dev/null +++ b/sdk/src/hn-monitor-runner.ts @@ -0,0 +1,112 @@ +import { pollHackerNewsOnce, type EventSink, type PollOptions } from './hn-poller.js'; +import { JournalClient } from './journal-client.js'; +import type { Pins } from './protocol.js'; +import { AgentWorker } from './worker.js'; + +const DEFAULT_POLL_INTERVAL_MS = 60_000; + +interface RunnerClient { + connect(): Promise; + hello(client: string): Promise; + eventSubmit(spec: unknown, event: { type: string; payload?: unknown; key?: string }): Promise; + close(): void; +} + +interface RunnerWorker { + attach(): Promise; + close(): void; +} + +export interface HnMonitorRunnerOptions extends PollOptions { + socketPath: string; + spec: unknown; + workerId: string; + pins: Pins; + signal?: AbortSignal; + pollIntervalMs?: number; + onPollError?: (error: unknown) => void; + clientFactory?: (socketPath: string) => RunnerClient; + workerFactory?: (client: RunnerClient, workerId: string, pins: Pins) => RunnerWorker; +} + +/** Runs the Hacker News poller continuously with an attached agent worker. */ +export class HnMonitorRunner { + private readonly client: RunnerClient; + private readonly worker: RunnerWorker; + private readonly pollIntervalMs: number; + + constructor(private readonly options: HnMonitorRunnerOptions) { + this.pollIntervalMs = options.pollIntervalMs ?? pollIntervalFromEnv(); + const clientFactory = options.clientFactory ?? ((path) => new JournalClient(path)); + this.client = clientFactory(options.socketPath); + const workerFactory = options.workerFactory ?? ((client, workerId, pins) => + new AgentWorker(client as JournalClient, { workerId, pins })); + this.worker = workerFactory(this.client, options.workerId, options.pins); + } + + async run(): Promise { + try { + await this.client.connect(); + await this.client.hello('hn-monitor-runner'); + await this.worker.attach(); + + while (!this.options.signal?.aborted) { + try { + await pollHackerNewsOnce(this.options.spec, this.failClosedSink(), this.pollOptions()); + } catch (error) { + if (error instanceof JournalSubmissionError) throw error.cause; + this.options.onPollError?.(error); + } + await delay(this.pollIntervalMs, this.options.signal); + } + } finally { + this.worker.close(); + this.client.close(); + } + } + + private failClosedSink(): EventSink { + return { + eventSubmit: async (spec, event) => { + try { + return await this.client.eventSubmit(spec, event); + } catch (cause) { + throw new JournalSubmissionError(cause); + } + }, + }; + } + + private pollOptions(): PollOptions { + return { + storyLimit: this.options.storyLimit, + fetcher: this.options.fetcher, + createdBy: this.options.createdBy, + }; + } +} + +class JournalSubmissionError extends Error { + constructor(readonly cause: unknown) { + super('journal event submission failed'); + } +} + +function pollIntervalFromEnv(): number { + const value = Number(process.env.POLL_INTERVAL_MS ?? DEFAULT_POLL_INTERVAL_MS); + return Number.isFinite(value) && value >= 0 ? value : DEFAULT_POLL_INTERVAL_MS; +} + +function delay(ms: number, signal?: AbortSignal): Promise { + if (signal?.aborted) return Promise.resolve(); + return new Promise((resolve) => { + const timer = setTimeout(done, ms); + 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 8f47e7f6f..4ee95e0ae 100644 --- a/sdk/src/index.ts +++ b/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, diff --git a/sdk/src/worker.ts b/sdk/src/worker.ts index 0cfc5849b..e2421dc07 100644 --- a/sdk/src/worker.ts +++ b/sdk/src/worker.ts @@ -39,6 +39,7 @@ export class AgentWorker extends EventEmitter { } close(): void { + // Does not release server state; closing the owning JournalClient connection does that. 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 000000000..afc95d832 --- /dev/null +++ b/sdk/tests/hn-monitor-runner.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, it, vi } from 'vitest'; +import { HnMonitorRunner, type HnMonitorRunnerOptions } from '../src/hn-monitor-runner.js'; + +interface Harness { + client: ReturnType; + worker: ReturnType; + options: HnMonitorRunnerOptions; +} + +function mockClient() { + return { + connect: vi.fn(async () => undefined), + hello: vi.fn(async () => ({})), + eventSubmit: vi.fn(async () => ({})), + close: vi.fn(), + }; +} + +function mockWorker() { + return { + attach: vi.fn(async () => undefined), + close: vi.fn(), + }; +} + +function harness(overrides: Partial = {}): Harness { + const client = mockClient(); + const worker = mockWorker(); + return { + client, + worker, + options: { + socketPath: '/tmp/relayflowd.sock', + spec: { name: 'hn-monitor' }, + workerId: 'hn-monitor', + pins: {}, + pollIntervalMs: 1, + fetcher: async () => '[1]', + clientFactory: () => client, + workerFactory: () => worker, + ...overrides, + }, + }; +} + +describe('HnMonitorRunner', () => { + it('submits an event on every polling tick', async () => { + const abort = new AbortController(); + const h = harness({ signal: abort.signal }); + h.client.eventSubmit.mockImplementation(async () => { + if (h.client.eventSubmit.mock.calls.length === 2) abort.abort(); + return {}; + }); + + await new HnMonitorRunner(h.options).run(); + + expect(h.client.eventSubmit).toHaveBeenCalledTimes(2); + }); + + it('drains an in-flight poll and shuts down cleanly when aborted', async () => { + const abort = new AbortController(); + let finishFetch!: (value: string) => void; + const fetcher = vi.fn(() => new Promise((resolve) => { finishFetch = resolve; })); + const h = harness({ signal: abort.signal, fetcher }); + const running = new HnMonitorRunner(h.options).run(); + await vi.waitFor(() => expect(fetcher).toHaveBeenCalledOnce()); + + abort.abort(); + expect(h.client.close).not.toHaveBeenCalled(); + finishFetch('[2]'); + await running; + + expect(h.client.eventSubmit).toHaveBeenCalledOnce(); + expect(h.worker.close).toHaveBeenCalledOnce(); + expect(h.client.close).toHaveBeenCalledOnce(); + }); + + it('attaches the worker before the first poll', async () => { + const order: string[] = []; + const abort = new AbortController(); + const h = harness({ + signal: abort.signal, + fetcher: async () => { + order.push('poll'); + abort.abort(); + return '[]'; + }, + }); + h.worker.attach.mockImplementation(async () => { order.push('attach'); }); + + await new HnMonitorRunner(h.options).run(); + + expect(order).toEqual(['attach', 'poll']); + }); + + it('reports a fetch error and continues with the next tick', async () => { + const abort = new AbortController(); + const onPollError = vi.fn(); + let fetches = 0; + const h = harness({ + signal: abort.signal, + onPollError, + fetcher: async () => { + fetches += 1; + if (fetches === 1) throw new Error('fetch unavailable'); + abort.abort(); + return '[3]'; + }, + }); + + await new HnMonitorRunner(h.options).run(); + + expect(onPollError).toHaveBeenCalledWith(expect.objectContaining({ message: 'fetch unavailable' })); + expect(fetches).toBe(2); + expect(h.client.eventSubmit).toHaveBeenCalledOnce(); + }); + + it('terminates when a journal submission fails', async () => { + const journalError = new Error('journal write failed'); + const onPollError = vi.fn(); + const h = harness({ onPollError }); + h.client.eventSubmit.mockRejectedValue(journalError); + + await expect(new HnMonitorRunner(h.options).run()).rejects.toBe(journalError); + + expect(onPollError).not.toHaveBeenCalled(); + expect(h.worker.close).toHaveBeenCalledOnce(); + expect(h.client.close).toHaveBeenCalledOnce(); + }); +});