feat(sdk): dir-watcher poller — non-provider 2nd workload for gate 2 (Track E redo)

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<unknown>;
+}
+
+/** Injected so I/O stays deterministic in tests. */
+export interface DirLister {
+  (dir: string): Promise<Array<{ name: string; size: number; mtimeMs: number; isFile: boolean }>>;
+}
+
+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<string>;
+  /**
+   * 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<unknown[]> {
+  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;
+}

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';

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<string>();
+    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<string>();
+    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<string>();
+    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<string>();
+
+    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<string>();
+
+    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);
+  });
+});

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