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";