diff --git a/AGENTS.md b/AGENTS.md index 4474bef4..ae356ce8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -233,6 +233,12 @@ During normal plan review, an Archive sidebar tab provides the same browsing via | `/api/external-annotations` | POST | Add external annotations (single or batch `{ annotations: [...] }`) | | `/api/external-annotations` | PATCH | Update fields on a single annotation (`?id=`) | | `/api/external-annotations` | DELETE | Remove by `?id=`, `?source=`, or clear all | +| `/api/agents/capabilities` | GET | Check available agent providers (claude, codex, shell) | +| `/api/agents/jobs/stream` | GET | SSE stream for real-time agent job status updates | +| `/api/agents/jobs` | GET | Snapshot of agent jobs (polling fallback, `?since=N` for version gating) | +| `/api/agents/jobs` | POST | Launch an agent job (body: `{ provider, command, label }`) | +| `/api/agents/jobs` | DELETE | Kill all running agent jobs | +| `/api/agents/jobs/:id` | DELETE | Kill a specific agent job | ### Annotate Server (`packages/server/annotate.ts`) diff --git a/apps/pi-extension/server/agent-jobs.ts b/apps/pi-extension/server/agent-jobs.ts new file mode 100644 index 00000000..4bd03440 --- /dev/null +++ b/apps/pi-extension/server/agent-jobs.ts @@ -0,0 +1,331 @@ +/** + * Agent Jobs — Pi (node:http) server handler. + * + * Manages background agent processes (spawn, monitor, kill) and exposes + * HTTP routes + SSE broadcasting for job status updates. + * + * Mirrors packages/server/agent-jobs.ts but uses node:http primitives. + */ + +import type { IncomingMessage, ServerResponse } from "node:http"; +import { spawn, execFileSync, type ChildProcess } from "node:child_process"; +import { + type AgentJobInfo, + type AgentJobEvent, + type AgentCapability, + type AgentCapabilities, + isTerminalStatus, + jobSource, + serializeAgentSSEEvent, + AGENT_HEARTBEAT_COMMENT, + AGENT_HEARTBEAT_INTERVAL_MS, +} from "../generated/agent-jobs.js"; +import { json, parseBody } from "./helpers.js"; + +// --------------------------------------------------------------------------- +// Route prefixes +// --------------------------------------------------------------------------- + +const BASE = "/api/agents"; +const JOBS = `${BASE}/jobs`; +const JOBS_STREAM = `${JOBS}/stream`; +const CAPABILITIES = `${BASE}/capabilities`; + +// --------------------------------------------------------------------------- +// which() helper for Node.js +// --------------------------------------------------------------------------- + +function whichCmd(cmd: string): boolean { + try { + const bin = process.platform === "win32" ? "where" : "which"; + execFileSync(bin, [cmd], { encoding: "utf-8", stdio: ["pipe", "pipe", "pipe"] }); + return true; + } catch { + return false; + } +} + +// --------------------------------------------------------------------------- +// Factory +// --------------------------------------------------------------------------- + +export interface AgentJobHandlerOptions { + mode: "plan" | "review" | "annotate"; + getServerUrl: () => string; + getCwd: () => string; +} + +export function createAgentJobHandler(options: AgentJobHandlerOptions) { + const { mode, getServerUrl, getCwd } = options; + + // --- State --- + const jobs = new Map(); + const subscribers = new Set(); + let version = 0; + + // --- Capability detection (run once) --- + const capabilities: AgentCapability[] = [ + { id: "claude", name: "Claude Code", available: whichCmd("claude") }, + { id: "codex", name: "Codex CLI", available: whichCmd("codex") }, + { id: "shell", name: "Shell Command", available: true }, + ]; + const capabilitiesResponse: AgentCapabilities = { + mode, + providers: capabilities, + available: capabilities.some((c) => c.available), + }; + + // --- SSE broadcasting --- + function broadcast(event: AgentJobEvent): void { + version++; + const data = serializeAgentSSEEvent(event); + for (const res of subscribers) { + try { + res.write(data); + } catch { + subscribers.delete(res); + } + } + } + + // --- Process lifecycle --- + function spawnJob( + provider: string, + command: string[], + label: string, + ): AgentJobInfo { + const id = crypto.randomUUID(); + const source = jobSource(id); + + const info: AgentJobInfo = { + id, + source, + provider, + label, + status: "starting", + startedAt: Date.now(), + command, + }; + + let proc: ChildProcess | null = null; + + try { + proc = spawn(command[0], command.slice(1), { + cwd: getCwd(), + stdio: ["ignore", "ignore", "pipe"], + env: { + ...process.env, + PLANNOTATOR_AGENT_SOURCE: source, + PLANNOTATOR_API_URL: getServerUrl(), + }, + }); + + info.status = "running"; + jobs.set(id, { info, proc }); + broadcast({ type: "job:started", job: { ...info } }); + + // Accumulate stderr continuously (must attach before exit fires) + let stderrBuf = ""; + if (proc.stderr) { + proc.stderr.on("data", (chunk: Buffer) => { + stderrBuf = (stderrBuf + chunk.toString()).slice(-500); + }); + } + + // Monitor process exit + proc.on("exit", (exitCode) => { + const entry = jobs.get(id); + if (!entry || isTerminalStatus(entry.info.status)) return; + + entry.info.endedAt = Date.now(); + entry.info.exitCode = exitCode ?? undefined; + entry.info.status = exitCode === 0 ? "done" : "failed"; + + if (exitCode !== 0 && stderrBuf) { + entry.info.error = stderrBuf; + } + + broadcast({ type: "job:completed", job: { ...entry.info } }); + }); + + // Handle spawn errors after process starts + proc.on("error", (err) => { + const entry = jobs.get(id); + if (!entry || isTerminalStatus(entry.info.status)) return; + + entry.info.status = "failed"; + entry.info.endedAt = Date.now(); + entry.info.error = err.message; + broadcast({ type: "job:completed", job: { ...entry.info } }); + }); + } catch (err) { + jobs.set(id, { info, proc: null }); + broadcast({ type: "job:started", job: { ...info } }); + + info.status = "failed"; + info.endedAt = Date.now(); + info.error = err instanceof Error ? err.message : String(err); + broadcast({ type: "job:completed", job: { ...info } }); + } + + return { ...info }; + } + + function killJob(id: string): boolean { + const entry = jobs.get(id); + if (!entry || isTerminalStatus(entry.info.status)) return false; + + if (entry.proc) { + try { + entry.proc.kill(); + } catch { + // Process may have already exited + } + } + + entry.info.status = "killed"; + entry.info.endedAt = Date.now(); + broadcast({ type: "job:completed", job: { ...entry.info } }); + return true; + } + + function killAll(): number { + let count = 0; + for (const [id, entry] of jobs) { + if (!isTerminalStatus(entry.info.status)) { + killJob(id); + count++; + } + } + return count; + } + + function getAllJobs(): AgentJobInfo[] { + return Array.from(jobs.values()).map((e) => ({ ...e.info })); + } + + // --- HTTP handler --- + return { + killAll, + + async handle( + req: IncomingMessage, + res: ServerResponse, + url: URL, + ): Promise { + // --- GET /api/agents/capabilities --- + if (url.pathname === CAPABILITIES && req.method === "GET") { + json(res, capabilitiesResponse); + return true; + } + + // --- SSE stream --- + if (url.pathname === JOBS_STREAM && req.method === "GET") { + res.writeHead(200, { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }); + + res.setTimeout(0); + + // Send current state as snapshot + const snapshot: AgentJobEvent = { + type: "snapshot", + jobs: getAllJobs(), + }; + res.write(serializeAgentSSEEvent(snapshot)); + + subscribers.add(res); + + // Heartbeat to keep connection alive + const heartbeatTimer = setInterval(() => { + try { + res.write(AGENT_HEARTBEAT_COMMENT); + } catch { + clearInterval(heartbeatTimer); + subscribers.delete(res); + } + }, AGENT_HEARTBEAT_INTERVAL_MS); + + // Clean up on disconnect + res.on("close", () => { + clearInterval(heartbeatTimer); + subscribers.delete(res); + }); + + return true; + } + + // --- GET /api/agents/jobs (snapshot / polling fallback) --- + if (url.pathname === JOBS && req.method === "GET") { + const since = url.searchParams.get("since"); + if (since !== null) { + const sinceVersion = parseInt(since, 10); + if (!isNaN(sinceVersion) && sinceVersion === version) { + res.writeHead(304); + res.end(); + return true; + } + } + json(res, { jobs: getAllJobs(), version }); + return true; + } + + // --- POST /api/agents/jobs (launch) --- + if (url.pathname === JOBS && req.method === "POST") { + try { + const body = await parseBody(req); + const provider = typeof body.provider === "string" ? body.provider : "shell"; + const rawCommand = Array.isArray(body.command) ? body.command : []; + const command = rawCommand.filter((c: unknown): c is string => typeof c === "string"); + const label = typeof body.label === "string" ? body.label : `${provider} agent`; + + // Validate provider is a known, available capability + const cap = capabilities.find((c) => c.id === provider); + if (!cap || !cap.available) { + json(res, { error: `Unknown or unavailable provider: ${provider}` }, 400); + return true; + } + + if (command.length === 0) { + json(res, { error: 'Missing "command" array' }, 400); + return true; + } + + const job = spawnJob(provider, command, label); + json(res, { job }, 201); + } catch { + json(res, { error: "Invalid JSON" }, 400); + } + return true; + } + + // --- DELETE /api/agents/jobs/:id (kill one) --- + if (url.pathname.startsWith(JOBS + "/") && url.pathname !== JOBS_STREAM && req.method === "DELETE") { + const id = url.pathname.slice(JOBS.length + 1); + if (!id) { + json(res, { error: "Missing job ID" }, 400); + return true; + } + const found = killJob(id); + if (!found) { + json(res, { error: "Job not found or already terminal" }, 404); + return true; + } + json(res, { ok: true }); + return true; + } + + // --- DELETE /api/agents/jobs (kill all) --- + if (url.pathname === JOBS && req.method === "DELETE") { + const count = killAll(); + json(res, { ok: true, killed: count }); + return true; + } + + // Not handled + return false; + }, + }; +} diff --git a/apps/pi-extension/server/serverReview.ts b/apps/pi-extension/server/serverReview.ts index c63baa27..2fdb1e43 100644 --- a/apps/pi-extension/server/serverReview.ts +++ b/apps/pi-extension/server/serverReview.ts @@ -36,6 +36,7 @@ import { } from "../generated/review-core.js"; import { createEditorAnnotationHandler } from "./annotations.js"; +import { createAgentJobHandler } from "./agent-jobs.js"; import { createExternalAnnotationHandler } from "./external-annotations.js"; import { handleDraftRequest, @@ -145,10 +146,25 @@ export async function startReviewServer(options: { : getRepoInfo(); const editorAnnotations = createEditorAnnotationHandler(); const externalAnnotations = createExternalAnnotationHandler("review"); + let currentPatch = options.rawPatch; let currentGitRef = options.gitRef; let currentDiffType: DiffType = options.diffType || "uncommitted"; let currentError = options.error; + + // Agent jobs — background process manager (late-binds serverUrl via getter) + let serverUrl = ""; + const agentJobs = createAgentJobHandler({ + mode: "review", + getServerUrl: () => serverUrl, + getCwd: () => { + if (currentDiffType.startsWith("worktree:")) { + const parsed = parseWorktreeDiffType(currentDiffType); + if (parsed) return parsed.path; + } + return options.gitContext?.cwd ?? process.cwd(); + }, + }); const sharingEnabled = options.sharingEnabled ?? process.env.PLANNOTATOR_SHARE !== "disabled"; const shareBaseUrl = @@ -519,6 +535,8 @@ export async function startReviewServer(options: { return; } else if (await externalAnnotations.handle(req, res, url)) { return; + } else if (await agentJobs.handle(req, res, url)) { + return; } else if (aiEndpoints && url.pathname.startsWith("/api/ai/")) { const handler = aiEndpoints[url.pathname]; if (handler) { @@ -568,13 +586,18 @@ export async function startReviewServer(options: { }); const { port, portSource } = await listenOnPort(server); + serverUrl = `http://localhost:${port}`; + const exitHandler = () => agentJobs.killAll(); + process.once("exit", exitHandler); return { port, portSource, - url: `http://localhost:${port}`, + url: serverUrl, waitForDecision: () => decisionPromise, stop: () => { + process.removeListener("exit", exitHandler); + agentJobs.killAll(); aiSessionManager?.disposeAll(); aiRegistry?.disposeAll(); server.close(); diff --git a/apps/pi-extension/vendor.sh b/apps/pi-extension/vendor.sh index 9f2626c9..c383d592 100755 --- a/apps/pi-extension/vendor.sh +++ b/apps/pi-extension/vendor.sh @@ -6,7 +6,7 @@ cd "$(dirname "$0")" mkdir -p generated generated/ai/providers -for f in feedback-templates review-core storage draft project pr-provider pr-github pr-gitlab checklist integrations-common repo reference-common favicon resolve-file config external-annotation; do +for f in feedback-templates review-core storage draft project pr-provider pr-github pr-gitlab checklist integrations-common repo reference-common favicon resolve-file config external-annotation agent-jobs; do src="../../packages/shared/$f.ts" printf '// @generated — DO NOT EDIT. Source: packages/shared/%s.ts\n' "$f" | cat - "$src" > "generated/$f.ts" done diff --git a/packages/review-editor/App.tsx b/packages/review-editor/App.tsx index 05855875..063827c0 100644 --- a/packages/review-editor/App.tsx +++ b/packages/review-editor/App.tsx @@ -29,6 +29,7 @@ import { extractLinesFromPatch } from './utils/patchParser'; import { isTypingTarget, useReviewSearch } from './hooks/useReviewSearch'; import { useEditorAnnotations } from '@plannotator/ui/hooks/useEditorAnnotations'; import { useExternalAnnotations } from '@plannotator/ui/hooks/useExternalAnnotations'; +import { useAgentJobs } from '@plannotator/ui/hooks/useAgentJobs'; import { exportEditorAnnotations } from '@plannotator/ui/utils/parser'; import { ResizeHandle } from '@plannotator/ui/components/ResizeHandle'; import { DiffViewer } from './components/DiffViewer'; @@ -215,6 +216,7 @@ const ReviewApp: React.FC = () => { // The same !!origin proxy is used elsewhere in this file (draft hook, feedback guard, conditional UI) // so this should be addressed as a broader refactor. const { externalAnnotations, updateExternalAnnotation, deleteExternalAnnotation } = useExternalAnnotations({ enabled: !!origin }); + const agentJobs = useAgentJobs({ enabled: !!origin }); // Merge local + SSE annotations, deduping draft-restored externals against // live SSE versions. Prefer the SSE version when both exist (same source, @@ -1568,6 +1570,12 @@ const ReviewApp: React.FC = () => { aiConfig={aiConfig} onAIConfigChange={handleAIConfigChange} hasAISession={!!aiChat.sessionId} + agentJobs={agentJobs.jobs} + agentCapabilities={agentJobs.capabilities} + onAgentLaunch={agentJobs.launchJob} + onAgentKillJob={agentJobs.killJob} + onAgentKillAll={agentJobs.killAll} + externalAnnotations={externalAnnotations} /> diff --git a/packages/review-editor/components/ReviewAgentsIcon.tsx b/packages/review-editor/components/ReviewAgentsIcon.tsx new file mode 100644 index 00000000..04045222 --- /dev/null +++ b/packages/review-editor/components/ReviewAgentsIcon.tsx @@ -0,0 +1,18 @@ +import React from 'react'; + +interface ReviewAgentsIconProps { + className?: string; +} + +/** + * Magnifying glass with cog — icon for the Review Agents tab. + * Uses currentColor for stroke so it inherits from parent text color. + */ +export const ReviewAgentsIcon: React.FC = ({ className = 'w-3.5 h-3.5' }) => ( + + + + + + +); diff --git a/packages/review-editor/components/ReviewPanel.tsx b/packages/review-editor/components/ReviewPanel.tsx index 90746e47..107e401c 100644 --- a/packages/review-editor/components/ReviewPanel.tsx +++ b/packages/review-editor/components/ReviewPanel.tsx @@ -12,11 +12,14 @@ import { PRCommentsTab } from './PRCommentsTab'; import { PRChecksTab } from './PRChecksTab'; import { AITab } from './AITab'; import { SparklesIcon } from './SparklesIcon'; +import { ReviewAgentsIcon } from './ReviewAgentsIcon'; +import { AgentsTab } from '@plannotator/ui/components/AgentsTab'; import type { PRMetadata } from '@plannotator/shared/pr-provider'; import type { AIChatEntry } from '../hooks/useAIChat'; +import type { AgentJobInfo, AgentCapabilities } from '@plannotator/ui/types'; import type { DiffFile } from '../types'; -type ReviewPanelTab = 'annotations' | 'ai' | 'summary' | 'comments' | 'checks'; +type ReviewPanelTab = 'annotations' | 'ai' | 'agents' | 'summary' | 'comments' | 'checks'; interface ReviewPanelProps { isOpen: boolean; @@ -48,6 +51,13 @@ interface ReviewPanelProps { aiConfig?: { providerId: string | null; model: string | null; reasoningEffort?: string | null }; onAIConfigChange?: (config: { providerId?: string | null; model?: string | null; reasoningEffort?: string | null }) => void; hasAISession?: boolean; + // Agent props + agentJobs?: AgentJobInfo[]; + agentCapabilities?: AgentCapabilities | null; + onAgentLaunch?: (params: { provider?: string; command?: string[]; label?: string }) => void; + onAgentKillJob?: (id: string) => void; + onAgentKillAll?: () => void; + externalAnnotations?: Array<{ source?: string }>; } const SuggestionPreview: React.FC<{ code: string; originalCode?: string; language?: string }> = ({ code, originalCode, language }) => { @@ -147,11 +157,19 @@ export const ReviewPanel: React.FC = ({ aiConfig, onAIConfigChange, hasAISession, + agentJobs, + agentCapabilities, + onAgentLaunch, + onAgentKillJob, + onAgentKillAll, + externalAnnotations, }) => { const totalCount = annotations.length + (editorAnnotations?.length ?? 0); const [copied, setCopied] = useState(false); const [activeTab, setActiveTab] = useState('annotations'); const hasPRTabs = !!prMetadata; + const hasAgents = !!agentCapabilities?.available; + const runningAgentCount = (agentJobs ?? []).filter(j => j.status === 'running' || j.status === 'starting').length; // Allow parent to control the active tab (e.g., switch to AI tab on ask) useEffect(() => { @@ -164,7 +182,7 @@ export const ReviewPanel: React.FC = ({ const handleTabChange = (tab: ReviewPanelTab) => { setActiveTab(tab); onTabChange?.(tab); - if (tab !== 'annotations' && tab !== 'ai' && !prContext && !isPRContextLoading) { + if (tab !== 'annotations' && tab !== 'ai' && tab !== 'agents' && !prContext && !isPRContextLoading) { fetchContext(); } }; @@ -213,13 +231,18 @@ export const ReviewPanel: React.FC = ({

- {activeTab === 'annotations' ? 'Annotations' : activeTab === 'ai' ? 'AI' : activeTab === 'summary' ? 'Summary' : activeTab === 'comments' ? 'Comments' : 'Checks'} + {activeTab === 'annotations' ? 'Annotations' : activeTab === 'ai' ? 'AI' : activeTab === 'agents' ? 'Review Agents' : activeTab === 'summary' ? 'Summary' : activeTab === 'comments' ? 'Comments' : 'Checks'}

{activeTab === 'annotations' && totalCount > 0 && ( {totalCount} )} + {activeTab === 'agents' && (agentJobs?.length ?? 0) > 0 && ( + + {agentJobs!.length} + + )} {activeTab === 'ai' && aiMessages.length > 0 && ( {aiMessages.length} @@ -260,6 +283,24 @@ export const ReviewPanel: React.FC = ({ )} + {/* Agents tab (visible when agent capabilities available) */} + {hasAgents && ( + + )} + {/* PR tabs (only in PR mode) */} {hasPRTabs && PR_TABS.filter(t => t.id !== 'annotations').map((tab) => ( + )} +
+ + {/* Error details */} + {job.status === 'failed' && job.error && expanded && ( +
+ 
+            {job.error}
+
+
+ )} +
+ ); +} + +// --- Main component --- + +export const AgentsTab: React.FC = ({ + jobs, + capabilities, + onLaunch, + onKillJob, + onKillAll, + externalAnnotations, +}) => { + const [selectedProvider, setSelectedProvider] = useState(''); + const [expandedJobId, setExpandedJobId] = useState(null); + const initializedRef = useRef(false); + + // Set default provider once capabilities load + useEffect(() => { + if (capabilities && !initializedRef.current) { + const firstAvailable = capabilities.providers.find((p) => p.available); + if (firstAvailable) { + setSelectedProvider(firstAvailable.id); + initializedRef.current = true; + } + } + }, [capabilities]); + + const availableProviders = useMemo( + () => capabilities?.providers.filter((p) => p.available) ?? [], + [capabilities], + ); + + // Annotation counts per job source + const annotationCounts = useMemo(() => { + const counts = new Map(); + for (const ann of externalAnnotations) { + if (ann.source) { + counts.set(ann.source, (counts.get(ann.source) ?? 0) + 1); + } + } + return counts; + }, [externalAnnotations]); + + // Sort: running first, then by startedAt descending + const sortedJobs = useMemo(() => { + return [...jobs].sort((a, b) => { + const aRunning = !isTerminalStatus(a.status); + const bRunning = !isTerminalStatus(b.status); + if (aRunning !== bRunning) return aRunning ? -1 : 1; + return b.startedAt - a.startedAt; + }); + }, [jobs]); + + const runningCount = useMemo( + () => jobs.filter((j) => !isTerminalStatus(j.status)).length, + [jobs], + ); + + const handleLaunch = () => { + if (!selectedProvider) return; + const provider = availableProviders.find((p) => p.id === selectedProvider); + onLaunch({ + provider: selectedProvider, + // Stub command — will be replaced with real agent invocation + command: ['sleep', String(10 + Math.floor(Math.random() * 21))], + label: provider ? `${provider.name}` : selectedProvider, + }); + }; + + return ( +
+ {/* Launch bar */} + {availableProviders.length > 0 && ( +
+
+ {availableProviders.length > 1 ? ( + + ) : ( + + {availableProviders[0]?.name} + + )} + +
+
+ )} + + {/* Job list */} +
+ {sortedJobs.length === 0 ? ( +
+
+ + + +
+

+ No agent jobs yet +

+

+ Launch an agent to get automated review findings +

+
+ ) : ( + sortedJobs.map((job) => ( + onKillJob(job.id)} + expanded={expandedJobId === job.id} + onToggle={() => setExpandedJobId(expandedJobId === job.id ? null : job.id)} + /> + )) + )} +
+ + {/* Kill All footer */} + {runningCount >= 2 && ( +
+ +
+ )} +
+ ); +}; diff --git a/packages/ui/hooks/useAgentJobs.ts b/packages/ui/hooks/useAgentJobs.ts new file mode 100644 index 00000000..dce7b087 --- /dev/null +++ b/packages/ui/hooks/useAgentJobs.ts @@ -0,0 +1,193 @@ +/** + * Real-time agent job state via SSE with polling fallback. + * + * Primary transport: EventSource on /api/agents/jobs/stream. + * Fallback: version-gated GET polling if SSE fails. + * + * Mirrors packages/ui/hooks/useExternalAnnotations.ts in structure. + * + * Gated by an `enabled` option — callers pass their API-mode signal + * to avoid SSE/polling in static or demo contexts where there is no server. + */ + +import { useState, useEffect, useCallback, useRef } from 'react'; +import type { AgentJobInfo, AgentJobEvent, AgentCapabilities } from '../types'; + +const POLL_INTERVAL_MS = 500; +const STREAM_URL = '/api/agents/jobs/stream'; +const JOBS_URL = '/api/agents/jobs'; +const CAPABILITIES_URL = '/api/agents/capabilities'; + +interface UseAgentJobsReturn { + jobs: AgentJobInfo[]; + capabilities: AgentCapabilities | null; + launchJob: (params: { provider?: string; command?: string[]; label?: string }) => Promise; + killJob: (id: string) => Promise; + killAll: () => Promise; +} + +export function useAgentJobs( + options?: { enabled?: boolean }, +): UseAgentJobsReturn { + const enabled = options?.enabled ?? true; + const [jobs, setJobs] = useState([]); + const [capabilities, setCapabilities] = useState(null); + const versionRef = useRef(0); + const fallbackRef = useRef(false); + const pollTimerRef = useRef | null>(null); + const receivedSnapshotRef = useRef(false); + + // Fetch capabilities once on mount + useEffect(() => { + if (!enabled) return; + + fetch(CAPABILITIES_URL) + .then((res) => res.json()) + .then((data) => { + if (data && Array.isArray(data.providers)) { + setCapabilities(data as AgentCapabilities); + } + }) + .catch(() => { + // Silent — capabilities unavailable + }); + }, [enabled]); + + // SSE + polling for job state + useEffect(() => { + if (!enabled) return; + let cancelled = false; + receivedSnapshotRef.current = false; + fallbackRef.current = false; + + // --- SSE primary transport --- + const es = new EventSource(STREAM_URL); + + es.onmessage = (event) => { + if (cancelled) return; + + try { + const parsed: AgentJobEvent = JSON.parse(event.data); + + switch (parsed.type) { + case 'snapshot': + receivedSnapshotRef.current = true; + setJobs(parsed.jobs); + break; + case 'job:started': + setJobs((prev) => [...prev, parsed.job]); + break; + case 'job:updated': + case 'job:completed': + setJobs((prev) => + prev.map((j) => (j.id === parsed.job.id ? parsed.job : j)), + ); + break; + case 'jobs:cleared': + // No-op: killAll() already broadcasts individual job:completed events + // for each killed job, so the UI updates incrementally. + break; + } + } catch { + // Ignore malformed events (e.g., heartbeat comments) + } + }; + + es.onerror = () => { + // If we never received a snapshot, SSE isn't working — fall back to polling + if (!receivedSnapshotRef.current && !fallbackRef.current) { + fallbackRef.current = true; + es.close(); + startPolling(); + } + // Otherwise, EventSource will auto-reconnect and we'll get a fresh snapshot + }; + + // --- Polling fallback --- + function startPolling() { + if (cancelled) return; + + fetchSnapshot(); + + pollTimerRef.current = setInterval(() => { + if (cancelled) return; + fetchSnapshot(); + }, POLL_INTERVAL_MS); + } + + async function fetchSnapshot() { + try { + const url = + versionRef.current > 0 + ? `${JOBS_URL}?since=${versionRef.current}` + : JOBS_URL; + + const res = await fetch(url); + + if (res.status === 304) return; + if (!res.ok) return; + + const data = await res.json(); + if (Array.isArray(data.jobs)) { + setJobs(data.jobs); + } + if (typeof data.version === 'number') { + versionRef.current = data.version; + } + } catch { + // Silent — next poll will retry + } + } + + return () => { + cancelled = true; + es.close(); + if (pollTimerRef.current) { + clearInterval(pollTimerRef.current); + pollTimerRef.current = null; + } + }; + }, [enabled]); + + const launchJob = useCallback( + async (params: { + provider?: string; + command?: string[]; + label?: string; + }): Promise => { + try { + const res = await fetch(JOBS_URL, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(params), + }); + if (!res.ok) return null; + const data = await res.json(); + return data.job ?? null; + } catch { + return null; + } + }, + [], + ); + + const killJob = useCallback(async (id: string) => { + try { + await fetch(`${JOBS_URL}/${encodeURIComponent(id)}`, { + method: 'DELETE', + }); + } catch { + // SSE will reconcile + } + }, []); + + const killAll = useCallback(async () => { + try { + await fetch(JOBS_URL, { method: 'DELETE' }); + } catch { + // SSE will reconcile + } + }, []); + + return { jobs, capabilities, launchJob, killJob, killAll }; +} diff --git a/packages/ui/types.ts b/packages/ui/types.ts index 951ff18d..ba6338cc 100644 --- a/packages/ui/types.ts +++ b/packages/ui/types.ts @@ -138,3 +138,11 @@ export type { EditorAnnotation } from '@plannotator/shared/types'; export type { ExternalAnnotationEvent, } from '@plannotator/shared/external-annotation'; + +export type { + AgentJobInfo, + AgentJobEvent, + AgentJobStatus, + AgentCapability, + AgentCapabilities, +} from '@plannotator/shared/agent-jobs';