From 301a881903cbbec6ec9233c1deee5d932836f485 Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Tue, 1 Sep 2026 23:19:32 -0300 Subject: [PATCH 1/4] feat(dashboard): agents WS channel + agent.task.updated event --- src/lib/events/types.ts | 19 +++++++++++++++++-- src/server/ws/types.ts | 6 +++--- tests/unit/ws-agents-channel.test.ts | 20 ++++++++++++++++++++ 3 files changed, 40 insertions(+), 5 deletions(-) create mode 100644 tests/unit/ws-agents-channel.test.ts diff --git a/src/lib/events/types.ts b/src/lib/events/types.ts index 874278e93ae..bfb5b2df2c2 100644 --- a/src/lib/events/types.ts +++ b/src/lib/events/types.ts @@ -17,7 +17,8 @@ export type DashboardEventName = | "combo.target.succeeded" | "credential.health.changed" | "compression.completed" - | "compression.step"; + | "compression.step" + | "agent.task.updated"; // ── Event Payloads ──────────────────────────────────────────────────────── @@ -138,6 +139,18 @@ export interface CompressionStepPayload { timestamp: number; } +/** + * Real-time transition of an orchestrated task tracked by the orchestration dashboard + * (Fase 2). `source` distinguishes which subsystem raised the transition so the canvas can + * route it to the right node type. + */ +export interface AgentTaskUpdatedPayload { + source: "cloud-agent" | "a2a" | "conductor"; + taskId: string; + state: string; + timestamp: number; +} + // ── Event Map ───────────────────────────────────────────────────────────── export interface DashboardEventMap { @@ -151,6 +164,7 @@ export interface DashboardEventMap { "credential.health.changed": CredentialHealthChangedPayload; "compression.completed": CompressionCompletedPayload; "compression.step": CompressionStepPayload; + "agent.task.updated": AgentTaskUpdatedPayload; } // ── Event Bus Listener ──────────────────────────────────────────────────── @@ -162,7 +176,7 @@ export type DashboardEventListener = ( // ── Channel Definitions ─────────────────────────────────────────────────── /** Available subscription channels */ -export type DashboardChannel = "requests" | "combo" | "credentials" | "compression"; +export type DashboardChannel = "requests" | "combo" | "credentials" | "compression" | "agents"; /** Map channels to their events */ export const CHANNEL_EVENTS: Record = { @@ -170,6 +184,7 @@ export const CHANNEL_EVENTS: Record = { combo: ["combo.target.attempt", "combo.target.failed", "combo.target.succeeded"], credentials: ["credential.health.changed"], compression: ["compression.completed", "compression.step"], + agents: ["agent.task.updated"], }; /** Get channel for an event */ diff --git a/src/server/ws/types.ts b/src/server/ws/types.ts index 2fa678e7680..d4728dd21b7 100644 --- a/src/server/ws/types.ts +++ b/src/server/ws/types.ts @@ -8,7 +8,7 @@ export interface WsSubscribeMessage { type: "subscribe"; - channels: Array<"requests" | "combo" | "credentials" | "compression">; + channels: Array<"requests" | "combo" | "credentials" | "compression" | "agents">; } export interface WsPingMessage { @@ -21,7 +21,7 @@ export type WsClientMessage = WsSubscribeMessage | WsPingMessage; export interface WsEventMessage { type: "event"; - channel: "requests" | "combo" | "credentials" | "compression"; + channel: "requests" | "combo" | "credentials" | "compression" | "agents"; event: string; data: unknown; } @@ -35,7 +35,7 @@ export interface WsWelcomeMessage { version: string; sessionId: string; serverTime: number; - channels: Array<"requests" | "combo" | "credentials" | "compression">; + channels: Array<"requests" | "combo" | "credentials" | "compression" | "agents">; /** Number of buffered events since last reconnect */ backlog: number; } diff --git a/tests/unit/ws-agents-channel.test.ts b/tests/unit/ws-agents-channel.test.ts new file mode 100644 index 00000000000..8b908c60173 --- /dev/null +++ b/tests/unit/ws-agents-channel.test.ts @@ -0,0 +1,20 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { CHANNEL_EVENTS, getChannelForEvent } from "../../src/lib/events/types.ts"; +import { emit, on } from "../../src/lib/events/eventBus.ts"; + +describe("agents WS channel (B1)", () => { + it("agents channel maps agent.task.updated", () => { + assert.deepEqual(CHANNEL_EVENTS.agents, ["agent.task.updated"]); + assert.equal(getChannelForEvent("agent.task.updated"), "agents"); + }); + + it("emit/on round-trip", () => { + const seen: unknown[] = []; + const off = on("agent.task.updated", (p) => seen.push(p)); + emit("agent.task.updated", { source: "a2a", taskId: "t1", state: "working", timestamp: 1 }); + off(); + assert.equal(seen.length, 1); + }); +}); From 2d3bfd746c11ba3ee0d8f508fe832c55deddca23 Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Tue, 1 Sep 2026 23:26:11 -0300 Subject: [PATCH 2/4] feat(a2a): publish agent.task.updated on cloud-agent and a2a task writes --- src/lib/a2a/taskManager.ts | 17 ++ src/lib/cloudAgent/db.ts | 15 ++ tests/unit/agents-channel-publish.test.ts | 239 ++++++++++++++++++++++ 3 files changed, 271 insertions(+) create mode 100644 tests/unit/agents-channel-publish.test.ts diff --git a/src/lib/a2a/taskManager.ts b/src/lib/a2a/taskManager.ts index a21ac572076..d8e6f70346c 100644 --- a/src/lib/a2a/taskManager.ts +++ b/src/lib/a2a/taskManager.ts @@ -13,6 +13,20 @@ import { randomUUID } from "crypto"; +import { emit } from "@/lib/events/eventBus"; + +/** + * Publish an `agent.task.updated` transition for the orchestration canvas (Fase 2, Task B2). + * Best-effort: a listener throwing must never break the task write path that triggered it. + */ +function emitAgentTaskUpdated(source: "cloud-agent" | "a2a", taskId: string, state: string): void { + try { + emit("agent.task.updated", { source, taskId, state, timestamp: Date.now() }); + } catch { + /* listeners never derail the write path */ + } +} + // ============ Types ============ export type TaskState = "submitted" | "working" | "completed" | "failed" | "cancelled"; @@ -114,6 +128,7 @@ export class A2ATaskManager { ...(owner !== undefined ? { owner } : {}), }; this.tasks.set(task.id, task); + emitAgentTaskUpdated("a2a", task.id, "submitted"); return task; } @@ -158,6 +173,7 @@ export class A2ATaskManager { task.events.push({ timestamp: now, state, message }); if (artifacts) task.artifacts.push(...artifacts); + emitAgentTaskUpdated("a2a", taskId, state); return task; } @@ -243,6 +259,7 @@ export class A2ATaskManager { task.state = "failed"; task.updatedAt = now.toISOString(); task.events.push({ timestamp: now.toISOString(), state: "failed", message: "TTL expired" }); + emitAgentTaskUpdated("a2a", id, "failed"); } // Remove terminal tasks older than 2x TTL if ( diff --git a/src/lib/cloudAgent/db.ts b/src/lib/cloudAgent/db.ts index 91667f54220..9d7f539078d 100644 --- a/src/lib/cloudAgent/db.ts +++ b/src/lib/cloudAgent/db.ts @@ -1,4 +1,17 @@ import { getDbInstance } from "@/lib/db/core.ts"; +import { emit } from "@/lib/events/eventBus"; + +/** + * Publish an `agent.task.updated` transition for the orchestration canvas (Fase 2, Task B2). + * Best-effort: a listener throwing must never break the DB write path that triggered it. + */ +function emitAgentTaskUpdated(source: "cloud-agent" | "a2a", taskId: string, state: string): void { + try { + emit("agent.task.updated", { source, taskId, state, timestamp: Date.now() }); + } catch { + /* listeners never derail the write path */ + } +} export interface CloudAgentTaskRow { id: string; @@ -66,6 +79,7 @@ export function insertCloudAgentTask(task: CloudAgentTaskRow): void { ) ` ).run(task); + emitAgentTaskUpdated("cloud-agent", task.id, task.status); } // Whitelist of allowed columns for update operations @@ -107,6 +121,7 @@ export function updateCloudAgentTask( WHERE id = @id ` ).run({ id, ...validUpdates }); + emitAgentTaskUpdated("cloud-agent", id, (validUpdates.status as string) ?? "updated"); } export function getCloudAgentTaskById(id: string): CloudAgentTaskRow | null { diff --git a/tests/unit/agents-channel-publish.test.ts b/tests/unit/agents-channel-publish.test.ts new file mode 100644 index 00000000000..3c5f64df120 --- /dev/null +++ b/tests/unit/agents-channel-publish.test.ts @@ -0,0 +1,239 @@ +/** + * Task B2 (Orchestration Canvas Fase 2): the cloud-agent and A2A task writers must publish + * `agent.task.updated` on every write, best-effort (a throwing listener must never break the + * write path). Covers: + * (a) A2ATaskManager.createTask / updateTask / cleanupExpired (TTL branch) + * (b) cloud-agent insertCloudAgentTask / updateCloudAgentTask + * (c) a throwing listener does not break the write path + */ +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 { on } from "../../src/lib/events/eventBus.ts"; +import type { AgentTaskUpdatedPayload } from "../../src/lib/events/types.ts"; +import { A2ATaskManager } from "../../src/lib/a2a/taskManager.ts"; + +// ── DB test hygiene (AGENTS.md "PII & Stream Sanitization Learnings" §3): temp DATA_DIR set +// BEFORE importing src/lib/db/core.ts (SQLITE_FILE is resolved from DATA_DIR at import time), +// resetDbInstance()+rm the temp dir in test.after so the node:test runner does not hang. +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-agents-channel-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../src/lib/db/core.ts"); +const cloudAgentDb = await import("../../src/lib/cloudAgent/db.ts"); + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); +}); + +// ── (a) A2ATaskManager ─────────────────────────────────────────────────────────────────── + +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(); + } +}); + +test("A2ATaskManager.createTask emits agent.task.updated {source: a2a, state: submitted}", () => { + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + const tm = createManager(); + const task = tm.createTask({ + skill: "smart-routing", + messages: [{ role: "user", content: "hello" }], + }); + + assert.equal(events.length, 1); + assert.equal(events[0].source, "a2a"); + assert.equal(events[0].taskId, task.id); + assert.equal(events[0].state, "submitted"); + assert.equal(typeof events[0].timestamp, "number"); + } finally { + unsubscribe(); + } +}); + +test("A2ATaskManager.updateTask emits agent.task.updated with the new state", () => { + const tm = createManager(); + const task = tm.createTask({ + skill: "smart-routing", + messages: [{ role: "user", content: "hello" }], + }); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + tm.updateTask(task.id, "working"); + + assert.equal(events.length, 1); + assert.equal(events[0].source, "a2a"); + assert.equal(events[0].taskId, task.id); + assert.equal(events[0].state, "working"); + } finally { + unsubscribe(); + } +}); + +test("A2ATaskManager.cleanupExpired emits agent.task.updated {state: failed} on TTL expiry", () => { + const tm = createManager(); + const task = tm.createTask({ + skill: "smart-routing", + messages: [{ role: "user", content: "hello" }], + }); + task.expiresAt = new Date(Date.now() - 1_000).toISOString(); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + // private in TS only; callable at runtime for regression test (matches + // tests/unit/t09-a2a-lifecycle.test.ts precedent). + (tm as unknown as { cleanupExpired(): void }).cleanupExpired(); + + assert.equal(events.length, 1); + assert.equal(events[0].source, "a2a"); + assert.equal(events[0].taskId, task.id); + assert.equal(events[0].state, "failed"); + } finally { + unsubscribe(); + } +}); + +test("a throwing agent.task.updated listener does not break A2ATaskManager.createTask", () => { + const unsubscribe = on("agent.task.updated", () => { + throw new Error("listener boom"); + }); + try { + const tm = createManager(); + let task: ReturnType | undefined; + assert.doesNotThrow(() => { + task = tm.createTask({ + skill: "smart-routing", + messages: [{ role: "user", content: "hello" }], + }); + }); + assert.ok(task); + assert.equal(tm.getTask(task!.id)?.id, task!.id); + } finally { + unsubscribe(); + } +}); + +// ── (b) cloud-agent DB writers ────────────────────────────────────────────────────────── + +function makeTaskRow(overrides: Partial[0]> = {}) { + const now = new Date().toISOString(); + return { + id: `task-${Math.random().toString(36).slice(2)}`, + provider_id: "codex-cloud", + external_id: null, + status: "queued", + prompt: "do something", + source: "dashboard", + options: "{}", + result: null, + activities: "[]", + error: null, + created_at: now, + updated_at: now, + completed_at: null, + ...overrides, + }; +} + +test.beforeEach(() => { + core.resetDbInstance(); + cloudAgentDb.createCloudAgentTaskTable(); +}); + +test("insertCloudAgentTask emits agent.task.updated {source: cloud-agent, state: queued}", () => { + const row = makeTaskRow({ status: "queued" }); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + cloudAgentDb.insertCloudAgentTask(row); + + assert.equal(events.length, 1); + assert.equal(events[0].source, "cloud-agent"); + assert.equal(events[0].taskId, row.id); + assert.equal(events[0].state, "queued"); + } finally { + unsubscribe(); + } +}); + +test("updateCloudAgentTask emits agent.task.updated with the new status", () => { + const row = makeTaskRow({ status: "queued" }); + cloudAgentDb.insertCloudAgentTask(row); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + cloudAgentDb.updateCloudAgentTask(row.id, { status: "running" }); + + assert.equal(events.length, 1); + assert.equal(events[0].source, "cloud-agent"); + assert.equal(events[0].taskId, row.id); + assert.equal(events[0].state, "running"); + } finally { + unsubscribe(); + } +}); + +test("updateCloudAgentTask without a status field emits state 'updated'", () => { + const row = makeTaskRow({ status: "queued" }); + cloudAgentDb.insertCloudAgentTask(row); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + cloudAgentDb.updateCloudAgentTask(row.id, { result: "partial output" }); + + assert.equal(events.length, 1); + assert.equal(events[0].source, "cloud-agent"); + assert.equal(events[0].taskId, row.id); + assert.equal(events[0].state, "updated"); + } finally { + unsubscribe(); + } +}); + +test("updateCloudAgentTask with no valid fields does not emit (no-op write)", () => { + const row = makeTaskRow({ status: "queued" }); + cloudAgentDb.insertCloudAgentTask(row); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + cloudAgentDb.updateCloudAgentTask(row.id, {}); + + assert.equal(events.length, 0); + } finally { + unsubscribe(); + } +}); + +test("a throwing agent.task.updated listener does not break insertCloudAgentTask", () => { + const row = makeTaskRow({ status: "queued" }); + const unsubscribe = on("agent.task.updated", () => { + throw new Error("listener boom"); + }); + try { + assert.doesNotThrow(() => cloudAgentDb.insertCloudAgentTask(row)); + assert.equal(cloudAgentDb.getCloudAgentTaskById(row.id)?.id, row.id); + } finally { + unsubscribe(); + } +}); From 7221c345cf4d753d01379553130c3104619ff823 Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Tue, 1 Sep 2026 23:38:42 -0300 Subject: [PATCH 3/4] feat(dashboard): mirror conductor fleet transitions into agents channel --- src/lib/conductor/hubProxy.ts | 40 ++++++ tests/unit/conductor-fleet-mirror.test.ts | 154 ++++++++++++++++++++++ 2 files changed, 194 insertions(+) create mode 100644 tests/unit/conductor-fleet-mirror.test.ts diff --git a/src/lib/conductor/hubProxy.ts b/src/lib/conductor/hubProxy.ts index e9b8dd7e736..8992aa92571 100644 --- a/src/lib/conductor/hubProxy.ts +++ b/src/lib/conductor/hubProxy.ts @@ -9,6 +9,8 @@ import { z } from "zod"; +import { emit } from "@/lib/events/eventBus"; + // ============ Whitelisted client-facing shapes ============ export interface FleetRunner { @@ -112,6 +114,43 @@ function toFleetTask(t: z.infer): FleetTask { }; } +// ============ Fleet task mirror (Orchestration Canvas Fase 2, Task B3) ============ +// +// Module-level cache of the last known status per fleet task, so `getFleetSnapshot` can +// diff-on-fetch and mirror Conductor task transitions into the `agents` WS channel without a +// dedicated poller — it piggybacks on the dashboard's existing poll. `null` means "no snapshot +// observed yet" (first-ever call): that call only seeds the cache, it never emits, since the +// dashboard already fetches the full snapshot on its initial poll. An offline snapshot never +// touches this cache (see call site below), so a hub flap does not cause a re-seed burst once +// the hub comes back — only the real delta since the last successful snapshot is emitted. +let lastFleetTaskStates: Map | null = null; + +function emitFleetTransitions(tasks: FleetTask[]): void { + const next = new Map(tasks.map((t) => [t.id, t.status])); + if (lastFleetTaskStates) { + for (const [id, status] of next) { + if (lastFleetTaskStates.get(id) !== status) { + try { + emit("agent.task.updated", { + source: "conductor", + taskId: id, + state: status, + timestamp: Date.now(), + }); + } catch { + /* best-effort */ + } + } + } + } + lastFleetTaskStates = next; +} + +/** Test-only seam: resets the fleet task mirror cache so tests are order-independent. */ +export function __resetFleetMirrorForTests(): void { + lastFleetTaskStates = null; +} + /** Fleet snapshot for the dashboard panel. Degraded ({offline: true}) on any failure. */ export async function getFleetSnapshot(opts: HubProxyOptions = {}): Promise { try { @@ -128,6 +167,7 @@ export async function getFleetSnapshot(opts: HubProxyOptions = {}): Promise) { + const impl = (async (url: string | URL | Request) => { + const u = String(url); + 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; +} + +function snapshotWith(tasks: ReturnType[]) { + return fakeHub({ + "/v1/runners": { status: 200, body: [] }, + "/v1/tasks": { status: 200, body: tasks }, + }); +} + +const offlineFetch = (async () => { + throw new Error("ECONNREFUSED"); +}) as unknown as typeof fetch; + +test.beforeEach(() => { + process.env.CONDUCTOR_HUB_URL = "http://hub.test:7910"; + process.env.CONDUCTOR_HUB_TOKEN = "tok-secreto"; + __resetFleetMirrorForTests(); +}); + +test.after(() => { + delete process.env.CONDUCTOR_HUB_URL; + delete process.env.CONDUCTOR_HUB_TOKEN; + __resetFleetMirrorForTests(); +}); + +test("fleet mirror: 1a foto semeia o cache sem emitir nada", async () => { + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + const snap = await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "working"), hubTask("t_2", "queued")]), + }); + assert.equal(snap.offline, false); + assert.equal(events.length, 0, "primeira foto (cache null) só semeia, não emite"); + } finally { + unsubscribe(); + } +}); + +test("fleet mirror: status mudado emite exatamente 1 evento agent.task.updated", async () => { + await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "working"), hubTask("t_2", "queued")]), + }); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "completed"), hubTask("t_2", "queued")]), + }); + assert.equal(events.length, 1); + assert.deepEqual(events[0], { + source: "conductor", + taskId: "t_1", + state: "completed", + timestamp: events[0].timestamp, + }); + assert.equal(typeof events[0].timestamp, "number"); + } finally { + unsubscribe(); + } +}); + +test("fleet mirror: foto idêntica à anterior não emite nada", async () => { + await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "working"), hubTask("t_2", "queued")]), + }); + await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "completed"), hubTask("t_2", "queued")]), + }); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "completed"), hubTask("t_2", "queued")]), + }); + assert.equal(events.length, 0); + } finally { + unsubscribe(); + } +}); + +test("fleet mirror: foto offline entre duas fotos não zera o cache — próxima foto só emite o delta real", async () => { + // Seed. + await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "working"), hubTask("t_2", "queued")]), + }); + // Establish a known baseline (t_1 -> completed). + await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "completed"), hubTask("t_2", "queued")]), + }); + + // Hub flaps offline in between — must not clear the cache. + const offlineSnap = await getFleetSnapshot({ fetchImpl: offlineFetch }); + assert.deepEqual(offlineSnap, { offline: true, runners: [], tasks: [] }); + + const events: AgentTaskUpdatedPayload[] = []; + const unsubscribe = on("agent.task.updated", (payload) => events.push(payload)); + try { + // Back online: only t_2 actually changed since the last successful snapshot (t_1 unchanged). + await getFleetSnapshot({ + fetchImpl: snapshotWith([hubTask("t_1", "completed"), hubTask("t_2", "completed")]), + }); + assert.equal(events.length, 1, "só o delta real (t_2) deve emitir, não um re-seed de tudo"); + assert.equal(events[0].taskId, "t_2"); + assert.equal(events[0].state, "completed"); + assert.equal(events[0].source, "conductor"); + } finally { + unsubscribe(); + } +}); From e7c68202fb3cdc86a6450f58bbd2997bcfab9d2a Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Wed, 2 Sep 2026 00:07:11 -0300 Subject: [PATCH 4/4] feat(dashboard): orchestration snapshot rides the agents WS channel (2.1) --- .../features/orchestration-agents-ws.md | 5 + .../hooks/useOrchestrationSnapshot.ts | 26 +++- .../unit/ui/useOrchestrationSnapshot.test.tsx | 132 ++++++++++++++++-- 3 files changed, 147 insertions(+), 16 deletions(-) create mode 100644 changelog.d/features/orchestration-agents-ws.md diff --git a/changelog.d/features/orchestration-agents-ws.md b/changelog.d/features/orchestration-agents-ws.md new file mode 100644 index 00000000000..737b072be10 --- /dev/null +++ b/changelog.d/features/orchestration-agents-ws.md @@ -0,0 +1,5 @@ +- **feat(dashboard):** the `/dashboard/orchestration` snapshot hook now subscribes to the + `agents` WebSocket channel (`agent.task.updated`) instead of `requests` as its refetch + trigger, and relaxes its background poll from 5s to 30s while that WS connection is up — + falling back to the tighter 5s cadence, reprogrammed live on any connect/disconnect + transition, whenever the socket is down. diff --git a/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts b/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts index fea94f73d15..5fc2182a93e 100644 --- a/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts +++ b/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts @@ -1,5 +1,9 @@ "use client"; -/** Polls the 3 agent sources (allSettled), listens to the `requests` WS channel as a refetch trigger. */ +/** + * Polls the 3 agent sources (allSettled), listens to the `agents` WS channel as a refetch + * trigger, and relaxes the poll interval from 5s to 30s while that WS connection is up (the + * channel event still forces an immediate debounced refetch either way). + */ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useLiveDashboard } from "@/hooks/useLiveDashboard"; import type { CloudAgentTask } from "@/lib/cloudAgent/types"; @@ -12,6 +16,7 @@ import { mergeSnapshot } from "../model/mergeSnapshot"; import type { OrchSnapshot, SourceStatus } from "../model/orchestrationTypes"; export const POLL_MS = 5_000; +export const POLL_MS_WS_CONNECTED = 30_000; export const WS_REFETCH_DEBOUNCE_MS = 1_000; interface Raw { @@ -135,9 +140,7 @@ export function useOrchestrationSnapshot() { pollRef.current = () => void poll(); void poll(); - const id = setInterval(() => void poll(), POLL_MS); return () => { - clearInterval(id); controller.abort(); if (debounceRef.current) { clearTimeout(debounceRef.current); @@ -150,10 +153,10 @@ export function useOrchestrationSnapshot() { pollRef.current(); }, []); - useLiveDashboard({ - channels: ["requests"], + const { connection } = useLiveDashboard({ + channels: ["agents"], onEvent: (payload) => { - if (payload.channel !== "requests") return; + if (payload.channel !== "agents") return; if (debounceRef.current) return; // debounce burst → one refetch debounceRef.current = setTimeout(() => { debounceRef.current = null; @@ -162,6 +165,17 @@ export function useOrchestrationSnapshot() { }, }); + // Adaptive poll interval, separated from the mount effect above: a connected `agents` WS + // already pushes refetches on change, so the background poll only needs to be a slow safety + // net (30s) — it falls back to the tighter 5s cadence while the WS is down. Declared AFTER + // the mount effect so `pollRef.current` is already populated (its initial value is a safe + // no-op) by the time this effect's first tick can fire. + const wsConnected = connection.isConnected; + useEffect(() => { + const id = setInterval(() => pollRef.current(), wsConnected ? POLL_MS_WS_CONNECTED : POLL_MS); + return () => clearInterval(id); + }, [wsConnected]); + const snapshot: OrchSnapshot = useMemo( () => mergeSnapshot( diff --git a/tests/unit/ui/useOrchestrationSnapshot.test.tsx b/tests/unit/ui/useOrchestrationSnapshot.test.tsx index 5160f8039dc..549d7ca1eff 100644 --- a/tests/unit/ui/useOrchestrationSnapshot.test.tsx +++ b/tests/unit/ui/useOrchestrationSnapshot.test.tsx @@ -3,12 +3,14 @@ import React, { act } from "react"; import { createRoot } from "react-dom/client"; import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; -// Capture the onEvent handler the hook registers on the requests channel. +// Capture the onEvent handler the hook registers on the agents channel, and let each test +// control the mocked WS connection state that drives the adaptive poll interval. let capturedOnEvent: ((p: { channel: string }) => void) | null = null; +const connectionState: { isConnected: boolean } = { isConnected: false }; vi.mock("@/hooks/useLiveDashboard", () => ({ useLiveDashboard: (opts: { onEvent?: (p: { channel: string }) => void }) => { capturedOnEvent = opts.onEvent ?? null; - return { connection: { isConnected: true }, events: [] }; + return { connection: connectionState, events: [] }; }, })); @@ -26,11 +28,21 @@ function HookProbe({ const okJson = (body: unknown) => Promise.resolve({ ok: true, json: () => Promise.resolve(body) } as Response); +const okFetchMock = () => + vi.fn((url: string) => { + if (url.startsWith("/api/v1/agents/tasks")) return okJson({ data: [] }); + if (url.startsWith("/api/a2a/tasks")) + return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); + return okJson({ offline: false, runners: [], tasks: [] }); + }); + describe("useOrchestrationSnapshot", () => { let container: HTMLDivElement; let root: ReturnType; beforeEach(() => { vi.useFakeTimers(); + connectionState.isConnected = false; + capturedOnEvent = null; container = document.createElement("div"); document.body.appendChild(container); root = createRoot(container); @@ -92,13 +104,30 @@ describe("useOrchestrationSnapshot", () => { expect(st?.ok).toBe(false); }); - it("a requests-channel WS event triggers a debounced immediate refetch", async () => { - const fetchMock = vi.fn((url: string) => { - if (url.startsWith("/api/v1/agents/tasks")) return okJson({ data: [] }); - if (url.startsWith("/api/a2a/tasks")) - return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); - return okJson({ offline: false, runners: [], tasks: [] }); + it("an agents-channel WS event triggers a debounced immediate refetch", async () => { + const fetchMock = okFetchMock(); + vi.stubGlobal("fetch", fetchMock); + await act(async () => { + root.render( {}} />); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(10); + }); + const callsAfterMount = fetchMock.mock.calls.length; + + act(() => { + capturedOnEvent?.({ channel: "agents" }); + capturedOnEvent?.({ channel: "agents" }); }); + await act(async () => { + await vi.advanceTimersByTimeAsync(1_100); + }); + // Two burst events → exactly ONE extra round of 3 fetches (debounce), not two. + expect(fetchMock.mock.calls.length).toBe(callsAfterMount + 3); + }); + + it("a WS event on a different channel does not trigger a refetch", async () => { + const fetchMock = okFetchMock(); vi.stubGlobal("fetch", fetchMock); await act(async () => { root.render( {}} />); @@ -110,12 +139,95 @@ describe("useOrchestrationSnapshot", () => { act(() => { capturedOnEvent?.({ channel: "requests" }); - capturedOnEvent?.({ channel: "requests" }); }); await act(async () => { await vi.advanceTimersByTimeAsync(1_100); }); - // Two burst events → exactly ONE extra round of 3 fetches (debounce), not two. + expect(fetchMock.mock.calls.length).toBe(callsAfterMount); + }); + + it("polls every 30s while the WS connection is up", async () => { + connectionState.isConnected = true; + const fetchMock = okFetchMock(); + vi.stubGlobal("fetch", fetchMock); + await act(async () => { + root.render( {}} />); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(10); + }); + const callsAfterMount = fetchMock.mock.calls.length; + + await act(async () => { + await vi.advanceTimersByTimeAsync(29_000); + }); + expect(fetchMock.mock.calls.length).toBe(callsAfterMount); + + await act(async () => { + await vi.advanceTimersByTimeAsync(1_000); + }); + expect(fetchMock.mock.calls.length).toBe(callsAfterMount + 3); + }); + + it("polls every 5s while the WS connection is down", async () => { + connectionState.isConnected = false; + const fetchMock = okFetchMock(); + vi.stubGlobal("fetch", fetchMock); + await act(async () => { + root.render( {}} />); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(10); + }); + const callsAfterMount = fetchMock.mock.calls.length; + + await act(async () => { + await vi.advanceTimersByTimeAsync(4_900); + }); + expect(fetchMock.mock.calls.length).toBe(callsAfterMount); + + await act(async () => { + await vi.advanceTimersByTimeAsync(200); + }); + expect(fetchMock.mock.calls.length).toBe(callsAfterMount + 3); + }); + + it("reprograms the interval when the WS connection transitions from connected to disconnected", async () => { + connectionState.isConnected = true; + let latest: ReturnType | null = null; + const onRender = (v: ReturnType) => { + latest = v; + }; + const fetchMock = okFetchMock(); + vi.stubGlobal("fetch", fetchMock); + await act(async () => { + root.render(); + }); + await act(async () => { + await vi.advanceTimersByTimeAsync(10); + }); + const callsAfterMount = fetchMock.mock.calls.length; + + // 10s into the 30s (connected) cycle — no extra poll yet. + await act(async () => { + await vi.advanceTimersByTimeAsync(10_000); + }); + expect(fetchMock.mock.calls.length).toBe(callsAfterMount); + + // Connection drops — force a re-render so the hook observes the new value and + // reprograms its interval effect (deps: [wsConnected]). + connectionState.isConnected = false; + await act(async () => { + root.render(); + }); + void latest; // keep the probe referenced + + // The old 30s timer would not have fired yet at the 30s mark either way, but the + // reprogrammed 5s timer must fire on its OWN schedule, starting from the flip — + // i.e. 5.1s after the flip (well before the original 30s mark at t=30s). + await act(async () => { + await vi.advanceTimersByTimeAsync(5_100); + }); expect(fetchMock.mock.calls.length).toBe(callsAfterMount + 3); });