diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 3c2428c2dc..0166d460f4 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1417,9 +1417,8 @@ export async function createPaseoDaemon( serviceProxyPublicBaseUrl, resolveScriptHealth: (hostname) => scriptHealthMonitor.getHealthForHostname(hostname), logger, - // MCP operations do not belong to one WebSocket session, so lifecycle - // status updates fan out to every connected client. emit: (message) => wsServer?.broadcast(wrapSessionMessage(message)), + publishStatusUpdate: (message) => wsServer?.publishScriptStatusUpdate(message), spawnWorkspaceScript, assertAutomationAllowed: (workspaceId) => assertWorkspaceAutomationAllowedForWorkspace(workspaceRegistry, workspaceId), diff --git a/packages/server/src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts b/packages/server/src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts index 9c27d2039c..19ccdf45e8 100644 --- a/packages/server/src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts @@ -192,3 +192,55 @@ test("listing by workspace ID keeps terminals in a shared directory separate", a expect.objectContaining({ id: created.terminal?.id, workspaceId: second, cwd }), ]); }); + +test("script status started and stopped by one client reaches every subscribed client", async () => { + await writeFile( + path.join(cwd, "paseo.json"), + JSON.stringify({ + scripts: { sleeper: { type: "script", command: 'node -e "setInterval(() => {}, 60000)"' } }, + }), + ); + const observer = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + const bystander = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + try { + await observer.connect(); + await bystander.connect(); + const bystanderEvents: string[] = []; + bystander.subscribeRawMessages((message) => bystanderEvents.push(message.type)); + const opened = await client.openProject(cwd); + expect(opened.error).toBeNull(); + const workspaceId = opened.workspace!.id; + const actorSeen = await observeScriptLifecycles(client, "sleeper"); + const observerSeen = await observeScriptLifecycles(observer, "sleeper"); + + expect((await client.startWorkspaceScript(workspaceId, "sleeper")).error).toBeNull(); + await expect.poll(() => actorSeen.at(-1)).toBe("running"); + await expect.poll(() => observerSeen.at(-1)).toBe("running"); + + expect((await client.stopWorkspaceScript(workspaceId, "sleeper")).error).toBeNull(); + await expect.poll(() => actorSeen.at(-1)).toBe("stopped"); + await expect.poll(() => observerSeen.at(-1)).toBe("stopped"); + expect(bystanderEvents).not.toContain("script_status_update"); + } finally { + await observer.close(); + await bystander.close(); + } +}, 60_000); + +async function observeScriptLifecycles( + daemonClient: DaemonClient, + scriptName: string, +): Promise { + const lifecycles: string[] = []; + const subscription = daemonClient.observeEvents(["script_status_update"]); + subscription.subscribe({ + snapshot: () => {}, + update: (message) => { + if (message.type !== "script_status_update") return; + const script = message.payload.scripts.find((entry) => entry.scriptName === scriptName); + if (script) lifecycles.push(script.lifecycle); + }, + }); + await subscription.ready; + return lifecycles; +} diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 1307cf808f..4e57ee0578 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -26,6 +26,7 @@ import { type FirstAgentContext, type SessionInboundMessage, type SessionOutboundMessage, + type ScriptStatusUpdateMessage, type GitSetupOptions, type StartWorkspaceScriptRequest, type WorkspaceScriptListRequest, @@ -450,6 +451,7 @@ export interface SessionOptions { getTransportBufferedAmount?: (source?: object) => number | null; onLifecycleIntent?: (intent: SessionLifecycleIntent) => void; onWorkspaceRecovered?: (workspace: PersistedWorkspaceRecord) => Promise; + publishScriptStatusUpdate?: (message: ScriptStatusUpdateMessage) => void; logger: pino.Logger; downloadTokenStore: DownloadTokenStore; pushNotifications: PushNotifications; @@ -810,6 +812,7 @@ export class Session { getTransportBufferedAmount, onLifecycleIntent, onWorkspaceRecovered, + publishScriptStatusUpdate, logger, downloadTokenStore, pushNotifications, @@ -1148,8 +1151,11 @@ export class Session { resolveScriptHealth: this.resolveScriptHealth, logger: this.sessionLogger, emit: (message) => this.emit(message), + publishStatusUpdate: (message) => { + if (publishScriptStatusUpdate) publishScriptStatusUpdate(message); + else this.emit(message); + }, spawnWorkspaceScript, - wantsStatusUpdates: () => this.wantsEvent("script_status_update"), assertAutomationAllowed: (workspaceId) => assertWorkspaceAutomationAllowedForWorkspace(this.workspaceRegistry, workspaceId), globalServicePorts: loadPersistedConfig(this.paseoHome).worktrees?.servicePorts, diff --git a/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.test.ts b/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.test.ts index 0ba8a0169b..aa366a3495 100644 --- a/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.test.ts +++ b/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.test.ts @@ -3,7 +3,11 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { pino } from "pino"; import { afterEach, describe, expect, test } from "vitest"; -import type { SessionOutboundMessage, StartWorkspaceScriptRequest } from "../../messages.js"; +import type { + ScriptStatusUpdateMessage, + SessionOutboundMessage, + StartWorkspaceScriptRequest, +} from "../../messages.js"; import { createServiceProxySubsystem, type ServiceProxySubsystem } from "../../service-proxy.js"; import type { TerminalManager } from "../../../terminal/terminal-manager.js"; import type { @@ -82,6 +86,7 @@ interface BuildOptions { function buildService(options: BuildOptions = {}) { const emitted: SessionOutboundMessage[] = []; + const published: ScriptStatusUpdateMessage[] = []; const spawnCalls: SpawnWorkspaceScriptOptions[] = []; const workspace = options.workspace === undefined @@ -108,6 +113,7 @@ function buildService(options: BuildOptions = {}) { resolveScriptHealth: null, logger, emit: (message) => emitted.push(message), + publishStatusUpdate: (message) => published.push(message), async spawnWorkspaceScript(spawnOptions): Promise { spawnCalls.push(spawnOptions); if (options.spawnThrows) { @@ -126,7 +132,7 @@ function buildService(options: BuildOptions = {}) { }, }); - return { service, emitted, spawnCalls }; + return { service, emitted, published, spawnCalls }; } const request: StartWorkspaceScriptRequest = { @@ -215,10 +221,11 @@ describe("buildSnapshot", () => { }); describe("emitStatusUpdate", () => { - test("emits one script_status_update carrying the snapshot", async () => { - const { service, emitted } = buildService(); + test("publishes one script_status_update carrying the snapshot", async () => { + const { service, emitted, published } = buildService(); await service.emitStatusUpdate("ws-1", "/tmp/repo"); - expect(emitted).toEqual([ + expect(emitted).toEqual([]); + expect(published).toEqual([ { type: "script_status_update", payload: { workspaceId: "ws-1", scripts: [] } }, ]); }); @@ -332,7 +339,7 @@ describe("start", () => { }); test("spawns the script with resolved git metadata and reports success", async () => { - const { service, emitted, spawnCalls } = buildService(); + const { service, emitted, published, spawnCalls } = buildService(); await service.start(request); expect(spawnCalls).toHaveLength(1); @@ -345,7 +352,7 @@ describe("start", () => { daemonPort: 6767, daemonListenHost: "127.0.0.1", }); - expect(emitted).toContainEqual({ + expect(published).toContainEqual({ type: "script_status_update", payload: { workspaceId: "ws-1", scripts: [] }, }); diff --git a/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.ts b/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.ts index 2116e82b11..d449a48855 100644 --- a/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.ts +++ b/packages/server/src/server/session/workspace-scripts/workspace-scripts-service.ts @@ -1,5 +1,6 @@ import type pino from "pino"; import type { + ScriptStatusUpdateMessage, SessionOutboundMessage, StartWorkspaceScriptRequest, WorkspaceDescriptorPayload, @@ -31,7 +32,7 @@ type WorkspaceScriptsPayload = WorkspaceDescriptorPayload["scripts"]; /** * The service-proxy-backed scripts a workspace exposes: build the scripts payload - * snapshot, emit a script_status_update to clients, and start a script. + * snapshot, publish a script_status_update to every client, and start a script. * * The workspace descriptor builder, the script-status emission path, and the * start-script RPC all funnel through one assembly of buildWorkspaceScriptPayloads' @@ -66,7 +67,8 @@ export function createWorkspaceScriptsService(deps: { globalServicePorts?: PaseoServicePortAllocation; logger: pino.Logger; emit: (message: SessionOutboundMessage) => void; - wantsStatusUpdates?: () => boolean; + // Script status is daemon state, so it reaches every client, not only the requester. + publishStatusUpdate: (message: ScriptStatusUpdateMessage) => void; spawnWorkspaceScript: (options: SpawnWorkspaceScriptOptions) => Promise; assertAutomationAllowed: (workspaceId: string) => Promise; }): WorkspaceScriptsService { @@ -84,6 +86,7 @@ export function createWorkspaceScriptsService(deps: { globalServicePorts, logger, emit, + publishStatusUpdate, spawnWorkspaceScript, assertAutomationAllowed, } = deps; @@ -130,12 +133,11 @@ export function createWorkspaceScriptsService(deps: { } async function emitStatusUpdate(workspaceId: string, _workspaceDirectory: string): Promise { - if (deps.wantsStatusUpdates && !deps.wantsStatusUpdates()) return; try { const workspace = await workspaceRegistry.get(workspaceId); if (!workspace) return; const project = await projectRegistry.get(workspace.projectId); - emit({ + publishStatusUpdate({ type: "script_status_update", payload: { workspaceId, scripts: buildSnapshot(workspace, project) }, }); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 67c78184fb..4dcdc680ab 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -20,6 +20,7 @@ import type { CheckoutDiffManager, CheckoutDiffMetrics } from "./checkout-diff-m import type { DaemonConfigStore, MutableDaemonConfig } from "./daemon-config-store.js"; import { type ServerInfoStatusPayload, + type ScriptStatusUpdateMessage, type SessionOutboundMessage, type WorkspaceSetupSnapshot, type WSHelloMessage, @@ -978,6 +979,12 @@ export class VoiceAssistantWebSocketServer { } } + public publishScriptStatusUpdate(message: ScriptStatusUpdateMessage): void { + for (const session of this.listSessions()) { + session.emitServerMessage(message); + } + } + public publishSpeechReadiness(readiness: SpeechReadinessSnapshot | null): void { this.updateServerCapabilities(buildServerCapabilities({ readiness })); } @@ -1465,6 +1472,7 @@ export class VoiceAssistantWebSocketServer { ), ); }, + publishScriptStatusUpdate: (message) => this.publishScriptStatusUpdate(message), downloadTokenStore: this.downloadTokenStore, pushNotifications: this.pushNotifications, paseoHome: this.paseoHome,