diff --git a/.env.example b/.env.example index 5a58eb3fc7..1c36d1125c 100644 --- a/.env.example +++ b/.env.example @@ -2234,3 +2234,6 @@ QUOTA_STORE_DRIVER=sqlite # sqlite | redis # Faro (OmniConductor spokesperson) — chat/voice proxy for the Conductor panel. # Used by: src/lib/conductor/faroProxy.ts # CONDUCTOR_SPOKESPERSON_URL=http://127.0.0.1:7920 +# Optional dedicated credential for inbound A2A delegation (falls back to CONDUCTOR_HUB_TOKEN). +# Used by: src/lib/conductor/hubProxy.ts +# CONDUCTOR_ORCHESTRATOR_TOKEN= diff --git a/changelog.d/features/conductor-a2a-in.md b/changelog.d/features/conductor-a2a-in.md new file mode 100644 index 0000000000..0922d620fc --- /dev/null +++ b/changelog.d/features/conductor-a2a-in.md @@ -0,0 +1 @@ +- feat(a2a): inbound delegation to the OmniConductor fleet — `POST /api/a2a/tasks` translates an external A2A task into the hub's `POST /v1/tasks` (fleet skills only, repo required, `CONDUCTOR_ORCHESTRATOR_TOKEN` with hub-token fallback); states flow back through the SSE→A2A mirror diff --git a/docs/frameworks/A2A-SERVER.md b/docs/frameworks/A2A-SERVER.md index 9cc405fd2b..ea451e42d3 100644 --- a/docs/frameworks/A2A-SERVER.md +++ b/docs/frameworks/A2A-SERVER.md @@ -178,6 +178,9 @@ The JSON-RPC endpoint `/a2a` is the canonical A2A entry point. The REST endpoint | `/api/a2a/tasks/[id]` | GET | Get task by ID | management | | `/api/a2a/tasks/[id]/cancel` | POST | Cancel running task | management | | `/.well-known/agent.json` | GET | Agent Card (A2A discovery) | (public, cached 3600s) | +| `/api/a2a/tasks` | POST | Inbound delegation to the OmniConductor fleet (Conductor PRD RF5) | Bearer vs `OMNIROUTE_API_KEY` + `a2aEnabled` | + +**Inbound Conductor delegation (`POST /api/a2a/tasks`):** external A2A agents delegate coding work to the OmniConductor fleet through OmniRoute. Body: `{ skill: "conductor" | "conductor-cli-", messages: [{role, content}], metadata: { conductor: { repo: { url, base_ref? }, mode?, cli?, model? } } }` — only Conductor fleet skills (the ones announced on the Agent Card) are delegable; `metadata.conductor.repo.url` is required (the fleet works on git repos). The route translates to the hub's `POST /v1/tasks` using the server-side `CONDUCTOR_ORCHESTRATOR_TOKEN` (fallback `CONDUCTOR_HUB_TOKEN`) and returns `201 { conductor_task_id, state: "submitted" }`; task states flow back through the SSE→A2A mirror (RF1) and are visible via `GET /api/a2a/tasks?skill=conductor`. --- diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index 2340f439da..795354e079 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/docs/reference/ENVIRONMENT.md @@ -1244,3 +1244,4 @@ Long-lived SSE consumer that mirrors OmniConductor hub tasks into the local A2A | `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). | | `CONDUCTOR_SPOKESPERSON_URL` | `http://127.0.0.1:7920` | `src/lib/conductor/faroProxy.ts` | Base URL of Faro (the Conductor spokesperson) for the dashboard chat/voice proxy. | +| `CONDUCTOR_ORCHESTRATOR_TOKEN` | _(empty)_ | `src/lib/conductor/hubProxy.ts` | Hub credential used for INBOUND A2A delegation (`POST /api/a2a/tasks` → hub `POST /v1/tasks`). Emit an `orchestrator`-kind peer; falls back to `CONDUCTOR_HUB_TOKEN`. | diff --git a/src/app/api/a2a/tasks/route.ts b/src/app/api/a2a/tasks/route.ts index 73bd0c9694..02e264c85a 100644 --- a/src/app/api/a2a/tasks/route.ts +++ b/src/app/api/a2a/tasks/route.ts @@ -1,5 +1,9 @@ import { NextResponse } from "next/server"; +import { z } from "zod"; + import { getTaskManager, type TaskState } from "@/lib/a2a/taskManager"; +import { createConductorTask } from "@/lib/conductor/hubProxy"; +import { getSettings } from "@/lib/db/settings"; const VALID_TASK_STATES = new Set([ "submitted", @@ -44,3 +48,97 @@ export async function GET(request: Request) { return NextResponse.json({ error: message }, { status: 500 }); } } + +// ============ POST — delegação de entrada à frota do Conductor (PRD Conductor RF5) ============ + +const delegationSchema = z.object({ + skill: z.string().default("conductor"), + messages: z.array(z.object({ role: z.string(), content: z.string() })).min(1), + metadata: z + .object({ + conductor: z + .object({ + repo: z.object({ url: z.string().min(1), base_ref: z.string().optional() }).optional(), + mode: z.string().optional(), + cli: z.string().optional(), + model: z.string().optional(), + }) + .optional(), + }) + .optional(), +}); + +/** Mesma semântica de auth do JSON-RPC A2A (src/app/a2a/route.ts): Bearer vs OMNIROUTE_API_KEY; aberto se não configurada. */ +function authenticateA2A(request: Request): boolean { + const configuredKey = process.env.OMNIROUTE_API_KEY; + if (!configuredKey) return true; + const token = (request.headers.get("authorization") || "").replace(/^Bearer\s+/i, ""); + return token === configuredKey; +} + +/** + * Traduz uma task A2A externa em `POST /v1/tasks` do hub do OmniConductor. + * Só skills da frota (`conductor` / `conductor-cli-` — as anunciadas no + * Agent Card) são delegáveis; os estados voltam pelo espelho SSE→A2A (RF1). + */ +export async function POST(request: Request) { + if (!authenticateA2A(request)) { + return NextResponse.json({ error: "Unauthorized: missing or invalid API key" }, { status: 401 }); + } + const settings = await getSettings(); + if (settings.a2aEnabled !== true) { + return NextResponse.json( + { error: "A2A endpoint is disabled. Enable it from the Endpoints page." }, + { status: 503 } + ); + } + + let raw: unknown; + try { + raw = await request.json(); + } catch { + return NextResponse.json({ error: "Invalid JSON body" }, { status: 400 }); + } + const parsed = delegationSchema.safeParse(raw); + if (!parsed.success) { + return NextResponse.json({ error: "Invalid A2A task: provide messages[] (and metadata.conductor)" }, { status: 400 }); + } + const { skill, messages, metadata } = parsed.data; + if (skill !== "conductor" && !skill.startsWith("conductor-cli-")) { + return NextResponse.json( + { error: "Only Conductor fleet skills are delegable here (conductor / conductor-cli-)" }, + { status: 400 } + ); + } + const conductor = metadata?.conductor; + if (!conductor?.repo?.url) { + return NextResponse.json( + { error: "Delegation requires metadata.conductor.repo.url (the fleet works on git repos)" }, + { status: 400 } + ); + } + const prompt = [...messages].reverse().find((m) => m.role === "user")?.content ?? messages[messages.length - 1].content; + + const created = await createConductorTask({ + repoUrl: conductor.repo.url, + baseRef: conductor.repo.base_ref, + prompt, + mode: conductor.mode, + cli: skill.startsWith("conductor-cli-") ? skill.slice("conductor-cli-".length) : conductor.cli, + model: conductor.model, + }); + if (!created.ok) { + return NextResponse.json( + { error: `Conductor hub refused the delegation (HTTP ${created.status})` }, + { status: created.status } + ); + } + return NextResponse.json( + { + conductor_task_id: created.task_id, + state: "submitted", + note: "States flow back through the SSE→A2A mirror (GET /api/a2a/tasks, skill=conductor).", + }, + { status: 201 } + ); +} diff --git a/src/lib/conductor/hubProxy.ts b/src/lib/conductor/hubProxy.ts index c8f3217ed5..d684c68c08 100644 --- a/src/lib/conductor/hubProxy.ts +++ b/src/lib/conductor/hubProxy.ts @@ -156,6 +156,52 @@ export async function getConductorTaskDetail( } } +export interface DelegationInput { + repoUrl: string; + prompt: string; + baseRef?: string; + mode?: string; + cli?: string; + model?: string; +} + +/** + * Delegates work to the fleet: translates an external A2A task into the hub's + * `POST /v1/tasks` (Conductor PRD RF5). Uses the orchestrator credential when + * set (CONDUCTOR_ORCHESTRATOR_TOKEN), falling back to the hub token. States + * flow back through the RF1 mirror — this call only creates. + */ +export async function createConductorTask( + input: DelegationInput, + opts: HubProxyOptions = {} +): Promise<{ ok: boolean; status: number; task_id?: string }> { + const cfg = hubConfig(); + if (!cfg) return { ok: false, status: 503 }; + const token = process.env.CONDUCTOR_ORCHESTRATOR_TOKEN?.trim() || cfg.token; + const body: Record = { + repo: { url: input.repoUrl, base_ref: input.baseRef?.trim() || "main" }, + spec: { prompt: input.prompt }, + mode: input.mode?.trim() || "solo", + }; + const requirements: Record = {}; + if (input.cli?.trim()) requirements.cli = input.cli.trim(); + if (input.model?.trim()) requirements.model = input.model.trim(); + if (Object.keys(requirements).length) body.requirements = requirements; + try { + const doFetch = opts.fetchImpl ?? fetch; + const res = await doFetch(`${cfg.url}/v1/tasks`, { + method: "POST", + headers: { authorization: `Bearer ${token}`, "content-type": "application/json" }, + body: JSON.stringify(body), + }); + if (!res.ok) return { ok: false, status: res.status }; + const created = z.object({ id: z.string() }).parse(await res.json()); + return { ok: true, status: res.status, task_id: created.id }; + } catch { + return { ok: false, status: 503 }; + } +} + /** Cancels a task on the hub. Returns the hub's verdict without leaking its body on error. */ export async function cancelConductorTask( taskId: string, diff --git a/src/lib/env/runtimeEnv.ts b/src/lib/env/runtimeEnv.ts index 738fa99865..632d94ee05 100644 --- a/src/lib/env/runtimeEnv.ts +++ b/src/lib/env/runtimeEnv.ts @@ -71,6 +71,7 @@ export const webRuntimeEnvSchema = z.object({ NEXT_PUBLIC_BASE_URL: optionalHttpUrl, CONDUCTOR_HUB_URL: optionalHttpUrl, CONDUCTOR_SPOKESPERSON_URL: optionalHttpUrl, + CONDUCTOR_ORCHESTRATOR_TOKEN: optionalTrimmedString, CONDUCTOR_HUB_TOKEN: optionalTrimmedString, OMNIROUTE_PORT: optionalPortEnv, API_PORT: optionalPortEnv, diff --git a/tests/unit/conductor-a2a-post.test.ts b/tests/unit/conductor-a2a-post.test.ts new file mode 100644 index 0000000000..7cbae0e4d6 --- /dev/null +++ b/tests/unit/conductor-a2a-post.test.ts @@ -0,0 +1,112 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { createServer, type Server } from "node:http"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-conductor-a2a-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../src/lib/db/core.ts"); +const settings = await import("../../src/lib/db/settings.ts"); +const tasksRoute = await import("../../src/app/api/a2a/tasks/route.ts"); + +const servers: Server[] = []; + +async function enableA2A() { + await settings.updateSettings({ a2aEnabled: true }); +} + +function delegationRequest(body: unknown, bearer?: string) { + return new Request("http://localhost/api/a2a/tasks", { + method: "POST", + headers: { + "content-type": "application/json", + ...(bearer ? { authorization: `Bearer ${bearer}` } : {}), + }, + body: JSON.stringify(body), + }); +} + +const VALID_BODY = { + skill: "conductor-cli-claude", + messages: [{ role: "user", content: "adicione um README com a seção Sobre" }], + metadata: { conductor: { repo: { url: "https://git.x/repo", base_ref: "dev" }, mode: "solo", model: "cc/claude-sonnet-5" } }, +}; + +test.beforeEach(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); + delete process.env.CONDUCTOR_HUB_URL; + delete process.env.OMNIROUTE_API_KEY; +}); + +test.after(async () => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + delete process.env.CONDUCTOR_HUB_URL; + delete process.env.OMNIROUTE_API_KEY; + while (servers.length > 0) { + const s = servers.pop(); + await new Promise((resolve) => s?.close(resolve)); + } +}); + +test("A2A desabilitado → 503 (mesmo gate do JSON-RPC)", async () => { + const res = await tasksRoute.POST(delegationRequest(VALID_BODY)); + assert.equal(res.status, 503); +}); + +test("com OMNIROUTE_API_KEY configurada, bearer errado → 401 e bearer certo passa", async () => { + await enableA2A(); + process.env.OMNIROUTE_API_KEY = "chave-certa"; + const denied = await tasksRoute.POST(delegationRequest(VALID_BODY, "chave-errada")); + assert.equal(denied.status, 401); +}); + +test("delegação válida → 201 com o task_id do hub; requirements derivados da skill do card", async () => { + await enableA2A(); + const bodies: unknown[] = []; + await new Promise((resolve) => { + const server = createServer((req, res) => { + let raw = ""; + req.on("data", (c) => (raw += c)); + req.on("end", () => { + bodies.push(JSON.parse(raw)); + res.writeHead(201, { "content-type": "application/json" }); + res.end(JSON.stringify({ id: "t_delegada", status: "submitted" })); + }); + }); + servers.push(server); + server.listen(0, "127.0.0.1", () => { + const addr = server.address(); + process.env.CONDUCTOR_HUB_URL = `http://127.0.0.1:${typeof addr === "object" && addr ? addr.port : 0}`; + resolve(); + }); + }); + const res = await tasksRoute.POST(delegationRequest(VALID_BODY)); + assert.equal(res.status, 201); + const out = await res.json(); + assert.equal(out.conductor_task_id, "t_delegada"); + assert.equal(out.state, "submitted"); + const sent = bodies[0] as { repo: { url: string; base_ref: string }; spec: { prompt: string }; requirements: { cli: string; model: string } }; + assert.equal(sent.repo.url, "https://git.x/repo"); + assert.equal(sent.repo.base_ref, "dev"); + assert.equal(sent.spec.prompt, "adicione um README com a seção Sobre"); + assert.equal(sent.requirements.cli, "claude", "skill conductor-cli-claude vira requirements.cli"); + assert.equal(sent.requirements.model, "cc/claude-sonnet-5"); +}); + +test("sem repo na metadata → 400 (delegação exige repo); skill não-conductor → 400", async () => { + await enableA2A(); + const noRepo = await tasksRoute.POST( + delegationRequest({ skill: "conductor", messages: [{ role: "user", content: "p" }] }) + ); + assert.equal(noRepo.status, 400); + const wrongSkill = await tasksRoute.POST( + delegationRequest({ ...VALID_BODY, skill: "smart-routing" }) + ); + assert.equal(wrongSkill.status, 400); +}); diff --git a/tests/unit/conductor-delegate.test.ts b/tests/unit/conductor-delegate.test.ts new file mode 100644 index 0000000000..d356f5110d --- /dev/null +++ b/tests/unit/conductor-delegate.test.ts @@ -0,0 +1,71 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { createConductorTask } from "../../src/lib/conductor/hubProxy.ts"; + +function fakeHub(body: unknown, status = 201) { + const calls: { url: string; method: string; auth: string | null; body: unknown }[] = []; + const impl = (async (url: string | URL | Request, init?: RequestInit) => { + calls.push({ + url: String(url), + method: init?.method ?? "GET", + auth: (init?.headers as Record | undefined)?.authorization ?? null, + body: JSON.parse(String(init?.body ?? "{}")), + }); + return new Response(JSON.stringify(body), { status }); + }) as typeof fetch; + return { impl, calls }; +} + +test.beforeEach(() => { + process.env.CONDUCTOR_HUB_URL = "http://hub.test:7910"; + process.env.CONDUCTOR_HUB_TOKEN = "tok-hub"; + delete process.env.CONDUCTOR_ORCHESTRATOR_TOKEN; +}); + +test.after(() => { + delete process.env.CONDUCTOR_HUB_URL; + delete process.env.CONDUCTOR_HUB_TOKEN; + delete process.env.CONDUCTOR_ORCHESTRATOR_TOKEN; +}); + +test("traduz a delegação A2A no POST /v1/tasks do hub (shape exato do contrato)", async () => { + const { impl, calls } = fakeHub({ id: "t_novo", status: "submitted" }); + const r = await createConductorTask( + { repoUrl: "https://git.x/repo", baseRef: "dev", prompt: "adicione um README", mode: "council-3", cli: "claude", model: "cc/claude-sonnet-5" }, + { fetchImpl: impl } + ); + assert.deepEqual(r, { ok: true, status: 201, task_id: "t_novo" }); + assert.equal(calls[0].url, "http://hub.test:7910/v1/tasks"); + assert.equal(calls[0].method, "POST"); + assert.deepEqual(calls[0].body, { + repo: { url: "https://git.x/repo", base_ref: "dev" }, + spec: { prompt: "adicione um README" }, + mode: "council-3", + requirements: { cli: "claude", model: "cc/claude-sonnet-5" }, + }); +}); + +test("defaults: base_ref main, mode solo, sem requirements quando cli/model ausentes", async () => { + const { impl, calls } = fakeHub({ id: "t_d", status: "submitted" }); + await createConductorTask({ repoUrl: "https://git.x/r", prompt: "p" }, { fetchImpl: impl }); + assert.deepEqual(calls[0].body, { repo: { url: "https://git.x/r", base_ref: "main" }, spec: { prompt: "p" }, mode: "solo" }); +}); + +test("credencial: prefere CONDUCTOR_ORCHESTRATOR_TOKEN; fallback é o token do hub", async () => { + const a = fakeHub({ id: "t_1" }); + await createConductorTask({ repoUrl: "https://x/r", prompt: "p" }, { fetchImpl: a.impl }); + assert.equal(a.calls[0].auth, "Bearer tok-hub"); + process.env.CONDUCTOR_ORCHESTRATOR_TOKEN = "tok-orch"; + const b = fakeHub({ id: "t_2" }); + await createConductorTask({ repoUrl: "https://x/r", prompt: "p" }, { fetchImpl: b.impl }); + assert.equal(b.calls[0].auth, "Bearer tok-orch"); +}); + +test("recusa do hub → {ok:false, status} sem lançar nem vazar corpo; env ausente → 503", async () => { + const { impl } = fakeHub({ error: "segredo do hub" }, 422); + const r = await createConductorTask({ repoUrl: "https://x/r", prompt: "p" }, { fetchImpl: impl }); + assert.deepEqual(r, { ok: false, status: 422 }); + delete process.env.CONDUCTOR_HUB_URL; + assert.deepEqual(await createConductorTask({ repoUrl: "https://x/r", prompt: "p" }, {}), { ok: false, status: 503 }); +});