feat(signals): register Centaur callbacks on centaur_review watch

server/__tests__/signal-processor.test.ts

@@ -380,6 +380,108 @@ describe('SignalProcessor', () => {
     });
   });
 
+  describe('centaur callback registration', () => {
+    it('registers callback with Centaur on watch for centaur_review', async () => {
+      const fetchSpy = vi
+        .spyOn(globalThis, 'fetch')
+        .mockResolvedValue(new Response(JSON.stringify({ ok: true }), { status: 200 }));
+
+      const goal = store.create({ title: 'Goal' });
+      const task = store.create({
+        title: 'Wait for Centaur review',
+        parentId: goal.id,
+        stageType: 'wait_for_signal',
+        gateConfig: { type: 'centaur_review', pr_url: 'https://github.com/org/repo/pull/1' },
+      });
+      store.update(task.id, { status: 'active' });
+
+      processor.watch(task.id, task.gateConfig!);
+
+      // Allow async registration to complete
+      await vi.waitFor(() => {
+        expect(fetchSpy).toHaveBeenCalledWith(
+          'http://localhost:8642/api/signals/register',
+          expect.objectContaining({
+            method: 'POST',
+            body: expect.stringContaining(task.id),
+          }),
+        );
+      });
+
+      fetchSpy.mockRestore();
+    });
+
+    it('deregisters callback on unwatch for centaur_review', async () => {
+      const fetchSpy = vi
+        .spyOn(globalThis, 'fetch')
+        .mockResolvedValue(new Response(JSON.stringify({ ok: true }), { status: 200 }));
+
+      const goal = store.create({ title: 'Goal' });
+      const task = store.create({
+        title: 'Wait for Centaur review',
+        parentId: goal.id,
+        stageType: 'wait_for_signal',
+        gateConfig: { type: 'centaur_review', pr_url: 'https://github.com/org/repo/pull/1' },
+      });
+      store.update(task.id, { status: 'active' });
+
+      processor.watch(task.id, task.gateConfig!);
+      processor.unwatch(task.id);
+
+      await vi.waitFor(() => {
+        expect(fetchSpy).toHaveBeenCalledWith(
+          `http://localhost:8642/api/signals/${task.id}`,
+          expect.objectContaining({ method: 'DELETE' }),
+        );
+      });
+
+      fetchSpy.mockRestore();
+    });
+
+    it('does not register callback for non-centaur gate types', () => {
+      const fetchSpy = vi.spyOn(globalThis, 'fetch');
+
+      const goal = store.create({ title: 'Goal' });
+      const task = store.create({
+        title: 'Wait for CI',
+        parentId: goal.id,
+        stageType: 'wait_for_signal',
+        gateConfig: { type: 'gh_ci', repo: 'org/repo', pr: 1 },
+      });
+      store.update(task.id, { status: 'active' });
+
+      processor.watch(task.id, task.gateConfig!);
+
+      // fetch should not have been called for registration (only polling uses execFile, not fetch)
+      const registerCalls = fetchSpy.mock.calls.filter(
+        (call) => typeof call[0] === 'string' && call[0].includes('/api/signals/'),
+      );
+      expect(registerCalls).toHaveLength(0);
+
+      fetchSpy.mockRestore();
+    });
+
+    it('handles Centaur being unavailable gracefully', () => {
+      vi.spyOn(globalThis, 'fetch').mockRejectedValue(new Error('ECONNREFUSED'));
+
+      const goal = store.create({ title: 'Goal' });
+      const task = store.create({
+        title: 'Wait for Centaur review',
+        parentId: goal.id,
+        stageType: 'wait_for_signal',
+        gateConfig: { type: 'centaur_review', pr_url: 'https://github.com/org/repo/pull/1' },
+      });
+      store.update(task.id, { status: 'active' });
+
+      // Should not throw — fire-and-forget registration
+      expect(() => processor.watch(task.id, task.gateConfig!)).not.toThrow();
+      // Polling interval should still be set
+      expect(processor.isWatching(task.id)).toBe(true);
+
+      vi.restoreAllMocks();
+    });
+  });
+
   it('stores failure artifacts in annotations on retry', () => {
     const goal = store.create({ title: 'Goal' });
     const agent = store.create({

server/index.ts

@@ -179,9 +179,14 @@ setTemplateStore(wfTemplateStore);
 // SignalProcessor + orchestrator have a circular dep: signal resolution triggers tick(),
 // tick() registers watches. Break the cycle with a late-bound callback.
 let orchestratorRef: TaskOrchestrator | null = null;
-const signalProc = new SignalProcessor(taskStore, () => {
-  orchestratorRef?.tick();
-});
+const signalProc = new SignalProcessor(
+  taskStore,
+  () => {
+    orchestratorRef?.tick();
+  },
+  process.env.CENTAUR_URL || 'http://localhost:8642',
+  process.env.MITZO_URL || `http://localhost:${PORT}`,
+);
 setSignalProcessor(signalProc);
 
 // --- Task Orchestrator ---

server/signal-processor.ts

@@ -32,12 +32,22 @@ const POLL_INTERVALS: Record<string, number> = {
  */
 export class SignalProcessor {
   private watches = new Map<string, WatchEntry>();
+  private pendingRegistrations = new Map<string, Promise<void>>();
   private store: TaskStore;
   private onSignalResolved: (taskId: string) => void;
-
-  constructor(store: TaskStore, onSignalResolved: (taskId: string) => void) {
+  private centaurBaseUrl: string;
+  private mitzoBaseUrl: string;
+
+  constructor(
+    store: TaskStore,
+    onSignalResolved: (taskId: string) => void,
+    centaurBaseUrl = 'http://localhost:8642',
+    mitzoBaseUrl = 'http://localhost:3100',
+  ) {
     this.store = store;
     this.onSignalResolved = onSignalResolved;
+    this.centaurBaseUrl = centaurBaseUrl;
+    this.mitzoBaseUrl = mitzoBaseUrl;
   }
 
   watch(taskId: string, gateConfig: GateConfig): void {
@@ -57,12 +67,28 @@ export class SignalProcessor {
     }
 
     this.watches.set(taskId, { taskId, gateConfig, intervalId });
+
+    // Register callback with Centaur for push-based resolution
+    if (gateConfig.type === 'centaur_review') {
+      const registration = this.registerCentaurCallback(
+        taskId,
+        gateConfig as GateConfig & { pr_url: string },
+      );
+      this.pendingRegistrations.set(taskId, registration);
+      registration.finally(() => this.pendingRegistrations.delete(taskId));
+    }
   }
 
   unwatch(taskId: string): void {
     const entry = this.watches.get(taskId);
     if (!entry) return;
     if (entry.intervalId) clearInterval(entry.intervalId);
+
+    // Deregister callback with Centaur
+    if (entry.gateConfig.type === 'centaur_review') {
+      this.deregisterCentaurCallback(taskId);
+    }
+
     this.watches.delete(taskId);
     log.info('unwatched task', { taskId });
   }
@@ -154,6 +180,51 @@ export class SignalProcessor {
     }
   }
 
+  /** Register a callback URL with Centaur so it pushes ReviewCompleted events. */
+  private async registerCentaurCallback(taskId: string, config: { pr_url: string }): Promise<void> {
+    const callbackUrl = `${this.mitzoBaseUrl}/api/tasks/${taskId}/signal`;
+    try {
+      const res = await fetch(`${this.centaurBaseUrl}/api/signals/register`, {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({
+          task_id: taskId,
+          pr_url: config.pr_url,
+          callback_url: callbackUrl,
+        }),
+      });
+      if (res.ok) {
+        log.info('registered centaur callback', { taskId, pr_url: config.pr_url });
+      } else {
+        log.warn('centaur callback registration failed', { taskId, status: res.status });
+      }
+    } catch (err) {
+      // Centaur might not be running — polling fallback covers this case
+      log.warn('centaur callback registration error', {
+        taskId,
+        error: err instanceof Error ? err.message : String(err),
+      });
+    }
+  }
+
+  /** Deregister a callback with Centaur. Best-effort. */
+  private async deregisterCentaurCallback(taskId: string): Promise<void> {
+    // Wait for any in-flight registration to finish before sending DELETE,
+    // otherwise DELETE arrives first and the registration creates a dangling entry.
+    const pending = this.pendingRegistrations.get(taskId);
+    if (pending) {
+      await pending.catch(() => {});
+    }
+    try {
+      await fetch(`${this.centaurBaseUrl}/api/signals/${taskId}`, {
+        method: 'DELETE',
+      });
+      log.info('deregistered centaur callback', { taskId });
+    } catch {
+      // Best-effort — Centaur might be down
+    }
+  }
+
   /** Reset the most recent agent_work sibling before this task to pending. */
   private resetPrecedingSibling(task: { id: string; parentId: string | null }): void {
     if (!task.parentId) return;