From 08ed416c1f556cc90435b8457143a9caa237e2f1 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Mon, 2 Nov 2026 22:17:57 +0300 Subject: [PATCH] feat(studio): D15 studio event bus Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../server/__tests__/eventBus.test.ts | 52 ++++ .../server/__tests__/eventsWs.test.ts | 279 ++++++++++++++++++ .../server/__tests__/runRegistry.test.ts | 3 +- .../server/__tests__/workspaceSeam.test.ts | 2 + apps/agent-forge/server/app.ts | 33 ++- apps/agent-forge/server/events/eventBus.ts | 40 +++ apps/agent-forge/server/events/eventsWs.ts | 35 +++ apps/agent-forge/server/index.ts | 16 +- apps/agent-forge/server/routes/agentRoutes.ts | 13 +- apps/agent-forge/server/routes/draftRoutes.ts | 5 + .../server/routes/versionRoutes.ts | 3 + .../workspace/__tests__/middleware.test.ts | 2 + apps/agent-forge/server/workspace/context.ts | 8 + .../server/ws/__tests__/upgradeRouter.test.ts | 151 ++++++++++ apps/agent-forge/server/ws/upgradeRouter.ts | 86 ++++++ apps/agent-forge/server/wsServer.ts | 47 +-- apps/agent-forge/shared/wireTypes.ts | 15 + apps/agent-forge/src/runtime/runtimeClient.ts | 31 ++ .../state/__tests__/useStudioEvents.test.tsx | 125 ++++++++ .../agent-forge/src/state/useAgentDocument.ts | 34 +++ apps/agent-forge/src/state/useStudioEvents.ts | 39 +++ apps/agent-forge/vite.config.ts | 6 + 22 files changed, 984 insertions(+), 41 deletions(-) create mode 100644 apps/agent-forge/server/__tests__/eventBus.test.ts create mode 100644 apps/agent-forge/server/__tests__/eventsWs.test.ts create mode 100644 apps/agent-forge/server/events/eventBus.ts create mode 100644 apps/agent-forge/server/events/eventsWs.ts create mode 100644 apps/agent-forge/server/ws/__tests__/upgradeRouter.test.ts create mode 100644 apps/agent-forge/server/ws/upgradeRouter.ts create mode 100644 apps/agent-forge/src/state/__tests__/useStudioEvents.test.tsx create mode 100644 apps/agent-forge/src/state/useStudioEvents.ts diff --git a/apps/agent-forge/server/__tests__/eventBus.test.ts b/apps/agent-forge/server/__tests__/eventBus.test.ts new file mode 100644 index 000000000..8b930b1e9 --- /dev/null +++ b/apps/agent-forge/server/__tests__/eventBus.test.ts @@ -0,0 +1,52 @@ +import { describe, it, expect, vi } from 'vitest'; +import { StudioEventBus } from '../events/eventBus'; +import type { StudioEvent } from '../../shared/wireTypes'; + +/** + * The per-workspace Studio event bus (server/events/eventBus.ts) that + * `WS /events` fans out from: a publish reaches every current subscriber, + * an unsubscribed listener hears nothing. + */ +describe('StudioEventBus', () => { + const AGENT_CHANGED: StudioEvent = { type: 'agent.changed', agentId: 'a1', reason: 'save' }; + + it('delivers a published event to every subscriber', () => { + const bus = new StudioEventBus(); + const first = vi.fn(); + const second = vi.fn(); + bus.subscribe(first); + bus.subscribe(second); + bus.publish(AGENT_CHANGED); + expect(first).toHaveBeenCalledWith(AGENT_CHANGED); + expect(second).toHaveBeenCalledWith(AGENT_CHANGED); + }); + + it('an unsubscribed listener receives nothing', () => { + const bus = new StudioEventBus(); + const listener = vi.fn(); + const unsubscribe = bus.subscribe(listener); + unsubscribe(); + bus.publish(AGENT_CHANGED); + expect(listener).not.toHaveBeenCalled(); + }); + + it('unsubscribing one listener leaves the others subscribed', () => { + const bus = new StudioEventBus(); + const kept = vi.fn(); + const dropped = vi.fn(); + bus.subscribe(kept); + const unsubscribe = bus.subscribe(dropped); + unsubscribe(); + bus.publish(AGENT_CHANGED); + expect(kept).toHaveBeenCalledTimes(1); + expect(dropped).not.toHaveBeenCalled(); + }); + + it('tolerates many subscribers (the WS endpoint adds one per open tab)', () => { + const bus = new StudioEventBus(); + const listeners = Array.from({ length: 25 }, () => vi.fn()); + for (const listener of listeners) bus.subscribe(listener); + bus.publish(AGENT_CHANGED); + for (const listener of listeners) expect(listener).toHaveBeenCalledTimes(1); + }); +}); diff --git a/apps/agent-forge/server/__tests__/eventsWs.test.ts b/apps/agent-forge/server/__tests__/eventsWs.test.ts new file mode 100644 index 000000000..877558c87 --- /dev/null +++ b/apps/agent-forge/server/__tests__/eventsWs.test.ts @@ -0,0 +1,279 @@ +import { describe, it, expect, beforeAll, beforeEach, afterAll, afterEach } from 'vitest'; +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import request from 'supertest'; +import { WebSocket } from 'ws'; +import type { AgentSpecV2 } from 'lousho'; +import { startStudioServer, type StudioServerHandle } from '../index'; +import { STUDIO_TOKEN_HEADER } from '../auth'; +import { createApp } from '../app'; +import { RunManager } from '../runRegistry'; +import { newTestHost } from './testHost'; +import { FileCheckpointStore } from '../checkpointStore'; +import { FileApprovalStore } from '../approvalStore'; +import { createFsAgentStore } from '../../src/persistence/fsAgentStore'; +import { AgentRepo } from '../storage/agentRepo'; +import { StudioEventBus } from '../events/eventBus'; +import type { AgentStore } from '../../src/persistence/AgentStore'; +import type { StudioEvent, StreamMessage } from '../../shared/wireTypes'; + +const TOKEN = 'events-token-0123456789'; + +const V2_SPEC: AgentSpecV2 = { + version: 2, + name: 'ev-agent', + model: 'mock/mock-1', + instructions: 'Check the queues.', +}; + +/** + * `WS /events` end to end: a real `startStudioServer` pushes the + * `StudioEvent`s its routes publish to every connected tab, under the + * same token gate as `WS /agents/:id/stream` - which keeps working off + * the same single upgrade router. + */ +describe('WS /events (startStudioServer)', () => { + let baseDir: string; + let handle: StudioServerHandle; + let origin: string; + + beforeAll(async () => { + baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'studio-events-')); + handle = await startStudioServer({ + baseDir, + port: 0, + observe: false, + judge: false, + staticDir: false, + token: TOKEN, + }); + origin = `http://127.0.0.1:${handle.port}`; + }); + + afterAll(async () => { + await handle.close(); + fs.rmSync(baseDir, { recursive: true, force: true }); + }); + + function wsUrl(path: string): string { + return `ws://127.0.0.1:${handle.port}${path}`; + } + + /** Awaits one StudioEvent matching `pred` from a list 'message' appends to. */ + async function waitForEvent( + events: StudioEvent[], + pred: (event: StudioEvent) => boolean, + timeoutMs = 3000 + ): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const found = events.find(pred); + if (found) return found; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error('timed out waiting for a StudioEvent'); + } + + it('a client on /events?token receives agent.changed after a PUT /agents/:id', async () => { + const ws = new WebSocket(wsUrl(`/events?token=${TOKEN}`)); + const events: StudioEvent[] = []; + ws.on('message', (data) => events.push(JSON.parse(String(data)) as StudioEvent)); + try { + await new Promise((resolve, reject) => { + ws.once('open', () => resolve()); + ws.once('error', reject); + }); + const res = await fetch(`${origin}/agents/ev-agent`, { + method: 'PUT', + headers: { 'Content-Type': 'application/json', [STUDIO_TOKEN_HEADER]: TOKEN }, + body: JSON.stringify({ spec: V2_SPEC }), + }); + expect(res.status).toBe(204); + const frame = await waitForEvent( + events, + (e) => e.type === 'agent.changed' && e.agentId === 'ev-agent' + ); + expect(frame).toEqual({ type: 'agent.changed', agentId: 'ev-agent', reason: 'save' }); + } finally { + ws.close(); + } + }); + + it('rejects /events without the token with 401, like the stream', async () => { + const status = await new Promise((resolve) => { + const ws = new WebSocket(wsUrl('/events')); + ws.on('open', () => { + ws.close(); + resolve(0); + }); + ws.on('unexpected-response', (_req, res) => resolve(res.statusCode ?? 0)); + ws.on('error', () => resolve(0)); + }); + expect(status).toBe(401); + }); + + it('destroys an upgrade to a path no WS endpoint owns', async () => { + const outcome = await new Promise<'open' | 'error' | number>((resolve) => { + const ws = new WebSocket(wsUrl(`/other?token=${TOKEN}`)); + ws.on('open', () => { + ws.close(); + resolve('open'); + }); + ws.on('unexpected-response', (_req, res) => resolve(res.statusCode ?? 0)); + ws.on('error', () => resolve('error')); + }); + expect(outcome).toBe('error'); + }); + + it('the agent stream still answers status on connect', async () => { + const ws = new WebSocket(wsUrl(`/agents/ev-agent/stream?token=${TOKEN}`)); + const messages: StreamMessage[] = []; + ws.on('message', (data) => messages.push(JSON.parse(String(data)) as StreamMessage)); + try { + const status = await new Promise>((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('no status frame')), 3000); + ws.on('message', (data) => { + const message = JSON.parse(String(data)) as StreamMessage; + if (message.type === 'status') { + clearTimeout(timer); + resolve(message); + } + }); + ws.on('error', reject); + }); + expect(status.type).toBe('status'); + expect(status.payload.agentId).toBe('ev-agent'); + } finally { + ws.close(); + } + }); +}); + +/** + * Which routes publish which StudioEvents, asserted against an injected + * bus on `createApp` (the `eventBus` test seam) rather than a real socket. + */ +describe('route publishes (injected eventBus)', () => { + let baseDir: string; + let agentStore: AgentStore; + let runManager: RunManager; + let repo: AgentRepo; + let events: StudioEvent[]; + let eventBus: StudioEventBus; + + const app = () => + createApp({ agentStore, runManager, baseDir, eventBus }); + + const lastEvents = () => events.splice(0, events.length); + + beforeEach(() => { + baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lou-d15-events-')); + agentStore = createFsAgentStore(baseDir); + repo = new AgentRepo(baseDir); + runManager = new RunManager({ + baseDir, + checkpointStore: new FileCheckpointStore(baseDir), + approvalStore: new FileApprovalStore(baseDir), + host: newTestHost(baseDir), + loadSpec: (id) => repo.readSpec(id), + saveSpec: (id, spec) => repo.writeSpec(id, spec), + }); + events = []; + eventBus = new StudioEventBus(); + eventBus.subscribe((event) => events.push(event)); + }); + + afterEach(() => { + fs.rmSync(baseDir, { recursive: true, force: true }); + }); + + const putAgent = (id: string, spec: unknown = { ...V2_SPEC, name: id }) => + request(app()).put(`/agents/${id}`).send({ spec }); + + it('PUT /agents/:id publishes agent.changed reason save', async () => { + expect((await putAgent('a1')).status).toBe(204); + expect(lastEvents()).toEqual([{ type: 'agent.changed', agentId: 'a1', reason: 'save' }]); + }); + + it('DELETE /agents/:id publishes agent.changed reason save', async () => { + await putAgent('a1'); + lastEvents(); + expect((await request(app()).delete('/agents/a1')).status).toBe(204); + expect(lastEvents()).toEqual([{ type: 'agent.changed', agentId: 'a1', reason: 'save' }]); + }); + + it('PATCH /agents/:id/meta publishes agent.changed reason meta', async () => { + await putAgent('a1'); + lastEvents(); + expect((await request(app()).patch('/agents/a1/meta').send({ enabled: false })).status).toBe(200); + expect(lastEvents()).toEqual([{ type: 'agent.changed', agentId: 'a1', reason: 'meta' }]); + }); + + it('POST /agents publishes agent.changed reason save for the new id', async () => { + const res = await request(app()).post('/agents').send({ from: 'blank', name: 'Fresh One' }); + expect(res.status).toBe(201); + expect(lastEvents()).toEqual([{ type: 'agent.changed', agentId: res.body.id, reason: 'save' }]); + }); + + it('POST /agents/:id/versions publishes agent.changed reason version when a version is written', async () => { + await putAgent('a1'); + lastEvents(); + expect((await request(app()).post('/agents/a1/versions').send({})).status).toBe(201); + expect(lastEvents()).toEqual([{ type: 'agent.changed', agentId: 'a1', reason: 'version' }]); + // A no-op save (nothing changed since the version) publishes nothing. + expect((await request(app()).post('/agents/a1/versions').send({})).status).toBe(200); + expect(lastEvents()).toEqual([]); + }); + + it('POST /agents/:id/versions/:n/restore publishes agent.changed reason restore', async () => { + await putAgent('a1'); + await request(app()).post('/agents/a1/versions').send({}); + await putAgent('a1', { ...V2_SPEC, name: 'a1', description: 'changed' }); + lastEvents(); + expect((await request(app()).post('/agents/a1/versions/1/restore')).status).toBe(201); + expect(lastEvents()).toEqual([{ type: 'agent.changed', agentId: 'a1', reason: 'restore' }]); + }); + + const draftBody = (description: string) => ({ patch: { description }, origin: 'manual', reason: 'try it' }); + + it('draft create/withdraw/approve/reject publish draft.changed (+ agent.changed draft on approve)', async () => { + await putAgent('a1'); + lastEvents(); + + // Create -> pending. + const created = await request(app()).post('/agents/a1/draft').send(draftBody('Drafted')); + expect(created.status).toBe(201); + expect(lastEvents()).toEqual([ + { type: 'draft.changed', agentId: 'a1', draftId: created.body.id, status: 'pending' }, + ]); + + // Approve -> approved, plus the agent record changed. + const approved = await request(app()).post('/agents/a1/draft/approve').send({}); + expect(approved.status).toBe(200); + expect(lastEvents()).toEqual([ + { type: 'draft.changed', agentId: 'a1', draftId: created.body.id, status: 'approved' }, + { type: 'agent.changed', agentId: 'a1', reason: 'draft' }, + ]); + + // Create + reject -> rejected. + const second = await request(app()).post('/agents/a1/draft').send(draftBody('Drafted again')); + expect(second.status).toBe(201); + lastEvents(); + const rejected = await request(app()).post('/agents/a1/draft/reject').send({}); + expect(rejected.status).toBe(200); + expect(lastEvents()).toEqual([ + { type: 'draft.changed', agentId: 'a1', draftId: second.body.id, status: 'rejected' }, + ]); + + // Create + withdraw -> withdrawn. + const third = await request(app()).post('/agents/a1/draft').send(draftBody('Drafted thrice')); + expect(third.status).toBe(201); + lastEvents(); + const withdrawn = await request(app()).delete('/agents/a1/draft'); + expect(withdrawn.status).toBe(200); + expect(lastEvents()).toEqual([ + { type: 'draft.changed', agentId: 'a1', draftId: third.body.id, status: 'withdrawn' }, + ]); + }); +}); diff --git a/apps/agent-forge/server/__tests__/runRegistry.test.ts b/apps/agent-forge/server/__tests__/runRegistry.test.ts index d75d639d9..fd8cbe900 100644 --- a/apps/agent-forge/server/__tests__/runRegistry.test.ts +++ b/apps/agent-forge/server/__tests__/runRegistry.test.ts @@ -9,6 +9,7 @@ import { upgradeSpec, type AgentSpecV2 } from 'lousho'; import { RunManager } from '../runRegistry'; import { newTestHost } from './testHost'; import { attachWebSocketServer } from '../wsServer'; +import { createUpgradeRouter } from '../ws/upgradeRouter'; import { FileCheckpointStore } from '../checkpointStore'; import { FileApprovalStore } from '../approvalStore'; import { graphToSpec } from '../../src/graph/graphToSpec'; @@ -129,7 +130,7 @@ describe('RunManager', () => { it('streams each model step over the WebSocket: several text.delta, one text.done per step', async () => { const server = http.createServer(); - const wss = attachWebSocketServer(server, runManager); + const wss = attachWebSocketServer(createUpgradeRouter(server), runManager); await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); const { port } = server.address() as AddressInfo; const ws = new WebSocket(`ws://127.0.0.1:${port}/agents/agent-ws/stream`); diff --git a/apps/agent-forge/server/__tests__/workspaceSeam.test.ts b/apps/agent-forge/server/__tests__/workspaceSeam.test.ts index 1d96c61e0..1f068cb1a 100644 --- a/apps/agent-forge/server/__tests__/workspaceSeam.test.ts +++ b/apps/agent-forge/server/__tests__/workspaceSeam.test.ts @@ -17,6 +17,7 @@ import { WORKSPACE_ID_RE } from '../../shared/workspace'; import { workspaceIdFor, workspaceRefFor } from '../workspace/workspaceId'; import { DraftStore } from '../storage/draftStore'; import { VersionStore } from '../storage/versionStore'; +import { StudioEventBus } from '../events/eventBus'; import type { WorkspaceContext } from '../workspace/context'; const SPEC: AgentSpec = { @@ -119,6 +120,7 @@ describe('workspace seam', () => { host: newTestHost(baseDir), runs: { status } as unknown as RunManager, chatTriggers: new TriggerRegistry(), + events: new StudioEventBus(), }; const app = createApp({ workspace, baseDir }); diff --git a/apps/agent-forge/server/app.ts b/apps/agent-forge/server/app.ts index b303d1a04..d615e24cf 100644 --- a/apps/agent-forge/server/app.ts +++ b/apps/agent-forge/server/app.ts @@ -42,9 +42,10 @@ * POST/GET /assistant/sessions · GET/DELETE /assistant/sessions/:id * -> Agent Author chat sessions, Studio-global (author/routes.ts) * - * `WS /agents/:id/stream` is wired separately in wsServer.ts (attached to - * the same underlying http.Server, since `ws` needs the raw HTTP upgrade - * event rather than an Express route). + * `WS /agents/:id/stream` and `WS /events` are wired separately + * (wsServer.ts, events/eventsWs.ts - both registered on the one + * `UpgradeRouter` attached to the same underlying http.Server, since `ws` + * needs the raw HTTP upgrade event rather than an Express route). * * Every route reads its workspace from workspaceOf(req) (header * x-lousho-workspace or ?ws=, else the launch folder). @@ -109,6 +110,7 @@ import { asyncRoute, paramId, sendError } from './http/routeUtils'; import { AgentRepo } from './storage/agentRepo'; import { DraftStore } from './storage/draftStore'; import { VersionStore } from './storage/versionStore'; +import { StudioEventBus } from './events/eventBus'; import { mergeLegacySave } from './storage/legacySpec'; import { FileCheckpointStore } from './checkpointStore'; import { FileApprovalStore } from './approvalStore'; @@ -179,6 +181,14 @@ interface CreateAppBaseOptions { judge?: JudgeFn; /** The `connectMcp` `POST /catalog/mcp/probe` uses (tests); default the SDK's own (routes/mcpProbe.ts). */ mcpConnect?: typeof connectMcp; + /** + * The `StudioEventBus` the default workspace context publishes + * `StudioEvent`s on - what `workspaceOf(req).events` answers and + * `WS /events` fans out. A test seam: `startStudioServer` passes the + * launch context's own bus; a test injects one to spy on published + * events. Defaults to the context's own `new StudioEventBus()`. + */ + eventBus?: StudioEventBus; /** * The app's one TemplateCatalog - Studio-global (the templates folder * ships in the SDK package), shared between `GET /templates*` and the @@ -224,6 +234,7 @@ const API_PREFIXES = [ '/catalog', '/assistant', '/templates', + '/events', ]; /** The only methods served without the token, and only off the API (`/health`, the static client). */ @@ -760,7 +771,7 @@ function registerStaticClient(app: Express, staticDir: string | undefined): void // `App.tsx` view state, but still refresh-safe) - serve `index.html` and // let the client app take over, the same as `vite preview`/most SPA // hosts do. - app.get(/^(?!\/health|\/agents|\/runs|\/workspaces|\/observe|\/playground|\/catalog|\/assistant|\/templates).*/, (_req, res) => { + app.get(/^(?!\/health|\/agents|\/runs|\/workspaces|\/observe|\/playground|\/catalog|\/assistant|\/templates|\/events).*/, (_req, res) => { res.sendFile(indexHtml); }); } @@ -772,7 +783,18 @@ function registerStaticClient(app: Express, staticDir: string | undefined): void * `workspace`: the launch folder's ref around those two stores. */ function defaultWorkspaceContext(options: CreateAppOptions): WorkspaceContext { - if (options.workspace) return options.workspace; + const workspace = options.workspace ?? legacyWorkspaceContext(options); + // `eventBus` is a test seam: it sets the default context's bus - the + // one `workspaceOf(req).events` publishes on and `WS /events` reads. + return options.eventBus === undefined ? workspace : { ...workspace, events: options.eventBus }; +} + +/** + * The context synthesized when the caller passed the legacy + * `{agentStore, runManager}` pair instead of a ready-built `workspace`: + * the launch folder's ref around those two stores. + */ +function legacyWorkspaceContext(options: CreateAppOptions & { workspace?: undefined }): WorkspaceContext { const repo = new AgentRepo(options.baseDir); return { ref: workspaceRefFor(options.baseDir), @@ -791,6 +813,7 @@ function defaultWorkspaceContext(options: CreateAppOptions): WorkspaceContext { }), runs: options.runManager, chatTriggers: chatTriggersFor(options.runManager), + events: new StudioEventBus(), }; } diff --git a/apps/agent-forge/server/events/eventBus.ts b/apps/agent-forge/server/events/eventBus.ts new file mode 100644 index 000000000..1f21f626f --- /dev/null +++ b/apps/agent-forge/server/events/eventBus.ts @@ -0,0 +1,40 @@ +/** + * The per-workspace Studio event bus. + * + * Write routes publish `StudioEvent`s here (`workspaceOf(req).events`), + * and `WS /events` (eventsWs.ts) forwards them to every open tab so the + * agents list refreshes on change rather than polling. One bus per + * `WorkspaceContext` - each workspace's writes only reach tabs watching + * that workspace. + */ +import { EventEmitter } from 'node:events'; +import type { StudioEvent } from '../../shared/wireTypes'; + +/** One open tab's `message` frame producer - returns the unsubscribe. */ +export type StudioEventListener = (event: StudioEvent) => void; + +/** + * A synchronous fan-out over `EventEmitter`: `publish` runs every current + * subscriber before returning. `setMaxListeners(0)` because every open + * `/events` socket adds one listener and tabs come and go freely. + */ +export class StudioEventBus { + private readonly emitter = new EventEmitter(); + + constructor() { + this.emitter.setMaxListeners(0); + } + + /** Delivers `event` to every current subscriber, synchronously. */ + publish(event: StudioEvent): void { + this.emitter.emit('event', event); + } + + /** Subscribes `listener` to every published event; the returned function unsubscribes it. */ + subscribe(listener: StudioEventListener): () => void { + this.emitter.on('event', listener); + return () => { + this.emitter.off('event', listener); + }; + } +} diff --git a/apps/agent-forge/server/events/eventsWs.ts b/apps/agent-forge/server/events/eventsWs.ts new file mode 100644 index 000000000..7a5e929d4 --- /dev/null +++ b/apps/agent-forge/server/events/eventsWs.ts @@ -0,0 +1,35 @@ +/** + * `WS /events` - pushes every `StudioEvent` published on the workspace's + * `StudioEventBus` to every connected tab, as JSON text frames. Nothing + * is sent on connect: events are live notifications only (a client that + * missed one re-reads the REST state it refreshes). + * + * Registered on the server's one `UpgradeRouter` (ws/upgradeRouter.ts), + * so the shared Host/Origin/token gate runs before this handler - the + * `?token=` query the gate checks rides along on the WS URL. + */ +import { WebSocketServer, WebSocket } from 'ws'; +import type { StudioEvent } from '../../shared/wireTypes'; +import type { UpgradeRouter } from '../ws/upgradeRouter'; +import type { StudioEventBus } from './eventBus'; + +/** The exact pathname `WS /events` registers on the upgrade router. */ +const EVENTS_PATH = '/events'; + +/** + * Registers `WS /events` on `router`: each accepted socket subscribes to + * `events` until it closes. Returns the (noServer) WebSocketServer so the + * caller can close it with the http server. + */ +export function attachEventsWebSocket(router: UpgradeRouter, events: StudioEventBus): WebSocketServer { + const wss = new WebSocketServer({ noServer: true }); + router.register(EVENTS_PATH, (req, socket, head) => { + wss.handleUpgrade(req, socket, head, (ws) => { + const unsubscribe = events.subscribe((event: StudioEvent) => { + if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify(event)); + }); + ws.on('close', unsubscribe); + }); + }); + return wss; +} diff --git a/apps/agent-forge/server/index.ts b/apps/agent-forge/server/index.ts index 21e141a35..dcc2ed61c 100644 --- a/apps/agent-forge/server/index.ts +++ b/apps/agent-forge/server/index.ts @@ -11,6 +11,8 @@ import * as path from 'node:path'; import { fileURLToPath, pathToFileURL } from 'node:url'; import { createApp } from './app'; import { attachWebSocketServer } from './wsServer'; +import { attachEventsWebSocket } from './events/eventsWs'; +import { createUpgradeRouter, type UpgradeRouter } from './ws/upgradeRouter'; import { RunManager } from './runRegistry'; import { SecretsStore } from './secretsStore'; import { SettingsStore } from './settingsStore'; @@ -24,6 +26,12 @@ export interface StudioServerHandle { server: http.Server; port: number; runManager: RunManager; + /** + * The server's one WS upgrade router (ws/upgradeRouter.ts): `/agents/:id/stream` + * and `/events` are already registered; later WS endpoints (the Author + * stream) call `upgradeRouter.register(...)` on it. + */ + upgradeRouter: UpgradeRouter; /** The per-launch token every API/WebSocket request must carry. */ token: string; /** The URL to open: the studio's own origin with `?token=` appended. */ @@ -141,9 +149,14 @@ export async function startStudioServer( staticDir, docsDir: resolveDocsDir(moduleDir), access, + eventBus: workspace.events, }); const server = http.createServer(app); - attachWebSocketServer(server, workspace.runs, access); + // One upgrade listener for every WS endpoint: the shared Host/Origin/ + // token gate runs in the router, then the first registered route wins. + const upgradeRouter = createUpgradeRouter(server, access); + attachWebSocketServer(upgradeRouter, workspace.runs); + attachEventsWebSocket(upgradeRouter, workspace.events); await listen(server, port, host); const judgeOption = options.judge ?? (options.observe === false ? false : {}); @@ -155,6 +168,7 @@ export async function startStudioServer( server, port: boundPort, runManager: workspace.runs, + upgradeRouter, token, url: `http://${urlHost}:${boundPort}/?token=${encodeURIComponent(token)}`, close: () => { diff --git a/apps/agent-forge/server/routes/agentRoutes.ts b/apps/agent-forge/server/routes/agentRoutes.ts index 4f161bc72..37c8d792a 100644 --- a/apps/agent-forge/server/routes/agentRoutes.ts +++ b/apps/agent-forge/server/routes/agentRoutes.ts @@ -334,6 +334,7 @@ export function registerAgentRoutes(app: Express, deps: AgentRouteDeps): void { return; } res.status(201).json(record); + ws.events.publish({ type: 'agent.changed', agentId: record.id, reason: 'save' }); return; } if (parsed.kind === 'copy') { @@ -343,9 +344,12 @@ export function registerAgentRoutes(app: Express, deps: AgentRouteDeps): void { return; } res.status(201).json(record); + ws.events.publish({ type: 'agent.changed', agentId: record.id, reason: 'save' }); return; } - res.status(201).json(await createBlankAgent(ws, parsed.name)); + const record = await createBlankAgent(ws, parsed.name); + res.status(201).json(record); + ws.events.publish({ type: 'agent.changed', agentId: record.id, reason: 'save' }); } catch (error) { if (error instanceof InvalidSpecError) { sendError(res, 400, error.message, error.code, { issues: error.issues }); @@ -411,6 +415,7 @@ export function registerAgentRoutes(app: Express, deps: AgentRouteDeps): void { throw error; } res.status(204).end(); + ws.events.publish({ type: 'agent.changed', agentId: id, reason: 'save' }); }) ); @@ -461,6 +466,7 @@ export function registerAgentRoutes(app: Express, deps: AgentRouteDeps): void { if (labels !== undefined) next.labels = labels; await ws.repo.writeMeta(id, next); res.json(next); + ws.events.publish({ type: 'agent.changed', agentId: id, reason: 'meta' }); }); }) ); @@ -489,8 +495,11 @@ export function registerAgentRoutes(app: Express, deps: AgentRouteDeps): void { app.delete( '/agents/:id', asyncRoute(async (req, res) => { - await workspaceOf(req).repo.remove(paramId(req)); + const ws = workspaceOf(req); + const id = paramId(req); + await ws.repo.remove(id); res.status(204).end(); + ws.events.publish({ type: 'agent.changed', agentId: id, reason: 'save' }); }) ); } diff --git a/apps/agent-forge/server/routes/draftRoutes.ts b/apps/agent-forge/server/routes/draftRoutes.ts index 4a4fe46ab..5d816e518 100644 --- a/apps/agent-forge/server/routes/draftRoutes.ts +++ b/apps/agent-forge/server/routes/draftRoutes.ts @@ -186,6 +186,7 @@ export function registerDraftRoutes(app: Express): void { baseSpecHash: specHash(spec), }); res.status(201).json(draft); + ws.events.publish({ type: 'draft.changed', agentId: id, draftId: draft.id, status: draft.status }); }); }) ); @@ -199,6 +200,7 @@ export function registerDraftRoutes(app: Express): void { try { const settled = await withAgentLock(id, () => ws.drafts.settle(id, 'withdrawn')); res.json(settled); + ws.events.publish({ type: 'draft.changed', agentId: id, draftId: settled.id, status: settled.status }); } catch (error) { sendDraftStoreError(res, error); } @@ -243,6 +245,8 @@ export function registerDraftRoutes(app: Express): void { await ws.versions.capture(id, { kind: 'update', source: 'draft' }); const settled = await ws.drafts.settle(id, 'approved', { draftId: cur.id, note: parsed.note }); res.json(settled); + ws.events.publish({ type: 'draft.changed', agentId: id, draftId: settled.id, status: settled.status }); + ws.events.publish({ type: 'agent.changed', agentId: id, reason: 'draft' }); }); } catch (error) { sendDraftStoreError(res, error); @@ -266,6 +270,7 @@ export function registerDraftRoutes(app: Express): void { const cur = await pendingDraft(ws, id, parsed.draftId); const settled = await ws.drafts.settle(id, 'rejected', { draftId: cur.id, note: parsed.note }); res.json(settled); + ws.events.publish({ type: 'draft.changed', agentId: id, draftId: settled.id, status: settled.status }); }); } catch (error) { sendDraftStoreError(res, error); diff --git a/apps/agent-forge/server/routes/versionRoutes.ts b/apps/agent-forge/server/routes/versionRoutes.ts index be0cd4b5b..672ca6e9f 100644 --- a/apps/agent-forge/server/routes/versionRoutes.ts +++ b/apps/agent-forge/server/routes/versionRoutes.ts @@ -100,6 +100,8 @@ export function registerVersionRoutes(app: Express): void { const latest = await ws.versions.latest(id); const item = await ws.versions.capture(id, { kind: latest ? 'update' : 'create', source: 'editor' }); res.status(item === null ? 200 : 201).json(item ?? latest); + // A no-op save writes nothing, so it publishes nothing. + if (item !== null) ws.events.publish({ type: 'agent.changed', agentId: id, reason: 'version' }); }); }) ); @@ -191,6 +193,7 @@ export function registerVersionRoutes(app: Express): void { restoredFromVersion: n, }); res.status(201).json(item); + ws.events.publish({ type: 'agent.changed', agentId: id, reason: 'restore' }); }); }) ); diff --git a/apps/agent-forge/server/workspace/__tests__/middleware.test.ts b/apps/agent-forge/server/workspace/__tests__/middleware.test.ts index a1bef229c..b73330a6c 100644 --- a/apps/agent-forge/server/workspace/__tests__/middleware.test.ts +++ b/apps/agent-forge/server/workspace/__tests__/middleware.test.ts @@ -7,6 +7,7 @@ import type { AgentStore } from '../../../src/persistence/AgentStore'; import type { RunManager } from '../../runRegistry'; import { AgentRepo } from '../../storage/agentRepo'; import { DraftStore } from '../../storage/draftStore'; +import { StudioEventBus } from '../../events/eventBus'; import type { AgentHost } from '../../host/agentHost'; import { VersionStore } from '../../storage/versionStore'; import { singleWorkspace, type WorkspaceContext, type WorkspaceResolver } from '../context'; @@ -30,6 +31,7 @@ function ctx(id: string): WorkspaceContext { host: {} as AgentHost, runs: {} as RunManager, chatTriggers: new TriggerRegistry(), + events: new StudioEventBus(), }; } diff --git a/apps/agent-forge/server/workspace/context.ts b/apps/agent-forge/server/workspace/context.ts index 3bbfdbfaf..0b0bfa48b 100644 --- a/apps/agent-forge/server/workspace/context.ts +++ b/apps/agent-forge/server/workspace/context.ts @@ -24,6 +24,7 @@ import { VersionStore } from '../storage/versionStore'; import { ChatTriggerAdapter } from '../chatTriggerAdapter'; import { FileApprovalStore } from '../approvalStore'; import { FileCheckpointStore } from '../checkpointStore'; +import { StudioEventBus } from '../events/eventBus'; import { InProcessAgentHost } from '../host/inProcessAgentHost'; import type { AgentHost } from '../host/agentHost'; import { RunManager } from '../runRegistry'; @@ -46,6 +47,12 @@ export interface WorkspaceContext { runs: RunManager; /** `'chat'` → `ChatTriggerAdapter(runs)`; `POST /agents/:id/message` goes through it (chatTriggerAdapter.ts). */ chatTriggers: TriggerRegistry; + /** + * This workspace's Studio event bus (events/eventBus.ts): write routes + * publish `StudioEvent`s here, `WS /events` fans them out to that + * workspace's tabs. Per workspace, like the stores. + */ + events: StudioEventBus; } /** @@ -139,5 +146,6 @@ export function createWorkspaceContext(ref: WorkspaceRef, services: WorkspaceSer host, runs, chatTriggers: chatTriggersFor(runs), + events: new StudioEventBus(), }; } diff --git a/apps/agent-forge/server/ws/__tests__/upgradeRouter.test.ts b/apps/agent-forge/server/ws/__tests__/upgradeRouter.test.ts new file mode 100644 index 000000000..52e27642c --- /dev/null +++ b/apps/agent-forge/server/ws/__tests__/upgradeRouter.test.ts @@ -0,0 +1,151 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import * as http from 'node:http'; +import type { IncomingMessage } from 'node:http'; +import type { AddressInfo, Socket } from 'node:net'; +import { WebSocket, WebSocketServer } from 'ws'; +import { createUpgradeRouter, type UpgradeRouter } from '../upgradeRouter'; + +/** + * The single `upgrade` listener every WS endpoint registers on + * (server/ws/upgradeRouter.ts): exact-path and RegExp routing in + * registration order, duplicate registrations rejected, unmatched paths + * destroyed, and the shared Host/Origin/token gate run before any handler. + */ +describe('createUpgradeRouter', () => { + const TOKEN = 'router-token-0123456789'; + let server: http.Server; + let port: number; + + beforeEach(async () => { + server = http.createServer(); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + port = (server.address() as AddressInfo).port; + }); + + afterEach(async () => { + // Open upgrade sockets keep `server.close` waiting - drop them first. + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + }); + + /** Opens a WS upgrade attempt and reports how it ended: 'open', 'error' (socket destroyed), or the HTTP status written back. */ + function attempt(path: string): Promise<'open' | 'error' | number> { + return new Promise((resolve) => { + const ws = new WebSocket(`ws://127.0.0.1:${port}${path}`); + ws.on('open', () => { + ws.close(); + resolve('open'); + }); + ws.on('unexpected-response', (_req, res) => resolve(res.statusCode ?? 0)); + ws.on('error', () => resolve('error')); + }); + } + + /** A handler that records the matched pathname, then drops the socket. */ + function recording(seen: string[]) { + return (_req: IncomingMessage, socket: Socket, _head: Buffer, url: URL) => { + seen.push(url.pathname); + socket.destroy(); + }; + } + + it('routes an exact pathname to its handler and nothing else', async () => { + const router = createUpgradeRouter(server); + const seen: string[] = []; + router.register('/events', recording(seen)); + expect(await attempt('/events?x=1')).toBe('error'); // handled, then destroyed by the recording handler + expect(seen).toEqual(['/events']); + expect(await attempt('/events/extra')).toBe('error'); // unmatched -> destroyed + expect(seen).toEqual(['/events']); + }); + + it('routes a RegExp pattern and passes the parsed URL through', async () => { + const router = createUpgradeRouter(server); + const seen: string[] = []; + let seenUrl: URL | undefined; + router.register(/^\/agents\/([^/]+)\/stream$/, (_req, socket, _head, url) => { + seen.push(url.pathname); + seenUrl = url; + socket.destroy(); + }); + expect(await attempt('/agents/a1/stream?token=x')).toBe('error'); + expect(seen).toEqual(['/agents/a1/stream']); + expect(seenUrl?.searchParams.get('token')).toBe('x'); + expect(await attempt('/agents/stream')).toBe('error'); + expect(seen).toEqual(['/agents/a1/stream']); + }); + + it('dispatches the first matching route in registration order', async () => { + const router: UpgradeRouter = createUpgradeRouter(server); + const exact: string[] = []; + const anyPath: string[] = []; + router.register('/events', recording(exact)); + router.register(/^\//, recording(anyPath)); + await attempt('/events'); + await attempt('/other'); + expect(exact).toEqual(['/events']); + expect(anyPath).toEqual(['/other']); + }); + + it('throws on a duplicate route (same string or same RegExp)', () => { + const router = createUpgradeRouter(server); + router.register('/events', () => {}); + router.register(/^\/agents\/x\/stream$/, () => {}); + expect(() => router.register('/events', () => {})).toThrow(/duplicate/i); + expect(() => router.register(/^\/agents\/x\/stream$/, () => {})).toThrow(/duplicate/i); + // A string and a RegExp spelling the same path are different routes. + expect(() => router.register(/^\/events$/, () => {})).not.toThrow(); + }); + + it('destroys an upgrade whose path matches no route', async () => { + const router = createUpgradeRouter(server); + router.register('/events', () => {}); + expect(await attempt('/other')).toBe('error'); + }); + + it('lets a handler complete the upgrade with its own WebSocketServer', async () => { + const router = createUpgradeRouter(server); + const wss = new WebSocketServer({ noServer: true }); + router.register('/events', (req, socket, head) => { + wss.handleUpgrade(req, socket, head, (ws) => ws.send('hello')); + }); + const ws = new WebSocket(`ws://127.0.0.1:${port}/events`); + const got = new Promise((resolve, reject) => { + ws.on('message', (data) => resolve(String(data))); + ws.on('error', reject); + }); + try { + await expect(got).resolves.toBe('hello'); + } finally { + ws.terminate(); + wss.close(); + } + }); + + it('rejects with 401 before any handler runs when the token is missing', async () => { + const router = createUpgradeRouter(server, { token: TOKEN }); + const seen: string[] = []; + router.register('/events', recording(seen)); + router.register(/^\//, recording(seen)); // catch-all: only routing can be skipped, not the gate + expect(await attempt('/events')).toBe(401); + expect(await attempt('/events?token=wrong')).toBe(401); + expect(seen).toEqual([]); + expect(await attempt(`/events?token=${TOKEN}`)).toBe('error'); + expect(seen).toEqual(['/events']); + }); + + it('rejects a foreign Origin with 403 before any handler runs', async () => { + const router = createUpgradeRouter(server, { token: TOKEN }); + const seen: string[] = []; + router.register(/^\//, recording(seen)); + const status = await new Promise((resolve) => { + const ws = new WebSocket(`ws://127.0.0.1:${port}/events?token=${TOKEN}`, { + headers: { Origin: 'https://example.com' }, + }); + ws.on('unexpected-response', (_req, res) => resolve(res.statusCode ?? 0)); + ws.on('error', () => resolve(0)); + }); + expect(status).toBe(403); + expect(seen).toEqual([]); + }); +}); diff --git a/apps/agent-forge/server/ws/upgradeRouter.ts b/apps/agent-forge/server/ws/upgradeRouter.ts new file mode 100644 index 000000000..0e9f331e4 --- /dev/null +++ b/apps/agent-forge/server/ws/upgradeRouter.ts @@ -0,0 +1,86 @@ +/** + * The ONE `upgrade` listener on Studio's `http.Server`. + * + * Two `upgrade` listeners on one server both fire for every request, so + * giving each WS endpoint its own listener means the wrong one destroys + * the other's paths. Instead, `createUpgradeRouter` owns the single + * listener; every WS endpoint (`/agents/:id/stream`, `/events`, later + * `/assistant/stream`) registers a route on it. On each upgrade the + * router runs the shared Host/Origin/token gate (the same checks the HTTP + * API applies - see auth.ts) before any handler, parses the URL once, + * then dispatches by pathname to the first registered route that matches + * (registration order). An unmatched path's socket is destroyed. + */ +import type { Server as HttpServer, IncomingMessage } from 'node:http'; +import type { Socket } from 'node:net'; +import { checkApiAccess, checkHostAndOrigin, type StudioAccessOptions } from '../auth'; + +/** One WS endpoint's upgrade handler: completes the upgrade itself (its own `WebSocketServer({ noServer: true })`) or destroys the socket. */ +export type UpgradeHandler = (req: IncomingMessage, socket: Socket, head: Buffer, url: URL) => void; + +/** The registry `createUpgradeRouter` returns - each WS endpoint calls `register` exactly once. */ +export interface UpgradeRouter { + /** + * `path` is an exact pathname ('/events', '/assistant/stream') or a + * RegExp matched against the pathname ('/^\\/agents\\/([^/]+)\\/stream$/') + * - matching never sees the query string, which stays on `url` for the + * handler. Throws on a duplicate route (same string, or a RegExp with + * the same source and flags). + */ + register(path: string | RegExp, handler: UpgradeHandler): void; +} + +interface UpgradeRoute { + path: string | RegExp; + handler: UpgradeHandler; +} + +/** + * The shared Host/Origin/token gate as the HTTP API applies it (see + * auth.ts); omitted `access` skips only the token. Writes the HTTP + * rejection straight to the socket and answers `true` when the upgrade + * must go no further. + */ +function rejectUpgrade(req: IncomingMessage, socket: Socket, access: StudioAccessOptions | undefined): boolean { + const decision = access ? checkApiAccess(req, access) : checkHostAndOrigin(req, { token: '' }); + if (decision.ok) return false; + const reason = decision.status === 401 ? 'Unauthorized' : 'Forbidden'; + const CRLF = '\r\n'; + socket.end(`HTTP/1.1 ${decision.status} ${reason}${CRLF}Connection: close${CRLF}Content-Length: 0${CRLF}${CRLF}`); + return true; +} + +function sameRoute(a: string | RegExp, b: string | RegExp): boolean { + if (typeof a !== typeof b) return false; + return typeof a === 'string' ? a === b : a.source === (b as RegExp).source && a.flags === (b as RegExp).flags; +} + +/** + * Attaches the single `upgrade` listener to `server` and returns the + * router every WS endpoint registers on. The listener runs the shared + * gate first - a rejection is written before any handler runs - then the + * first matching route gets the upgrade; no match destroys the socket. + */ +export function createUpgradeRouter(server: HttpServer, access?: StudioAccessOptions): UpgradeRouter { + const routes: UpgradeRoute[] = []; + server.on('upgrade', (req: IncomingMessage, socket: Socket, head: Buffer) => { + if (rejectUpgrade(req, socket, access)) return; + const url = new URL(req.url ?? '', 'http://localhost'); + const route = routes.find((r) => + typeof r.path === 'string' ? r.path === url.pathname : r.path.test(url.pathname) + ); + if (!route) { + socket.destroy(); + return; + } + route.handler(req, socket, head, url); + }); + return { + register(path, handler) { + if (routes.some((r) => sameRoute(r.path, path))) { + throw new Error(`Duplicate WS upgrade route '${String(path)}'`); + } + routes.push({ path, handler }); + }, + }; +} diff --git a/apps/agent-forge/server/wsServer.ts b/apps/agent-forge/server/wsServer.ts index 848de13e5..9a1272d9d 100644 --- a/apps/agent-forge/server/wsServer.ts +++ b/apps/agent-forge/server/wsServer.ts @@ -6,17 +6,17 @@ * next transition) plus best-effort `event` messages forwarding the run's * `AgentEvent`s for lightweight in-progress visibility. * - * Attached to the same underlying `http.Server` the Express app listens on - * (see index.ts) via a manual `upgrade` handler, since matching a path - * with a `:id` param needs a tiny bit of parsing `ws` doesn't do for you. + * Registered on the server's one `UpgradeRouter` (ws/upgradeRouter.ts - + * the only `upgrade` listener the http.Server has) rather than attaching + * its own, so the shared Host/Origin/token gate runs once for every WS + * endpoint; this handler only re-validates the `:id` the RegExp captured, + * the same boundary app.ts's `/agents/:id` middleware applies. */ -import type { Server as HttpServer, IncomingMessage } from 'node:http'; -import type { Socket } from 'node:net'; import { WebSocketServer, WebSocket } from 'ws'; import type { RunManager } from './runRegistry'; import { isValidAgentId } from './types'; import type { StreamMessage } from '../shared/wireTypes'; -import { checkApiAccess, checkHostAndOrigin, type StudioAccessOptions } from './auth'; +import type { UpgradeRouter } from './ws/upgradeRouter'; const STREAM_PATH_RE = /^\/agents\/([^/]+)\/stream$/; @@ -27,36 +27,20 @@ function send(ws: WebSocket, message: StreamMessage): void { } /** - * Parses `/agents/:id/stream` out of an upgrade request URL, or returns - * undefined when the path doesn't match or the id fails the same - * path-traversal boundary as app.ts's `/agents/:id` middleware - this - * handler parses `:id` itself (see the module doc comment) rather than - * going through Express routing, so it needs its own check before the id is - * used as a registry/status key alongside the HTTP routes. + * The `:id` `STREAM_PATH_RE` captured, or `undefined` when it fails the + * same path-traversal boundary as app.ts's `/agents/:id` middleware - this + * handler parses `:id` itself (it runs outside Express routing) so it + * needs its own check before the id is used as a registry/status key + * alongside the HTTP routes. */ -function streamAgentId(req: IncomingMessage): string | undefined { - const url = new URL(req.url ?? '', 'http://localhost'); +function streamAgentId(url: URL): string | undefined { const match = STREAM_PATH_RE.exec(url.pathname); if (!match) return undefined; const agentId = decodeURIComponent(match[1]); return isValidAgentId(agentId) ? agentId : undefined; } -/** The same Host/Origin/token gate as the HTTP API (see auth.ts); omitted `access` skips only the token. */ -function rejectUpgrade(req: IncomingMessage, socket: Socket, access: StudioAccessOptions | undefined): boolean { - const decision = access ? checkApiAccess(req, access) : checkHostAndOrigin(req, { token: '' }); - if (decision.ok) return false; - const reason = decision.status === 401 ? 'Unauthorized' : 'Forbidden'; - const CRLF = '\r\n'; - socket.end(`HTTP/1.1 ${decision.status} ${reason}${CRLF}Connection: close${CRLF}Content-Length: 0${CRLF}${CRLF}`); - return true; -} - -export function attachWebSocketServer( - server: HttpServer, - runManager: RunManager, - access?: StudioAccessOptions -): WebSocketServer { +export function attachWebSocketServer(router: UpgradeRouter, runManager: RunManager): WebSocketServer { const wss = new WebSocketServer({ noServer: true }); // agentId -> connected sockets, so status/event broadcasts only fan out // to clients watching that one agent's run. @@ -80,9 +64,8 @@ export function attachWebSocketServer( ); } - server.on('upgrade', (req: IncomingMessage, socket: Socket, head: Buffer) => { - if (rejectUpgrade(req, socket, access)) return; - const agentId = streamAgentId(req); + router.register(STREAM_PATH_RE, (req, socket, head, url) => { + const agentId = streamAgentId(url); if (!agentId) { socket.destroy(); return; diff --git a/apps/agent-forge/shared/wireTypes.ts b/apps/agent-forge/shared/wireTypes.ts index 119f38306..a25741e96 100644 --- a/apps/agent-forge/shared/wireTypes.ts +++ b/apps/agent-forge/shared/wireTypes.ts @@ -222,6 +222,21 @@ export type StreamMessage = | { type: 'debug'; payload: DebugStatePayload } | { type: 'chat'; payload: ChatStatePayload }; +/** + * One `WS /events` push - Studio-wide lifecycle notifications every open + * tab receives, so the agents list (and later the summaries, drafts, + * suggestions and inbox badge) refreshes on change rather than polling. + * `agent.changed` fires whenever an agent's spec/meta/draft state is + * written, `reason` saying which write caused it. Events carry ids and + * names only - never secret values. + */ +export type StudioEvent = + | { type: 'agent.changed'; agentId: string; reason: 'save' | 'version' | 'meta' | 'draft' | 'restore' | 'import' } + | { type: 'version.summary'; agentId: string; version: number; summary: string } + | { type: 'draft.changed'; agentId: string; draftId: string; status: ChangeDraft['status'] } + | { type: 'suggestion.new'; agentId: string; suggestionId: string } + | { type: 'inbox.count'; count: number }; + /** Masked provider key status. Never carries a real key. */ export interface ProviderKeyStatus { provider: 'openai' | 'anthropic'; diff --git a/apps/agent-forge/src/runtime/runtimeClient.ts b/apps/agent-forge/src/runtime/runtimeClient.ts index 1d09d5310..0ee95261a 100644 --- a/apps/agent-forge/src/runtime/runtimeClient.ts +++ b/apps/agent-forge/src/runtime/runtimeClient.ts @@ -44,6 +44,7 @@ import type { SpecChange, StreamMessage, SpanEvent, + StudioEvent, TemplateDetail, TemplateSummary, TraceDetailPayload, @@ -608,6 +609,36 @@ class RuntimeClient implements AgentApi { ws.close(); }; } + + /** + * Subscribes to `WS /events`, the Studio-wide event bus (the server's + * per-workspace `StudioEventBus` fanned out to every tab). Returns an + * unsubscribe function. Like `subscribe()`, reconnects are the caller's + * job (`connection.onLost`) - `useStudioEvents` is that caller. + */ + subscribeEvents( + onEvent: (event: StudioEvent) => void, + connection?: StreamConnectionHandlers + ): () => void { + const query = this.token ? `?${new URLSearchParams({ token: this.token })}` : ''; + const ws = new WebSocket(`${this.wsBaseUrl()}/events${query}`); + let closedByCaller = false; + ws.addEventListener('message', (event) => { + try { + onEvent(JSON.parse(event.data as string) as StudioEvent); + } catch { + // Ignore malformed frames rather than crashing the UI over one bad message. + } + }); + if (connection?.onOpen) ws.addEventListener('open', connection.onOpen); + ws.addEventListener('close', () => { + if (!closedByCaller) connection?.onLost?.(); + }); + return () => { + closedByCaller = true; + ws.close(); + }; + } } export { RuntimeClient }; diff --git a/apps/agent-forge/src/state/__tests__/useStudioEvents.test.tsx b/apps/agent-forge/src/state/__tests__/useStudioEvents.test.tsx new file mode 100644 index 000000000..cdb60b647 --- /dev/null +++ b/apps/agent-forge/src/state/__tests__/useStudioEvents.test.tsx @@ -0,0 +1,125 @@ +import { act } from 'react'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { cleanup, render } from '../../ui/__tests__/render'; +import { useStudioEvents } from '../useStudioEvents'; +import { useAgentDocument } from '../useAgentDocument'; +import type { AgentStore } from '../../persistence/AgentStore'; +import type { StreamConnectionHandlers } from '../../runtime/runtimeClient'; +import type { StudioEvent } from '../../../shared/wireTypes'; + +/** + * `useStudioEvents` (src/state/useStudioEvents.ts) and the + * `agent.changed` -> debounced agents-list refresh `useAgentDocument` + * wires on top of it, against a mocked `runtimeClient.subscribeEvents` + * (the socket itself is covered by server/__tests__/eventsWs.test.ts). + */ + +const captured = vi.hoisted(() => ({ + calls: 0, + onEvent: undefined as ((event: StudioEvent) => void) | undefined, + connection: undefined as StreamConnectionHandlers | undefined, + unsubscribe: vi.fn(), +})); + +vi.mock('../../runtime/runtimeClient', () => ({ + runtimeClient: { + subscribeEvents: (onEvent: (event: StudioEvent) => void, connection?: StreamConnectionHandlers) => { + captured.calls += 1; + captured.onEvent = onEvent; + captured.connection = connection; + return captured.unsubscribe; + }, + }, +})); + +function Probe({ onEvent }: { onEvent: (event: StudioEvent) => void }) { + useStudioEvents(onEvent); + return null; +} + +function DocProbe({ store }: { store: AgentStore }) { + useAgentDocument(store); + return null; +} + +const emit = (event: StudioEvent) => act(() => captured.onEvent?.(event)); + +beforeEach(() => { + vi.useFakeTimers(); + captured.calls = 0; + captured.onEvent = undefined; + captured.connection = undefined; + captured.unsubscribe.mockClear(); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('useStudioEvents', () => { + it('subscribes once and delivers each published event to the callback', () => { + const spy = vi.fn(); + render(); + expect(captured.calls).toBe(1); + const event: StudioEvent = { type: 'agent.changed', agentId: 'a1', reason: 'save' }; + emit(event); + expect(spy).toHaveBeenCalledTimes(1); + expect(spy).toHaveBeenCalledWith(event); + }); + + it('resubscribes 2 s after the socket is lost', () => { + render(); + act(() => captured.connection?.onLost?.()); + act(() => { + vi.advanceTimersByTime(1999); + }); + expect(captured.calls).toBe(1); + act(() => { + vi.advanceTimersByTime(1); + }); + expect(captured.calls).toBe(2); + }); + + it('unsubscribes on unmount so a stale listener receives nothing', () => { + render(); + cleanup(); + expect(captured.unsubscribe).toHaveBeenCalled(); + }); +}); + +describe('agent.changed -> agents list refresh', () => { + it('an agent.changed triggers one refresh after the debounce', async () => { + const store: AgentStore = { + list: vi.fn(async () => []), + load: vi.fn(async () => undefined), + save: vi.fn(async () => {}), + remove: vi.fn(async () => {}), + }; + render(); + await act(async () => { + await Promise.resolve(); + }); + expect(store.list).toHaveBeenCalledTimes(1); // the mount-time load + + // Two agent.changed events inside one 300 ms window coalesce. + emit({ type: 'agent.changed', agentId: 'a1', reason: 'save' }); + emit({ type: 'agent.changed', agentId: 'a2', reason: 'version' }); + act(() => { + vi.advanceTimersByTime(299); + }); + expect(store.list).toHaveBeenCalledTimes(1); + await act(async () => { + vi.advanceTimersByTime(1); + await Promise.resolve(); + }); + expect(store.list).toHaveBeenCalledTimes(2); + + // Other StudioEvent types don't touch the agents list. + emit({ type: 'inbox.count', count: 3 }); + await act(async () => { + vi.advanceTimersByTime(500); + await Promise.resolve(); + }); + expect(store.list).toHaveBeenCalledTimes(2); + }); +}); diff --git a/apps/agent-forge/src/state/useAgentDocument.ts b/apps/agent-forge/src/state/useAgentDocument.ts index cb4dbf53a..d65e7aac6 100644 --- a/apps/agent-forge/src/state/useAgentDocument.ts +++ b/apps/agent-forge/src/state/useAgentDocument.ts @@ -6,8 +6,12 @@ import { specToGraph } from '../graph/specToGraph'; import type { AgentGraphSpec } from '../graph/types'; import { graphFromTemplate, type TemplateId } from '../canvas/templates'; import { agentIdProblem } from '../../shared/agentId'; +import type { StudioEvent } from '../../shared/wireTypes'; +import { useStudioEvents } from './useStudioEvents'; const AUTOSAVE_DEBOUNCE_MS = 800; +/** How long `agent.changed` events collect before the agents list re-fetches - a burst (rename = save+delete) is one refresh. */ +const AGENTS_REFRESH_DEBOUNCE_MS = 300; const DEFAULT_SPEC: AgentSpec = { name: 'untitled-agent', @@ -159,6 +163,34 @@ async function deleteSavedAgent( await refreshAgents(); } +/** + * The `WS /events` handler `useAgentDocument` subscribes: an `agent.changed` + * (any workspace write - another tab's save, a version capture, a draft + * approval) re-fetches the agents list, debounced so a burst of writes is + * one `refreshAgents()`. Other event types are for the later summary, + * draft, suggestion and inbox consumers. + */ +function useAgentChangedRefresh(refreshAgents: () => Promise): (event: StudioEvent) => void { + const timer = useRef | undefined>(undefined); + useEffect( + () => () => { + if (timer.current !== undefined) clearTimeout(timer.current); + }, + [] + ); + return useCallback( + (event: StudioEvent) => { + if (event.type !== 'agent.changed') return; + if (timer.current !== undefined) clearTimeout(timer.current); + timer.current = setTimeout(() => { + timer.current = undefined; + void refreshAgents(); + }, AGENTS_REFRESH_DEBOUNCE_MS); + }, + [refreshAgents] + ); +} + /** * The saved-agent list and its refresh; on mount it loads the list and opens * the first saved agent (`initialId` when it is saved). @@ -273,6 +305,8 @@ export function useAgentDocument(store: AgentStore, initialId?: string) { const lastValidSpec = useRef(DEFAULT_SPEC); const doc: DocumentSetters = { setAgentId, setGraphState, lastValidSpec, setSelectedNodeId, setDirty }; const { agents, setAgents, refreshAgents, agentsLoaded } = useAgentList(store, initialId ?? agentId, doc); + // `WS /events`: another tab's write refreshes this tab's agents list. + useStudioEvents(useAgentChangedRefresh(refreshAgents)); const spec = useMemo(() => { const derived = deriveSpec(graph, lastValidSpec.current); diff --git a/apps/agent-forge/src/state/useStudioEvents.ts b/apps/agent-forge/src/state/useStudioEvents.ts new file mode 100644 index 000000000..879734da5 --- /dev/null +++ b/apps/agent-forge/src/state/useStudioEvents.ts @@ -0,0 +1,39 @@ +import { useEffect, useRef, useState } from 'react'; +import { runtimeClient } from '../runtime/runtimeClient'; +import type { StudioEvent } from '../../shared/wireTypes'; + +/** How long to wait before re-opening a dropped `/events` socket - same as useAgentStream's stream reconnect. */ +const RECONNECT_DELAY_MS = 2000; + +/** + * One `WS /events` subscription for the whole app: `onEvent` sees every + * `StudioEvent` the server publishes. The socket reconnects + * `RECONNECT_DELAY_MS` after it drops (the same pattern + * `useAgentStream` uses for `/agents/:id/stream`); a missed event only + * means a stale list until the next one - the REST reads it refreshes + * stay the source of truth. + * + * `onEvent` may be a fresh closure each render - the subscription is + * opened once and always calls the latest callback rather than + * re-subscribing when its identity changes. + */ +export function useStudioEvents(onEvent: (event: StudioEvent) => void): void { + const onEventRef = useRef(onEvent); + useEffect(() => { + onEventRef.current = onEvent; + }); + // `attempt` bumps on each socket loss so the effect re-subscribes. + const [attempt, setAttempt] = useState(0); + useEffect(() => { + let retry: ReturnType | undefined; + const unsubscribe = runtimeClient.subscribeEvents((event) => onEventRef.current(event), { + onLost: () => { + retry = setTimeout(() => setAttempt((n) => n + 1), RECONNECT_DELAY_MS); + }, + }); + return () => { + if (retry) clearTimeout(retry); + unsubscribe(); + }; + }, [attempt]); +} diff --git a/apps/agent-forge/vite.config.ts b/apps/agent-forge/vite.config.ts index 61952482c..a3a160ce6 100644 --- a/apps/agent-forge/vite.config.ts +++ b/apps/agent-forge/vite.config.ts @@ -47,6 +47,12 @@ export default defineConfig({ }, // The SDK's shipped templates (list + detail). '/templates': { target: process.env.LOUSHO_STUDIO_API_URL ?? 'http://127.0.0.1:4750', changeOrigin: true }, + // The Studio event bus (WS /events). + '/events': { + target: process.env.LOUSHO_STUDIO_API_URL ?? 'http://127.0.0.1:4750', + changeOrigin: true, + ws: true, + }, }, }, test: {