feat(kanban): persist board, reference pages, and run agents
Replace the localStorage kanban with the backend-backed board, add typed clients and a React hook with optimistic updates. Cards can reference static doc pages and user-created custom pages (new "custom" reference type with purple chips). Add the agentic orchestrator UI: a per-card panel to launch `pi` runs, watch a live tool/thought stream over SSE, steer mid-run, and stop — while the board stays fully interactive. The board page wires the orchestrator and custom-pages stores into every card.
This commit is contained in:
226
apps/docs/src/components/kanban/useOrchestrator.ts
Normal file
226
apps/docs/src/components/kanban/useOrchestrator.ts
Normal file
@@ -0,0 +1,226 @@
|
||||
import { useCallback, useEffect, useRef, useState } from 'react';
|
||||
import {
|
||||
orchestratorApi,
|
||||
type AgentRun,
|
||||
type RunEvent,
|
||||
} from '../../lib/orchestratorApi';
|
||||
|
||||
/**
|
||||
* Drives the agentic orchestrator from the board UI.
|
||||
*
|
||||
* Holds the latest run per card plus a live, replayable event log for any run
|
||||
* currently being "watched" (typically the expanded card's active run). Live
|
||||
* updates arrive over Server-Sent Events; the board stays fully interactive
|
||||
* while an agent works a card — start, steer, and stop at any time.
|
||||
*/
|
||||
|
||||
export interface UseOrchestrator {
|
||||
/** Newest run for a card (active if one is running, else the last settled). */
|
||||
runForCard: (cardId: string) => AgentRun | undefined;
|
||||
/** True while a run is actively working the given card. */
|
||||
isRunning: (cardId: string) => boolean;
|
||||
/** All known runs (newest first), for a global activity view. */
|
||||
runs: AgentRun[];
|
||||
/** Watched events keyed by run id (ordered). */
|
||||
eventsForRun: (runId: string) => RunEvent[];
|
||||
/** Load initial state. */
|
||||
reload: () => Promise<void>;
|
||||
loading: boolean;
|
||||
error: string | null;
|
||||
/** Begin a run for a card. Returns the created run. */
|
||||
start: (input: { cardId: string; prompt?: string }) => Promise<AgentRun>;
|
||||
/** Send a steer/follow-up message to an active run. */
|
||||
message: (runId: string, text: string, mode: 'steer' | 'followUp') => Promise<void>;
|
||||
/** Stop an active run. */
|
||||
stop: (runId: string) => Promise<void>;
|
||||
/** Open the live event stream for a run (ref-counted; safe to call repeatedly). */
|
||||
watch: (runId: string) => void;
|
||||
/** Release a watch (ref-counted; closes the stream when the last watcher leaves). */
|
||||
unwatch: (runId: string) => void;
|
||||
/** Remove a settled run from the UI (and reclaim its worktree). */
|
||||
remove: (runId: string) => Promise<void>;
|
||||
}
|
||||
|
||||
export function useOrchestrator(): UseOrchestrator {
|
||||
const [runs, setRuns] = useState<AgentRun[]>([]);
|
||||
const [eventsByRun, setEventsByRun] = useState<Record<string, RunEvent[]>>({});
|
||||
const [loading, setLoading] = useState(true);
|
||||
const [error, setError] = useState<string | null>(null);
|
||||
|
||||
// Refcounted EventSource subscriptions + replay cursors, kept in refs so the
|
||||
// SSE callbacks always see fresh state without re-subscribing.
|
||||
const sources = useRef(new Map<string, EventSource>());
|
||||
const refcounts = useRef(new Map<string, number>());
|
||||
const cursors = useRef(new Map<string, number>());
|
||||
|
||||
const reload = useCallback(async () => {
|
||||
setError(null);
|
||||
try {
|
||||
const { runs: list } = await orchestratorApi.listRuns();
|
||||
setRuns(list);
|
||||
} catch (e) {
|
||||
setError(e instanceof Error ? e.message : 'Failed to load runs');
|
||||
} finally {
|
||||
setLoading(false);
|
||||
}
|
||||
}, []);
|
||||
|
||||
useEffect(() => {
|
||||
void reload();
|
||||
}, [reload]);
|
||||
|
||||
const upsertRun = useCallback((run: AgentRun) => {
|
||||
setRuns((prev) => {
|
||||
const next = prev.filter((r) => r.id !== run.id);
|
||||
next.unshift(run);
|
||||
return next;
|
||||
});
|
||||
}, []);
|
||||
|
||||
const appendEvent = useCallback((runId: string, ev: RunEvent) => {
|
||||
setEventsByRun((prev) => {
|
||||
const cur = prev[runId] ?? [];
|
||||
// Dedup by seq when present (history flush vs. live may overlap).
|
||||
if (typeof ev.seq === 'number' && cur.some((e) => e.seq === ev.seq)) return prev;
|
||||
return { ...prev, [runId]: [...cur, ev] };
|
||||
});
|
||||
}, []);
|
||||
|
||||
const start = useCallback(
|
||||
async (input: { cardId: string; prompt?: string }) => {
|
||||
const { run } = await orchestratorApi.startRun(input);
|
||||
upsertRun(run);
|
||||
return run;
|
||||
},
|
||||
[upsertRun],
|
||||
);
|
||||
|
||||
const message = useCallback(
|
||||
async (runId: string, text: string, mode: 'steer' | 'followUp') => {
|
||||
await orchestratorApi.messageRun(runId, text, mode);
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
const stop = useCallback(
|
||||
async (runId: string) => {
|
||||
await orchestratorApi.stopRun(runId);
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
const remove = useCallback(async (runId: string) => {
|
||||
await orchestratorApi.deleteRun(runId);
|
||||
setRuns((prev) => prev.filter((r) => r.id !== runId));
|
||||
setEventsByRun((prev) => {
|
||||
const next = { ...prev };
|
||||
delete next[runId];
|
||||
return next;
|
||||
});
|
||||
}, []);
|
||||
|
||||
/** (Re)open the SSE stream for a run, replaying persisted history first. */
|
||||
const openStream = useCallback(
|
||||
(runId: string) => {
|
||||
if (sources.current.has(runId)) return;
|
||||
const since = cursors.current.get(runId) ?? 0;
|
||||
const es = new EventSource(`/api/orchestrator/runs/${runId}/stream?since=${since}`);
|
||||
sources.current.set(runId, es);
|
||||
es.addEventListener('event', (msg) => {
|
||||
const ev = JSON.parse((msg as MessageEvent).data) as RunEvent;
|
||||
if (typeof ev.seq === 'number') cursors.current.set(runId, ev.seq);
|
||||
appendEvent(runId, ev);
|
||||
// Reflect status changes onto the run record.
|
||||
if (ev.type === 'status') {
|
||||
const status = ev.data.status as AgentRun['status'];
|
||||
setRuns((prev) =>
|
||||
prev.map((r) =>
|
||||
r.id === runId
|
||||
? {
|
||||
...r,
|
||||
status,
|
||||
finishedAt: ['completed', 'failed', 'stopped'].includes(status)
|
||||
? new Date().toISOString()
|
||||
: r.finishedAt,
|
||||
}
|
||||
: r,
|
||||
),
|
||||
);
|
||||
}
|
||||
if (ev.type === 'done' && typeof ev.data.summary === 'string') {
|
||||
setRuns((prev) =>
|
||||
prev.map((r) => (r.id === runId ? { ...r, summary: ev.data.summary as string } : r)),
|
||||
);
|
||||
}
|
||||
});
|
||||
es.onerror = () => {
|
||||
// EventSource auto-reconnects; nothing to do here.
|
||||
};
|
||||
},
|
||||
[appendEvent],
|
||||
);
|
||||
|
||||
const watch = useCallback(
|
||||
(runId: string) => {
|
||||
const n = (refcounts.current.get(runId) ?? 0) + 1;
|
||||
refcounts.current.set(runId, n);
|
||||
if (n === 1) openStream(runId);
|
||||
},
|
||||
[openStream],
|
||||
);
|
||||
|
||||
const unwatch = useCallback((runId: string) => {
|
||||
const n = (refcounts.current.get(runId) ?? 0) - 1;
|
||||
if (n > 0) {
|
||||
refcounts.current.set(runId, n);
|
||||
return;
|
||||
}
|
||||
refcounts.current.delete(runId);
|
||||
const es = sources.current.get(runId);
|
||||
if (es) {
|
||||
es.close();
|
||||
sources.current.delete(runId);
|
||||
}
|
||||
}, []);
|
||||
|
||||
// Close all streams on unmount.
|
||||
useEffect(() => {
|
||||
return () => {
|
||||
for (const es of sources.current.values()) es.close();
|
||||
sources.current.clear();
|
||||
refcounts.current.clear();
|
||||
};
|
||||
}, []);
|
||||
|
||||
const runForCard = useCallback(
|
||||
(cardId: string) => {
|
||||
const forCard = runs.filter((r) => r.cardId === cardId);
|
||||
const active = forCard.find((r) => r.status === 'running');
|
||||
return active ?? forCard[0];
|
||||
},
|
||||
[runs],
|
||||
);
|
||||
|
||||
const isRunning = useCallback(
|
||||
(cardId: string) => Boolean(runs.find((r) => r.cardId === cardId && r.status === 'running')),
|
||||
[runs],
|
||||
);
|
||||
|
||||
const eventsForRun = useCallback((runId: string) => eventsByRun[runId] ?? [], [eventsByRun]);
|
||||
|
||||
return {
|
||||
runForCard,
|
||||
isRunning,
|
||||
runs,
|
||||
eventsForRun,
|
||||
reload,
|
||||
loading,
|
||||
error,
|
||||
start,
|
||||
message,
|
||||
stop,
|
||||
watch,
|
||||
unwatch,
|
||||
remove,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user