From 79f72da153e4121eee70b90fc9b6230554952633 Mon Sep 17 00:00:00 2001 From: qinkangdeid Date: Wed, 23 Sep 2026 16:54:54 +0800 Subject: [PATCH 1/2] fix(tell-agent): follow the agent directory so new sessions get a pill Since Paseo 0.9 (getpaseo/paseo#4596), agents.subscribe(handler) only registers a local listener on that API instance's observations and does not request data from the daemon; the plugin also seeds once via loadAgents() at startup, so agents created afterwards only get the "Tell agent" pill after the plugin is reloaded. - client/agents.ts: add followAgentDirectory. It feature-detects via paseo.observeEvents (added in 0.9.0-beta.1 together with the new subscription model). On 0.9 clients it opens an observation with list({ sort: updated_at desc, page: 200, subscribe: {}, signal }); the snapshot (first connect and every reconnect) only covers the first page, so remaining pages are read with plain list() calls under the existing 200x10 cap, updates arriving meanwhile override stale page data, and a fresh snapshot after a reconnect invalidates in-flight continuation reads. Pills for agents missing from the fully-read snapshot are removed; on truncation or continuation failure only upsert is performed. When the observation is released by the client it is reopened with 2s to 60s backoff; on stop the signal is aborted and the subscription released. On 0.8 clients the subscribe option is not sent (the legacy connection keeps one subscription slot per kind and later sends overwrite earlier ones, which would clobber the host app's own agent subscription), so the original agents.subscribe() + loadAgents() path is kept. - message-agent.tsx: pills are now driven by followAgentDirectory, reusing syncAgent (no re-registration when the workspace is unchanged); the popover target list still loads on demand via useQuery and is unchanged. - tests/agents.test.ts: cover the 0.9 single-page, multi-page, update override, cap truncation, reconnect invalidation, and failure reopen paths, plus the 0.8 legacy path. --- tell-agent/client/agents.ts | 170 +++++++++++++++++- tell-agent/client/message-agent.tsx | 25 ++- tell-agent/tests/agents.test.ts | 262 ++++++++++++++++++++++++++++ 3 files changed, 443 insertions(+), 14 deletions(-) create mode 100644 tell-agent/tests/agents.test.ts diff --git a/tell-agent/client/agents.ts b/tell-agent/client/agents.ts index 0017c485..b44f453d 100644 --- a/tell-agent/client/agents.ts +++ b/tell-agent/client/agents.ts @@ -1,4 +1,9 @@ -import type { PluginClientContext } from "@getpaseo/plugin/client"; +import type { + PaseoAgent, + PaseoAgentListResult, + PaseoAgentUpdate, + PaseoApi, +} from "@getpaseo/client"; type PaseoApi = PluginClientContext["paseo"]; export type AgentEntry = Awaited>["entries"][number]; @@ -6,6 +11,8 @@ type AgentSnapshot = AgentEntry["agent"]; 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 +34,167 @@ export async function loadAgents(paseo: PaseoApi): Promise { 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, + current: () => boolean, +): Promise { + 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 }, + }); + 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; +}; + +/** + * `observeEvents` shipped together with owned observations (0.9.0-beta.1). A + * 0.8 client must not send `subscribe`: the daemon keeps one agents slot per + * legacy connection, last query wins, so it would replace the app's own. + */ +function ownsObservations(paseo: PaseoApi): boolean { + return typeof (paseo as { observeEvents?: unknown }).observeEvents === "function"; +} + +/** + * Keeps `follower` in step with the host's agents until the returned cleanup. + * + * On 0.9, `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. + * + * On 0.8 it listens and reads the directory once, as before. + */ +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); + }; + + if (!ownsObservations(paseo)) { + const unsubscribe = paseo.agents.subscribe((update) => { + if (!stopped) apply(update); + }); + void loadAgents(paseo) + .then(({ entries }) => { + if (stopped) return; + const agents = entries.map((entry) => entry.agent); + follower.snapshot(agents, false); + }) + .catch(() => undefined); + return () => { + stopped = true; + unsubscribe(); + }; + } + + const lifetime = new AbortController(); + let observation: { release(): Promise } | null = null; + let reopenTimer: ReturnType | 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 | null = null; + + const applySnapshot = async (first: PaseoAgentListResult) => { + const current = ++generation; + const isCurrent = () => !stopped && generation === current; + const touched = new Map(); + landed = touched; + let listed: AgentDirectoryPage | null; + try { + listed = await readRemainingPages(paseo, first, 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; + 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) 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); diff --git a/tell-agent/client/message-agent.tsx b/tell-agent/client/message-agent.tsx index 8a2ec4e8..0e4ef681 100644 --- a/tell-agent/client/message-agent.tsx +++ b/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", diff --git a/tell-agent/tests/agents.test.ts b/tell-agent/tests/agents.test.ts new file mode 100644 index 00000000..b4a80258 --- /dev/null +++ b/tell-agent/tests/agents.test.ts @@ -0,0 +1,262 @@ +import type { PaseoApi } from "@getpaseo/client"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { + type AgentDirectoryFollower, + followAgentDirectory, + MAX_AGENT_PAGES, +} from "../client/agents"; + +type Agent = { id: string; workspaceId: string }; +type ListRequest = { + subscribe?: object; + signal?: AbortSignal; + sort?: unknown; + page?: { limit: number; cursor?: string }; +}; +type Observer = { + snapshot(list: unknown): void; + update(message: { type: string; payload: unknown }): void; + error?(error: unknown): void; +}; + +function agent(id: string): Agent { + return { id, workspaceId: `ws-${id}` }; +} + +function page(agents: Agent[], next?: string) { + return { + requestId: "req", + subscriptionId: "sub", + entries: agents.map((item) => ({ agent: item, project: {} })), + pageInfo: { hasMore: next !== undefined, nextCursor: next ?? null }, + }; +} + +/** + * A host whose directory is `pages()`, one array per page. `observing` makes it + * a 0.9 client (observation from `list({ subscribe })`); otherwise a 0.8 one. + * `gate` holds plain page reads until released, to land updates mid-paging. + */ +function fakeHost(options: { observing: boolean; pages: () => Agent[][] }) { + const requests: ListRequest[] = []; + const observers: Observer[] = []; + let listener: ((update: unknown) => void) | null = null; + let released = 0; + let gate: Promise | null = null; + const pageAt = (cursor?: string) => { + const pages = options.pages(); + const index = cursor ? Number(cursor) : 0; + return page(pages[index] ?? [], index + 1 < pages.length ? String(index + 1) : undefined); + }; + const agents = { + subscribe(handler: (update: unknown) => void) { + listener = handler; + return () => { + listener = null; + }; + }, + async list(request: ListRequest = {}) { + requests.push(request); + if (!request.subscribe) { + if (gate) await gate; + return pageAt(request.page?.cursor); + } + return { + ...pageAt(), + subscription: { + subscribe(next: Observer) { + observers.push(next); + next.snapshot(pageAt()); + return () => undefined; + }, + async release() { + released += 1; + }, + }, + }; + }, + }; + const observer = () => observers.at(-1); + return { + paseo: (options.observing ? { observeEvents() {}, agents } : { agents }) as unknown as PaseoApi, + requests, + get released() { + return released; + }, + get listening() { + return listener !== null; + }, + hold() { + let open = () => {}; + gate = new Promise((resolve) => { + open = resolve; + }); + return () => { + gate = null; + open(); + }; + }, + update(payload: unknown) { + observer()?.update({ type: "agent_update", payload }); + listener?.(payload); + }, + reconnect() { + observer()?.snapshot(pageAt()); + }, + fail(error: unknown) { + observer()?.error?.(error); + }, + }; +} + +function recorder() { + const snapshots: { ids: string[]; complete: boolean }[] = []; + const upserts: string[] = []; + const removes: string[] = []; + const follower: AgentDirectoryFollower = { + snapshot(agents, complete) { + snapshots.push({ ids: agents.map((item) => item.id).sort(), complete }); + }, + upsert(item) { + upserts.push(item.id); + }, + remove(agentId) { + removes.push(agentId); + }, + }; + return { follower, snapshots, upserts, removes }; +} + +const flush = () => vi.advanceTimersByTimeAsync(0); + +describe("followAgentDirectory on a 0.9 client", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + test("opens one observation and applies its snapshot and updates", async () => { + const host = fakeHost({ observing: true, pages: () => [[agent("a1"), agent("a2")]] }); + const seen = recorder(); + const stop = followAgentDirectory(host.paseo, seen.follower); + await flush(); + + expect(host.requests).toHaveLength(1); + expect(host.requests[0]).toMatchObject({ + subscribe: {}, + sort: [{ key: "updated_at", direction: "desc" }], + page: { limit: 200 }, + }); + expect(host.requests[0]?.signal).toBeInstanceOf(AbortSignal); + expect(host.listening).toBe(false); + expect(seen.snapshots).toEqual([{ ids: ["a1", "a2"], complete: true }]); + + host.update({ kind: "upsert", agent: agent("a3") }); + host.update({ kind: "remove", agentId: "a1" }); + expect(seen.upserts).toEqual(["a3"]); + expect(seen.removes).toEqual(["a1"]); + + stop(); + await flush(); + expect(host.requests[0]?.signal?.aborted).toBe(true); + expect(host.released).toBe(1); + host.update({ kind: "upsert", agent: agent("a4") }); + expect(seen.upserts).toEqual(["a3"]); + }); + + test("reads later pages plainly and lets updates that land meanwhile win", async () => { + const host = fakeHost({ + observing: true, + pages: () => [[agent("a1")], [agent("a2"), agent("a3")]], + }); + const release = host.hold(); + const seen = recorder(); + const stop = followAgentDirectory(host.paseo, seen.follower); + await flush(); + expect(seen.snapshots).toEqual([]); + + // The second page was computed before these; it must not resurrect a3 or drop a9. + host.update({ kind: "remove", agentId: "a3" }); + host.update({ kind: "upsert", agent: agent("a9") }); + release(); + await flush(); + + expect(host.requests[1]).toMatchObject({ page: { limit: 200, cursor: "1" } }); + expect(host.requests[1]?.subscribe).toBeUndefined(); + expect(seen.snapshots).toEqual([{ ids: ["a1", "a2", "a9"], complete: true }]); + stop(); + }); + + test("stops at the page cap and marks the snapshot incomplete", async () => { + const pages = Array.from({ length: MAX_AGENT_PAGES + 2 }, (_, index) => [agent(`a${index}`)]); + const host = fakeHost({ observing: true, pages: () => pages }); + const seen = recorder(); + const stop = followAgentDirectory(host.paseo, seen.follower); + await flush(); + + expect(host.requests).toHaveLength(MAX_AGENT_PAGES); + expect(seen.snapshots).toHaveLength(1); + expect(seen.snapshots[0]?.complete).toBe(false); + expect(seen.snapshots[0]?.ids).toHaveLength(MAX_AGENT_PAGES); + stop(); + }); + + test("a reconnect snapshot supersedes paging still in flight", async () => { + let pages = [[agent("a1")], [agent("a2")]]; + const host = fakeHost({ observing: true, pages: () => pages }); + const release = host.hold(); + const seen = recorder(); + const stop = followAgentDirectory(host.paseo, seen.follower); + await flush(); + + pages = [[agent("b1")]]; + host.reconnect(); + await flush(); + release(); + await flush(); + + expect(seen.snapshots).toEqual([{ ids: ["b1"], complete: true }]); + stop(); + }); + + test("reopens a failed observation with backoff and leaves no timer behind", async () => { + let pages = [[agent("a1")]]; + const host = fakeHost({ observing: true, pages: () => pages }); + const seen = recorder(); + const stop = followAgentDirectory(host.paseo, seen.follower); + await flush(); + + pages = [[agent("a2")]]; + host.fail(new Error("reconnect request failed")); + await vi.advanceTimersByTimeAsync(1_999); + expect(host.requests).toHaveLength(1); + await vi.advanceTimersByTimeAsync(1); + expect(host.requests).toHaveLength(2); + expect(seen.snapshots.at(-1)).toEqual({ ids: ["a2"], complete: true }); + + host.fail(new Error("again")); + stop(); + expect(vi.getTimerCount()).toBe(0); + }); +}); + +describe("followAgentDirectory on a 0.8 client", () => { + test("keeps the listener and one upsert-only read, and never subscribes", async () => { + const host = fakeHost({ observing: false, pages: () => [[agent("a1")]] }); + const seen = recorder(); + const stop = followAgentDirectory(host.paseo, seen.follower); + await vi.waitFor(() => expect(seen.snapshots).toHaveLength(1)); + + expect(host.requests.every((request) => request.subscribe === undefined)).toBe(true); + expect(host.listening).toBe(true); + expect(seen.snapshots).toEqual([{ ids: ["a1"], complete: false }]); + + host.update({ kind: "upsert", agent: agent("a2") }); + expect(seen.upserts).toEqual(["a2"]); + + stop(); + expect(host.listening).toBe(false); + }); +}); From 6a3b7338fcd6c23e848ff333b033dce59468ce88 Mon Sep 17 00:00:00 2001 From: Omer Cohen <639682+omercnet@users.noreply.github.com> Date: Wed, 23 Sep 2026 18:58:02 +0000 Subject: [PATCH 2/2] fix(tell-agent): require owned agent observation --- tell-agent/client/agents.ts | 70 ++++++++++++------------------ tell-agent/paseo-plugin.json | 2 +- tell-agent/tests/agents.test.ts | 76 ++++++++++++++------------------- 3 files changed, 62 insertions(+), 86 deletions(-) diff --git a/tell-agent/client/agents.ts b/tell-agent/client/agents.ts index b44f453d..8cd55a38 100644 --- a/tell-agent/client/agents.ts +++ b/tell-agent/client/agents.ts @@ -1,13 +1,20 @@ -import type { - PaseoAgent, - PaseoAgentListResult, - PaseoAgentUpdate, - PaseoApi, -} from "@getpaseo/client"; +import type { PluginClientContext } from "@getpaseo/plugin/client"; type PaseoApi = PluginClientContext["paseo"]; -export type AgentEntry = Awaited>["entries"][number]; -type AgentSnapshot = AgentEntry["agent"]; +type PaseoAgentUpdate = Parameters[0]>[0]; +type AgentSnapshot = Extract["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; @@ -41,6 +48,7 @@ export async function loadAgents(paseo: PaseoApi): Promise { async function readRemainingPages( paseo: PaseoApi, first: PaseoAgentListResult, + signal: AbortSignal, current: () => boolean, ): Promise { const entries = [...first.entries]; @@ -50,6 +58,7 @@ async function readRemainingPages( 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); @@ -68,25 +77,14 @@ export type AgentDirectoryFollower = { remove(agentId: string): void; }; -/** - * `observeEvents` shipped together with owned observations (0.9.0-beta.1). A - * 0.8 client must not send `subscribe`: the daemon keeps one agents slot per - * legacy connection, last query wins, so it would replace the app's own. - */ -function ownsObservations(paseo: PaseoApi): boolean { - return typeof (paseo as { observeEvents?: unknown }).observeEvents === "function"; -} - /** * Keeps `follower` in step with the host's agents until the returned cleanup. * - * On 0.9, `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. + * `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. - * - * On 0.8 it listens and reads the directory once, as before. */ export function followAgentDirectory( paseo: PaseoApi, @@ -98,23 +96,6 @@ export function followAgentDirectory( else follower.upsert(update.agent); }; - if (!ownsObservations(paseo)) { - const unsubscribe = paseo.agents.subscribe((update) => { - if (!stopped) apply(update); - }); - void loadAgents(paseo) - .then(({ entries }) => { - if (stopped) return; - const agents = entries.map((entry) => entry.agent); - follower.snapshot(agents, false); - }) - .catch(() => undefined); - return () => { - stopped = true; - unsubscribe(); - }; - } - const lifetime = new AbortController(); let observation: { release(): Promise } | null = null; let reopenTimer: ReturnType | null = null; @@ -131,7 +112,7 @@ export function followAgentDirectory( landed = touched; let listed: AgentDirectoryPage | null; try { - listed = await readRemainingPages(paseo, first, isCurrent); + listed = await readRemainingPages(paseo, first, lifetime.signal, isCurrent); } catch { listed = { entries: [...first.entries], truncated: true }; } @@ -147,6 +128,8 @@ export function followAgentDirectory( const reopen = () => { observation = null; + generation += 1; + landed = null; if (stopped || reopenTimer !== null) return; reopenTimer = setTimeout(() => { reopenTimer = null; @@ -164,7 +147,10 @@ export function followAgentDirectory( signal: lifetime.signal, }) .then(({ subscription }) => { - if (stopped) return; + if (stopped) { + void subscription.release().catch(() => undefined); + return; + } observation = subscription; subscription.subscribe({ snapshot: (first) => { diff --git a/tell-agent/paseo-plugin.json b/tell-agent/paseo-plugin.json index 85dff5c7..81b895b3 100644 --- a/tell-agent/paseo-plugin.json +++ b/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" } } diff --git a/tell-agent/tests/agents.test.ts b/tell-agent/tests/agents.test.ts index b4a80258..29fee920 100644 --- a/tell-agent/tests/agents.test.ts +++ b/tell-agent/tests/agents.test.ts @@ -1,4 +1,4 @@ -import type { PaseoApi } from "@getpaseo/client"; +import type { PluginClientContext } from "@getpaseo/plugin/client"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { type AgentDirectoryFollower, @@ -32,15 +32,12 @@ function page(agents: Agent[], next?: string) { }; } -/** - * A host whose directory is `pages()`, one array per page. `observing` makes it - * a 0.9 client (observation from `list({ subscribe })`); otherwise a 0.8 one. - * `gate` holds plain page reads until released, to land updates mid-paging. +/** A host whose directory is `pages()`, one array per page. `gate` holds + * plain page reads until released, to land updates or failures mid-paging. */ -function fakeHost(options: { observing: boolean; pages: () => Agent[][] }) { +function fakeHost(options: { pages: () => Agent[][] }) { const requests: ListRequest[] = []; const observers: Observer[] = []; - let listener: ((update: unknown) => void) | null = null; let released = 0; let gate: Promise | null = null; const pageAt = (cursor?: string) => { @@ -49,17 +46,15 @@ function fakeHost(options: { observing: boolean; pages: () => Agent[][] }) { return page(pages[index] ?? [], index + 1 < pages.length ? String(index + 1) : undefined); }; const agents = { - subscribe(handler: (update: unknown) => void) { - listener = handler; - return () => { - listener = null; - }; + subscribe() { + return () => undefined; }, async list(request: ListRequest = {}) { requests.push(request); if (!request.subscribe) { + const result = pageAt(request.page?.cursor); if (gate) await gate; - return pageAt(request.page?.cursor); + return result; } return { ...pageAt(), @@ -78,14 +73,11 @@ function fakeHost(options: { observing: boolean; pages: () => Agent[][] }) { }; const observer = () => observers.at(-1); return { - paseo: (options.observing ? { observeEvents() {}, agents } : { agents }) as unknown as PaseoApi, + paseo: { agents } as unknown as PluginClientContext["paseo"], requests, get released() { return released; }, - get listening() { - return listener !== null; - }, hold() { let open = () => {}; gate = new Promise((resolve) => { @@ -98,7 +90,6 @@ function fakeHost(options: { observing: boolean; pages: () => Agent[][] }) { }, update(payload: unknown) { observer()?.update({ type: "agent_update", payload }); - listener?.(payload); }, reconnect() { observer()?.snapshot(pageAt()); @@ -138,7 +129,7 @@ describe("followAgentDirectory on a 0.9 client", () => { }); test("opens one observation and applies its snapshot and updates", async () => { - const host = fakeHost({ observing: true, pages: () => [[agent("a1"), agent("a2")]] }); + const host = fakeHost({ pages: () => [[agent("a1"), agent("a2")]] }); const seen = recorder(); const stop = followAgentDirectory(host.paseo, seen.follower); await flush(); @@ -150,7 +141,6 @@ describe("followAgentDirectory on a 0.9 client", () => { page: { limit: 200 }, }); expect(host.requests[0]?.signal).toBeInstanceOf(AbortSignal); - expect(host.listening).toBe(false); expect(seen.snapshots).toEqual([{ ids: ["a1", "a2"], complete: true }]); host.update({ kind: "upsert", agent: agent("a3") }); @@ -168,7 +158,6 @@ describe("followAgentDirectory on a 0.9 client", () => { test("reads later pages plainly and lets updates that land meanwhile win", async () => { const host = fakeHost({ - observing: true, pages: () => [[agent("a1")], [agent("a2"), agent("a3")]], }); const release = host.hold(); @@ -185,13 +174,14 @@ describe("followAgentDirectory on a 0.9 client", () => { expect(host.requests[1]).toMatchObject({ page: { limit: 200, cursor: "1" } }); expect(host.requests[1]?.subscribe).toBeUndefined(); + expect(host.requests[1]?.signal).toBe(host.requests[0]?.signal); expect(seen.snapshots).toEqual([{ ids: ["a1", "a2", "a9"], complete: true }]); stop(); }); test("stops at the page cap and marks the snapshot incomplete", async () => { const pages = Array.from({ length: MAX_AGENT_PAGES + 2 }, (_, index) => [agent(`a${index}`)]); - const host = fakeHost({ observing: true, pages: () => pages }); + const host = fakeHost({ pages: () => pages }); const seen = recorder(); const stop = followAgentDirectory(host.paseo, seen.follower); await flush(); @@ -205,7 +195,7 @@ describe("followAgentDirectory on a 0.9 client", () => { test("a reconnect snapshot supersedes paging still in flight", async () => { let pages = [[agent("a1")], [agent("a2")]]; - const host = fakeHost({ observing: true, pages: () => pages }); + const host = fakeHost({ pages: () => pages }); const release = host.hold(); const seen = recorder(); const stop = followAgentDirectory(host.paseo, seen.follower); @@ -221,9 +211,28 @@ describe("followAgentDirectory on a 0.9 client", () => { stop(); }); + test("drops a paging continuation after an observation failure", async () => { + let pages = [[agent("a1")], [agent("a2")]]; + const host = fakeHost({ pages: () => pages }); + const release = host.hold(); + const seen = recorder(); + const stop = followAgentDirectory(host.paseo, seen.follower); + await flush(); + + host.fail(new Error("connection lost")); + pages = [[agent("b1")]]; + release(); + await flush(); + expect(seen.snapshots).toEqual([]); + + await vi.advanceTimersByTimeAsync(2_000); + expect(seen.snapshots).toEqual([{ ids: ["b1"], complete: true }]); + stop(); + }); + test("reopens a failed observation with backoff and leaves no timer behind", async () => { let pages = [[agent("a1")]]; - const host = fakeHost({ observing: true, pages: () => pages }); + const host = fakeHost({ pages: () => pages }); const seen = recorder(); const stop = followAgentDirectory(host.paseo, seen.follower); await flush(); @@ -241,22 +250,3 @@ describe("followAgentDirectory on a 0.9 client", () => { expect(vi.getTimerCount()).toBe(0); }); }); - -describe("followAgentDirectory on a 0.8 client", () => { - test("keeps the listener and one upsert-only read, and never subscribes", async () => { - const host = fakeHost({ observing: false, pages: () => [[agent("a1")]] }); - const seen = recorder(); - const stop = followAgentDirectory(host.paseo, seen.follower); - await vi.waitFor(() => expect(seen.snapshots).toHaveLength(1)); - - expect(host.requests.every((request) => request.subscribe === undefined)).toBe(true); - expect(host.listening).toBe(true); - expect(seen.snapshots).toEqual([{ ids: ["a1"], complete: false }]); - - host.update({ kind: "upsert", agent: agent("a2") }); - expect(seen.upserts).toEqual(["a2"]); - - stop(); - expect(host.listening).toBe(false); - }); -});