diff --git a/sdk/src/dir-watcher-poller.ts b/sdk/src/dir-watcher-poller.ts new file mode 100644 index 000000000..f69c8c28c --- /dev/null +++ b/sdk/src/dir-watcher-poller.ts @@ -0,0 +1,122 @@ +/** + * Directory watcher -> relayflow events. + * + * Second proactive workload on gate 2 primitives (hn-monitor is the first). + * Deliberately non-provider: no HTTP, no API tokens, no gate-6 dependency — + * just a directory poll. This proves the runner pattern generalizes beyond + * `hn-poller` without regressing RFC-0001 §6 (which assigns providers to + * relayfile adapters, not SDK code). + * + * How it works: each poll lists the target directory, dedupes against a + * caller-supplied `seen` set (or an internal Map if none provided), and + * submits a `dir.file_appeared` event for each unseen entry through the + * journal protocol. The kernel then dispatches the flow's agent step for + * each new file. + * + * Deduplication is still ultimately the kernel's job (flow's + * `dedupeKeyTemplate` + the (flow, subscription, key) claim). This layer's + * `seen` set is a cheap pre-filter so we don't spam `event.submit` with the + * same paths on every poll — an optimization, not a correctness contract. + * + * Real-world analog: an "inbox" directory that a human or another system + * drops files into, triggering a per-file flow (summarize, ingest, route, + * whatever the step declares). + */ + +import { promises as fsp } from 'node:fs'; +import { join } from 'node:path'; + +/** Anything that can submit an event through the journal protocol. */ +export interface EventSink { + eventSubmit(spec: unknown, event: { type: string; payload?: unknown; key?: string }): Promise; +} + +/** Injected so I/O stays deterministic in tests. */ +export interface DirLister { + (dir: string): Promise>; +} + +const defaultLister: DirLister = async (dir) => { + const entries = await fsp.readdir(dir, { withFileTypes: true }); + const out: Array<{ name: string; size: number; mtimeMs: number; isFile: boolean }> = []; + for (const ent of entries) { + if (!ent.isFile()) continue; + const full = join(dir, ent.name); + const stat = await fsp.stat(full); + out.push({ + name: ent.name, + size: stat.size, + mtimeMs: stat.mtimeMs, + isFile: true, + }); + } + return out; +}; + +export interface PollOptions { + /** Directory to watch. Required. */ + dir: string; + /** + * Set of relative paths already seen. The poller mutates it, adding each + * new file it submits. Callers persist this across polls to avoid + * re-submitting; internal callers can pass a fresh Set each poll if + * they'd rather rely on the kernel's dedupe claim. + */ + seen: Set; + /** + * Lister override — tests inject a deterministic fake. Production uses + * fs.readdir. + */ + lister?: DirLister; + /** + * Cap on files per poll (safety valve against dropping thousands into + * the directory at once). Default 100. + */ + fileLimit?: number; +} + +const DEFAULT_FILE_LIMIT = 100; + +/** + * List the directory once and submit a `dir.file_appeared` event for each + * unseen file. Adds each submitted path to `seen`. + * + * Returns the submit outcomes (one per new file). Journal errors from + * `eventSubmit` propagate; empty result is not an error; a missing + * directory throws (the caller decides whether that's a fetch error or + * a real failure — the runner classifies). + */ +export async function pollDirectoryOnce( + spec: unknown, + sink: EventSink, + options: PollOptions, +): Promise { + const lister = options.lister ?? defaultLister; + const fileLimit = options.fileLimit ?? DEFAULT_FILE_LIMIT; + const seen = options.seen; + + const entries = await lister(options.dir); + const fresh = entries + .filter((e) => e.isFile && !seen.has(e.name)) + .slice(0, fileLimit); + + const outcomes: unknown[] = []; + for (const entry of fresh) { + outcomes.push( + await sink.eventSubmit(spec, { + type: 'dir.file_appeared', + payload: { + type: 'file', + path: entry.name, + size: entry.size, + mtime_ms: entry.mtimeMs, + }, + }), + ); + // Only add to `seen` AFTER a successful submit — a journal failure + // means the event didn't reach the kernel, so the next poll should + // retry submission. + seen.add(entry.name); + } + return outcomes; +} diff --git a/sdk/src/index.ts b/sdk/src/index.ts index 59875f542..8f47e7f6f 100644 --- a/sdk/src/index.ts +++ b/sdk/src/index.ts @@ -144,3 +144,13 @@ export { type Fetcher, type PollOptions, } from './hn-poller.js'; + +// Directory watcher — second proactive workload for gate 2 primitives. +// Non-provider: no HTTP, no API tokens, no gate-6 dependency. Proves the +// pattern generalizes without regressing RFC-0001 §6 (providers = relayfile +// adapters, not SDK code). +export { + pollDirectoryOnce, + type DirLister, + type PollOptions as DirWatcherPollOptions, +} from './dir-watcher-poller.js'; diff --git a/sdk/tests/dir-watcher-poller.test.ts b/sdk/tests/dir-watcher-poller.test.ts new file mode 100644 index 000000000..334f24840 --- /dev/null +++ b/sdk/tests/dir-watcher-poller.test.ts @@ -0,0 +1,124 @@ +import { describe, expect, it } from 'vitest'; +import { pollDirectoryOnce, type DirLister } from '../src/dir-watcher-poller.js'; + +function recordingSink() { + const submitted: Array<{ spec: unknown; event: { type: string; payload?: unknown } }> = []; + return { + submitted, + async eventSubmit(spec: unknown, event: { type: string; payload?: unknown }) { + submitted.push({ spec, event }); + return { matched: true, deduped: false }; + }, + }; +} + +const listerFor = (files: Array<{ name: string; size?: number; mtimeMs?: number }>): DirLister => + async () => files.map((f) => ({ + name: f.name, + size: f.size ?? 100, + mtimeMs: f.mtimeMs ?? Date.now(), + isFile: true, + })); + +describe('dir-watcher poller', () => { + it('submits one event per NEW file and adds it to the seen set', async () => { + const sink = recordingSink(); + const seen = new Set(); + const spec = { name: 'dir-watcher' }; + + await pollDirectoryOnce(spec, sink, { + dir: '/tmp/watch', + seen, + lister: listerFor([{ name: 'a.txt' }, { name: 'b.log' }]), + }); + + expect(sink.submitted).toHaveLength(2); + expect(sink.submitted[0].event.type).toBe('dir.file_appeared'); + expect((sink.submitted[0].event.payload as any).path).toBe('a.txt'); + expect((sink.submitted[0].event.payload as any).type).toBe('file'); + expect(seen.has('a.txt')).toBe(true); + expect(seen.has('b.log')).toBe(true); + }); + + it('does NOT re-submit files already in the seen set', async () => { + const sink = recordingSink(); + const seen = new Set(['a.txt']); + + await pollDirectoryOnce({}, sink, { + dir: '/tmp/watch', + seen, + lister: listerFor([{ name: 'a.txt' }, { name: 'b.log' }]), + }); + + // Only b.log is new. + expect(sink.submitted).toHaveLength(1); + expect((sink.submitted[0].event.payload as any).path).toBe('b.log'); + }); + + it('does NOT add a file to `seen` if its eventSubmit throws (retry on next poll)', async () => { + const seen = new Set(); + let attempts = 0; + const sink = { + async eventSubmit() { + attempts++; + throw new Error('journal client: connection closed'); + }, + }; + + await expect(pollDirectoryOnce({}, sink, { + dir: '/tmp/watch', + seen, + lister: listerFor([{ name: 'a.txt' }]), + })).rejects.toThrow(/journal client/); + + expect(attempts).toBe(1); + // The file did NOT enter seen — the next poll must retry. + expect(seen.has('a.txt')).toBe(false); + }); + + it('respects the fileLimit cap', async () => { + const sink = recordingSink(); + const seen = new Set(); + const files = Array.from({ length: 25 }, (_, i) => ({ name: `f${i}.txt` })); + + await pollDirectoryOnce({}, sink, { + dir: '/tmp/watch', + seen, + lister: listerFor(files), + fileLimit: 10, + }); + + expect(sink.submitted).toHaveLength(10); + // Only the first 10 got submitted; the remaining 15 are still un-seen. + expect(seen.size).toBe(10); + }); + + it('propagates a lister error (missing directory, permission denied)', async () => { + const sink = recordingSink(); + const seen = new Set(); + + await expect(pollDirectoryOnce({}, sink, { + dir: '/nonexistent', + seen, + lister: async () => { throw new Error('ENOENT: no such directory'); }, + })).rejects.toThrow(/ENOENT/); + + expect(sink.submitted).toHaveLength(0); + }); + + it('carries file metadata (size + mtime) in the event payload', async () => { + const sink = recordingSink(); + const seen = new Set(); + + await pollDirectoryOnce({}, sink, { + dir: '/tmp/watch', + seen, + lister: listerFor([{ name: 'a.txt', size: 4096, mtimeMs: 1717000000000 }]), + }); + + const payload = sink.submitted[0].event.payload as any; + expect(payload.path).toBe('a.txt'); + expect(payload.size).toBe(4096); + expect(payload.mtime_ms).toBe(1717000000000); + }); +}); diff --git a/testdata/dir-watcher.flow.yaml b/testdata/dir-watcher.flow.yaml new file mode 100644 index 000000000..2f4c7dcdd --- /dev/null +++ b/testdata/dir-watcher.flow.yaml @@ -0,0 +1,38 @@ +version: '0.1.0' +name: dir-watcher +description: >- + Fire an agent step for each new file that appears in a watched directory. + Second proactive workload on gate 2 primitives (hn-monitor is the first) — + proves the pattern generalizes to a NON-provider input. No external network, + no API tokens, no gate-6 dependency. +triggers: + - id: file-appeared + executor: agent-worker + eventType: dir.file_appeared + pattern: + type: file + dedupeKeyTemplate: '{{event.type}}:{{payload.path}}' +steps: + - id: describe-file + type: agent + instruction: >- + A new file was reported in the watched directory. Read the wake context + (which includes the file's path and size), then output a JSON summary + naming: the file path, a one-line description of what the filename + suggests, and whether the file appears to require follow-up action. + recoveryMode: reset + verification: + type: json_schema + schema: + type: object + required: + - path + - description + - needs_followup + properties: + path: + type: string + description: + type: string + needs_followup: + type: boolean