diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index ee8483dc8..426b564a2 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -16,6 +16,11 @@ import { createEmptyTranscriptState, processTranscriptEvent, } from "./ui/agentSessionTranscript"; +import { + type InstanceLeadership, + LEADERSHIP_EVENT_KIND, + buildLeadership, +} from "./ui/leadershipHelpers"; const MAX_OBSERVER_EVENTS = 800; @@ -32,11 +37,13 @@ const IDLE_SNAPSHOT: ObserverSnapshot = { }; const EMPTY_TRANSCRIPT: TranscriptItem[] = []; +const EMPTY_LEADERSHIP: InstanceLeadership[] = []; const listeners = new Set<() => void>(); const eventsByAgent = new Map(); const transcriptByAgent = new Map(); const snapshotByAgent = new Map(); +const leadershipByAgent = new Map(); // Normalized pubkeys of agents we are actively managing. Only events whose // "agent" tag matches an entry here will be decrypted (defense-in-depth). @@ -109,6 +116,14 @@ function appendAgentEvent(agentPubkey: string, event: ObserverEvent) { transcriptByAgent.set(key, buildTranscriptState(final)); } + // Rebuild the cached leadership array only when a leadership frame lands, so + // `getAgentLeadership` stays a stable map lookup (referential stability is + // required by `useSyncExternalStore`). The rebuild walks the trimmed window, + // so instances whose latest frame aged out are pruned automatically. + if (event.kind === LEADERSHIP_EVENT_KIND) { + leadershipByAgent.set(key, buildLeadership(final)); + } + // Invalidate cached snapshot for this agent invalidateSnapshot(key); @@ -272,6 +287,20 @@ export function getAgentTranscript( return state?.items ?? EMPTY_TRANSCRIPT; } +export type { InstanceLeadership }; + +export function getAgentLeadership( + agentPubkey?: string | null, + enabled?: boolean, +): InstanceLeadership[] { + if (!enabled || !agentPubkey) { + return EMPTY_LEADERSHIP; + } + return ( + leadershipByAgent.get(normalizePubkey(agentPubkey)) ?? EMPTY_LEADERSHIP + ); +} + export function useManagedAgentObserverBridge( agents: readonly Pick[], ) { @@ -308,6 +337,7 @@ export function resetAgentObserverStore() { eventsByAgent.clear(); transcriptByAgent.clear(); snapshotByAgent.clear(); + leadershipByAgent.clear(); knownAgentPubkeys.clear(); connectionState = "idle"; errorMessage = null; diff --git a/desktop/src/features/agents/ui/ManagedAgentRow.tsx b/desktop/src/features/agents/ui/ManagedAgentRow.tsx index 33ca2e5bb..70207aab7 100644 --- a/desktop/src/features/agents/ui/ManagedAgentRow.tsx +++ b/desktop/src/features/agents/ui/ManagedAgentRow.tsx @@ -4,6 +4,7 @@ import { ChevronDown, ChevronRight, Clipboard, + Crown, Ellipsis, FileText, Pencil, @@ -33,13 +34,23 @@ import { DropdownMenuContent, DropdownMenuItem, DropdownMenuSeparator, + DropdownMenuSub, + DropdownMenuSubContent, + DropdownMenuSubTrigger, DropdownMenuTrigger, } from "@/shared/ui/dropdown-menu"; import { EditAgentDialog } from "./EditAgentDialog"; import { friendlyAgentLastError } from "@/features/agents/lib/friendlyAgentLastError"; import { ManagedAgentLogPanel } from "./ManagedAgentLogPanel"; import { ModelPicker } from "./ModelPicker"; -import { truncatePubkey } from "./agentUi"; +import { truncateInstanceId, truncatePubkey } from "./agentUi"; +import { useAgentLeadership } from "./useObserverEvents"; +import { + type InstanceLeadership, + filterStaleInstances, + selectFreshestLeader, +} from "./leadershipHelpers"; +import { claimManagedAgentLeadership } from "@/shared/api/agentControl"; export function ManagedAgentRow({ agent, @@ -115,6 +126,21 @@ export function ManagedAgentRow({ // crash. Generic exits stay verbatim so we don't lie about other failures. const friendlyError = friendlyAgentLastError(agent.lastError); + // Leadership frames flow into the owner-wide observer store regardless of + // session-panel state, so this is enabled on row visibility (gated only on a + // pubkey). The 5s clock drives stale eviction without a new frame arriving — + // a crashed leader's last frame ages out and the badge drops within 15s. + const leadership = useAgentLeadership(true, agent.pubkey); + const leadershipNow = useNow(5000); + const liveInstances = React.useMemo( + () => filterStaleInstances(leadership, leadershipNow), + [leadership, leadershipNow], + ); + const leaderInstanceId = React.useMemo( + () => selectFreshestLeader(liveInstances)?.instanceId ?? null, + [liveInstances], + ); + return (
onSelectLogAgent(pubkey)} @@ -335,6 +365,7 @@ function WorkingBadge({ function StatusBlock({ friendlyError, isWorking, + leaderInstanceId, presenceLoaded, presenceStatus, processDetail, @@ -342,6 +373,7 @@ function StatusBlock({ }: { friendlyError: ReturnType; isWorking: boolean; + leaderInstanceId: string | null; presenceLoaded: boolean; presenceStatus: PresenceStatus | undefined; processDetail: string; @@ -352,12 +384,20 @@ function StatusBlock({

Status

- +
+ + {leaderInstanceId ? ( + + + Leader + + ) : null} +

{processDetail}

{friendlyError ? (

void; onDelete: (pubkey: string) => void; onOpenLogs: (pubkey: string) => void; @@ -423,6 +467,8 @@ function AgentActionsMenu({ onToggleStartOnAppLaunch: (pubkey: string, startOnAppLaunch: boolean) => void; }) { const [editOpen, setEditOpen] = React.useState(false); + // Nothing to steal unless at least two instances are racing. + const showLeadershipSubmenu = instances.length > 1; return ( <> @@ -522,6 +568,57 @@ function AgentActionsMenu({ ) : null} + {showLeadershipSubmenu ? ( + + + + Leadership + + + {instances.map((instance) => { + const isLeader = instance.instanceId === leaderInstanceId; + return ( + { + try { + await claimManagedAgentLeadership( + agent.pubkey, + instance.instanceId, + ); + toast.success( + `Leadership request sent to ${agent.name}.`, + ); + } catch (error) { + toast.error( + error instanceof Error + ? error.message + : `Failed to send leadership request to ${agent.name}.`, + ); + } + }} + > + {isLeader ? ( + + ) : ( + + )} + + {truncateInstanceId(instance.instanceId)} + + + {isLeader + ? "Leader" + : `${formatElapsed(Date.now() - instance.lastSeen)} ago`} + + + ); + })} + + + ) : null} + new Date(epochMs).toISOString(); + +function leadershipEvent({ seq, instanceId, isLeader, at, kind, payload }) { + return { + seq, + timestamp: iso(at), + kind: kind ?? "leadership_status", + agentIndex: null, + channelId: null, + sessionId: null, + turnId: null, + payload: payload ?? { type: "leadership_status", instanceId, isLeader }, + }; +} + +// --- parseLeadershipPayload --- + +test("parseLeadershipPayload accepts a well-formed payload", () => { + const result = parseLeadershipPayload({ + type: "leadership_status", + instanceId: "123-456", + isLeader: true, + }); + assert.deepEqual(result, { instanceId: "123-456", isLeader: true }); +}); + +test("parseLeadershipPayload rejects non-object payloads", () => { + for (const bad of [null, undefined, "string", 42, true, []]) { + // Arrays are objects but lack the required string/boolean fields, so they + // must also be rejected. + assert.equal(parseLeadershipPayload(bad), null); + } +}); + +test("parseLeadershipPayload rejects a missing or non-string instanceId", () => { + assert.equal(parseLeadershipPayload({ isLeader: true }), null); + assert.equal(parseLeadershipPayload({ instanceId: 5, isLeader: true }), null); +}); + +test("parseLeadershipPayload rejects a non-boolean isLeader", () => { + assert.equal( + parseLeadershipPayload({ instanceId: "a", isLeader: "yes" }), + null, + ); + assert.equal(parseLeadershipPayload({ instanceId: "a" }), null); +}); + +// --- buildLeadership --- + +test("buildLeadership keeps the latest frame per instanceId", () => { + const events = [ + leadershipEvent({ seq: 1, instanceId: "A", isLeader: true, at: 1000 }), + leadershipEvent({ seq: 2, instanceId: "B", isLeader: false, at: 1500 }), + leadershipEvent({ seq: 3, instanceId: "A", isLeader: false, at: 2000 }), + ]; + const result = buildLeadership(events); + assert.equal(result.length, 2); + const a = result.find((i) => i.instanceId === "A"); + assert.deepEqual(a, { instanceId: "A", isLeader: false, lastSeen: 2000 }); +}); + +test("buildLeadership ignores non-leadership events", () => { + const events = [ + leadershipEvent({ seq: 1, kind: "turn_started", payload: {}, at: 500 }), + leadershipEvent({ seq: 2, instanceId: "A", isLeader: true, at: 1000 }), + ]; + const result = buildLeadership(events); + assert.deepEqual(result, [ + { instanceId: "A", isLeader: true, lastSeen: 1000 }, + ]); +}); + +test("buildLeadership drops frames that fail the payload guard", () => { + const events = [ + leadershipEvent({ + seq: 1, + payload: { instanceId: 5, isLeader: true }, + at: 1000, + }), + leadershipEvent({ seq: 2, instanceId: "A", isLeader: true, at: 1500 }), + ]; + const result = buildLeadership(events); + assert.deepEqual(result, [ + { instanceId: "A", isLeader: true, lastSeen: 1500 }, + ]); +}); + +test("buildLeadership drops frames with an unparseable timestamp", () => { + const bad = leadershipEvent({ + seq: 1, + instanceId: "A", + isLeader: true, + at: 1000, + }); + bad.timestamp = "not-a-date"; + const good = leadershipEvent({ + seq: 2, + instanceId: "B", + isLeader: false, + at: 1500, + }); + const result = buildLeadership([bad, good]); + assert.deepEqual(result, [ + { instanceId: "B", isLeader: false, lastSeen: 1500 }, + ]); +}); + +test("buildLeadership returns an empty array for no leadership frames", () => { + assert.deepEqual(buildLeadership([]), []); +}); + +test("buildLeadership prunes a zombie instance whose frame aged out of the window", () => { + // Simulates the trimmed event window: the dead instance's frame is gone, so + // only the survivor's frame remains in the input. The reduction therefore + // never re-surfaces the zombie instanceId. + const events = [ + leadershipEvent({ + seq: 9, + instanceId: "survivor", + isLeader: true, + at: 5000, + }), + ]; + const result = buildLeadership(events); + assert.deepEqual( + result.map((i) => i.instanceId), + ["survivor"], + ); +}); + +// --- filterStaleInstances --- + +test("filterStaleInstances drops instances past the stale threshold", () => { + const now = 100_000; + const fresh = { instanceId: "fresh", isLeader: true, lastSeen: now - 1000 }; + const stale = { + instanceId: "stale", + isLeader: false, + lastSeen: now - LEADERSHIP_STALE_MS - 1, + }; + const result = filterStaleInstances([fresh, stale], now); + assert.deepEqual(result, [fresh]); +}); + +test("filterStaleInstances keeps an instance exactly at the threshold", () => { + const now = 100_000; + const boundary = { + instanceId: "boundary", + isLeader: true, + lastSeen: now - LEADERSHIP_STALE_MS, + }; + assert.deepEqual(filterStaleInstances([boundary], now), [boundary]); +}); + +test("filterStaleInstances treats a NaN lastSeen as stale", () => { + const now = 100_000; + const nan = { instanceId: "nan", isLeader: true, lastSeen: Number.NaN }; + // now - NaN === NaN, and `NaN <= threshold` is false, so it is excluded. + assert.deepEqual(filterStaleInstances([nan], now), []); +}); + +// --- selectFreshestLeader --- + +test("selectFreshestLeader returns null when no instance leads", () => { + const instances = [ + { instanceId: "A", isLeader: false, lastSeen: 1000 }, + { instanceId: "B", isLeader: false, lastSeen: 2000 }, + ]; + assert.equal(selectFreshestLeader(instances), null); +}); + +test("selectFreshestLeader picks the freshest among multiple leaders", () => { + // The transient two-leader window after a crash: the dead leader's stale + // isLeader:true and the survivor's fresh one coexist. Freshest wins. + const dead = { instanceId: "dead", isLeader: true, lastSeen: 1000 }; + const survivor = { instanceId: "survivor", isLeader: true, lastSeen: 9000 }; + assert.equal(selectFreshestLeader([dead, survivor]), survivor); +}); + +test("selectFreshestLeader ignores non-leaders even if fresher", () => { + const leader = { instanceId: "leader", isLeader: true, lastSeen: 1000 }; + const followerFresher = { + instanceId: "follower", + isLeader: false, + lastSeen: 9000, + }; + assert.equal(selectFreshestLeader([leader, followerFresher]), leader); +}); + +test("selectFreshestLeader returns null for an empty list", () => { + assert.equal(selectFreshestLeader([]), null); +}); diff --git a/desktop/src/features/agents/ui/leadershipHelpers.ts b/desktop/src/features/agents/ui/leadershipHelpers.ts new file mode 100644 index 000000000..9055d504a --- /dev/null +++ b/desktop/src/features/agents/ui/leadershipHelpers.ts @@ -0,0 +1,102 @@ +import type { ObserverEvent } from "./agentSessionTypes"; + +/** Per-window-instance leadership state derived from `leadership_status` frames. */ +export type InstanceLeadership = { + instanceId: string; + isLeader: boolean; + lastSeen: number; // epoch ms — Date.parse(event.timestamp) +}; + +export const LEADERSHIP_EVENT_KIND = "leadership_status"; + +/** + * An instance is stale once it has missed 3 consecutive 5s emit ticks. A + * surviving instance re-emits within 5s, so 15s tolerates a single dropped + * relay frame without the badge flickering. + */ +export const LEADERSHIP_STALE_MS = 15_000; + +/** + * Narrows the untrusted `unknown` payload of a `leadership_status` frame. + * Harness emits arbitrary JSON (`observer.rs`), so the contents are validated + * here at the boundary; malformed frames are dropped rather than producing + * `undefined`/`NaN` entries. + */ +export function parseLeadershipPayload( + payload: unknown, +): { instanceId: string; isLeader: boolean } | null { + if (typeof payload !== "object" || payload === null) { + return null; + } + const record = payload as Record; + if ( + typeof record.instanceId !== "string" || + typeof record.isLeader !== "boolean" + ) { + return null; + } + return { instanceId: record.instanceId, isLeader: record.isLeader }; +} + +/** + * Reduces an agent's observer events to the latest `leadership_status` frame + * per `instanceId`. `events` must be sorted ascending (the store keeps + * `eventsByAgent` sorted by `compareObserverEvents`), so a simple + * last-write-wins walk in iteration order is correct — no comparator needed. + * + * Instances whose latest frame fell out of the trimmed event window are + * naturally absent, so this also prunes zombie instanceIds. Frames that fail + * the payload guard or carry an unparseable timestamp are dropped. + */ +export function buildLeadership( + events: readonly ObserverEvent[], +): InstanceLeadership[] { + const latestByInstance = new Map(); + for (const event of events) { + if (event.kind !== LEADERSHIP_EVENT_KIND) { + continue; + } + const parsed = parseLeadershipPayload(event.payload); + if (!parsed) { + continue; + } + const lastSeen = Date.parse(event.timestamp); + if (Number.isNaN(lastSeen)) { + continue; + } + latestByInstance.set(parsed.instanceId, { ...parsed, lastSeen }); + } + return [...latestByInstance.values()]; +} + +/** Drops instances whose last frame is older than the stale threshold. */ +export function filterStaleInstances( + instances: readonly InstanceLeadership[], + now: number, +): InstanceLeadership[] { + return instances.filter( + (instance) => now - instance.lastSeen <= LEADERSHIP_STALE_MS, + ); +} + +/** + * The instance to surface as leader: the freshest (`max(lastSeen)`) among + * those reporting `isLeader`. After a leader window crashes, the survivor's + * `isLeader: true` and the dead window's stale `isLeader: true` coexist for up + * to one stale window; picking the freshest converges to the survivor without + * a "contested" UI state. Returns null when no instance currently leads. + */ +export function selectFreshestLeader( + instances: readonly InstanceLeadership[], +): InstanceLeadership | null { + let leader: InstanceLeadership | null = null; + for (const instance of instances) { + if (!instance.isLeader) { + continue; + } + if (!leader || instance.lastSeen > leader.lastSeen) { + leader = instance; + } + } + return leader; +} diff --git a/desktop/src/features/agents/ui/useObserverEvents.ts b/desktop/src/features/agents/ui/useObserverEvents.ts index 6631b1677..9771c9a57 100644 --- a/desktop/src/features/agents/ui/useObserverEvents.ts +++ b/desktop/src/features/agents/ui/useObserverEvents.ts @@ -2,10 +2,12 @@ import * as React from "react"; import { ensureRelayObserverSubscription, + getAgentLeadership, getAgentObserverSnapshot, getAgentTranscript, subscribeAgentObserverStore, } from "@/features/agents/observerRelayStore"; +import type { InstanceLeadership } from "@/features/agents/observerRelayStore"; import type { TranscriptItem } from "./agentSessionTypes"; // Stable subscribe reference shared by all useSyncExternalStore hooks. @@ -45,3 +47,15 @@ export function useAgentTranscript( return React.useSyncExternalStore(subscribeToStore, getSnapshot); } + +export function useAgentLeadership( + enabled: boolean, + agentPubkey?: string | null, +): InstanceLeadership[] { + const getSnapshot = React.useCallback( + () => getAgentLeadership(agentPubkey, enabled), + [agentPubkey, enabled], + ); + + return React.useSyncExternalStore(subscribeToStore, getSnapshot); +} diff --git a/desktop/src/shared/api/agentControl.ts b/desktop/src/shared/api/agentControl.ts index 923d93922..46a828102 100644 --- a/desktop/src/shared/api/agentControl.ts +++ b/desktop/src/shared/api/agentControl.ts @@ -11,3 +11,18 @@ export async function cancelManagedAgentTurn( }); return { status: "sent" }; } + +// Best-effort cooperative-steal request. The harness gates its `control_result` +// ack on a successful lock acquire, so this ack only means "frame sent" — the +// `leadership_status` stream remains the source of truth for who actually +// leads. The UI must not optimistically flip on this return value. +export async function claimManagedAgentLeadership( + pubkey: string, + targetInstanceId: string, +): Promise<{ status: "sent" }> { + await sendAgentObserverControl(pubkey, { + type: "claim_leadership", + targetInstanceId, + }); + return { status: "sent" }; +}