diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 77ce691..e304952 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -1,7 +1,7 @@ # Contributing Use Node.js 24 LTS, npm, and the BB CLI. Release preparation uses Node 24.20.0; -older Node versions have not been verified. The public SDK is locked at 0.4.34. +older Node versions have not been verified. The public SDK is locked at 0.6.132. ```sh npm ci diff --git a/PLUGIN_OVERVIEW.md b/PLUGIN_OVERVIEW.md index 3a0c5e7..46f4140 100644 --- a/PLUGIN_OVERVIEW.md +++ b/PLUGIN_OVERVIEW.md @@ -17,7 +17,7 @@ reopen its notes from the dashboard whenever needed. ## Requirements -Requires BB 0.40 or newer; BB 0.46.0 is recommended. Ollama Fleet 0.6.0 or newer and Redteam's public observability RPC +Requires BB 0.47 or newer. Ollama Fleet 0.6.0 or newer and Redteam's public observability RPC are optional integrations. Missing integrations leave native BB observation available. Mobile layouts and reduced-motion preferences are supported. Hidden reasoning is excluded. System/push notifications are unavailable in diff --git a/README.md b/README.md index fbbe221..cd6e6d3 100644 --- a/README.md +++ b/README.md @@ -66,8 +66,8 @@ bb plugin build bb plugin install . ``` -Neither Fleet nor Redteam is required. Declared compatibility is BB >=0.40 and -SDK >=0.4.21; the development SDK is pinned to 0.4.34. See the honest +Neither Fleet nor Redteam is required. Declared compatibility is BB >=0.47 and +SDK >=0.6.73; the development SDK is pinned to 0.6.132. See the honest [compatibility matrix](RELEASE_AUDIT.md), including historical runtime evidence. For the prepared, unpublished 0.7.0 source, see [compatibility and upgrade notes](docs/upgrade-0.7.0.md). diff --git a/app.tsx b/app.tsx index 001c3d9..056e885 100644 --- a/app.tsx +++ b/app.tsx @@ -8,7 +8,7 @@ // surface-agnostic: it takes a rootThreadId, so a future entry surface // (header action, nav panel, message action) can mount it unchanged. import { definePluginApp } from "@get-bb/plugin-sdk/app"; -import type { PluginNavPanelProps, PluginThreadPanelProps } from "@get-bb/plugin-sdk"; +import type { ExperimentalThreadActionTabProps, PluginNavPanelProps } from "@get-bb/plugin-sdk"; import { Board } from "@/components/board/board"; import { GlobalDashboard } from "@/components/global/global-dashboard"; import "./app.css"; @@ -18,7 +18,7 @@ import "./app.css"; * persisted `params` (used when another surface opens this panel scoped to a * different root: {"rootThreadId": "thr_..."}). */ -function BoardPanel({ threadId, params }: PluginThreadPanelProps) { +function BoardPanel({ threadId, params }: ExperimentalThreadActionTabProps) { const fromParams = params !== null && typeof params === "object" && @@ -42,7 +42,9 @@ export default definePluginApp((app) => { path: "operations", component: GlobalDashboardPage, }); - app.slots.threadPanelAction({ + app.slots.experimental_tab({ + context: { kind: "thread" }, + availability: "action", id: "agent-board", title: "Agent Board", icon: "ListTodo", diff --git a/contract/rpc.ts b/contract/rpc.ts index 030830f..cca31a6 100644 --- a/contract/rpc.ts +++ b/contract/rpc.ts @@ -183,7 +183,7 @@ export const boardRootSchema = z.object({ goal: z .object({ objective: z.string(), - status: z.enum(["active", "paused", "budgetLimited", "complete"]), + status: z.enum(["active", "paused", "blocked", "usageLimited", "budgetLimited", "complete"]), tokensUsed: z.number(), tokenBudget: z.number().nullable(), timeUsedSeconds: z.number(), @@ -395,6 +395,21 @@ export const boardChangedPayloadSchema = z.object({ global: z.boolean().optional(), }); +/** Query tags: every board read, one root's board read, and the global page read. */ +export const BOARD_TAG = "board"; +export const GLOBAL_DASHBOARD_TAG = "global-dashboard"; +export function boardTag(rootThreadId: string): string { + return `board:${rootThreadId}`; +} + +/** Tags made stale by one watcher invalidation payload (same rules the realtime listeners apply). */ +export function boardChangedTags(payload: z.infer): string[] { + if (payload.roots.length === 0) { + return payload.global === true ? [GLOBAL_DASHBOARD_TAG] : [BOARD_TAG, GLOBAL_DASHBOARD_TAG]; + } + return [...payload.roots.map(boardTag), ...(payload.global === true ? [GLOBAL_DASHBOARD_TAG] : [])]; +} + export const rpcContract = { release_announcement: { input: z.object({}), @@ -412,6 +427,7 @@ export const rpcContract = { board_snapshot: { input: z.object({ threadId: z.string().min(1), fresh: z.boolean().optional() }), output: boardSnapshotSchema, + tags: (input: { threadId: string }) => [BOARD_TAG, boardTag(input.threadId)], }, /** Bounded recent-activity tail for one thread (details expansion). */ board_activity: { @@ -430,6 +446,7 @@ export const rpcContract = { global_dashboard: { input: z.object({ fresh: z.boolean().optional() }), output: globalDashboardSnapshotSchema, + tags: () => [GLOBAL_DASHBOARD_TAG], }, /** * Presentation-only: hide one attention item from the global dashboard. diff --git a/docs/architecture.md b/docs/architecture.md index bca105b..c3ca470 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -36,9 +36,13 @@ Redteam timelines reuse the same public observability snapshot and generic UI. O ## Realtime and polling -BB changes use the existing public thread lifecycle/entity stream, a 400 ms coalescing watcher, and the plugin's `agent-board-changed` frontend signal. +BB changes use the existing public thread lifecycle/entity stream and a 400 ms coalescing watcher. The board and global page read their snapshots with `experimental_useRpcQuery`; the watcher invalidates the `board:` and `global-dashboard` tags with `bb.rpc.experimental_invalidate`, and the host refetches open views and refreshes them after reconnects. The `agent-board-changed` realtime signal remains for the Timeline dialog and the global page's monitoring health line. -SDK 0.4.21 has no supported cross-plugin realtime subscription. BB changes therefore opportunistically re-check Redteam. After a snapshot discovers a queued/running Redteam scan, a server-side fallback probes that project/global snapshot every 3 seconds. Requests share the Redteam source's inflight guard and 1.5-second TTL and never overlap. Polling stops when observed scans are terminal or the watched lease expires. While the global page is mounted, one slow 20-second snapshot refresh discovers brand-new Redteam work that has no BB event; unmounting clears that timer. BB thread changes remain event-driven. +A board or global page counts as open while a client watches its read: `bb.rpc.experimental_whileWatched` starts the watcher, Redteam and Fleet bookkeeping for that read and aborts it when the last client leaves or hides the tab. A thread the board has not seen yet (for example a grandchild created between snapshots) is matched to its root with one batched `experimental_listAncestors` read per tick. + +There is no supported cross-plugin realtime subscription. BB changes therefore opportunistically re-check Redteam. After a snapshot discovers a queued/running Redteam scan, a server-side fallback probes that project/global snapshot every 3 seconds. Requests share the Redteam source's inflight guard and 1.5-second TTL and never overlap. Polling stops when observed scans are terminal or the view's watched read ends. While the global page is watched, the backend invalidates it every 20 seconds to discover brand-new Redteam work that has no BB event. BB thread changes remain event-driven. + +The thread tree is read with one `experimental_listDescendants` call and one batched `threads.list({ threadIds })` row read; the per-thread timeline, model, output, event and queue reads run in parallel, six threads at a time. ## Bounds @@ -52,11 +56,11 @@ The shared Timeline uses the existing responsive dialog/drawer primitive. Rows a ## Compatibility -- BB engine: `>=0.40` -- Runtime Plugin SDK floor: `>=0.4.21` -- Current development types: 0.4.34 +- BB engine: `>=0.47` +- Runtime Plugin SDK floor: `>=0.6.73` +- Current development types: 0.6.132 -Server modules use relative imports and avoid `@/` aliases so the BB 0.40 daemon loader can load source when artifact SDK versions differ. Newer lifecycle names are registered opportunistically, and missing newer thread fields degrade gracefully. +Server modules use relative imports and avoid `@/` aliases so the daemon loader can load source when artifact SDK versions differ. Newer lifecycle names are registered opportunistically, and missing newer thread fields degrade gracefully. ## Commands @@ -119,7 +123,7 @@ All views share a Fleet cache and one physical inflight request. Idle cache TTL is 20 seconds, matching the existing global discovery cycle. New route-tool activity can refresh discovery at a bounded two-second TTL. Once active jobs are discovered, one 2.5-second loop polls regardless of job count, stops at terminal state or -source failure, and expires unused watches after 90 seconds. No idle Fleet timer +source failure, and stops when no board or global page is watched. No idle Fleet timer is created. Meaningful changes invalidate board/timeline generations and publish Agent Board's normal signal. Invalidation during a source read discards its result and retries sequentially; stale RUNNING cannot re-enter the cache. diff --git a/hooks/useBoard.ts b/hooks/useBoard.ts index 16a58fa..20ad9dd 100644 --- a/hooks/useBoard.ts +++ b/hooks/useBoard.ts @@ -1,22 +1,19 @@ // bb-plugin-agent-board — the board data hook. // -// One bounded fetch per invalidation signal: the backend publishes -// "agent-board-changed" and this hook refetches board_snapshot. No polling. -// A 1s local ticker drives live elapsed timers between snapshots. -import { useCallback, useEffect, useMemo, useRef, useState } from "react"; -import { useRealtime, useRealtimeConnectionState, useRpc } from "@get-bb/plugin-sdk/app"; -import { - BOARD_CHANGED_CHANNEL, - boardChangedPayloadSchema, - type BoardSnapshot, - type rpcContract, -} from "@/contract/rpc"; +// The board reads board_snapshot through BB's shared RPC query cache. The +// backend invalidates the board's tag when a watched thread changes, and the +// host refetches on invalidation and after reconnects. No polling. Holding +// the read is also what keeps the backend's watcher and pollers running for +// this root. A 1s local ticker drives live elapsed timers between snapshots. +import { useCallback, useEffect, useMemo, useState } from "react"; +import { experimental_useRpcQuery, useRpc } from "@get-bb/plugin-sdk/app"; +import { rpcContract, type BoardSnapshot } from "@/contract/rpc"; interface BoardState { snapshot: BoardSnapshot | null; - /** Set after the first fetch for a threadId resolves or rejects. */ + /** True until the first fetch for a threadId resolves or rejects. */ loading: boolean; - /** Fetch in flight after the first (manual refresh / signal refetch). */ + /** Fetch in flight after the first (manual refresh / invalidation refetch). */ refreshing: boolean; error: string | null; /** Advances every second while any card is still live (running/waiting). */ @@ -24,100 +21,36 @@ interface BoardState { refetch: (options?: { fresh?: boolean }) => void; } -/** - * Parse a realtime payload defensively: only the `roots` array is trusted, - * and anything malformed means "refetch" (the safe default). - */ -export function isForThisBoard(payload: unknown, rootThreadId: string): boolean { - const parsed = boardChangedPayloadSchema.safeParse(payload); - if (!parsed.success) return true; // unreadable signal: assume relevant - const roots = parsed.data.roots; - if (roots.length === 0) return parsed.data.global !== true; - return roots.includes(rootThreadId); -} - export function useBoard(rootThreadId: string): BoardState { const rpc = useRpc(); - const [snapshot, setSnapshot] = useState(null); - const [loading, setLoading] = useState(true); - const [refreshing, setRefreshing] = useState(false); - const [error, setError] = useState(null); + const query = experimental_useRpcQuery({ + contract: rpcContract, + method: "board_snapshot", + input: { threadId: rootThreadId }, + }); + const snapshot = query.data ?? null; + const [freshPending, setFreshPending] = useState(false); const [tick, setTick] = useState(0); - const requestId = useRef(0); + const refetchQuery = query.refetch; + // `fresh` bypasses the backend's short-lived caches; the follow-up read + // then serves that fresh snapshot to every view of this board. const refetch = useCallback( (options?: { fresh?: boolean }) => { - const id = ++requestId.current; - if (snapshot === null) setLoading(true); - else setRefreshing(true); - rpc - .call("board_snapshot", { - threadId: rootThreadId, - fresh: options?.fresh === true, - }) - .then( - (result) => { - if (id !== requestId.current) return; // a newer fetch superseded this one - setSnapshot(result); - setError(null); - }, - (cause: unknown) => { - if (id !== requestId.current) return; - setError(cause instanceof Error ? cause.message : String(cause)); - }, - ) - .finally(() => { - if (id !== requestId.current) return; - setLoading(false); - setRefreshing(false); - }); + if (options?.fresh !== true) { + void refetchQuery(); + return; + } + setFreshPending(true); + void rpc + .call("board_snapshot", { threadId: rootThreadId, fresh: true }) + .catch(() => undefined) + .then(() => refetchQuery()) + .finally(() => setFreshPending(false)); }, - [rpc, rootThreadId, snapshot], + [rpc, rootThreadId, refetchQuery], ); - // Initial fetch when the board opens (and on thread change). - useEffect(() => { - requestId.current++; - setSnapshot(null); - setError(null); - setLoading(true); - rpc - .call("board_snapshot", { threadId: rootThreadId }) - .then( - (result) => { - setSnapshot(result); - setError(null); - }, - (cause: unknown) => { - setError(cause instanceof Error ? cause.message : String(cause)); - }, - ) - .finally(() => { - setLoading(false); - }); - }, [rpc, rootThreadId]); - - // Backend invalidation → refetch (bounded: debounced server-side, cached - // server-side, race-guarded here). - useRealtime(BOARD_CHANGED_CHANNEL, (payload) => { - if (isForThisBoard(payload, rootThreadId)) refetch(); - }); - - // After a realtime reconnect, reconcile: signals emitted while offline - // were missed, so refetch once (but not on the first connect — the mount - // effect already fetched). - const connectionState = useRealtimeConnectionState(); - const everConnected = useRef(false); - const prevState = useRef(connectionState); - useEffect(() => { - const was = prevState.current; - prevState.current = connectionState; - if (connectionState === "connected" && was !== "connected" && everConnected.current) { - refetch(); - } - if (connectionState === "connected") everConnected.current = true; - }, [connectionState, refetch]); - // Local timer: only while work is live, so idle boards cost nothing. const hasLive = useMemo( () => @@ -133,5 +66,12 @@ export function useBoard(rootThreadId: string): BoardState { return () => clearInterval(timer); }, [hasLive]); - return { snapshot, loading, refreshing, error, tick, refetch }; + return { + snapshot, + loading: query.isLoading, + refreshing: freshPending || (query.isFetching && snapshot !== null), + error: query.error?.message ?? null, + tick, + refetch, + }; } diff --git a/hooks/useGlobalDashboard.ts b/hooks/useGlobalDashboard.ts index e9defb7..6fa7dde 100644 --- a/hooks/useGlobalDashboard.ts +++ b/hooks/useGlobalDashboard.ts @@ -1,17 +1,16 @@ -// Global dashboard data hook. Core thread events drive immediate refreshes; -// one slow bounded discovery check catches unrelated optional-plugin work. -import { useCallback, useEffect, useRef, useState } from "react"; -import { useRealtime, useRealtimeConnectionState, useRpc } from "@get-bb/plugin-sdk/app"; +// Global dashboard data hook. The page reads global_dashboard through BB's +// shared RPC query cache; the backend invalidates its tag on BB changes and, +// while the page is watched, every 20 seconds to discover new optional-plugin +// work. The host refetches on invalidation and after reconnects. +import { useCallback, useEffect, useState } from "react"; +import { experimental_useRpcQuery, useRealtime, useRealtimeConnectionState, useRpc } from "@get-bb/plugin-sdk/app"; import { BOARD_CHANGED_CHANNEL, boardChangedPayloadSchema, + rpcContract, type AttentionItem, - type GlobalDashboardSnapshot, - type rpcContract, } from "@/contract/rpc"; -export const GLOBAL_DISCOVERY_INTERVAL_MS = 20_000; - export function isForGlobalDashboard(payload: unknown): boolean { const parsed = boardChangedPayloadSchema.safeParse(payload); if (!parsed.success) return true; @@ -20,80 +19,44 @@ export function isForGlobalDashboard(payload: unknown): boolean { export function useGlobalDashboard() { const rpc = useRpc(); - const [snapshot, setSnapshot] = useState(null); - const [loading, setLoading] = useState(true); - const [refreshing, setRefreshing] = useState(false); - const [error, setError] = useState(null); + const query = experimental_useRpcQuery({ + contract: rpcContract, + method: "global_dashboard", + input: {}, + }); + const snapshot = query.data ?? null; + const [freshPending, setFreshPending] = useState(false); + const [actionError, setActionError] = useState(null); const [tick, setTick] = useState(0); const [lastSignalAt, setLastSignalAt] = useState(null); const [lastRefreshAt, setLastRefreshAt] = useState(null); - const snapshotRef = useRef(null); - const inflight = useRef(false); - /** - * A refetch signal arrived while a fetch was in flight. The signal is not - * dropped (the old bug: a terminal-state signal lost this way never - * returned, leaving the dashboard showing RUNNING forever) — instead one - * trailing refetch runs when the in-flight request settles. Bounded: the - * flag collapses any signal burst into a single retry, and the retry only - * runs when a signal actually arrived, so it can never loop or storm. - */ - const pendingSignal = useRef(false); - const requestId = useRef(0); + const refetchQuery = query.refetch; + const fetchedAt = snapshot?.fetchedAt; + + useEffect(() => { + if (fetchedAt === undefined) return; + setLastRefreshAt(Date.now()); + setActionError(null); + }, [fetchedAt]); const refetch = useCallback((options?: { fresh?: boolean }) => { - if (inflight.current) { - pendingSignal.current = true; + if (options?.fresh !== true) { + void refetchQuery(); return; } - inflight.current = true; - const id = ++requestId.current; - if (snapshotRef.current === null) setLoading(true); - else setRefreshing(true); - rpc.call("global_dashboard", { fresh: options?.fresh === true }).then( - (result) => { - if (id !== requestId.current) return; - snapshotRef.current = result; - setSnapshot(result); - setLastRefreshAt(Date.now()); - setError(null); - }, - (cause: unknown) => { - if (id !== requestId.current) return; - setError(cause instanceof Error ? cause.message : String(cause)); - }, - ).finally(() => { - if (id !== requestId.current) return; - inflight.current = false; - setLoading(false); - setRefreshing(false); - if (pendingSignal.current) { - pendingSignal.current = false; - refetch(options); - } - }); - }, [rpc]); - - useEffect(() => { - refetch(); - return () => { - requestId.current += 1; - inflight.current = false; - }; - }, [refetch]); + setFreshPending(true); + void rpc + .call("global_dashboard", { fresh: true }) + .catch(() => undefined) + .then(() => refetchQuery()) + .finally(() => setFreshPending(false)); + }, [rpc, refetchQuery]); useRealtime(BOARD_CHANGED_CHANNEL, (payload) => { - if (isForGlobalDashboard(payload)) { setLastSignalAt(Date.now()); refetch(); } + if (isForGlobalDashboard(payload)) setLastSignalAt(Date.now()); }); const connectionState = useRealtimeConnectionState(); - const everConnected = useRef(false); - const previousConnection = useRef(connectionState); - useEffect(() => { - const was = previousConnection.current; - previousConnection.current = connectionState; - if (connectionState === "connected" && was !== "connected" && everConnected.current) refetch(); - if (connectionState === "connected") everConnected.current = true; - }, [connectionState, refetch]); /** Presentation-only: hide one attention item (persists via plugin storage). */ const dismissAttention = useCallback( @@ -105,12 +68,10 @@ export function useGlobalDashboard() { updatedAt: item.updatedAt, }); } catch (cause: unknown) { - setError(cause instanceof Error ? cause.message : String(cause)); - return; + setActionError(cause instanceof Error ? cause.message : String(cause)); } - refetch(); }, - [rpc, refetch], + [rpc], ); /** Presentation-only: forget every stored attention dismissal. */ @@ -118,19 +79,9 @@ export function useGlobalDashboard() { try { await rpc.call("clear_dismissed_attentions", {}); } catch (cause: unknown) { - setError(cause instanceof Error ? cause.message : String(cause)); - return; + setActionError(cause instanceof Error ? cause.message : String(cause)); } - refetch(); - }, [rpc, refetch]); - - // BB lifecycle events are immediate. This slow mounted-page check exists - // only to discover a new Redteam run, whose realtime channel cannot be - // subscribed to cross-plugin on runtime SDK 0.4.21. - useEffect(() => { - const timer = setInterval(() => refetch(), GLOBAL_DISCOVERY_INTERVAL_MS); - return () => clearInterval(timer); - }, [refetch]); + }, [rpc]); useEffect(() => { const timer = setInterval(() => setTick((value) => value + 1), 1_000); @@ -139,9 +90,9 @@ export function useGlobalDashboard() { return { snapshot, - loading, - refreshing, - error, + loading: query.isLoading, + refreshing: freshPending || (query.isFetching && snapshot !== null), + error: actionError ?? query.error?.message ?? null, tick, refetch, dismissAttention, diff --git a/lib/bb-source.ts b/lib/bb-source.ts index c7e812a..28682b8 100644 --- a/lib/bb-source.ts +++ b/lib/bb-source.ts @@ -16,12 +16,15 @@ // - Statuses are mapped losslessly enough to drive column placement. import type { BbPluginApi } from "@get-bb/plugin-sdk"; import { reportedCompletion } from "./dashboard-view"; -import type { - BoardActivity, - BoardActivityFeed, - BoardActivityFeedEntry, - BoardCard, - BoardCardStatus, +import { z } from "zod"; +import { + boardRootSchema, + type BoardActivity, + type BoardActivityFeed, + type BoardActivityFeedEntry, + type BoardCard, + type BoardCardStatus, + type BoardRoot, } from "../contract/rpc"; import { BOARD_LIMITS, @@ -45,7 +48,7 @@ type Threads = Sdk["threads"]; export type ThreadListRow = Awaited>[number]; /** threads.get() result (with or without includes). */ export type ThreadGetResult = Awaited>; -export type TimelineResponse = Awaited>; +export type TimelineResponse = Exclude>, { delta: unknown }>; export type TimelineRow = TimelineResponse["rows"][number]; export type TimelineSystemRow = Extract; export type TimelineConversationRow = Extract; @@ -945,6 +948,23 @@ export function activityFromEvents(events: readonly ThreadEventRow[]): BoardActi // Root summary // --------------------------------------------------------------------------- +/** + * BB 0.49 moved the Codex goal out of the timeline's typed `goal` field into + * the Codex plugin's extension state. Read the extension state first and fall + * back to the field older hosts still send. + */ +const CODEX_GOAL_EXTENSION_KIND = "provider-codex/goal"; +const timelineGoalSchema = boardRootSchema.shape.goal.unwrap().extend({ + tokenBudget: z.number().nullable().optional().transform((value) => value ?? null), +}); + +export function timelineGoal(timeline: TimelineResponse): BoardRoot["goal"] { + const extensionGoal = timeline.extensionStates?.[CODEX_GOAL_EXTENSION_KIND]; + const legacyGoal = "goal" in timeline ? timeline.goal : undefined; + const parsed = timelineGoalSchema.safeParse(extensionGoal ?? legacyGoal); + return parsed.success ? parsed.data : null; +} + export function rootSummary( facts: ThreadFacts, ctx: NormalizeContext, @@ -959,17 +979,11 @@ export function rootSummary( projectId: string | null; createdAt: number | null; updatedAt: number | null; - goal: { - objective: string; - status: "active" | "paused" | "budgetLimited" | "complete"; - tokensUsed: number; - tokenBudget: number | null; - timeUsedSeconds: number; - } | null; + goal: BoardRoot["goal"]; contextWindowUsedTokens: number | null; contextWindowTotalTokens: number | null; } { - const goal = facts.timeline.goal ?? null; + const goal = timelineGoal(facts.timeline); const context = facts.timeline.contextWindowUsage ?? null; return { threadId: facts.row.id, @@ -982,15 +996,7 @@ export function rootSummary( projectId: facts.row.projectId ?? null, createdAt: facts.row.createdAt, updatedAt: facts.row.updatedAt, - goal: goal - ? { - objective: goal.objective, - status: goal.status, - tokensUsed: goal.tokensUsed, - tokenBudget: goal.tokenBudget ?? null, - timeUsedSeconds: goal.timeUsedSeconds, - } - : null, + goal, contextWindowUsedTokens: context?.usedTokens ?? null, contextWindowTotalTokens: context?.modelContextWindow ?? null, }; diff --git a/lib/collect.ts b/lib/collect.ts index 124b1d2..045960e 100644 --- a/lib/collect.ts +++ b/lib/collect.ts @@ -23,7 +23,6 @@ import { type ThreadEventRow, type ThreadFacts, type ThreadGetResult, - type ThreadListRow, type ThreadRowLike, type TimelineResponse, } from "./bb-source"; @@ -61,7 +60,7 @@ export interface BoardSnapshot { } const TTL = { - children: 1_000, + tree: 1_000, timeline: 2_000, events: 2_000, output: 3_000, @@ -83,7 +82,7 @@ export const PROJECT_METADATA_TIMEOUT_MS = 750; export class BoardCollector { private readonly sdk: Sdk; private readonly now: () => number; - private readonly childrenCache = new Map>(); + private readonly treeCache = new Map>(); private readonly timelineCache = new Map>(); private readonly eventsCache = new Map>(); private readonly outputCache = new Map>(); @@ -137,7 +136,7 @@ export class BoardCollector { if (invalidateFleet) this.fleet.invalidate(); this.generation += 1; this.globalGeneration += 1; - this.childrenCache.clear(); + this.treeCache.clear(); this.timelineCache.clear(); this.eventsCache.clear(); this.outputCache.clear(); @@ -158,12 +157,6 @@ export class BoardCollector { // -- individual reads ---------------------------------------------------- - private childrenOf(threadId: string): Promise { - return this.cached(this.childrenCache, threadId, TTL.children, async () => - this.sdk.threads.list({ parentThreadId: threadId, includeHidden: true }), - ); - } - private timelineOf(threadId: string): Promise { return this.cached(this.timelineCache, threadId, TTL.timeline, async () => this.sdk.threads.timeline({ @@ -228,6 +221,11 @@ export class BoardCollector { } // -- tree walk ----------------------------------------------------------- + // + // One recursive descendants read plus one batched row read replace the + // per-node children listing. Rows come back in BB's thread-list order, so + // siblings keep the order the per-parent listing produced; the bounded + // breadth-first placement below then applies the same depth and count caps. private async walkTree( rootRow: ThreadRowLike, @@ -237,43 +235,25 @@ export class BoardCollector { childIds: Map; partial: boolean; }> { - const rows: ThreadRowLike[] = [rootRow]; - const depths = new Map([[rootRow.id, 0]]); - const childIds = new Map([[rootRow.id, []]]); - const queue: Array<{ id: string; depth: number }> = [{ id: rootRow.id, depth: 0 }]; - const seen = new Set([rootRow.id]); - let partial = false; - while (queue.length > 0) { - const { id, depth } = queue.shift()!; - if (depth >= BOARD_LIMITS.maxDepth) { - // Depth frontier: one bounded probe so `partial` tells the truth about - // unseen deeper threads (we never walk them). - const deeper = await this.childrenOf(id); - if (deeper.length > 0) partial = true; - continue; - } - const children = await this.childrenOf(id); - const ids: string[] = []; - for (const child of children) { - if (seen.has(child.id)) continue; - if (rows.length >= BOARD_LIMITS.maxThreads) { - partial = true; - break; - } - seen.add(child.id); - rows.push(child); - depths.set(child.id, depth + 1); - ids.push(child.id); - childIds.set(child.id, []); - queue.push({ id: child.id, depth: depth + 1 }); - } - childIds.set(id, ids); - if (rows.length >= BOARD_LIMITS.maxThreads) { - partial = true; - break; - } - } - return { rows, depths, childIds, partial }; + const descendants = await this.cached(this.treeCache, rootRow.id, TTL.tree, async () => { + const listed = await this.sdk.threads.experimental_listDescendants({ + threadIds: [rootRow.id], + includeHidden: true, + }); + const descendantIds = + listed.threads.find((entry) => entry.threadId === rootRow.id)?.descendantIds ?? []; + const candidateIds = descendantIds.slice(0, TREE_ROW_READ_LIMIT); + const rows: ThreadRowLike[] = + candidateIds.length === 0 + ? [] + : await this.sdk.threads.list({ + threadIds: candidateIds, + includeHidden: true, + limit: candidateIds.length, + }); + return { rows, truncated: descendantIds.length > candidateIds.length }; + }); + return placeTree(rootRow, descendants.rows, descendants.truncated); } // -- snapshot ------------------------------------------------------------ @@ -323,9 +303,23 @@ export class BoardCollector { string, { label: string | null; output: string | null } >(); + const facts = await mapWithConcurrency(rows, SNAPSHOT_READ_CONCURRENCY, async (row) => { + const isRoot = row.id === rootThreadId; + const needsOutput = row.status === "idle" || (!isRoot && row.status === "error"); + const needsEvents = row.status === "error"; + const [timeline, model, output, events, queuedMessages] = await Promise.all([ + this.timelineOf(row.id), + this.modelOf(row.id), + needsOutput ? this.outputOf(row.id) : Promise.resolve(null), + needsEvents ? this.eventsTail(row.id) : Promise.resolve(undefined), + (row.queuedMessageCount ?? 0) > 0 + ? this.queuedMessagesOf(row.id) + : Promise.resolve(undefined), + ]); + return { row, isRoot, timeline, model, output, events, queuedMessages }; + }); const timelines = new Map(); - for (const row of rows) { - const timeline = await this.timelineOf(row.id); + for (const { row, timeline } of facts) { timelines.set(row.id, timeline); for (const item of delegationIndex(timeline)) { if (item.childRef && !delegationsByChild.has(item.childRef)) { @@ -337,21 +331,8 @@ export class BoardCollector { } } - for (const row of rows) { - const timeline = timelines.get(row.id); - if (!timeline) continue; - const isRoot = row.id === rootThreadId; - const needsOutput = row.status === "idle" || (!isRoot && row.status === "error"); - const needsEvents = row.status === "error"; - const [model, output, events, queuedMessages] = await Promise.all([ - this.modelOf(row.id), - needsOutput ? this.outputOf(row.id) : Promise.resolve(null), - needsEvents ? this.eventsTail(row.id) : Promise.resolve(undefined), - (row.queuedMessageCount ?? 0) > 0 - ? this.queuedMessagesOf(row.id) - : Promise.resolve(undefined), - ]); - const facts: ThreadFacts = { + for (const { row, isRoot, timeline, model, output, events, queuedMessages } of facts) { + const threadFacts: ThreadFacts = { row, timeline, events, @@ -363,7 +344,7 @@ export class BoardCollector { childThreadIds: childIds.get(row.id) ?? [], }; const delegation = isRoot ? null : (delegationsByChild.get(row.id) ?? null); - for (const card of normalizeThread(facts, ctx, delegation)) cards.push(card); + for (const card of normalizeThread(threadFacts, ctx, delegation)) cards.push(card); } // Root summary carries goal + context window usage from the root timeline. @@ -837,6 +818,71 @@ function globalSummaryCard( } /** Small fixed-concurrency mapper so bounded enrichment does not become a request burst. */ +/** Per-snapshot parallelism for the per-thread reads. */ +const SNAPSHOT_READ_CONCURRENCY = 6; +/** One batched row read; `threads.list({ threadIds })` accepts 200 ids. */ +const TREE_ROW_READ_LIMIT = 200; + +/** + * Bounded breadth-first placement of a root's descendants. Only threads + * reachable from the root through placed parents are kept (a child of an + * archived thread is not), matching a walk of per-parent listings. `partial` + * is true when a reachable thread was left out by the depth or count cap, or + * when the descendant list was longer than one row read. + */ +export function placeTree( + rootRow: ThreadRowLike, + descendantRows: readonly ThreadRowLike[], + truncated: boolean, +): { + rows: ThreadRowLike[]; + depths: Map; + childIds: Map; + partial: boolean; +} { + const childrenByParent = new Map(); + for (const row of descendantRows) { + const parentId = row.parentThreadId ?? null; + if (parentId === null || row.id === rootRow.id) continue; + const siblings = childrenByParent.get(parentId); + if (siblings) siblings.push(row); + else childrenByParent.set(parentId, [row]); + } + const rows: ThreadRowLike[] = [rootRow]; + const depths = new Map([[rootRow.id, 0]]); + const childIds = new Map([[rootRow.id, []]]); + let partial = truncated; + const queue: Array<{ id: string; depth: number }> = [{ id: rootRow.id, depth: 0 }]; + for (let index = 0; index < queue.length; index += 1) { + const { id, depth } = queue[index]!; + const children = childrenByParent.get(id) ?? []; + if (children.length === 0) continue; + if (depth >= BOARD_LIMITS.maxDepth) { + partial = true; + continue; + } + const ids: string[] = []; + for (const child of children) { + if (depths.has(child.id)) continue; + if (rows.length >= BOARD_LIMITS.maxThreads) { + partial = true; + break; + } + rows.push(child); + depths.set(child.id, depth + 1); + childIds.set(child.id, []); + ids.push(child.id); + queue.push({ id: child.id, depth: depth + 1 }); + } + childIds.set(id, ids); + if (rows.length >= BOARD_LIMITS.maxThreads) { + if (queue.slice(index + 1).some((pending) => childrenByParent.has(pending.id))) partial = true; + break; + } + } + return { rows, depths, childIds, partial }; +} + async function mapWithConcurrency( values: readonly T[], concurrency: number, diff --git a/lib/fleet-poll.ts b/lib/fleet-poll.ts index a992f52..5773a4b 100644 --- a/lib/fleet-poll.ts +++ b/lib/fleet-poll.ts @@ -1,28 +1,35 @@ -// Exactly one active-only loop, independent of job/parent count. Idle discovery -// belongs to the existing global 20-second refresh and ordinary BB changes. +// Exactly one active-only loop, independent of job/parent count, running only +// while a client watches a board or the global page. Idle discovery belongs to +// the global page's 20-second refresh and ordinary BB changes. export class FleetActivePoller { private timer: ReturnType | null = null; private inflight = false; private disposed = false; private active = false; - private touchedAt = 0; + private watchers = 0; constructor( private deps: { poll: () => Promise<{ changed: boolean; active: boolean }>; changed: () => void; - now?: () => number; }, ) {} - private now() { - return this.deps.now?.() ?? Date.now(); + /** Allow polling until the watched read's signal aborts. */ + watch(signal: AbortSignal) { + if (this.disposed || signal.aborted) return; + this.watchers += 1; + signal.addEventListener("abort", () => { + this.watchers = Math.max(0, this.watchers - 1); + if (this.watchers > 0) return; + this.active = false; + this.stop(); + }, { once: true }); } observe(active: boolean) { - if (this.disposed) return; + if (this.disposed || this.watchers === 0) return; // Preserve one scheduled tick to publish a terminal result another view // discovered before the poll. The tick consumes the shared cache and stops. if (!active && this.active && (this.timer || this.inflight)) return; this.active = active; - this.touchedAt = this.now(); if (active) this.schedule(); else this.stop(); } @@ -39,16 +46,11 @@ export class FleetActivePoller { this.timer.unref?.(); } async tick() { - if (this.inflight || this.disposed || !this.active) return; - if (this.now() - this.touchedAt > 90000) { - this.active = false; - this.stop(); - return; - } + if (this.inflight || this.disposed || !this.active || this.watchers === 0) return; this.inflight = true; try { const result = await this.deps.poll(); - if (this.disposed) return; + if (this.disposed || this.watchers === 0) return; this.active = result.active; if (result.changed) this.deps.changed(); } catch { diff --git a/lib/redteam-poll.ts b/lib/redteam-poll.ts index e2b8829..0d9cef1 100644 --- a/lib/redteam-poll.ts +++ b/lib/redteam-poll.ts @@ -1,51 +1,69 @@ // Conservative fallback for SDK 0.4.21, which has cross-plugin RPC but no // supported cross-plugin realtime subscription. Polls only after an active -// Redteam scan has been discovered for an open board. +// Redteam scan has been discovered for a board or global page that a client +// is watching; the read's watched signal ends polling when the view closes. export interface RedteamPollDeps { poll: (projectId: string) => Promise<{ changed: boolean; active: boolean }>; changed: (projectId: string) => void; pollGlobal?: () => Promise<{ changed: boolean; active: boolean }>; changedGlobal?: () => void; - now?: () => number; intervalMs?: number; - watchedTtlMs?: number; } export class RedteamActivePoller { - private readonly projects = new Map>(); + private readonly projects = new Map>(); + private readonly watchedRoots = new Map(); private readonly inflight = new Set(); - private globalTouchedAt: number | null = null; + private globalWatchers = 0; private globalActive = false; private globalInflight = false; - private readonly now: () => number; private readonly watchedTtlMs: number; private timer: ReturnType | null = null; private disposed = false; constructor(private readonly deps: RedteamPollDeps) { - this.now = deps.now ?? (() => Date.now()); this.watchedTtlMs = deps.watchedTtlMs ?? 90_000; this.timer = setInterval(() => { void this.tick(); }, deps.intervalMs ?? 3_000); if (typeof this.timer === "object" && this.timer !== null && "unref" in this.timer) (this.timer as unknown as { unref(): void }).unref(); } + /** Keep a board's root eligible for polling until its watched read ends. */ + watch(rootThreadId: string, signal: AbortSignal): void { + if (this.disposed || signal.aborted) return; + this.watchedRoots.set(rootThreadId, (this.watchedRoots.get(rootThreadId) ?? 0) + 1); + signal.addEventListener("abort", () => { + const count = (this.watchedRoots.get(rootThreadId) ?? 0) - 1; + if (count > 0) { this.watchedRoots.set(rootThreadId, count); return; } + this.watchedRoots.delete(rootThreadId); + for (const [projectId, roots] of this.projects) { roots.delete(rootThreadId); if (roots.size === 0) this.projects.delete(projectId); } + }, { once: true }); + } + + /** Keep the global page eligible for polling until its watched read ends. */ + watchGlobal(signal: AbortSignal): void { + if (this.disposed || signal.aborted) return; + this.globalWatchers += 1; + signal.addEventListener("abort", () => { + this.globalWatchers = Math.max(0, this.globalWatchers - 1); + if (this.globalWatchers === 0) this.globalActive = false; + }, { once: true }); + } + observe(rootThreadId: string, projectId: string | null, active: boolean): void { if (this.disposed || !projectId) return; - if (!active) { + if (!active || !this.watchedRoots.has(rootThreadId)) { const roots = this.projects.get(projectId); roots?.delete(rootThreadId); if (roots?.size === 0) this.projects.delete(projectId); return; } - const roots = this.projects.get(projectId) ?? new Map(); roots.set(rootThreadId, this.now()); this.projects.set(projectId, roots); + const roots = this.projects.get(projectId) ?? new Set(); roots.add(rootThreadId); this.projects.set(projectId, roots); } - /** Track an open global page only while it has an active Redteam scan. */ + /** Track a watched global page only while it has an active Redteam scan. */ observeGlobal(active: boolean): void { - if (this.disposed) return; - this.globalTouchedAt = this.now(); + if (this.disposed || this.globalWatchers === 0) return; this.globalActive = active; } /** One bounded polling pass; public as a deterministic test seam. */ async tick(): Promise { - if (this.disposed) return; const now = this.now(); const tasks: Promise[] = []; + if (this.disposed) return; const tasks: Promise[] = []; for (const [projectId, roots] of this.projects) { - for (const [rootId, touchedAt] of roots) if (now - touchedAt > this.watchedTtlMs) roots.delete(rootId); if (roots.size === 0) { this.projects.delete(projectId); continue; } if (this.inflight.has(projectId)) continue; this.inflight.add(projectId); @@ -55,30 +73,21 @@ export class RedteamActivePoller { }).catch(() => { /* optional source failure: retry only while previously active */ }).finally(() => this.inflight.delete(projectId)); tasks.push(task); } - if ( - this.globalTouchedAt !== null && - now - this.globalTouchedAt <= this.watchedTtlMs && - this.globalActive && - !this.globalInflight && - this.deps.pollGlobal - ) { + if (this.globalWatchers > 0 && this.globalActive && !this.globalInflight && this.deps.pollGlobal) { this.globalInflight = true; const task = this.deps.pollGlobal().then((result) => { if (result.changed) this.deps.changedGlobal?.(); - this.globalActive = result.active; + if (this.globalWatchers > 0) this.globalActive = result.active; }).catch(() => { // Optional source failures never affect BB work; the page's slow // discovery refresh will reconcile availability. }).finally(() => { this.globalInflight = false; }); tasks.push(task); - } else if (this.globalTouchedAt !== null && now - this.globalTouchedAt > this.watchedTtlMs) { - this.globalTouchedAt = null; - this.globalActive = false; } await Promise.all(tasks); } isPolling(projectId: string): boolean { return this.projects.has(projectId); } - isGlobalPolling(): boolean { return this.globalActive && this.globalTouchedAt !== null; } - dispose(): void { this.disposed = true; if (this.timer !== null) clearInterval(this.timer); this.timer = null; this.projects.clear(); this.inflight.clear(); this.globalTouchedAt = null; this.globalActive = false; this.globalInflight = false; } + isGlobalPolling(): boolean { return this.globalActive && this.globalWatchers > 0; } + dispose(): void { this.disposed = true; if (this.timer !== null) clearInterval(this.timer); this.timer = null; this.projects.clear(); this.watchedRoots.clear(); this.inflight.clear(); this.globalWatchers = 0; this.globalActive = false; this.globalInflight = false; } } diff --git a/lib/watch.ts b/lib/watch.ts index bafab03..10972ff 100644 --- a/lib/watch.ts +++ b/lib/watch.ts @@ -137,19 +137,17 @@ export interface WatchedRoot { threads: Set; /** Project filter for parentage lookups of unseen threads. */ projectId: string | null; - /** Last time a snapshot was requested for this root. */ - lastTouchedAt: number; + /** Open board clients currently watching this root. */ + watchers: number; } export interface WatcherDeps { /** Publishes the invalidation signal. */ publish: (payload: { roots: string[]; global?: boolean }) => void; - /** Resolves a thread's parentThreadId (bounded parentage check). */ - getParent: (threadId: string) => Promise; + /** Resolves a thread's ancestor ids, parent first (bounded parentage check). */ + getAncestors: (threadId: string) => Promise; /** Clock, injectable for tests. Defaults to Date.now. */ now?: () => number; - /** How long a watched root stays live without a snapshot request. */ - watchedTtlMs?: number; /** Debounce window for coalescing signals. */ debounceMs?: number; /** Min gap between parentage lookups for the same unseen thread. */ @@ -174,7 +172,7 @@ export class BoardWatcher { private readonly watched = new Map(); private readonly dirty = new Set(); private dirtyGlobal = false; - private globalTouchedAt: number | null = null; + private globalWatchers = 0; private readonly lookupAt = new Map(); private timer: ReturnType | null = null; /** Pending terminal reconciliation: at most one, replaced by newer terminal flushes. */ @@ -188,13 +186,11 @@ export class BoardWatcher { private lastEventAt: number | null = null; private lastPublishedAt: number | null = null; private lastErrorAt: number | null = null; - private readonly watchedTtlMs: number; private readonly debounceMs: number; private readonly lookupGapMs: number; private readonly reconcileDelayMs: number; constructor(private readonly deps: WatcherDeps) { - this.watchedTtlMs = deps.watchedTtlMs ?? 90_000; this.debounceMs = deps.debounceMs ?? 400; this.lookupGapMs = deps.lookupGapMs ?? 2_000; this.reconcileDelayMs = deps.reconcileDelayMs ?? 1_500; @@ -216,42 +212,59 @@ export class BoardWatcher { catch { this.recordError(); return false; } } - /** Called by the RPC handler whenever a board is fetched for a root. */ - touch(rootThreadId: string, threads: string[], projectId: string | null): void { - if (this.disposed) return; - const now = this.clock(); + /** + * Watch a root while an open board reads it; the read's watched signal + * ends the watch when its last client leaves. + */ + watch(rootThreadId: string, signal: AbortSignal): void { + if (this.disposed || signal.aborted) return; const existing = this.watched.get(rootThreadId); - if (existing) { - existing.threads = new Set(threads); - existing.projectId = projectId; - existing.lastTouchedAt = now; - } else { + if (existing) existing.watchers += 1; + else { this.watched.set(rootThreadId, { rootThreadId, - threads: new Set(threads), - projectId, - lastTouchedAt: now, + threads: new Set([rootThreadId]), + projectId: null, + watchers: 1, }); } - this.prune(now); + signal.addEventListener("abort", () => { + const root = this.watched.get(rootThreadId); + if (!root) return; + root.watchers -= 1; + if (root.watchers <= 0) this.watched.delete(rootThreadId); + }, { once: true }); } - /** Lease the global dashboard as watched; refreshed by its bounded RPC. */ - touchGlobal(): void { + /** Watch the global dashboard while an open page reads it. */ + watchGlobal(signal: AbortSignal): void { + if (this.disposed || signal.aborted) return; + this.globalWatchers += 1; + signal.addEventListener("abort", () => { + this.globalWatchers = Math.max(0, this.globalWatchers - 1); + }, { once: true }); + } + + /** Called by the RPC handler with the threads a watched root's snapshot observed. */ + touch(rootThreadId: string, threads: string[], projectId: string | null): void { if (this.disposed) return; - this.globalTouchedAt = this.clock(); - this.prune(this.globalTouchedAt); + const root = this.watched.get(rootThreadId); + if (!root) return; + root.threads = new Set(threads); + root.projectId = projectId; } - private prune(now: number): void { - for (const [id, root] of this.watched) { - if (now - root.lastTouchedAt > this.watchedTtlMs) this.watched.delete(id); - } + isWatching(rootThreadId: string): boolean { + return this.watched.has(rootThreadId); + } + + isWatchingGlobal(): boolean { + return this.globalWatchers > 0; + } + + private pruneLookups(now: number): void { for (const [id, at] of this.lookupAt) { - if (now - at > this.watchedTtlMs) this.lookupAt.delete(id); - } - if (this.globalTouchedAt !== null && now - this.globalTouchedAt > this.watchedTtlMs) { - this.globalTouchedAt = null; + if (now - at >= this.lookupGapMs) this.lookupAt.delete(id); } } @@ -291,19 +304,7 @@ export class BoardWatcher { for (const root of this.watched.values()) this.markDirty(root.rootThreadId); return; } - const projectId = event.metadata?.projectId ?? null; - for (const root of this.candidatesFor(event.id, projectId)) { - if (root.threads.has(event.id) || event.id === root.rootThreadId) { - this.markDirty(root.rootThreadId); - continue; - } - if (this.lookupRecently(event.id)) continue; - const parent = await this.deps.getParent(event.id); - if (parent !== null && (root.threads.has(parent) || parent === root.rootThreadId)) { - root.threads.add(event.id); // a new descendant appeared - this.markDirty(root.rootThreadId); - } - } + await this.markAffectedRoots(event.id, event.metadata?.projectId ?? null, undefined); } /** Handle a plugin thread lifecycle event (thread.created/failed/idle/…). */ @@ -319,31 +320,46 @@ export class BoardWatcher { this.lastEventAt = this.clock(); if (isTerminalLifecycleKind(kind)) this.dirtyTerminal = true; this.markGlobalDirty(); - for (const root of this.candidatesFor(thread.id, thread.projectId ?? null)) { - if (root.threads.has(thread.id) || thread.id === root.rootThreadId) { + await this.markAffectedRoots(thread.id, thread.projectId ?? null, thread.parentThreadId); + } + + /** + * Mark every watched root containing `threadId`. Known threads and children + * of known threads match directly; any other candidate needs one ancestor + * lookup, so a thread created below an unseen parent still reaches its + * root. A known null parent means a root thread, which needs no lookup. + */ + private async markAffectedRoots( + threadId: string, + projectId: string | null, + parentThreadId: string | null | undefined, + ): Promise { + const unresolved: WatchedRoot[] = []; + for (const root of this.candidatesFor(threadId, projectId)) { + if (root.threads.has(threadId) || threadId === root.rootThreadId) { this.markDirty(root.rootThreadId); - continue; - } - const parent = thread.parentThreadId ?? null; - if (parent !== null) { - if (root.threads.has(parent) || parent === root.rootThreadId) { - root.threads.add(thread.id); - this.markDirty(root.rootThreadId); - } - continue; + } else if (parentThreadId != null && (root.threads.has(parentThreadId) || parentThreadId === root.rootThreadId)) { + root.threads.add(threadId); + this.markDirty(root.rootThreadId); + } else if (parentThreadId !== null) { + unresolved.push(root); } - if (this.lookupRecently(thread.id)) continue; - const resolved = await this.deps.getParent(thread.id); - if (resolved !== null && (root.threads.has(resolved) || resolved === root.rootThreadId)) { - root.threads.add(thread.id); + } + if (unresolved.length === 0 || this.lookupRecently(threadId)) return; + const ancestors = await this.deps.getAncestors(threadId); + if (this.disposed) return; + for (const root of unresolved) { + if (ancestors.some((id) => id === root.rootThreadId || root.threads.has(id))) { + root.threads.add(threadId); this.markDirty(root.rootThreadId); } } } private lookupRecently(threadId: string): boolean { - const at = this.lookupAt.get(threadId); const now = this.clock(); + this.pruneLookups(now); + const at = this.lookupAt.get(threadId); if (at !== undefined && now - at < this.lookupGapMs) return true; this.lookupAt.set(threadId, now); return false; @@ -355,9 +371,7 @@ export class BoardWatcher { } private markGlobalDirty(): void { - const now = this.clock(); - this.prune(now); - if (this.globalTouchedAt === null) return; + if (this.globalWatchers === 0) return; this.dirtyGlobal = true; this.ensureTimer(); } @@ -376,7 +390,6 @@ export class BoardWatcher { /** Optional-source invalidation for every currently watched root in a project. */ invalidateProject(projectId: string): void { if (this.disposed) return; - const now = this.clock(); this.prune(now); for (const root of this.watched.values()) if (root.projectId === projectId) this.markDirty(root.rootThreadId); this.markGlobalDirty(); } @@ -437,7 +450,7 @@ export class BoardWatcher { this.dirty.clear(); this.dirtyGlobal = false; this.dirtyTerminal = false; - this.globalTouchedAt = null; + this.globalWatchers = 0; } /** Realtime channel the frontend listens on. */ diff --git a/package-lock.json b/package-lock.json index e24972a..b649b39 100644 --- a/package-lock.json +++ b/package-lock.json @@ -7,6 +7,7 @@ "": { "name": "bb-plugin-agent-board", "version": "0.7.0", + "license": "MIT", "dependencies": { "@hugeicons/core-free-icons": "^4.1.3", "@hugeicons/react": "^1.1.6", @@ -15,7 +16,7 @@ "zod": "^4.3.6" }, "devDependencies": { - "@get-bb/plugin-sdk": "0.4.34", + "@get-bb/plugin-sdk": "0.6.132", "@pierre/diffs": "^1.2.9", "@radix-ui/react-alert-dialog": "^1.1.19", "@radix-ui/react-context-menu": "^2.3.3", @@ -45,10 +46,9 @@ "vitest": "^5.0.0" }, "engines": { - "bb": ">=0.40", - "bbPluginSdk": ">=0.4.21" - }, - "license": "MIT" + "bb": ">=0.47", + "bbPluginSdk": ">=0.6.73" + } }, "node_modules/@asamuzakjp/css-color": { "version": "6.0.7", @@ -334,10 +334,15 @@ "license": "MIT" }, "node_modules/@get-bb/plugin-sdk": { - "version": "0.4.34", - "resolved": "https://registry.npmjs.org/@get-bb/plugin-sdk/-/plugin-sdk-0.4.34.tgz", - "integrity": "sha512-2ugZuMXdASpmRUrGPaj09sk0t21xDJTudqzT0ucFW8pYY8PMyVmy7RCTRAsNo1oCicvOvD97rw8C1v3k5hYqFA==", + "version": "0.6.132", + "resolved": "https://registry.npmjs.org/@get-bb/plugin-sdk/-/plugin-sdk-0.6.132.tgz", + "integrity": "sha512-gxRXAikswX3+gKcLODVz8uIl8fPU2M5KYApZOjZ6VX38o3nmP0Iv27FJE7AGybvK6DQSc9USBYmXIVBaEEIgcw==", "dev": true, + "dependencies": { + "@tanstack/react-query": "^5.106.1", + "cross-spawn": "^7.0.6", + "zod": "^4.6.7" + }, "peerDependencies": { "@testing-library/react": "^16.3.2", "@types/better-sqlite3": "^7.6.12", @@ -346,8 +351,7 @@ "cron-parser": "^5.5.0", "hono": "^4.11.9", "react": "^19.0.0", - "react-dom": "^19.0.0", - "zod": "^4.3.6" + "react-dom": "^19.0.0" }, "peerDependenciesMeta": { "@testing-library/react": { @@ -373,9 +377,6 @@ }, "react-dom": { "optional": true - }, - "zod": { - "optional": true } } }, @@ -1549,9 +1550,6 @@ "arm64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -1570,9 +1568,6 @@ "arm64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -1591,9 +1586,6 @@ "ppc64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -1612,9 +1604,6 @@ "s390x" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -1633,9 +1622,6 @@ "x64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MIT", "optional": true, "os": [ @@ -1654,9 +1640,6 @@ "x64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MIT", "optional": true, "os": [ @@ -1851,6 +1834,40 @@ "dev": true, "license": "MIT" }, + "node_modules/@tanstack/query-core": { + "version": "5.107.1", + "resolved": "https://registry.npmjs.org/@tanstack/query-core/-/query-core-5.107.1.tgz", + "integrity": "sha512-PSyAik11oQBEVOiVy5hjZWHmoUqOXqK1WYy4v55mcxozEfKaQgpUMsgnXhWjuESzSht8JfzqMS1m6M/9CpNgvg==", + "dev": true, + "license": "MIT", + "funding": { + "type": "github", + "url": "https://github.com/sponsors/tannerlinsley" + } + }, + "node_modules/@tanstack/react-query": { + "version": "5.107.1", + "resolved": "https://registry.npmjs.org/@tanstack/react-query/-/react-query-5.107.1.tgz", + "integrity": "sha512-udCOOLp3ZAheUrIHLCe+ZZge6nr2ZUrGMSWcCuXsjY1Bf2JeeE09oxst97PILiCY6N8xo/7bIo3Hvz13zh06FA==", + "dev": true, + "license": "MIT", + "dependencies": { + "@tanstack/query-core": "5.107.1" + }, + "funding": { + "type": "github", + "url": "https://github.com/sponsors/tannerlinsley" + }, + "peerDependencies": { + "@types/react": "^18 || ^19", + "react": "^18 || ^19" + }, + "peerDependenciesMeta": { + "@types/react": { + "optional": true + } + } + }, "node_modules/@testing-library/dom": { "version": "10.4.1", "resolved": "https://registry.npmjs.org/@testing-library/dom/-/dom-10.4.1.tgz", @@ -2294,6 +2311,21 @@ "node": ">=18" } }, + "node_modules/cross-spawn": { + "version": "7.0.6", + "resolved": "https://registry.npmjs.org/cross-spawn/-/cross-spawn-7.0.6.tgz", + "integrity": "sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==", + "dev": true, + "license": "MIT", + "dependencies": { + "path-key": "^3.1.0", + "shebang-command": "^2.0.0", + "which": "^2.0.1" + }, + "engines": { + "node": ">= 8" + } + }, "node_modules/css-tree": { "version": "3.2.1", "resolved": "https://registry.npmjs.org/css-tree/-/css-tree-3.2.1.tgz", @@ -2675,6 +2707,13 @@ "dev": true, "license": "MIT" }, + "node_modules/isexe": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/isexe/-/isexe-2.0.0.tgz", + "integrity": "sha512-RHxMLp9lnKHGHRng9QFhRCMbYAcVpn69smSGcq3f36xjgVVWThj4qqLbTLlq7Ssj8B+fIQ1EuCEGI2lKsyQeIw==", + "dev": true, + "license": "ISC" + }, "node_modules/js-tokens": { "version": "4.0.0", "resolved": "https://registry.npmjs.org/js-tokens/-/js-tokens-4.0.0.tgz", @@ -2873,9 +2912,6 @@ "arm64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MPL-2.0", "optional": true, "os": [ @@ -2898,9 +2934,6 @@ "arm64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MPL-2.0", "optional": true, "os": [ @@ -2923,9 +2956,6 @@ "x64" ], "dev": true, - "libc": [ - "glibc" - ], "license": "MPL-2.0", "optional": true, "os": [ @@ -2948,9 +2978,6 @@ "x64" ], "dev": true, - "libc": [ - "musl" - ], "license": "MPL-2.0", "optional": true, "os": [ @@ -3306,6 +3333,16 @@ "url": "https://github.com/inikulin/parse5?sponsor=1" } }, + "node_modules/path-key": { + "version": "3.1.1", + "resolved": "https://registry.npmjs.org/path-key/-/path-key-3.1.1.tgz", + "integrity": "sha512-ojmeN0qd+y0jszEtoY48r0Peq5dwMEkIlCOu6Q5f41lfkswXuKtYrhgoTpLnyIcHm24Uhqx+5Tqm2InSwLhE6Q==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=8" + } + }, "node_modules/picocolors": { "version": "1.1.1", "resolved": "https://registry.npmjs.org/picocolors/-/picocolors-1.1.1.tgz", @@ -3693,6 +3730,29 @@ "node": ">=10" } }, + "node_modules/shebang-command": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/shebang-command/-/shebang-command-2.0.0.tgz", + "integrity": "sha512-kHxr2zZpYtdmrN1qDjrrX/Z1rR1kG8Dx+gkpK1G4eXmvXswmcE1hTWBWYUzlraYw1/yZp6YuDY77YtvbN0dmDA==", + "dev": true, + "license": "MIT", + "dependencies": { + "shebang-regex": "^3.0.0" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/shebang-regex": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/shebang-regex/-/shebang-regex-3.0.0.tgz", + "integrity": "sha512-7++dFhtcx3353uBaq8DDR4NuxBetBzC7ZQOhmTQInHEd6bSrXdiEyzCvG07Z44UYdLShWUyXt5M/yhz8ekcb1A==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=8" + } + }, "node_modules/shiki": { "version": "4.4.3", "resolved": "https://registry.npmjs.org/shiki/-/shiki-4.4.3.tgz", @@ -4420,6 +4480,22 @@ "node": "^22.14.0 || >=24.0.0" } }, + "node_modules/which": { + "version": "2.0.2", + "resolved": "https://registry.npmjs.org/which/-/which-2.0.2.tgz", + "integrity": "sha512-BLI3Tl1TW3Pvl70l3yq3Y64i+awpwXqsGBYWkkqMtnbXgrMD+yj7rhW0kuEDxzJaYXGjEW5ogapKNMEKNMjibA==", + "dev": true, + "license": "ISC", + "dependencies": { + "isexe": "^2.0.0" + }, + "bin": { + "node-which": "bin/node-which" + }, + "engines": { + "node": ">= 8" + } + }, "node_modules/why-is-node-running": { "version": "2.3.0", "resolved": "https://registry.npmjs.org/why-is-node-running/-/why-is-node-running-2.3.0.tgz", @@ -4462,9 +4538,9 @@ "license": "MIT" }, "node_modules/zod": { - "version": "4.5.4", - "resolved": "https://registry.npmjs.org/zod/-/zod-4.5.4.tgz", - "integrity": "sha512-sC95tT5iHHH9gtpj6A81kh+NEaRAUFN+qlUPDUbRfOMvNf5QCBqsb3WgvnpVtK5Y+4UfA6KqufotuTvMGiTlsA==", + "version": "4.6.7", + "resolved": "https://registry.npmjs.org/zod/-/zod-4.6.7.tgz", + "integrity": "sha512-g8W2D2kU6a8untc053fX1uVBKzM7yqAxW/3zwk83+6RyF2+SKOuV+c0wBDhUwzV1PaGapmZH+rmcUzPf+3CGwg==", "license": "MIT", "funding": { "url": "https://github.com/sponsors/colinhacks" diff --git a/package.json b/package.json index 09991ce..35c1530 100644 --- a/package.json +++ b/package.json @@ -26,8 +26,8 @@ "verify:package": "node scripts/verify-package.mjs" }, "engines": { - "bb": ">=0.40", - "bbPluginSdk": ">=0.4.21" + "bb": ">=0.47", + "bbPluginSdk": ">=0.6.73" }, "bb": { "name": "Agent Board", @@ -46,7 +46,7 @@ "zod": "^4.3.6" }, "devDependencies": { - "@get-bb/plugin-sdk": "0.4.34", + "@get-bb/plugin-sdk": "0.6.132", "@pierre/diffs": "^1.2.9", "@radix-ui/react-alert-dialog": "^1.1.19", "@radix-ui/react-context-menu": "^2.3.3", diff --git a/server.ts b/server.ts index c77f1a1..5fa7fd0 100644 --- a/server.ts +++ b/server.ts @@ -5,8 +5,8 @@ // bb.sdk (bounded), normalizes what BB reports into BoardCards, and returns // a snapshot. Sources remain authoritative; only attention/release dismissal // preferences are durable. Collection and watcher state are ephemeral caches. -import { defineRpcContract, type BbPluginApi } from "@get-bb/plugin-sdk"; -import { rpcContract } from "./contract/rpc"; +import type { BbPluginApi } from "@get-bb/plugin-sdk"; +import { boardChangedTags, GLOBAL_DASHBOARD_TAG, rpcContract } from "./contract/rpc"; import { BoardCollector } from "./lib/collect"; import { attachWatcher, BoardWatcher } from "./lib/watch"; import { FleetActivePoller } from "./lib/fleet-poll"; @@ -27,16 +27,10 @@ export default async function plugin(bb: BbPluginApi) { const watcher = new BoardWatcher({ publish: (payload) => { collector.invalidate(); + bb.rpc.experimental_invalidate(boardChangedTags(payload)); bb.realtime.publish(BoardWatcher.channel, payload); }, - getParent: async (threadId) => { - try { - const thread = await bb.sdk.threads.get({ threadId }); - return thread.parentThreadId ?? null; - } catch { - return null; // deleted or not visible to this plugin - } - }, + getAncestors: createAncestorLoader(bb), }); const redteamPoller = new RedteamActivePoller({ poll: (projectId) => collector.pollRedteam(projectId), @@ -56,10 +50,29 @@ export default async function plugin(bb: BbPluginApi) { changed: () => { collector.invalidate(false); // Empty roots without global-only marks all scoped and global views stale. - bb.realtime.publish(BoardWatcher.channel, { roots: [] }); + const payload = { roots: [] }; + bb.rpc.experimental_invalidate(boardChangedTags(payload)); + bb.realtime.publish(BoardWatcher.channel, payload); }, }); + /** Feed the watcher and pollers what one board snapshot observed. */ + const observeBoard = (threadId: string, snapshot: Awaited>) => { + // The observed tree lets change signals be attributed to open boards + // (ephemeral, rebuilt every fetch). + const observed = [ + threadId, + ...[...new Set(snapshot.cards.map((card) => card.threadId).filter((id) => id !== null))], + ]; + watcher.touch(threadId, observed, snapshot.root.projectId); + redteamPoller.observe( + threadId, + snapshot.root.projectId, + collector.isRedteamActive(snapshot.root.projectId), + ); + fleetPoller.observe(collector.isFleetActive()); + }; + // BB threads/workflows/plans/delegations plus optional adapter cards share // one source-discriminated normalized contract. bb.rpc.register(rpcContract, { @@ -67,19 +80,7 @@ export default async function plugin(bb: BbPluginApi) { dismiss_release_announcement: ({ version }) => releases.dismiss(version), board_snapshot: async ({ threadId, fresh }) => { const snapshot = await collector.snapshot(threadId, { fresh }); - // Feed the watcher the observed tree so change signals can be - // attributed to open boards (ephemeral, rebuilt every fetch). - const observed = [ - threadId, - ...[...new Set(snapshot.cards.map((card) => card.threadId).filter((id) => id !== null))], - ]; - watcher.touch(threadId, observed, snapshot.root.projectId); - redteamPoller.observe( - threadId, - snapshot.root.projectId, - collector.isRedteamActive(snapshot.root.projectId), - ); - fleetPoller.observe(collector.isFleetActive()); + observeBoard(threadId, snapshot); return snapshot; }, board_activity: ({ threadId, limit }) => collector.activity(threadId, limit), @@ -91,7 +92,6 @@ export default async function plugin(bb: BbPluginApi) { global_dashboard: async ({ fresh }) => { const dismissedKeys = await dismissals.keys(); const snapshot = await collector.globalDashboard({ fresh, dismissedKeys }); - watcher.touchGlobal(); redteamPoller.observeGlobal(collector.isGlobalRedteamActive()); fleetPoller.observe(collector.isFleetActive()); return { ...snapshot, monitoring: watcher.health() }; @@ -102,15 +102,47 @@ export default async function plugin(bb: BbPluginApi) { ); // Only the global snapshot changed; per-thread caches stay warm. collector.invalidateGlobal(); + bb.rpc.experimental_invalidate([GLOBAL_DASHBOARD_TAG]); return { dismissed }; }, clear_dismissed_attentions: async () => { await dismissals.clear(); collector.invalidateGlobal(); + bb.rpc.experimental_invalidate([GLOBAL_DASHBOARD_TAG]); return { dismissed: 0 }; }, }); + // A board or global page counts as open while a client watches its read: + // the host aborts the signal when the last client leaves or hides the tab. + bb.rpc.experimental_whileWatched("board_snapshot", async (read, signal) => { + if (!("input" in read)) return; + const input = rpcContract.board_snapshot.input.safeParse(read.input); + if (!input.success) return; + const { threadId } = input.data; + watcher.watch(threadId, signal); + redteamPoller.watch(threadId, signal); + fleetPoller.watch(signal); + try { + const snapshot = await collector.snapshot(threadId); + if (!signal.aborted) observeBoard(threadId, snapshot); + } catch { + // The read itself reports the failure; the next successful read observes. + } + }); + bb.rpc.experimental_whileWatched("global_dashboard", (_read, signal) => { + watcher.watchGlobal(signal); + redteamPoller.watchGlobal(signal); + fleetPoller.watch(signal); + redteamPoller.observeGlobal(collector.isGlobalRedteamActive()); + fleetPoller.observe(collector.isFleetActive()); + // BB lifecycle events are immediate. This slow check exists only to + // discover a new Redteam run, whose changes cannot be subscribed to + // cross-plugin. + const discovery = setInterval(() => bb.rpc.experimental_invalidate([GLOBAL_DASHBOARD_TAG]), GLOBAL_DISCOVERY_INTERVAL_MS); + signal.addEventListener("abort", () => clearInterval(discovery), { once: true }); + }); + // Invalidate open boards on real BB changes (coalesced, no polling). const unsubscribe = attachWatcher(bb, watcher); @@ -161,6 +193,49 @@ export default async function plugin(bb: BbPluginApi) { }); } +const GLOBAL_DISCOVERY_INTERVAL_MS = 20_000; +const ANCESTOR_BATCH_LIMIT = 200; + +/** + * Batch ancestor lookups made in the same tick into one + * `experimental_listAncestors` call (200 ids per request). A failed or + * omitted lookup resolves to no ancestors. + */ +function createAncestorLoader(bb: BbPluginApi): (threadId: string) => Promise { + let pending = new Map void>>(); + let scheduled = false; + const flush = async () => { + scheduled = false; + const batch = pending; + pending = new Map(); + const ids = [...batch.keys()]; + const ancestorsById = new Map(); + for (let start = 0; start < ids.length; start += ANCESTOR_BATCH_LIMIT) { + try { + const result = await bb.sdk.threads.experimental_listAncestors({ + threadIds: ids.slice(start, start + ANCESTOR_BATCH_LIMIT), + }); + for (const entry of result.threads) ancestorsById.set(entry.threadId, entry.ancestorIds); + } catch { + // Unresolvable threads are treated as outside every watched tree. + } + } + for (const [id, resolvers] of batch) { + for (const resolve of resolvers) resolve(ancestorsById.get(id) ?? []); + } + }; + return (threadId) => + new Promise((resolve) => { + const resolvers = pending.get(threadId); + if (resolvers) resolvers.push(resolve); + else pending.set(threadId, [resolve]); + if (!scheduled) { + scheduled = true; + setTimeout(() => void flush(), 0); + } + }); +} + /** Compact text rendering of a snapshot for CLI use. */ function renderText( snapshot: Awaited>, diff --git a/tests/app.test.tsx b/tests/app.test.tsx index f98c33c..d4a9fc1 100644 --- a/tests/app.test.tsx +++ b/tests/app.test.tsx @@ -1,11 +1,12 @@ // @vitest-environment jsdom -// Frontend tests: render the threadPanelAction slot through the official -// plugin app testing harness — columns/cards, realtime signal refetch, +// Frontend tests: render the thread action tab through the official +// plugin app testing harness — columns/cards, tag invalidation refetch, // reconnect reconcile, navigation, loading/error/empty states. import { describe, expect, it, beforeEach } from "vitest"; +import { waitFor } from "@testing-library/react"; import { loadPluginApp, renderSlot, type CapturedPluginApp, type RenderedSlot } from "@get-bb/plugin-sdk/testing/app"; +import { boardTabComponent } from "./board-tab"; import type { BoardCard, BoardSnapshot, ExecutionTimeline } from "@/contract/rpc"; -import { BOARD_CHANGED_CHANNEL } from "@/contract/rpc"; let app: CapturedPluginApp; @@ -135,9 +136,8 @@ function timelineFixture(overrides: Partial = {}): ExecutionT function mountBoard( options: Parameters[2] = {}, ): RenderedSlot { - const registration = app.threadPanelActions[0]!; return renderSlot( - { component: registration.component }, + { component: boardTabComponent(app) }, { threadId: "thr_root", params: null } as never, options, ); @@ -168,11 +168,16 @@ function kanbanSnapshot(cards: BoardCard[]): BoardSnapshot { } describe("slot registration", () => { - it("registers exactly one thread panel action", () => { - expect(app.threadPanelActions).toHaveLength(1); - expect(app.threadPanelActions[0]!.id).toBe("agent-board"); - expect(app.threadPanelActions[0]!.title).toBe("Agent Board"); - expect(app.threadPanelActions[0]!.icon).toBe("ListTodo"); + it("registers exactly one thread action tab", () => { + const threadTabs = app.tabs.filter((tab) => tab.context.kind === "thread"); + expect(threadTabs).toHaveLength(1); + expect(threadTabs[0]).toMatchObject({ + id: "agent-board", + availability: "action", + title: "Agent Board", + icon: "ListTodo", + layout: "flush", + }); }); }); @@ -242,66 +247,57 @@ describe("board rendering", () => { }); }); -describe("realtime invalidation", () => { - it("refetches when a signal names this board", async () => { +describe("board invalidation", () => { + it("refetches when this board's tag is invalidated", async () => { const slot = mountBoard({ rpc: { board_snapshot: () => snapshotFixture() }, }); await slot.findAllByText("Refactor the auth flow"); expect(slot.inspection.rpcCalls.length).toBe(1); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: ["thr_root"] }); - await slot.findAllByText("Refactor the auth flow"); - expect(slot.inspection.rpcCalls.length).toBe(2); + await slot.behavior.experimental_invalidateRpc(["board:thr_root"]); + await waitFor(() => expect(slot.inspection.rpcCalls.length).toBe(2)); slot.unmount(); }); - it("ignores signals scoped to other roots", async () => { - const slot = mountBoard({ - rpc: { board_snapshot: () => snapshotFixture() }, - }); + it("refetches when every board is invalidated", async () => { + const slot = mountBoard({ rpc: { board_snapshot: () => snapshotFixture() } }); await slot.findAllByText("Refactor the auth flow"); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: ["thr_other"] }); - await new Promise((resolve) => setTimeout(resolve, 50)); - expect(slot.inspection.rpcCalls.length).toBe(1); - - slot.unmount(); - }); + await slot.behavior.experimental_invalidateRpc(["board"]); + await waitFor(() => expect(slot.inspection.rpcCalls.length).toBe(2)); - it("ignores global-only invalidations", async () => { - const slot = mountBoard({ rpc: { board_snapshot: () => snapshotFixture() } }); - await slot.findAllByText("Refactor the auth flow"); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: [], global: true }); - await new Promise((resolve) => setTimeout(resolve, 25)); - expect(slot.inspection.rpcCalls.length).toBe(1); slot.unmount(); }); - it("refetches defensively on a malformed signal", async () => { + it("ignores invalidations of other roots and the global page", async () => { const slot = mountBoard({ rpc: { board_snapshot: () => snapshotFixture() }, }); await slot.findAllByText("Refactor the auth flow"); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { nonsense: true }); - expect(slot.inspection.rpcCalls.length).toBe(2); + await slot.behavior.experimental_invalidateRpc(["board:thr_other"]); + await slot.behavior.experimental_invalidateRpc(["global-dashboard"]); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(slot.inspection.rpcCalls.length).toBe(1); slot.unmount(); }); - it("reconciles after a realtime reconnect", async () => { + it("refreshes with fresh: true, then serves that snapshot to the board read", async () => { const slot = mountBoard({ rpc: { board_snapshot: () => snapshotFixture() }, - realtimeConnectionState: "connected", }); await slot.findAllByText("Refactor the auth flow"); - expect(slot.inspection.rpcCalls.length).toBe(1); - await slot.behavior.setRealtimeConnectionState("reconnecting"); - await slot.behavior.setRealtimeConnectionState("connected"); - expect(slot.inspection.rpcCalls.length).toBe(2); + slot.getByRole("button", { name: /refresh/i }).click(); + await waitFor(() => expect(slot.inspection.rpcCalls.length).toBe(3)); + expect(slot.inspection.rpcCalls.map((call) => call.input)).toEqual([ + { threadId: "thr_root" }, + { threadId: "thr_root", fresh: true }, + { threadId: "thr_root" }, + ]); slot.unmount(); }); @@ -668,10 +664,11 @@ describe("single-agent focus mode", () => { await slot.findByTestId("focus-hero"); const hero = slot.getByTestId("focus-hero"); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: ["thr_root"] }); + await slot.behavior.experimental_invalidateRpc(["board:thr_root"]); // The outgoing and incoming lines coexist only during the one-shot swap. const activity = slot.getByTestId("hero-activity"); + await waitFor(() => expect(activity.textContent).toContain("Writing report.md")); expect(activity.textContent).toContain("Reading src/dispatcher.js"); expect(activity.textContent).toContain("Writing report.md"); // The hero itself never remounts for a text change. @@ -696,7 +693,7 @@ describe("single-agent focus mode", () => { }); await slot.findByTestId("single-agent-focus"); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: ["thr_root"] }); + await slot.behavior.experimental_invalidateRpc(["board:thr_root"]); expect(await slot.findByText("ACTIVE AGENTS")).toBeTruthy(); expect(slot.queryByTestId("single-agent-focus")).toBeNull(); @@ -722,7 +719,7 @@ describe("single-agent focus mode", () => { await slot.findByText("ACTIVE AGENTS"); expect(slot.getAllByTestId("kanban-live-strip")).toHaveLength(2); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: ["thr_root"] }); + await slot.behavior.experimental_invalidateRpc(["board:thr_root"]); const focus = await slot.findByTestId("single-agent-focus"); expect(slot.queryByText("ACTIVE AGENTS")).toBeNull(); diff --git a/tests/bb-source.test.ts b/tests/bb-source.test.ts index 6b8fb7d..ea77941 100644 --- a/tests/bb-source.test.ts +++ b/tests/bb-source.test.ts @@ -8,6 +8,7 @@ import { normalizeThread, recentActivityFeed, threadBoardStatus, + timelineGoal, waitingOnLabel, workKindLabel, type NormalizeContext, @@ -915,3 +916,24 @@ describe("latestPublicUpdate", () => { expect(latestPublicUpdate(rows)).toBeNull(); }); }); + +describe("timelineGoal", () => { + const goal = { objective: "Ship it", status: "usageLimited", tokensUsed: 10, tokenBudget: null, timeUsedSeconds: 5 }; + const base = { rows: [], extensionStates: {} } as unknown as TimelineResponse; + + it("reads the Codex goal from the timeline's extension states", () => { + const timeline = { ...base, extensionStates: { "provider-codex/goal": goal } } as unknown as TimelineResponse; + expect(timelineGoal(timeline)).toEqual(goal); + }); + + it("falls back to the goal field hosts before BB 0.49 send", () => { + const timeline = { rows: [], goal: { ...goal, status: "active", tokenBudget: undefined } } as unknown as TimelineResponse; + expect(timelineGoal(timeline)).toEqual({ ...goal, status: "active", tokenBudget: null }); + }); + + it("returns null for a cleared or unreadable goal", () => { + expect(timelineGoal({ ...base, extensionStates: { "provider-codex/goal": null } } as unknown as TimelineResponse)).toBeNull(); + expect(timelineGoal({ ...base, extensionStates: { "provider-codex/goal": { objective: 1 } } } as unknown as TimelineResponse)).toBeNull(); + expect(timelineGoal(base)).toBeNull(); + }); +}); diff --git a/tests/board-tab.ts b/tests/board-tab.ts new file mode 100644 index 0000000..0f74ced --- /dev/null +++ b/tests/board-tab.ts @@ -0,0 +1,12 @@ +// The Agent Board thread action tab's component, typed with the props the +// host passes a thread action tab. +import type { ComponentType } from "react"; +import type { ExperimentalThreadActionTabProps } from "@get-bb/plugin-sdk"; + +export function boardTabComponent(app: { + tabs: ReadonlyArray<{ id: string; component: unknown }>; +}): ComponentType { + const tab = app.tabs.find((candidate) => candidate.id === "agent-board"); + if (!tab) throw new Error("Agent Board tab is not registered"); + return tab.component as ComponentType; +} diff --git a/tests/collect.test.ts b/tests/collect.test.ts index a24acc5..a287349 100644 --- a/tests/collect.test.ts +++ b/tests/collect.test.ts @@ -5,7 +5,8 @@ import { beforeEach, describe, expect, it } from "vitest"; import { createFakeSdk } from "@get-bb/plugin-sdk/testing"; import type { BbPluginApi } from "@get-bb/plugin-sdk"; import type { FakeSdkOverrides } from "@get-bb/plugin-sdk/testing"; -import { BoardCollector } from "@/lib/collect"; +import { BoardCollector, placeTree } from "@/lib/collect"; +import type { ThreadRowLike } from "@/lib/bb-source"; /** Real sdk arg types (wire DTOs are not exported individually). */ type ThreadsArea = BbPluginApi["sdk"]["threads"]; @@ -120,8 +121,54 @@ function twoThreadOverrides(): FakeSdkOverrides { } as unknown as FakeSdkOverrides; } +type ThreadRowFixture = ReturnType; + +/** + * Fixtures describe the tree as per-parent listings. Derive the recursive + * descendants read and the batched `threadIds` row read from that listing, + * so harness call counts show only what the collector requested. + */ +function withTreeReads(overrides: FakeSdkOverrides): FakeSdkOverrides { + const threads = (overrides as { threads?: Record }).threads ?? {}; + const listByParent = threads.list as ((args: ListArgsLike) => Promise) | undefined; + if (!listByParent) return overrides; + const descendantsOf = async (threadId: string): Promise => { + const found: ThreadRowFixture[] = []; + const seen = new Set([threadId]); + for (let index = -1; index < found.length; index += 1) { + const parentId = index < 0 ? threadId : found[index]!.id; + for (const child of await listByParent({ parentThreadId: parentId, includeHidden: true })) { + if (seen.has(child.id)) continue; + seen.add(child.id); + found.push(child); + } + } + return found; + }; + let lastDescendants: ThreadRowFixture[] = []; + return { + ...overrides, + threads: { + ...threads, + experimental_listDescendants: async ({ threadIds }: { threadIds: readonly string[] }) => ({ + threads: await Promise.all( + threadIds.map(async (threadId) => { + lastDescendants = await descendantsOf(threadId); + return { threadId, descendantIds: lastDescendants.map((row) => row.id) }; + }), + ), + }), + list: async (args: ListArgsLike) => { + if (args.threadIds === undefined) return listByParent(args); + const wanted = new Set(args.threadIds); + return lastDescendants.filter((row) => wanted.has(row.id)); + }, + }, + } as unknown as FakeSdkOverrides; +} + function makeCollector(overrides: FakeSdkOverrides) { - const { sdk, harness } = createFakeSdk({ pluginId: "agent-board", overrides }); + const { sdk, harness } = createFakeSdk({ pluginId: "agent-board", overrides: withTreeReads(overrides) }); const bb = { sdk } as unknown as BbPluginApi; return { collector: new BoardCollector(bb, clock), harness }; } @@ -203,10 +250,11 @@ describe("bounded fetching and caching", () => { const snap = await collector.snapshot("thr_root", { fresh: true }); expect(snap.root.threadId).toBe("thr_root"); - // Underlying 1s/2s caches still hold: children/timelines/models are not + // Underlying 1s/2s caches still hold: tree/timelines/models are not // refetched. Only the uncached threads.get runs again. expect(harness.calls.length).toBe(afterFirst + 2); // threads.get + explicit fresh Fleet snapshot - expect(harness.callsTo("threads.list").length).toBe(2); + expect(harness.callsTo("threads.experimental_listDescendants").length).toBe(1); + expect(harness.callsTo("threads.list").length).toBe(1); expect(harness.callsTo("threads.timeline").length).toBe(2); expect(harness.callsTo("threads.defaultExecutionOptions").length).toBe(2); }); @@ -231,7 +279,8 @@ describe("bounded fetching and caching", () => { const [a, b] = await Promise.all([first, second]); expect(a).toEqual(b); expect(harness.callsTo("threads.get").length).toBe(1); - expect(harness.callsTo("threads.list").length).toBe(2); // root + child, once each + expect(harness.callsTo("threads.experimental_listDescendants").length).toBe(1); + expect(harness.callsTo("threads.list").length).toBe(1); // one batched row read for the tree }); it("invalidate() forces a full refetch including models", async () => { @@ -679,7 +728,8 @@ describe("terminal-state cache correctness (generation guards)", () => { const [a, b] = await Promise.all([first, second]); expect(a).toEqual(b); // one collect served both callers expect(harness.callsTo("threads.get").length).toBe(1); - expect(harness.callsTo("threads.list").length).toBe(2); // root + child, once each + expect(harness.callsTo("threads.experimental_listDescendants").length).toBe(1); + expect(harness.callsTo("threads.list").length).toBe(1); // one batched row read for the tree }); it("does not serve a pre-invalidation in-flight global dashboard after invalidateGlobal()", async () => { @@ -754,3 +804,50 @@ describe("terminal-state cache correctness (generation guards)", () => { expect(scopedAgain.cards.find((card) => card.threadId === "thr_child")).toBeDefined(); }); }); + +describe("tree placement", () => { + const row = (id: string, parentThreadId: string | null) => + thread(id, { parentThreadId }) as unknown as ThreadRowLike; + + it("keeps a tree of exactly the thread cap whole and not partial", () => { + const root = row("thr_root", null); + const rows = Array.from({ length: 23 }, (_, index) => row(`thr_${index}`, "thr_root")); + const placed = placeTree(root, rows, false); + expect(placed.rows).toHaveLength(24); + expect(placed.partial).toBe(false); + }); + + it("flags partial when one more reachable thread does not fit", () => { + const root = row("thr_root", null); + const rows = Array.from({ length: 24 }, (_, index) => row(`thr_${index}`, "thr_root")); + const placed = placeTree(root, rows, false); + expect(placed.rows).toHaveLength(24); + expect(placed.partial).toBe(true); + }); + + it("keeps the row order for siblings and depth from parent links", () => { + const root = row("thr_root", null); + const placed = placeTree( + root, + [row("thr_b", "thr_root"), row("thr_b1", "thr_b"), row("thr_a", "thr_root")], + false, + ); + expect(placed.rows.map((placedRow) => placedRow.id)).toEqual(["thr_root", "thr_b", "thr_a", "thr_b1"]); + expect(placed.childIds.get("thr_root")).toEqual(["thr_b", "thr_a"]); + expect(placed.depths.get("thr_b1")).toBe(2); + }); + + it("leaves out threads whose parent is not in the tree (below an archived thread)", () => { + const root = row("thr_root", null); + const placed = placeTree(root, [row("thr_child", "thr_root"), row("thr_orphan", "thr_archived")], false); + expect(placed.rows.map((placedRow) => placedRow.id)).toEqual(["thr_root", "thr_child"]); + expect(placed.partial).toBe(false); + }); + + it("flags partial for threads below the depth cap", () => { + const chain = [row("thr_1", "thr_root"), row("thr_2", "thr_1"), row("thr_3", "thr_2"), row("thr_4", "thr_3"), row("thr_5", "thr_4")]; + const placed = placeTree(row("thr_root", null), chain, false); + expect(placed.rows.map((placedRow) => placedRow.id)).toEqual(["thr_root", "thr_1", "thr_2", "thr_3", "thr_4"]); + expect(placed.partial).toBe(true); + }); +}); diff --git a/tests/fleet-app.test.tsx b/tests/fleet-app.test.tsx index 5c1aed4..508b2a7 100644 --- a/tests/fleet-app.test.tsx +++ b/tests/fleet-app.test.tsx @@ -1,5 +1,6 @@ // @vitest-environment jsdom import { beforeEach, afterEach, describe, it, expect } from "vitest"; +import { boardTabComponent } from "./board-tab"; import { loadPluginApp, renderSlot, @@ -20,7 +21,7 @@ afterEach(() => { function board(jobs = [job()]) { const snapshot = scoped([parent(), ...jobs.map(fleetCard)]); const s = renderSlot( - { component: app.threadPanelActions[0].component }, + { component: boardTabComponent(app) }, { threadId: "parent", params: null } as never, { rpc: { board_snapshot: () => snapshot } }, ); diff --git a/tests/fleet-collect.test.ts b/tests/fleet-collect.test.ts index 76d3034..1bba4da 100644 --- a/tests/fleet-collect.test.ts +++ b/tests/fleet-collect.test.ts @@ -36,6 +36,9 @@ async function setup(fleet: () => unknown) { threads: { get: async () => root, list: async (args) => (args?.parentThreadId ? [] : [root]), + experimental_listDescendants: async (args) => ({ + threads: args.threadIds.map((threadId) => ({ threadId, descendantIds: [] })), + }), defaultExecutionOptions: async () => ({ model: "parent-model" }), timeline: async () => ({ rows: [], @@ -126,6 +129,9 @@ describe("Fleet through Agent Board RPCs", () => { await h.harness.behavior.callRpc("board_snapshot", { threadId: "parent", }); + h.harness.behavior.experimental_watchRpc("board_snapshot", { + input: { threadId: "parent" }, + }); current = job({ state, revision: 3, diff --git a/tests/fleet-poll.test.ts b/tests/fleet-poll.test.ts index 773a4c2..278ecaa 100644 --- a/tests/fleet-poll.test.ts +++ b/tests/fleet-poll.test.ts @@ -1,6 +1,11 @@ import { afterEach, describe, it, expect, vi } from "vitest"; import { FleetActivePoller } from "../lib/fleet-poll"; afterEach(() => vi.useRealTimers()); +function watched(poller: FleetActivePoller): AbortController { + const controller = new AbortController(); + poller.watch(controller.signal); + return controller; +} const settle = async () => { await Promise.resolve(); await Promise.resolve(); @@ -10,6 +15,7 @@ describe("one bounded active Fleet loop", () => { vi.useFakeTimers(); const poll = vi.fn(); const p = new FleetActivePoller({ poll, changed: () => {} }); + watched(p); p.observe(false); await vi.advanceTimersByTimeAsync(60000); expect(poll).not.toHaveBeenCalled(); @@ -21,6 +27,7 @@ describe("one bounded active Fleet loop", () => { const poll = vi.fn().mockResolvedValue({ active: false, changed: true }); const changed = vi.fn(); const p = new FleetActivePoller({ poll, changed }); + watched(p); p.observe(true); await vi.advanceTimersByTimeAsync(2499); expect(poll).not.toHaveBeenCalled(); @@ -36,6 +43,7 @@ describe("one bounded active Fleet loop", () => { const poll = vi.fn().mockResolvedValue({ active: false, changed: true }); const changed = vi.fn(); const p = new FleetActivePoller({ poll, changed }); + watched(p); p.observe(true); p.observe(false); await vi.advanceTimersByTimeAsync(10000); @@ -48,6 +56,7 @@ describe("one bounded active Fleet loop", () => { vi.useFakeTimers(); const poll = vi.fn().mockResolvedValue({ active: true, changed: false }); const p = new FleetActivePoller({ poll, changed: () => {} }); + watched(p); for (let i = 0; i < 20; i++) p.observe(true); await vi.advanceTimersByTimeAsync(10000); expect(poll).toHaveBeenCalledTimes(4); @@ -63,6 +72,7 @@ describe("one bounded active Fleet loop", () => { ), ); const p = new FleetActivePoller({ poll, changed: () => {} }); + watched(p); p.observe(true); await vi.advanceTimersByTimeAsync(2500); for (let i = 0; i < 3; i++) { @@ -83,6 +93,7 @@ describe("one bounded active Fleet loop", () => { poll: () => new Promise((r) => (resolve = r)), changed, }); + watched(p); p.observe(true); await vi.advanceTimersByTimeAsync(2500); p.dispose(); @@ -91,21 +102,38 @@ describe("one bounded active Fleet loop", () => { expect(changed).not.toHaveBeenCalled(); expect(vi.getTimerCount()).toBe(0); }); - it("stops stale watchers after 90 seconds", async () => { + it("stops when the last watched view closes", async () => { vi.useFakeTimers(); const poll = vi.fn().mockResolvedValue({ active: true, changed: false }); const p = new FleetActivePoller({ poll, changed: () => {} }); + const board = watched(p); + const page = watched(p); p.observe(true); - await vi.advanceTimersByTimeAsync(95000); - const count = poll.mock.calls.length; + await vi.advanceTimersByTimeAsync(2500); + expect(poll).toHaveBeenCalledTimes(1); + board.abort(); + await vi.advanceTimersByTimeAsync(2500); + expect(poll).toHaveBeenCalledTimes(2); + page.abort(); await vi.advanceTimersByTimeAsync(30000); - expect(poll).toHaveBeenCalledTimes(count); + expect(poll).toHaveBeenCalledTimes(2); + expect(vi.getTimerCount()).toBe(0); + p.dispose(); + }); + it("ignores observations while no view is watched", async () => { + vi.useFakeTimers(); + const poll = vi.fn().mockResolvedValue({ active: true, changed: false }); + const p = new FleetActivePoller({ poll, changed: () => {} }); + p.observe(true); + await vi.advanceTimersByTimeAsync(10000); + expect(poll).not.toHaveBeenCalled(); p.dispose(); }); it("RPC errors stop active polling until ordinary discovery", async () => { vi.useFakeTimers(); const poll = vi.fn().mockRejectedValue(Error("offline")); const p = new FleetActivePoller({ poll, changed: () => {} }); + watched(p); p.observe(true); await vi.advanceTimersByTimeAsync(10000); expect(poll).toHaveBeenCalledTimes(1); diff --git a/tests/focus-motion.test.tsx b/tests/focus-motion.test.tsx index 36f2538..e55f069 100644 --- a/tests/focus-motion.test.tsx +++ b/tests/focus-motion.test.tsx @@ -5,6 +5,7 @@ // module state: a poisoned entry from an earlier test would mask the stub. import { describe, expect, it } from "vitest"; import { loadPluginApp, renderSlot, type CapturedPluginApp, type RenderedSlot } from "@get-bb/plugin-sdk/testing/app"; +import { boardTabComponent } from "./board-tab"; import type { BoardCard, BoardSnapshot } from "@/contract/rpc"; let app: CapturedPluginApp; @@ -99,9 +100,8 @@ function scanFocusSnapshot(): BoardSnapshot { } function mountBoard(options: Parameters[2] = {}): RenderedSlot { - const registration = app.threadPanelActions[0]!; return renderSlot( - { component: registration.component }, + { component: boardTabComponent(app) }, { threadId: "thr_root", params: null } as never, options, ); diff --git a/tests/global-app.test.tsx b/tests/global-app.test.tsx index b92e9d0..56be868 100644 --- a/tests/global-app.test.tsx +++ b/tests/global-app.test.tsx @@ -1,10 +1,10 @@ // @vitest-environment jsdom import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { waitFor } from "@testing-library/react"; import { readFileSync } from "node:fs"; import { loadPluginApp, renderSlot, type CapturedPluginApp, type RenderedSlot } from "@get-bb/plugin-sdk/testing/app"; import type { AttentionItem, BoardCard, ExecutionTimeline, GlobalDashboardSnapshot, GlobalExecution, TimelineEvent } from "@/contract/rpc"; import { BOARD_CHANGED_CHANNEL } from "@/contract/rpc"; -const GLOBAL_DISCOVERY_INTERVAL_MS = 20_000; const LONG_TOKEN = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; const LONG_MODEL = "provider/some-extremely-long-model-name-with-many-segments:latest"; const LONG_PATH = "/home/user/projects/some-extremely-long-directory-name/src/components/another-extremely-long-directory/ExecutionCard.tsx"; @@ -67,7 +67,7 @@ describe("global page registration", () => { it("registers the stable nav panel while retaining the thread action", () => { expect(app.navPanels).toHaveLength(1); expect(app.navPanels[0]).toMatchObject({ id: "agent-board-global", title: "Agent Board", icon: "ListTodo", path: "operations" }); - expect(app.threadPanelActions).toHaveLength(1); + expect(app.tabs.filter((tab) => tab.id === "agent-board")).toHaveLength(1); }); }); @@ -236,33 +236,43 @@ describe("global operations rendering", () => { }); describe("global refresh behavior", () => { - it("refetches on a global invalidation but ignores a scoped-only signal", async () => { + it("refetches on a global invalidation but ignores a scoped-only one", async () => { const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot() } }); await slot.findByText("Auth refactor"); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement")).toHaveLength(1); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: ["thr_other"] }); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement")).toHaveLength(1); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: [], global: true }); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement")).toHaveLength(2); + const reads = () => slot.inspection.rpcCalls.filter(call => call.method === "global_dashboard"); + expect(reads()).toHaveLength(1); + await slot.behavior.experimental_invalidateRpc(["board:thr_other"]); + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(reads()).toHaveLength(1); + await slot.behavior.experimental_invalidateRpc(["global-dashboard"]); + await waitFor(() => expect(reads()).toHaveLength(2)); slot.unmount(); }); - it("uses one slow discovery timer and stops it on unmount", async () => { + it("does not poll from the client; discovery refreshes come from the backend", async () => { vi.useFakeTimers(); const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot() } }); await vi.runOnlyPendingTimersAsync(); - const initial = slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement").length; - await vi.advanceTimersByTimeAsync(GLOBAL_DISCOVERY_INTERVAL_MS); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement").length).toBeGreaterThan(initial); - const beforeUnmount = slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement").length; + const reads = () => slot.inspection.rpcCalls.filter(call => call.method === "global_dashboard").length; + const initial = reads(); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads()).toBe(initial); + slot.unmount(); + }); + + it("records a change signal without refetching on it", async () => { + const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot() } }); + await slot.findByText("Auth refactor"); + expect(slot.getByLabelText("Monitoring health").textContent).toContain("No change signal received"); + await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: [], global: true }); + expect(slot.getByLabelText("Monitoring health").textContent).toContain("Last change signal"); + expect(slot.inspection.rpcCalls.filter(call => call.method === "global_dashboard")).toHaveLength(1); slot.unmount(); - await vi.advanceTimersByTimeAsync(GLOBAL_DISCOVERY_INTERVAL_MS * 2); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement")).toHaveLength(beforeUnmount); }); }); describe("terminal-state self-correction in the hook", () => { - it("queues exactly one trailing refetch when a signal lands mid-flight (no dropped terminal signals)", async () => { + it("ends a mid-flight invalidation with a read that started after it (no dropped terminal signals)", async () => { let calls = 0; let release!: () => void; const gate = new Promise((resolve) => { @@ -272,151 +282,24 @@ describe("terminal-state self-correction in the hook", () => { rpc: { global_dashboard: async () => { calls += 1; - if (calls === 1) await gate; // first fetch is slow - return snapshot(); + const call = calls; + if (call === 1) await gate; // first fetch is slow + return snapshot({ active: [execution("bb:auth", { card: boardCard({ key: "card:bb:auth", title: `Read ${call}` }) })] }); }, }, }); - // The first fetch is in flight; a terminal signal arrives mid-flight. - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement")).toHaveLength(1); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: [], global: true }); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: [], global: true }); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: [], global: true }); - expect(calls).toBe(1); // nothing new yet: the fetch is still in flight - + expect(calls).toBe(1); + // The host's invalidation waits for the refetch it starts, so release the + // slow first read before awaiting it. + const invalidation = slot.behavior.experimental_invalidateRpc(["global-dashboard"]); release(); - await new Promise((resolve) => setTimeout(resolve, 50)); - // The in-flight fetch settled, then exactly ONE trailing refetch ran — - // the signal burst collapsed into a single retry, not a storm. + await invalidation; expect(calls).toBe(2); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement")).toHaveLength(2); - - // Signals after settle go back to the normal immediate path. - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: [], global: true }); - await new Promise((resolve) => setTimeout(resolve, 25)); - expect(calls).toBe(3); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement")).toHaveLength(3); // no retry loop - slot.unmount(); - }); -}); - -describe("latestPublicUpdate excerpt rendering", () => { - it("renders a 2–3-line excerpt on active execution cards only", async () => { - const withExcerpt = execution("bb:auth", { - card: boardCard({ latestPublicUpdate: "Reconciled the terminal state; all counts are now truthful." }), - }); - const recentToo = execution("bb:done", { - card: boardCard({ key: "done", title: "Finished run", status: "completed", completedAt: Date.now() - 1_000, durationMs: 5_000, activity: null, latestPublicUpdate: "Recent text should not appear as an excerpt." }), - }); - const slot = mountGlobal({ - rpc: { - global_dashboard: () => - snapshot({ - active: [withExcerpt], - recent: [recentToo], - counts: { active: 1, waiting: 0, attention: 0, recent: 1, failures: 0 }, - }), - }, - }); - await slot.findByText("Auth refactor"); - const excerpt = slot.container.querySelector('[data-part="public-update"]'); - expect(excerpt?.textContent).toBe("Reconciled the terminal state; all counts are now truthful."); - expect(excerpt?.className).toContain("line-clamp-3"); - expect(slot.container.textContent).not.toContain("Recent text should not appear as an excerpt."); - slot.unmount(); - }); - - it("omits the excerpt entirely when there is no public update", async () => { - const noText = execution("bb:quiet", { card: boardCard({ key: "quiet", latestPublicUpdate: null }) }); - const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot({ active: [noText] }) } }); - await slot.findByText("Auth refactor"); - expect(slot.container.querySelector('[data-part="public-update"]')).toBeNull(); - slot.unmount(); - }); -}); - -describe("lazy shared execution timeline", () => { - it("adds zero initial requests, then renders chronological operational events from one timeline RPC", async () => { - const rpcTimeline = vi.fn(async () => executionTimeline({ - events: [ - timelineEvent("start", 100), - timelineEvent("model", 200, { kind: "model-selected", title: "Model selected", model: "qwen3:8b" }), - timelineEvent("tool", 300, { kind: "tool-complete", status: "completed", title: "Read file completed", toolName: "read_file", target: "src/auth.ts", durationMs: 1_200 }), - timelineEvent("update", 400, { kind: "public-update", status: null, title: "Public update", summary: "The collector remains bounded." }), - timelineEvent("done", 500, { kind: "execution-complete", status: "completed", title: "Execution finished", durationMs: 400 }), - ], - })); - const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot(), execution_timeline: rpcTimeline } }); - await slot.findByText("Auth refactor"); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement")).toHaveLength(1); - expect(rpcTimeline).not.toHaveBeenCalled(); - - const trigger = slot.getByRole("button", { name: "Open timeline for Auth refactor" }); - expect(trigger.tagName).toBe("BUTTON"); - expect(trigger.getAttribute("aria-haspopup")).toBe("dialog"); - trigger.click(); - expect(await slot.findByText("The collector remains bounded.")).toBeTruthy(); - expect(rpcTimeline).toHaveBeenCalledTimes(1); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement").filter((call) => call.method === "execution_timeline")).toHaveLength(1); - const rows = Array.from(document.querySelectorAll("[data-event-kind]")); - expect(rows.map((row) => row.getAttribute("data-event-kind"))).toEqual(["execution-start", "model-selected", "tool-complete", "public-update", "execution-complete"]); - expect(document.body.textContent).toContain("read_file · src/auth.ts · 1s"); - expect(document.body.textContent).toContain("qwen3:8b"); - expect(document.querySelector('[data-icon="Bot"]')).toBeTruthy(); - expect(document.querySelector('[aria-label="Chronological execution events"]')?.getAttribute("tabindex")).toBe("0"); - slot.unmount(); - }); - - it("opens the same Timeline action for a recent execution", async () => { - const recent = execution("bb:done", { card: boardCard({ key: "done", title: "Finished run", status: "completed", completedAt: Date.now(), activity: null }) }); - const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot({ active: [], recent: [recent], counts: { active: 0, waiting: 0, attention: 0, recent: 1, failures: 0 } }), execution_timeline: () => executionTimeline({ executionKey: "bb:done", events: [] }) } }); - const trigger = await slot.findByRole("button", { name: "Open timeline for Finished run" }); - trigger.click(); - expect(await slot.findByText("No public operational events are available.")).toBeTruthy(); - slot.unmount(); - }); - - it("opens a Redteam timeline without inventing a BB thread", async () => { - const redteam = execution("redteam:scan-1", { card: boardCard({ key: "scan", source: "redteam", kind: "phase", title: "Redteam — Tiny lab", threadId: null, scanId: "scan-1" }) }); - const rpcTimeline = vi.fn(async () => executionTimeline({ executionKey: "redteam:scan-1", source: "redteam", events: [timelineEvent("phase", 100, { executionKey: "redteam:scan-1", source: "redteam", kind: "phase-complete", status: "completed", title: "Code review completed", phase: "Code review" })] })); - const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot({ active: [redteam] }), execution_timeline: rpcTimeline } }); - (await slot.findByRole("button", { name: "Open timeline for Redteam — Tiny lab" })).click(); - expect(await slot.findByText("Code review completed")).toBeTruthy(); - expect(rpcTimeline).toHaveBeenCalledWith({ source: "redteam", executionKey: "redteam:scan-1", scanId: "scan-1" }); - slot.unmount(); - }); - - it("renders failure, waiting, truncation, warning, and privacy-safe text without raw JSON", async () => { - const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot(), execution_timeline: () => executionTimeline({ - active: true, truncated: true, sourceWarning: "Some lifecycle details are temporarily unavailable.", - events: [ - timelineEvent("wait", 100, { kind: "approval-requested", status: "waiting", title: "Approval requested", summary: "Permission needed", attentionType: "action-required" }), - timelineEvent("fail", 200, { kind: "tool-failed", status: "failed", title: "Command failed", summary: "Safe error preview", toolName: "command", target: "npm test" }), - ], - }) } }); - (await slot.findByRole("button", { name: "Open timeline for Auth refactor" })).click(); - expect(await slot.findByText(/Earlier activity omitted/)).toBeTruthy(); - expect(slot.getByText("Some lifecycle details are temporarily unavailable.")).toBeTruthy(); - expect(slot.getByText("Action required")).toBeTruthy(); - expect(slot.getByText("Safe error preview")).toBeTruthy(); - expect(document.body.textContent).not.toContain("sourceMetadata"); - expect(document.body.textContent).not.toContain("SECRET REASONING"); - expect(slot.getAllByText("Live").length).toBeGreaterThan(0); - slot.unmount(); - }); - - it("shows loading and a recoverable source error", async () => { - let reject!: (reason: unknown) => void; - const gate = new Promise((_resolve, rejectPromise) => { reject = rejectPromise; }); - const slot = mountGlobal({ rpc: { global_dashboard: () => snapshot(), execution_timeline: () => gate } }); - (await slot.findByRole("button", { name: "Open timeline for Auth refactor" })).click(); - expect(await slot.findByText("Loading timeline…")).toBeTruthy(); - reject(new Error("source offline")); - expect((await slot.findByRole("alert")).textContent).toContain("Timeline unavailable: source offline"); - expect(slot.getByRole("button", { name: "Try again" })).toBeTruthy(); + expect(await slot.findAllByText("Read 2")).not.toHaveLength(0); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(calls).toBe(2); // no retry loop slot.unmount(); }); - it("collapses a burst of active invalidations into one trailing request and stops refreshing after completion", async () => { let calls = 0; let release!: (value: ExecutionTimeline) => void; @@ -480,7 +363,9 @@ describe("attention dismissal UI", () => { type: item.type, updatedAt: item.updatedAt, }); - expect(slot.inspection.rpcCalls.filter(call => call.method !== "release_announcement").filter((call) => call.method === "global_dashboard")).toHaveLength(2); + // The backend invalidates the global page's tag after a dismissal. + await slot.behavior.experimental_invalidateRpc(["global-dashboard"]); + await waitFor(() => expect(slot.inspection.rpcCalls.filter((call) => call.method === "global_dashboard")).toHaveLength(2)); slot.unmount(); }); @@ -504,7 +389,8 @@ describe("attention dismissal UI", () => { clearButton.click(); await new Promise((resolve) => setTimeout(resolve, 25)); expect(clear).toHaveBeenCalledWith({}); - expect(slot.container.textContent).not.toContain("Clear dismissed"); + await slot.behavior.experimental_invalidateRpc(["global-dashboard"]); + await waitFor(() => expect(slot.container.textContent).not.toContain("Clear dismissed")); slot.unmount(); }); diff --git a/tests/redteam-poll.test.ts b/tests/redteam-poll.test.ts index a0bbbff..320cdad 100644 --- a/tests/redteam-poll.test.ts +++ b/tests/redteam-poll.test.ts @@ -1,36 +1,55 @@ import { describe, expect, it, vi } from "vitest"; import { RedteamActivePoller } from "@/lib/redteam-poll"; +function watched(poller: RedteamActivePoller, rootThreadId: string): AbortController { + const controller = new AbortController(); + poller.watch(rootThreadId, controller.signal); + return controller; +} + +function watchedGlobal(poller: RedteamActivePoller): AbortController { + const controller = new AbortController(); + poller.watchGlobal(controller.signal); + return controller; +} + describe("active Redteam polling", () => { it("starts only after an active scan is observed and stops at terminal state", async () => { const poll = vi.fn().mockResolvedValueOnce({ changed: true, active: true }).mockResolvedValueOnce({ changed: true, active: false }); const changed = vi.fn(); const poller = new RedteamActivePoller({ poll, changed, intervalMs: 60_000 }); - expect(poller.isPolling("p1")).toBe(false); poller.observe("thr-1", "p1", true); expect(poller.isPolling("p1")).toBe(true); + expect(poller.isPolling("p1")).toBe(false); watched(poller, "thr-1"); poller.observe("thr-1", "p1", true); expect(poller.isPolling("p1")).toBe(true); await poller.tick(); expect(changed).toHaveBeenCalledWith("p1"); expect(poller.isPolling("p1")).toBe(true); await poller.tick(); expect(poller.isPolling("p1")).toBe(false); poller.dispose(); }); it("does not overlap slow requests for the same project", async () => { let release!: (value: { changed: boolean; active: boolean }) => void; const gate = new Promise<{ changed: boolean; active: boolean }>((resolve) => { release = resolve; }); - const poll = vi.fn(async () => gate); const poller = new RedteamActivePoller({ poll, changed: vi.fn(), intervalMs: 60_000 }); poller.observe("thr-1", "p1", true); + const poll = vi.fn(async () => gate); const poller = new RedteamActivePoller({ poll, changed: vi.fn(), intervalMs: 60_000 }); watched(poller, "thr-1"); poller.observe("thr-1", "p1", true); const first = poller.tick(); const second = poller.tick(); expect(poll).toHaveBeenCalledTimes(1); release({ changed: false, active: true }); await Promise.all([first, second]); poller.dispose(); }); - it("does not poll idle/completed scans and expires closed boards", async () => { - let now = 0; const poll = vi.fn(async () => ({ changed: false, active: true })); const poller = new RedteamActivePoller({ poll, changed: vi.fn(), now: () => now, intervalMs: 60_000, watchedTtlMs: 100 }); + it("does not poll idle/completed scans and stops when the board's watched read ends", async () => { + const poll = vi.fn(async () => ({ changed: false, active: true })); const poller = new RedteamActivePoller({ poll, changed: vi.fn(), intervalMs: 60_000 }); + const board = watched(poller, "thr-1"); poller.observe("thr-1", "p1", false); await poller.tick(); expect(poll).not.toHaveBeenCalled(); - poller.observe("thr-1", "p1", true); now = 101; await poller.tick(); expect(poll).not.toHaveBeenCalled(); expect(poller.isPolling("p1")).toBe(false); poller.dispose(); + poller.observe("thr-1", "p1", true); board.abort(); await poller.tick(); expect(poll).not.toHaveBeenCalled(); expect(poller.isPolling("p1")).toBe(false); poller.dispose(); + }); + + it("ignores observations of boards nobody watches", async () => { + const poll = vi.fn(async () => ({ changed: false, active: true })); const poller = new RedteamActivePoller({ poll, changed: vi.fn(), intervalMs: 60_000 }); + poller.observe("thr-1", "p1", true); await poller.tick(); expect(poll).not.toHaveBeenCalled(); poller.dispose(); }); it("dispose stops future activity", async () => { const poll = vi.fn(async () => ({ changed: false, active: true })); const poller = new RedteamActivePoller({ poll, changed: vi.fn(), intervalMs: 60_000 }); - poller.observe("thr-1", "p1", true); poller.dispose(); await poller.tick(); expect(poll).not.toHaveBeenCalled(); + watched(poller, "thr-1"); poller.observe("thr-1", "p1", true); poller.dispose(); await poller.tick(); expect(poll).not.toHaveBeenCalled(); }); it("polls a globally observed active scan and stops at terminal state", async () => { const pollGlobal = vi.fn().mockResolvedValueOnce({ changed: true, active: true }).mockResolvedValueOnce({ changed: true, active: false }); const changedGlobal = vi.fn(); const poller = new RedteamActivePoller({ poll: vi.fn(), changed: vi.fn(), pollGlobal, changedGlobal, intervalMs: 60_000 }); + watchedGlobal(poller); poller.observeGlobal(true); expect(poller.isGlobalPolling()).toBe(true); await poller.tick(); @@ -45,6 +64,7 @@ describe("active Redteam polling", () => { const gate = new Promise<{ changed: boolean; active: boolean }>((resolve) => { release = resolve; }); const pollGlobal = vi.fn(async () => gate); const poller = new RedteamActivePoller({ poll: vi.fn(), changed: vi.fn(), pollGlobal, intervalMs: 60_000 }); + watchedGlobal(poller); poller.observeGlobal(true); const first = poller.tick(); const second = poller.tick(); @@ -54,9 +74,22 @@ describe("active Redteam polling", () => { poller.dispose(); }); + it("stops global polling when the global page's watched read ends", async () => { + const pollGlobal = vi.fn(async () => ({ changed: false, active: true })); + const poller = new RedteamActivePoller({ poll: vi.fn(), changed: vi.fn(), pollGlobal, intervalMs: 60_000 }); + const page = watchedGlobal(poller); + poller.observeGlobal(true); + page.abort(); + await poller.tick(); + expect(pollGlobal).not.toHaveBeenCalled(); + expect(poller.isGlobalPolling()).toBe(false); + poller.dispose(); + }); + it("does not poll Redteam globally while no active scan is known", async () => { const pollGlobal = vi.fn(async () => ({ changed: false, active: false })); const poller = new RedteamActivePoller({ poll: vi.fn(), changed: vi.fn(), pollGlobal, intervalMs: 60_000 }); + watchedGlobal(poller); poller.observeGlobal(false); await poller.tick(); expect(pollGlobal).not.toHaveBeenCalled(); diff --git a/tests/server.test.ts b/tests/server.test.ts index 4e83a8f..faf0020 100644 --- a/tests/server.test.ts +++ b/tests/server.test.ts @@ -52,6 +52,15 @@ async function makeHost(overrides: FakeSdkOverrides = {}) { }); }, list: async () => [], + experimental_listDescendants: async (args: { threadIds: readonly string[] }) => ({ + threads: args.threadIds.map((threadId) => ({ threadId, descendantIds: [] })), + }), + experimental_listAncestors: async (args: { threadIds: readonly string[] }) => ({ + threads: args.threadIds.map((threadId) => ({ + threadId, + ancestorIds: threadId === "thr_grandchild" ? ["thr_unseen_child", "thr_root"] : [], + })), + }), timeline: async () => ({ rows: [], activeWorkflows: [], @@ -98,6 +107,21 @@ async function makeHost(overrides: FakeSdkOverrides = {}) { const settle = (ms = 550) => new Promise((resolve) => setTimeout(resolve, ms)); +/** Open a board as the app does: read it and hold its watched read. */ +async function openBoard(host: FakePluginHost, threadId: string): Promise<() => void> { + await host.harness.behavior.callRpc("board_snapshot", { threadId }); + const close = host.harness.behavior.experimental_watchRpc("board_snapshot", { input: { threadId } }); + await settle(20); + return close; +} + +async function openGlobalPage(host: FakePluginHost): Promise<() => void> { + await host.harness.behavior.callRpc("global_dashboard", {}); + const close = host.harness.behavior.experimental_watchRpc("global_dashboard", { input: {} }); + await settle(20); + return close; +} + describe("rpc registration", () => { it("serves a schema-validated board_snapshot", async () => { const { host } = await makeHost(); @@ -172,7 +196,7 @@ describe("rpc registration", () => { describe("watcher wiring", () => { it("publishes an invalidation signal when an observed thread changes", async () => { const { host, changed } = await makeHost(); - await host.harness.behavior.callRpc("board_snapshot", { threadId: "thr_root" }); + await openBoard(host, "thr_root"); expect(host.harness.inspection.realtimeSignals).toEqual([]); changed({ @@ -187,12 +211,69 @@ describe("watcher wiring", () => { expect(signals).toHaveLength(1); expect(signals[0]!.channel).toBe(BOARD_CHANGED_CHANNEL); expect(signals[0]!.payload).toEqual({ roots: ["thr_root"] }); + expect(host.harness.inspection.experimental_rpcInvalidations).toEqual([["board:thr_root"]]); }); - it("coalesces chatty changes into one signal per debounce window", async () => { + it("invalidates the root board for a thread created below a parent the board has not seen", async () => { + const { host } = await makeHost(); + await openBoard(host, "thr_root"); + + await host.harness.behavior.emitThreadEvent("thread.created", { + thread: makeThreadResponse({ id: "thr_grandchild", parentThreadId: "thr_unseen_child" }), + }); + await settle(); + + expect(host.harness.inspection.sdk.callsTo("threads.experimental_listAncestors")).toHaveLength(1); + expect(host.harness.inspection.sdk.callsTo("threads.get")).toHaveLength(1); // the board read only + expect(host.harness.inspection.experimental_rpcInvalidations).toEqual([["board:thr_root"]]); + }); + + it("resolves several unseen threads with one batched ancestor read", async () => { + const { host } = await makeHost(); + await openBoard(host, "thr_root"); + + await Promise.all([ + host.harness.behavior.emitThreadEvent("thread.created", { + thread: makeThreadResponse({ id: "thr_grandchild", parentThreadId: "thr_unseen_child" }), + }), + host.harness.behavior.emitThreadEvent("thread.created", { + thread: makeThreadResponse({ id: "thr_stranger", parentThreadId: "thr_elsewhere" }), + }), + ]); + await settle(); + + const lookups = host.harness.inspection.sdk.callsTo("threads.experimental_listAncestors"); + expect(lookups).toHaveLength(1); + expect(lookups[0]).toEqual([{ threadIds: ["thr_grandchild", "thr_stranger"] }]); + expect(host.harness.inspection.experimental_rpcInvalidations).toEqual([["board:thr_root"]]); + }); + + it("stops invalidating a board once its last watched read ends", async () => { + const { host, changed } = await makeHost(); + const close = await openBoard(host, "thr_root"); + close(); + + changed({ entity: "thread", type: "changed", id: "thr_root", changes: ["status-changed"] }); + await settle(); + + expect(host.harness.inspection.realtimeSignals).toEqual([]); + expect(host.harness.inspection.experimental_rpcInvalidations).toEqual([]); + }); + + it("does not watch a board that was only read, never watched", async () => { const { host, changed } = await makeHost(); await host.harness.behavior.callRpc("board_snapshot", { threadId: "thr_root" }); + changed({ entity: "thread", type: "changed", id: "thr_root", changes: ["status-changed"] }); + await settle(); + + expect(host.harness.inspection.realtimeSignals).toEqual([]); + }); + + it("coalesces chatty changes into one signal per debounce window", async () => { + const { host, changed } = await makeHost(); + await openBoard(host, "thr_root"); + changed({ entity: "thread", type: "changed", @@ -212,7 +293,7 @@ describe("watcher wiring", () => { it("stays quiet for cosmetic and reasoning-only changes", async () => { const { host, changed } = await makeHost(); - await host.harness.behavior.callRpc("board_snapshot", { threadId: "thr_root" }); + await openBoard(host, "thr_root"); changed({ entity: "thread", @@ -234,7 +315,7 @@ describe("watcher wiring", () => { it("reacts to lifecycle events for observed threads", async () => { const { host } = await makeHost(); - await host.harness.behavior.callRpc("board_snapshot", { threadId: "thr_root" }); + await openBoard(host, "thr_root"); await host.harness.behavior.emitThreadEvent("thread.idle", { thread: makeThreadResponse({ id: "thr_root", status: "idle" }), @@ -249,13 +330,29 @@ describe("watcher wiring", () => { it("publishes a global invalidation after the global page is observed", async () => { const { host, changed } = await makeHost(); - await host.harness.behavior.callRpc("global_dashboard", {}); + await openGlobalPage(host); changed({ entity: "thread", type: "changed", id: "thr_unrelated", changes: ["status-changed"] }); await settle(); expect(host.harness.inspection.realtimeSignals).toContainEqual({ channel: BOARD_CHANGED_CHANNEL, payload: { roots: [], global: true }, }); + expect(host.harness.inspection.experimental_rpcInvalidations).toContainEqual(["global-dashboard"]); + }); + + it("refreshes the watched global page every 20 seconds for optional-source discovery, and stops when it closes", async () => { + vi.useFakeTimers(); + try { + const { host } = await makeHost(); + const close = host.harness.behavior.experimental_watchRpc("global_dashboard", { input: {} }); + await vi.advanceTimersByTimeAsync(20_000); + expect(host.harness.inspection.experimental_rpcInvalidations).toEqual([["global-dashboard"]]); + close(); + await vi.advanceTimersByTimeAsync(60_000); + expect(host.harness.inspection.experimental_rpcInvalidations).toEqual([["global-dashboard"]]); + } finally { + vi.useRealTimers(); + } }); }); @@ -301,7 +398,7 @@ describe("cli", () => { describe("dispose", () => { it("stops the watcher and unsubscribes on plugin dispose", async () => { const { host, changed } = await makeHost(); - await host.harness.behavior.callRpc("board_snapshot", { threadId: "thr_root" }); + await openBoard(host, "thr_root"); await host.harness.lifecycle.dispose(); changed({ @@ -392,6 +489,7 @@ describe("attention dismissal RPCs", () => { updatedAt: item.updatedAt, })) as { dismissed: number }; expect(dismissed.dismissed).toBe(1); + expect(host.harness.inspection.experimental_rpcInvalidations).toContainEqual(["global-dashboard"]); const second = (await host.harness.behavior.callRpc("global_dashboard", { fresh: true, diff --git a/tests/vnext-app.test.tsx b/tests/vnext-app.test.tsx index 26bb897..cbd1885 100644 --- a/tests/vnext-app.test.tsx +++ b/tests/vnext-app.test.tsx @@ -38,7 +38,7 @@ describe("Phase 1 dashboard interactions", () => { let result = phaseSnapshot(); const slot = mount({ rpc: { global_dashboard: () => result } }); await slot.findByText("Task A"); expect(slot.container.querySelector("[data-empty-attention]")?.className).not.toContain("py-4"); result = phaseSnapshot({ attention: [{ id: "a", executionKey: "a", type: "action-required", status: "open", title: "Approval", message: "Approve tool", firstObservedAt: 10, updatedAt: 20, threadId: "thread", source: "bb" }] }); - await slot.behavior.emitRealtime(BOARD_CHANGED_CHANNEL, { roots: [], global: true }); await slot.findByText("Approve tool"); expect(slot.container.querySelector("[data-empty-attention]")).toBeNull(); slot.unmount(); + await slot.behavior.experimental_invalidateRpc(["global-dashboard"]); await slot.findByText("Approve tool"); expect(slot.container.querySelector("[data-empty-attention]")).toBeNull(); slot.unmount(); }); it("shows watcher errors independently of transport and recovers on reconnect", async () => { const fetch = vi.fn(() => phaseSnapshot({ monitoring: { status: "degraded", startedAt: 1, lastEventAt: 2, lastPublishedAt: null, lastErrorAt: 3 } })); diff --git a/tests/watch.test.ts b/tests/watch.test.ts index 7a9e9fd..b55ad9f 100644 --- a/tests/watch.test.ts +++ b/tests/watch.test.ts @@ -1,5 +1,5 @@ // Watcher tests: relevance filtering, candidate selection, parentage -// lookups, rate-limited lookups, debounce coalescing, TTL pruning, dispose. +// lookups, rate-limited lookups, debounce coalescing, watched lifetimes, dispose. import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { BoardWatcher, @@ -15,27 +15,49 @@ interface Recorded { lookups: string[]; } +const ANCESTORS: Record = { + thr_new: ["thr_root"], + thr_grandchild: ["thr_unseen_child", "thr_root"], +}; + function makeWatcher(overrides?: { now?: () => number; debounceMs?: number; lookupGapMs?: number; - watchedTtlMs?: number; }) { const recorded: Recorded = { publishes: [], lookups: [] }; const watcher = new BoardWatcher({ publish: (payload) => recorded.publishes.push(payload), - getParent: async (threadId) => { + getAncestors: async (threadId) => { recorded.lookups.push(threadId); - return threadId === "thr_new" ? "thr_root" : null; + return ANCESTORS[threadId] ?? []; }, now: overrides?.now, debounceMs: overrides?.debounceMs, lookupGapMs: overrides?.lookupGapMs, - watchedTtlMs: overrides?.watchedTtlMs, }); return { watcher, recorded }; } +/** Watch a root as an open board's watched read does, then record its tree. */ +function watchRoot( + watcher: BoardWatcher, + rootThreadId: string, + threads: string[], + projectId: string | null, +): AbortController { + const controller = new AbortController(); + watcher.watch(rootThreadId, controller.signal); + watcher.touch(rootThreadId, threads, projectId); + return controller; +} + +function watchGlobal(watcher: BoardWatcher): AbortController { + const controller = new AbortController(); + watcher.watchGlobal(controller.signal); + return controller; +} + function changed(id: string | undefined, fields: Partial = {}): ThreadChangedEvent { return { entity: "thread", type: "changed", id, ...fields }; } @@ -51,8 +73,8 @@ describe("watcher health", () => { }); it("records publication errors and recovers after a successful signal", async () => { let fail = true; let now = 10; - const watcher = new BoardWatcher({ publish: () => { if (fail) throw new Error("private source detail"); }, getParent: async () => null, now: () => now }); - watcher.touchGlobal(); await watcher.handleChanged(changed("thread")); expect(() => watcher.flush()).not.toThrow(); + const watcher = new BoardWatcher({ publish: () => { if (fail) throw new Error("private source detail"); }, getAncestors: async () => [], now: () => now }); + watchGlobal(watcher); await watcher.handleChanged(changed("thread")); expect(() => watcher.flush()).not.toThrow(); expect(watcher.health()).toMatchObject({ status: "degraded", lastErrorAt: 10 }); fail = false; now = 20; await watcher.handleChanged(changed("thread")); watcher.flush(); expect(watcher.health()).toMatchObject({ status: "watching", lastEventAt: 20, lastPublishedAt: 20 }); expect(JSON.stringify(watcher.health())).not.toContain("private"); watcher.dispose(); @@ -70,7 +92,7 @@ afterEach(() => { describe("invalidation for observed threads", () => { it("marks a watched root dirty when an observed thread changes", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root", "thr_child"], null); + watchRoot(watcher, "thr_root", ["thr_root", "thr_child"], null); await watcher.handleChanged(changed("thr_child", { changes: ["status-changed"] })); watcher.flush(); @@ -80,7 +102,7 @@ describe("invalidation for observed threads", () => { it("ignores cosmetic changes and pure reasoning deltas", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged(changed("thr_root", { changes: ["read-state-changed"] })); await watcher.handleChanged(changed("thr_root", { changes: ["pin-state-changed"] })); await watcher.handleChanged( @@ -95,8 +117,8 @@ describe("invalidation for observed threads", () => { it("refreshes every board on an id-less broadcast", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_a", ["thr_a"], null); - watcher.touch("thr_b", ["thr_b"], null); + watchRoot(watcher, "thr_a", ["thr_a"], null); + watchRoot(watcher, "thr_b", ["thr_b"], null); await watcher.handleChanged(changed(undefined, { changes: ["history-rewritten"] })); watcher.flush(); expect(recorded.publishes).toEqual([{ roots: ["thr_a", "thr_b"] }]); @@ -106,7 +128,7 @@ describe("invalidation for observed threads", () => { describe("candidate filtering and parentage", () => { it("skips a project-mismatched root without a parentage lookup", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], "p1"); + watchRoot(watcher, "thr_root", ["thr_root"], "p1"); await watcher.handleChanged( changed("thr_new", { changes: ["status-changed"], metadata: { projectId: "p2" } }), ); @@ -117,7 +139,7 @@ describe("candidate filtering and parentage", () => { it("resolves a new descendant through a bounded parentage lookup", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged(changed("thr_new", { changes: ["status-changed"] })); watcher.flush(); @@ -131,7 +153,7 @@ describe("candidate filtering and parentage", () => { it("rate-limits parentage lookups for unresolved threads", async () => { const { watcher, recorded } = makeWatcher({ lookupGapMs: 2_000 }); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged(changed("thr_unknown", { changes: ["status-changed"] })); await watcher.handleChanged(changed("thr_unknown", { changes: ["events-appended"] })); expect(recorded.lookups).toEqual(["thr_unknown"]); // one lookup within the gap @@ -143,7 +165,7 @@ describe("candidate filtering and parentage", () => { it("keeps unrelated threads out of the board", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root", "thr_child"], null); + watchRoot(watcher, "thr_root", ["thr_root", "thr_child"], null); await watcher.handleChanged(changed("thr_elsewhere", { changes: ["status-changed"] })); watcher.flush(); expect(recorded.lookups).toEqual(["thr_elsewhere"]); @@ -154,7 +176,7 @@ describe("candidate filtering and parentage", () => { describe("lifecycle events", () => { it("uses the event's parent when provided (no lookup)", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleLifecycle({ id: "thr_new", parentThreadId: "thr_root", @@ -165,9 +187,27 @@ describe("lifecycle events", () => { expect(recorded.publishes).toEqual([{ roots: ["thr_root"] }]); }); + it("reaches the root for a thread created below a parent the board has not seen", async () => { + const { watcher, recorded } = makeWatcher(); + watchRoot(watcher, "thr_root", ["thr_root"], null); + await watcher.handleLifecycle({ id: "thr_grandchild", parentThreadId: "thr_unseen_child" }, "created"); + watcher.flush(); + expect(recorded.lookups).toEqual(["thr_grandchild"]); + expect(recorded.publishes).toEqual([{ roots: ["thr_root"] }]); + }); + + it("never looks up a root thread's ancestors", async () => { + const { watcher, recorded } = makeWatcher(); + watchRoot(watcher, "thr_root", ["thr_root"], null); + await watcher.handleLifecycle({ id: "thr_other_root", parentThreadId: null }, "created"); + watcher.flush(); + expect(recorded.lookups).toEqual([]); + expect(recorded.publishes).toEqual([]); + }); + it("ignores lifecycle events for threads outside the tree", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleLifecycle({ id: "thr_far", parentThreadId: "thr_other" }); watcher.flush(); expect(recorded.publishes).toEqual([]); @@ -177,7 +217,7 @@ describe("lifecycle events", () => { describe("debounce and coalescing", () => { it("coalesces many changes into one publish per window", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); await watcher.handleChanged(changed("thr_root", { changes: ["events-appended"] })); await watcher.handleChanged(changed("thr_root", { changes: ["events-appended"] })); @@ -189,8 +229,8 @@ describe("debounce and coalescing", () => { it("debounces across roots into a single signal", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_a", ["thr_a"], null); - watcher.touch("thr_b", ["thr_b"], null); + watchRoot(watcher, "thr_a", ["thr_a"], null); + watchRoot(watcher, "thr_b", ["thr_b"], null); await watcher.handleChanged(changed("thr_a", { changes: ["status-changed"] })); vi.advanceTimersByTime(200); await watcher.handleChanged(changed("thr_b", { changes: ["status-changed"] })); @@ -201,20 +241,44 @@ describe("debounce and coalescing", () => { }); describe("registry lifecycle", () => { - it("prunes watched roots after the TTL", async () => { - let now = 0; - const { watcher, recorded } = makeWatcher({ now: () => now, watchedTtlMs: 90_000 }); + it("stops watching a root when its last watched read ends", async () => { + const { watcher, recorded } = makeWatcher(); + const first = watchRoot(watcher, "thr_root", ["thr_root"], null); + const second = watchRoot(watcher, "thr_root", ["thr_root"], null); + first.abort(); + await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); + watcher.flush(); + expect(recorded.publishes).toEqual([{ roots: ["thr_root"] }]); // one client still watches + second.abort(); + await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); + watcher.flush(); + expect(recorded.publishes).toEqual([{ roots: ["thr_root"] }]); + expect(watcher.isWatching("thr_root")).toBe(false); + }); + + it("ignores snapshots of roots nobody watches", async () => { + const { watcher, recorded } = makeWatcher(); watcher.touch("thr_root", ["thr_root"], null); - now = 91_000; - watcher.touch("thr_other", ["thr_other"], null); // triggers prune await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); watcher.flush(); - expect(recorded.publishes).toEqual([]); // thr_root was pruned + expect(recorded.publishes).toEqual([]); + }); + + it("stops global signals when the global page's watched read ends", async () => { + const { watcher, recorded } = makeWatcher(); + const global = watchGlobal(watcher); + await watcher.handleChanged(changed("thr_any", { changes: ["status-changed"] })); + watcher.flush(); + expect(recorded.publishes).toEqual([{ roots: [], global: true }]); + global.abort(); + await watcher.handleChanged(changed("thr_any", { changes: ["status-changed"] })); + watcher.flush(); + expect(recorded.publishes).toHaveLength(1); }); it("stops handling events and publishing after dispose", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); watcher.dispose(); await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); watcher.flush(); @@ -223,7 +287,7 @@ describe("registry lifecycle", () => { it("cancels the pending debounce on dispose", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); watcher.dispose(); vi.advanceTimersByTime(1_000); @@ -234,7 +298,7 @@ describe("registry lifecycle", () => { describe("terminal-state reconciliation", () => { it("re-publishes the same payload once, after the flush, for a terminal status change", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged( changed("thr_root", { changes: ["status-changed"], @@ -252,7 +316,7 @@ describe("terminal-state reconciliation", () => { it("treats turn/completed appended events as terminal", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged( changed("thr_root", { changes: ["events-appended"], @@ -277,7 +341,7 @@ describe("terminal-state reconciliation", () => { ]; for (const [kind, expected] of cases) { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleLifecycle({ id: "thr_new", parentThreadId: "thr_root" }, kind); watcher.flush(); vi.advanceTimersByTime(1_500); @@ -288,7 +352,7 @@ describe("terminal-state reconciliation", () => { it("does not schedule a reconciliation for non-terminal flushes", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); watcher.flush(); vi.advanceTimersByTime(10_000); @@ -297,8 +361,8 @@ describe("terminal-state reconciliation", () => { it("replaces — never stacks — the pending reconciliation across terminal flushes", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_a", ["thr_a"], null); - watcher.touch("thr_b", ["thr_b"], null); + watchRoot(watcher, "thr_a", ["thr_a"], null); + watchRoot(watcher, "thr_b", ["thr_b"], null); await watcher.handleChanged( changed("thr_a", { changes: ["status-changed"], metadata: { statusChange: { status: "idle" } } }), ); @@ -318,8 +382,8 @@ describe("terminal-state reconciliation", () => { it("merges a scoped and global terminal flush into a single reconciled payload", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); - watcher.touchGlobal(); + watchRoot(watcher, "thr_root", ["thr_root"], null); + watchGlobal(watcher); await watcher.handleChanged( changed("thr_root", { changes: ["status-changed"], @@ -336,7 +400,7 @@ describe("terminal-state reconciliation", () => { it("cancels the pending reconciliation on dispose", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged( changed("thr_root", { changes: ["status-changed"], metadata: { statusChange: { status: "idle" } } }), ); @@ -356,7 +420,7 @@ describe("terminal-state reconciliation", () => { watcher.flush(); expect(recorded.publishes).toEqual([]); // Later, a non-terminal change must not inherit the earlier terminal flag. - watcher.touch("thr_root", ["thr_root"], null); + watchRoot(watcher, "thr_root", ["thr_root"], null); await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); watcher.flush(); vi.advanceTimersByTime(5_000); @@ -386,7 +450,7 @@ describe("channel contract", () => { describe("global dashboard invalidation", () => { it("invalidates a watched global page for any relevant thread change", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touchGlobal(); + watchGlobal(watcher); await watcher.handleChanged(changed("thr_any", { changes: ["status-changed"] })); watcher.flush(); expect(recorded.publishes).toEqual([{ roots: [], global: true }]); @@ -395,8 +459,8 @@ describe("global dashboard invalidation", () => { it("coalesces scoped and global invalidation in one signal", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touch("thr_root", ["thr_root"], null); - watcher.touchGlobal(); + watchRoot(watcher, "thr_root", ["thr_root"], null); + watchGlobal(watcher); await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); watcher.flush(); expect(recorded.publishes).toEqual([{ roots: ["thr_root"], global: true }]); @@ -404,19 +468,10 @@ describe("global dashboard invalidation", () => { it("does not wake the global page for ignored reasoning deltas", async () => { const { watcher, recorded } = makeWatcher(); - watcher.touchGlobal(); + watchGlobal(watcher); await watcher.handleChanged(changed("thr_root", { changes: ["events-appended"], metadata: { eventTypes: ["item/reasoning/textDelta"] } })); watcher.flush(); expect(recorded.publishes).toEqual([]); }); - it("expires the global watch lease", async () => { - let now = 0; - const { watcher, recorded } = makeWatcher({ now: () => now, watchedTtlMs: 100 }); - watcher.touchGlobal(); - now = 101; - await watcher.handleChanged(changed("thr_root", { changes: ["status-changed"] })); - watcher.flush(); - expect(recorded.publishes).toEqual([]); - }); });