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>

packages/server/src/server/agent/agent-manager.ts

@@ -2517,12 +2517,30 @@ export class AgentManager {
   }
 
   async clearAgentAttention(agentId: string): Promise<void> {
-    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<void> {

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();
 });
 

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();
+    }
+  },
+);

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);
         }
 

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