From ac08077f7ae69dc61e45a694b927e12f0a7f64ae Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Sun, 8 Nov 2026 06:54:33 +0000 Subject: [PATCH] Keep workspace attention in sync after agent runtimes close (#9087) * fix(server): publish attention clears for unloaded agents Stored attention clears bypassed agent state events and subscription ownership, leaving connected directory observers stale. Use the existing manager mutation and publish path for unloaded agents, and cover independent filtered subscribers through real daemon requests. * test(server): run attention regressions in subscription CI Keep full clear-response coverage alongside independent subscriber checks. Allow asynchronous native deletion classification beyond the default one-second polling deadline. --------- Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../server/src/server/agent/agent-manager.ts | 28 ++- .../src/server/file-observer/index.test.ts | 6 +- .../server/owned-subscriptions.e2e.test.ts | 112 +++++++++++ packages/server/src/server/session.ts | 19 +- .../src/server/session.workspaces.test.ts | 179 ------------------ 5 files changed, 141 insertions(+), 203 deletions(-) diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index b13de1930a..8b78f3706a 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2517,12 +2517,30 @@ export class AgentManager { } async clearAgentAttention(agentId: string): Promise { - const agent = this.requireAgent(agentId); - if (agent.attention.requiresAttention) { - agent.attention = { requiresAttention: false }; - await this.persistSnapshot(agent); - this.emitState(agent, { persist: false }); + const agent = this.agents.get(agentId); + if (agent) { + if (agent.attention.requiresAttention) { + agent.attention = { requiresAttention: false }; + await this.persistSnapshot(agent); + this.emitState(agent, { persist: false }); + } + return; } + + const registry = this.requireRegistry(); + const record = await registry.get(agentId); + if (!record || record.internal || record.archivedAt || record.requiresAttention !== true) { + return; + } + const nextRecord: StoredAgentRecord = { + ...record, + updatedAt: this.nextStoredUpdatedAt(record), + requiresAttention: false, + attentionReason: null, + attentionTimestamp: null, + }; + await registry.upsert(nextRecord); + this.dispatchStoredAgentState(nextRecord); } async markAgentUnread(agentId: string): Promise { diff --git a/packages/server/src/server/file-observer/index.test.ts b/packages/server/src/server/file-observer/index.test.ts index 95fe416c54..ccb451632a 100644 --- a/packages/server/src/server/file-observer/index.test.ts +++ b/packages/server/src/server/file-observer/index.test.ts @@ -165,7 +165,11 @@ test("classifies a removed file as deleted", async () => { }); await rm(target); - await expect.poll(() => events.find((event) => event.path === target)?.type).toBe("delete"); + // Native recursive delivery can be coalesced before the backend's deletion audit. + // Assert eventual classification without imposing Vitest's one-second default. + await expect + .poll(() => events.find((event) => event.path === target)?.type, { timeout: 10_000 }) + .toBe("delete"); await subscription.unsubscribe(); }); diff --git a/packages/server/src/server/owned-subscriptions.e2e.test.ts b/packages/server/src/server/owned-subscriptions.e2e.test.ts index fc73e44db4..175b245dfc 100644 --- a/packages/server/src/server/owned-subscriptions.e2e.test.ts +++ b/packages/server/src/server/owned-subscriptions.e2e.test.ts @@ -1873,3 +1873,115 @@ test("public project subscriptions request updates and release their producer de await daemon.close(); } }); + +test.each([ + { closed: false, batch: false }, + { closed: true, batch: false }, + { closed: false, batch: true }, + { closed: true, batch: true }, +])( + "clearing workspace attention reaches independent subscribers (closed=$closed, batch=$batch)", + async ({ closed, batch }) => { + const daemon = await createTestPaseoDaemon({ mcpEnabled: false }); + const clients = Array.from( + { length: 4 }, + () => new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }), + ); + const [actor, observer, unreadObserver, quiet] = clients; + const updates: SessionOutboundMessage[] = []; + const unreadUpdates: SessionOutboundMessage[] = []; + const quietUpdates: SessionOutboundMessage[] = []; + const responses: SessionOutboundMessage[] = []; + try { + await Promise.all(clients.map((client) => client.connect())); + const agentIds: string[] = []; + for (let index = 0; index < (batch ? 2 : 1); index += 1) { + const cwd = path.join(daemon.staticDir, `workspace-${index}`); + await mkdir(cwd); + const agent = await actor.createAgent({ provider: "codex", cwd }); + await actor.sendMessage(agent.id, "Reply with OK."); + await actor.waitForFinish(agent.id, 5000); + await daemon.daemon.agentManager.clearAgentAttention(agent.id); + if (closed) await daemon.daemon.agentManager.closeAgent(agent.id); + agentIds.push(agent.id); + } + const snapshot = await observer.fetchAgents({ + filter: { includeArchived: true }, + subscribe: {}, + }); + const workspaceIds = agentIds.map( + (id) => snapshot.entries.find((entry) => entry.agent.id === id)!.agent.workspaceId!, + ); + expect(new Set(workspaceIds).size).toBe(agentIds.length); + await unreadObserver.fetchAgents({ filter: { requiresAttention: true }, subscribe: {} }); + actor.subscribeRawMessages((message) => responses.push(message)); + observer.subscribeRawMessages((message) => updates.push(message)); + unreadObserver.subscribeRawMessages((message) => unreadUpdates.push(message)); + quiet.subscribeRawMessages((message) => quietUpdates.push(message)); + const attention = (id: string) => + updates + .flatMap((message) => + message.type === "agent_update" && + message.payload.kind === "upsert" && + message.payload.agent.id === id + ? [message.payload.agent.requiresAttention] + : [], + ) + .at(-1); + for (const [index, id] of agentIds.entries()) { + await actor.markWorkspaceUnread(workspaceIds[index]); + await expect.poll(() => attention(id)).toBe(true); + await expect + .poll(() => + unreadUpdates.some( + (message) => + message.type === "agent_update" && + message.payload.kind === "upsert" && + message.payload.agent.id === id, + ), + ) + .toBe(true); + } + updates.length = unreadUpdates.length = quietUpdates.length = 0; + await actor.clearWorkspaceAttention(batch ? workspaceIds : workspaceIds[0]); + const response = responses.findLast( + (message) => message.type === "workspace.clear_attention.response", + ); + expect(response?.payload).toEqual({ + requestId: expect.any(String), + workspaceId: batch ? workspaceIds : workspaceIds[0], + clearedAgentIds: agentIds, + results: workspaceIds.map((workspaceId, index) => ({ + workspaceId, + clearedAgentIds: [agentIds[index]], + success: true, + error: null, + })), + success: true, + error: null, + }); + const fresh = await actor.fetchAgents(); + for (const id of agentIds) { + expect(fresh.entries.find((entry) => entry.agent.id === id)?.agent.requiresAttention).toBe( + false, + ); + expect(Boolean(daemon.daemon.agentManager.getAgent(id))).toBe(!closed); + await expect.poll(() => attention(id)).toBe(false); + await expect + .poll(() => + unreadUpdates.some( + (message) => + message.type === "agent_update" && + message.payload.kind === "remove" && + message.payload.agentId === id, + ), + ) + .toBe(true); + } + expect(quietUpdates.filter((message) => message.type === "agent_update")).toEqual([]); + } finally { + await Promise.all(clients.map((client) => client.close())); + await daemon.close(); + } + }, +); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 0de1d64687..a09fc070be 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -8076,24 +8076,7 @@ export class Session { ) { continue; } - const nextRecord: StoredAgentRecord = { - ...record, - updatedAt: new Date().toISOString(), - requiresAttention: false, - attentionReason: null, - attentionTimestamp: null, - }; - await this.agentStorage.upsert(nextRecord); - const agent = this.buildStoredAgentPayload(nextRecord); - const project = await this.buildProjectPlacementForWorkspace(workspace); - this.emit({ - type: "agent_update", - payload: { - kind: "upsert", - agent, - project, - }, - }); + await this.agentManager.clearAgentAttention(agentId); clearedAgentIds.push(agentId); } diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 928131ab3a..7511938390 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -1847,81 +1847,6 @@ test("archive emits an authoritative agent_update upsert for subscribed clients" }); }); -test("workspace clear attention clears stored-only agents and responds", async () => { - const emitted: SessionOutboundMessage[] = []; - const workspace = createPersistedWorkspaceRecord({ - workspaceId: REPO_CWD, - projectId: REPO_CWD, - cwd: REPO_CWD, - kind: "directory", - displayName: "repo", - createdAt: "2026-03-30T15:00:00.000Z", - updatedAt: "2026-03-30T15:00:00.000Z", - }); - const project = createPersistedProjectRecord({ - projectId: REPO_CWD, - rootPath: REPO_CWD, - kind: "non_git", - displayName: "repo", - createdAt: "2026-03-30T15:00:00.000Z", - updatedAt: "2026-03-30T15:00:00.000Z", - }); - let storedRecord = makeStoredAgent({ - id: "stored-agent-1", - cwd: REPO_CWD, - updatedAt: "2026-03-30T15:00:00.000Z", - requiresAttention: true, - attentionReason: "finished", - }); - const session = createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) }); - - session.workspaceRegistry.listAll = session.workspaceRegistry.listActive = async () => [ - workspace, - ]; - session.workspaceRegistry.get = async (id: string) => - id === workspace.workspaceId ? workspace : null; - session.projectRegistry.list = async () => [project]; - session.projectRegistry.get = async (id: string) => (id === project.projectId ? project : null); - session.agentStorage.get = async (agentId: string) => - agentId === storedRecord.id ? storedRecord : null; - session.agentStorage.upsert = async (record: unknown) => { - storedRecord = record as StoredAgentRecord; - }; - session.listAgentPayloads = async () => [ - makeAgent({ - id: storedRecord.id, - cwd: storedRecord.cwd, - workspaceId: workspace.workspaceId, - status: "closed", - updatedAt: storedRecord.updatedAt, - requiresAttention: true, - attentionReason: "finished", - }), - ]; - - await session.handleMessage({ - type: "workspace.clear_attention.request", - workspaceId: workspace.workspaceId, - requestId: "req-1", - }); - - expect(storedRecord.requiresAttention).toBe(false); - expect(storedRecord.attentionReason).toBeNull(); - expect(storedRecord.attentionTimestamp).toBeNull(); - expect(findByType(emitted, "workspace.clear_attention.response").payload).toMatchObject({ - requestId: "req-1", - workspaceId: workspace.workspaceId, - clearedAgentIds: [storedRecord.id], - success: true, - error: null, - }); - const agentUpdate = findByType(emitted, "agent_update"); - expect(agentUpdate.payload.kind).toBe("upsert"); - if (agentUpdate.payload.kind === "upsert") { - expect(agentUpdate.payload.agent.requiresAttention).toBe(false); - } -}); - test("workspace clear attention responds with an error instead of timing out", async () => { const emitted: SessionOutboundMessage[] = []; const session = createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) }); @@ -2200,110 +2125,6 @@ test("workspace mark unread rejects workspaces without a read root agent", async }); }); -test("workspace clear attention can clear multiple workspaces in one request", async () => { - const emitted: SessionOutboundMessage[] = []; - const workspaces = [ - createPersistedWorkspaceRecord({ - workspaceId: "ws-repo-a", - projectId: "/tmp/repo-a", - cwd: "/tmp/repo-a", - kind: "directory", - displayName: "repo-a", - createdAt: "2026-03-30T15:00:00.000Z", - updatedAt: "2026-03-30T15:00:00.000Z", - }), - createPersistedWorkspaceRecord({ - workspaceId: "ws-repo-b", - projectId: "/tmp/repo-b", - cwd: "/tmp/repo-b", - kind: "directory", - displayName: "repo-b", - createdAt: "2026-03-30T15:00:00.000Z", - updatedAt: "2026-03-30T15:00:00.000Z", - }), - ]; - const projects = workspaces.map((workspace) => - createPersistedProjectRecord({ - projectId: workspace.projectId, - rootPath: workspace.cwd, - kind: "non_git", - displayName: workspace.displayName, - createdAt: "2026-03-30T15:00:00.000Z", - updatedAt: "2026-03-30T15:00:00.000Z", - }), - ); - const storedRecords = new Map( - workspaces.map((workspace, index) => [ - `stored-agent-${index + 1}`, - makeStoredAgent({ - id: `stored-agent-${index + 1}`, - cwd: workspace.cwd, - updatedAt: "2026-03-30T15:00:00.000Z", - requiresAttention: true, - attentionReason: "finished", - }), - ]), - ); - const session = createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) }); - - session.workspaceRegistry.listAll = session.workspaceRegistry.listActive = async () => workspaces; - session.workspaceRegistry.get = async (id: string) => - workspaces.find((workspace) => workspace.workspaceId === id) ?? null; - session.projectRegistry.list = async () => projects; - session.projectRegistry.get = async (id: string) => - projects.find((project) => project.projectId === id) ?? null; - session.agentStorage.get = async (agentId: string) => storedRecords.get(agentId) ?? null; - session.agentStorage.upsert = async (record: unknown) => { - const storedRecord = record as StoredAgentRecord; - storedRecords.set(storedRecord.id, storedRecord); - }; - session.listAgentPayloads = async () => - Array.from(storedRecords.values()).map((record) => { - const owner = workspaces.find((workspace) => workspace.cwd === record.cwd); - return makeAgent({ - id: record.id, - cwd: record.cwd, - ...(owner ? { workspaceId: owner.workspaceId } : {}), - status: "closed", - updatedAt: record.updatedAt, - requiresAttention: record.requiresAttention, - attentionReason: record.attentionReason, - }); - }); - - await session.handleMessage({ - type: "workspace.clear_attention.request", - workspaceId: workspaces.map((workspace) => workspace.workspaceId), - requestId: "req-1", - }); - - expect(Array.from(storedRecords.values()).map((record) => record.requiresAttention)).toEqual([ - false, - false, - ]); - expect(findByType(emitted, "workspace.clear_attention.response").payload).toMatchObject({ - requestId: "req-1", - workspaceId: workspaces.map((workspace) => workspace.workspaceId), - clearedAgentIds: ["stored-agent-1", "stored-agent-2"], - results: [ - { - workspaceId: workspaces[0].workspaceId, - clearedAgentIds: ["stored-agent-1"], - success: true, - error: null, - }, - { - workspaceId: workspaces[1].workspaceId, - clearedAgentIds: ["stored-agent-2"], - success: true, - error: null, - }, - ], - success: true, - error: null, - }); -}); - test("close_items_request archives agents and kills terminals in one batch", async () => { const emitted: SessionOutboundMessage[] = []; const archivedAt = "2026-04-01T00:00:00.000Z";