feat(a2a): start conductor bridge at boot behind CONDUCTOR_HUB_URL

.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=

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`)

src/instrumentation-node.ts

@@ -475,6 +475,15 @@ export async function registerNodejs(): Promise<void> {
           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.

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<ConductorBridgeOptions> = {}): 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;
+}

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

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,

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)

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