diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index e4697dfaf8..44b9189425 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -762,6 +762,8 @@ test("advertises client capabilities in hello", async () => { reasoning_merge_enum: true, terminal_reflowable_snapshot: true, timeline_notifications: true, + plugin_timeline_items: true, + workspace_setup_blocked: true, browser_host: { supportedCommands: ["list_tabs"], hostKind: "desktop app", diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index f67a4e7023..576e111a71 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -5681,6 +5681,8 @@ export class DaemonClient { ? { [CLIENT_CAPS.providerSnapshotReferences]: true } : {}), [CLIENT_CAPS.timelineNotifications]: true, + [CLIENT_CAPS.pluginTimelineItems]: true, + [CLIENT_CAPS.workspaceSetupBlocked]: true, ...this.config.capabilities, }, ...(this.config.appVersion ? { appVersion: this.config.appVersion } : {}), diff --git a/packages/protocol/src/client-capabilities.ts b/packages/protocol/src/client-capabilities.ts index d07ffdf242..f15c3ea7d9 100644 --- a/packages/protocol/src/client-capabilities.ts +++ b/packages/protocol/src/client-capabilities.ts @@ -34,6 +34,10 @@ export const CLIENT_CAPS = { // timeline items for older clients whose strict timeline union rejects them. // Remove after 2027-03-03 once the supported client floor is >= v0.7.2. timelineNotifications: "timeline_notifications", + // COMPAT(pluginTimelineItems): added in v0.8.0, remove after 2027-03-07 once client floor >= v0.8.0. + pluginTimelineItems: "plugin_timeline_items", + // COMPAT(workspaceSetupBlocked): added in v0.8.0, remove after 2027-03-07 once client floor >= v0.8.0. + workspaceSetupBlocked: "workspace_setup_blocked", browserHost: "browser_host", } as const; diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 3c8d09d5b9..6be32a3525 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -3404,9 +3404,9 @@ export const ServerInfoStatusPayloadSchema = z // COMPAT(providersSnapshot): added in v0.1.48, remove gating when all clients use snapshot features: z .object({ - // COMPAT(agentRequestReceipts): added in v0.7.3; remove gate after 2027-03-05. + // COMPAT(agentRequestReceipts): added in v0.8.0; remove gate after 2027-03-05. agentRequestReceipts: z.boolean().optional(), - // COMPAT(hubAgentRpc): added in v0.7.3; remove gate after 2027-03-05. + // COMPAT(hubAgentRpc): added in v0.8.0; remove gate after 2027-03-05. hubAgentRpc: z.boolean().optional(), providersSnapshot: z.boolean().optional(), // COMPAT(providersSnapshotCwd): added in v0.3.2, remove gate after 2027-02-10. @@ -3415,9 +3415,9 @@ export const ServerInfoStatusPayloadSchema = z directorySync: z.boolean().optional(), // COMPAT(workspaceLabels): added in v0.5.0, remove after 2027-08-14. workspaceLabels: z.boolean().optional(), - // COMPAT(workspaceSetupRun): added in v0.7.3, remove gate after 2027-09-02. + // COMPAT(workspaceSetupRun): added in v0.8.0, remove gate after 2027-09-02. workspaceSetupRun: z.boolean().optional(), - // COMPAT(workspaceTerminals): added in v0.7.3, remove gate after 2027-09-05. + // COMPAT(workspaceTerminals): added in v0.8.0, remove gate after 2027-09-05. workspaceTerminals: z.boolean().optional(), // COMPAT(checkoutForgeSetAutoMerge): added in v0.2.0-beta.1. Remove the // feature gate and checkoutGithubSetAutoMerge fallback after 2027-01-17 @@ -3527,7 +3527,7 @@ export const ServerInfoStatusPayloadSchema = z providerRemoval: z.boolean().optional(), // COMPAT(importSessionWorkspaceTarget): added in v0.1.110, remove gate after 2027-01-16. importSessionWorkspaceTarget: z.boolean().optional(), - // COMPAT(importSessionSearch): added in v0.7.3, remove gate after 2027-03-02. + // COMPAT(importSessionSearch): added in v0.8.0, remove gate after 2027-03-02. importSessionSearch: z.boolean().optional(), // COMPAT(forgeProviders): added in v0.2.0-beta.1. Drop the gate after // 2027-01-17 once the supported daemon floor is >= v0.2.0. diff --git a/packages/protocol/src/messages.wire-compat.test.ts b/packages/protocol/src/messages.wire-compat.test.ts index bcfc37d5f8..fe2a79916f 100644 --- a/packages/protocol/src/messages.wire-compat.test.ts +++ b/packages/protocol/src/messages.wire-compat.test.ts @@ -6,6 +6,9 @@ import { ServerInfoStatusPayloadSchema, SessionOutboundMessageSchema, WSHelloMessageSchema, + WorkspaceSetupSnapshotSchema, + WorkspaceSetupProgressMessageSchema, + AgentTimelineEntryPayloadSchema, } from "./messages.js"; test("terminal listings accept older rows and retain new per-terminal directories", () => { @@ -302,3 +305,87 @@ describe("wire schema compatibility", () => { }); }); }); + +test("0.8 timeline and setup capabilities remain optional in the hello", () => { + const hello = { type: "hello", clientId: "compat", clientType: "mobile", protocolVersion: 1 }; + expect(WSHelloMessageSchema.safeParse(hello).success).toBe(true); + expect( + WSHelloMessageSchema.parse({ + ...hello, + capabilities: { plugin_timeline_items: true, workspace_setup_blocked: true }, + }).capabilities, + ).toEqual({ plugin_timeline_items: true, workspace_setup_blocked: true }); +}); + +test("plugin rows and identity merges require a capable receiver", () => { + const item = { + type: "plugin", + id: "task", + pluginId: "tasks", + kind: "tasks", + version: 1, + data: { text: "Working" }, + }; + expect(AgentTimelineItemPayloadSchema.parse(item)).toEqual(item); + const legacyItems = z.object({ + type: z.enum([ + "user_message", + "assistant_message", + "reasoning", + "tool_call", + "todo", + "error", + "notification", + "compaction", + ]), + }); + expect(legacyItems.safeParse(item).success).toBe(false); + const entry = { + provider: "codex", + item, + timestamp: "2026-09-07T00:00:00.000Z", + seqStart: 1, + seqEnd: 2, + sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }], + collapsed: ["identity"], + }; + expect(AgentTimelineEntryPayloadSchema.parse(entry)).toEqual(entry); + const legacyCollapsed = z.array(z.enum(["assistant_merge", "reasoning_merge", "tool_lifecycle"])); + expect(legacyCollapsed.safeParse(entry.collapsed).success).toBe(false); +}); + +test("blocked setup preserves the legacy failed shape and optional provenance", () => { + const snapshot = { + status: "blocked", + error: null, + detail: { + type: "worktree_setup", + worktreePath: "/workspace", + branchName: "fork", + log: "", + commands: [], + }, + blockedSource: { + kind: "change_request", + forge: "github", + number: 42, + headRepository: "contributor/project", + }, + }; + expect(WorkspaceSetupSnapshotSchema.parse(snapshot)).toEqual(snapshot); + const legacyStatus = z.enum(["running", "completed", "failed"]); + const legacySnapshot = WorkspaceSetupSnapshotSchema.omit({ blockedSource: true }).extend({ + status: legacyStatus, + }); + expect(legacySnapshot.safeParse(snapshot).success).toBe(false); + const failed = { ...snapshot, status: "failed", error: "Update Paseo to review and run setup." }; + expect(legacySnapshot.safeParse(failed).success).toBe(true); + const progress = { + type: "workspace_setup_progress", + payload: { ...failed, workspaceId: "workspace" }, + }; + expect(WorkspaceSetupProgressMessageSchema.parse(progress)).toEqual(progress); + expect(WorkspaceSetupSnapshotSchema.parse(legacySnapshot.parse(failed))).toEqual( + legacySnapshot.parse(failed), + ); +}); diff --git a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts index 2bbe48662a..a06b7a32c8 100644 --- a/packages/server/src/server/selective-timeline-delivery.e2e.test.ts +++ b/packages/server/src/server/selective-timeline-delivery.e2e.test.ts @@ -1,9 +1,17 @@ +import type { AgentStreamEvent, AgentSessionConfig } from "./agent/agent-sdk-types.js"; +import { mkdtemp, mkdir, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createPersistedWorkspaceRecord } from "./workspace-registry.js"; import { afterEach, beforeEach, expect, test } from "vitest"; import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities"; import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { DaemonClient } from "./test-utils/daemon-client.js"; import { createTestPaseoDaemon, type TestPaseoDaemon } from "./test-utils/paseo-daemon.js"; -import { MockLoadTestAgentClient } from "./agent/providers/mock-load-test-agent.js"; +import { + MockLoadTestAgentClient, + MockLoadTestAgentSession, +} from "./agent/providers/mock-load-test-agent.js"; interface MessageWaiter { predicate(message: SessionOutboundMessage): boolean; @@ -130,12 +138,16 @@ async function connect(input: { selective: boolean; timelineReplacementInvalidation?: boolean; timelineNotifications?: boolean; + pluginTimelineItems?: boolean; + workspaceSetupBlocked?: boolean; }): Promise { const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, clientId: input.clientId, capabilities: { [CLIENT_CAPS.selectiveAgentTimeline]: input.selective, + [CLIENT_CAPS.pluginTimelineItems]: input.pluginTimelineItems ?? false, + [CLIENT_CAPS.workspaceSetupBlocked]: input.workspaceSetupBlocked ?? false, ...(input.timelineNotifications === undefined ? {} : { [CLIENT_CAPS.timelineNotifications]: input.timelineNotifications }), @@ -228,6 +240,94 @@ test("notification timeline items are sent only to clients that advertise suppor expect(legacyTimeline.endCursor).toEqual(capableTimeline.endCursor); }); +test("plugin timeline items are sent only to clients that advertise support", async () => { + await daemon.close(); + daemon = await createTestPaseoDaemon({ + isDev: true, + agentClients: { mock: new MockLoadTestAgentClient() }, + }); + const capable = await connect({ + clientId: "plugin-shared", + selective: false, + pluginTimelineItems: true, + }); + const legacy = await connect({ + clientId: "plugin-shared", + selective: false, + pluginTimelineItems: false, + }); + const agent = await capable.client.createAgent({ + provider: "mock", + cwd: "/tmp", + title: "Plugin compatibility", + model: "ten-second-stream", + }); + capable.clear(); + legacy.clear(); + + await daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "plugin", + id: "compat-row", + pluginId: "timeline-items", + kind: "notice", + version: 1, + data: { text: "Capable clients only" }, + }); + await daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "plugin", + id: "compat-row", + pluginId: "timeline-items", + kind: "notice", + version: 1, + data: { text: "Updated" }, + }); + await daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "assistant_message", + text: "Visible to every client", + }); + await Promise.all([ + capable.next(isAgentStream(agent.id), "capable timeline delivery"), + legacy.next(isAgentStream(agent.id), "legacy timeline delivery"), + ]); + await Promise.all([capable.barrier("plugin-capable"), legacy.barrier("plugin-legacy")]); + + const capableLiveItems = capable.messages.flatMap((message) => + message.type === "agent_stream" && message.payload.event.type === "timeline" + ? [message.payload.event.item] + : [], + ); + const legacyLiveItems = legacy.messages.flatMap((message) => + message.type === "agent_stream" && message.payload.event.type === "timeline" + ? [message.payload.event.item] + : [], + ); + expect(capableLiveItems.some((item) => item.type === "plugin")).toBe(true); + expect(legacyLiveItems.some((item) => item.type === "plugin")).toBe(false); + expect(legacyLiveItems).toContainEqual( + expect.objectContaining({ type: "assistant_message", text: "Visible to every client" }), + ); + + for (const projection of ["canonical", "projected"] as const) { + const [capableTimeline, legacyTimeline] = await Promise.all([ + capable.client.fetchAgentTimeline(agent.id, { direction: "tail", projection }), + legacy.client.fetchAgentTimeline(agent.id, { direction: "tail", projection }), + ]); + expect(capableTimeline.entries.some((entry) => entry.item.type === "plugin")).toBe(true); + expect(legacyTimeline.entries.some((entry) => entry.item.type === "plugin")).toBe(false); + expect(legacyTimeline.entries).toContainEqual( + expect.objectContaining({ + item: expect.objectContaining({ + type: "assistant_message", + text: "Visible to every client", + }), + }), + ); + expect(legacyTimeline.window).toEqual(capableTimeline.window); + expect(legacyTimeline.endCursor).toEqual(capableTimeline.endCursor); + expect(legacyTimeline.entries.flatMap((entry) => entry.collapsed)).not.toContain("identity"); + } +}); + test("rewind routes replacement completion by source capability and subscription", async () => { await daemon.close(); daemon = await createTestPaseoDaemon({ @@ -320,12 +420,14 @@ test("subscription acknowledgements stay on the requesting socket of a retained test("real WebSocket sessions enforce selective delivery, retained resets, downgrade, and dedicated attention", async () => { const legacy = await connect({ clientId: "legacy-client", selective: false }); let capable = await connect({ clientId: "capable-client", selective: true }); + const workspaceId = await createAttentionWorkspace(legacy.client); const agents = await Promise.all( ["A", "B", "C"].map((title) => legacy.client.createAgent({ provider: "codex", - cwd: "/tmp", + cwd: daemon.paseoHome, title: `Selective ${title}`, + workspaceId, modeId: "full-access", }), ), @@ -397,6 +499,10 @@ test("real WebSocket sessions enforce selective delivery, retained resets, downg await capable.barrier("resumed-membership-reset"); expect(capable.hasTimeline(agentB.id)).toBe(false); + await Promise.all([ + legacy.client.fetchAgents({ subscribe: { subscriptionId: "legacy-directory" } }), + capable.client.fetchAgents({ subscribe: { subscriptionId: "capable-directory" } }), + ]); capable.client.sendHeartbeat({ deviceType: "mobile", focusedAgentId: null, @@ -451,3 +557,190 @@ test("real WebSocket sessions enforce selective delivery, retained resets, downg expect(downgradedDelivery.type).toBe("agent_stream"); }, 30_000); + +test("blocked setup remains readable on mixed-capability sockets sharing a session", async () => { + await daemon.close(); + const root = await mkdtemp(path.join(os.tmpdir(), "paseo-blocked-compat-")); + const projects = path.join(root, ".paseo", "projects"); + await mkdir(projects, { recursive: true }); + const workspace = createPersistedWorkspaceRecord({ + workspaceId: "blocked-workspace", + projectId: "project", + cwd: root, + kind: "directory", + displayName: "Fork PR", + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + untrustedSource: { + kind: "change_request", + forge: "github", + number: 42, + headRepository: "contributor/project", + }, + }); + await writeFile(path.join(projects, "workspaces.json"), JSON.stringify([workspace])); + daemon = await createTestPaseoDaemon({ paseoHomeRoot: root }); + const legacy = await connect({ clientId: "setup-shared", selective: false }); + const capable = await connect({ + clientId: "setup-shared", + selective: false, + workspaceSetupBlocked: true, + }); + const oldStatus = await legacy.client.fetchWorkspaceSetupStatus(workspace.workspaceId); + const newStatus = await capable.client.fetchWorkspaceSetupStatus(workspace.workspaceId); + expect(oldStatus.snapshot).toMatchObject({ + status: "failed", + error: + "Workspace setup is blocked pending approval of code from a fork pull request. Update Paseo to review and run setup.", + }); + expect(newStatus.snapshot).toMatchObject({ + status: "blocked", + error: null, + blockedSource: workspace.untrustedSource, + }); + expect(oldStatus.snapshot?.detail).toEqual(newStatus.snapshot?.detail); +}); + +// A provider adapter with controllable events and durable provider history. +class CompatibilityProviderSession extends MockLoadTestAgentSession { + private readonly observers = new Set<(event: AgentStreamEvent) => void>(); + private readonly injectedHistory: AgentStreamEvent[] = []; + override subscribe(callback: (event: AgentStreamEvent) => void): () => void { + const unsubscribe = super.subscribe(callback); + this.observers.add(callback); + return () => { + unsubscribe(); + this.observers.delete(callback); + }; + } + push(event: AgentStreamEvent): void { + this.injectedHistory.push(event); + for (const callback of this.observers) callback(event); + } + override async *streamHistory(): AsyncGenerator { + yield* this.injectedHistory; + yield* super.streamHistory(); + } +} + +class CompatibilityProvider extends MockLoadTestAgentClient { + session!: CompatibilityProviderSession; + override async createSession(config: AgentSessionConfig): Promise { + this.session = new CompatibilityProviderSession({ + config, + sessionId: "compat-provider-session", + }); + return this.session; + } +} + +test("plugin items are gated in provider child streams, child fetches, and rewind replay", async () => { + await daemon.close(); + const provider = new CompatibilityProvider(); + daemon = await createTestPaseoDaemon({ isDev: true, agentClients: { mock: provider } }); + const legacy = await connect({ clientId: "provider-shared", selective: false }); + const capable = await connect({ + clientId: "provider-shared", + selective: false, + pluginTimelineItems: true, + }); + const agent = await capable.client.createAgent({ + provider: "mock", + cwd: "/tmp", + model: "ten-second-stream", + }); + const plugin = { + type: "plugin" as const, + id: "provider-row", + pluginId: "provider", + kind: "notice", + version: 1, + data: { text: "Provider notice" }, + }; + provider.session.push({ type: "timeline", provider: "mock", item: plugin }); + provider.session.push({ + type: "provider_subagent", + provider: "mock", + event: { type: "upsert", id: "child", title: "Child", status: "running" }, + }); + provider.session.push({ + type: "provider_subagent", + provider: "mock", + event: { type: "timeline", id: "child", item: plugin }, + }); + provider.session.push({ + type: "provider_subagent", + provider: "mock", + event: { + type: "timeline", + id: "child", + item: { type: "assistant_message", text: "Child result" }, + }, + }); + function childResult(message: SessionOutboundMessage): boolean { + return ( + message.type === "agent.provider_subagents.update" && + message.payload.kind === "timeline" && + message.payload.item.type === "assistant_message" + ); + } + await Promise.all([ + capable.next(childResult, "capable child result"), + legacy.next(childResult, "legacy child result"), + ]); + function childItems(client: ConnectedClient) { + return client.messages.flatMap((message) => + message.type === "agent.provider_subagents.update" && message.payload.kind === "timeline" + ? [message.payload.item] + : [], + ); + } + expect(childItems(capable)).toContainEqual(plugin); + expect(childItems(legacy)).toEqual([{ type: "assistant_message", text: "Child result" }]); + const oldChild = await legacy.client.fetchProviderSubagentTimeline(agent.id, "child"); + const newChild = await capable.client.fetchProviderSubagentTimeline(agent.id, "child"); + expect(newChild.rows.map((row) => row.item)).toContainEqual(plugin); + expect(oldChild.rows.map((row) => row.item)).toEqual([ + { type: "assistant_message", text: "Child result" }, + ]); + expect(oldChild.window).toEqual(newChild.window); + + await capable.client.sendMessage(agent.id, "Rewind target"); + await capable.client.cancelAgent(agent.id); + const timeline = await capable.client.fetchAgentTimeline(agent.id, { + direction: "tail", + projection: "canonical", + }); + const targetMessageId = rewindMessageId(timeline); + capable.clear(); + legacy.clear(); + await capable.client.rewindAgent(agent.id, targetMessageId, "conversation"); + await Promise.all([capable.barrier("provider-rewind"), legacy.barrier("provider-rewind")]); + function replayItems(client: ConnectedClient) { + return client.messages.flatMap((message) => + message.type === "agent_stream" && message.payload.event.type === "timeline" + ? [message.payload.event.item] + : [], + ); + } + expect(replayItems(capable)).toContainEqual(plugin); + expect(replayItems(legacy).some((item) => item.type === "plugin")).toBe(false); + expect(replayItems(legacy)).toContainEqual(expect.objectContaining({ type: "user_message" })); +}); + +async function createAttentionWorkspace(client: DaemonClient): Promise { + const result = await client.createWorkspace({ + source: { kind: "directory", path: daemon.paseoHome }, + }); + if (!result.workspace) throw new Error(result.error ?? "Expected workspace"); + return result.workspace.id; +} + +function rewindMessageId( + timeline: Awaited>, +): string { + const target = timeline.entries.find((entry) => entry.item.type === "user_message"); + if (target?.item.type !== "user_message" || !target.item.messageId) + throw new Error("Expected rewind target"); + return target.item.messageId; +} diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 6a6d9ba030..3bda1ec7cc 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1175,20 +1175,23 @@ export class Session { return true; } - private supportsTimelineNotifications(source?: object): boolean { - return source - ? this.supportsForSource(CLIENT_CAPS.timelineNotifications, source) - : this.supports(CLIENT_CAPS.timelineNotifications); + // COMPAT(timelineItemCapabilities): plugin items added in v0.8.0, notifications in v0.7.2. + // Remove after 2027-03-07 once the supported client floor is >= v0.8.0. + private supportsTimelineItem(item: { type: string }, source?: object): boolean { + let capability: ClientCapability; + if (item.type === "notification") capability = CLIENT_CAPS.timelineNotifications; + else if (item.type === "plugin") capability = CLIENT_CAPS.pluginTimelineItems; + else return true; + return source ? this.supportsForSource(capability, source) : this.supports(capability); } private forwardAgentStream( event: Extract, serializedEvent: Extract["payload"]["event"], ): void { - const isNotification = - serializedEvent.type === "timeline" && serializedEvent.item.type === "notification"; if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { - if (isNotification && !this.supportsTimelineNotifications()) return; + if (serializedEvent.type === "timeline" && !this.supportsTimelineItem(serializedEvent.item)) + return; if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") { this.emit({ type: "agent_attention_required", @@ -1213,7 +1216,11 @@ export class Session { } for (const [source, capabilities] of this.clientCapabilitiesBySource) { - if (isNotification && !capabilities.has(CLIENT_CAPS.timelineNotifications)) continue; + if ( + serializedEvent.type === "timeline" && + !this.supportsTimelineItem(serializedEvent.item, source) + ) + continue; const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline); if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") { this.onMessageToSource(source, { @@ -1679,11 +1686,10 @@ export class Session { }; } - const isNotification = update.type === "timeline" && update.row.item.type === "notification"; if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { if ( this.supports(CLIENT_CAPS.providerSubagents) && - (!isNotification || this.supportsTimelineNotifications()) + (update.type !== "timeline" || this.supportsTimelineItem(update.row.item)) ) { this.emit(message); } @@ -1691,7 +1697,8 @@ export class Session { } for (const [source, capabilities] of this.clientCapabilitiesBySource) { if (!capabilities.has(CLIENT_CAPS.providerSubagents)) continue; - if (isNotification && !capabilities.has(CLIENT_CAPS.timelineNotifications)) continue; + if (update.type === "timeline" && !this.supportsTimelineItem(update.row.item, source)) + continue; this.onMessageToSource(source, message); } } @@ -4128,7 +4135,7 @@ export class Session { source?: object, ): void { for (const row of rows) { - if (row.item.type === "notification" && !this.supportsTimelineNotifications(source)) { + if (!this.supportsTimelineItem(row.item, source)) { continue; } const event = serializeAgentStreamEvent({ @@ -7175,9 +7182,9 @@ export class Session { selectedTimeline.endSeq !== null ? { epoch: selectedTimeline.timeline.epoch, seq: selectedTimeline.endSeq } : null; - const entries = this.supportsTimelineNotifications(source) - ? selectedTimeline.entries - : selectedTimeline.entries.filter((entry) => entry.item.type !== "notification"); + const entries = selectedTimeline.entries.filter((entry) => + this.supportsTimelineItem(entry.item, source), + ); this.emitForSource( { @@ -7376,9 +7383,7 @@ export class Session { limit: msg.limit ?? (direction === "after" ? 0 : 200), }, ); - const rows = this.supportsTimelineNotifications(source) - ? timeline.rows - : timeline.rows.filter((row) => row.item.type !== "notification"); + const rows = timeline.rows.filter((row) => this.supportsTimelineItem(row.item, source)); this.emitForSource( { type: "agent.provider_subagents.timeline.get.response", @@ -7714,9 +7719,43 @@ export class Session { "agent.session.outbound", ); } + if (msg.type === "workspace_setup_progress" || msg.type === "workspace_setup_status_response") { + if (this.clientCapabilitiesBySource.size > 0 && this.onMessageToSource) { + for (const [source] of this.clientCapabilitiesBySource) { + this.onMessageToSource(source, this.workspaceSetupMessageForClient(msg, source)); + } + return; + } + msg = this.workspaceSetupMessageForClient(msg); + } this.onMessage(msg); } + // COMPAT(workspaceSetupBlocked): added in v0.8.0, remove after 2027-03-07 once client floor >= v0.8.0. + private workspaceSetupMessageForClient( + message: Extract< + SessionOutboundMessage, + { type: "workspace_setup_progress" | "workspace_setup_status_response" } + >, + source?: object, + ): SessionOutboundMessage { + const supportsBlocked = source + ? this.supportsForSource(CLIENT_CAPS.workspaceSetupBlocked, source) + : this.supports(CLIENT_CAPS.workspaceSetupBlocked); + const snapshot = + message.type === "workspace_setup_progress" ? message.payload : message.payload.snapshot; + if (supportsBlocked || snapshot?.status !== "blocked") return message; + const legacySnapshot = { + ...snapshot, + status: "failed" as const, + error: + "Workspace setup is blocked pending approval of code from a fork pull request. Update Paseo to review and run setup.", + }; + return message.type === "workspace_setup_progress" + ? { ...message, payload: { ...message.payload, ...legacySnapshot } } + : { ...message, payload: { ...message.payload, snapshot: legacySnapshot } }; + } + private emitBinary(frame: Uint8Array): void { if (!this.onBinaryMessage) { return; diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index a51c4646f7..baf720b51b 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -185,6 +185,7 @@ function createSessionForWireCompatTest(options?: { clientCapabilities?: Record | null; directorySync?: DirectorySyncService; messages?: SessionOutboundMessage[]; + onMessageToSource?: SessionOptions["onMessageToSource"]; rows?: AgentTimelineRow[]; }): Session { const messages = options?.messages ?? []; @@ -212,6 +213,7 @@ function createSessionForWireCompatTest(options?: { permissions: OWNER_PERMISSIONS, clientCapabilities: options?.clientCapabilities ?? null, onMessage: (message) => messages.push(message), + onMessageToSource: options?.onMessageToSource, logger: pino({ level: "silent" }), downloadTokenStore: {} as SessionOptions["downloadTokenStore"], pushNotifications: {} as SessionOptions["pushNotifications"], @@ -543,3 +545,51 @@ describe("wire compatibility", () => { }); }); }); + +test("setup progress is adapted per socket without changing the canonical snapshot", async () => { + const legacy = {}; + const capable = {}; + const delivered = new Map(); + const session = createSessionForWireCompatTest({ + onMessageToSource: (source, message) => + delivered.set(source, [...(delivered.get(source) ?? []), message]), + }); + session.updateClientCapabilities({}, legacy); + session.updateClientCapabilities({ workspace_setup_blocked: true }, capable); + const message = { + type: "workspace_setup_progress" as const, + payload: { + workspaceId: "fork-workspace", + status: "blocked" as const, + error: null, + detail: { + type: "worktree_setup" as const, + worktreePath: "/workspace", + branchName: "fork", + log: "", + commands: [], + }, + blockedSource: { + kind: "change_request" as const, + forge: "github", + number: 42, + headRepository: "contributor/project", + }, + }, + }; + session.publish(message); + expect(delivered.get(capable)).toEqual([message]); + expect(delivered.get(legacy)).toEqual([ + { + ...message, + payload: { + ...message.payload, + status: "failed", + error: + "Workspace setup is blocked pending approval of code from a fork pull request. Update Paseo to review and run setup.", + }, + }, + ]); + expect(message.payload.status).toBe("blocked"); + await session.cleanup(); +}); diff --git a/public-docs/hub/security.md b/public-docs/hub/security.md index 7b3b9f94a6..f2e5d5d2f2 100644 --- a/public-docs/hub/security.md +++ b/public-docs/hub/security.md @@ -20,10 +20,11 @@ The host, provider credentials, filesystem, network, and resulting actions remai Connecting a daemon does not grant Hub permission to run agents. Grant `hub.execute` when you want Hub automations to use it. This permission allows Hub to create and continue agents, inspect their -state and timelines, and archive or restore workspaces on that daemon. It also applies to existing +state and timelines, configure their model, thinking, and provider permission mode, and archive or +restore workspaces on that daemon. Provider modes can include bypassing approval prompts. It also applies to existing agents and workspaces; it is not an isolation boundary around Hub-created work. -Daemon administration, terminal control, and permission management require separate permissions. +Daemon administration, terminal control, and daemon credential permission management require separate permissions. You can revoke `hub.execute` without granting Hub the ability to change its own permissions. ## Treat requests as untrusted