WIP: async action fixes and interest manager

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

foundry/packages/client/src/backend-client.ts

@@ -6,6 +6,9 @@ import type {
   FoundryAppSnapshot,
   FoundryBillingPlanId,
   CreateTaskInput,
+  AppEvent,
+  SessionEvent,
+  SandboxProcessesEvent,
   TaskRecord,
   TaskSummary,
   TaskWorkbenchChangeModelInput,
@@ -20,6 +23,12 @@ import type {
   TaskWorkbenchSnapshot,
   TaskWorkbenchTabInput,
   TaskWorkbenchUpdateDraftInput,
+  TaskEvent,
+  WorkbenchTaskDetail,
+  WorkbenchTaskSummary,
+  WorkbenchSessionDetail,
+  WorkspaceEvent,
+  WorkspaceSummarySnapshot,
   HistoryEvent,
   HistoryQueryInput,
   ProviderId,
@@ -34,7 +43,7 @@ import type {
 } from "@sandbox-agent/foundry-shared";
 import type { ProcessCreateRequest, ProcessInfo, ProcessLogFollowQuery, ProcessLogsResponse, ProcessSignalQuery } from "sandbox-agent";
 import { createMockBackendClient } from "./mock/backend-client.js";
-import { sandboxInstanceKey, workspaceKey } from "./keys.js";
+import { sandboxInstanceKey, taskKey, workspaceKey } from "./keys.js";
 
 export type TaskAction = "push" | "sync" | "merge" | "archive" | "kill";
 
@@ -60,7 +69,14 @@ export interface SandboxSessionEventRecord {
 
 export type SandboxProcessRecord = ProcessInfo;
 
+export interface ActorConn {
+  on(event: string, listener: (payload: any) => void): () => void;
+  onError(listener: (error: unknown) => void): () => void;
+  dispose(): Promise<void>;
+}
+
 interface WorkspaceHandle {
+  connect(): ActorConn;
   addRepo(input: AddRepoInput): Promise<RepoRecord>;
   listRepos(input: { workspaceId: string }): Promise<RepoRecord[]>;
   createTask(input: CreateTaskInput): Promise<TaskRecord>;
@@ -78,7 +94,10 @@ interface WorkspaceHandle {
   killTask(input: { workspaceId: string; taskId: string; reason?: string }): Promise<void>;
   useWorkspace(input: { workspaceId: string }): Promise<{ workspaceId: string }>;
   starSandboxAgentRepo(input: StarSandboxAgentRepoInput): Promise<StarSandboxAgentRepoResult>;
-  getWorkbench(input: { workspaceId: string }): Promise<TaskWorkbenchSnapshot>;
+  getWorkspaceSummary(input: { workspaceId: string }): Promise<WorkspaceSummarySnapshot>;
+  applyTaskSummaryUpdate(input: { taskSummary: WorkbenchTaskSummary }): Promise<void>;
+  removeTaskSummary(input: { taskId: string }): Promise<void>;
+  reconcileWorkbenchState(input: { workspaceId: string }): Promise<WorkspaceSummarySnapshot>;
   createWorkbenchTask(input: TaskWorkbenchCreateTaskInput): Promise<TaskWorkbenchCreateTaskResponse>;
   markWorkbenchUnread(input: TaskWorkbenchSelectInput): Promise<void>;
   renameWorkbenchTask(input: TaskWorkbenchRenameInput): Promise<void>;
@@ -95,7 +114,15 @@ interface WorkspaceHandle {
   revertWorkbenchFile(input: TaskWorkbenchDiffInput): Promise<void>;
 }
 
+interface TaskHandle {
+  getTaskSummary(): Promise<WorkbenchTaskSummary>;
+  getTaskDetail(): Promise<WorkbenchTaskDetail>;
+  getSessionDetail(input: { sessionId: string }): Promise<WorkbenchSessionDetail>;
+  connect(): ActorConn;
+}
+
 interface SandboxInstanceHandle {
+  connect(): ActorConn;
   createSession(input: {
     prompt: string;
     cwd?: string;
@@ -119,6 +146,10 @@ interface RivetClient {
   workspace: {
     getOrCreate(key?: string | string[], opts?: { createWithInput?: unknown }): WorkspaceHandle;
   };
+  task: {
+    get(key?: string | string[]): TaskHandle;
+    getOrCreate(key?: string | string[], opts?: { createWithInput?: unknown }): TaskHandle;
+  };
   sandboxInstance: {
     getOrCreate(key?: string | string[], opts?: { createWithInput?: unknown }): SandboxInstanceHandle;
   };
@@ -132,6 +163,9 @@ export interface BackendClientOptions {
 
 export interface BackendClient {
   getAppSnapshot(): Promise<FoundryAppSnapshot>;
+  connectWorkspace(workspaceId: string): Promise<ActorConn>;
+  connectTask(workspaceId: string, repoId: string, taskId: string): Promise<ActorConn>;
+  connectSandbox(workspaceId: string, providerId: ProviderId, sandboxId: string): Promise<ActorConn>;
   subscribeApp(listener: () => void): () => void;
   signInWithGithub(): Promise<void>;
   signOutApp(): Promise<FoundryAppSnapshot>;
@@ -222,6 +256,9 @@ export interface BackendClient {
     sandboxId: string,
   ): Promise<{ providerId: ProviderId; sandboxId: string; state: string; at: number }>;
   getSandboxAgentConnection(workspaceId: string, providerId: ProviderId, sandboxId: string): Promise<{ endpoint: string; token?: string }>;
+  getWorkspaceSummary(workspaceId: string): Promise<WorkspaceSummarySnapshot>;
+  getTaskDetail(workspaceId: string, repoId: string, taskId: string): Promise<WorkbenchTaskDetail>;
+  getSessionDetail(workspaceId: string, repoId: string, taskId: string, sessionId: string): Promise<WorkbenchSessionDetail>;
   getWorkbench(workspaceId: string): Promise<TaskWorkbenchSnapshot>;
   subscribeWorkbench(workspaceId: string, listener: () => void): () => void;
   createWorkbenchTask(workspaceId: string, input: TaskWorkbenchCreateTaskInput): Promise<TaskWorkbenchCreateTaskResponse>;
@@ -337,6 +374,8 @@ export function createBackendClient(options: BackendClientOptions): BackendClien
       createWithInput: workspaceId,
     });
 
+  const task = async (workspaceId: string, repoId: string, taskId: string): Promise<TaskHandle> => client.task.get(taskKey(workspaceId, repoId, taskId));
+
   const sandboxByKey = async (workspaceId: string, providerId: ProviderId, sandboxId: string): Promise<SandboxInstanceHandle> => {
     return (client as any).sandboxInstance.get(sandboxInstanceKey(workspaceId, providerId, sandboxId));
   };
@@ -400,6 +439,91 @@ export function createBackendClient(options: BackendClientOptions): BackendClien
     }
   };
 
+  const connectWorkspace = async (workspaceId: string): Promise<ActorConn> => {
+    return (await workspace(workspaceId)).connect() as ActorConn;
+  };
+
+  const connectTask = async (workspaceId: string, repoId: string, taskIdValue: string): Promise<ActorConn> => {
+    return (await task(workspaceId, repoId, taskIdValue)).connect() as ActorConn;
+  };
+
+  const connectSandbox = async (workspaceId: string, providerId: ProviderId, sandboxId: string): Promise<ActorConn> => {
+    try {
+      return (await sandboxByKey(workspaceId, providerId, sandboxId)).connect() as ActorConn;
+    } catch (error) {
+      if (!isActorNotFoundError(error)) {
+        throw error;
+      }
+      const fallback = await sandboxByActorIdFromTask(workspaceId, providerId, sandboxId);
+      if (!fallback) {
+        throw error;
+      }
+      return fallback.connect() as ActorConn;
+    }
+  };
+
+  const getWorkbenchCompat = async (workspaceId: string): Promise<TaskWorkbenchSnapshot> => {
+    const summary = await (await workspace(workspaceId)).getWorkspaceSummary({ workspaceId });
+    const tasks = await Promise.all(
+      summary.taskSummaries.map(async (taskSummary) => {
+        const detail = await (await task(workspaceId, taskSummary.repoId, taskSummary.id)).getTaskDetail();
+        const sessionDetails = await Promise.all(
+          detail.sessionsSummary.map(async (session) => {
+            const full = await (await task(workspaceId, detail.repoId, detail.id)).getSessionDetail({ sessionId: session.id });
+            return [session.id, full] as const;
+          }),
+        );
+        const sessionDetailsById = new Map(sessionDetails);
+        return {
+          id: detail.id,
+          repoId: detail.repoId,
+          title: detail.title,
+          status: detail.status,
+          repoName: detail.repoName,
+          updatedAtMs: detail.updatedAtMs,
+          branch: detail.branch,
+          pullRequest: detail.pullRequest,
+          tabs: detail.sessionsSummary.map((session) => {
+            const full = sessionDetailsById.get(session.id);
+            return {
+              id: session.id,
+              sessionId: session.sessionId,
+              sessionName: session.sessionName,
+              agent: session.agent,
+              model: session.model,
+              status: session.status,
+              thinkingSinceMs: session.thinkingSinceMs,
+              unread: session.unread,
+              created: session.created,
+              draft: full?.draft ?? { text: "", attachments: [], updatedAtMs: null },
+              transcript: full?.transcript ?? [],
+            };
+          }),
+          fileChanges: detail.fileChanges,
+          diffs: detail.diffs,
+          fileTree: detail.fileTree,
+          minutesUsed: detail.minutesUsed,
+        };
+      }),
+    );
+
+    const projects = summary.repos
+      .map((repo) => ({
+        id: repo.id,
+        label: repo.label,
+        updatedAtMs: tasks.filter((task) => task.repoId === repo.id).reduce((latest, task) => Math.max(latest, task.updatedAtMs), repo.latestActivityMs),
+        tasks: tasks.filter((task) => task.repoId === repo.id).sort((left, right) => right.updatedAtMs - left.updatedAtMs),
+      }))
+      .filter((repo) => repo.tasks.length > 0);
+
+    return {
+      workspaceId,
+      repos: summary.repos.map((repo) => ({ id: repo.id, label: repo.label })),
+      projects,
+      tasks: tasks.sort((left, right) => right.updatedAtMs - left.updatedAtMs),
+    };
+  };
+
   const subscribeWorkbench = (workspaceId: string, listener: () => void): (() => void) => {
     let entry = workbenchSubscriptions.get(workspaceId);
     if (!entry) {
@@ -544,6 +668,18 @@ export function createBackendClient(options: BackendClientOptions): BackendClien
       return await appRequest<FoundryAppSnapshot>("/app/snapshot");
     },
 
+    async connectWorkspace(workspaceId: string): Promise<ActorConn> {
+      return await connectWorkspace(workspaceId);
+    },
+
+    async connectTask(workspaceId: string, repoId: string, taskIdValue: string): Promise<ActorConn> {
+      return await connectTask(workspaceId, repoId, taskIdValue);
+    },
+
+    async connectSandbox(workspaceId: string, providerId: ProviderId, sandboxId: string): Promise<ActorConn> {
+      return await connectSandbox(workspaceId, providerId, sandboxId);
+    },
+
     subscribeApp(listener: () => void): () => void {
       return subscribeApp(listener);
     },
@@ -861,8 +997,20 @@ export function createBackendClient(options: BackendClientOptions): BackendClien
       return await withSandboxHandle(workspaceId, providerId, sandboxId, async (handle) => handle.sandboxAgentConnection());
     },
 
+    async getWorkspaceSummary(workspaceId: string): Promise<WorkspaceSummarySnapshot> {
+      return (await workspace(workspaceId)).getWorkspaceSummary({ workspaceId });
+    },
+
+    async getTaskDetail(workspaceId: string, repoId: string, taskIdValue: string): Promise<WorkbenchTaskDetail> {
+      return (await task(workspaceId, repoId, taskIdValue)).getTaskDetail();
+    },
+
+    async getSessionDetail(workspaceId: string, repoId: string, taskIdValue: string, sessionId: string): Promise<WorkbenchSessionDetail> {
+      return (await task(workspaceId, repoId, taskIdValue)).getSessionDetail({ sessionId });
+    },
+
     async getWorkbench(workspaceId: string): Promise<TaskWorkbenchSnapshot> {
-      return (await workspace(workspaceId)).getWorkbench({ workspaceId });
+      return await getWorkbenchCompat(workspaceId);
     },
 
     subscribeWorkbench(workspaceId: string, listener: () => void): () => void {

foundry/packages/client/src/index.ts

@@ -1,5 +1,10 @@
 export * from "./app-client.js";
 export * from "./backend-client.js";
+export * from "./interest/manager.js";
+export * from "./interest/mock-manager.js";
+export * from "./interest/remote-manager.js";
+export * from "./interest/topics.js";
+export * from "./interest/use-interest.js";
 export * from "./keys.js";
 export * from "./mock-app.js";
 export * from "./view-model.js";

foundry/packages/client/src/interest/manager.ts

@@ -0,0 +1,24 @@
+import type { TopicData, TopicKey, TopicParams } from "./topics.js";
+
+export type TopicStatus = "loading" | "connected" | "error";
+
+export interface TopicState<K extends TopicKey> {
+  data: TopicData<K> | undefined;
+  status: TopicStatus;
+  error: Error | null;
+}
+
+/**
+ * The InterestManager owns all realtime actor connections and cached state.
+ *
+ * Multiple subscribers to the same topic share one connection and one cache
+ * entry. After the last subscriber leaves, a short grace period keeps the
+ * connection warm so navigation does not thrash actor connections.
+ */
+export interface InterestManager {
+  subscribe<K extends TopicKey>(topicKey: K, params: TopicParams<K>, listener: () => void): () => void;
+  getSnapshot<K extends TopicKey>(topicKey: K, params: TopicParams<K>): TopicData<K> | undefined;
+  getStatus<K extends TopicKey>(topicKey: K, params: TopicParams<K>): TopicStatus;
+  getError<K extends TopicKey>(topicKey: K, params: TopicParams<K>): Error | null;
+  dispose(): void;
+}

foundry/packages/client/src/interest/mock-manager.ts

@@ -0,0 +1,12 @@
+import { createMockBackendClient } from "../mock/backend-client.js";
+import { RemoteInterestManager } from "./remote-manager.js";
+
+/**
+ * Mock implementation shares the same interest-manager harness as the remote
+ * path, but uses the in-memory mock backend that synthesizes actor events.
+ */
+export class MockInterestManager extends RemoteInterestManager {
+  constructor() {
+    super(createMockBackendClient());
+  }
+}

foundry/packages/client/src/interest/remote-manager.ts

@@ -0,0 +1,167 @@
+import type { BackendClient } from "../backend-client.js";
+import type { InterestManager, TopicStatus } from "./manager.js";
+import { topicDefinitions, type TopicData, type TopicDefinition, type TopicKey, type TopicParams } from "./topics.js";
+
+const GRACE_PERIOD_MS = 30_000;
+
+/**
+ * Remote implementation of InterestManager.
+ * Each cache entry owns one actor connection plus one materialized snapshot.
+ */
+export class RemoteInterestManager implements InterestManager {
+  private entries = new Map<string, TopicEntry<any, any, any>>();
+
+  constructor(private readonly backend: BackendClient) {}
+
+  subscribe<K extends TopicKey>(topicKey: K, params: TopicParams<K>, listener: () => void): () => void {
+    const definition = topicDefinitions[topicKey] as unknown as TopicDefinition<any, any, any>;
+    const cacheKey = definition.key(params as any);
+    let entry = this.entries.get(cacheKey);
+
+    if (!entry) {
+      entry = new TopicEntry(definition, this.backend, params as any);
+      this.entries.set(cacheKey, entry);
+    }
+
+    entry.cancelTeardown();
+    entry.addListener(listener);
+    entry.ensureStarted();
+
+    return () => {
+      const current = this.entries.get(cacheKey);
+      if (!current) {
+        return;
+      }
+      current.removeListener(listener);
+      if (current.listenerCount === 0) {
+        current.scheduleTeardown(GRACE_PERIOD_MS, () => {
+          this.entries.delete(cacheKey);
+        });
+      }
+    };
+  }
+
+  getSnapshot<K extends TopicKey>(topicKey: K, params: TopicParams<K>): TopicData<K> | undefined {
+    return this.entries.get((topicDefinitions[topicKey] as any).key(params))?.data as TopicData<K> | undefined;
+  }
+
+  getStatus<K extends TopicKey>(topicKey: K, params: TopicParams<K>): TopicStatus {
+    return this.entries.get((topicDefinitions[topicKey] as any).key(params))?.status ?? "loading";
+  }
+
+  getError<K extends TopicKey>(topicKey: K, params: TopicParams<K>): Error | null {
+    return this.entries.get((topicDefinitions[topicKey] as any).key(params))?.error ?? null;
+  }
+
+  dispose(): void {
+    for (const entry of this.entries.values()) {
+      entry.dispose();
+    }
+    this.entries.clear();
+  }
+}
+
+class TopicEntry<TData, TParams, TEvent> {
+  data: TData | undefined;
+  status: TopicStatus = "loading";
+  error: Error | null = null;
+  listenerCount = 0;
+
+  private readonly listeners = new Set<() => void>();
+  private conn: Awaited<ReturnType<TopicDefinition<TData, TParams, TEvent>["connect"]>> | null = null;
+  private unsubscribeEvent: (() => void) | null = null;
+  private unsubscribeError: (() => void) | null = null;
+  private teardownTimer: ReturnType<typeof setTimeout> | null = null;
+  private startPromise: Promise<void> | null = null;
+  private started = false;
+
+  constructor(
+    private readonly definition: TopicDefinition<TData, TParams, TEvent>,
+    private readonly backend: BackendClient,
+    private readonly params: TParams,
+  ) {}
+
+  addListener(listener: () => void): void {
+    this.listeners.add(listener);
+    this.listenerCount = this.listeners.size;
+  }
+
+  removeListener(listener: () => void): void {
+    this.listeners.delete(listener);
+    this.listenerCount = this.listeners.size;
+  }
+
+  ensureStarted(): void {
+    if (this.started || this.startPromise) {
+      return;
+    }
+    this.startPromise = this.start().finally(() => {
+      this.startPromise = null;
+    });
+  }
+
+  scheduleTeardown(ms: number, onTeardown: () => void): void {
+    this.teardownTimer = setTimeout(() => {
+      this.dispose();
+      onTeardown();
+    }, ms);
+  }
+
+  cancelTeardown(): void {
+    if (this.teardownTimer) {
+      clearTimeout(this.teardownTimer);
+      this.teardownTimer = null;
+    }
+  }
+
+  dispose(): void {
+    this.cancelTeardown();
+    this.unsubscribeEvent?.();
+    this.unsubscribeError?.();
+    if (this.conn) {
+      void this.conn.dispose();
+    }
+    this.conn = null;
+    this.data = undefined;
+    this.status = "loading";
+    this.error = null;
+    this.started = false;
+  }
+
+  private async start(): Promise<void> {
+    this.status = "loading";
+    this.error = null;
+    this.notify();
+
+    try {
+      this.conn = await this.definition.connect(this.backend, this.params);
+      this.unsubscribeEvent = this.conn.on(this.definition.event, (event: TEvent) => {
+        if (this.data === undefined) {
+          return;
+        }
+        this.data = this.definition.applyEvent(this.data, event);
+        this.notify();
+      });
+      this.unsubscribeError = this.conn.onError((error: unknown) => {
+        this.status = "error";
+        this.error = error instanceof Error ? error : new Error(String(error));
+        this.notify();
+      });
+      this.data = await this.definition.fetchInitial(this.backend, this.params);
+      this.status = "connected";
+      this.started = true;
+      this.notify();
+    } catch (error) {
+      this.status = "error";
+      this.error = error instanceof Error ? error : new Error(String(error));
+      this.started = false;
+      this.notify();
+    }
+  }
+
+  private notify(): void {
+    for (const listener of [...this.listeners]) {
+      listener();
+    }
+  }
+}

foundry/packages/client/src/interest/topics.ts

@@ -0,0 +1,131 @@
+import type {
+  AppEvent,
+  FoundryAppSnapshot,
+  ProviderId,
+  SandboxProcessesEvent,
+  SessionEvent,
+  TaskEvent,
+  WorkbenchSessionDetail,
+  WorkbenchTaskDetail,
+  WorkspaceEvent,
+  WorkspaceSummarySnapshot,
+} from "@sandbox-agent/foundry-shared";
+import type { ActorConn, BackendClient, SandboxProcessRecord } from "../backend-client.js";
+
+/**
+ * Topic definitions for the interest manager.
+ *
+ * Each topic describes one actor connection plus one materialized read model.
+ * Events always carry full replacement payloads for the changed entity so the
+ * client can replace cached state directly instead of reconstructing patches.
+ */
+export interface TopicDefinition<TData, TParams, TEvent> {
+  key: (params: TParams) => string;
+  event: string;
+  connect: (backend: BackendClient, params: TParams) => Promise<ActorConn>;
+  fetchInitial: (backend: BackendClient, params: TParams) => Promise<TData>;
+  applyEvent: (current: TData, event: TEvent) => TData;
+}
+
+export interface AppTopicParams {}
+export interface WorkspaceTopicParams {
+  workspaceId: string;
+}
+export interface TaskTopicParams {
+  workspaceId: string;
+  repoId: string;
+  taskId: string;
+}
+export interface SessionTopicParams {
+  workspaceId: string;
+  repoId: string;
+  taskId: string;
+  sessionId: string;
+}
+export interface SandboxProcessesTopicParams {
+  workspaceId: string;
+  providerId: ProviderId;
+  sandboxId: string;
+}
+
+function upsertById<T extends { id: string }>(items: T[], nextItem: T, sort: (left: T, right: T) => number): T[] {
+  const filtered = items.filter((item) => item.id !== nextItem.id);
+  return [...filtered, nextItem].sort(sort);
+}
+
+export const topicDefinitions = {
+  app: {
+    key: () => "app",
+    event: "appUpdated",
+    connect: (backend: BackendClient, _params: AppTopicParams) => backend.connectWorkspace("app"),
+    fetchInitial: (backend: BackendClient, _params: AppTopicParams) => backend.getAppSnapshot(),
+    applyEvent: (_current: FoundryAppSnapshot, event: AppEvent) => event.snapshot,
+  } satisfies TopicDefinition<FoundryAppSnapshot, AppTopicParams, AppEvent>,
+
+  workspace: {
+    key: (params: WorkspaceTopicParams) => `workspace:${params.workspaceId}`,
+    event: "workspaceUpdated",
+    connect: (backend: BackendClient, params: WorkspaceTopicParams) => backend.connectWorkspace(params.workspaceId),
+    fetchInitial: (backend: BackendClient, params: WorkspaceTopicParams) => backend.getWorkspaceSummary(params.workspaceId),
+    applyEvent: (current: WorkspaceSummarySnapshot, event: WorkspaceEvent) => {
+      switch (event.type) {
+        case "taskSummaryUpdated":
+          return {
+            ...current,
+            taskSummaries: upsertById(current.taskSummaries, event.taskSummary, (left, right) => right.updatedAtMs - left.updatedAtMs),
+          };
+        case "taskRemoved":
+          return {
+            ...current,
+            taskSummaries: current.taskSummaries.filter((task) => task.id !== event.taskId),
+          };
+        case "repoAdded":
+        case "repoUpdated":
+          return {
+            ...current,
+            repos: upsertById(current.repos, event.repo, (left, right) => right.latestActivityMs - left.latestActivityMs),
+          };
+        case "repoRemoved":
+          return {
+            ...current,
+            repos: current.repos.filter((repo) => repo.id !== event.repoId),
+          };
+      }
+    },
+  } satisfies TopicDefinition<WorkspaceSummarySnapshot, WorkspaceTopicParams, WorkspaceEvent>,
+
+  task: {
+    key: (params: TaskTopicParams) => `task:${params.workspaceId}:${params.taskId}`,
+    event: "taskUpdated",
+    connect: (backend: BackendClient, params: TaskTopicParams) => backend.connectTask(params.workspaceId, params.repoId, params.taskId),
+    fetchInitial: (backend: BackendClient, params: TaskTopicParams) => backend.getTaskDetail(params.workspaceId, params.repoId, params.taskId),
+    applyEvent: (_current: WorkbenchTaskDetail, event: TaskEvent) => event.detail,
+  } satisfies TopicDefinition<WorkbenchTaskDetail, TaskTopicParams, TaskEvent>,
+
+  session: {
+    key: (params: SessionTopicParams) => `session:${params.workspaceId}:${params.taskId}:${params.sessionId}`,
+    event: "sessionUpdated",
+    connect: (backend: BackendClient, params: SessionTopicParams) => backend.connectTask(params.workspaceId, params.repoId, params.taskId),
+    fetchInitial: (backend: BackendClient, params: SessionTopicParams) =>
+      backend.getSessionDetail(params.workspaceId, params.repoId, params.taskId, params.sessionId),
+    applyEvent: (current: WorkbenchSessionDetail, event: SessionEvent) => {
+      if (event.session.sessionId !== current.sessionId) {
+        return current;
+      }
+      return event.session;
+    },
+  } satisfies TopicDefinition<WorkbenchSessionDetail, SessionTopicParams, SessionEvent>,
+
+  sandboxProcesses: {
+    key: (params: SandboxProcessesTopicParams) => `sandbox:${params.workspaceId}:${params.providerId}:${params.sandboxId}`,
+    event: "processesUpdated",
+    connect: (backend: BackendClient, params: SandboxProcessesTopicParams) => backend.connectSandbox(params.workspaceId, params.providerId, params.sandboxId),
+    fetchInitial: async (backend: BackendClient, params: SandboxProcessesTopicParams) =>
+      (await backend.listSandboxProcesses(params.workspaceId, params.providerId, params.sandboxId)).processes,
+    applyEvent: (_current: SandboxProcessRecord[], event: SandboxProcessesEvent) => event.processes,
+  } satisfies TopicDefinition<SandboxProcessRecord[], SandboxProcessesTopicParams, SandboxProcessesEvent>,
+} as const;
+
+export type TopicKey = keyof typeof topicDefinitions;
+export type TopicParams<K extends TopicKey> = Parameters<(typeof topicDefinitions)[K]["fetchInitial"]>[1];
+export type TopicData<K extends TopicKey> = Awaited<ReturnType<(typeof topicDefinitions)[K]["fetchInitial"]>>;

foundry/packages/client/src/interest/use-interest.ts

@@ -0,0 +1,56 @@
+import { useMemo, useRef, useSyncExternalStore } from "react";
+import type { InterestManager, TopicState } from "./manager.js";
+import { topicDefinitions, type TopicKey, type TopicParams } from "./topics.js";
+
+/**
+ * React bridge for the interest manager.
+ *
+ * `null` params disable the subscription entirely, which is how screens express
+ * conditional interest in task/session/sandbox topics.
+ */
+export function useInterest<K extends TopicKey>(manager: InterestManager, topicKey: K, params: TopicParams<K> | null): TopicState<K> {
+  const paramsKey = params ? (topicDefinitions[topicKey] as any).key(params) : null;
+  const paramsRef = useRef<TopicParams<K> | null>(params);
+  paramsRef.current = params;
+
+  const subscribe = useMemo(() => {
+    return (listener: () => void) => {
+      const currentParams = paramsRef.current;
+      if (!currentParams) {
+        return () => {};
+      }
+      return manager.subscribe(topicKey, currentParams, listener);
+    };
+  }, [manager, topicKey, paramsKey]);
+
+  const getSnapshot = useMemo(() => {
+    let lastSnapshot: TopicState<K> | null = null;
+
+    return (): TopicState<K> => {
+      const currentParams = paramsRef.current;
+      const nextSnapshot: TopicState<K> = currentParams
+        ? {
+            data: manager.getSnapshot(topicKey, currentParams),
+            status: manager.getStatus(topicKey, currentParams),
+            error: manager.getError(topicKey, currentParams),
+          }
+        : {
+            data: undefined,
+            status: "loading",
+            error: null,
+          };
+
+      // `useSyncExternalStore` requires referentially-stable snapshots when the
+      // underlying store has not changed. Reuse the previous object whenever
+      // the topic data/status/error triplet is unchanged.
+      if (lastSnapshot && lastSnapshot.data === nextSnapshot.data && lastSnapshot.status === nextSnapshot.status && lastSnapshot.error === nextSnapshot.error) {
+        return lastSnapshot;
+      }
+
+      lastSnapshot = nextSnapshot;
+      return nextSnapshot;
+    };
+  }, [manager, topicKey, paramsKey]);
+
+  return useSyncExternalStore(subscribe, getSnapshot, getSnapshot);
+}

foundry/packages/client/src/mock/backend-client.ts

@@ -1,7 +1,10 @@
 import type {
   AddRepoInput,
+  AppEvent,
   CreateTaskInput,
   FoundryAppSnapshot,
+  SandboxProcessesEvent,
+  SessionEvent,
   TaskRecord,
   TaskSummary,
   TaskWorkbenchChangeModelInput,
@@ -16,6 +19,12 @@ import type {
   TaskWorkbenchSnapshot,
   TaskWorkbenchTabInput,
   TaskWorkbenchUpdateDraftInput,
+  TaskEvent,
+  WorkbenchSessionDetail,
+  WorkbenchTaskDetail,
+  WorkbenchTaskSummary,
+  WorkspaceEvent,
+  WorkspaceSummarySnapshot,
   HistoryEvent,
   HistoryQueryInput,
   ProviderId,
@@ -27,7 +36,7 @@ import type {
   SwitchResult,
 } from "@sandbox-agent/foundry-shared";
 import type { ProcessCreateRequest, ProcessLogFollowQuery, ProcessLogsResponse, ProcessSignalQuery } from "sandbox-agent";
-import type { BackendClient, SandboxProcessRecord, SandboxSessionEventRecord, SandboxSessionRecord } from "../backend-client.js";
+import type { ActorConn, BackendClient, SandboxProcessRecord, SandboxSessionEventRecord, SandboxSessionRecord } from "../backend-client.js";
 import { getSharedMockWorkbenchClient } from "./workbench-client.js";
 
 interface MockProcessRecord extends SandboxProcessRecord {
@@ -86,6 +95,7 @@ export function createMockBackendClient(defaultWorkspaceId = "default"): Backend
   const workbench = getSharedMockWorkbenchClient();
   const listenersBySandboxId = new Map<string, Set<() => void>>();
   const processesBySandboxId = new Map<string, MockProcessRecord[]>();
+  const connectionListeners = new Map<string, Set<(payload: any) => void>>();
   let nextPid = 4000;
   let nextProcessId = 1;
 
@@ -110,11 +120,174 @@ export function createMockBackendClient(defaultWorkspaceId = "default"): Backend
   const notifySandbox = (sandboxId: string): void => {
     const listeners = listenersBySandboxId.get(sandboxId);
     if (!listeners) {
+      emitSandboxProcessesUpdate(sandboxId);
       return;
     }
     for (const listener of [...listeners]) {
       listener();
     }
+    emitSandboxProcessesUpdate(sandboxId);
+  };
+
+  const connectionChannel = (scope: string, event: string): string => `${scope}:${event}`;
+
+  const emitConnectionEvent = (scope: string, event: string, payload: any): void => {
+    const listeners = connectionListeners.get(connectionChannel(scope, event));
+    if (!listeners) {
+      return;
+    }
+    for (const listener of [...listeners]) {
+      listener(payload);
+    }
+  };
+
+  const createConn = (scope: string): ActorConn => ({
+    on(event: string, listener: (payload: any) => void): () => void {
+      const channel = connectionChannel(scope, event);
+      let listeners = connectionListeners.get(channel);
+      if (!listeners) {
+        listeners = new Set();
+        connectionListeners.set(channel, listeners);
+      }
+      listeners.add(listener);
+      return () => {
+        const current = connectionListeners.get(channel);
+        if (!current) {
+          return;
+        }
+        current.delete(listener);
+        if (current.size === 0) {
+          connectionListeners.delete(channel);
+        }
+      };
+    },
+    onError(): () => void {
+      return () => {};
+    },
+    async dispose(): Promise<void> {},
+  });
+
+  const buildTaskSummary = (task: TaskWorkbenchSnapshot["tasks"][number]): WorkbenchTaskSummary => ({
+    id: task.id,
+    repoId: task.repoId,
+    title: task.title,
+    status: task.status,
+    repoName: task.repoName,
+    updatedAtMs: task.updatedAtMs,
+    branch: task.branch,
+    pullRequest: task.pullRequest,
+    sessionsSummary: task.tabs.map((tab) => ({
+      id: tab.id,
+      sessionId: tab.sessionId,
+      sessionName: tab.sessionName,
+      agent: tab.agent,
+      model: tab.model,
+      status: tab.status,
+      thinkingSinceMs: tab.thinkingSinceMs,
+      unread: tab.unread,
+      created: tab.created,
+    })),
+  });
+
+  const buildTaskDetail = (task: TaskWorkbenchSnapshot["tasks"][number]): WorkbenchTaskDetail => ({
+    ...buildTaskSummary(task),
+    task: task.title,
+    agentType: task.tabs[0]?.agent === "Codex" ? "codex" : "claude",
+    runtimeStatus: toTaskStatus(task.status === "archived" ? "archived" : "running", task.status === "archived"),
+    statusMessage: task.status === "archived" ? "archived" : "mock sandbox ready",
+    activeSessionId: task.tabs[0]?.sessionId ?? null,
+    diffStat: task.fileChanges.length > 0 ? `+${task.fileChanges.length}/-${task.fileChanges.length}` : "+0/-0",
+    prUrl: task.pullRequest ? `https://example.test/pr/${task.pullRequest.number}` : null,
+    reviewStatus: null,
+    fileChanges: task.fileChanges,
+    diffs: task.diffs,
+    fileTree: task.fileTree,
+    minutesUsed: task.minutesUsed,
+    sandboxes: [
+      {
+        providerId: "local",
+        sandboxId: task.id,
+        cwd: mockCwd(task.repoName, task.id),
+      },
+    ],
+    activeSandboxId: task.id,
+  });
+
+  const buildSessionDetail = (task: TaskWorkbenchSnapshot["tasks"][number], tabId: string): WorkbenchSessionDetail => {
+    const tab = task.tabs.find((candidate) => candidate.id === tabId);
+    if (!tab) {
+      throw new Error(`Unknown mock tab ${tabId} for task ${task.id}`);
+    }
+    return {
+      sessionId: tab.id,
+      tabId: tab.id,
+      sandboxSessionId: tab.sessionId,
+      sessionName: tab.sessionName,
+      agent: tab.agent,
+      model: tab.model,
+      status: tab.status,
+      thinkingSinceMs: tab.thinkingSinceMs,
+      unread: tab.unread,
+      created: tab.created,
+      draft: tab.draft,
+      transcript: tab.transcript,
+    };
+  };
+
+  const buildWorkspaceSummary = (): WorkspaceSummarySnapshot => {
+    const snapshot = workbench.getSnapshot();
+    const taskSummaries = snapshot.tasks.map(buildTaskSummary);
+    return {
+      workspaceId: defaultWorkspaceId,
+      repos: snapshot.repos.map((repo) => {
+        const repoTasks = taskSummaries.filter((task) => task.repoId === repo.id);
+        return {
+          id: repo.id,
+          label: repo.label,
+          taskCount: repoTasks.length,
+          latestActivityMs: repoTasks.reduce((latest, task) => Math.max(latest, task.updatedAtMs), 0),
+        };
+      }),
+      taskSummaries,
+    };
+  };
+
+  const workspaceScope = (workspaceId: string): string => `workspace:${workspaceId}`;
+  const taskScope = (workspaceId: string, repoId: string, taskId: string): string => `task:${workspaceId}:${repoId}:${taskId}`;
+  const sandboxScope = (workspaceId: string, providerId: string, sandboxId: string): string => `sandbox:${workspaceId}:${providerId}:${sandboxId}`;
+
+  const emitWorkspaceSnapshot = (): void => {
+    const summary = buildWorkspaceSummary();
+    const latestTask = [...summary.taskSummaries].sort((left, right) => right.updatedAtMs - left.updatedAtMs)[0] ?? null;
+    if (latestTask) {
+      emitConnectionEvent(workspaceScope(defaultWorkspaceId), "workspaceUpdated", {
+        type: "taskSummaryUpdated",
+        taskSummary: latestTask,
+      } satisfies WorkspaceEvent);
+    }
+  };
+
+  const emitTaskUpdate = (taskId: string): void => {
+    const task = requireTask(taskId);
+    emitConnectionEvent(taskScope(defaultWorkspaceId, task.repoId, task.id), "taskUpdated", {
+      type: "taskDetailUpdated",
+      detail: buildTaskDetail(task),
+    } satisfies TaskEvent);
+  };
+
+  const emitSessionUpdate = (taskId: string, tabId: string): void => {
+    const task = requireTask(taskId);
+    emitConnectionEvent(taskScope(defaultWorkspaceId, task.repoId, task.id), "sessionUpdated", {
+      type: "sessionUpdated",
+      session: buildSessionDetail(task, tabId),
+    } satisfies SessionEvent);
+  };
+
+  const emitSandboxProcessesUpdate = (sandboxId: string): void => {
+    emitConnectionEvent(sandboxScope(defaultWorkspaceId, "local", sandboxId), "processesUpdated", {
+      type: "processesUpdated",
+      processes: ensureProcessList(sandboxId).map((process) => cloneProcess(process)),
+    } satisfies SandboxProcessesEvent);
   };
 
   const buildTaskRecord = (taskId: string): TaskRecord => {
@@ -192,7 +365,19 @@ export function createMockBackendClient(defaultWorkspaceId = "default"): Backend
       return unsupportedAppSnapshot();
     },
 
-    subscribeApp(_listener: () => void): () => void {
+    async connectWorkspace(workspaceId: string): Promise<ActorConn> {
+      return createConn(workspaceScope(workspaceId));
+    },
+
+    async connectTask(workspaceId: string, repoId: string, taskId: string): Promise<ActorConn> {
+      return createConn(taskScope(workspaceId, repoId, taskId));
+    },
+
+    async connectSandbox(workspaceId: string, providerId: ProviderId, sandboxId: string): Promise<ActorConn> {
+      return createConn(sandboxScope(workspaceId, providerId, sandboxId));
+    },
+
+    subscribeApp(): () => void {
       return () => {};
     },
 
@@ -462,6 +647,18 @@ export function createMockBackendClient(defaultWorkspaceId = "default"): Backend
       return { endpoint: "mock://terminal-unavailable" };
     },
 
+    async getWorkspaceSummary(): Promise<WorkspaceSummarySnapshot> {
+      return buildWorkspaceSummary();
+    },
+
+    async getTaskDetail(_workspaceId: string, _repoId: string, taskId: string): Promise<WorkbenchTaskDetail> {
+      return buildTaskDetail(requireTask(taskId));
+    },
+
+    async getSessionDetail(_workspaceId: string, _repoId: string, taskId: string, sessionId: string): Promise<WorkbenchSessionDetail> {
+      return buildSessionDetail(requireTask(taskId), sessionId);
+    },
+
     async getWorkbench(): Promise<TaskWorkbenchSnapshot> {
       return workbench.getSnapshot();
     },
@@ -471,59 +668,99 @@ export function createMockBackendClient(defaultWorkspaceId = "default"): Backend
     },
 
     async createWorkbenchTask(_workspaceId: string, input: TaskWorkbenchCreateTaskInput): Promise<TaskWorkbenchCreateTaskResponse> {
-      return await workbench.createTask(input);
+      const created = await workbench.createTask(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(created.taskId);
+      if (created.tabId) {
+        emitSessionUpdate(created.taskId, created.tabId);
+      }
+      return created;
     },
 
     async markWorkbenchUnread(_workspaceId: string, input: TaskWorkbenchSelectInput): Promise<void> {
       await workbench.markTaskUnread(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
     },
 
     async renameWorkbenchTask(_workspaceId: string, input: TaskWorkbenchRenameInput): Promise<void> {
       await workbench.renameTask(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
     },
 
     async renameWorkbenchBranch(_workspaceId: string, input: TaskWorkbenchRenameInput): Promise<void> {
       await workbench.renameBranch(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
     },
 
     async createWorkbenchSession(_workspaceId: string, input: TaskWorkbenchSelectInput & { model?: string }): Promise<{ tabId: string }> {
-      return await workbench.addTab(input);
+      const created = await workbench.addTab(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
+      emitSessionUpdate(input.taskId, created.tabId);
+      return created;
     },
 
     async renameWorkbenchSession(_workspaceId: string, input: TaskWorkbenchRenameSessionInput): Promise<void> {
       await workbench.renameSession(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
+      emitSessionUpdate(input.taskId, input.tabId);
     },
 
     async setWorkbenchSessionUnread(_workspaceId: string, input: TaskWorkbenchSetSessionUnreadInput): Promise<void> {
       await workbench.setSessionUnread(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
+      emitSessionUpdate(input.taskId, input.tabId);
     },
 
     async updateWorkbenchDraft(_workspaceId: string, input: TaskWorkbenchUpdateDraftInput): Promise<void> {
       await workbench.updateDraft(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
+      emitSessionUpdate(input.taskId, input.tabId);
     },
 
     async changeWorkbenchModel(_workspaceId: string, input: TaskWorkbenchChangeModelInput): Promise<void> {
       await workbench.changeModel(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
+      emitSessionUpdate(input.taskId, input.tabId);
     },
 
     async sendWorkbenchMessage(_workspaceId: string, input: TaskWorkbenchSendMessageInput): Promise<void> {
       await workbench.sendMessage(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
+      emitSessionUpdate(input.taskId, input.tabId);
     },
 
     async stopWorkbenchSession(_workspaceId: string, input: TaskWorkbenchTabInput): Promise<void> {
       await workbench.stopAgent(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
+      emitSessionUpdate(input.taskId, input.tabId);
     },
 
     async closeWorkbenchSession(_workspaceId: string, input: TaskWorkbenchTabInput): Promise<void> {
       await workbench.closeTab(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
     },
 
     async publishWorkbenchPr(_workspaceId: string, input: TaskWorkbenchSelectInput): Promise<void> {
       await workbench.publishPr(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
     },
 
     async revertWorkbenchFile(_workspaceId: string, input: TaskWorkbenchDiffInput): Promise<void> {
       await workbench.revertFile(input);
+      emitWorkspaceSnapshot();
+      emitTaskUpdate(input.taskId);
     },
 
     async health(): Promise<{ ok: true }> {