diff --git a/apps/ui/src/app.css b/apps/ui/src/app.css index 853739106..1aafd2738 100644 --- a/apps/ui/src/app.css +++ b/apps/ui/src/app.css @@ -252,6 +252,26 @@ font-variant-numeric: tabular-nums; } +.agent-halt-dot { + width: 6px; + height: 6px; + flex: none; + background: var(--cx-st-error); +} +.agent-halt-dot[data-halt="done"] { + background: var(--cx-st-done); +} +.agent-halt-dot[data-halt="awaiting_user"] { + background: var(--cx-st-waiting); +} + +.agent-waiting-count { + flex: none; + color: var(--cx-st-waiting); + font-size: 10px; + font-variant-numeric: tabular-nums; +} + /* Presence activity string on an agent leaf (T8): a small dim run after the * badge, truncated so a long status never widens the tree row. */ .agent-activity { @@ -1908,6 +1928,14 @@ overflow: hidden; text-overflow: ellipsis; } + +.fleet-waiting-count { + align-self: center; + margin-left: 10px; + color: var(--cx-st-waiting); + font-size: 11px; + white-space: nowrap; +} /* A fleet tab's pane: a compact open-workspace header over the inline home-DM conversation (design: architecture-lineage). */ .fleet-pane { diff --git a/apps/ui/src/components/LeftSidebar.live.test.tsx b/apps/ui/src/components/LeftSidebar.live.test.tsx index 8946f487b..9a14a10a0 100644 --- a/apps/ui/src/components/LeftSidebar.live.test.tsx +++ b/apps/ui/src/components/LeftSidebar.live.test.tsx @@ -2,7 +2,12 @@ import { describe, expect, test } from "bun:test"; import { AccountSchema, AgentAccountSchema, + AgentBoardSchema, + AgentBoardSummarySchema, AgentPresence, + BoardHaltKind, + BoardHaltSchema, + BoardTaskSchema, create, RosterEntrySchema, UserAccountSchema, @@ -168,4 +173,52 @@ describe("LeftSidebar — live roster (T4/T5)", () => { expect(container.querySelector(".tree-empty")).toBeNull(); expect(container.querySelectorAll(".tree-agent").length).toBeGreaterThan(0); }); + + test("An agentBoardChanged event updates the sidebar badge", async () => { + const entry = (waitingOnOperator: number, halt?: BoardHaltKind) => + create(RosterEntrySchema, { + agentAccountId: "acc-cook", + presence: AgentPresence.WORKING, + board: create(AgentBoardSummarySchema, { + taskCount: 1, + waitingOnOperator, + halt: halt ? create(BoardHaltSchema, { kind: halt }) : undefined, + }), + }); + const fake = createFakeComms({ + accounts: [wireUserAccount(CALLER), wireAgentAccount("acc-cook")], + roster: [entry(2)], + }); + const { container } = await mountLive(fake); + const count = () => + container + .querySelector(".tree-agent .agent-waiting-count") + ?.getAttribute("aria-label"); + const leaf = () => container.querySelector(".tree-agent"); + expect(count()).toBe("2 waiting on you"); + expect(leaf()?.querySelector(".agent-halt-dot")).toBeNull(); + fake.setRoster([entry(1, BoardHaltKind.BLOCKED)]); + await fake.emit( + { + case: "agentBoardChanged", + value: { + agentAccountId: "acc-cook", + board: create(AgentBoardSchema, { + agentAccountId: "acc-cook", + halt: create(BoardHaltSchema, { + kind: BoardHaltKind.BLOCKED, + ref: "RIG-9", + }), + tasks: [create(BoardTaskSchema, { key: "a", title: "A" })], + writtenAtUnixMs: 1n, + }), + }, + }, + 1n, + ); + expect( + leaf()?.querySelector(".agent-halt-dot")?.getAttribute("data-halt"), + ).toBe("blocked"); + expect(count()).toBe("1 waiting on you"); + }); }); diff --git a/apps/ui/src/components/LeftSidebar.tsx b/apps/ui/src/components/LeftSidebar.tsx index b53187247..818df82e4 100644 --- a/apps/ui/src/components/LeftSidebar.tsx +++ b/apps/ui/src/components/LeftSidebar.tsx @@ -70,6 +70,29 @@ const AgentLeaf: Component<{ agent: Agent; badge?: number }> = (props) => { + + {(halt) => ( + + )} + + + {(count) => ( + + {count()} + + )} + {props.badge} diff --git a/apps/ui/src/components/RightSidebar.tsx b/apps/ui/src/components/RightSidebar.tsx index 864fcd46c..e52ac2647 100644 --- a/apps/ui/src/components/RightSidebar.tsx +++ b/apps/ui/src/components/RightSidebar.tsx @@ -483,6 +483,11 @@ const FleetPane: Component<{ item: AvatarTabItem }> = (props) => { > Open {a().account.handle}'s workspace + 0}> + + {a().board?.waitingOnOperator} waiting on you + + diff --git a/apps/ui/src/live/adapt.test.ts b/apps/ui/src/live/adapt.test.ts index 16132fbaf..cd8bbdb97 100644 --- a/apps/ui/src/live/adapt.test.ts +++ b/apps/ui/src/live/adapt.test.ts @@ -1,8 +1,11 @@ import { describe, expect, test } from "bun:test"; +import { fromBinary } from "@bufbuild/protobuf"; import type { Ask as WireAsk } from "@compass/client"; import { AccountSchema, AgentAccountSchema, + AgentBoardSchema, + AgentBoardSummarySchema, AgentPlanEntryStatus, AgentPresence, AgentSessionStatusSchema, @@ -10,6 +13,17 @@ import { AskOptionSchema, AskQuestionSchema, AskSchema, + BoardAskSchema, + BoardHaltKind, + BoardHaltSchema, + BoardPrGate, + BoardRefKind, + BoardRefSchema, + BoardSubagentSchema, + BoardTaskSchema, + BoardTaskState, + BoardWaitingOn, + BoardWaitingSchema, ChannelGroupSchema, ChannelGroupVisibility, ChannelKind, @@ -37,6 +51,8 @@ import type { IssueState as DomainIssueState } from "../stub-data"; import { type Account, type Agent, agentTree } from "../stub-data"; import { adaptAccount, + adaptAgentBoard, + adaptAgentBoardSummary, adaptAsk, adaptChannel, adaptChannelGroup, @@ -1043,6 +1059,168 @@ describe("adaptIssue", () => { }); }); +describe("adaptAgentBoard", () => { + test("maps board enums, bigint timestamps, and empty optional strings", () => { + const wire = create(AgentBoardSchema, { + agentAccountId: "acc-cook", + handle: "cook", + focusTaskKey: "task-a", + halt: create(BoardHaltSchema, { + kind: BoardHaltKind.UNSPECIFIED, + ref: "", + reason: "", + }), + tasks: [ + create(BoardTaskSchema, { + key: "task-a", + title: "Ship the board", + primaryRef: "", + state: BoardTaskState.ACTIVE, + note: "", + waiting: [ + create(BoardWaitingSchema, { + on: BoardWaitingOn.OPERATOR, + agentHandle: "", + what: "", + ref: "", + }), + create(BoardWaitingSchema, { on: BoardWaitingOn.UNSPECIFIED }), + ], + subagents: [ + create(BoardSubagentSchema, { + id: "sub-1", + agent: "builder", + description: "", + status: "working", + startedAtUnixMs: 10n, + endedAtUnixMs: 0n, + }), + ], + }), + ], + refs: [ + create(BoardRefSchema, { + ref: "RIG-12", + kind: BoardRefKind.UNSPECIFIED, + title: "", + url: "", + gate: BoardPrGate.UNSPECIFIED, + stackTipRef: "", + owned: true, + staleApproval: true, + mergeQueueState: "", + trackerStatus: "", + }), + ], + asks: [ + create(BoardAskSchema, { + askId: "ask-1", + messageId: "msg-1", + channelId: "ch-1", + ref: "", + header: "Pick a fix", + postedAtUnixMs: 77n, + }), + ], + writtenAtUnixMs: 1234n, + stale: true, + }); + const board = adaptAgentBoard(wire); + expect(board.agentId).toBe("acc-cook"); + expect(board.focusTaskKey).toBe("task-a"); + expect(board.halt).toBeUndefined(); + expect(board.writtenAtUnixMs).toBe(1234); + expect(board.tasks[0]).toMatchObject({ + state: "active", + primaryRef: undefined, + note: undefined, + waiting: [ + { + on: "operator", + agentHandle: undefined, + what: undefined, + ref: undefined, + }, + { on: undefined }, + ], + subagents: [{ startedAtUnixMs: 10, endedAtUnixMs: undefined }], + }); + expect(board.refs[0]).toMatchObject({ + kind: "unknown", + title: undefined, + url: undefined, + gate: undefined, + stackTipRef: undefined, + mergeQueueState: undefined, + trackerStatus: undefined, + }); + expect(board.asks).toEqual([ + { + askId: "ask-1", + messageId: "msg-1", + channelId: "ch-1", + ref: undefined, + header: "Pick a fix", + answered: false, + postedAtUnixMs: 77, + }, + ]); + }); + + test("maps a ref kind from a newer server to unknown", () => { + // Field 2 (kind) = 99: a value this client's enum does not define. + const ref = fromBinary(BoardRefSchema, new Uint8Array([0x10, 0x63])); + const board = adaptAgentBoard(create(AgentBoardSchema, { refs: [ref] })); + expect(board.refs[0]?.kind).toBe("unknown"); + }); + + test("maps wire board summaries", () => { + expect( + adaptAgentBoardSummary( + create(AgentBoardSummarySchema, { + focus: "", + taskCount: 3, + waitingOnOperator: 2, + halt: create(BoardHaltSchema, { + kind: BoardHaltKind.AWAITING_USER, + ref: "RIG-9", + reason: "Need a decision", + }), + writtenAtUnixMs: 100n, + }), + ), + ).toEqual({ + focus: undefined, + taskCount: 3, + waitingOnOperator: 2, + halt: { kind: "awaiting_user", ref: "RIG-9", reason: "Need a decision" }, + writtenAtUnixMs: 100, + stale: false, + }); + }); +}); + +test("adaptRosterEntry carries the board summary", () => { + const [, info] = adaptRosterEntry( + create(RosterEntrySchema, { + agentAccountId: "acc-cook", + board: create(AgentBoardSummarySchema, { + focus: "Review pull request", + taskCount: 2, + waitingOnOperator: 1, + writtenAtUnixMs: 100n, + }), + }), + ); + expect(info.board).toEqual({ + focus: "Review pull request", + taskCount: 2, + waitingOnOperator: 1, + halt: undefined, + writtenAtUnixMs: 100, + stale: false, + }); +}); describe("presenceLifecycle", () => { test("maps each modeled AgentPresence to its AgentState dot", () => { expect(presenceLifecycle(AgentPresence.WORKING)).toBe("working"); diff --git a/apps/ui/src/live/adapt.ts b/apps/ui/src/live/adapt.ts index e86edc185..b700b3cb6 100644 --- a/apps/ui/src/live/adapt.ts +++ b/apps/ui/src/live/adapt.ts @@ -7,6 +7,11 @@ import { AgentPlanEntryStatus, AgentPresence, AgentToolCallStatus, + BoardHaltKind, + BoardPrGate, + BoardRefKind, + BoardTaskState, + BoardWaitingOn, ChannelGroupVisibility, ChannelKind, ChannelMembershipMode, @@ -19,9 +24,17 @@ import { SessionInjectionKind, type Account as WireAccount, type AgentAttribution as WireAgentAttribution, + type AgentBoard as WireAgentBoard, + type AgentBoardSummary as WireAgentBoardSummary, type AgentSessionStatus as WireAgentSessionStatus, type Ask as WireAsk, type AskQuestion as WireAskQuestion, + type BoardAsk as WireBoardAsk, + type BoardHalt as WireBoardHalt, + type BoardRef as WireBoardRef, + type BoardSubagent as WireBoardSubagent, + type BoardTask as WireBoardTask, + type BoardWaiting as WireBoardWaiting, type ChangedStats as WireChangedStats, type Channel as WireChannel, type ChannelGroup as WireChannelGroup, @@ -63,8 +76,17 @@ import type { } from "../session-events"; import type { Account, + AgentBoard, + AgentBoardSummary, AgentState, + BoardAsk, + BoardRef, + BoardSubagent, + BoardTask, + BoardWaiting, AgentAttribution as DomainAgentAttribution, + BoardHalt as DomainBoardHalt, + BoardPrGate as DomainBoardPrGate, ChangedStats as DomainChangedStats, Check as DomainCheck, ChecksSummary as DomainChecksSummary, @@ -114,6 +136,181 @@ const MEMBERSHIP_MODE: Record< [ChannelMembershipMode.TREE]: "tree", }; +const TASK_STATE: Record< + BoardTaskState, + NonNullable | undefined +> = { + [BoardTaskState.UNSPECIFIED]: undefined, + [BoardTaskState.ACTIVE]: "active", + [BoardTaskState.BLOCKED]: "blocked", + [BoardTaskState.WAITING]: "waiting", + [BoardTaskState.PARKED]: "parked", + [BoardTaskState.DONE]: "done", +} satisfies Record | undefined>; + +const HALT_KIND: Record = { + [BoardHaltKind.UNSPECIFIED]: undefined, + [BoardHaltKind.DONE]: "done", + [BoardHaltKind.BLOCKED]: "blocked", + [BoardHaltKind.AWAITING_USER]: "awaiting_user", +} satisfies Record; + +const WAITING_ON: Record< + BoardWaitingOn, + NonNullable | undefined +> = { + [BoardWaitingOn.UNSPECIFIED]: undefined, + [BoardWaitingOn.OPERATOR]: "operator", + [BoardWaitingOn.AGENT]: "agent", + [BoardWaitingOn.EXTERNAL]: "external", +} satisfies Record | undefined>; + +const REF_KIND: Record = { + [BoardRefKind.UNSPECIFIED]: "unknown", + [BoardRefKind.ISSUE]: "issue", + [BoardRefKind.PULL_REQUEST]: "pr", +} satisfies Record; + +const PR_GATE: Record = { + [BoardPrGate.UNSPECIFIED]: undefined, + [BoardPrGate.DRAFT]: "draft", + [BoardPrGate.CHECKS_PENDING]: "checks_pending", + [BoardPrGate.CHECKS_FAILED]: "checks_failed", + [BoardPrGate.CHANGES_REQUESTED]: "changes_requested", + [BoardPrGate.AWAITING_REVIEW]: "awaiting_review", + [BoardPrGate.AWAITING_ENQUEUE]: "awaiting_enqueue", + [BoardPrGate.MERGED]: "merged", + [BoardPrGate.CLOSED]: "closed", + [BoardPrGate.QUEUED]: "queued", + [BoardPrGate.QUEUE_FAILED]: "queue_failed", + [BoardPrGate.QUEUE_REMOVED]: "queue_removed", + [BoardPrGate.CONFLICTS]: "conflicts", +} satisfies Record; + +function optionalText(value: string): string | undefined { + return value || undefined; +} + +function adaptBoardHalt( + w: WireBoardHalt | undefined, +): DomainBoardHalt | undefined { + if (!w) return undefined; + const kind = HALT_KIND[w.kind]; + if (!kind) return undefined; + return { kind, ref: optionalText(w.ref), reason: optionalText(w.reason) }; +} + +function adaptBoardWaiting(w: WireBoardWaiting): BoardWaiting { + return { + on: WAITING_ON[w.on], + agentHandle: optionalText(w.agentHandle), + what: optionalText(w.what), + ref: optionalText(w.ref), + blocks: w.blocks, + }; +} + +function adaptBoardSubagent(w: WireBoardSubagent): BoardSubagent { + return { + id: w.id, + agent: w.agent, + description: w.description, + status: w.status, + startedAtUnixMs: Number(w.startedAtUnixMs), + endedAtUnixMs: w.endedAtUnixMs === 0n ? undefined : Number(w.endedAtUnixMs), + }; +} + +function adaptBoardTask(w: WireBoardTask): BoardTask { + return { + key: w.key, + title: w.title, + primaryRef: optionalText(w.primaryRef), + refs: w.refs, + state: TASK_STATE[w.state], + note: optionalText(w.note), + waiting: w.waiting.map(adaptBoardWaiting), + subagents: w.subagents.map(adaptBoardSubagent), + }; +} + +function adaptBoardRef(w: WireBoardRef): BoardRef { + return { + ref: w.ref, + kind: REF_KIND[w.kind] ?? "unknown", + title: optionalText(w.title), + url: optionalText(w.url), + gate: PR_GATE[w.gate], + draft: w.draft, + checksState: optionalText(w.checksState), + stackTipRef: optionalText(w.stackTipRef), + stackPosition: w.stackPosition || undefined, + owned: w.owned, + staleApproval: w.staleApproval, + mergeQueueState: optionalText(w.mergeQueueState), + trackerStatus: optionalText(w.trackerStatus), + }; +} + +function adaptBoardAsk(w: WireBoardAsk): BoardAsk { + return { + askId: w.askId, + messageId: w.messageId, + channelId: w.channelId, + ref: optionalText(w.ref), + header: w.header, + answered: w.answered, + postedAtUnixMs: Number(w.postedAtUnixMs), + }; +} + +export function adaptAgentBoardSummary( + w: WireAgentBoardSummary, +): AgentBoardSummary { + return { + focus: optionalText(w.focus), + taskCount: w.taskCount, + waitingOnOperator: w.waitingOnOperator, + halt: adaptBoardHalt(w.halt), + writtenAtUnixMs: Number(w.writtenAtUnixMs), + stale: w.stale, + }; +} + +export function adaptAgentBoard(w: WireAgentBoard): AgentBoard { + return { + agentId: w.agentAccountId, + handle: w.handle, + focusTaskKey: optionalText(w.focusTaskKey), + halt: adaptBoardHalt(w.halt), + tasks: w.tasks.map(adaptBoardTask), + refs: w.refs.map(adaptBoardRef), + untrackedRefs: w.untrackedRefs, + asks: w.asks.map(adaptBoardAsk), + writtenAtUnixMs: Number(w.writtenAtUnixMs), + stale: w.stale, + }; +} + +/** Fold an AgentBoardChanged board into the roster summary. The event carries + * no derived refs, so the server's waiting-on-you count stands until a re-read. */ +export function applyBoardEvent( + prev: AgentBoardSummary | undefined, + board: AgentBoard, +): AgentBoardSummary { + const focusTask = board.tasks.find((task) => task.key === board.focusTaskKey); + return { + focus: focusTask + ? [focusTask.primaryRef, focusTask.title].filter(Boolean).join(" ") + : undefined, + taskCount: board.tasks.length, + waitingOnOperator: prev?.waitingOnOperator ?? 0, + halt: board.halt, + writtenAtUnixMs: board.writtenAtUnixMs, + stale: board.stale, + }; +} + /** Map a wire PinnedEntry to the domain one: the board pointer (message id + * position) the strip renders. The wire's audit extras (`pinnedAtUnixMs`, * `pinnedByAccountId`) are not in the domain contract — the strip renders the @@ -555,6 +752,7 @@ export function presenceLifecycle(p: AgentPresence): AgentState | undefined { export interface AgentPresenceInfo { readonly lifecycle?: AgentState; readonly activity?: string; + readonly board?: AgentBoardSummary; } /** Map one wire `RosterEntry` to its presence-map entry: `[agentAccountId, @@ -573,6 +771,7 @@ export function adaptRosterEntry( { lifecycle: presenceLifecycle(w.presence), activity: w.activity || undefined, + board: w.board ? adaptAgentBoardSummary(w.board) : undefined, }, ]; } diff --git a/apps/ui/src/live/comms-fake.ts b/apps/ui/src/live/comms-fake.ts index ff453fac4..37d5a0c68 100644 --- a/apps/ui/src/live/comms-fake.ts +++ b/apps/ui/src/live/comms-fake.ts @@ -61,6 +61,8 @@ export interface FakeComms { emit: (payload: unknown, seq: bigint) => Promise; /** Close the scripted stream so `runCommsStream` stops tailing it. */ close: () => void; + /** Replace what later GetRoster calls serve, as a server write would. */ + setRoster: (entries: readonly unknown[]) => void; /** Reject the next PostMessage with `error` (one-shot) — the composer's * keep-the-user's-text path. */ failNextPost: (error: Error) => void; @@ -116,6 +118,7 @@ export function createFakeComms(snapshot: FakeCommsSnapshot = {}): FakeComms { const askResponses: RecordedAskResponse[] = []; let postFailure: Error | undefined; let askFailure: Error | undefined; + let roster = snapshot.roster ?? []; // A held-in-flight RespondToAsk: the promise the RPC parks on until the // test's gate settles it. One at a time — the store issues at most one // respond per ask, and a test gates exactly the one it is interleaving on. @@ -175,7 +178,7 @@ export function createFakeComms(snapshot: FakeCommsSnapshot = {}): FakeComms { listAccounts: async () => ({ accounts: snapshot.accounts ?? [] }), listChannelGroups: async () => ({ groups: snapshot.channelGroups ?? [] }), listChannels: async () => ({ channels: snapshot.channels ?? [] }), - getRoster: async () => ({ entries: snapshot.roster ?? [] }), + getRoster: async () => ({ entries: roster }), listTopics: async (req: { channelId: string; includeArchived: boolean; @@ -329,6 +332,9 @@ export function createFakeComms(snapshot: FakeCommsSnapshot = {}): FakeComms { wake?.(); wake = undefined; }, + setRoster: (entries) => { + roster = entries; + }, failNextPost: (error) => { postFailure = error; }, diff --git a/apps/ui/src/live/comms-state.test.ts b/apps/ui/src/live/comms-state.test.ts index c00f728da..9088b2448 100644 --- a/apps/ui/src/live/comms-state.test.ts +++ b/apps/ui/src/live/comms-state.test.ts @@ -11,6 +11,8 @@ import { UserAccountSchema, } from "@compass/client"; import type { Account, Channel, ChannelGroup, Message } from "../comms-stub"; +import type { AgentBoard } from "../stub-data"; +import type { AgentPresenceInfo } from "./adapt"; import { applyEvent, type CommsEvent, @@ -336,3 +338,77 @@ describe("applyEvent — presence upsert", () => { expect(next).not.toBe(base); }); }); + +describe("applyEvent — board update", () => { + const board: AgentBoard = { + agentId: "acc-cook", + handle: "cook", + focusTaskKey: "task-a", + halt: { kind: "blocked", ref: "RIG-9" }, + tasks: [ + { + key: "task-a", + title: "Write adapter", + primaryRef: "RIG-12", + refs: [], + state: "active", + waiting: [], + subagents: [], + }, + { key: "task-b", title: "Docs", refs: [], waiting: [], subagents: [] }, + ], + refs: [], + untrackedRefs: [], + asks: [], + writtenAtUnixMs: 20, + stale: false, + }; + const presence: AgentPresenceInfo = { + lifecycle: "working", + activity: "writing tests", + board: { + taskCount: 1, + waitingOnOperator: 3, + writtenAtUnixMs: 5, + stale: true, + }, + }; + + test("updates the written summary and keeps the server's waiting count", () => { + const seeded = { + ...baseState(), + presence: new Map([["acc-cook", presence]]), + }; + const next = applyEvent(seeded, { + kind: "boardChanged", + accountId: "acc-cook", + board, + }); + expect(next.presence.get("acc-cook")).toEqual({ + lifecycle: "working", + activity: "writing tests", + board: { + focus: "RIG-12 Write adapter", + taskCount: 2, + waitingOnOperator: 3, + halt: { kind: "blocked", ref: "RIG-9" }, + writtenAtUnixMs: 20, + stale: false, + }, + }); + expect(next.presence).not.toBe(seeded.presence); + }); + + test("a presence delta keeps the board summary", () => { + const seeded = { + ...baseState(), + presence: new Map([["acc-cook", presence]]), + }; + const next = applyEvent(seeded, { + kind: "presenceChanged", + accountId: "acc-cook", + info: { lifecycle: "waiting", activity: "asked" }, + }); + expect(next.presence.get("acc-cook")?.board).toBe(presence.board); + }); +}); diff --git a/apps/ui/src/live/comms-state.ts b/apps/ui/src/live/comms-state.ts index 0db6fd8c4..071707518 100644 --- a/apps/ui/src/live/comms-state.ts +++ b/apps/ui/src/live/comms-state.ts @@ -17,6 +17,7 @@ import type { Message, Topic, } from "../comms-stub"; +import type { AgentBoard, AgentBoardSummary } from "../stub-data"; import { type AgentPresenceInfo, adaptAccount, @@ -25,6 +26,7 @@ import { adaptRosterEntry, adaptTopic, agentHomeChannelIds, + applyBoardEvent, } from "./adapt"; /** Maps a wire Message to the domain Message. Injected so this engine stays @@ -190,6 +192,16 @@ export type CommsEvent = readonly kind: "presenceChanged"; readonly accountId: string; readonly info: AgentPresenceInfo; + } + | { + readonly kind: "boardChanged"; + readonly accountId: string; + readonly board: AgentBoard; + } + | { + readonly kind: "boardSummary"; + readonly accountId: string; + readonly board: AgentBoardSummary | undefined; }; /** Apply one decoded event to the state, returning the next state. Pure: upserts @@ -235,7 +247,27 @@ export function applyEvent(state: CommsState, event: CommsEvent): CommsState { // collections keeping identity (structural sharing). case "presenceChanged": { const presence = new Map(state.presence); - presence.set(event.accountId, event.info); + presence.set(event.accountId, { + ...presence.get(event.accountId), + ...event.info, + }); + return { ...state, presence }; + } + case "boardChanged": { + const presence = new Map(state.presence); + const prev = presence.get(event.accountId); + presence.set(event.accountId, { + ...prev, + board: applyBoardEvent(prev?.board, event.board), + }); + return { ...state, presence }; + } + case "boardSummary": { + const presence = new Map(state.presence); + presence.set(event.accountId, { + ...presence.get(event.accountId), + board: event.board, + }); return { ...state, presence }; } } diff --git a/apps/ui/src/live/stream.test.ts b/apps/ui/src/live/stream.test.ts index b572b25c8..6744a4a9b 100644 --- a/apps/ui/src/live/stream.test.ts +++ b/apps/ui/src/live/stream.test.ts @@ -2,6 +2,8 @@ import { describe, expect, test } from "bun:test"; import { AccountSchema, AgentAccountSchema, + AgentBoardSchema, + AgentBoardSummarySchema, AgentPresence, ChannelGroupSchema, ChannelKind, @@ -11,6 +13,7 @@ import { RosterEntrySchema, UserAccountSchema, type Account as WireAccount, + type AgentBoard as WireAgentBoard, type Channel as WireChannel, type ChannelGroup as WireChannelGroup, type RosterEntry as WireRosterEntry, @@ -105,6 +108,13 @@ type FakePayload = readonly activity: string; }; } + | { + readonly case: "agentBoardChanged"; + readonly value: { + readonly agentAccountId: string; + readonly board?: WireAgentBoard; + }; + } | { readonly case: "resyncRequired"; readonly value: Record } | { readonly case: undefined; readonly value?: undefined }; interface FakeResponse { @@ -217,6 +227,12 @@ interface FakeConfig { readonly subscribeScripts?: Array<() => AsyncGenerator>; /** Roster entries GetRoster returns for the snapshot presence seed. */ readonly roster?: WireRosterEntry[]; + /** When set, every GetRoster after the first returns these entries instead. */ + readonly rosterAfter?: WireRosterEntry[]; + /** When set, every GetRoster after the first rejects with this error. */ + readonly rosterAfterError?: Error; + /** Runs inside every GetRoster after the first, before it returns. */ + readonly onRosterAfter?: () => void; /** When set, GetRoster rejects with this error instead of returning — the * read-failure path (the snapshot must abort, not seed an empty map). */ readonly getRosterError?: Error; @@ -267,7 +283,11 @@ function createFakeClient( getRoster: async ({ scope }: { scope: number }) => { calls.getRosterScopes.push(scope); if (config.getRosterError) throw config.getRosterError; - return { entries: config.roster ?? [] }; + if (calls.getRosterScopes.length === 1) + return { entries: config.roster ?? [] }; + config.onRosterAfter?.(); + if (config.rosterAfterError) throw config.rosterAfterError; + return { entries: config.rosterAfter ?? config.roster ?? [] }; }, listTopics: async (req: { channelId: string }) => ({ topics: config.topicsByChannel?.[req.channelId] ?? [], @@ -1133,3 +1153,97 @@ describe("runCommsStream — presence seed + tail", () => { expect(states).toHaveLength(0); }); }); + +describe("runCommsStream — board summary re-read", () => { + const entry = (waitingOnOperator: number) => + create(RosterEntrySchema, { + agentAccountId: "acc-cook", + presence: AgentPresence.WORKING, + board: create(AgentBoardSummarySchema, { waitingOnOperator }), + }); + const boardChanged = (seq: bigint): FakeResponse => ({ + seq, + atUnixMs: 0n, + instanceEpoch: 1n, + snapshotSeq: 0n, + payload: { + case: "agentBoardChanged", + value: { + agentAccountId: "acc-cook", + board: create(AgentBoardSchema, { agentAccountId: "acc-cook" }), + }, + }, + }); + const run = async ( + config: Omit, + controller = new AbortController(), + ) => { + const { client, calls } = createFakeClient(controller, { + accounts: [wireUserAccount(CALLER)], + ...config, + subscribeScripts: [ + async function* () { + yield boundary(100n, 1n); + yield boundary(0n, 2n); + yield boardChanged(3n); + controller.abort(); + }, + ], + }); + const states: CommsState[] = []; + const errors: unknown[] = []; + await runCommsStream({ + client, + callerId: CALLER, + mapMessage, + onState: (s) => states.push(s), + onError: (e) => errors.push(e), + signal: controller.signal, + }); + const waiting = states.at(-1)?.presence.get("acc-cook") + ?.board?.waitingOnOperator; + return { calls, waiting, errors }; + }; + + test("a re-read that resolves after teardown applies nothing", async () => { + const controller = new AbortController(); + const { calls, waiting } = await run( + { + roster: [entry(1)], + rosterAfter: [entry(0)], + onRosterAfter: () => controller.abort(), + }, + controller, + ); + expect(calls.getRosterScopes).toEqual([2, 2]); + expect(waiting).toBe(1); + }); + + test("an agentBoardChanged re-reads GetRoster for the server's waiting count", async () => { + const { calls, waiting } = await run({ + roster: [entry(1)], + rosterAfter: [entry(0)], + }); + expect(calls.getRosterScopes).toEqual([2, 2]); + expect(waiting).toBe(0); + }); + + test("a failed re-read reports the error and keeps the event's summary", async () => { + const boom = new Error("getRoster boom"); + const { waiting, errors } = await run({ + roster: [entry(1)], + rosterAfterError: boom, + }); + expect(errors).toEqual([boom]); + expect(waiting).toBe(1); + }); + + test("an agent the re-read no longer lists waits on no one", async () => { + const { calls, waiting } = await run({ + roster: [entry(2)], + rosterAfter: [], + }); + expect(calls.getRosterScopes).toEqual([2, 2]); + expect(waiting).toBe(0); + }); +}); diff --git a/apps/ui/src/live/stream.ts b/apps/ui/src/live/stream.ts index 7e95a1590..100bcd36f 100644 --- a/apps/ui/src/live/stream.ts +++ b/apps/ui/src/live/stream.ts @@ -5,11 +5,14 @@ import type { CommsClient, SubscribeCommsResponse } from "@compass/client"; import { RosterScope } from "@compass/client"; +import type { AgentBoardSummary } from "../stub-data"; import { type AgentPresenceInfo, adaptAccount, + adaptAgentBoard, adaptChannel, adaptChannelGroup, + adaptRosterEntry, adaptTopic, agentHomeChannelIds, presenceLifecycle, @@ -32,8 +35,8 @@ const SNAPSHOT_MESSAGE_PAGE = 200; /** What the driver needs to run: the comms client, who the caller is (scopes * membership derivation), the injected message mapper, and the sink for each * new reduced state. `signal` cancels the whole run (component unmount / app - * teardown); `onError` observes a non-fatal stream error before the driver - * reconnects (fatal = aborted). */ + * teardown); `onError` observes a non-fatal error: a stream error before the + * driver reconnects, or a failed board re-read (fatal = aborted). */ export interface CommsStreamOptions { readonly client: CommsClient; readonly callerId: string; @@ -42,8 +45,8 @@ export interface CommsStreamOptions { * once per applied tail event. The store sets its signals from this. */ readonly onState: (state: CommsState) => void; readonly signal?: AbortSignal; - /** Observed on a stream error before reconnect; the run continues (retries - * the subscribe). Not called on a clean abort. */ + /** Observed on a stream error before reconnect, or on a failed board + * re-read that leaves the stream running. Not called on a clean abort. */ readonly onError?: (error: unknown) => void; } @@ -103,6 +106,29 @@ export async function fetchSnapshot( }; } +/** Re-read the board summary an AgentBoardChanged cannot carry; an agent not in + * the caller's OWNER roster waits on no one. Undefined on failure. */ +async function readBoardSummary( + client: CommsClient, + accountId: string, + current: AgentBoardSummary | undefined, + signal: AbortSignal | undefined, + onError: ((error: unknown) => void) | undefined, +): Promise<{ readonly board: AgentBoardSummary | undefined } | undefined> { + try { + const resp = await client.getRoster( + { scope: RosterScope.OWNER }, + { signal }, + ); + const entry = resp.entries.find((e) => e.agentAccountId === accountId); + if (entry) return { board: adaptRosterEntry(entry)[1].board }; + return { board: current && { ...current, waitingOnOperator: 0 } }; + } catch (error) { + if (!signal?.aborted) onError?.(error); + return undefined; + } +} + /** Page one channel's messages fully at the snapshot boundary. ListMessages is * newest-first, paged backward by `beforeMessageId` (exclusive); loop until a * page is short. Returns wire messages (the caller maps them). */ @@ -152,6 +178,13 @@ function decodeEvent( mapMessage: MapMessage, ): CommsEvent | null { switch (payload.case) { + case "agentBoardChanged": + if (!payload.value.board) return null; + return { + kind: "boardChanged", + accountId: payload.value.agentAccountId, + board: adaptAgentBoard(payload.value.board), + }; case "messagePosted": case "messageUpdated": { // Singular message fields carry proto3 presence (X | undefined); a @@ -334,6 +367,23 @@ export async function runCommsStream(opts: CommsStreamOptions): Promise { state = applyEvent(state, event); onState(state); } + if (event?.kind === "boardChanged") { + const read = await readBoardSummary( + client, + event.accountId, + state.presence.get(event.accountId)?.board, + signal, + onError, + ); + if (read && !signal?.aborted) { + state = applyEvent(state, { + kind: "boardSummary", + accountId: event.accountId, + board: read.board, + }); + onState(state); + } + } } // A clean close that made no forward tail progress — zero responses, // only the snapshot boundary, or pure seq replays — is a spin a tight diff --git a/apps/ui/src/roster.test.ts b/apps/ui/src/roster.test.ts index bd590bb94..dafff4afe 100644 --- a/apps/ui/src/roster.test.ts +++ b/apps/ui/src/roster.test.ts @@ -100,6 +100,23 @@ describe("joinAgents", () => { // The join is UI-agnostic about terminals: a live agent has none. expect(agents[0]?.terminals).toEqual([]); }); + test("joins a board summary from roster presence", () => { + const board = { + focus: "Write adapter", + taskCount: 2, + waitingOnOperator: 1, + writtenAtUnixMs: 100, + stale: false, + }; + const [agent] = joinAgents( + [agentAccount("acc-cook")], + new Map([["acc-cook", { board }]]), + new Map(), + emptySessions, + emptyLastOpened, + ); + expect(agent?.board).toEqual(board); + }); test("a presence-map MISS projects lifecycle 'stopped' (R2/DL-194)", () => { // An account present in `accounts` but absent from the presence seed — a diff --git a/apps/ui/src/roster.ts b/apps/ui/src/roster.ts index 7100dee2f..1b725b041 100644 --- a/apps/ui/src/roster.ts +++ b/apps/ui/src/roster.ts @@ -37,6 +37,7 @@ export function joinAgents( lifecycle, activity: info?.activity, runtime: runtime.get(account.id), + board: info?.board, terminals: [], }); } diff --git a/apps/ui/src/stub-data.ts b/apps/ui/src/stub-data.ts index 4a3587001..7b71b96ad 100644 --- a/apps/ui/src/stub-data.ts +++ b/apps/ui/src/stub-data.ts @@ -344,6 +344,102 @@ export interface Account { parentAgentId?: string; } +export interface BoardHalt { + kind: "done" | "blocked" | "awaiting_user"; + ref?: string; + reason?: string; +} + +export interface AgentBoardSummary { + focus?: string; + taskCount: number; + waitingOnOperator: number; + halt?: BoardHalt; + writtenAtUnixMs: number; + stale: boolean; +} + +export interface BoardWaiting { + on?: "operator" | "agent" | "external"; + agentHandle?: string; + what?: string; + ref?: string; + blocks: string[]; +} + +export interface BoardSubagent { + id: string; + agent: string; + description: string; + status: string; + startedAtUnixMs: number; + endedAtUnixMs?: number; +} + +export type BoardPrGate = + | "draft" + | "checks_pending" + | "checks_failed" + | "changes_requested" + | "awaiting_review" + | "awaiting_enqueue" + | "merged" + | "closed" + | "queued" + | "queue_failed" + | "queue_removed" + | "conflicts"; + +export interface BoardRef { + ref: string; + kind: "issue" | "pr" | "unknown"; + title?: string; + url?: string; + gate?: BoardPrGate; + draft: boolean; + checksState?: string; + stackTipRef?: string; + stackPosition?: number; + owned: boolean; + staleApproval: boolean; + mergeQueueState?: string; + trackerStatus?: string; +} + +export interface BoardTask { + key: string; + title: string; + primaryRef?: string; + refs: string[]; + state?: "active" | "blocked" | "waiting" | "parked" | "done"; + note?: string; + waiting: BoardWaiting[]; + subagents: BoardSubagent[]; +} + +export interface BoardAsk { + askId: string; + messageId: string; + channelId: string; + ref?: string; + header: string; + answered: boolean; + postedAtUnixMs: number; +} + +export interface AgentBoard { + agentId: string; + handle: string; + focusTaskKey?: string; + halt?: BoardHalt; + tasks: BoardTask[]; + refs: BoardRef[]; + untrackedRefs: string[]; + asks: BoardAsk[]; + writtenAtUnixMs: number; + stale: boolean; +} + /** The agent's ephemeral lifecycle — SubscribeEvents.AgentSessionStatus.state * (`compass.proto:126-129`), keyed by account id. Absent = created but no * session has run. This is the ONLY agent-object field SubscribeEvents feeds. */ @@ -375,6 +471,7 @@ export interface Agent { model?: string; /** UI-only. Optional for a live agent. */ cwd?: string; + board?: AgentBoardSummary; /** Fixture-only. */ terminals: Terminal[]; } diff --git a/packages/compass-client/src/index.ts b/packages/compass-client/src/index.ts index 306d25268..476b6bb4b 100644 --- a/packages/compass-client/src/index.ts +++ b/packages/compass-client/src/index.ts @@ -424,10 +424,18 @@ export { create } from "@bufbuild/protobuf"; export type { Account, AgentAccount, + AgentBoard, + AgentBoardSummary, Ask, AskOption, AskQuestion, AskQuestionAnswer, + BoardAsk, + BoardHalt, + BoardRef, + BoardSubagent, + BoardTask, + BoardWaiting, Channel, ChannelGroup, ListTopicsRequest, @@ -447,11 +455,24 @@ export type { export { AccountSchema, AgentAccountSchema, + AgentBoardSchema, + AgentBoardSummarySchema, AgentPresence, AskOptionSchema, AskQuestionAnswerSchema, AskQuestionSchema, AskSchema, + BoardAskSchema, + BoardHaltKind, + BoardHaltSchema, + BoardPrGate, + BoardRefKind, + BoardRefSchema, + BoardSubagentSchema, + BoardTaskSchema, + BoardTaskState, + BoardWaitingOn, + BoardWaitingSchema, ChannelGroupSchema, ChannelGroupVisibility, ChannelKind,