diff --git a/CHANGELOG.md b/CHANGELOG.md index 21640861..0945d765 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Added +- **The A2A switchboard — the org chart as pair cards.** `/a2a`'s desktop default is now a grid of agent-pair cards (every pair the permission matrix allows — 70 pairs across cell/PM-chain/board/cross sections, derived statically from `can_a2a_direct` at import time), each lighting up when either side messages the other (45s CSS fade, driven purely by the live `a2a.message` frames — A2A only, never verbs, by CEO ruling). Clicking a card opens the existing transcript + chime-in drawer; never-talked pairs render dimmed with an explicit empty state; the v1 list stays as the mobile/compact fallback. Backed by one CEO-gated `GET /a2a/chat/admin/pairs` route joining the static matrix against conversations in a single bulk query. +- **Secretary full task access; PM lighter editing — and a closed over-permission hole.** The Secretary's CEO-gated `edit` directive now covers the full content surface (title/description/AC/priority/team/complexity/nature plus claim-aware reassignment through the real reassign paths), and `read_task` returns full detail. Scouting the PM side found `has_higher_perms` gave PM identities UNRESTRICTED admin on `PATCH /tasks/{id}` (ASSIGN is not team-scoped) — now cell PMs hard-403 outside their team and both PM roles are capped to the content allowlist with zero status changes via that surface; CEO/Board/Auditor keep full admin. + - **Prompter memory — intake remembers the task history.** An intake session's prompt now carries a compact chronological digest of the scoped project's recent tasks (per-project for MegaTask scopes; hard-capped at ~1,000 tokens worst case, typically ~300), and the interviewer gains a `search_past_tasks` tool (bounded, both runtimes share one implementation) to check precedent mid-conversation — so a new task can be described and sequenced against what actually happened before. Informational only: the sequencing analyzer keeps ownership of ordering. - **A2A live view — watch the fleet talk, and chime in.** New panel page (`/a2a`): live conversation list + transcript, updated in real time via a new `A2A_MESSAGE_SENT` event fanned through the existing `/ws/system` bridge (frames carry capped excerpts; full bodies stay on REST). The CEO can reply into any task-linked conversation as themselves. **Agent→CEO communication is reply-only and hard-budgeted in code**: no agent may initiate toward the CEO (stateless matrix block), and inside a conversation the CEO has posted in, each agent may send at most one message per CEO message (per-conversation, per-agent — 1:1:1 with multiple agents), with a rejection envelope that says to wait rather than retry. CEO→agent stays unrestricted — the one asymmetric rule in the matrix. @@ -48,6 +51,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Fixed +- **MegaTask intakes get their architectural-conventions block.** `_resolve_intake_ambient` threaded the multi-project `project_ids` scope to the history digest but not to the conventions resolver, so a MegaTask intake saw no conventions ambient even with the flag on. Both sub-resolvers now share one id→project resolution path and cover all three scopes; regression test pins the threading to both. + - **MegaTask root-subtasks can complete through the Main-PM path.** `_main_pm_complete_guard` and `TaskService.escalate_to_ceo` refused ANY parented task as "not a root" — but a batch root-subtask is parented (the umbrella) BY DESIGN while carrying its own project/branch/PR. Both sites now consult `is_batch_root_subtask` (the single-source identity predicate the other exemption sites already use), so `complete` → CEO escalation works for batch roots while plain subtasks stay refused. Found by e2e scenario 4 on its first run; live root-subtasks previously needed CEO god-mode to close. - **Every `agent.spawned` audit row names its dispatcher.** A rogue spawner could not be identified live — the audit row carried container/model but not which of the ~27 dispatch loops launched it. `spawn_agent` now takes `spawned_by`, stamps it into the `agent.spawned` / `agent.spawn_failed` details (`"unspecified"` when absent so audit queries never miss the field), every call site passes its loop name, and a whole-package AST sweep test fails any future caller that omits it. diff --git a/docs/map/_front.md b/docs/map/_front.md index 6e85aa36..526e3d77 100644 --- a/docs/map/_front.md +++ b/docs/map/_front.md @@ -612,3 +612,9 @@ Backend: `EventType.A2A_MESSAGE_SENT` published from `A2AService.send` (excerpt- ## Delta 2026-07-03 (3) — prompter memory (branch `feat/wave-2b`, SDD/Sonnet 5, reviewed) `TaskService.list_recent_for_project` (recency = coalesce(completed,updated,created)); pure digest builders in `prompter.py` (15 lines/project, 70-char titles, 4000-char total cap); orchestrator `_resolve_history_digest_ambient` (+`_resolve_intake_ambient` merge, best-effort/non-blocking) injected at `_spawn_intake_container` → `_generate_composed_prompt(ambient=…)`; `GET /prompter/live/{id}/search-tasks` (session-liveness = trust boundary, q 2–200, limit ≤10); `query_past_tasks`/`format_search_results` shared by the grok MCP tool AND the Claude SDK in-process tool (full parity, one implementation). FLAGGED pre-existing gap (untouched): `_resolve_conventions_ambient` doesn't cover the MegaTask `project_ids` scope — conventions ambient absent on MegaTask intakes. + +--- +## Delta 2026-07-03 (4) — switchboard + task access (branch `feat/wave-2c`, SDD/Sonnet 5, reviewed) + +1. **Switchboard**: `A2A_ALLOWED_PAIRS` (agents_config, import-time; 70 pairs: 15×3 cells, 6 pm-chain, 3 board, 16 cross), `A2AService.list_admin_pairs` (bulk tuple_ IN join, latest conversation per pair), `GET /a2a/chat/admin/pairs` (CEO-gated); panel `a2a-switchboard{,-utils,-pair-card}` (pure lighting utils, rAF/CSS 45s fade, no timers), /a2a defaults to switchboard w/ v1-list toggle. FLAGGED pre-existing: `agent-utils.ts` static maps miss the per-cell pr-reviewer slugs. +2. **Task access**: Secretary `_EDITABLE_TASK_FIELDS` full content surface + enum coercion + claim-aware reassign (`reassign_active_claim` when claimed/in_progress); `read_task` full detail (progress bounded 50). PM lighter: `_pm_editor_scope`/`_enforce_pm_lighter_fields` in routes/tasks.py — closed the pre-existing PM-unrestricted-admin hole (cross-team cell PM hard-403; content allowlist; zero status via PATCH). submit_directive tool docs never mentioned `edit` (fixed — undiscoverable). diff --git a/panel/src/app/(dashboard)/a2a/__tests__/page.test.tsx b/panel/src/app/(dashboard)/a2a/__tests__/page.test.tsx index 1bcb2707..dfe0244a 100644 --- a/panel/src/app/(dashboard)/a2a/__tests__/page.test.tsx +++ b/panel/src/app/(dashboard)/a2a/__tests__/page.test.tsx @@ -1,21 +1,28 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; -import { render, screen } from "@testing-library/react"; -import type { AdminConversationSummary, A2AChatMessage } from "@/lib/api/a2a"; +import { render, screen, fireEvent } from "@testing-library/react"; +import type { + AdminConversationSummary, + AdminPairSummary, + A2AChatMessage, +} from "@/lib/api/a2a"; const { useA2AConversations, useA2AMessages, + useA2AAdminPairs, useA2ALiveStream, invalidateQueries, a2aLiveKeys, } = vi.hoisted(() => ({ useA2AConversations: vi.fn(), useA2AMessages: vi.fn(), + useA2AAdminPairs: vi.fn(), useA2ALiveStream: vi.fn(), invalidateQueries: vi.fn(), a2aLiveKeys: { all: ["a2a-live"] as const, conversations: ["a2a-live", "conversations"] as const, + pairs: ["a2a-live", "pairs"] as const, messages: (conversationId: string) => ["a2a-live", "messages", conversationId] as const, }, @@ -30,6 +37,7 @@ vi.mock("@/hooks/use-a2a-live", () => ({ a2aLiveKeys, useA2AConversations, useA2AMessages, + useA2AAdminPairs, useReplyAsCeo: () => ({ mutate: vi.fn(), isPending: false }), })); @@ -74,6 +82,22 @@ function buildConversation( }; } +function buildPair(overrides: Partial = {}): AdminPairSummary { + return { + agent_a: "be-dev-1", + role_a: "developer", + team_a: "backend", + agent_b: "be-qa", + role_b: "qa", + team_b: "backend", + group_key: "cell-backend", + conversation_id: "conv-1", + last_message_at: "2026-07-02T09:00:00Z", + message_count: 2, + ...overrides, + }; +} + function buildMessage(): A2AChatMessage { return { id: "m1", @@ -103,8 +127,14 @@ describe("A2APage", () => { isLoading: false, refetch: vi.fn(), }); + useA2AAdminPairs.mockReturnValue({ + data: { items: [], total: 0 }, + isLoading: false, + refetch: vi.fn(), + }); useA2ALiveStream.mockReturnValue({ lastMessage: null, + a2aMessages: [], isConnected: true, }); }); @@ -150,7 +180,7 @@ describe("A2APage", () => { ).toBeInTheDocument(); }); - it("invalidates conversations + selected messages on a matching a2a.message frame", () => { + it("invalidates conversations + pairs + selected messages on a matching a2a.message frame", () => { useA2ALiveStream.mockReturnValue({ lastMessage: { type: "a2a.message", @@ -161,33 +191,69 @@ describe("A2APage", () => { body_excerpt: "capped", timestamp: "2026-07-02T10:00:00Z", }, + a2aMessages: [], isConnected: true, }); render(); expect(invalidateQueries).toHaveBeenCalledWith({ queryKey: a2aLiveKeys.conversations, }); + expect(invalidateQueries).toHaveBeenCalledWith({ + queryKey: a2aLiveKeys.pairs, + }); expect(invalidateQueries).toHaveBeenCalledWith({ queryKey: a2aLiveKeys.messages("conv-1"), }); }); - it("only invalidates the conversation list for frames of other conversations", () => { + it("only invalidates the conversation + pair lists for frames of other conversations", () => { useA2ALiveStream.mockReturnValue({ lastMessage: { type: "a2a.message", conversation_id: "conv-other", timestamp: "2026-07-02T10:00:00Z", }, + a2aMessages: [], isConnected: false, }); render(); expect(invalidateQueries).toHaveBeenCalledWith({ queryKey: a2aLiveKeys.conversations, }); + expect(invalidateQueries).toHaveBeenCalledWith({ + queryKey: a2aLiveKeys.pairs, + }); expect(invalidateQueries).not.toHaveBeenCalledWith({ queryKey: a2aLiveKeys.messages("conv-1"), }); expect(screen.getByText("Offline")).toBeInTheDocument(); }); + + it("shows the switchboard by default with pair cards grouped into sections", () => { + useA2AAdminPairs.mockReturnValue({ + data: { items: [buildPair()], total: 1 }, + isLoading: false, + refetch: vi.fn(), + }); + render(); + expect(screen.getByText("Switchboard")).toBeInTheDocument(); + expect(screen.getByText(/Backend Cell/)).toBeInTheDocument(); + }); + + it("toggles to the classic conversation list and back", () => { + useA2AAdminPairs.mockReturnValue({ + data: { items: [buildPair()], total: 1 }, + isLoading: false, + refetch: vi.fn(), + }); + render(); + expect(screen.getByText("Switchboard")).toBeInTheDocument(); + + fireEvent.click(screen.getByTitle("Classic conversation list")); + expect(screen.getByText("Conversations")).toBeInTheDocument(); + expect(screen.queryByText(/Backend Cell/)).not.toBeInTheDocument(); + + fireEvent.click(screen.getByTitle("Switchboard: org-chart pair cards")); + expect(screen.getByText("Switchboard")).toBeInTheDocument(); + }); }); diff --git a/panel/src/app/(dashboard)/a2a/page.tsx b/panel/src/app/(dashboard)/a2a/page.tsx index 10b4c6e8..fb0fbcff 100644 --- a/panel/src/app/(dashboard)/a2a/page.tsx +++ b/panel/src/app/(dashboard)/a2a/page.tsx @@ -1,17 +1,21 @@ "use client"; -import { Suspense, useCallback, useEffect } from "react"; +import { Suspense, useCallback, useEffect, useMemo, useState } from "react"; import { useRouter, useSearchParams } from "next/navigation"; import { useQueryClient } from "@tanstack/react-query"; import { a2aLiveKeys, + useA2AAdminPairs, useA2AConversations, useA2AMessages, } from "@/hooks/use-a2a-live"; import { useA2ALiveStream } from "@/hooks/use-websocket"; import { A2AConversationList } from "@/components/a2a/a2a-conversation-list"; +import { A2ASwitchboard } from "@/components/a2a/a2a-switchboard"; import { A2ATranscript } from "@/components/a2a/a2a-transcript"; import { A2AReplyComposer } from "@/components/a2a/a2a-reply-composer"; +import { latestPulseTimestamps } from "@/components/a2a/a2a-switchboard-utils"; +import type { AdminPairSummary } from "@/lib/api/a2a"; import { Card, CardContent } from "@/components/ui/card"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; @@ -20,9 +24,22 @@ import { OfflineState } from "@/components/ui/offline-state"; import { getAgentDisplayName } from "@/lib/agent-utils"; import { lastSenderOf } from "@/components/a2a/a2a-utils"; import { cn } from "@/lib/utils"; -import { MessagesSquare, Radio, RefreshCw } from "lucide-react"; +import { + LayoutGrid, + List as ListIcon, + MessagesSquare, + Radio, + RefreshCw, +} from "lucide-react"; import { formatDistanceToNow } from "date-fns"; +type A2AView = "switchboard" | "list"; + +interface PeekedPair { + agent_a: string; + agent_b: string; +} + function EmptyPanel({ icon: Icon, message, @@ -47,12 +64,25 @@ function A2APageContent() { const selectedId = searchParams.get("conversation"); + // Desktop default is the switchboard (org-chart pair cards); the classic + // list stays one click away as the mobile/compact fallback. + const [view, setView] = useState("switchboard"); + // A pair with no conversation yet, clicked from the switchboard — there is + // nothing to select via `?conversation=`, so it's tracked separately and + // shown as an explicit "no A2A yet" state in the drill-in panel. + const [peekedPair, setPeekedPair] = useState(null); + const { data: conversationData, isLoading: loadingConversations, error, refetch: refetchConversations, - } = useA2AConversations(); + } = useA2AConversations(100); + const { + data: pairsData, + isLoading: loadingPairs, + refetch: refetchPairs, + } = useA2AAdminPairs(); const { data: messagesData, isLoading: loadingMessages, @@ -63,10 +93,11 @@ function A2APageContent() { // `a2a.message` frame. Invalidate-on-frame (the session-detail idiom) — the // frame's excerpt is capped by design, so REST stays the source of truth and // react-query refetches the affected queries. - const { lastMessage, isConnected } = useA2ALiveStream(); + const { lastMessage, a2aMessages, isConnected } = useA2ALiveStream(); useEffect(() => { if (lastMessage?.type !== "a2a.message") return; queryClient.invalidateQueries({ queryKey: a2aLiveKeys.conversations }); + queryClient.invalidateQueries({ queryKey: a2aLiveKeys.pairs }); if (selectedId && lastMessage.conversation_id === selectedId) { queryClient.invalidateQueries({ queryKey: a2aLiveKeys.messages(selectedId), @@ -74,8 +105,17 @@ function A2APageContent() { } }, [lastMessage, queryClient, selectedId]); + const pairs = useMemo(() => pairsData?.items ?? [], [pairsData]); + // Activity = A2A only: derived purely from a2a.message frames, never from + // verb/flow traffic on the same /ws/system stream. + const pulses = useMemo( + () => latestPulseTimestamps(a2aMessages, pairs), + [a2aMessages, pairs], + ); + const handleSelect = useCallback( (id: string) => { + setPeekedPair(null); const params = new URLSearchParams(searchParams.toString()); params.set("conversation", id); router.push(`/a2a?${params.toString()}`); @@ -83,8 +123,26 @@ function A2APageContent() { [router, searchParams], ); + const handleOpenPair = useCallback( + (pair: AdminPairSummary) => { + if (pair.conversation_id) { + handleSelect(pair.conversation_id); + return; + } + // Never-talked pair: nothing to select, clear any prior selection and + // show the pair's own empty state instead. + setPeekedPair({ agent_a: pair.agent_a, agent_b: pair.agent_b }); + const params = new URLSearchParams(searchParams.toString()); + params.delete("conversation"); + const qs = params.toString(); + router.push(qs ? `/a2a?${qs}` : "/a2a"); + }, + [handleSelect, router, searchParams], + ); + const handleRefresh = () => { refetchConversations(); + refetchPairs(); if (selectedId) refetchMessages(); }; @@ -137,20 +195,56 @@ function A2APageContent() { /> ) : (
- {/* Panel 1: Conversations */} + {/* Panel 1: Switchboard (default) / classic conversation list */}
- Conversations + + {view === "switchboard" ? "Switchboard" : "Conversations"} + +
+ + +
- + {view === "switchboard" ? ( + + ) : ( + + )}
@@ -211,6 +305,17 @@ function A2APageContent() { )}
+ ) : peekedPair ? ( +
+
+ +

+ {getAgentDisplayName(peekedPair.agent_a)} and{" "} + {getAgentDisplayName(peekedPair.agent_b)} haven't + A2A'd each other yet. +

+
+
) : ( = {}): AdminPairSummary { + return { + agent_a: "be-dev-1", + role_a: "developer", + team_a: "backend", + agent_b: "be-qa", + role_b: "qa", + team_b: "backend", + group_key: "cell-backend", + conversation_id: "conv-1", + last_message_at: "2026-07-02T09:00:00Z", + message_count: 5, + ...overrides, + }; +} + +describe("A2APairCard", () => { + it("renders both display names, message count, and relative time", () => { + render( + , + ); + expect(screen.getByText(/Backend Dev 1/)).toBeInTheDocument(); + expect(screen.getByText(/Backend QA/)).toBeInTheDocument(); + expect(screen.getByText("5")).toBeInTheDocument(); + expect(screen.getByText(/ago$/)).toBeInTheDocument(); + }); + + it("fires onOpen when clicked", () => { + const onOpen = vi.fn(); + render(); + fireEvent.click(screen.getByTestId("pair-card")); + expect(onOpen).toHaveBeenCalledTimes(1); + }); + + it("renders dimmed and shows 'No A2A yet' for a never-talked pair", () => { + render( + , + ); + expect(screen.getByText("No A2A yet")).toBeInTheDocument(); + expect(screen.getByTestId("pair-card")).toHaveClass("opacity-60"); + // No message-count badge for a pair with no history. + expect(screen.queryByText("5")).not.toBeInTheDocument(); + }); + + it("marks the card as selected via aria-pressed", () => { + render( + , + ); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "aria-pressed", + "true", + ); + }); +}); + +describe("A2APairCard pulsing (frame -> matching card lights up, then fades)", () => { + // Capture the rAF callback instead of letting it fire on the real event + // loop, so the activation/fade transition is fully deterministic. + let rafCallback: FrameRequestCallback | null = null; + + beforeEach(() => { + rafCallback = null; + vi.spyOn(window, "requestAnimationFrame").mockImplementation((cb) => { + rafCallback = cb; + return 1; + }); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("starts cold with no pulse", () => { + render(); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "false", + ); + }); + + it("goes hot the instant a matching pulsedAt is received", () => { + const { rerender } = render( + , + ); + + rerender( + , + ); + + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "true", + ); + }); + + it("fades back to cold on the next paint frame (CSS transition then does the decay)", () => { + const { rerender } = render( + , + ); + rerender( + , + ); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "true", + ); + + act(() => { + rafCallback?.(0); + }); + + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "false", + ); + }); + + it("does not re-trigger the pulse for an unchanged pulsedAt", () => { + const { rerender } = render( + , + ); + act(() => { + rafCallback?.(0); + }); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "false", + ); + + // Same pulsedAt as before (e.g. an unrelated parent re-render) — must + // stay cooled, not flash again. + rerender( + , + ); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "false", + ); + }); + + it("re-triggers on a newer pulsedAt after cooling down", () => { + const { rerender } = render( + , + ); + act(() => { + rafCallback?.(0); + }); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "false", + ); + + rerender( + , + ); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "true", + ); + }); +}); diff --git a/panel/src/components/a2a/__tests__/a2a-switchboard-utils.test.ts b/panel/src/components/a2a/__tests__/a2a-switchboard-utils.test.ts new file mode 100644 index 00000000..525a1fb9 --- /dev/null +++ b/panel/src/components/a2a/__tests__/a2a-switchboard-utils.test.ts @@ -0,0 +1,219 @@ +import { describe, it, expect } from "vitest"; +import type { AdminPairSummary } from "@/lib/api/a2a"; +import { + pairKey, + pairMatchesFrame, + latestPulseTimestamps, + groupPairsBySection, + sortPairsForSection, + SECTION_LABELS, +} from "../a2a-switchboard-utils"; + +function buildPair(overrides: Partial = {}): AdminPairSummary { + return { + agent_a: "be-dev-1", + role_a: "developer", + team_a: "backend", + agent_b: "be-qa", + role_b: "qa", + team_b: "backend", + group_key: "cell-backend", + conversation_id: null, + last_message_at: null, + message_count: 0, + ...overrides, + }; +} + +describe("pairKey", () => { + it("is order-independent", () => { + expect(pairKey("be-dev-1", "be-qa")).toBe(pairKey("be-qa", "be-dev-1")); + }); + + it("always puts the lexically smaller slug first", () => { + expect(pairKey("be-qa", "be-dev-1")).toBe("be-dev-1|be-qa"); + }); +}); + +describe("pairMatchesFrame", () => { + it("matches regardless of from/to direction", () => { + expect(pairMatchesFrame("be-dev-1", "be-qa", "be-qa", "be-dev-1")).toBe( + true, + ); + expect(pairMatchesFrame("be-dev-1", "be-qa", "be-dev-1", "be-qa")).toBe( + true, + ); + }); + + it("does not match a different pair", () => { + expect(pairMatchesFrame("be-dev-1", "be-qa", "fe-dev-1", "fe-qa")).toBe( + false, + ); + }); + + it("does not match when either side is missing", () => { + expect(pairMatchesFrame("be-dev-1", "be-qa", undefined, "be-qa")).toBe( + false, + ); + expect(pairMatchesFrame("be-dev-1", "be-qa", "be-dev-1", null)).toBe( + false, + ); + }); +}); + +describe("latestPulseTimestamps", () => { + const pairs = [ + buildPair({ agent_a: "be-dev-1", agent_b: "be-qa" }), + buildPair({ + agent_a: "fe-dev-1", + agent_b: "fe-qa", + group_key: "cell-frontend", + }), + ]; + + it("returns the epoch ms of the matching frame for a pulsed pair", () => { + const result = latestPulseTimestamps( + [ + { + from_agent: "be-dev-1", + to_agent: "be-qa", + timestamp: "2026-07-02T10:00:00Z", + }, + ], + pairs, + ); + expect(result[pairKey("be-dev-1", "be-qa")]).toBe( + new Date("2026-07-02T10:00:00Z").getTime(), + ); + }); + + it("omits pairs with no matching frame", () => { + const result = latestPulseTimestamps( + [ + { + from_agent: "be-dev-1", + to_agent: "be-qa", + timestamp: "2026-07-02T10:00:00Z", + }, + ], + pairs, + ); + expect(result[pairKey("fe-dev-1", "fe-qa")]).toBeUndefined(); + }); + + it("keeps only the most recent matching frame per pair", () => { + const result = latestPulseTimestamps( + [ + { + from_agent: "be-dev-1", + to_agent: "be-qa", + timestamp: "2026-07-02T09:00:00Z", + }, + { + from_agent: "be-qa", + to_agent: "be-dev-1", + timestamp: "2026-07-02T11:00:00Z", + }, + ], + pairs, + ); + expect(result[pairKey("be-dev-1", "be-qa")]).toBe( + new Date("2026-07-02T11:00:00Z").getTime(), + ); + }); + + it("ignores frames with an unparseable timestamp", () => { + const result = latestPulseTimestamps( + [{ from_agent: "be-dev-1", to_agent: "be-qa", timestamp: "not-a-date" }], + pairs, + ); + expect(result[pairKey("be-dev-1", "be-qa")]).toBeUndefined(); + }); + + it("returns an empty map for an empty frame list", () => { + expect(latestPulseTimestamps([], pairs)).toEqual({}); + }); +}); + +describe("sortPairsForSection", () => { + it("sorts pairs with history before never-talked pairs, most recent first", () => { + const older = buildPair({ + agent_a: "be-dev-1", + agent_b: "be-doc", + conversation_id: "c1", + last_message_at: "2026-07-01T00:00:00Z", + }); + const newer = buildPair({ + agent_a: "be-dev-2", + agent_b: "be-doc", + conversation_id: "c2", + last_message_at: "2026-07-02T00:00:00Z", + }); + const neverTalked = buildPair({ + agent_a: "be-pm", + agent_b: "be-qa", + conversation_id: null, + last_message_at: null, + }); + + const sorted = sortPairsForSection([older, neverTalked, newer]); + expect(sorted.map((p) => p.conversation_id)).toEqual(["c2", "c1", null]); + }); + + it("breaks ties between never-talked pairs alphabetically", () => { + const b = buildPair({ agent_a: "fe-dev-1", agent_b: "fe-qa" }); + const a = buildPair({ agent_a: "be-dev-1", agent_b: "be-qa" }); + const sorted = sortPairsForSection([b, a]); + expect(sorted[0]).toBe(a); + expect(sorted[1]).toBe(b); + }); +}); + +describe("groupPairsBySection", () => { + it("groups pairs by group_key and orders sections canonically", () => { + const pairs = [ + buildPair({ group_key: "cross", agent_a: "be-pm", agent_b: "fe-pm" }), + buildPair({ + group_key: "board", + agent_a: "auditor", + agent_b: "product-owner", + }), + buildPair({ group_key: "cell-backend" }), + buildPair({ + group_key: "pm-chain", + agent_a: "be-pm", + agent_b: "main-pm", + }), + ]; + + const sections = groupPairsBySection(pairs); + + expect(sections.map((s) => s.groupKey)).toEqual([ + "cell-backend", + "pm-chain", + "board", + "cross", + ]); + for (const section of sections) { + expect(section.label).toBe(SECTION_LABELS[section.groupKey]); + expect(section.pairs.length).toBe(1); + } + }); + + it("appends unrecognized group keys after the canonical sections", () => { + const pairs = [ + buildPair({ group_key: "cell-backend" }), + buildPair({ group_key: "mystery-group" }), + ]; + const sections = groupPairsBySection(pairs); + expect(sections.map((s) => s.groupKey)).toEqual([ + "cell-backend", + "mystery-group", + ]); + expect(sections[1].label).toBe("mystery-group"); + }); + + it("returns no sections for an empty pair list", () => { + expect(groupPairsBySection([])).toEqual([]); + }); +}); diff --git a/panel/src/components/a2a/__tests__/a2a-switchboard.test.tsx b/panel/src/components/a2a/__tests__/a2a-switchboard.test.tsx new file mode 100644 index 00000000..21540470 --- /dev/null +++ b/panel/src/components/a2a/__tests__/a2a-switchboard.test.tsx @@ -0,0 +1,134 @@ +import { describe, it, expect, vi } from "vitest"; +import { render, screen, fireEvent } from "@testing-library/react"; +import type { AdminPairSummary } from "@/lib/api/a2a"; +import { A2ASwitchboard } from "../a2a-switchboard"; + +function buildPair(overrides: Partial = {}): AdminPairSummary { + return { + agent_a: "be-dev-1", + role_a: "developer", + team_a: "backend", + agent_b: "be-qa", + role_b: "qa", + team_b: "backend", + group_key: "cell-backend", + conversation_id: null, + last_message_at: null, + message_count: 0, + ...overrides, + }; +} + +describe("A2ASwitchboard", () => { + it("shows a loading skeleton grid", () => { + render( + , + ); + // No section headings or empty-state copy while loading. + expect(screen.queryByText(/Backend Cell/)).not.toBeInTheDocument(); + expect( + screen.queryByText(/No allowed A2A pairs configured/), + ).not.toBeInTheDocument(); + }); + + it("shows an empty state when there are no pairs", () => { + render( + , + ); + expect( + screen.getByText("No allowed A2A pairs configured"), + ).toBeInTheDocument(); + }); + + it("groups pairs into labeled sections with counts", () => { + const pairs = [ + buildPair({ group_key: "cell-backend" }), + buildPair({ + group_key: "cell-backend", + agent_a: "be-dev-2", + agent_b: "be-doc", + }), + buildPair({ + group_key: "board", + agent_a: "auditor", + agent_b: "product-owner", + }), + ]; + render( + , + ); + expect(screen.getByText(/Backend Cell/)).toBeInTheDocument(); + expect(screen.getByText("(2)")).toBeInTheDocument(); + expect(screen.getByText(/^Board$/)).toBeInTheDocument(); + expect(screen.getByText("(1)")).toBeInTheDocument(); + expect(screen.getAllByTestId("pair-card")).toHaveLength(3); + }); + + it("calls onOpenPair with the clicked pair", () => { + const onOpenPair = vi.fn(); + const pair = buildPair({ conversation_id: "conv-9" }); + render( + , + ); + fireEvent.click(screen.getByTestId("pair-card")); + expect(onOpenPair).toHaveBeenCalledWith(pair); + }); + + it("marks the card matching selectedConversationId as selected", () => { + const pair = buildPair({ conversation_id: "conv-9" }); + render( + , + ); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "aria-pressed", + "true", + ); + }); + + it("passes each pair's pulse timestamp through by canonical pair key", () => { + const pair = buildPair(); + render( + , + ); + expect(screen.getByTestId("pair-card")).toHaveAttribute( + "data-pulsing", + "true", + ); + }); +}); diff --git a/panel/src/components/a2a/a2a-pair-card.tsx b/panel/src/components/a2a/a2a-pair-card.tsx new file mode 100644 index 00000000..c7f67004 --- /dev/null +++ b/panel/src/components/a2a/a2a-pair-card.tsx @@ -0,0 +1,117 @@ +"use client"; + +import { useEffect, useState } from "react"; +import { Badge } from "@/components/ui/badge"; +import { getAgentDisplayName, getAgentInitials } from "@/lib/agent-utils"; +import type { AdminPairSummary } from "@/lib/api/a2a"; +import { cn } from "@/lib/utils"; +import { formatDistanceToNow } from "date-fns"; +import { PAIR_PULSE_FADE_MS } from "./a2a-switchboard-utils"; + +interface A2APairCardProps { + pair: AdminPairSummary; + /** Epoch ms of the latest a2a.message frame matching this pair, or null + * when neither side has ever A2A'd the other (this session). */ + pulsedAt: number | null; + isSelected?: boolean; + onOpen: () => void; +} + +function PairAvatar({ slug }: { slug: string }) { + return ( +
+ + {getAgentInitials(slug)} + +
+ ); +} + +/** + * One pair card in the A2A switchboard. Lights up on a matching a2a.message + * frame — the card jumps to full "hot" intensity, then a plain CSS + * transition (no animation library) fades it back to baseline over + * PAIR_PULSE_FADE_MS. A never-talked pair (no conversation_id) renders + * dimmed/compact. + */ +export function A2APairCard({ + pair, + pulsedAt, + isSelected, + onOpen, +}: A2APairCardProps) { + const hasHistory = pair.conversation_id !== null; + const [isPulsing, setIsPulsing] = useState(false); + + // Render-phase derivation, not an Effect (react.dev/learn/you-might-not- + // need-an-effect#adjusting-some-state-when-a-prop-changes): flash hot in + // the very same render that receives a new pulsedAt, comparing against the + // last value we've seen. No cascading extra render from an Effect body. + // Seeded to null (not the initial pulsedAt) so a card that *mounts* + // already carrying a live pulse — e.g. switching into switchboard view + // right after a frame arrived — still flashes hot instead of looking cold. + const [lastSeenPulse, setLastSeenPulse] = useState(null); + if (pulsedAt !== lastSeenPulse) { + setLastSeenPulse(pulsedAt); + if (pulsedAt !== null) setIsPulsing(true); + } + + // Flip back on the next paint frame — the long CSS transition-duration + // below then animates the decay from "hot" to baseline over + // PAIR_PULSE_FADE_MS. The setState here is inside the (async) rAF + // callback, not the Effect body itself, so it's the intended "subscribe to + // an external clock" use of an Effect. + useEffect(() => { + if (!isPulsing) return; + const raf = requestAnimationFrame(() => setIsPulsing(false)); + return () => cancelAnimationFrame(raf); + }, [isPulsing]); + + return ( + + ); +} diff --git a/panel/src/components/a2a/a2a-switchboard-utils.ts b/panel/src/components/a2a/a2a-switchboard-utils.ts new file mode 100644 index 00000000..9ecb1a68 --- /dev/null +++ b/panel/src/components/a2a/a2a-switchboard-utils.ts @@ -0,0 +1,135 @@ +/** + * Pure helpers for the A2A switchboard (org-chart pair cards) — extracted + * for direct unit testing, same idiom as a2a-utils.ts. + */ + +import type { AdminPairSummary } from "@/lib/api/a2a"; + +/** How long a card stays visibly "hot" after a matching frame (CSS transition). */ +export const PAIR_PULSE_FADE_MS = 45_000; + +/** Canonical, order-independent key for a pair — stable regardless of which + * slug is passed first. */ +export function pairKey(agentA: string, agentB: string): string { + return agentA < agentB ? `${agentA}|${agentB}` : `${agentB}|${agentA}`; +} + +/** True when an A2A live-stream frame's from/to agents are this unordered pair. */ +export function pairMatchesFrame( + agentA: string, + agentB: string, + frameFrom?: string | null, + frameTo?: string | null, +): boolean { + if (!frameFrom || !frameTo) return false; + return ( + (frameFrom === agentA && frameTo === agentB) || + (frameFrom === agentB && frameTo === agentA) + ); +} + +interface PulseFrame { + from_agent?: string | null; + to_agent?: string | null; + timestamp?: string | null; +} + +/** + * For each pair, the epoch ms of the most recent matching frame — or absent + * from the map when no frame has matched. Bounded: O(pairs * frames), both + * small (the switchboard's static matrix, and the live-stream's capped + * message buffer). + */ +export function latestPulseTimestamps( + frames: ReadonlyArray, + pairs: ReadonlyArray>, +): Record { + const out: Record = {}; + for (const pair of pairs) { + const key = pairKey(pair.agent_a, pair.agent_b); + for (const frame of frames) { + if ( + !pairMatchesFrame( + pair.agent_a, + pair.agent_b, + frame.from_agent, + frame.to_agent, + ) + ) { + continue; + } + const ts = frame.timestamp ? new Date(frame.timestamp).getTime() : NaN; + if (Number.isNaN(ts)) continue; + if (out[key] === undefined || ts > out[key]) out[key] = ts; + } + } + return out; +} + +/** Stable section ordering — cells first (org-chart top-down), then the PM + * chain, board, and finally the lateral catch-all. */ +export const SECTION_ORDER = [ + "cell-backend", + "cell-frontend", + "cell-ux_ui", + "pm-chain", + "board", + "cross", +] as const; + +export const SECTION_LABELS: Record = { + "cell-backend": "Backend Cell", + "cell-frontend": "Frontend Cell", + "cell-ux_ui": "UX/UI Cell", + "pm-chain": "PM Chain", + board: "Board", + cross: "Cross-Team", +}; + +export interface PairSection { + groupKey: string; + label: string; + pairs: AdminPairSummary[]; +} + +/** Pairs with history sort first (most recent activity first); never-talked + * pairs follow in a stable alphabetical order. */ +export function sortPairsForSection( + pairs: ReadonlyArray, +): AdminPairSummary[] { + return [...pairs].sort((a, b) => { + const aTime = a.last_message_at + ? new Date(a.last_message_at).getTime() + : null; + const bTime = b.last_message_at + ? new Date(b.last_message_at).getTime() + : null; + if (aTime !== null && bTime !== null) return bTime - aTime; + if (aTime !== null) return -1; + if (bTime !== null) return 1; + return `${a.agent_a}${a.agent_b}`.localeCompare(`${b.agent_a}${b.agent_b}`); + }); +} + +/** Group pairs into ordered, labeled sections for the switchboard grid. */ +export function groupPairsBySection( + pairs: ReadonlyArray, +): PairSection[] { + const byGroup = new Map(); + for (const pair of pairs) { + const list = byGroup.get(pair.group_key) ?? []; + list.push(pair); + byGroup.set(pair.group_key, list); + } + + const known = SECTION_ORDER.filter((key) => byGroup.has(key)); + const unknown = [...byGroup.keys()].filter( + (key) => !(SECTION_ORDER as readonly string[]).includes(key), + ); + + return [...known, ...unknown].map((groupKey) => ({ + groupKey, + label: SECTION_LABELS[groupKey] ?? groupKey, + pairs: sortPairsForSection(byGroup.get(groupKey) ?? []), + })); +} diff --git a/panel/src/components/a2a/a2a-switchboard.tsx b/panel/src/components/a2a/a2a-switchboard.tsx new file mode 100644 index 00000000..75e45048 --- /dev/null +++ b/panel/src/components/a2a/a2a-switchboard.tsx @@ -0,0 +1,86 @@ +"use client"; + +import { Radio } from "lucide-react"; +import { Skeleton } from "@/components/ui/skeleton"; +import type { AdminPairSummary } from "@/lib/api/a2a"; +import { A2APairCard } from "./a2a-pair-card"; +import { groupPairsBySection, pairKey } from "./a2a-switchboard-utils"; + +interface A2ASwitchboardProps { + pairs: AdminPairSummary[]; + /** pairKey(agent_a, agent_b) -> epoch ms of the latest matching frame. */ + pulses: Record; + selectedConversationId: string | null; + isLoading: boolean; + onOpenPair: (pair: AdminPairSummary) => void; +} + +const SKELETON_COUNT = 9; + +/** + * The org-chart switchboard: every allowed agent pair as a card, grouped + * into sections (each cell, the PM chain, board, cross-team) and sorted so + * pairs with history come first within their section. + */ +export function A2ASwitchboard({ + pairs, + pulses, + selectedConversationId, + isLoading, + onOpenPair, +}: A2ASwitchboardProps) { + if (isLoading) { + return ( +
+ {Array.from({ length: SKELETON_COUNT }).map((_, i) => ( + + ))} +
+ ); + } + + if (pairs.length === 0) { + return ( +
+
+ +

No allowed A2A pairs configured

+
+
+ ); + } + + const sections = groupPairsBySection(pairs); + + return ( +
+ {sections.map((section) => ( +
+

+ {section.label} + + ({section.pairs.length}) + +

+
+ {section.pairs.map((pair) => { + const key = pairKey(pair.agent_a, pair.agent_b); + return ( + onOpenPair(pair)} + /> + ); + })} +
+
+ ))} +
+ ); +} diff --git a/panel/src/hooks/use-a2a-live.ts b/panel/src/hooks/use-a2a-live.ts index ec4c46b3..72102654 100644 --- a/panel/src/hooks/use-a2a-live.ts +++ b/panel/src/hooks/use-a2a-live.ts @@ -6,6 +6,7 @@ import { a2aApi, type AdminReplyRequest } from "@/lib/api/a2a"; export const a2aLiveKeys = { all: ["a2a-live"] as const, conversations: ["a2a-live", "conversations"] as const, + pairs: ["a2a-live", "pairs"] as const, messages: (conversationId: string) => ["a2a-live", "messages", conversationId] as const, }; @@ -20,6 +21,17 @@ export function useA2AConversations(limit?: number) { }); } +// Switchboard pair cards (the org-chart view) — every allowed agent pair +// joined with its representative conversation stats. Refreshed by WS +// `a2a.message` invalidation, same as the conversation list. +export function useA2AAdminPairs() { + return useQuery({ + queryKey: a2aLiveKeys.pairs, + queryFn: () => a2aApi.listAdminPairs(), + staleTime: 30_000, + }); +} + // Transcript for one conversation. WS frames for the selected conversation // invalidate this key; full bodies always come from REST (excerpts are capped). export function useA2AMessages(conversationId: string | null) { diff --git a/panel/src/lib/api/a2a.ts b/panel/src/lib/api/a2a.ts index 88863d0a..4de0ab7a 100644 --- a/panel/src/lib/api/a2a.ts +++ b/panel/src/lib/api/a2a.ts @@ -103,6 +103,29 @@ export interface AdminConversationListResponse { total: number; } +/** + * One pair card for the CEO's A2A switchboard (org-chart view) — the static + * can_a2a_direct matrix joined with the pair's representative conversation + * stats when one exists. + */ +export interface AdminPairSummary { + agent_a: string; + role_a: string; + team_a: string; + agent_b: string; + role_b: string; + team_b: string; + group_key: string; + conversation_id: string | null; + last_message_at: string | null; + message_count: number; +} + +export interface AdminPairListResponse { + items: AdminPairSummary[]; + total: number; +} + export interface AdminMessageListResponse { items: A2AChatMessage[]; total: number; @@ -322,6 +345,49 @@ export const a2aApi = { return data; }, + /** + * List the org-chart switchboard's pair cards (CEO-only): every agent pair + * allowed to A2A directly, joined with each pair's representative + * conversation stats when one exists. + */ + listAdminPairs: async (): Promise => { + if (isMockMode()) { + return { + items: [ + { + agent_a: "be-dev-1", + role_a: "developer", + team_a: "backend", + agent_b: "be-qa", + role_b: "qa", + team_b: "backend", + group_key: "cell-backend", + conversation_id: "mock-conversation-1", + last_message_at: new Date().toISOString(), + message_count: 3, + }, + { + agent_a: "auditor", + role_a: "auditor", + team_a: "board", + agent_b: "product-owner", + role_b: "product_owner", + team_b: "board", + group_key: "board", + conversation_id: null, + last_message_at: null, + message_count: 0, + }, + ], + total: 2, + }; + } + const { data } = await api.get( + "/a2a/chat/admin/pairs", + ); + return data; + }, + /** * Send a CEO reply. Lands in the CEO<->to_agent pairwise conversation (the * A2A model is strictly pairwise), NOT inside the watched transcript. diff --git a/roboco/agent_sdk/secretary_driver.py b/roboco/agent_sdk/secretary_driver.py index 63e5f0e6..488e4777 100644 --- a/roboco/agent_sdk/secretary_driver.py +++ b/roboco/agent_sdk/secretary_driver.py @@ -132,7 +132,12 @@ def build_secretary_options( async def _t_read_state(_args: dict[str, Any]) -> dict[str, Any]: return _text_result(await _do_read_state()) - @tool("read_task", "Read one task's detail by its id.", {"task_id": str}) + @tool( + "read_task", + "Read one task's full detail by its id — content, notes, plan, " + "progress, and PR reference (Secretary FULL task access).", + {"task_id": str}, + ) async def _t_read_task(args: dict[str, Any]) -> dict[str, Any]: return _text_result(await _do_read_task(str(args["task_id"]))) @@ -140,8 +145,11 @@ def build_secretary_options( "submit_directive", "Act on the CEO's command. 'kind' is one of: relay_message " "(payload: channel, text), update_charter (payload: charter), " - "control_task (payload: task_id, action[start|cancel|override], " - "status?), approve_pitch (payload: pitch_id, notes?), announce " + "control_task (payload: task_id, action[start|cancel|override|edit], " + "status? for override, fields? for edit — edit accepts title/" + "description/acceptance_criteria/priority/team/estimated_complexity/" + "nature/assigned_to; assigned_to may be a UUID or an agent slug like " + '"be-dev-1"), approve_pitch (payload: pitch_id, notes?), announce ' "(payload: text). High-impact kinds (charter, control_task, " "approve_pitch, announce) are queued for the CEO's explicit " "confirmation; relay_message runs directly.", diff --git a/roboco/agents_config.py b/roboco/agents_config.py index d2d6a674..4f60510b 100644 --- a/roboco/agents_config.py +++ b/roboco/agents_config.py @@ -28,6 +28,7 @@ different purposes. MCP is coarse-grained (tool-level), API is fine-grained import hashlib import hmac import os +from dataclasses import dataclass from typing import Final from roboco.foundation import identity as _foundation @@ -716,3 +717,110 @@ def get_a2a_route_hint(from_agent: str, to_agent: str) -> str: return f"Route: {from_agent}→{cell_pm}→main-pm→board" return "Use escalate_up() for proper escalation." + + +# ============================================================================= +# A2A SWITCHBOARD — ALLOWED AGENT PAIRS (CEO admin view) +# ============================================================================= +# Static, stateless derivation from can_a2a_direct(): every unordered pair of +# real (non-human, non-sentinel) agents where at least one direction is +# permitted. This is the org-chart the CEO's A2A switchboard renders as pair +# cards — computed once at import time, since the matrix never changes at +# runtime. The route/service layer joins this list against live DB +# conversation data per request. + + +@dataclass(frozen=True) +class A2AAllowedPair: + """One CEO-visible pair of agents allowed to A2A directly (>=1 direction). + + ``agent_a`` < ``agent_b`` lexically — the same canonical ordering + A2AConversationTable and A2AService._canonical_pair use. + """ + + agent_a: str + agent_b: str + role_a: str + team_a: str + role_b: str + team_b: str + group_key: str + + +# Slugs eligible for the switchboard: excludes the system sentinel and the +# human-only roles (CEO, prompter, secretary) — none of those are real A2A +# participants in the org chart the CEO is browsing. +_SWITCHBOARD_SLUGS: Final[list[str]] = sorted( + slug + for slug, row in _foundation.AGENTS.items() + if slug != "system" and not _foundation.is_human_only_role(row.role) +) + +_BOARD_ROLE_VALUES: Final[frozenset[str]] = frozenset( + r.value for r in _foundation.BOARD_ROLES +) +_CELL_TEAM_VALUES: Final[frozenset[str]] = frozenset( + t.value for t in _foundation.CELL_TEAMS +) + + +def _a2a_group_key(role_a: str, team_a: str, role_b: str, team_b: str) -> str: + """Classify a pair into a stable section for the panel switchboard. + + - ``cell-``: both agents share a delivery-cell team — each cell's + own section (dev/qa/doc/pm/pr-reviewer talking within their cell). + - ``pm-chain``: the coordination spine — cell_pm<->main_pm, or + main_pm<->a board role (mirrors ESCALATION_CHAIN: cell_pm -> main-pm + -> board). + - ``board``: pure board-to-board pairs (product_owner/head_marketing/ + auditor). + - ``cross``: everything else — chiefly a PR reviewer's lateral reach + outside its own cell/pm (delivering a gate verdict to another cell's + PM or to main-pm), which isn't part of the escalation spine. + """ + roles = {role_a, role_b} + if team_a == team_b and team_a in _CELL_TEAM_VALUES: + return f"cell-{team_a}" + if roles == {"cell_pm", "main_pm"}: + return "pm-chain" + if "main_pm" in roles and ( + role_a in _BOARD_ROLE_VALUES or role_b in _BOARD_ROLE_VALUES + ): + return "pm-chain" + if role_a in _BOARD_ROLE_VALUES and role_b in _BOARD_ROLE_VALUES: + return "board" + return "cross" + + +def _compute_a2a_allowed_pairs() -> tuple[A2AAllowedPair, ...]: + """Enumerate every unordered pair with >=1 allowed A2A direction.""" + pairs: list[A2AAllowedPair] = [] + for i, a in enumerate(_SWITCHBOARD_SLUGS): + row_a = _foundation.AGENTS[a] + for b in _SWITCHBOARD_SLUGS[i + 1 :]: + row_b = _foundation.AGENTS[b] + allowed_ab, _ = can_a2a_direct(a, b) + allowed_ba, _ = can_a2a_direct(b, a) + if not (allowed_ab or allowed_ba): + continue + pairs.append( + A2AAllowedPair( + agent_a=a, + agent_b=b, + role_a=row_a.role.value, + team_a=row_a.team.value, + role_b=row_b.role.value, + team_b=row_b.team.value, + group_key=_a2a_group_key( + row_a.role.value, + row_a.team.value, + row_b.role.value, + row_b.team.value, + ), + ) + ) + return tuple(pairs) + + +# Computed once at module load — see module docstring above. +A2A_ALLOWED_PAIRS: Final[tuple[A2AAllowedPair, ...]] = _compute_a2a_allowed_pairs() diff --git a/roboco/api/routes/a2a.py b/roboco/api/routes/a2a.py index c21b543c..c9ba394a 100644 --- a/roboco/api/routes/a2a.py +++ b/roboco/api/routes/a2a.py @@ -35,6 +35,8 @@ from roboco.api.routes.v1._role_dep import require_any_authenticated_agent from roboco.api.schemas.a2a_chat import ( AdminConversationListResponse, AdminConversationSummaryResponse, + AdminPairListResponse, + AdminPairResponse, AdminReplyRequest, ConversationCloseRequest, ConversationCreateRequest, @@ -975,6 +977,44 @@ async def list_admin_conversations( ) +@router.get("/chat/admin/pairs") +async def list_admin_pairs( + db: DbSession, + agent: CurrentAgentContext, +) -> AdminPairListResponse: + """CEO-only: the org-chart switchboard. + + Every agent pair allowed to A2A directly per the static + ``agents_config.can_a2a_direct`` matrix (>=1 direction), joined with each + pair's representative conversation stats when one exists — the pair + cards the panel groups into sections (each cell, the PM chain, board). + """ + _require_ceo(agent) + service = A2AService(db) + pairs = await service.list_admin_pairs() + + return AdminPairListResponse( + items=[ + AdminPairResponse( + agent_a=p.agent_a, + role_a=p.role_a, + team_a=p.team_a, + agent_b=p.agent_b, + role_b=p.role_b, + team_b=p.team_b, + group_key=p.group_key, + conversation_id=( + require_uuid(p.conversation_id) if p.conversation_id else None + ), + last_message_at=p.last_message_at, + message_count=p.message_count, + ) + for p in pairs + ], + total=len(pairs), + ) + + @router.get("/chat/admin/conversations/{conversation_id}/messages") async def list_admin_chat_messages( conversation_id: str, diff --git a/roboco/api/routes/secretary.py b/roboco/api/routes/secretary.py index 581846fd..bd55d2b9 100644 --- a/roboco/api/routes/secretary.py +++ b/roboco/api/routes/secretary.py @@ -76,7 +76,8 @@ async def search_tasks( async def read_task( task_id: UUID, db: DbSession, agent: CurrentAgentContext ) -> dict[str, object]: - """Read one task's detail (Secretary or CEO).""" + """Read one task's full detail — content, notes, plan, progress, PR ref + (Secretary or CEO). Secretary FULL task access.""" _require(agent, _SECRETARY_OR_CEO) try: return await get_secretary_service(db).read_task(task_id) diff --git a/roboco/api/routes/tasks.py b/roboco/api/routes/tasks.py index fd5c37e3..d86c1f1f 100644 --- a/roboco/api/routes/tasks.py +++ b/roboco/api/routes/tasks.py @@ -234,6 +234,73 @@ _PRIVILEGED_UPDATE_FIELDS: frozenset[str] = frozenset( } ) +# The CEO's "PM lighter" scope: cell_pm/main_pm may PATCH this content-only +# slice — the same allowlist the Secretary's edit directive originally had +# (before it grew to Secretary FULL). No status changes, no structural/ +# ownership fields (_PRIVILEGED_UPDATE_FIELDS), no git fields — those stay on +# the lifecycle-verb surface (delegate/reassign/complete/...). +_PM_LIGHTER_UPDATE_FIELDS: frozenset[str] = frozenset( + {"title", "description", "acceptance_criteria", "priority"} +) + +# Roles that get the lighter slice above instead of the full ASSIGN-holding +# admin bypass. TaskAction.ASSIGN is not team-scoped (see +# can_perform_task_action), so a cell_pm would otherwise ride the same +# unrestricted bypass CEO/Board/Auditor get, on any team's task — the +# own-team restriction below is enforced independently of that permission. +_PM_LIGHTER_ROLES: frozenset[AgentRole] = frozenset( + {AgentRole.CELL_PM, AgentRole.MAIN_PM} +) + + +def _pm_editor_scope( + agent: AgentContext, task: TaskTable, *, has_higher_perms: bool +) -> bool: + """Return True if ``agent`` gets the "PM lighter" content-only slice. + + Raises 403 outright for a cell PM outside its own team — ASSIGN itself + is not team-scoped (see ``can_perform_task_action``), so without this + check a cross-team cell PM would fall through to the wider CEO/Board/ + Auditor admin bypass on ``has_higher_perms`` alone. + """ + is_pm_editor = has_higher_perms and agent.role in _PM_LIGHTER_ROLES + if is_pm_editor and agent.role == AgentRole.CELL_PM and agent.team != task.team: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=( + "Cell PM may only update tasks belonging to their own team " + f"({agent.team}); this task is on {task.team}." + ), + ) + return is_pm_editor + + +def _enforce_pm_lighter_fields( + updates: dict[str, Any], + null_clears: dict[str, Any], + new_status: TaskStatus | None, +) -> None: + """Refuse anything past the content-only allowlist for a PM-lighter editor. + + "No status changes beyond what they already have" — status rides the + lifecycle verbs, never this PATCH surface, for cell_pm/main_pm. + """ + disallowed = (updates.keys() | null_clears.keys()) - _PM_LIGHTER_UPDATE_FIELDS + if not disallowed and new_status is None: + return + reasons = [] + if disallowed: + reasons.append(f"disallowed fields {sorted(disallowed)}") + if new_status is not None: + reasons.append("status changes are not part of the PM PATCH surface") + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=( + f"PM roles may only edit {sorted(_PM_LIGHTER_UPDATE_FIELDS)} via " + "PATCH; " + "; ".join(reasons) + ), + ) + def _translate_error(e: ServiceError) -> HTTPException: """Service errors → HTTP status. Kept at route layer; everything else moves.""" @@ -1110,6 +1177,11 @@ async def update_task( agent, TaskAction.ASSIGN, task.team ) + # PM roles (cell_pm/main_pm) ride the ASSIGN-holding bypass above like + # CEO/Board/Auditor, but get the narrower "PM lighter" content-only slice + # instead of unrestricted admin access (see _pm_editor_scope). + is_pm_editor = _pm_editor_scope(agent, task, has_higher_perms=has_higher_perms) + if not ((can_update_own and is_owner) or has_higher_perms): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, @@ -1136,6 +1208,12 @@ async def update_task( # on the ORM object after the update returns. null_clears = _pop_null_clears(updates) + # PM-lighter: restrict to the content-only allowlist, and refuse a status + # change outright — "no status changes beyond what they already have" + # (the lifecycle verbs), not a new capability riding this PATCH surface. + if is_pm_editor: + _enforce_pm_lighter_fields(updates, null_clears, new_status) + # A bare task owner (UPDATE_OWN) may edit dev-facing fields only. The # structural / ownership fields are gated to ASSIGN/PM; an owner PATCHing # any of them (set or explicitly nulled) without higher perms is refused — diff --git a/roboco/api/schemas/a2a_chat.py b/roboco/api/schemas/a2a_chat.py index f59c86e3..387bbb0c 100644 --- a/roboco/api/schemas/a2a_chat.py +++ b/roboco/api/schemas/a2a_chat.py @@ -194,3 +194,30 @@ class AdminReplyRequest(BaseModel): to_agent: str = Field(..., description="Which participant to address") content: str = Field(..., min_length=1, max_length=10000) skill: str | None = None + + +# ============================================================================= +# SWITCHBOARD SCHEMAS (CEO-only) — org-chart pair cards +# ============================================================================= + + +class AdminPairResponse(BaseModel): + """One agent pair for the CEO's A2A switchboard (org-chart pair cards).""" + + agent_a: str + role_a: str + team_a: str + agent_b: str + role_b: str + team_b: str + group_key: str + conversation_id: UUID | None + last_message_at: datetime | None + message_count: int + + +class AdminPairListResponse(BaseModel): + """List of switchboard pairs.""" + + items: list[AdminPairResponse] + total: int diff --git a/roboco/mcp/secretary_server.py b/roboco/mcp/secretary_server.py index 7df88813..f42a6693 100644 --- a/roboco/mcp/secretary_server.py +++ b/roboco/mcp/secretary_server.py @@ -42,7 +42,8 @@ async def read_company_state() -> str: @mcp.tool() async def read_task(task_id: str) -> str: - """Read one task's detail by its id.""" + """Read one task's full detail by its id — content, notes, plan, + progress, and PR reference (Secretary FULL task access).""" return json.dumps(await _do_read_task(task_id)) @@ -52,7 +53,10 @@ async def submit_directive(kind: str, payload: dict[str, Any]) -> str: 'kind' is one of: relay_message (payload: channel, text), update_charter (payload: charter), control_task (payload: task_id, action[start|cancel| - override], status?), approve_pitch (payload: pitch_id, notes?), announce + override|edit], status? for override, fields? for edit — edit accepts + title/description/acceptance_criteria/priority/team/estimated_complexity/ + nature/assigned_to; assigned_to may be a UUID or an agent slug like + "be-dev-1"), approve_pitch (payload: pitch_id, notes?), announce (payload: text). High-impact kinds (charter, control_task, approve_pitch, announce) are queued for the CEO's explicit confirmation; relay_message runs directly. diff --git a/roboco/models/a2a.py b/roboco/models/a2a.py index d617e2a5..2007d046 100644 --- a/roboco/models/a2a.py +++ b/roboco/models/a2a.py @@ -615,3 +615,23 @@ class A2APair(RobocoBase): ) total_unread: int = Field(default=0, description="Total unread across all convos") last_activity: datetime | None = None + + +class A2AAdminPairSummary(RobocoBase): + """One CEO-visible org-chart pair for the A2A switchboard. + + Joins the static allowed-pair matrix (roboco.agents_config.A2A_ALLOWED_PAIRS + — role/team/group_key) with the pair's representative conversation stats + when one exists. + """ + + agent_a: str + role_a: str + team_a: str + agent_b: str + role_b: str + team_b: str + group_key: str + conversation_id: str | None = None + last_message_at: datetime | None = None + message_count: int = 0 diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index fb29e2aa..eec60b2b 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -2959,15 +2959,17 @@ class AgentOrchestrator: project_slug: str | None, task_id: str | None = None, product_id: str | None = None, + project_ids: list[str] | None = None, ) -> str | None: """Resolve the architectural-standard ambient block for the spawn. - Covers a delivery role's single project (via ``project_slug``) AND a - PO / Intake working a product, whose per-cell projects are resolved from - the task's ``product_id`` or a directly-supplied ``product_id``. - Best-effort + flag-gated: returns None (no ambient layer) when the - subsystem is off, no project is in scope, or anything fails — a prompt - compose must never be blocked by conventions resolution. + Covers a delivery role's single project (via ``project_slug``), a PO / + Intake working a product (per-cell projects resolved from the task's + ``product_id`` or a directly-supplied ``product_id``), AND a MegaTask + intake's explicit ``project_ids`` scope. Best-effort + flag-gated: + returns None (no ambient layer) when the subsystem is off, no project is + in scope, or anything fails — a prompt compose must never be blocked by + conventions resolution. """ from roboco.config import settings @@ -2984,6 +2986,7 @@ class AgentOrchestrator: project_slug=project_slug, task_id=task_id, product_id=product_id, + project_ids=project_ids, ) return await conventions_ambient_layer(db, projects) except Exception as exc: @@ -3001,9 +3004,12 @@ class AgentOrchestrator: project_slug: str | None, task_id: str | None, product_id: str | None, + project_ids: list[str] | None = None, ) -> list[Any]: - """The in-scope projects for the ambient block (single repo, product, or - ad-hoc cell map).""" + """The in-scope projects for the ambient block: single repo, product, + an explicit MegaTask ``project_ids`` set, or an ad-hoc cell map.""" + if project_ids: + return await self._projects_by_ids(db, project_ids) if product_id is not None: return await self._ambient_product_projects(db, product_id) if task_id is not None: @@ -3017,6 +3023,23 @@ class AgentOrchestrator: return [project] if project is not None else [] return [] + @staticmethod + async def _projects_by_ids(db: Any, project_ids: list[str]) -> list[Any]: + """Resolve an explicit id list to project rows, in order, skipping any + that don't resolve — best-effort ambient resolution, not the hard + clone-scope resolver (which fails loud on a missing id).""" + from uuid import UUID + + from roboco.services.project import get_project_service + + project_svc = get_project_service(db) + out = [] + for pid in project_ids: + p = await project_svc.get(UUID(pid)) + if p is not None: + out.append(p) + return out + @staticmethod async def _ambient_projects_for_task(db: Any, task_id: str) -> list[Any]: """The in-scope projects for a task's ambient block, from its product OR @@ -3068,11 +3091,10 @@ class AgentOrchestrator: ) -> str | None: """Resolve the prompter's task-history-digest ambient block for this scope. - Unlike the conventions ambient resolver, this covers all three intake - scopes including ``project_ids`` (a MegaTask) — the digest is meant to - span every project the intake agent is reading. Best-effort: returns None - on any failure or empty scope so history resolution can never block a - spawn. + Covers all three intake scopes including ``project_ids`` (a MegaTask) — + the digest is meant to span every project the intake agent is reading. + Best-effort: returns None on any failure or empty scope so history + resolution can never block a spawn. """ try: from roboco.db.base import get_session_factory @@ -3106,17 +3128,7 @@ class AgentOrchestrator: """The in-scope ProjectTable rows for the history digest — single repo, product (all cell projects), or an explicit MegaTask project_ids set.""" if project_ids: - from uuid import UUID - - from roboco.services.project import get_project_service - - project_svc = get_project_service(db) - out = [] - for pid in project_ids: - p = await project_svc.get(UUID(pid)) - if p is not None: - out.append(p) - return out + return await AgentOrchestrator._projects_by_ids(db, project_ids) if product_id is not None: return await AgentOrchestrator._ambient_product_projects(db, product_id) if project_slug: @@ -3136,7 +3148,7 @@ class AgentOrchestrator: """The intake spawn's full ambient block: conventions + history digest, joined with ``compose_prompt``'s own layer separator.""" conventions_ambient = await self._resolve_conventions_ambient( - project_slug, product_id=product_id + project_slug, product_id=product_id, project_ids=project_ids ) history_ambient = await self._resolve_history_digest_ambient( project_slug, product_id=product_id, project_ids=project_ids diff --git a/roboco/services/a2a.py b/roboco/services/a2a.py index 0ae74c7f..1a88e9f5 100644 --- a/roboco/services/a2a.py +++ b/roboco/services/a2a.py @@ -15,7 +15,12 @@ import structlog from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from roboco.agents_config import ALL_AGENTS, get_agent_skills, get_agent_team +from roboco.agents_config import ( + A2A_ALLOWED_PAIRS, + ALL_AGENTS, + get_agent_skills, + get_agent_team, +) from roboco.config import settings from roboco.db.tables import ( A2AConversationTable, @@ -26,6 +31,7 @@ from roboco.db.tables import ( from roboco.enforcement import A2AAccessDeniedError, validate_a2a_access from roboco.events import Event, EventType, get_event_bus from roboco.models.a2a import ( + A2AAdminPairSummary, A2AArtifact, A2AChatMessage, A2AConversation, @@ -1092,6 +1098,52 @@ class A2AService: return summaries + async def list_admin_pairs(self) -> list[A2AAdminPairSummary]: + """CEO switchboard: every allowed agent pair (static matrix, see + ``agents_config.A2A_ALLOWED_PAIRS``) joined with its representative + conversation when one exists. + + One bulk query over the bounded static pair count — never N+1. When a + pair has more than one conversation (distinct topics), the most + recently updated one is treated as "the" conversation for that pair. + """ + from sqlalchemy import tuple_ + + canonical_keys = [(p.agent_a, p.agent_b) for p in A2A_ALLOWED_PAIRS] + conv_by_pair: dict[tuple[str, str], A2AConversationTable] = {} + if canonical_keys: + result = await self.session.execute( + select(A2AConversationTable).where( + tuple_( + A2AConversationTable.agent_a, A2AConversationTable.agent_b + ).in_(canonical_keys) + ) + ) + for conv in result.scalars().all(): + key = (conv.agent_a, conv.agent_b) + current = conv_by_pair.get(key) + if current is None or conv.updated_at > current.updated_at: + conv_by_pair[key] = conv + + summaries: list[A2AAdminPairSummary] = [] + for p in A2A_ALLOWED_PAIRS: + rep = conv_by_pair.get((p.agent_a, p.agent_b)) + summaries.append( + A2AAdminPairSummary( + agent_a=p.agent_a, + role_a=p.role_a, + team_a=p.team_a, + agent_b=p.agent_b, + role_b=p.role_b, + team_b=p.team_b, + group_key=p.group_key, + conversation_id=str(rep.id) if rep else None, + last_message_at=rep.last_message_at if rep else None, + message_count=rep.message_count if rep else 0, + ) + ) + return summaries + async def close_conversation( self, conversation_id: UUID, diff --git a/roboco/services/secretary.py b/roboco/services/secretary.py index 4cbee760..d2dfcbe2 100644 --- a/roboco/services/secretary.py +++ b/roboco/services/secretary.py @@ -20,7 +20,7 @@ from sqlalchemy import select from roboco.db.tables import SecretaryDirectiveTable from roboco.foundation.identity import AGENTS -from roboco.models.base import TaskStatus +from roboco.models.base import Complexity, TaskNature, TaskStatus, Team from roboco.models.secretary import GATED_KINDS, DirectiveKind, DirectiveStatus from roboco.services.base import ( BaseService, @@ -31,14 +31,28 @@ from roboco.services.base import ( from roboco.services.company_goals import get_company_goals_service from roboco.services.messaging import get_messaging_service from roboco.services.pitch import get_pitch_service +from roboco.services.repositories.query_helpers import get_agent_by_slug from roboco.services.task import get_task_service -from roboco.utils.converters import require_uuid +from roboco.utils.converters import InvalidIdentifierError, require_uuid if TYPE_CHECKING: from uuid import UUID from sqlalchemy.ext.asyncio import AsyncSession + from roboco.db.tables import TaskTable + from roboco.services.task import TaskService + +# Sentinel distinguishing "assigned_to not present in the edit payload" from +# an explicit ``None`` (unassign) — a plain ``None`` default would conflate +# the two and silently skip a deliberate unassign. +_UNSET: Any = object() + +# Full-detail read is bounded: a long-running task's progress history could +# otherwise blow up the payload. Mirrors the spirit of other list caps in the +# codebase (e.g. get_subtasks' [:500]). +_MAX_PROGRESS_UPDATES = 50 + _CEO_ID = AGENTS["ceo"].uuid _ANNOUNCE_CHANNEL = "announcements" @@ -77,7 +91,16 @@ class SecretaryService(BaseService): } async def read_task(self, task_id: UUID) -> dict[str, Any]: - task = await get_task_service(self.session).get(task_id) + """Full-detail task read — the Secretary's FULL task access. + + Beyond identity/status/description, carries everything the panel's + full ``TaskResponse`` has that this previously omitted: acceptance + criteria, plan, notes (dev/qa/auditor/pr-reviewer/doc/quick-context), + and the PR/branch reference. ``progress_updates`` is bounded to the + most recent entries (see ``_MAX_PROGRESS_UPDATES``) so a long-running + task's history can't blow up the payload. + """ + task: TaskTable | None = await get_task_service(self.session).get(task_id) if task is None: raise NotFoundError("task", str(task_id)) return { @@ -87,6 +110,25 @@ class SecretaryService(BaseService): "team": str(task.team) if task.team else None, "assigned_to": str(task.assigned_to) if task.assigned_to else None, "description": task.description, + "acceptance_criteria": list(task.acceptance_criteria or []), + "priority": task.priority, + "estimated_complexity": ( + str(task.estimated_complexity) if task.estimated_complexity else None + ), + "nature": str(task.nature) if task.nature else None, + "plan": task.plan, + "progress_updates": list(task.progress_updates or [])[ + -_MAX_PROGRESS_UPDATES: + ], + "dev_notes": task.dev_notes, + "qa_notes": task.qa_notes, + "auditor_notes": task.auditor_notes, + "pr_reviewer_notes": task.pr_reviewer_notes, + "doc_notes": task.doc_notes, + "quick_context": task.quick_context, + "branch_name": task.branch_name, + "pr_number": task.pr_number, + "pr_url": task.pr_url, } # ------------------------------------------------------------------ # @@ -226,29 +268,44 @@ class SecretaryService(BaseService): return "pitch approved and provisioned" return await self._control_task(payload) - # Content fields the Secretary may edit on CEO confirmation. Status, - # ownership, and git fields never ride an edit — they have their own - # audited paths (override, reassign, the git workflow). + # Content fields the Secretary may edit on CEO confirmation — the FULL + # surface (Secretary FULL task access). Status is never set here — it has + # its own audited path (the "start"/"cancel"/"override" actions below); + # git fields (branch/PR) are never editable — they follow the git + # workflow. ``assigned_to`` rides the edit too but is handled separately + # (see ``_edit_task``) since it needs claim-aware reassignment, not a + # plain field set. _EDITABLE_TASK_FIELDS: ClassVar[frozenset[str]] = frozenset( - {"title", "description", "acceptance_criteria", "priority"} + { + "title", + "description", + "acceptance_criteria", + "priority", + "team", + "estimated_complexity", + "nature", + "assigned_to", + } ) + # Enum-typed content fields need coercion from the raw JSON string before + # they reach TaskService.update() (a generic setattr passthrough with no + # type coercion of its own). + _ENUM_TASK_FIELDS: ClassVar[dict[str, Any]] = { + "team": Team, + "estimated_complexity": Complexity, + "nature": TaskNature, + } + + _ASSIGNMENT_FIELD = "assigned_to" + async def _control_task(self, payload: dict[str, Any]) -> str: task_svc = get_task_service(self.session) task_id = require_uuid(payload["task_id"]) action = str(payload["action"]) notes = str(payload.get("notes", "via Secretary on CEO command")) if action == "edit": - fields = dict(payload.get("fields") or {}) - illegal = set(fields) - self._EDITABLE_TASK_FIELDS - if not fields or illegal: - raise ValidationError( - "edit accepts only " - f"{sorted(self._EDITABLE_TASK_FIELDS)}; got " - f"{sorted(fields) or 'nothing'}" - ) - await task_svc.update(task_id, **fields) - return f"task fields updated: {', '.join(sorted(fields))}" + return await self._edit_task(task_svc, task_id, payload) if action == "start": await task_svc.approve_and_start(task_id, notes) return "task started" @@ -265,6 +322,84 @@ class SecretaryService(BaseService): return f"task set to {new_status.value}" raise ValidationError(f"unknown task action: {action!r}") + async def _edit_task( + self, task_svc: TaskService, task_id: UUID, payload: dict[str, Any] + ) -> str: + """Apply the Secretary's edit — content fields + optional reassign. + + ``assigned_to`` is popped out and routed through claim-aware + reassignment (``_reassign_task``) rather than a plain field set; + every other allowlisted field goes through ``TaskService.update`` + after enum coercion. + """ + fields = dict(payload.get("fields") or {}) + illegal = set(fields) - self._EDITABLE_TASK_FIELDS + if not fields or illegal: + raise ValidationError( + "edit accepts only " + f"{sorted(self._EDITABLE_TASK_FIELDS)}; got " + f"{sorted(fields) or 'nothing'}" + ) + reassign_to = fields.pop(self._ASSIGNMENT_FIELD, _UNSET) + for field, enum_cls in self._ENUM_TASK_FIELDS.items(): + if field in fields: + fields[field] = enum_cls(str(fields[field])) + + results: list[str] = [] + if fields: + await task_svc.update(task_id, **fields) + results.append(f"fields updated: {', '.join(sorted(fields))}") + if reassign_to is not _UNSET: + new_assignee = await self._resolve_assignee(reassign_to) + await self._reassign_task(task_svc, task_id, new_assignee) + results.append( + f"reassigned to {reassign_to}" + if new_assignee is not None + else "unassigned" + ) + return "; ".join(results) + + async def _reassign_task( + self, task_svc: TaskService, task_id: UUID, new_assignee: UUID | None + ) -> None: + """Route reassignment through the task service's claim-aware paths. + + An active claim (``claimed``/``in_progress``) reseeds the heartbeat + via ``reassign_active_claim`` so the new assignee isn't immediately + stale to the reaper; everything else (review-state handoffs, or an + explicit unassign) goes through the general ``reassign`` — never a + naive ``setattr`` on ``assigned_to``. + """ + if new_assignee is not None: + task = await task_svc.get(task_id) + if task is not None and task.status in ( + TaskStatus.CLAIMED, + TaskStatus.IN_PROGRESS, + ): + reassigned = await task_svc.reassign_active_claim(task_id, new_assignee) + if reassigned is not None: + return + await task_svc.reassign(task_id, new_assignee) + + async def _resolve_assignee(self, raw: Any) -> UUID | None: + """Resolve an edit's ``assigned_to`` value to an agent UUID. + + Accepts ``None`` (explicit unassign), a UUID string, or an agent slug + (e.g. ``"be-dev-1"``) — the same slug convention the REST PATCH path + resolves for the CEO's chat, which refers to agents by name. + """ + if raw is None: + return None + candidate = str(raw) + try: + return require_uuid(candidate) + except InvalidIdentifierError: + pass + agent_row = await get_agent_by_slug(self.session, candidate) + if agent_row is None: + raise ValidationError(f"no agent with slug or UUID {candidate!r}") + return require_uuid(agent_row.id) + async def _notify_ceo_pending(self, row: SecretaryDirectiveTable) -> None: from roboco.services.notification import NotificationService diff --git a/tests/integration/test_a2a_routes.py b/tests/integration/test_a2a_routes.py index a1afcbb1..3871d673 100644 --- a/tests/integration/test_a2a_routes.py +++ b/tests/integration/test_a2a_routes.py @@ -19,7 +19,7 @@ from roboco.api.routes.a2a import wellknown_router from roboco.db.tables import AgentTable, ProjectTable, TaskTable from roboco.enforcement import A2AAccessDeniedError from roboco.models import AgentRole, AgentStatus, Team -from roboco.models.a2a import A2ATask, A2ATaskState, A2ATaskStatus +from roboco.models.a2a import A2AAdminPairSummary, A2ATask, A2ATaskState, A2ATaskStatus from roboco.models.base import ( TaskNature, TaskStatus, @@ -35,6 +35,8 @@ if TYPE_CHECKING: _PAGE_TOKEN_OFFSET = 20 _MIN_STREAM_CHUNKS = 2 +_EXPECTED_PAIR_LIST_TOTAL = 2 +_EXPECTED_PAIR_MESSAGE_COUNT = 4 @pytest_asyncio.fixture @@ -1067,6 +1069,71 @@ async def test_admin_list_conversations_as_ceo(a2a_route_client: dict) -> None: assert body["items"][0]["agent_b"] == "fe-dev-1" +@pytest.mark.asyncio +async def test_admin_list_pairs_forbidden_for_non_ceo(a2a_route_client: dict) -> None: + client = a2a_route_client["client"] + response = await client.get("/api/a2a/chat/admin/pairs", headers=_HDR) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.asyncio +async def test_admin_list_pairs_as_ceo(a2a_route_client: dict) -> None: + """CEO gets the switchboard's pair cards — the static matrix joined with + each pair's representative conversation stats.""" + app = a2a_route_client["app"] + dev = a2a_route_client["dev"] + client = a2a_route_client["client"] + _set_ceo_context(app, dev) + + conv_id = uuid4() + pair_with_history = A2AAdminPairSummary( + agent_a="be-dev-1", + role_a="developer", + team_a="backend", + agent_b="be-qa", + role_b="qa", + team_b="backend", + group_key="cell-backend", + conversation_id=str(conv_id), + last_message_at=datetime.now(UTC), + message_count=4, + ) + pair_never_talked = A2AAdminPairSummary( + agent_a="auditor", + role_a="auditor", + team_a="board", + agent_b="product-owner", + role_b="product_owner", + team_b="board", + group_key="board", + conversation_id=None, + last_message_at=None, + message_count=0, + ) + with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: + instance = AsyncMock() + instance.list_admin_pairs = AsyncMock( + return_value=[pair_with_history, pair_never_talked] + ) + mock_service_cls.return_value = instance + response = await client.get("/api/a2a/chat/admin/pairs", headers=_HDR) + + assert response.status_code == HTTPStatus.OK + instance.list_admin_pairs.assert_awaited_once_with() + body = response.json() + assert body["total"] == _EXPECTED_PAIR_LIST_TOTAL + first = body["items"][0] + assert first["agent_a"] == "be-dev-1" + assert first["agent_b"] == "be-qa" + assert first["group_key"] == "cell-backend" + assert first["conversation_id"] == str(conv_id) + assert first["message_count"] == _EXPECTED_PAIR_MESSAGE_COUNT + second = body["items"][1] + assert second["group_key"] == "board" + assert second["conversation_id"] is None + assert second["message_count"] == 0 + + @pytest.mark.asyncio async def test_admin_get_messages_as_ceo_returns_full_transcript( a2a_route_client: dict, diff --git a/tests/integration/test_a2a_service.py b/tests/integration/test_a2a_service.py index c6324015..b59bf2b3 100644 --- a/tests/integration/test_a2a_service.py +++ b/tests/integration/test_a2a_service.py @@ -12,6 +12,7 @@ from uuid import uuid4 as _u import pytest import pytest_asyncio +from roboco.agents_config import A2A_ALLOWED_PAIRS from roboco.db.tables import ( A2AConversationTable, A2AMessageTable, @@ -637,6 +638,86 @@ async def test_get_conversation_admin_returns_none_for_unknown( assert await svc.get_conversation_admin(uuid4()) is None +# --------------------------------------------------------------------------- +# list_admin_pairs — the A2A switchboard's static-matrix + DB join +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_list_admin_pairs_bounded_by_static_matrix(a2a_setup: dict) -> None: + """With no conversations at all, every pair from the static matrix is + still returned (conversation-less), sized exactly to the matrix.""" + svc = a2a_setup["svc"] + pairs = await svc.list_admin_pairs() + + assert len(pairs) == len(A2A_ALLOWED_PAIRS) + assert all(p.conversation_id is None for p in pairs) + assert all(p.message_count == 0 for p in pairs) + assert all(p.last_message_at is None for p in pairs) + + +@pytest.mark.asyncio +async def test_list_admin_pairs_joins_representative_conversation( + a2a_setup: dict, +) -> None: + svc = a2a_setup["svc"] + conv = await svc.get_or_create_conversation("be-dev-1", "be-qa") + await svc.send_chat_message(UUID(conv.id), "be-dev-1", "hello") + + pairs = await svc.list_admin_pairs() + + match = next(p for p in pairs if {p.agent_a, p.agent_b} == {"be-dev-1", "be-qa"}) + assert match.conversation_id == conv.id + assert match.message_count == 1 + assert match.last_message_at is not None + assert match.group_key == "cell-backend" + assert match.role_a == "developer" + assert match.role_b == "qa" + + +@pytest.mark.asyncio +async def test_list_admin_pairs_picks_most_recently_updated_conversation( + a2a_setup: dict, +) -> None: + """A pair with two conversations (distinct topics) surfaces the more + recently active one as its representative conversation.""" + svc = a2a_setup["svc"] + db = a2a_setup["db"] + conv_old = await svc.get_or_create_conversation("be-dev-1", "be-qa", topic="t1") + conv_new = await svc.get_or_create_conversation("be-dev-1", "be-qa", topic="t2") + + now = datetime.now(UTC) + row_old = await db.get(A2AConversationTable, UUID(conv_old.id)) + row_new = await db.get(A2AConversationTable, UUID(conv_new.id)) + assert row_old is not None + assert row_new is not None + row_old.updated_at = now - timedelta(minutes=10) + row_new.updated_at = now + await db.flush() + + pairs = await svc.list_admin_pairs() + + match = next(p for p in pairs if {p.agent_a, p.agent_b} == {"be-dev-1", "be-qa"}) + assert match.conversation_id == conv_new.id + + +@pytest.mark.asyncio +async def test_list_admin_pairs_excludes_disallowed_pairs(a2a_setup: dict) -> None: + """A conversation row between two agents the matrix does NOT allow (dev + A2A is same-cell only — this should never legitimately exist, but the + join must be robust against it) never surfaces as a pair card: the + service iterates the static matrix, not "any conversation row".""" + svc = a2a_setup["svc"] + db = a2a_setup["db"] + stray = A2AConversationTable(agent_a="be-dev-1", agent_b="fe-dev-1") + db.add(stray) + await db.flush() + + pairs = await svc.list_admin_pairs() + + assert not any({p.agent_a, p.agent_b} == {"be-dev-1", "fe-dev-1"} for p in pairs) + + # --------------------------------------------------------------------------- # CEO reply-only budget — an agent may only reply to the CEO inside a # conversation the CEO itself opened, and only up to the CEO's own message diff --git a/tests/integration/test_tasks_route_pm_lighter_patch.py b/tests/integration/test_tasks_route_pm_lighter_patch.py new file mode 100644 index 00000000..6313d4dc --- /dev/null +++ b/tests/integration/test_tasks_route_pm_lighter_patch.py @@ -0,0 +1,241 @@ +"""PM-lighter PATCH surface on PATCH /api/tasks/{id}. + +The CEO's spec: PMs get a *lighter* content-only slice of the same REST PATCH +path the Secretary's edit directive uses — title/description/ +acceptance_criteria/priority — scoped to tasks in the PM's remit (cell_pm: +own team only; main_pm: any team). No status changes, no structural/ +ownership fields, no git fields — those stay on the lifecycle-verb surface. + +cell_pm/main_pm already hold TaskAction.ASSIGN (see TASK_PERMISSIONS), which +is *not* team-scoped in ``can_perform_task_action`` — so absent this gate a +PM would already ride the CEO/Board/Auditor "full admin" bypass on this +route (any field, any team). These tests pin the narrower behavior down. +""" + +from __future__ import annotations + +from http import HTTPStatus +from typing import TYPE_CHECKING, Any, cast +from uuid import UUID, uuid4 + +import pytest +import pytest_asyncio +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from roboco.api.deps import get_agent_context, get_db +from roboco.api.routes.tasks import router as tasks_router +from roboco.db.tables import AgentTable, ProjectTable, TaskTable +from roboco.models import AgentRole, AgentStatus, Team +from roboco.models.base import TaskNature, TaskStatus, TaskType +from roboco.models.permissions import AgentContext + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from sqlalchemy.ext.asyncio import AsyncSession + + +async def _make_client( + db_session: AsyncSession, *, role: AgentRole, team: Team | None +) -> dict[str, Any]: + pm = AgentTable( + id=uuid4(), + name="PM", + slug=f"pm-{uuid4().hex[:8]}", + role=role, + team=team, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt="pm", + capabilities=[], + permissions={}, + metrics={}, + ) + db_session.add(pm) + await db_session.flush() + project = ProjectTable( + id=uuid4(), + name="PL-Proj", + slug=f"pl-proj-{uuid4().hex[:6]}", + git_url="https://example.com/pl.git", + assigned_cell=Team.BACKEND, + created_by=pm.id, + ) + db_session.add(project) + await db_session.flush() + + app = FastAPI() + app.include_router(tasks_router, prefix="/api/tasks") + + async def _override_db() -> AsyncIterator[AsyncSession]: + yield db_session + + async def _override_agent() -> AgentContext: + return AgentContext(agent_id=cast("UUID", pm.id), role=role, team=team) + + app.dependency_overrides[get_db] = _override_db + app.dependency_overrides[get_agent_context] = _override_agent + + transport = ASGITransport(app=app) + client = AsyncClient(transport=transport, base_url="http://test") + return { + "client": client, + "app": app, + "agent": pm, + "project": project, + "db": db_session, + } + + +@pytest_asyncio.fixture +async def cell_pm_client(db_session: AsyncSession) -> AsyncIterator[dict]: + """A backend cell PM (ASSIGN, no UPDATE_OWN).""" + setup = await _make_client(db_session, role=AgentRole.CELL_PM, team=Team.BACKEND) + async with setup["client"]: + yield setup + setup["app"].dependency_overrides.clear() + + +@pytest_asyncio.fixture +async def main_pm_client(db_session: AsyncSession) -> AsyncIterator[dict]: + """The Main PM (ASSIGN, no team restriction).""" + setup = await _make_client(db_session, role=AgentRole.MAIN_PM, team=None) + async with setup["client"]: + yield setup + setup["app"].dependency_overrides.clear() + + +def _seed_task(setup: dict, **kw: Any) -> TaskTable: + team = kw.pop("team", Team.BACKEND) + task = TaskTable( + id=uuid4(), + title=kw.pop("title", "t"), + description=kw.pop("description", "d"), + acceptance_criteria=["ac"], + status=kw.pop("status", TaskStatus.IN_PROGRESS), + priority=2, + task_type=TaskType.CODE, + nature=TaskNature.TECHNICAL, + project_id=setup["project"].id, + created_by=setup["agent"].id, + assigned_to=None, + team=team, + ) + setup["db"].add(task) + return task + + +def _hdr(agent: AgentTable, role: AgentRole) -> dict[str, str]: + return {"X-Agent-ID": str(agent.id), "X-Agent-Role": role.value} + + +@pytest.mark.asyncio +async def test_cell_pm_can_patch_content_field_on_own_team_task( + cell_pm_client: dict, +) -> None: + setup = cell_pm_client + task = _seed_task(setup, team=Team.BACKEND) + await setup["db"].flush() + response = await setup["client"].patch( + f"/api/tasks/{task.id}", + json={"title": "Sharper title from the cell PM"}, + headers=_hdr(setup["agent"], AgentRole.CELL_PM), + ) + assert response.status_code == HTTPStatus.OK + assert response.json()["title"] == "Sharper title from the cell PM" + + +@pytest.mark.asyncio +async def test_cell_pm_cannot_patch_task_outside_own_team( + cell_pm_client: dict, +) -> None: + """ASSIGN is not team-scoped — without an explicit check a cell PM would + ride the same admin bypass CEO/Board get on ANY team's task.""" + setup = cell_pm_client + task = _seed_task(setup, team=Team.FRONTEND) + await setup["db"].flush() + response = await setup["client"].patch( + f"/api/tasks/{task.id}", + json={"title": "Should not land"}, + headers=_hdr(setup["agent"], AgentRole.CELL_PM), + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.parametrize( + "field,value", + [ + ("dev_notes", "trying to sneak a note in"), + ("team", "frontend"), + ("assigned_to", str(uuid4())), + ], +) +@pytest.mark.asyncio +async def test_cell_pm_cannot_patch_fields_outside_lighter_allowlist( + cell_pm_client: dict, field: str, value: object +) -> None: + setup = cell_pm_client + task = _seed_task(setup, team=Team.BACKEND) + await setup["db"].flush() + response = await setup["client"].patch( + f"/api/tasks/{task.id}", + json={field: value}, + headers=_hdr(setup["agent"], AgentRole.CELL_PM), + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.asyncio +async def test_cell_pm_cannot_change_status_via_patch(cell_pm_client: dict) -> None: + """No status changes beyond what the lifecycle verbs already grant — the + PM-lighter PATCH surface must not become a side-door status override.""" + setup = cell_pm_client + task = _seed_task(setup, team=Team.BACKEND, status=TaskStatus.IN_PROGRESS) + await setup["db"].flush() + response = await setup["client"].patch( + f"/api/tasks/{task.id}", + json={"status": "completed", "force": True}, + headers=_hdr(setup["agent"], AgentRole.CELL_PM), + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.asyncio +async def test_main_pm_can_patch_content_field_on_any_team_task( + main_pm_client: dict, +) -> None: + setup = main_pm_client + task = _seed_task(setup, team=Team.UX_UI) + await setup["db"].flush() + response = await setup["client"].patch( + f"/api/tasks/{task.id}", + json={"description": "A clarified description of at least twenty chars."}, + headers=_hdr(setup["agent"], AgentRole.MAIN_PM), + ) + assert response.status_code == HTTPStatus.OK + + +@pytest.mark.asyncio +async def test_main_pm_cannot_patch_privileged_field(main_pm_client: dict) -> None: + setup = main_pm_client + task = _seed_task(setup, team=Team.BACKEND) + await setup["db"].flush() + response = await setup["client"].patch( + f"/api/tasks/{task.id}", + json={"parent_task_id": str(uuid4())}, + headers=_hdr(setup["agent"], AgentRole.MAIN_PM), + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.asyncio +async def test_main_pm_cannot_change_status_via_patch(main_pm_client: dict) -> None: + setup = main_pm_client + task = _seed_task(setup, status=TaskStatus.AWAITING_PM_REVIEW) + await setup["db"].flush() + response = await setup["client"].patch( + f"/api/tasks/{task.id}", + json={"status": "completed"}, + headers=_hdr(setup["agent"], AgentRole.MAIN_PM), + ) + assert response.status_code == HTTPStatus.FORBIDDEN diff --git a/tests/integration/test_tasks_routes.py b/tests/integration/test_tasks_routes.py index 9271649b..b0aed703 100644 --- a/tests/integration/test_tasks_routes.py +++ b/tests/integration/test_tasks_routes.py @@ -102,6 +102,7 @@ async def task_client( "agent": main_pm, "project": project, "db": db_session, + "app": app, } app.dependency_overrides.clear() @@ -109,6 +110,24 @@ async def task_client( _HDR = {"X-Agent-ID": str(uuid4()), "X-Agent-Role": "main_pm"} +def _as_ceo(setup: dict) -> None: + """Re-override this client's agent identity to CEO for the rest of the + test — the general PATCH admin surface (status overrides, structural + fields, force hatches, ...) is CEO/Board/Auditor-only now that + cell_pm/main_pm get the narrower "PM lighter" content-only slice (see + test_tasks_route_pm_lighter_patch.py). ``task_client``'s default agent + stays main_pm so the many other tests that specifically exercise + PM-role behavior (or "not CEO" refusals) are unaffected. + """ + + async def _override_ceo() -> AgentContext: + return AgentContext( + agent_id=cast("UUID", setup["agent"].id), role=AgentRole.CEO, team=None + ) + + setup["app"].dependency_overrides[get_agent_context] = _override_ceo + + def _seed_task( setup: dict, *, status: TaskStatus = TaskStatus.PENDING, **kw: Any ) -> TaskTable: @@ -278,6 +297,7 @@ async def test_update_task_status_override_recovers_blocked(task_client: dict) - override, so an operator can recover a task wedged in ``blocked`` (which ``/complete`` refuses) instead of the status being silently dropped. The ``force`` flag acknowledges the bypass past the lifecycle gate (#13).""" + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.BLOCKED) await task_client["db"].flush() @@ -297,6 +317,7 @@ async def test_update_task_status_override_refused_without_force( """#13: pasting over the lifecycle gate into a terminal/final hatch state (completed / awaiting_qa / awaiting_pm_review) without ``force`` is refused with 400 — the bypass must be an explicit, acknowledged forced override.""" + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.IN_PROGRESS) await task_client["db"].flush() @@ -316,6 +337,7 @@ async def test_update_task_status_override_non_hatch_needs_no_force( ) -> None: """#13: a non-terminal recovery override (blocked -> pending) does NOT require ``force`` — only the terminal/final hatch states do.""" + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.BLOCKED) await task_client["db"].flush() @@ -336,6 +358,7 @@ async def test_update_task_override_gate_states_require_force( """The hatch set covers the CEO gate and the terminal cancel too (not just completed/awaiting_qa/awaiting_pm_review): a privileged PATCH into either without ``force`` is refused 400.""" + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.IN_PROGRESS) await task_client["db"].flush() @@ -353,6 +376,7 @@ async def test_update_task_override_gate_states_require_force( async def test_update_task_override_gate_states_with_force_succeeds( task_client: dict, hatch: str ) -> None: + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.IN_PROGRESS) await task_client["db"].flush() @@ -370,6 +394,7 @@ async def test_update_task_resurrect_terminal_requires_force(task_client: dict) """Resurrecting a COMPLETED task back to in_progress is a bypass of the merge decision; the target (in_progress) is not itself a hatch state, so the target-only gate would miss it — the source-terminal check requires force.""" + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.COMPLETED) await task_client["db"].flush() @@ -416,6 +441,7 @@ async def test_admin_complete_with_open_pr_names_the_pr(task_client: dict) -> No """Admin status→completed on a task whose PR is still OPEN strands its commits (bit the CEO twice live, 2026-07-02). The refusal must name the PR and the stranding — not just the generic lifecycle-gate text.""" + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.AWAITING_CEO_APPROVAL) await task_client["db"].flush() @@ -438,6 +464,7 @@ async def test_admin_complete_with_open_pr_force_still_escapes( ) -> None: """``force`` remains the deliberate, audited escape — an operator who KNOWS the PR should be stranded can still complete.""" + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.AWAITING_CEO_APPROVAL) await task_client["db"].flush() @@ -457,6 +484,7 @@ async def test_admin_complete_with_merged_pr_gets_generic_gate_only( ) -> None: """A merged PR strands nothing — the refusal stays the generic hatch text (no PR callout), and force completes as before.""" + _as_ceo(task_client) client = task_client["client"] task = _seed_task(task_client, status=TaskStatus.AWAITING_CEO_APPROVAL) await task_client["db"].flush() @@ -3163,6 +3191,7 @@ async def test_get_sessions_for_task_not_found(task_client: dict) -> None: @pytest.mark.asyncio async def test_patch_nature_persists(task_client: dict) -> None: """PATCH with nature=non_technical persists; GET returns updated value.""" + _as_ceo(task_client) task = _seed_task(task_client, nature=TaskNature.TECHNICAL) await task_client["db"].flush() response = await task_client["client"].patch( @@ -3178,6 +3207,7 @@ async def test_patch_nature_persists(task_client: dict) -> None: @pytest.mark.asyncio async def test_patch_task_type_persists(task_client: dict) -> None: """PATCH with task_type=research persists; GET returns updated value.""" + _as_ceo(task_client) task = _seed_task(task_client, task_type=TaskType.CODE) await task_client["db"].flush() response = await task_client["client"].patch( @@ -3193,6 +3223,7 @@ async def test_patch_task_type_persists(task_client: dict) -> None: @pytest.mark.asyncio async def test_patch_project_id_persists(task_client: dict) -> None: """PATCH with project_id= persists; GET returns updated value.""" + _as_ceo(task_client) task = _seed_task(task_client) # Create a second project to switch to second_project = ProjectTable( @@ -3250,6 +3281,7 @@ async def test_patch_title_only_changes_title(task_client: dict) -> None: @pytest.mark.asyncio async def test_patch_assigned_to_slug_resolves_to_uuid(task_client: dict) -> None: """PATCH assigned_to with agent slug resolves to agent UUID.""" + _as_ceo(task_client) dev = await _seed_agent(task_client) task = _seed_task(task_client) await task_client["db"].flush() @@ -3267,6 +3299,7 @@ async def test_patch_assigned_to_slug_resolves_to_uuid(task_client: dict) -> Non @pytest.mark.asyncio async def test_patch_assigned_to_null_unassigns(task_client: dict) -> None: """PATCH assigned_to: null sets assigned_to to null (unassign).""" + _as_ceo(task_client) dev = await _seed_agent(task_client) task = _seed_task(task_client, assigned_to=dev.id) await task_client["db"].flush() diff --git a/tests/unit/runtime/test_intake_spawn.py b/tests/unit/runtime/test_intake_spawn.py index 4976c183..306c8ea9 100644 --- a/tests/unit/runtime/test_intake_spawn.py +++ b/tests/unit/runtime/test_intake_spawn.py @@ -17,6 +17,7 @@ from unittest.mock import patch from uuid import UUID, uuid4 import pytest +from roboco.config import settings from roboco.runtime.orchestrator import ( INTAKE_AGENT_ID, AgentInstance, @@ -265,7 +266,7 @@ class TestIntakeScopeSlugs: # --------------------------------------------------------------------------- # _resolve_history_digest_projects — the prompter-memory ambient's project scope -# (covers all three intake scopes, unlike the conventions ambient resolver). +# (covers all three intake scopes: project_slug, product_id, project_ids). # --------------------------------------------------------------------------- @@ -375,6 +376,284 @@ class TestResolveHistoryDigestAmbient: assert result is None +# --------------------------------------------------------------------------- +# _resolve_ambient_projects — the conventions ambient's project scope. Mirrors +# _resolve_history_digest_projects's shape, including the MegaTask project_ids +# scope (the pre-existing gap: this resolver used to stop at project_slug / +# product_id / task_id and never saw a MegaTask's explicit project_ids). +# --------------------------------------------------------------------------- + + +class TestResolveAmbientProjects: + @pytest.mark.asyncio + async def test_project_slug_branch_resolves_single_project(self) -> None: + orch = _make_minimal_orchestrator() + + class _FakeProjectSvc: + async def get_by_slug(self, slug: str) -> Any: + return SimpleNamespace(slug=slug, id=uuid4()) + + with patch( + "roboco.services.project.get_project_service", + lambda _db: _FakeProjectSvc(), + ): + projects = await orch._resolve_ambient_projects( + object(), + project_slug="roboco", + task_id=None, + product_id=None, + ) + assert [p.slug for p in projects] == ["roboco"] + + @pytest.mark.asyncio + async def test_project_slug_missing_returns_empty(self) -> None: + orch = _make_minimal_orchestrator() + + class _FakeProjectSvc: + async def get_by_slug(self, _slug: str) -> Any: + return None + + with patch( + "roboco.services.project.get_project_service", + lambda _db: _FakeProjectSvc(), + ): + projects = await orch._resolve_ambient_projects( + object(), + project_slug="ghost", + task_id=None, + product_id=None, + ) + assert projects == [] + + @pytest.mark.asyncio + async def test_product_id_branch_delegates_to_ambient_product_projects( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + orch = _make_minimal_orchestrator() + sentinel = [SimpleNamespace(slug="p1", id=uuid4())] + + async def _fake_product_projects(_db: Any, product_id: str) -> list[Any]: + assert product_id == "prod-1" + return sentinel + + # Patched on the instance (not the class): _ambient_product_projects is + # a staticmethod, so accessing it via `self.` never binds `self` — but a + # plain function patched onto the *class* would, since it's no longer + # wrapped in `staticmethod`. Patching the instance attribute sidesteps + # the descriptor lookup entirely. + monkeypatch.setattr(orch, "_ambient_product_projects", _fake_product_projects) + projects = await orch._resolve_ambient_projects( + object(), + project_slug=None, + task_id=None, + product_id="prod-1", + ) + assert projects is sentinel + + @pytest.mark.asyncio + async def test_project_ids_branch_preserves_order_and_skips_missing( + self, + ) -> None: + """The MegaTask scope: an explicit project_ids list must resolve into + projects (order preserved, unresolvable ids skipped) — this is the gap + fix, mirroring the history digest resolver's own project_ids branch.""" + orch = _make_minimal_orchestrator() + good1 = "11111111-1111-1111-1111-111111111111" + missing = "22222222-2222-2222-2222-222222222222" + good2 = "33333333-3333-3333-3333-333333333333" + + class _FakeProjectSvc: + async def get(self, pid: Any) -> Any: + if str(pid) == missing: + return None + return SimpleNamespace(slug=f"proj-{str(pid)[0]}", id=pid) + + with patch( + "roboco.services.project.get_project_service", + lambda _db: _FakeProjectSvc(), + ): + projects = await orch._resolve_ambient_projects( + object(), + project_slug=None, + task_id=None, + product_id=None, + project_ids=[good1, missing, good2], + ) + assert [p.slug for p in projects] == ["proj-1", "proj-3"] + + @pytest.mark.asyncio + async def test_project_ids_takes_priority_over_other_scopes( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A MegaTask spawn passes project_slug/product_id as None in practice, + but the resolver must still prefer the explicit project_ids set over a + stray product_id/task_id if both were somehow present.""" + orch = _make_minimal_orchestrator() + pid = "11111111-1111-1111-1111-111111111111" + + class _FakeProjectSvc: + async def get(self, _pid: Any) -> Any: + return SimpleNamespace(slug="from-ids", id=_pid) + + async def _fail_product_projects(*_a: Any, **_k: Any) -> list[Any]: + raise AssertionError("product_id branch must not run") + + monkeypatch.setattr(orch, "_ambient_product_projects", _fail_product_projects) + with patch( + "roboco.services.project.get_project_service", + lambda _db: _FakeProjectSvc(), + ): + projects = await orch._resolve_ambient_projects( + object(), + project_slug=None, + task_id=None, + product_id="prod-should-be-ignored", + project_ids=[pid], + ) + assert [p.slug for p in projects] == ["from-ids"] + + @pytest.mark.asyncio + async def test_no_scope_given_returns_empty(self) -> None: + orch = _make_minimal_orchestrator() + projects = await orch._resolve_ambient_projects( + object(), + project_slug=None, + task_id=None, + product_id=None, + ) + assert projects == [] + + +# --------------------------------------------------------------------------- +# _resolve_conventions_ambient — flag-gated + best-effort, now MegaTask-aware. +# --------------------------------------------------------------------------- + + +class TestResolveConventionsAmbient: + @pytest.mark.asyncio + async def test_flag_off_returns_none(self, monkeypatch: pytest.MonkeyPatch) -> None: + orch = _make_minimal_orchestrator() + monkeypatch.setattr(settings, "conventions_enabled", False) + + result = await orch._resolve_conventions_ambient( + "roboco", project_ids=["11111111-1111-1111-1111-111111111111"] + ) + assert result is None + + @pytest.mark.asyncio + async def test_failure_returns_none_not_raises( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + orch = _make_minimal_orchestrator() + monkeypatch.setattr(settings, "conventions_enabled", True) + + def _boom() -> Any: + raise RuntimeError("db unavailable") + + monkeypatch.setattr("roboco.db.base.get_session_factory", _boom) + + result = await orch._resolve_conventions_ambient( + None, project_ids=["11111111-1111-1111-1111-111111111111"] + ) + assert result is None + + @pytest.mark.asyncio + async def test_project_ids_scope_reaches_conventions_layer( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The MegaTask project_ids scope must flow all the way through to + conventions_ambient_layer — the actual gap this fix closes.""" + orch = _make_minimal_orchestrator() + monkeypatch.setattr(settings, "conventions_enabled", True) + + class _FakeFactory: + def __call__(self) -> Any: + return self + + async def __aenter__(self) -> Any: + return "fake-db" + + async def __aexit__(self, *_a: Any) -> None: + return None + + monkeypatch.setattr("roboco.db.base.get_session_factory", _FakeFactory) + + sentinel_projects = [SimpleNamespace(slug="proj-1")] + captured: dict[str, Any] = {} + + async def _fake_resolve_projects(_self: Any, _db: Any, **kwargs: Any) -> Any: + captured["kwargs"] = kwargs + return sentinel_projects + + async def _fake_layer(_db: Any, projects: Any) -> str: + captured["projects"] = projects + return "RENDERED BLOCK" + + monkeypatch.setattr( + AgentOrchestrator, "_resolve_ambient_projects", _fake_resolve_projects + ) + monkeypatch.setattr( + "roboco.agents.factories._base.conventions_ambient_layer", _fake_layer + ) + + result = await orch._resolve_conventions_ambient( + None, project_ids=["11111111-1111-1111-1111-111111111111"] + ) + assert result == "RENDERED BLOCK" + assert captured["kwargs"]["project_ids"] == [ + "11111111-1111-1111-1111-111111111111" + ] + assert captured["projects"] is sentinel_projects + + +# --------------------------------------------------------------------------- +# _resolve_intake_ambient — must forward project_ids to BOTH sub-resolvers. +# Regression test for the gap: it used to thread project_ids only to the +# history-digest resolver, leaving a MegaTask intake with no conventions block. +# --------------------------------------------------------------------------- + + +class TestResolveIntakeAmbientThreadsProjectIds: + @pytest.mark.asyncio + async def test_project_ids_forwarded_to_conventions_and_history( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + orch = _make_minimal_orchestrator() + conventions_calls: list[dict[str, Any]] = [] + history_calls: list[dict[str, Any]] = [] + + async def _conventions(_slug: Any, **kwargs: Any) -> str | None: + conventions_calls.append(kwargs) + return "CONVENTIONS" + + async def _history(_slug: Any, **kwargs: Any) -> str | None: + history_calls.append(kwargs) + return "HISTORY" + + monkeypatch.setattr(orch, "_resolve_conventions_ambient", _conventions) + monkeypatch.setattr(orch, "_resolve_history_digest_ambient", _history) + + result = await orch._resolve_intake_ambient( + None, + product_id=None, + project_ids=["11111111-1111-1111-1111-111111111111"], + ) + + assert result == "CONVENTIONS\n\n---\n\nHISTORY" + assert conventions_calls == [ + { + "product_id": None, + "project_ids": ["11111111-1111-1111-1111-111111111111"], + } + ] + assert history_calls == [ + { + "product_id": None, + "project_ids": ["11111111-1111-1111-1111-111111111111"], + } + ] + + # --------------------------------------------------------------------------- # spawn_intake_session / reap_intake_session — orchestration (docker mocked). # --------------------------------------------------------------------------- diff --git a/tests/unit/services/test_secretary_service.py b/tests/unit/services/test_secretary_service.py index 88e57733..306777e1 100644 --- a/tests/unit/services/test_secretary_service.py +++ b/tests/unit/services/test_secretary_service.py @@ -2,12 +2,14 @@ from __future__ import annotations +from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 import pytest from roboco.db.tables import SecretaryDirectiveTable +from roboco.models.base import Complexity, TaskNature, TaskStatus, Team from roboco.models.secretary import DirectiveKind, DirectiveStatus from roboco.services import secretary as sec_module from roboco.services.base import ValidationError @@ -35,7 +37,12 @@ def _patch(monkeypatch: pytest.MonkeyPatch) -> dict[str, MagicMock]: task.approve_and_start = AsyncMock() task.admin_set_status = AsyncMock() task.update = AsyncMock() + task.get = AsyncMock(return_value=None) + task.reassign = AsyncMock() + task.reassign_active_claim = AsyncMock() monkeypatch.setattr(sec_module, "get_task_service", lambda _s: task) + agent_lookup = AsyncMock(return_value=None) + monkeypatch.setattr(sec_module, "get_agent_by_slug", agent_lookup) notifier = MagicMock() notifier.send_ack_notification = AsyncMock() monkeypatch.setattr( @@ -47,6 +54,7 @@ def _patch(monkeypatch: pytest.MonkeyPatch) -> dict[str, MagicMock]: "pitch": pitch, "task": task, "notifier": notifier, + "agent_lookup": agent_lookup, } @@ -224,3 +232,241 @@ async def test_control_task_edit_rejects_non_allowlisted_fields( out = await svc.confirm_directive(row.id, uuid4()) assert out.status == DirectiveStatus.FAILED.value svcs["task"].update.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_confirm_control_task_edit_extended_content_fields( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Secretary FULL: team/estimated_complexity/nature ride the edit too, + coerced into their proper enums before hitting TaskService.update.""" + svcs = _patch(monkeypatch) + svc = SecretaryService(_session()) + tid = uuid4() + row = _pending( + DirectiveKind.CONTROL_TASK, + { + "task_id": str(tid), + "action": "edit", + "fields": { + "team": "frontend", + "estimated_complexity": "high", + "nature": "technical", + }, + }, + ) + monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row)) + out = await svc.confirm_directive(row.id, uuid4()) + assert out.status == DirectiveStatus.EXECUTED.value + svcs["task"].update.assert_awaited_once() + _, kwargs = svcs["task"].update.await_args + assert kwargs["team"] == Team.FRONTEND + assert kwargs["estimated_complexity"] == Complexity.HIGH + assert kwargs["nature"] == TaskNature.TECHNICAL + + +@pytest.mark.asyncio +async def test_confirm_control_task_edit_bad_enum_value_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + svcs = _patch(monkeypatch) + svc = SecretaryService(_session()) + row = _pending( + DirectiveKind.CONTROL_TASK, + { + "task_id": str(uuid4()), + "action": "edit", + "fields": {"team": "not-a-real-team"}, + }, + ) + monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row)) + out = await svc.confirm_directive(row.id, uuid4()) + assert out.status == DirectiveStatus.FAILED.value + svcs["task"].update.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_confirm_control_task_edit_reassigns_active_claim( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reassigning an active (claimed/in_progress) task goes through + ``reassign_active_claim`` so the new assignee reseeds its heartbeat and + isn't immediately stale to the reaper — not a naive setattr.""" + svcs = _patch(monkeypatch) + new_assignee = uuid4() + svcs["task"].get = AsyncMock( + return_value=SimpleNamespace(status=TaskStatus.IN_PROGRESS) + ) + svcs["task"].reassign_active_claim = AsyncMock( + return_value=SimpleNamespace(status=TaskStatus.IN_PROGRESS) + ) + svc = SecretaryService(_session()) + tid = uuid4() + row = _pending( + DirectiveKind.CONTROL_TASK, + { + "task_id": str(tid), + "action": "edit", + "fields": {"assigned_to": str(new_assignee)}, + }, + ) + monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row)) + out = await svc.confirm_directive(row.id, uuid4()) + assert out.status == DirectiveStatus.EXECUTED.value + svcs["task"].reassign_active_claim.assert_awaited_once_with(tid, new_assignee) + svcs["task"].reassign.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_confirm_control_task_edit_reassigns_non_active_task( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A non-active task (e.g. pending) reassigns through the general + ``reassign`` path — no heartbeat to reseed.""" + svcs = _patch(monkeypatch) + new_assignee = uuid4() + svcs["task"].get = AsyncMock( + return_value=SimpleNamespace(status=TaskStatus.PENDING) + ) + svc = SecretaryService(_session()) + tid = uuid4() + row = _pending( + DirectiveKind.CONTROL_TASK, + { + "task_id": str(tid), + "action": "edit", + "fields": {"assigned_to": str(new_assignee)}, + }, + ) + monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row)) + out = await svc.confirm_directive(row.id, uuid4()) + assert out.status == DirectiveStatus.EXECUTED.value + svcs["task"].reassign.assert_awaited_once_with(tid, new_assignee) + svcs["task"].reassign_active_claim.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_confirm_control_task_edit_reassigns_by_slug( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The CEO refers to agents by slug (e.g. 'be-dev-1'); the edit resolves + it to a UUID the same way the REST PATCH path does.""" + svcs = _patch(monkeypatch) + agent_row_id = uuid4() + svcs["agent_lookup"].return_value = SimpleNamespace(id=agent_row_id) + svcs["task"].get = AsyncMock( + return_value=SimpleNamespace(status=TaskStatus.PENDING) + ) + svc = SecretaryService(_session()) + tid = uuid4() + row = _pending( + DirectiveKind.CONTROL_TASK, + { + "task_id": str(tid), + "action": "edit", + "fields": {"assigned_to": "be-dev-1"}, + }, + ) + monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row)) + out = await svc.confirm_directive(row.id, uuid4()) + assert out.status == DirectiveStatus.EXECUTED.value + svcs["task"].reassign.assert_awaited_once_with(tid, agent_row_id) + + +@pytest.mark.asyncio +async def test_confirm_control_task_edit_unknown_assignee_slug_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + svcs = _patch(monkeypatch) + svc = SecretaryService(_session()) + row = _pending( + DirectiveKind.CONTROL_TASK, + { + "task_id": str(uuid4()), + "action": "edit", + "fields": {"assigned_to": "no-such-agent"}, + }, + ) + monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row)) + out = await svc.confirm_directive(row.id, uuid4()) + assert out.status == DirectiveStatus.FAILED.value + svcs["task"].reassign.assert_not_awaited() + svcs["task"].reassign_active_claim.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_confirm_control_task_edit_combines_fields_and_reassign( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A single edit directive may both update content fields and reassign.""" + svcs = _patch(monkeypatch) + new_assignee = uuid4() + svcs["task"].get = AsyncMock( + return_value=SimpleNamespace(status=TaskStatus.PENDING) + ) + svc = SecretaryService(_session()) + tid = uuid4() + row = _pending( + DirectiveKind.CONTROL_TASK, + { + "task_id": str(tid), + "action": "edit", + "fields": {"title": "Renamed", "assigned_to": str(new_assignee)}, + }, + ) + monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row)) + out = await svc.confirm_directive(row.id, uuid4()) + assert out.status == DirectiveStatus.EXECUTED.value + svcs["task"].update.assert_awaited_once() + _, kwargs = svcs["task"].update.await_args + assert kwargs == {"title": "Renamed"} + svcs["task"].reassign.assert_awaited_once_with(tid, new_assignee) + + +_SEEDED_PR_NUMBER = 42 +_SEEDED_PROGRESS_UPDATE_COUNT = sec_module._MAX_PROGRESS_UPDATES + 10 + + +@pytest.mark.asyncio +async def test_read_task_includes_full_detail(monkeypatch: pytest.MonkeyPatch) -> None: + """Secretary FULL read breadth: notes/progress/plan/pr fields join the + brief identity fields that read_task already carried.""" + svcs = _patch(monkeypatch) + tid = uuid4() + fake_task = SimpleNamespace( + id=tid, + title="Ship the thing", + status=TaskStatus.IN_PROGRESS, + team="backend", + assigned_to=uuid4(), + description="A real description.", + acceptance_criteria=["works"], + priority=1, + estimated_complexity="high", + nature="technical", + plan={"approach": "do it"}, + progress_updates=[ + {"message": f"update {i}"} for i in range(_SEEDED_PROGRESS_UPDATE_COUNT) + ], + dev_notes="dev note", + qa_notes="qa note", + auditor_notes="audit note", + pr_reviewer_notes="pr note", + doc_notes="doc note", + quick_context="ctx", + branch_name="feature/x", + pr_number=_SEEDED_PR_NUMBER, + pr_url="https://example.com/pr/42", + ) + svcs["task"].get = AsyncMock(return_value=fake_task) + svc = SecretaryService(_session()) + out = await svc.read_task(tid) + assert out["title"] == "Ship the thing" + assert out["plan"] == {"approach": "do it"} + assert out["dev_notes"] == "dev note" + assert out["pr_number"] == _SEEDED_PR_NUMBER + assert out["branch_name"] == "feature/x" + # Bounded: only the most recent entries survive. + assert len(out["progress_updates"]) == sec_module._MAX_PROGRESS_UPDATES + last_index = _SEEDED_PROGRESS_UPDATE_COUNT - 1 + assert out["progress_updates"][-1]["message"] == f"update {last_index}" diff --git a/tests/unit/test_agents_config.py b/tests/unit/test_agents_config.py index 1960dd55..652aa474 100644 --- a/tests/unit/test_agents_config.py +++ b/tests/unit/test_agents_config.py @@ -2,9 +2,11 @@ from __future__ import annotations +from collections import Counter from typing import TYPE_CHECKING from roboco.agents_config import ( + A2A_ALLOWED_PAIRS, can_a2a_direct, can_assign_tasks, can_cancel_tasks, @@ -27,6 +29,7 @@ from roboco.agents_config import ( issue_panel_token, verify_agent_token, ) +from roboco.foundation import identity as foundation from roboco.seeds.initial_data import CEO_AGENT_ID if TYPE_CHECKING: @@ -441,3 +444,85 @@ def test_get_a2a_route_hint_unknown_from_agent_falls_through() -> None: """from_agent with no team falls through to escalate fallback (line 774).""" hint = get_a2a_route_hint("ghost", "be-dev-1") assert "escalate" in hint.lower() + + +# --------------------------------------------------------------------------- +# A2A_ALLOWED_PAIRS — the switchboard's static org-chart pair matrix +# --------------------------------------------------------------------------- + +_EXPECTED_PAIR_COUNT = 70 +_EXPECTED_GROUP_COUNTS = { + "board": 3, + "cell-backend": 15, + "cell-frontend": 15, + "cell-ux_ui": 15, + "cross": 16, + "pm-chain": 6, +} + + +def test_a2a_allowed_pairs_total_count() -> None: + assert len(A2A_ALLOWED_PAIRS) == _EXPECTED_PAIR_COUNT + + +def test_a2a_allowed_pairs_canonical_lexical_order() -> None: + """agent_a < agent_b always — matches A2AConversationTable's canonical + ordering, so the service's DB join keys line up.""" + for p in A2A_ALLOWED_PAIRS: + assert p.agent_a < p.agent_b + + +def test_a2a_allowed_pairs_no_duplicates() -> None: + keys = [(p.agent_a, p.agent_b) for p in A2A_ALLOWED_PAIRS] + assert len(keys) == len(set(keys)) + + +def test_a2a_allowed_pairs_excludes_human_only_and_sentinel_roles() -> None: + """CEO, the intake interviewer, the secretary, and the system sentinel + are not real A2A participants in the org chart.""" + slugs = {p.agent_a for p in A2A_ALLOWED_PAIRS} | { + p.agent_b for p in A2A_ALLOWED_PAIRS + } + for excluded in ("ceo", "intake-1", "secretary-1", "system"): + assert excluded not in slugs + + +def test_a2a_allowed_pairs_group_key_counts() -> None: + counts = Counter(p.group_key for p in A2A_ALLOWED_PAIRS) + assert dict(counts) == _EXPECTED_GROUP_COUNTS + + +def test_a2a_allowed_pairs_contains_same_cell_pair() -> None: + assert any( + {p.agent_a, p.agent_b} == {"be-dev-1", "be-qa"} for p in A2A_ALLOWED_PAIRS + ) + + +def test_a2a_allowed_pairs_contains_pm_chain_pair() -> None: + assert any( + {p.agent_a, p.agent_b} == {"be-pm", "main-pm"} for p in A2A_ALLOWED_PAIRS + ) + + +def test_a2a_allowed_pairs_contains_board_pair() -> None: + assert any( + {p.agent_a, p.agent_b} == {"auditor", "product-owner"} + for p in A2A_ALLOWED_PAIRS + ) + + +def test_a2a_allowed_pairs_reflects_can_a2a_direct_matrix() -> None: + """Every listed pair allows >=1 direction per the live matrix — catches + drift if can_a2a_direct changes without regenerating the static list.""" + for p in A2A_ALLOWED_PAIRS: + allowed_ab, _ = can_a2a_direct(p.agent_a, p.agent_b) + allowed_ba, _ = can_a2a_direct(p.agent_b, p.agent_a) + assert allowed_ab or allowed_ba + + +def test_a2a_allowed_pairs_role_team_fields_match_registry() -> None: + for p in A2A_ALLOWED_PAIRS: + assert p.role_a == foundation.AGENTS[p.agent_a].role.value + assert p.team_a == foundation.AGENTS[p.agent_a].team.value + assert p.role_b == foundation.AGENTS[p.agent_b].role.value + assert p.team_b == foundation.AGENTS[p.agent_b].team.value