feat(hub-mcp): add wait_for_change long-poll tool (#283)

gordonwoodhull · web-flow · commit 697e9b924cf6 · 2026-06-13T11:53:44.000-05:00
Add a `wait_for_change` MCP tool that blocks until a watched file is
edited by any collaborator, then returns its new content — letting an
agent react to a live collaborator without busy-polling `read_file`.

- ConnectionManager.waitForChange() registers a one-shot waiter against
  the existing onFileChanged/onFileRemoved sync callbacks and resolves
  with the new payload, or times out (default 25s, clamped to 1-55s).
- `since_hash` closes the gap between polls: if the file already differs
  from the caller's last-known hash the change returns immediately, so an
  edit landing between two calls is never missed.
- Read-only tool; available even under --read-only.
- README: add a Tools section documenting every tool plus wait_for_change.

Tests: 4 new ConnectionManager unit tests (resolve-on-change, timeout,
gap-close via since_hash, other-path isolation); protocol tests updated to
assert the tool is registered (read-write + read-only lists). Full suite
163 passing. Verified live end-to-end against a running hub this session
(watch -&gt; collaborator edits -&gt; agent appends -&gt; re-arm).
diff --git a/ts-packages/quarto-hub-mcp/README.md b/ts-packages/quarto-hub-mcp/README.md
@@ -16,6 +16,61 @@ The design lives in
 operators registering a hub for end-user use should consult
 [`claude-notes/instructions/hub-mcp-operator-runbook.md`](../../claude-notes/instructions/hub-mcp-operator-runbook.md).
 
+## Tools
+
+Once configured, the agent has the tools below. Each operates on a project
+identified by its Automerge **index document ID**, passed as the `project`
+argument.
+
+### Reading
+
+| Tool | What it does |
+|------|--------------|
+| `connect_project` | Connect to a project by its index doc ID; returns the file list. Triggers the sign-in flow if the hub requires auth. |
+| `list_files` | List the files in a connected project. |
+| `read_file` | Read a text file's content. |
+| `wait_for_change` | **Long-poll**: block until a file is edited by any collaborator, then return its new content (see below). |
+
+### Writing
+
+| Tool | What it does |
+|------|--------------|
+| `write_file` | Replace a text file's entire content (creates it if absent). |
+| `patch_file` | Replace one unique substring in a text file. |
+| `create_file` | Create a new text file. |
+| `delete_file` | Delete a file. |
+| `rename_file` | Rename / move a file. |
+| `create_project` | Create a new project on the sync server with optional initial files. |
+
+The write tools are hidden when the server is started with `--read-only`.
+
+### Authentication
+
+`authenticate` and `authenticate_clear` are present only when the OAuth env
+vars are configured (see [Setup](#setup)). On a no-auth hub they are unused.
+
+### `wait_for_change` — reacting to a live collaborator
+
+`wait_for_change` lets an agent respond to another editor without busy-polling
+`read_file`. It blocks until the watched `path` changes, then returns
+`{ changed: true, content, hash }`; on timeout it returns
+`{ changed: false, hash }`, so the agent just calls again to keep watching.
+
+```jsonc
+wait_for_change({
+  project: "<index-doc-id>",
+  path: "report.qmd",
+  timeout_seconds: 25,    // optional, clamped to 1–55
+  since_hash: "sha256:…"  // optional — see below
+})
+```
+
+Pass the `hash` from the previous result back as `since_hash`: if the file
+already differs from it, the call returns **immediately**. This closes the gap
+between polls, so an edit that lands *between* two `wait_for_change` calls is
+never missed. The tool is read-only and is available even in `--read-only`
+mode.
+
 ## Setup
 
 You need two values from your hub operator:
diff --git a/ts-packages/quarto-hub-mcp/src/connection-manager.test.ts b/ts-packages/quarto-hub-mcp/src/connection-manager.test.ts
@@ -605,6 +605,103 @@ describe('lastObservedAuthMode state machine', () => {
   });
 });
 
+// ---------------------------------------------------------------------------
+// waitForChange (long-poll)
+// ---------------------------------------------------------------------------
+
+/**
+ * Sync-client factory that captures the callbacks object so a test can
+ * drive `onFileChanged` / `onFileRemoved` to simulate remote edits. The
+ * `connect` stub seeds one text file ('test.qmd' = 'hello') via
+ * `onFileAdded`, mirroring a real initial sync.
+ */
+function capturingSyncClientFactory() {
+  let captured: SyncClientCallbacks | undefined;
+  const factory = (cbs: SyncClientCallbacks): SyncClient => {
+    captured = cbs;
+    const stub: Partial<SyncClient> = {
+      connect: vi.fn(async () => {
+        cbs.onFileAdded?.('test.qmd', { type: 'text', text: 'hello' });
+        return [];
+      }) as unknown as SyncClient['connect'],
+      disconnect: vi.fn().mockResolvedValue(undefined) as unknown as SyncClient['disconnect'],
+    };
+    return stub as SyncClient;
+  };
+  return { factory, cbs: () => captured! };
+}
+
+describe('waitForChange (long-poll)', () => {
+  it('resolves when the watched file changes', async () => {
+    const fetchSpy = scriptedFetch([200]);
+    const cap = capturingSyncClientFactory();
+    const mgr = new ConnectionManager({
+      serverUrl: 'wss://hub.example.com/ws',
+      fetch: fetchSpy.fetch,
+      syncClientFactory: cap.factory,
+    });
+    await mgr.connect('idx-1');
+
+    const pending = mgr.waitForChange('idx-1', 'test.qmd', 5000);
+    // Simulate a remote edit arriving on the watched path.
+    cap.cbs().onFileChanged('test.qmd', 'hello world', []);
+
+    const res = await pending;
+    expect(res.changed).toBe(true);
+    expect(res.payload?.type).toBe('text');
+    expect((res.payload as { type: 'text'; text: string }).text).toBe('hello world');
+    expect(res.hash).toMatch(/^sha256:/);
+  });
+
+  it('times out with changed=false when nothing changes', async () => {
+    const fetchSpy = scriptedFetch([200]);
+    const cap = capturingSyncClientFactory();
+    const mgr = new ConnectionManager({
+      serverUrl: 'wss://hub.example.com/ws',
+      fetch: fetchSpy.fetch,
+      syncClientFactory: cap.factory,
+    });
+    await mgr.connect('idx-1');
+
+    const res = await mgr.waitForChange('idx-1', 'test.qmd', 20);
+    expect(res.changed).toBe(false);
+  });
+
+  it('returns immediately when since_hash differs from current (gap-close)', async () => {
+    const fetchSpy = scriptedFetch([200]);
+    const cap = capturingSyncClientFactory();
+    const mgr = new ConnectionManager({
+      serverUrl: 'wss://hub.example.com/ws',
+      fetch: fetchSpy.fetch,
+      syncClientFactory: cap.factory,
+    });
+    await mgr.connect('idx-1');
+
+    // No onFileChanged fired; current is the seeded 'hello'. A stale
+    // baseline hash means a change already happened in the gap → resolve now.
+    const res = await mgr.waitForChange('idx-1', 'test.qmd', 5000, 'sha256:stale');
+    expect(res.changed).toBe(true);
+    expect((res.payload as { type: 'text'; text: string }).text).toBe('hello');
+  });
+
+  it('ignores changes to other paths', async () => {
+    const fetchSpy = scriptedFetch([200]);
+    const cap = capturingSyncClientFactory();
+    const mgr = new ConnectionManager({
+      serverUrl: 'wss://hub.example.com/ws',
+      fetch: fetchSpy.fetch,
+      syncClientFactory: cap.factory,
+    });
+    await mgr.connect('idx-1');
+
+    const pending = mgr.waitForChange('idx-1', 'test.qmd', 40);
+    // A change to a different file must not satisfy the waiter.
+    cap.cbs().onFileChanged('other.qmd', 'nope', []);
+    const res = await pending;
+    expect(res.changed).toBe(false);
+  });
+});
+
 // ---------------------------------------------------------------------------
 // Redaction invariants
 // ---------------------------------------------------------------------------
diff --git a/ts-packages/quarto-hub-mcp/src/connection-manager.ts b/ts-packages/quarto-hub-mcp/src/connection-manager.ts
@@ -23,6 +23,8 @@
  * not require auth.
  */
 
+import { createHash } from 'node:crypto';
+
 import {
   createSyncClient,
   type DisconnectOptions,
@@ -84,9 +86,58 @@ export interface ConnectionManagerDeps {
   readonly probePath?: string;
 }
 
+/**
+ * A one-shot listener registered by {@link ConnectionManager.waitForChange}.
+ * Fired (and removed) the next time the watched `path` changes.
+ */
+interface ChangeWaiter {
+  path: string;
+  fire: (payload: FilePayload | null) => void;
+}
+
 interface ProjectState {
   client: SyncClient;
   files: Map<string, FilePayload>;
+  /** Pending long-poll waiters, keyed implicitly by their `path` field. */
+  waiters: Set<ChangeWaiter>;
+}
+
+/**
+ * Result of a {@link ConnectionManager.waitForChange} long-poll.
+ * `changed: false` means the call timed out with no edit observed.
+ * `payload: null` means the file was removed.
+ */
+export interface ChangeResult {
+  changed: boolean;
+  payload: FilePayload | null;
+  /** sha256 (`sha256:<hex>`) of the payload, or null when absent/removed. */
+  hash: string | null;
+}
+
+/** Content hash used for the long-poll gap-close check. */
+function hashPayload(p: FilePayload | undefined | null): string | null {
+  if (!p) return null;
+  const h = createHash('sha256');
+  if (p.type === 'text') h.update(p.text, 'utf8');
+  else h.update(Buffer.from(p.data));
+  return `sha256:${h.digest('hex')}`;
+}
+
+/**
+ * Fire (and remove) every waiter registered for `path`, handing it the
+ * new payload (`null` on removal). Other paths' waiters are untouched.
+ */
+function fireWaiters(
+  waiters: Set<ChangeWaiter>,
+  path: string,
+  payload: FilePayload | null,
+): void {
+  for (const w of [...waiters]) {
+    if (w.path === path) {
+      waiters.delete(w);
+      w.fire(payload);
+    }
+  }
 }
 
 // ---------------------------------------------------------------------------
@@ -180,18 +231,25 @@ export class ConnectionManager {
     const auth = await this.resolveAuthForConnect();
 
     const files = new Map<string, FilePayload>();
+    const waiters = new Set<ChangeWaiter>();
     const callbacks: SyncClientCallbacks = {
       onFileAdded(path: string, file: FilePayload) {
         files.set(path, file);
+        fireWaiters(waiters, path, file);
       },
       onFileChanged(path: string, text: string, _patches: Patch[]) {
-        files.set(path, { type: 'text', text });
+        const payload: FilePayload = { type: 'text', text };
+        files.set(path, payload);
+        fireWaiters(waiters, path, payload);
       },
       onBinaryChanged(path: string, data: Uint8Array, mimeType: string) {
-        files.set(path, { type: 'binary', data, mimeType });
+        const payload: FilePayload = { type: 'binary', data, mimeType };
+        files.set(path, payload);
+        fireWaiters(waiters, path, payload);
       },
       onFileRemoved(path: string) {
         files.delete(path);
+        fireWaiters(waiters, path, null);
       },
       onError(err: Error) {
         console.error(
@@ -215,11 +273,56 @@ export class ConnectionManager {
       peerTimeoutMs: PEER_TIMEOUT_MS,
     });
 
-    const state: ProjectState = { client, files };
+    const state: ProjectState = { client, files, waiters };
     this.projects.set(indexDocId, state);
     return state;
   }
 
+  /**
+   * Long-poll: resolve the next time `path` changes in the project, or
+   * after `timeoutMs` with `changed: false`. Connects first if needed.
+   *
+   * `sinceHash` closes the gap between polls: if the file already differs
+   * from the caller's last-known hash, the change is returned immediately
+   * (so an edit that lands between two `waitForChange` calls is never
+   * missed). Pass back the `hash` from the previous result each call.
+   */
+  async waitForChange(
+    indexDocId: string,
+    path: string,
+    timeoutMs: number,
+    sinceHash?: string,
+  ): Promise<ChangeResult> {
+    // Register the waiter synchronously when already connected so an edit
+    // arriving immediately after the call can't slip through the await gap.
+    // (First-time connects still pay one await; `sinceHash` covers that gap.)
+    const state = this.projects.get(indexDocId) ?? (await this.connect(indexDocId));
+
+    const current = state.files.get(path);
+    const currentHash = hashPayload(current);
+    // Gap-close: a change already happened relative to the caller's baseline.
+    if (sinceHash !== undefined && sinceHash !== currentHash) {
+      return { changed: true, payload: current ?? null, hash: currentHash };
+    }
+
+    return await new Promise<ChangeResult>((resolve) => {
+      let timer: ReturnType<typeof setTimeout>;
+      const waiter: ChangeWaiter = {
+        path,
+        fire: (payload) => {
+          clearTimeout(timer);
+          resolve({ changed: true, payload, hash: hashPayload(payload) });
+        },
+      };
+      state.waiters.add(waiter);
+      timer = setTimeout(() => {
+        state.waiters.delete(waiter);
+        const latest = state.files.get(path);
+        resolve({ changed: false, payload: latest ?? null, hash: hashPayload(latest) });
+      }, timeoutMs);
+    });
+  }
+
   /**
    * Create a new project. Currently this path always runs *after* the
    * agent has authenticated (or the hub is no-auth), so we run the
@@ -231,18 +334,25 @@ export class ConnectionManager {
     const auth = await this.resolveAuthForConnect();
 
     const tempFiles = new Map<string, FilePayload>();
+    const waiters = new Set<ChangeWaiter>();
     const callbacks: SyncClientCallbacks = {
       onFileAdded(path: string, file: FilePayload) {
         tempFiles.set(path, file);
+        fireWaiters(waiters, path, file);
       },
       onFileChanged(path: string, text: string, _patches: Patch[]) {
-        tempFiles.set(path, { type: 'text', text });
+        const payload: FilePayload = { type: 'text', text };
+        tempFiles.set(path, payload);
+        fireWaiters(waiters, path, payload);
       },
       onBinaryChanged(path: string, data: Uint8Array, mimeType: string) {
-        tempFiles.set(path, { type: 'binary', data, mimeType });
+        const payload: FilePayload = { type: 'binary', data, mimeType };
+        tempFiles.set(path, payload);
+        fireWaiters(waiters, path, payload);
       },
       onFileRemoved(path: string) {
         tempFiles.delete(path);
+        fireWaiters(waiters, path, null);
       },
     };
 
@@ -261,7 +371,7 @@ export class ConnectionManager {
       peerTimeoutMs: PEER_TIMEOUT_MS,
     });
 
-    const state: ProjectState = { client, files: tempFiles };
+    const state: ProjectState = { client, files: tempFiles, waiters };
     this.projects.set(result.indexDocId, state);
     return { indexDocId: result.indexDocId, files: result.files };
   }
diff --git a/ts-packages/quarto-hub-mcp/src/hub-mcp.test.ts b/ts-packages/quarto-hub-mcp/src/hub-mcp.test.ts
@@ -45,6 +45,7 @@ describe('MCP protocol', () => {
       'patch_file',
       'read_file',
       'rename_file',
+      'wait_for_change',
       'write_file',
     ]);
   });
@@ -99,6 +100,7 @@ describe('MCP protocol (read-only mode)', () => {
       'connect_project',
       'list_files',
       'read_file',
+      'wait_for_change',
     ]);
   });
 
diff --git a/ts-packages/quarto-hub-mcp/src/tools.ts b/ts-packages/quarto-hub-mcp/src/tools.ts
