diff --git a/.env.example b/.env.example index d0fc4022ec..e6a434c816 100644 --- a/.env.example +++ b/.env.example @@ -2222,3 +2222,12 @@ QUOTA_STORE_DRIVER=sqlite # sqlite | redis # PROMPTQL_CREDITS_ENDPOINT=https://data.pro.ql.app/v1/graphql # PROMPTQL_TOKEN_REFRESH_URL=https://auth.pro.ql.app/ddn/project/token # PROMPTQL_POLL_TIMEOUT_MS=180000 + +# ── OmniConductor bridge (Conductor PRD RF1) ────────────────────────────────── +# Mirrors the OmniConductor hub's tasks into the local A2A TaskManager via SSE. +# Opt-in: the bridge only starts when CONDUCTOR_HUB_URL is set. +# Token: emit a `spokesperson`-kind credential on the hub (POST /v1/peers, admin) — +# server-side only, never exposed to the browser. +# Used by: src/lib/conductor/boot.ts, src/lib/conductor/bridge.ts +# CONDUCTOR_HUB_URL=http://127.0.0.1:7910 +# CONDUCTOR_HUB_TOKEN= diff --git a/changelog.d/features/conductor-bridge.md b/changelog.d/features/conductor-bridge.md new file mode 100644 index 0000000000..08050beda2 --- /dev/null +++ b/changelog.d/features/conductor-bridge.md @@ -0,0 +1 @@ +- feat(a2a): Conductor bridge — long-lived SSE consumer that mirrors OmniConductor hub tasks into the A2A TaskManager (explicit `canceled→cancelled` mapping with tests, persisted `last_event_id` cursor in the `key_value` table, exponential-backoff reconnection; opt-in via `CONDUCTOR_HUB_URL`/`CONDUCTOR_HUB_TOKEN`) diff --git a/src/instrumentation-node.ts b/src/instrumentation-node.ts index 6dad53b14d..0784499f1a 100755 --- a/src/instrumentation-node.ts +++ b/src/instrumentation-node.ts @@ -475,6 +475,15 @@ export async function registerNodejs(): Promise { console.warn("[STARTUP] Auto-refresh daemon failed to start (non-fatal):", msg); }), + // Conductor bridge (PRD Conductor RF1): mirrors OmniConductor hub tasks into the + // A2A TaskManager via the hub SSE. Opt-in — self-gated on CONDUCTOR_HUB_URL. + import("@/lib/conductor/boot").then((m) => { + if (m.initConductorBridge()) console.log("[STARTUP] Conductor bridge started"); + }).catch((err: unknown) => { + const msg = err instanceof Error ? err.message : String(err); + console.warn("[STARTUP] Conductor bridge failed to start (non-fatal):", msg); + }), + // Proactive connection-cooldown recovery (#8): re-validate connections whose // transient `rate_limited_until` window has elapsed OUTSIDE the request hot path, // so the first request after a cooldown does not pay the probe latency. diff --git a/src/lib/conductor/boot.ts b/src/lib/conductor/boot.ts new file mode 100644 index 0000000000..d920cb7d48 --- /dev/null +++ b/src/lib/conductor/boot.ts @@ -0,0 +1,36 @@ +/** + * Conductor bridge boot — production wiring for `createConductorBridge`. + * + * Opt-in: only starts when `CONDUCTOR_HUB_URL` is set. Called from + * `instrumentation-node.ts` (same non-fatal pattern as the other daemons). + * No graceful-shutdown registration needed: the cursor is persisted after each + * event, so an abrupt exit loses nothing — the hub replay converges the mirror. + */ + +import { getTaskManager } from "@/lib/a2a/taskManager"; +import { getConductorCursor, setConductorCursor } from "@/lib/db/conductorBridge"; + +import { createConductorBridge, type ConductorBridge, type ConductorBridgeOptions } from "./bridge"; + +let bridge: ConductorBridge | null = null; + +/** Starts the bridge once (idempotent). Returns null when CONDUCTOR_HUB_URL is unset. */ +export function initConductorBridge(overrides: Partial = {}): ConductorBridge | null { + const hubUrl = process.env.CONDUCTOR_HUB_URL?.trim(); + if (!hubUrl) return null; + if (bridge) return bridge; + bridge = createConductorBridge({ + hubUrl, + token: process.env.CONDUCTOR_HUB_TOKEN?.trim() ?? "", + tm: getTaskManager(), + cursor: { get: getConductorCursor, set: setConductorCursor }, + ...overrides, + }); + bridge.start(); + return bridge; +} + +export function stopConductorBridge(): void { + bridge?.stop(); + bridge = null; +} diff --git a/src/lib/db/conductorBridge.ts b/src/lib/db/conductorBridge.ts new file mode 100644 index 0000000000..edb0641aab --- /dev/null +++ b/src/lib/db/conductorBridge.ts @@ -0,0 +1,35 @@ +/** + * Conductor bridge persistence — SSE cursor (`last_event_id`) for the hub mirror. + * + * Uses the existing `key_value` table under a dedicated namespace (no migration + * needed — same pattern as settings.ts). The cursor lets the bridge resume the + * hub SSE from where it stopped; the hub replay covers the gap. + */ + +import { getDbInstance } from "./core"; + +const NAMESPACE = "conductor"; +const CURSOR_KEY = "last_event_id"; + +export function getConductorCursor(): string | null { + const db = getDbInstance(); + const row = db + .prepare("SELECT value FROM key_value WHERE namespace = ? AND key = ?") + .get(NAMESPACE, CURSOR_KEY) as { value: string } | undefined; + if (!row) return null; + try { + const parsed = JSON.parse(row.value); + return typeof parsed === "string" ? parsed : null; + } catch { + return null; + } +} + +export function setConductorCursor(value: string): void { + const db = getDbInstance(); + db.prepare("INSERT OR REPLACE INTO key_value (namespace, key, value) VALUES (?, ?, ?)").run( + NAMESPACE, + CURSOR_KEY, + JSON.stringify(value) + ); +} diff --git a/src/lib/env/runtimeEnv.ts b/src/lib/env/runtimeEnv.ts index 7b0fe12f52..577f228b62 100644 --- a/src/lib/env/runtimeEnv.ts +++ b/src/lib/env/runtimeEnv.ts @@ -69,6 +69,8 @@ export const webRuntimeEnvSchema = z.object({ OMNIROUTE_BASE_URL: optionalHttpUrl, BASE_URL: optionalHttpUrl, NEXT_PUBLIC_BASE_URL: optionalHttpUrl, + CONDUCTOR_HUB_URL: optionalHttpUrl, + CONDUCTOR_HUB_TOKEN: optionalTrimmedString, OMNIROUTE_PORT: optionalPortEnv, API_PORT: optionalPortEnv, DASHBOARD_PORT: optionalPortEnv, diff --git a/src/lib/localDb.ts b/src/lib/localDb.ts index 82b041f621..3176a21b27 100755 --- a/src/lib/localDb.ts +++ b/src/lib/localDb.ts @@ -805,3 +805,4 @@ export { markConnectionRateLimitedUntil, clearConnectionRateLimit } from "./db/p export * from "./db/paramFilters"; export * from "./db/interceptionRules"; // Per-model web-search/web-fetch interception rules (#3384) export * from "./db/relayProbeStats"; // Relay probe latency/health stats (#6909) +export * from "./db/conductorBridge"; // OmniConductor hub mirror — SSE cursor (PRD Conductor RF1) diff --git a/tests/unit/conductor-bridge-boot.test.ts b/tests/unit/conductor-bridge-boot.test.ts new file mode 100644 index 0000000000..cbde7ccb75 --- /dev/null +++ b/tests/unit/conductor-bridge-boot.test.ts @@ -0,0 +1,35 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { initConductorBridge, stopConductorBridge } from "../../src/lib/conductor/boot.ts"; + +test.afterEach(() => { + stopConductorBridge(); + delete process.env.CONDUCTOR_HUB_URL; + delete process.env.CONDUCTOR_HUB_TOKEN; +}); + +test("no-op without CONDUCTOR_HUB_URL (bridge is opt-in)", () => { + delete process.env.CONDUCTOR_HUB_URL; + assert.equal(initConductorBridge(), null); +}); + +test("starts once with CONDUCTOR_HUB_URL set and is idempotent", () => { + process.env.CONDUCTOR_HUB_URL = "http://127.0.0.1:1"; // porta inválida: conexão falha, mas o handle existe + process.env.CONDUCTOR_HUB_TOKEN = "tok"; + // cursor injetado em memória: o teste não toca o SQLite real + const first = initConductorBridge({ cursor: { get: () => null, set: () => {} } }); + assert.ok(first, "bridge iniciada"); + const second = initConductorBridge({ cursor: { get: () => null, set: () => {} } }); + assert.equal(second, first, "segunda chamada devolve a mesma instância (idempotente)"); +}); + +test("stopConductorBridge() stops and allows a fresh start", () => { + process.env.CONDUCTOR_HUB_URL = "http://127.0.0.1:1"; + const first = initConductorBridge({ cursor: { get: () => null, set: () => {} } }); + assert.ok(first); + stopConductorBridge(); + assert.equal(first!.state(), "stopped"); + const second = initConductorBridge({ cursor: { get: () => null, set: () => {} } }); + assert.ok(second && second !== first, "após stop, novo init cria instância nova"); +});