From 8349c6f211340039cf9e542ecae9da64820dca8b Mon Sep 17 00:00:00 2001 From: dimakis Date: Sat, 4 Jul 2026 10:56:57 +0100 Subject: [PATCH] feat(signals): register Centaur callbacks on centaur_review watch SignalProcessor registers a callback URL with Centaur's signal bridge when watching a centaur_review gate, enabling push-based resolution alongside the existing 30s polling fallback. Deregisters on unwatch. Gracefully handles Centaur being unavailable (fire-and-forget). Includes race condition fix for pending registrations and env var wiring for CENTAUR_URL and MITZO_URL. Co-Authored-By: Claude Opus 4.6 --- server/__tests__/signal-processor.test.ts | 102 ++++++++++++++++++++++ server/index.ts | 11 ++- server/signal-processor.ts | 75 +++++++++++++++- 3 files changed, 183 insertions(+), 5 deletions(-) diff --git a/server/__tests__/signal-processor.test.ts b/server/__tests__/signal-processor.test.ts index b910788e..20054025 100644 --- a/server/__tests__/signal-processor.test.ts +++ b/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({ diff --git a/server/index.ts b/server/index.ts index 8cb4640d..f075d098 100644 --- a/server/index.ts +++ b/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 --- diff --git a/server/signal-processor.ts b/server/signal-processor.ts index 0bc6f2e9..7a55f1ed 100644 --- a/server/signal-processor.ts +++ b/server/signal-processor.ts @@ -32,12 +32,22 @@ const POLL_INTERVALS: Record = { */ export class SignalProcessor { private watches = new Map(); + private pendingRegistrations = new Map>(); 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 { + 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 { + // 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;