fix(tell-agent): follow the agent directory so new sessions get a pill

tell-agent/client/agents.ts

@@ -1,11 +1,25 @@
 import type { PluginClientContext } from "@getpaseo/plugin/client";
 
 type PaseoApi = PluginClientContext["paseo"];
-export type AgentEntry = Awaited<ReturnType<PaseoApi["agents"]["list"]>>["entries"][number];
-type AgentSnapshot = AgentEntry["agent"];
+type PaseoAgentUpdate = Parameters<Parameters<PaseoApi["agents"]["subscribe"]>[0]>[0];
+type AgentSnapshot = Extract<PaseoAgentUpdate, { kind: "upsert" }>["agent"];
+export type AgentEntry = {
+  agent: AgentSnapshot;
+  project: {
+    projectName: string;
+    workspaceName?: string | null;
+    checkout: { isGit: boolean; currentBranch: string | null };
+  };
+};
+type PaseoAgentListResult = {
+  entries: AgentEntry[];
+  pageInfo: { hasMore: boolean; nextCursor: string | null };
+};
 
 export const AGENT_PAGE_LIMIT = 200;
 export const MAX_AGENT_PAGES = 10;
+const REOPEN_MIN_MS = 2_000;
+const REOPEN_MAX_MS = 60_000;
 
 export type AgentDirectoryPage = {
   entries: AgentEntry[];
@@ -27,6 +41,146 @@ export async function loadAgents(paseo: PaseoApi): Promise<AgentDirectoryPage> {
   return { entries, truncated: true };
 }
 
+/**
+ * Continues `loadAgents` from an observation's first page, to the same cap.
+ * Returns null once `current()` turns false, so a stale read is dropped.
+ */
+async function readRemainingPages(
+  paseo: PaseoApi,
+  first: PaseoAgentListResult,
+  signal: AbortSignal,
+  current: () => boolean,
+): Promise<AgentDirectoryPage | null> {
+  const entries = [...first.entries];
+  let cursor = first.pageInfo.hasMore ? (first.pageInfo.nextCursor ?? undefined) : undefined;
+  for (let page = 1; cursor; page += 1) {
+    if (page >= MAX_AGENT_PAGES) return { entries, truncated: true };
+    const result = await paseo.agents.list({
+      sort: [{ key: "updated_at", direction: "desc" }],
+      page: { limit: AGENT_PAGE_LIMIT, cursor },
+      signal,
+    });
+    if (!current()) return null;
+    entries.push(...result.entries);
+    cursor = result.pageInfo.hasMore ? (result.pageInfo.nextCursor ?? undefined) : undefined;
+  }
+  return { entries, truncated: false };
+}
+
+export type AgentDirectoryFollower = {
+  /**
+   * The directory as listed. `complete` means it holds every agent the host
+   * has, so anything missing is gone; otherwise apply it as upserts only.
+   */
+  snapshot(agents: AgentSnapshot[], complete: boolean): void;
+  upsert(agent: AgentSnapshot): void;
+  remove(agentId: string): void;
+};
+
+/**
+ * Keeps `follower` in step with the host's agents until the returned cleanup.
+ *
+ * `agents.subscribe()` only hears observations the same API instance opened,
+ * so this opens one with `list({ subscribe: {} })`. Its snapshot, the first
+ * page, arrives first and again after every reconnect; later pages are read
+ * plainly, and updates that land meanwhile win over what those pages say.
+ * Paseo releases an observation that fails, so it is reopened with backoff.
+ */
+export function followAgentDirectory(
+  paseo: PaseoApi,
+  follower: AgentDirectoryFollower,
+): () => void {
+  let stopped = false;
+  const apply = (update: PaseoAgentUpdate) => {
+    if (update.kind === "remove") follower.remove(update.agentId);
+    else follower.upsert(update.agent);
+  };
+
+  const lifetime = new AbortController();
+  let observation: { release(): Promise<void> } | null = null;
+  let reopenTimer: ReturnType<typeof setTimeout> | null = null;
+  let reopenDelay = REOPEN_MIN_MS;
+  /** Bumped per snapshot, so a reconnect abandons the paging of the one before. */
+  let generation = 0;
+  /** What updates said while the current snapshot's later pages were read. */
+  let landed: Map<string, AgentSnapshot | null> | null = null;
+
+  const applySnapshot = async (first: PaseoAgentListResult) => {
+    const current = ++generation;
+    const isCurrent = () => !stopped && generation === current;
+    const touched = new Map<string, AgentSnapshot | null>();
+    landed = touched;
+    let listed: AgentDirectoryPage | null;
+    try {
+      listed = await readRemainingPages(paseo, first, lifetime.signal, isCurrent);
+    } catch {
+      listed = { entries: [...first.entries], truncated: true };
+    }
+    if (!listed || !isCurrent()) return;
+    landed = null;
+    const agents = new Map(listed.entries.map((entry) => [entry.agent.id, entry.agent]));
+    for (const [agentId, agent] of touched) {
+      if (agent) agents.set(agentId, agent);
+      else agents.delete(agentId);
+    }
+    follower.snapshot([...agents.values()], !listed.truncated);
+  };
+
+  const reopen = () => {
+    observation = null;
+    generation += 1;
+    landed = null;
+    if (stopped || reopenTimer !== null) return;
+    reopenTimer = setTimeout(() => {
+      reopenTimer = null;
+      open();
+    }, reopenDelay);
+    reopenDelay = Math.min(reopenDelay * 2, REOPEN_MAX_MS);
+  };
+
+  const open = () => {
+    paseo.agents
+      .list({
+        sort: [{ key: "updated_at", direction: "desc" }],
+        page: { limit: AGENT_PAGE_LIMIT },
+        subscribe: {},
+        signal: lifetime.signal,
+      })
+      .then(({ subscription }) => {
+        if (stopped) {
+          void subscription.release().catch(() => undefined);
+          return;
+        }
+        observation = subscription;
+        subscription.subscribe({
+          snapshot: (first) => {
+            reopenDelay = REOPEN_MIN_MS;
+            void applySnapshot(first);
+          },
+          update: (message) => {
+            if (stopped || message.type !== "agent_update") return;
+            const update = message.payload;
+            if (update.kind === "remove") landed?.set(update.agentId, null);
+            else landed?.set(update.agent.id, update.agent);
+            apply(update);
+          },
+          error: reopen,
+        });
+      })
+      .catch(reopen);
+  };
+
+  open();
+  return () => {
+    stopped = true;
+    lifetime.abort();
+    if (reopenTimer !== null) clearTimeout(reopenTimer);
+    reopenTimer = null;
+    void observation?.release().catch(() => undefined);
+    observation = null;
+  };
+}
+
 export function title(entry: AgentEntry): string {
   const explicit = entry.agent.title?.trim();
   return explicit && explicit.length > 0 ? explicit : entry.agent.id.slice(0, 7);

tell-agent/client/message-agent.tsx

@@ -11,6 +11,7 @@ import { Pressable, ScrollView, Text, TextInput, View } from "react-native";
 import {
   AGENT_PAGE_LIMIT,
   type AgentEntry,
+  followAgentDirectory,
   loadAgents,
   MAX_AGENT_PAGES,
   placement,
@@ -328,20 +329,18 @@ export function contributeAgentMessaging(client: PluginClientContext) {
     pills.set(agent.id, { workspaceId: next.workspaceId, remove: pill.remove });
   }
 
-  const unsubscribe = client.paseo.agents.subscribe((update) => {
-    if (update.kind === "remove") {
-      removePill(update.agentId);
-      return;
-    }
-    syncAgent(update.agent);
-  });
-
-  void loadAgents(client.paseo)
-    .then(({ entries }) => {
+  const unsubscribe = followAgentDirectory(client.paseo, {
+    snapshot(agents, complete) {
       if (stopped) return;
-      for (const { agent } of entries) syncAgent(agent);
-    })
-    .catch(() => undefined);
+      if (complete) {
+        const listed = new Set(agents.map((agent) => agent.id));
+        for (const agentId of [...pills.keys()]) if (!listed.has(agentId)) removePill(agentId);
+      }
+      for (const agent of agents) syncAgent(agent);
+    },
+    upsert: syncAgent,
+    remove: removePill,
+  });
 
   const removeTellCommand = client.addSlashCommand({
     name: "tell",

tell-agent/paseo-plugin.json

@@ -1,4 +1,4 @@
 {
   "id": "tell-agent",
-  "requirements": { "paseo": ">=0.8.0 <0.10.0" }
+  "requirements": { "paseo": ">=0.9.0 <0.10.0" }
 }