diff --git a/src/apps/officer-web/App.tsx b/src/apps/officer-web/App.tsx index 949548eb..980be5d4 100644 --- a/src/apps/officer-web/App.tsx +++ b/src/apps/officer-web/App.tsx @@ -42,6 +42,7 @@ export function App() { } /> } /> } /> + } /> } /> } /> diff --git a/src/apps/officer-web/Screens/Dashboard/Activity/ActivityScreen.tsx b/src/apps/officer-web/Screens/Dashboard/Activity/ActivityScreen.tsx new file mode 100644 index 00000000..4e2d5b37 --- /dev/null +++ b/src/apps/officer-web/Screens/Dashboard/Activity/ActivityScreen.tsx @@ -0,0 +1,120 @@ +import { useEffect, useRef, useState } from 'react'; +import { useClient } from 'hooks/useClient'; +import { Radio, FileText, Activity as ActivityIcon } from 'lucide-react'; + +type TaskRow = { id: string; source: 'harness'; cwd: string; sizeBytes: number; updatedAt: number; active: boolean }; +type DetachedRow = { id: string; source: 'detached'; path: string; announcedAt: number }; +type Registry = { tasks: TaskRow[]; detached: DetachedRow[] }; +type ProgressLine = { phase?: string; status?: string; pct?: number; detail?: string; cap?: string; job?: string }; + +const POLL_MS = 3000; +const MAX_LINES = 600; + +export const ActivityScreen = () => { + const { token, get } = useClient(); + const [reg, setReg] = useState({ tasks: [], detached: [] }); + const [selected, setSelected] = useState<{ label: string; query: string } | null>(null); + const [lines, setLines] = useState([]); + const [progress, setProgress] = useState(null); + const esRef = useRef(null); + const scrollRef = useRef(null); + + // Poll the registry (harness task files + announced detached jobs). + useEffect(() => { + let alive = true; + const tick = () => get('/activity/tasks').then((r) => alive && setReg(r)).catch(() => {}); + tick(); + const iv = setInterval(tick, POLL_MS); + return () => { alive = false; clearInterval(iv); }; + }, []); + + // Live-tail the selected task via SSE (EventSource can't set headers → token in the query string). + useEffect(() => { + esRef.current?.close(); + setLines([]); + setProgress(null); + if (!selected) return; + const es = new EventSource(`/api/activity/stream?${selected.query}&token=${encodeURIComponent(token ?? '')}`); + esRef.current = es; + es.onmessage = (ev) => { + try { + const d = JSON.parse(ev.data) as { kind: string; text?: string; progress?: ProgressLine }; + if (d.kind === 'progress' && d.progress) setProgress(d.progress); + else if (d.kind === 'line' && typeof d.text === 'string') setLines((prev) => [...prev.slice(-(MAX_LINES - 1)), d.text!]); + } catch { /* ignore */ } + }; + return () => es.close(); + }, [selected, token]); + + useEffect(() => { scrollRef.current?.scrollTo(0, scrollRef.current.scrollHeight); }, [lines]); + + const rowCls = (active: boolean) => + `flex w-full items-center gap-2 rounded-md px-2 py-1.5 text-left text-sm ${active ? 'bg-muted text-foreground' : 'text-muted-foreground hover:bg-muted/60 hover:text-foreground'}`; + + return ( +
+ + +
+ {selected ? ( + <> +
+
+ + {selected.label} +
+ {progress && ( +
+
+ + {[progress.cap, progress.phase].filter(Boolean).join(' · ')} + {progress.status ? ` (${progress.status})` : ''} + + {progress.detail ?? (typeof progress.pct === 'number' ? `${progress.pct}%` : '')} +
+ {typeof progress.pct === 'number' && ( +
+
+
+ )} +
+ )} +
+
+ {lines.length === 0 ? ( + waiting for output… + ) : ( + lines.map((l, i) =>
{l}
) + )} +
+ + ) : ( +
Select a task to follow its live output
+ )} +
+
+ ); +}; diff --git a/src/apps/officer-web/Screens/Dashboard/Activity/index.tsx b/src/apps/officer-web/Screens/Dashboard/Activity/index.tsx new file mode 100644 index 00000000..83299756 --- /dev/null +++ b/src/apps/officer-web/Screens/Dashboard/Activity/index.tsx @@ -0,0 +1 @@ +export * from './ActivityScreen'; diff --git a/src/apps/officer-web/Screens/Dashboard/Layout/Dock.tsx b/src/apps/officer-web/Screens/Dashboard/Layout/Dock.tsx index 09da47c9..8a0dfc7a 100644 --- a/src/apps/officer-web/Screens/Dashboard/Layout/Dock.tsx +++ b/src/apps/officer-web/Screens/Dashboard/Layout/Dock.tsx @@ -130,6 +130,7 @@ import { Workflow, Music, Activity, + Radio, } from 'lucide-react'; export const ALL_DOCK_ITEMS: DockItem[] = [ @@ -147,6 +148,7 @@ export const ALL_DOCK_ITEMS: DockItem[] = [ { label: 'Browser', to: '/browser', icon: Globe, color: '#06b6d4' }, { label: 'Desktop', to: '/desktop', icon: MonitorSmartphone, color: '#ec4899' }, { label: 'Monitor', to: '/system-monitor', icon: Activity, color: '#0ea5e9' }, + { label: 'Activity', to: '/activity', icon: Radio, color: '#f59e0b' }, { label: 'Dashboards', to: '/dashboards', icon: LayoutGrid, color: '#8b5cf6' }, ]; diff --git a/src/apps/officer-web/Screens/Dashboard/index.tsx b/src/apps/officer-web/Screens/Dashboard/index.tsx index 6ff60bf3..17d8198e 100644 --- a/src/apps/officer-web/Screens/Dashboard/index.tsx +++ b/src/apps/officer-web/Screens/Dashboard/index.tsx @@ -12,6 +12,7 @@ export * from './Tasks'; export * from './Files'; export * from './Music'; export * from './SystemMonitor'; +export * from './Activity'; export * from './CodeEditor'; export * from './ChatHistory'; export * from './Dashboards'; diff --git a/src/servers/api/activity/progress.ts b/src/servers/api/activity/progress.ts new file mode 100644 index 00000000..877ce51e --- /dev/null +++ b/src/servers/api/activity/progress.ts @@ -0,0 +1,39 @@ +// Shared structured-progress contract for the Activity feature. +// +// Any capability that wants live progress bars / phase chips (instead of the UI regexing prose) appends +// NDJSON lines of this shape to its output/log file, alongside whatever human text it already writes: +// +// {"job":"dearly-devoted","cap":"split-audiobook","phase":"transcription","status":"running","pct":9,"detail":"0:50:00 / 9:28:24","ts":1730000000000} +// {"job":"dearly-devoted","cap":"split-audiobook","phase":"verdict","status":"complete","verdict":"CLEAN","chapters":31,"pieces":52,"ts":...} +// +// The parser is deliberately tolerant: a line is treated as structured progress iff it JSON-parses to +// an object carrying a `phase` or `status`; everything else is passed through as a raw log line. So a +// capability can dual-write (human lines + NDJSON) and both render correctly. + +export type ProgressLine = { + job?: string; + cap?: string; + phase?: string; + status?: 'running' | 'complete' | 'failed' | string; + pct?: number; + detail?: string; + ts?: number; + [k: string]: unknown; +}; + +export type TailEvent = { kind: 'progress'; progress: ProgressLine } | { kind: 'line'; text: string }; + +export function parseTailLine(line: string): TailEvent { + const trimmed = line.trim(); + if (trimmed.startsWith('{') && trimmed.endsWith('}')) { + try { + const obj = JSON.parse(trimmed) as Record; + if (obj && (typeof obj.phase === 'string' || typeof obj.status === 'string')) { + return { kind: 'progress', progress: obj as ProgressLine }; + } + } catch { + /* not JSON → fall through to raw line */ + } + } + return { kind: 'line', text: line }; +} diff --git a/src/servers/api/activity/router.ts b/src/servers/api/activity/router.ts new file mode 100644 index 00000000..c91f0a9e --- /dev/null +++ b/src/servers/api/activity/router.ts @@ -0,0 +1,191 @@ +import { readdir, stat, readFile, writeFile, mkdir, realpath, open } from 'node:fs/promises'; +import { join, dirname, resolve as resolvePath } from 'node:path'; +import { createRouter } from '../../create-router'; +import { parseTailLine } from './progress'; + +// "Activity" — follow the agent's background work live. Two producers, one primitive (tail a file): +// - Harness `run_in_background` tasks write /tmp/claude-//tasks/.output +// (append-only, live). Discovered by scanning; the FILE is the source of truth. +// - Detached (setsid) jobs write an arbitrary log; they self-register via POST /announce with the log +// path, so they're followable too (the setsid case is on the critical path, not a footnote). +// Owner-only (the account gate confines non-owners to /auth+/music). Paths are traversal-guarded. + +export const activityRouter = createRouter(); + +const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); +const HOME_DIR = process.env.HOME_DIR ?? process.env.HOME ?? ''; +const ANNOUNCED_PATH = join(DATA_PATH, 'activity', 'announced.json'); +const ALLOWED_ROOTS = ['/tmp', DATA_PATH, HOME_DIR].filter(Boolean); +const ACTIVE_WINDOW_MS = 120_000; // a task file touched within this is considered "active" + +type TaskFile = { taskId: string; path: string; cwdLabel: string; sizeBytes: number; mtimeMs: number }; +type Announced = { name: string; path: string; ts: number }; + +// Scan the claude tmp tree for harness background-task output files (…/tasks/.output). +async function listTaskFiles(): Promise { + const out: TaskFile[] = []; + let tmp: string[]; + try { + tmp = await readdir('/tmp'); + } catch { + return out; + } + for (const name of tmp) { + if (!name.startsWith('claude-')) continue; + const base = join('/tmp', name); + let cwdDirs: import('node:fs').Dirent[]; + try { + cwdDirs = await readdir(base, { withFileTypes: true }); + } catch { + continue; + } + for (const cwd of cwdDirs) { + if (!cwd.isDirectory()) continue; + const tasksDir = join(base, cwd.name, 'tasks'); + let files: string[]; + try { + files = await readdir(tasksDir); + } catch { + continue; + } + for (const f of files) { + if (!f.endsWith('.output')) continue; + const p = join(tasksDir, f); + const st = await stat(p).catch(() => null); + if (!st) continue; + out.push({ taskId: f.slice(0, -'.output'.length), path: p, cwdLabel: cwd.name, sizeBytes: st.size, mtimeMs: st.mtimeMs }); + } + } + } + return out; +} + +async function readAnnounced(): Promise { + try { + return JSON.parse(await readFile(ANNOUNCED_PATH, 'utf8')) as Announced[]; + } catch { + return []; + } +} + +/** Resolve a path only if it lands under an allowed root (guards traversal); the file need not exist yet. */ +async function resolveAllowed(p: string): Promise { + const under = (abs: string) => ALLOWED_ROOTS.some((root) => abs === root || abs.startsWith(root + '/')); + try { + const real = await realpath(p); + return under(real) ? real : null; + } catch { + const abs = resolvePath(p); + return under(abs) ? abs : null; + } +} + +// GET /tasks — registry of followable background work. +activityRouter.get('/tasks', async (ctx) => { + const [harness, announced] = await Promise.all([listTaskFiles(), readAnnounced()]); + const now = Date.now(); + return ctx.json({ + tasks: harness + .sort((a, b) => b.mtimeMs - a.mtimeMs) + .map((t) => ({ + id: t.taskId, + source: 'harness' as const, + cwd: t.cwdLabel, + sizeBytes: t.sizeBytes, + updatedAt: Math.round(t.mtimeMs), + active: now - t.mtimeMs < ACTIVE_WINDOW_MS, + })), + detached: announced.map((a) => ({ id: a.name, source: 'detached' as const, path: a.path, announcedAt: a.ts })), + }); +}); + +// POST /announce { name, path } — register a detached (setsid) job's log so it's followable too. +activityRouter.post('/announce', async (ctx) => { + const body = ctx.get('body') as { name?: string; path?: string } | undefined; + if (!body?.name || !body?.path) return ctx.text('name and path are required', 400); + const safe = await resolveAllowed(body.path); + if (!safe) return ctx.text('path is not under an allowed root', 403); + await mkdir(dirname(ANNOUNCED_PATH), { recursive: true }); + const list = await readAnnounced(); + const next = [{ name: body.name, path: safe, ts: Date.now() }, ...list.filter((a) => a.name !== body.name)].slice(0, 100); + await writeFile(ANNOUNCED_PATH, JSON.stringify(next)); + return ctx.json({ ok: true, name: body.name, path: safe }); +}); + +// GET /stream?task= | ?path= — SSE that tails the output file, emitting {kind:'line'|'progress'}. +activityRouter.get('/stream', async (ctx) => { + const task = ctx.req.query('task'); + const pathQ = ctx.req.query('path'); + + let target: string | null = null; + if (task) { + target = (await listTaskFiles()).find((f) => f.taskId === task)?.path ?? null; + } else if (pathQ) { + target = await resolveAllowed(pathQ); + } + if (!target) return ctx.text('a valid task id or path is required', 400); + + const filePath = target; + const enc = new TextEncoder(); + + const stream = new ReadableStream({ + async start(controller) { + let offset = 0; + let closed = false; + const send = (obj: unknown) => { + try { + controller.enqueue(enc.encode(`data: ${JSON.stringify(obj)}\n\n`)); + } catch { + /* controller closed */ + } + }; + + const drain = async () => { + try { + const st = await stat(filePath); + if (st.size < offset) offset = 0; // truncated / rotated + if (st.size > offset) { + const fh = await open(filePath, 'r'); + const buf = Buffer.alloc(st.size - offset); + await fh.read(buf, 0, buf.length, offset); + await fh.close(); + offset = st.size; + for (const line of buf.toString('utf8').split('\n')) { + if (line !== '') send(parseTailLine(line)); + } + } + } catch { + /* file may not exist yet — keep polling */ + } + }; + + await drain(); + const poll = setInterval(drain, 800); + const heartbeat = setInterval(() => { + try { + controller.enqueue(enc.encode(': hb\n\n')); + } catch { + /* closed */ + } + }, 15_000); + + const cleanup = () => { + if (closed) return; + closed = true; + clearInterval(poll); + clearInterval(heartbeat); + try { + controller.close(); + } catch { + /* already closed */ + } + }; + ctx.req.raw.signal.addEventListener('abort', cleanup); + setTimeout(cleanup, 60 * 60 * 1000); // safety cap + }, + }); + + return new Response(stream, { + headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive' }, + }); +}); diff --git a/src/servers/api/chat/retention.ts b/src/servers/api/chat/retention.ts new file mode 100644 index 00000000..c1a286aa --- /dev/null +++ b/src/servers/api/chat/retention.ts @@ -0,0 +1,22 @@ +import { pruneChatEventsOlderThan } from 'officerdb'; + +// The durable chat event queue (chat_session_events) grows with every turn. Prune old rows so it +// stays bounded — a session's transcript (Claude's .jsonl) is the real record; the queue only needs +// enough history to cover reconnect replay, so a few days is generous. +const RETENTION_MS = 7 * 24 * 60 * 60 * 1000; // keep 7 days +const INTERVAL_MS = 6 * 60 * 60 * 1000; // prune every 6h + +export function startChatEventRetention(): void { + const tick = async () => { + try { + await pruneChatEventsOlderThan(new Date(Date.now() - RETENTION_MS)); + } catch (err) { + console.error('[chat-retention] prune failed:', err instanceof Error ? err.message : err); + } + }; + void tick(); // sweep once at startup + setInterval(tick, INTERVAL_MS); + console.log( + `[chat-retention] pruning chat_session_events older than ${RETENTION_MS / 86_400_000}d, every ${INTERVAL_MS / 3_600_000}h`, + ); +} diff --git a/src/servers/bootstrap.ts b/src/servers/bootstrap.ts index 3a105346..6e856e05 100644 --- a/src/servers/bootstrap.ts +++ b/src/servers/bootstrap.ts @@ -5,6 +5,7 @@ import { ensureToolLoader } from './ensure-tool-loader'; import { startDiscordBotIfConfigured } from './channels/discord/bot'; import { startTelegramBotIfConfigured } from './channels/telegram/bot'; import { startWhatsAppBotIfConfigured } from './channels/whatsapp/bot'; +import { startChatEventRetention } from './api/chat/retention'; mkdirSync(DATA_PATH, { recursive: true }); ensureItemDirs(); @@ -13,6 +14,8 @@ ensureItemDirs(); ensureToolLoader(); // Queue is initialized by the sidecar process + startChatEventRetention(); + await startDiscordBotIfConfigured().catch((err) => { console.error('[channels] Failed to start Discord bot:', err); }); diff --git a/src/servers/hono.ts b/src/servers/hono.ts index 55e2d3fa..a98f2821 100644 --- a/src/servers/hono.ts +++ b/src/servers/hono.ts @@ -21,6 +21,7 @@ import { taskLogsRouter } from './api/task-logs/task-logs'; import { router as fileBrowserRouter } from './api/file-browser/router'; import { musicRouter } from './api/music/router'; import { systemMonitorRouter } from './api/system-monitor/system-monitor'; +import { activityRouter } from './api/activity/router'; import './api/music/sidecar-server'; // side-effect: capture the officer-music audio server port import { devServerRouter, devServerProxyRouter } from './api/dev-server/router'; import { dockRouter } from './api/dock/dock'; @@ -95,6 +96,7 @@ protectedRouter.route('/task-logs', taskLogsRouter); protectedRouter.route('/file-browser', fileBrowserRouter); protectedRouter.route('/music', musicRouter); protectedRouter.route('/system-monitor', systemMonitorRouter); +protectedRouter.route('/activity', activityRouter); protectedRouter.route('/dev-server', devServerRouter); protectedRouter.route('/dock', dockRouter); protectedRouter.route('/integrations', integrationsRouter);