From d42a58141b0f1b0f7deb9aa1b26e1ab06116bdeb Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:20:29 -0300 Subject: [PATCH 1/6] =?UTF-8?q?feat(a2a):=20conductor=20bridge=20core=20?= =?UTF-8?q?=E2=80=94=20event=20mapping=20with=20canceled->cancelled?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/lib/conductor/bridge.ts | 107 ++++++++++++++++++++ tests/unit/conductor-bridge-mapping.test.ts | 106 +++++++++++++++++++ 2 files changed, 213 insertions(+) create mode 100644 src/lib/conductor/bridge.ts create mode 100644 tests/unit/conductor-bridge-mapping.test.ts diff --git a/src/lib/conductor/bridge.ts b/src/lib/conductor/bridge.ts new file mode 100644 index 0000000000..610e7f4d1b --- /dev/null +++ b/src/lib/conductor/bridge.ts @@ -0,0 +1,107 @@ +/** + * Conductor Bridge — mirrors OmniConductor hub tasks into the local A2A TaskManager. + * + * The hub's SSE feed is the source of truth; this mirror is disposable and fully + * rebuildable by replay (`last_event_id`). State spelling differs on purpose: + * the Conductor follows A2A upstream with `canceled` (1 L) while OmniRoute's + * TaskManager uses `cancelled` (2 L) — the mapping here is explicit and tested. + */ + +import type { A2ATaskManager, TaskState } from "@/lib/a2a/taskManager"; + +export interface ConductorEvent { + id: string; + type: string; + payload: Record; +} + +/** Target A2A state for a Conductor event type; null = not a task-state event (tolerated). */ +export function conductorStateFor(type: string): TaskState | null { + switch (type) { + case "task.created": + return "submitted"; + case "task.scheduled": + case "task.input_required": + return "working"; + case "task.completed": + return "completed"; + case "task.failed": + return "failed"; + case "task.canceled": // Conductor/A2A upstream spelling (1 L) + return "cancelled"; // local TaskManager spelling (2 L) + default: + return null; // runner.*, council.*, terminal.*, future types — forward-compat + } +} + +/** Minimal legal path from a task's current state up to the target state. */ +const LADDER: TaskState[] = ["submitted", "working", "completed"]; + +function climb(tm: A2ATaskManager, a2aId: string, target: TaskState): void { + const task = tm.getTask(a2aId); + if (!task) return; + if (task.state === target) return; + if (task.state === "completed" || task.state === "failed" || task.state === "cancelled") return; // terminal: late events ignored + try { + if (target === "completed") { + // submitted → working → completed (the manager rejects skips) + let idx = LADDER.indexOf(task.state); + for (idx = idx + 1; idx < LADDER.length; idx++) tm.updateTask(a2aId, LADDER[idx]); + return; + } + tm.updateTask(a2aId, target); // working/failed/cancelled are reachable from any non-terminal state + } catch { + // A transition rejected by the manager must never take the bridge down; the + // hub replay will converge the mirror on the next connection. + } +} + +function ensureMirrored(tm: A2ATaskManager, index: Map, conductorId: string, payload: Record): string { + const known = index.get(conductorId); + if (known && tm.getTask(known)) return known; + const mode = typeof payload.mode === "string" ? payload.mode : "?"; + const task = tm.createTask({ + skill: "conductor", + messages: [{ role: "user", content: `Conductor task ${conductorId} (${mode})` }], + metadata: { + conductor: { + task_id: conductorId, + mode, + from: typeof payload.from === "string" ? payload.from : undefined, + }, + }, + }); + index.set(conductorId, task.id); + return task.id; +} + +function conductorMeta(tm: A2ATaskManager, a2aId: string): Record | null { + const task = tm.getTask(a2aId); + if (!task) return null; + // The manager has no public metadata mutator; direct mutation is the repo precedent + // (tests mutate task fields) and the object lives in the same process. + const meta = (task.metadata.conductor ??= {}) as Record; + return meta; +} + +/** Applies one hub event to the mirror. Never throws for event-shaped input. */ +export function applyConductorEvent(tm: A2ATaskManager, index: Map, ev: ConductorEvent): void { + const target = conductorStateFor(ev.type); + if (!target) return; + const conductorId = typeof ev.payload.task_id === "string" ? ev.payload.task_id : null; + if (!conductorId) return; + + const a2aId = ensureMirrored(tm, index, conductorId, ev.payload); + const meta = conductorMeta(tm, a2aId); + if (meta) { + if (ev.type === "task.scheduled" && typeof ev.payload.runner_id === "string") meta.runner = ev.payload.runner_id; + if (ev.type === "task.input_required") meta.input_required = true; + if (ev.type === "task.completed" || ev.type === "task.failed") { + const manifest = (ev.payload.manifest ?? {}) as Record; + if (typeof manifest.summary === "string") meta.summary = manifest.summary; + if (typeof manifest.branch === "string") meta.branch = manifest.branch; + if (typeof manifest.error === "string") meta.error = manifest.error; + } + } + if (target !== "submitted") climb(tm, a2aId, target); +} diff --git a/tests/unit/conductor-bridge-mapping.test.ts b/tests/unit/conductor-bridge-mapping.test.ts new file mode 100644 index 0000000000..ed4c69e518 --- /dev/null +++ b/tests/unit/conductor-bridge-mapping.test.ts @@ -0,0 +1,106 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { A2ATaskManager } from "../../src/lib/a2a/taskManager.ts"; +import { applyConductorEvent, conductorStateFor } from "../../src/lib/conductor/bridge.ts"; + +const managers: A2ATaskManager[] = []; + +function createManager(ttlMinutes = 5) { + const manager = new A2ATaskManager(ttlMinutes); + managers.push(manager); + return manager; +} + +test.afterEach(() => { + while (managers.length > 0) { + managers.pop()?.destroy(); + } +}); + +const ev = (id: number, type: string, payload: Record) => ({ + id: String(id), + type, + payload, +}); + +function mirrored(tm: A2ATaskManager, index: Map, conductorId: string) { + const a2aId = index.get(conductorId); + assert.ok(a2aId, `task ${conductorId} indexada`); + const task = tm.getTask(a2aId!); + assert.ok(task, `task A2A ${a2aId} existe`); + return task!; +} + +test("task.created mirrors as submitted with conductor metadata", () => { + const tm = createManager(); + const index = new Map(); + applyConductorEvent(tm, index, ev(1, "task.created", { task_id: "t_abc", mode: "solo", from: "orchestrator" })); + const task = mirrored(tm, index, "t_abc"); + assert.equal(task.state, "submitted"); + assert.equal(task.skill, "conductor"); + const meta = task.metadata.conductor as Record; + assert.equal(meta.task_id, "t_abc"); + assert.equal(meta.mode, "solo"); +}); + +test("created → scheduled → completed climbs the transition ladder without throwing", () => { + const tm = createManager(); + const index = new Map(); + applyConductorEvent(tm, index, ev(1, "task.created", { task_id: "t_ok", mode: "solo", from: "x" })); + applyConductorEvent(tm, index, ev(2, "task.scheduled", { task_id: "t_ok", runner_id: "r_1" })); + applyConductorEvent(tm, index, ev(3, "task.completed", { task_id: "t_ok", manifest: { summary: "feito", branch: "task/t_ok" } })); + const task = mirrored(tm, index, "t_ok"); + assert.equal(task.state, "completed"); + const meta = task.metadata.conductor as Record; + assert.equal(meta.runner, "r_1"); + assert.equal(meta.summary, "feito"); + assert.equal(meta.branch, "task/t_ok"); +}); + +test("task.canceled (Conductor, 1 L) maps to cancelled (A2A local, 2 L)", () => { + const tm = createManager(); + const index = new Map(); + assert.equal(conductorStateFor("task.canceled"), "cancelled"); + applyConductorEvent(tm, index, ev(1, "task.created", { task_id: "t_cx", mode: "solo", from: "x" })); + applyConductorEvent(tm, index, ev(2, "task.canceled", { task_id: "t_cx", by: "operator" })); + const task = mirrored(tm, index, "t_cx"); + assert.equal(task.state, "cancelled"); +}); + +test("unknown event types are tolerated and mirror nothing (forward-compat)", () => { + const tm = createManager(); + const index = new Map(); + applyConductorEvent(tm, index, ev(1, "runner.registered", { runner_id: "r_1", capabilities: {} })); + applyConductorEvent(tm, index, ev(2, "council.fanout", { task_id: "t_c", candidate_task_ids: [] })); + applyConductorEvent(tm, index, ev(3, "some.future.event", {})); + assert.equal(index.size, 0); + assert.equal(tm.listTasks().length, 0); +}); + +test("late-join: terminal event for a never-seen task creates it and lands terminal", () => { + const tm = createManager(); + const index = new Map(); + applyConductorEvent(tm, index, ev(9, "task.completed", { task_id: "t_late", manifest: { summary: "s", branch: null } })); + const task = mirrored(tm, index, "t_late"); + assert.equal(task.state, "completed"); +}); + +test("events after a terminal state are ignored without throwing", () => { + const tm = createManager(); + const index = new Map(); + applyConductorEvent(tm, index, ev(1, "task.created", { task_id: "t_t", mode: "solo", from: "x" })); + applyConductorEvent(tm, index, ev(2, "task.canceled", { task_id: "t_t" })); + assert.doesNotThrow(() => applyConductorEvent(tm, index, ev(3, "task.completed", { task_id: "t_t", manifest: {} }))); + assert.equal(mirrored(tm, index, "t_t").state, "cancelled"); +}); + +test("task.input_required stays working with input_required flag (A2A has no such state)", () => { + const tm = createManager(); + const index = new Map(); + applyConductorEvent(tm, index, ev(1, "task.created", { task_id: "t_i", mode: "solo", from: "x" })); + applyConductorEvent(tm, index, ev(2, "task.input_required", { task_id: "t_i", question: "qual branch?" })); + const task = mirrored(tm, index, "t_i"); + assert.equal(task.state, "working"); + assert.equal((task.metadata.conductor as Record).input_required, true); +}); From 33baf62b58c2c256f1743c68ec3533d82bae1405 Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:21:10 -0300 Subject: [PATCH 2/6] feat(a2a): incremental SSE parser for conductor bridge --- src/lib/conductor/bridge.ts | 143 ++++++++++++++++++++++++ tests/unit/conductor-bridge-sse.test.ts | 43 +++++++ 2 files changed, 186 insertions(+) create mode 100644 tests/unit/conductor-bridge-sse.test.ts diff --git a/src/lib/conductor/bridge.ts b/src/lib/conductor/bridge.ts index 610e7f4d1b..5839f9cb31 100644 --- a/src/lib/conductor/bridge.ts +++ b/src/lib/conductor/bridge.ts @@ -7,6 +7,8 @@ * TaskManager uses `cancelled` (2 L) — the mapping here is explicit and tested. */ +import { z } from "zod"; + import type { A2ATaskManager, TaskState } from "@/lib/a2a/taskManager"; export interface ConductorEvent { @@ -105,3 +107,144 @@ export function applyConductorEvent(tm: A2ATaskManager, index: Map void; + backoffBaseMs?: number; +} + +export interface ConductorBridge { + start(): void; + stop(): void; + state(): "connected" | "reconnecting" | "stopped"; +} + +const BACKOFF_CAP_MS = 30_000; + +/** + * Long-lived consumer: connects to the hub SSE with the persisted cursor, mirrors + * events, reconnects with exponential backoff. A hub outage never propagates — + * errors are logged and retried; `stop()` aborts for good. + */ +export function createConductorBridge(opts: ConductorBridgeOptions): ConductorBridge { + const log = opts.log ?? ((msg: string) => console.log(`[conductor-bridge] ${msg}`)); + const doFetch = opts.fetchImpl ?? fetch; + const backoffBase = opts.backoffBaseMs ?? 1_000; + const index = new Map(); + let state: "connected" | "reconnecting" | "stopped" = "stopped"; + let abort: AbortController | null = null; + let attempt = 0; + let running = false; + + async function readStream(res: Response): Promise { + const reader = res.body?.getReader(); + if (!reader) throw new Error("SSE response without body"); + const parser = new SseParser(); + const decoder = new TextDecoder(); + for (;;) { + const { value, done } = await reader.read(); + if (done) return; + for (const frame of parser.push(decoder.decode(value, { stream: true }))) { + let parsed: z.infer; + try { + parsed = hubEventSchema.parse(JSON.parse(frame.data)); + } catch { + log(`evento SSE malformado ignorado (id=${frame.id ?? "?"})`); + continue; + } + applyConductorEvent(opts.tm, index, parsed); + opts.cursor.set(parsed.id); // persisted per event — replay covers any gap on reconnect + } + } + } + + async function loop(): Promise { + while (running) { + abort = new AbortController(); + try { + const since = opts.cursor.get() ?? "0"; + const res = await doFetch(`${opts.hubUrl}/v1/events?last_event_id=${encodeURIComponent(since)}`, { + headers: { authorization: `Bearer ${opts.token}`, accept: "text/event-stream" }, + signal: abort.signal, + }); + if (!res.ok) throw new Error(`hub respondeu HTTP ${res.status}`); + state = "connected"; + attempt = 0; + log(`conectado ao hub (last_event_id=${since})`); + await readStream(res); // resolves when the hub closes the stream + throw new Error("stream encerrado pelo hub"); + } catch (err) { + if (!running) break; + state = "reconnecting"; + const wait = Math.min(backoffBase * 2 ** attempt, BACKOFF_CAP_MS); + attempt++; + log(`conexão caiu (${err instanceof Error ? err.message : String(err)}) — reconectando em ${wait}ms`); + await new Promise((resolve) => setTimeout(resolve, wait)); + } + } + state = "stopped"; + } + + return { + start() { + if (running) return; + running = true; + void loop(); + }, + stop() { + running = false; + state = "stopped"; + abort?.abort(); + }, + state: () => state, + }; +} diff --git a/tests/unit/conductor-bridge-sse.test.ts b/tests/unit/conductor-bridge-sse.test.ts new file mode 100644 index 0000000000..754d936c13 --- /dev/null +++ b/tests/unit/conductor-bridge-sse.test.ts @@ -0,0 +1,43 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { SseParser } from "../../src/lib/conductor/bridge.ts"; + +test("whole event in a single chunk", () => { + const p = new SseParser(); + const out = p.push('id: 7\nevent: task.created\ndata: {"task_id":"t_1"}\n\n'); + assert.equal(out.length, 1); + assert.deepEqual(out[0], { id: "7", event: "task.created", data: '{"task_id":"t_1"}' }); +}); + +test("event fragmented across three chunks (cut mid-data)", () => { + const p = new SseParser(); + assert.equal(p.push("id: 8\nevent: task.sch").length, 0); + assert.equal(p.push('eduled\ndata: {"task_id"').length, 0); + const out = p.push(':"t_2"}\n\n'); + assert.equal(out.length, 1); + assert.deepEqual(out[0], { id: "8", event: "task.scheduled", data: '{"task_id":"t_2"}' }); +}); + +test("two events in one chunk", () => { + const p = new SseParser(); + const out = p.push("id: 1\nevent: a\ndata: x\n\nid: 2\nevent: b\ndata: y\n\n"); + assert.equal(out.length, 2); + assert.equal(out[0].id, "1"); + assert.equal(out[1].data, "y"); +}); + +test("comment pings are ignored", () => { + const p = new SseParser(); + assert.equal(p.push(": ping\n\n").length, 0); + const out = p.push(": ping\n\nid: 3\nevent: c\ndata: z\n\n"); + assert.equal(out.length, 1); + assert.equal(out[0].id, "3"); +}); + +test("multi-line data joins with newline (SSE spec)", () => { + const p = new SseParser(); + const out = p.push("event: m\ndata: line1\ndata: line2\n\n"); + assert.equal(out.length, 1); + assert.equal(out[0].data, "line1\nline2"); +}); From 5eb10e896d02739529020bf97dc61ab08590dd90 Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:21:10 -0300 Subject: [PATCH 3/6] feat(a2a): conductor bridge connection loop with persisted cursor and backoff --- tests/unit/conductor-bridge-loop.test.ts | 140 +++++++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 tests/unit/conductor-bridge-loop.test.ts diff --git a/tests/unit/conductor-bridge-loop.test.ts b/tests/unit/conductor-bridge-loop.test.ts new file mode 100644 index 0000000000..37fb4d6abf --- /dev/null +++ b/tests/unit/conductor-bridge-loop.test.ts @@ -0,0 +1,140 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { createServer, type Server } from "node:http"; + +import { A2ATaskManager } from "../../src/lib/a2a/taskManager.ts"; +import { createConductorBridge } from "../../src/lib/conductor/bridge.ts"; + +const managers: A2ATaskManager[] = []; +const servers: Server[] = []; +const bridges: { stop(): void }[] = []; + +function createManager() { + const manager = new A2ATaskManager(5); + managers.push(manager); + return manager; +} + +test.afterEach(async () => { + while (bridges.length > 0) bridges.pop()?.stop(); + while (managers.length > 0) managers.pop()?.destroy(); + while (servers.length > 0) { + const s = servers.pop(); + await new Promise((resolve) => s?.close(resolve)); + } +}); + +const sse = (id: number, type: string, payload: Record) => + `id: ${id}\nevent: ${type}\ndata: ${JSON.stringify({ id: String(id), type, ts: "2026-07-22T00:00:00Z", payload })}\n\n`; + +interface FakeHub { + url: string; + requests: { lastEventId: string | null; auth: string | null }[]; +} + +/** Fake hub SSE: each handler invocation pops the next script entry {events, thenClose}. */ +function fakeHub(script: { events: string[]; thenClose: boolean }[]): Promise { + const requests: FakeHub["requests"] = []; + let call = 0; + const server = createServer((req, res) => { + const u = new URL(req.url ?? "/", "http://x"); + requests.push({ lastEventId: u.searchParams.get("last_event_id"), auth: req.headers.authorization ?? null }); + const step = script[Math.min(call, script.length - 1)]; + call++; + res.writeHead(200, { "content-type": "text/event-stream" }); + for (const e of step.events) res.write(e); + if (step.thenClose) res.end(); + // senão: conexão fica aberta (o teste encerra via bridge.stop() + server.close()) + }); + servers.push(server); + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + const addr = server.address(); + resolve({ url: `http://127.0.0.1:${typeof addr === "object" && addr ? addr.port : 0}`, requests }); + }); + }); +} + +function waitFor(cond: () => boolean, ms = 3000): Promise { + const deadline = Date.now() + ms; + return new Promise((resolve, reject) => { + const tick = () => { + if (cond()) return resolve(); + if (Date.now() > deadline) return reject(new Error("waitFor timeout")); + setTimeout(tick, 10); + }; + tick(); + }); +} + +test("mirrors events from the hub SSE, advancing the injected cursor (incl. canceled→cancelled)", async () => { + const hub = await fakeHub([ + { + events: [ + sse(1, "task.created", { task_id: "t_x", mode: "solo", from: "orch" }), + sse(2, "task.scheduled", { task_id: "t_x", runner_id: "r_1" }), + sse(3, "task.canceled", { task_id: "t_x" }), + ], + thenClose: false, + }, + ]); + const tm = createManager(); + let cursor: string | null = null; + const bridge = createConductorBridge({ + hubUrl: hub.url, + token: "tok-spk", + tm, + cursor: { get: () => cursor, set: (v) => (cursor = v) }, + backoffBaseMs: 10, + }); + bridges.push(bridge); + bridge.start(); + await waitFor(() => cursor === "3"); + assert.equal(hub.requests[0].lastEventId, "0", "sem cursor persistido, começa do 0 (replay total)"); + assert.equal(hub.requests[0].auth, "Bearer tok-spk"); + const tasks = tm.listTasks(); + assert.equal(tasks.length, 1); + assert.equal(tasks[0].state, "cancelled"); +}); + +test("reconnects with backoff and resumes from the persisted cursor", async () => { + const hub = await fakeHub([ + { events: [sse(1, "task.created", { task_id: "t_r", mode: "solo", from: "o" })], thenClose: true }, + { events: [sse(2, "task.completed", { task_id: "t_r", manifest: { summary: "ok" } })], thenClose: false }, + ]); + const tm = createManager(); + let cursor: string | null = null; + const bridge = createConductorBridge({ + hubUrl: hub.url, + token: "tok", + tm, + cursor: { get: () => cursor, set: (v) => (cursor = v) }, + backoffBaseMs: 10, + }); + bridges.push(bridge); + bridge.start(); + await waitFor(() => hub.requests.length >= 2 && cursor === "2"); + assert.equal(hub.requests[1].lastEventId, "1", "reconexão retoma do cursor persistido"); + assert.equal(tm.listTasks()[0].state, "completed"); +}); + +test("stop() aborts and never reconnects", async () => { + const hub = await fakeHub([{ events: [sse(1, "task.created", { task_id: "t_s", mode: "solo", from: "o" })], thenClose: true }]); + const tm = createManager(); + let cursor: string | null = null; + const bridge = createConductorBridge({ + hubUrl: hub.url, + token: "tok", + tm, + cursor: { get: () => cursor, set: (v) => (cursor = v) }, + backoffBaseMs: 10, + }); + bridges.push(bridge); + bridge.start(); + await waitFor(() => cursor === "1"); + bridge.stop(); + const before = hub.requests.length; + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.equal(hub.requests.length, before, "nenhuma reconexão após stop()"); + assert.equal(bridge.state(), "stopped"); +}); From 00e15af62216ba52339501191e13cd55390cf048 Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:23:28 -0300 Subject: [PATCH 4/6] feat(a2a): start conductor bridge at boot behind CONDUCTOR_HUB_URL --- .env.example | 9 ++++++ changelog.d/features/conductor-bridge.md | 1 + src/instrumentation-node.ts | 9 ++++++ src/lib/conductor/boot.ts | 36 ++++++++++++++++++++++++ src/lib/db/conductorBridge.ts | 35 +++++++++++++++++++++++ src/lib/env/runtimeEnv.ts | 2 ++ src/lib/localDb.ts | 1 + tests/unit/conductor-bridge-boot.test.ts | 35 +++++++++++++++++++++++ 8 files changed, 128 insertions(+) create mode 100644 changelog.d/features/conductor-bridge.md create mode 100644 src/lib/conductor/boot.ts create mode 100644 src/lib/db/conductorBridge.ts create mode 100644 tests/unit/conductor-bridge-boot.test.ts 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"); +}); From a75295f35986dfeef9fbbc1911c5b229aca166e7 Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:25:35 -0300 Subject: [PATCH 5/6] fix(a2a): conductor bridge parses the hub's real SSE wire format (id/type in frame, data={ts,payload}) --- src/lib/conductor/bridge.ts | 18 +++++++++--------- tests/unit/conductor-bridge-loop.test.ts | 4 +++- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/src/lib/conductor/bridge.ts b/src/lib/conductor/bridge.ts index 5839f9cb31..ed3396b472 100644 --- a/src/lib/conductor/bridge.ts +++ b/src/lib/conductor/bridge.ts @@ -145,10 +145,9 @@ export class SseParser { // ============ Connection loop ============ -/** Wire shape of one hub event (the SSE `data:` payload) — untrusted input, Zod-validated. */ -const hubEventSchema = z.object({ - id: z.string(), - type: z.string(), +/** Wire shape of the hub's SSE `data:` field — {ts, payload}; id/type live in the + * SSE frame fields (verified against the live hub, 2026-07-22). Untrusted input → Zod. */ +const hubDataSchema = z.object({ payload: z.record(z.string(), z.unknown()), }); @@ -194,15 +193,16 @@ export function createConductorBridge(opts: ConductorBridgeOptions): ConductorBr const { value, done } = await reader.read(); if (done) return; for (const frame of parser.push(decoder.decode(value, { stream: true }))) { - let parsed: z.infer; + if (!frame.id || !frame.event) continue; // keepalive/partial frame — nada a espelhar + let payload: Record; try { - parsed = hubEventSchema.parse(JSON.parse(frame.data)); + payload = hubDataSchema.parse(JSON.parse(frame.data)).payload; } catch { - log(`evento SSE malformado ignorado (id=${frame.id ?? "?"})`); + log(`evento SSE malformado ignorado (id=${frame.id})`); continue; } - applyConductorEvent(opts.tm, index, parsed); - opts.cursor.set(parsed.id); // persisted per event — replay covers any gap on reconnect + applyConductorEvent(opts.tm, index, { id: frame.id, type: frame.event, payload }); + opts.cursor.set(frame.id); // persisted per event — replay covers any gap on reconnect } } } diff --git a/tests/unit/conductor-bridge-loop.test.ts b/tests/unit/conductor-bridge-loop.test.ts index 37fb4d6abf..c03dafc493 100644 --- a/tests/unit/conductor-bridge-loop.test.ts +++ b/tests/unit/conductor-bridge-loop.test.ts @@ -24,8 +24,10 @@ test.afterEach(async () => { } }); +// Formato REAL do SSE do hub (verificado ao vivo 2026-07-22): id/type nos campos do +// frame; o `data:` carrega só {ts, payload}. const sse = (id: number, type: string, payload: Record) => - `id: ${id}\nevent: ${type}\ndata: ${JSON.stringify({ id: String(id), type, ts: "2026-07-22T00:00:00Z", payload })}\n\n`; + `id: ${id}\nevent: ${type}\ndata: ${JSON.stringify({ ts: "2026-07-22T00:00:00Z", payload })}\n\n`; interface FakeHub { url: string; From b97318d73bd8a6b6c4b0ee9567d7a4d875c08ee1 Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 22 Jul 2026 01:06:22 -0300 Subject: [PATCH 6/6] docs(reference): document CONDUCTOR_HUB_URL/TOKEN in ENVIRONMENT.md (env-doc-sync gate) --- docs/reference/ENVIRONMENT.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 66c08a73b1..b6b7ae462e 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -1232,3 +1232,14 @@ Not required for normal operation — developer tooling only. | Variable | Default | Source File | Description | | ---------------------------- | ------------ | ----------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------- | | `OMNIROUTE_EVAL_CREDENTIALS` | `{}` (empty) | `scripts/compression-eval/index.ts` | Operator-supplied JSON credentials for the provider exercised by the offline compression-eval CLI (parsed with `JSON.parse`). Leave unset for a dry run. | + +--- + +## OmniConductor Bridge + +Long-lived SSE consumer that mirrors OmniConductor hub tasks into the local A2A TaskManager (`src/lib/conductor/`). Opt-in — the bridge only starts when `CONDUCTOR_HUB_URL` is set. Server-side only: the hub token must never reach the browser. + +| Variable | Default | Source File | Description | +| --------------------- | ---------- | ---------------------------- | ------------------------------------------------------------------------------------------------------- | +| `CONDUCTOR_HUB_URL` | _(empty)_ | `src/lib/conductor/boot.ts` | Base URL of the OmniConductor hub (e.g. `http://127.0.0.1:7910`). Unset = bridge disabled. | +| `CONDUCTOR_HUB_TOKEN` | _(empty)_ | `src/lib/conductor/boot.ts` | Hub credential for the SSE feed — emit a `spokesperson`-kind peer on the hub (`POST /v1/peers`, admin). |