mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-07-31 04:12:10 +03:00
feat(dashboard): server-side hub proxy for the Conductor panel (whitelisted shapes, fail-open)
This commit is contained in:
176
src/lib/conductor/hubProxy.ts
Normal file
176
src/lib/conductor/hubProxy.ts
Normal file
@@ -0,0 +1,176 @@
|
||||
/**
|
||||
* Server-side proxy to the OmniConductor hub (Conductor PRD RF3).
|
||||
*
|
||||
* The browser NEVER talks to the hub: these helpers run only in API routes,
|
||||
* authenticate with the server-side env token, and return WHITELISTED shapes —
|
||||
* runner/hub tokens can never leak to the client. Fail-open: hub unset/offline
|
||||
* yields a degraded snapshot ({offline: true}) instead of an error.
|
||||
*/
|
||||
|
||||
import { z } from "zod";
|
||||
|
||||
// ============ Whitelisted client-facing shapes ============
|
||||
|
||||
export interface FleetRunner {
|
||||
id: string;
|
||||
name: string;
|
||||
clis: string[];
|
||||
online: boolean;
|
||||
draining: boolean;
|
||||
}
|
||||
|
||||
export interface FleetTask {
|
||||
id: string;
|
||||
status: string;
|
||||
mode: string;
|
||||
repo: string | null;
|
||||
runner: string | null;
|
||||
summary: string | null;
|
||||
branch: string | null;
|
||||
error: string | null;
|
||||
updated_at: string | null;
|
||||
}
|
||||
|
||||
export interface FleetSnapshot {
|
||||
offline: boolean;
|
||||
runners: FleetRunner[];
|
||||
tasks: FleetTask[];
|
||||
}
|
||||
|
||||
export interface ConductorTaskDetail extends FleetTask {
|
||||
prompt: string | null;
|
||||
base_ref: string | null;
|
||||
tests: unknown;
|
||||
council: unknown;
|
||||
created_at: string | null;
|
||||
}
|
||||
|
||||
// ============ Untrusted hub shapes (parse only what we read) ============
|
||||
|
||||
const hubRunnerSchema = z.object({
|
||||
id: z.string(),
|
||||
online: z.boolean().optional(),
|
||||
draining: z.boolean().optional(),
|
||||
capabilities: z.object({
|
||||
name: z.string().optional(),
|
||||
clis: z.array(z.object({ profile: z.string() })).optional(),
|
||||
}),
|
||||
});
|
||||
|
||||
const hubTaskSchema = z.object({
|
||||
id: z.string(),
|
||||
status: z.string(),
|
||||
mode: z.string().optional(),
|
||||
repo: z.object({ url: z.string().optional(), base_ref: z.string().optional() }).nullish(),
|
||||
spec: z.object({ prompt: z.string().optional() }).nullish(),
|
||||
assigned_runner: z.string().nullish(),
|
||||
manifest: z
|
||||
.object({
|
||||
summary: z.string().nullish(),
|
||||
branch: z.string().nullish(),
|
||||
error: z.string().nullish(),
|
||||
tests: z.unknown().optional(),
|
||||
})
|
||||
.nullish(),
|
||||
council: z.unknown().optional(),
|
||||
created_at: z.string().optional(),
|
||||
updated_at: z.string().optional(),
|
||||
});
|
||||
|
||||
export interface HubProxyOptions {
|
||||
fetchImpl?: typeof fetch;
|
||||
}
|
||||
|
||||
function hubConfig(): { url: string; token: string } | null {
|
||||
const url = process.env.CONDUCTOR_HUB_URL?.trim();
|
||||
if (!url) return null;
|
||||
return { url, token: process.env.CONDUCTOR_HUB_TOKEN?.trim() ?? "" };
|
||||
}
|
||||
|
||||
async function hubGet(path: string, opts: HubProxyOptions): Promise<unknown | null> {
|
||||
const cfg = hubConfig();
|
||||
if (!cfg) return null;
|
||||
const doFetch = opts.fetchImpl ?? fetch;
|
||||
const res = await doFetch(`${cfg.url}${path}`, {
|
||||
headers: { authorization: `Bearer ${cfg.token}` },
|
||||
});
|
||||
if (!res.ok) return null;
|
||||
return res.json();
|
||||
}
|
||||
|
||||
function toFleetTask(t: z.infer<typeof hubTaskSchema>): FleetTask {
|
||||
return {
|
||||
id: t.id,
|
||||
status: t.status,
|
||||
mode: t.mode ?? "solo",
|
||||
repo: t.repo?.url ?? null,
|
||||
runner: t.assigned_runner ?? null,
|
||||
summary: t.manifest?.summary ?? null,
|
||||
branch: t.manifest?.branch ?? null,
|
||||
error: t.manifest?.error ?? null,
|
||||
updated_at: t.updated_at ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
/** Fleet snapshot for the dashboard panel. Degraded ({offline: true}) on any failure. */
|
||||
export async function getFleetSnapshot(opts: HubProxyOptions = {}): Promise<FleetSnapshot> {
|
||||
try {
|
||||
const [rawRunners, rawTasks] = await Promise.all([
|
||||
hubGet("/v1/runners", opts),
|
||||
hubGet("/v1/tasks", opts),
|
||||
]);
|
||||
if (rawRunners === null || rawTasks === null) return { offline: true, runners: [], tasks: [] };
|
||||
const runners = z.array(hubRunnerSchema).parse(rawRunners).map((r) => ({
|
||||
id: r.id,
|
||||
name: r.capabilities.name ?? "?",
|
||||
clis: (r.capabilities.clis ?? []).map((c) => c.profile),
|
||||
online: r.online !== false,
|
||||
draining: r.draining === true,
|
||||
}));
|
||||
const tasks = z.array(hubTaskSchema).parse(rawTasks).map(toFleetTask);
|
||||
return { offline: false, runners, tasks };
|
||||
} catch {
|
||||
return { offline: true, runners: [], tasks: [] };
|
||||
}
|
||||
}
|
||||
|
||||
/** Full whitelisted detail of one task (manifest, prompt, council funnel data). */
|
||||
export async function getConductorTaskDetail(
|
||||
taskId: string,
|
||||
opts: HubProxyOptions = {}
|
||||
): Promise<ConductorTaskDetail | null> {
|
||||
try {
|
||||
const raw = await hubGet(`/v1/tasks/${encodeURIComponent(taskId)}`, opts);
|
||||
if (raw === null) return null;
|
||||
const t = hubTaskSchema.parse(raw);
|
||||
return {
|
||||
...toFleetTask(t),
|
||||
prompt: t.spec?.prompt ?? null,
|
||||
base_ref: t.repo?.base_ref ?? null,
|
||||
tests: t.manifest?.tests ?? null,
|
||||
council: t.council ?? null,
|
||||
created_at: t.created_at ?? null,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Cancels a task on the hub. Returns the hub's verdict without leaking its body on error. */
|
||||
export async function cancelConductorTask(
|
||||
taskId: string,
|
||||
opts: HubProxyOptions = {}
|
||||
): Promise<{ ok: boolean; status: number }> {
|
||||
const cfg = hubConfig();
|
||||
if (!cfg) return { ok: false, status: 503 };
|
||||
try {
|
||||
const doFetch = opts.fetchImpl ?? fetch;
|
||||
const res = await doFetch(`${cfg.url}/v1/tasks/${encodeURIComponent(taskId)}/cancel`, {
|
||||
method: "POST",
|
||||
headers: { authorization: `Bearer ${cfg.token}` },
|
||||
});
|
||||
return { ok: res.ok, status: res.status };
|
||||
} catch {
|
||||
return { ok: false, status: 503 };
|
||||
}
|
||||
}
|
||||
140
tests/unit/conductor-hub-proxy.test.ts
Normal file
140
tests/unit/conductor-hub-proxy.test.ts
Normal file
@@ -0,0 +1,140 @@
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
|
||||
import {
|
||||
getFleetSnapshot,
|
||||
getConductorTaskDetail,
|
||||
cancelConductorTask,
|
||||
} from "../../src/lib/conductor/hubProxy.ts";
|
||||
|
||||
const RUNNERS = [
|
||||
{
|
||||
id: "r_1",
|
||||
token: "NUNCA-VAZAR",
|
||||
online: true,
|
||||
draining: false,
|
||||
capabilities: { name: "devbox", clis: [{ profile: "claude" }, { profile: "codex" }], skills: [] },
|
||||
},
|
||||
];
|
||||
|
||||
const TASKS = [
|
||||
{
|
||||
id: "t_1",
|
||||
status: "completed",
|
||||
mode: "solo",
|
||||
from: "orchestrator",
|
||||
repo: { url: "https://git.x/repo", base_ref: "main" },
|
||||
spec: { prompt: "faz algo" },
|
||||
assigned_runner: "r_1",
|
||||
manifest: { summary: "feito", branch: "task/t_1", error: null },
|
||||
council: null,
|
||||
created_at: "2026-07-22T00:00:00Z",
|
||||
updated_at: "2026-07-22T00:01:00Z",
|
||||
},
|
||||
{
|
||||
id: "t_2",
|
||||
status: "working",
|
||||
mode: "council-3",
|
||||
from: "orchestrator",
|
||||
repo: { url: "https://git.x/repo", base_ref: "main" },
|
||||
spec: { prompt: "outra" },
|
||||
assigned_runner: null,
|
||||
manifest: null,
|
||||
council: { candidate_task_ids: ["t_2a", "t_2b"] },
|
||||
created_at: "2026-07-22T00:02:00Z",
|
||||
updated_at: "2026-07-22T00:02:30Z",
|
||||
},
|
||||
];
|
||||
|
||||
function fakeHub(routes: Record<string, { status: number; body: unknown }>) {
|
||||
const calls: { url: string; method: string; auth: string | null }[] = [];
|
||||
const impl = (async (url: string | URL | Request, init?: RequestInit) => {
|
||||
const u = String(url);
|
||||
calls.push({
|
||||
url: u,
|
||||
method: init?.method ?? "GET",
|
||||
auth: (init?.headers as Record<string, string> | undefined)?.authorization ?? null,
|
||||
});
|
||||
const hit = Object.entries(routes).find(([path]) => u.includes(path));
|
||||
if (!hit) return new Response("{}", { status: 404 });
|
||||
return new Response(JSON.stringify(hit[1].body), { status: hit[1].status });
|
||||
}) as typeof fetch;
|
||||
return { impl, calls };
|
||||
}
|
||||
|
||||
test.beforeEach(() => {
|
||||
process.env.CONDUCTOR_HUB_URL = "http://hub.test:7910";
|
||||
process.env.CONDUCTOR_HUB_TOKEN = "tok-secreto";
|
||||
});
|
||||
|
||||
test.after(() => {
|
||||
delete process.env.CONDUCTOR_HUB_URL;
|
||||
delete process.env.CONDUCTOR_HUB_TOKEN;
|
||||
});
|
||||
|
||||
test("snapshot: runners e tasks sanitizados (whitelist — token do runner NUNCA passa)", async () => {
|
||||
const { impl, calls } = fakeHub({
|
||||
"/v1/runners": { status: 200, body: RUNNERS },
|
||||
"/v1/tasks": { status: 200, body: TASKS },
|
||||
});
|
||||
const snap = await getFleetSnapshot({ fetchImpl: impl });
|
||||
assert.equal(snap.offline, false);
|
||||
assert.deepEqual(snap.runners, [
|
||||
{ id: "r_1", name: "devbox", clis: ["claude", "codex"], online: true, draining: false },
|
||||
]);
|
||||
assert.equal(snap.tasks.length, 2);
|
||||
assert.deepEqual(snap.tasks[0], {
|
||||
id: "t_1",
|
||||
status: "completed",
|
||||
mode: "solo",
|
||||
repo: "https://git.x/repo",
|
||||
runner: "r_1",
|
||||
summary: "feito",
|
||||
branch: "task/t_1",
|
||||
error: null,
|
||||
updated_at: "2026-07-22T00:01:00Z",
|
||||
});
|
||||
assert.ok(!JSON.stringify(snap).includes("NUNCA-VAZAR"), "token de runner não vaza");
|
||||
assert.ok(!JSON.stringify(snap).includes("tok-secreto"), "token do hub não vaza");
|
||||
assert.equal(calls.every((c) => c.auth === "Bearer tok-secreto"), true, "proxy autentica no hub");
|
||||
});
|
||||
|
||||
test("snapshot: hub fora do ar → degradado {offline:true} sem lançar", async () => {
|
||||
const failing = (async () => {
|
||||
throw new Error("ECONNREFUSED");
|
||||
}) as unknown as typeof fetch;
|
||||
const snap = await getFleetSnapshot({ fetchImpl: failing });
|
||||
assert.deepEqual(snap, { offline: true, runners: [], tasks: [] });
|
||||
});
|
||||
|
||||
test("snapshot: env ausente → degradado sem fetch", async () => {
|
||||
delete process.env.CONDUCTOR_HUB_URL;
|
||||
const { impl, calls } = fakeHub({});
|
||||
const snap = await getFleetSnapshot({ fetchImpl: impl });
|
||||
assert.equal(snap.offline, true);
|
||||
assert.equal(calls.length, 0);
|
||||
});
|
||||
|
||||
test("detalhe: manifest e council passam; spec.prompt vem; campos fora da whitelist não", async () => {
|
||||
const { impl } = fakeHub({ "/v1/tasks/t_2": { status: 200, body: TASKS[1] } });
|
||||
const detail = await getConductorTaskDetail("t_2", { fetchImpl: impl });
|
||||
assert.ok(detail);
|
||||
assert.equal(detail!.id, "t_2");
|
||||
assert.equal(detail!.prompt, "outra");
|
||||
assert.deepEqual(detail!.council, { candidate_task_ids: ["t_2a", "t_2b"] });
|
||||
assert.equal(detail!.base_ref, "main");
|
||||
});
|
||||
|
||||
test("detalhe: 404 do hub → null", async () => {
|
||||
const { impl } = fakeHub({});
|
||||
assert.equal(await getConductorTaskDetail("t_x", { fetchImpl: impl }), null);
|
||||
});
|
||||
|
||||
test("cancelar: POST no hub e repassa o status", async () => {
|
||||
const { impl, calls } = fakeHub({ "/v1/tasks/t_1/cancel": { status: 200, body: { ok: true } } });
|
||||
const r = await cancelConductorTask("t_1", { fetchImpl: impl });
|
||||
assert.deepEqual(r, { ok: true, status: 200 });
|
||||
assert.equal(calls[0].method, "POST");
|
||||
const miss = await cancelConductorTask("t_zzz", { fetchImpl: fakeHub({}).impl });
|
||||
assert.deepEqual(miss, { ok: false, status: 404 });
|
||||
});
|
||||
Reference in New Issue
Block a user