activity: follow the agent's background work live + chat-event retention

"Activity" (placeholder name — jobs/tasks were taken) = watch background tasks scroll in parallel with chat. Built DB-free; NOT restarted — deploy + test in the morning. - NDJSON progress contract (activity/progress.ts): capabilities append {job,cap,phase,status,pct,detail,ts,...} lines; tolerant parser treats any JSON object with phase/status as structured progress, else a raw log line. - Backend (activity/router.ts, owner-only, path-guarded): - GET /api/activity/tasks — registry by scanning /tmp/claude-*/<cwd>/tasks/ *.output (harness run_in_background) + announced detached jobs. - POST /api/activity/announce {name,path} — register a detached (setsid) job's log so it's followable too (the setsid case is on the critical path, since the warm worker now makes plain run_in_background the default for heavy jobs). - GET /api/activity/stream?task=<id>|path=<abs> — SSE tail (poll + offset), emitting {kind:'line'|'progress'} with NDJSON parsed. - Frontend /activity screen + Radio nav item: task list (active dot) → live tail with a phase/pct progress header + raw log, following the /system-monitor pattern. - Retention: startChatEventRetention() prunes chat_session_events >7d every 6h (wired in bootstrap) so the durable queue stays bounded. Verified headlessly (no restart): parseTailLine classification, and the scan finds 53 real task output files. Endpoints + UI untested until deploy. Deferred (see handoff): cross-device sync + OpenCode parity (both touch the now-stable chat path — won't ship un-restart-tested); task:progress into chat. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-27 00:56:10 +00:00
parent 6b3eb247a3
commit 35973a5505
10 changed files with 382 additions and 0 deletions
@@ -42,6 +42,7 @@ export function App() {
            <Route path="/files" element={<Dashboard.FilesScreen />} />
            <Route path="/music" element={<Dashboard.MusicScreen />} />
            <Route path="/system-monitor" element={<Dashboard.SystemMonitorScreen />} />
+            <Route path="/activity" element={<Dashboard.ActivityScreen />} />

            <Route path="/code-editor" element={<Dashboard.CodeEditor />} />
            <Route path="/skills" element={<Dashboard.Skills />} />
@@ -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<Registry>({ tasks: [], detached: [] });
+  const [selected, setSelected] = useState<{ label: string; query: string } | null>(null);
+  const [lines, setLines] = useState<string[]>([]);
+  const [progress, setProgress] = useState<ProgressLine | null>(null);
+  const esRef = useRef<EventSource | null>(null);
+  const scrollRef = useRef<HTMLDivElement | null>(null);
+
+  // Poll the registry (harness task files + announced detached jobs).
+  useEffect(() => {
+    let alive = true;
+    const tick = () => get<Registry>('/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 (
+    <div className="flex h-full w-full">
+      <aside className="flex w-72 shrink-0 flex-col overflow-y-auto border-r border-border p-3">
+        <div className="mb-3 flex items-center gap-2 text-sm font-semibold text-foreground">
+          <ActivityIcon size={16} className="text-primary" /> Activity
+        </div>
+
+        <div className="mb-1 px-1 text-xs font-medium uppercase tracking-wide text-muted-foreground">Background tasks</div>
+        {reg.tasks.length === 0 && <p className="px-2 py-1 text-xs text-muted-foreground">none running</p>}
+        {reg.tasks.map((t) => (
+          <button key={t.id} type="button" onClick={() => setSelected({ label: t.id, query: `task=${encodeURIComponent(t.id)}` })} className={rowCls(selected?.label === t.id)} title={t.cwd}>
+            <span className={`inline-block h-2 w-2 shrink-0 rounded-full ${t.active ? 'bg-emerald-500 animate-pulse' : 'bg-muted-foreground/40'}`} />
+            <span className="truncate font-mono text-xs">{t.id}</span>
+          </button>
+        ))}
+
+        {reg.detached.length > 0 && (
+          <div className="mb-1 mt-4 px-1 text-xs font-medium uppercase tracking-wide text-muted-foreground">Detached</div>
+        )}
+        {reg.detached.map((d) => (
+          <button key={d.id} type="button" onClick={() => setSelected({ label: d.id, query: `path=${encodeURIComponent(d.path)}` })} className={rowCls(selected?.label === d.id)} title={d.path}>
+            <FileText size={13} className="shrink-0" />
+            <span className="truncate">{d.id}</span>
+          </button>
+        ))}
+      </aside>
+
+      <main className="flex min-w-0 flex-1 flex-col">
+        {selected ? (
+          <>
+            <div className="shrink-0 border-b border-border p-3">
+              <div className="flex items-center gap-2 text-sm text-foreground">
+                <Radio size={14} className="text-primary" />
+                <span className="truncate font-mono">{selected.label}</span>
+              </div>
+              {progress && (
+                <div className="mt-2">
+                  <div className="flex items-center justify-between text-xs text-muted-foreground">
+                    <span className="truncate">
+                      {[progress.cap, progress.phase].filter(Boolean).join(' · ')}
+                      {progress.status ? ` (${progress.status})` : ''}
+                    </span>
+                    <span className="shrink-0 pl-2">{progress.detail ?? (typeof progress.pct === 'number' ? `${progress.pct}%` : '')}</span>
+                  </div>
+                  {typeof progress.pct === 'number' && (
+                    <div className="mt-1 h-1.5 w-full overflow-hidden rounded-full bg-muted">
+                      <div className="h-full rounded-full bg-primary transition-all" style={{ width: `${Math.min(100, Math.max(0, progress.pct))}%` }} />
+                    </div>
+                  )}
+                </div>
+              )}
+            </div>
+            <div ref={scrollRef} className="min-h-0 flex-1 overflow-y-auto bg-black/30 p-3 font-mono text-xs text-foreground/80">
+              {lines.length === 0 ? (
+                <span className="text-muted-foreground">waiting for output…</span>
+              ) : (
+                lines.map((l, i) => <div key={i} className="whitespace-pre-wrap break-words">{l}</div>)
+              )}
+            </div>
+          </>
+        ) : (
+          <div className="flex h-full items-center justify-center text-sm text-muted-foreground">Select a task to follow its live output</div>
+        )}
+      </main>
+    </div>
+  );
+};
@@ -0,0 +1 @@
+export * from './ActivityScreen';
@@ -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' },
 ];

@@ -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';
@@ -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<string, unknown>;
+      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 };
+}
@@ -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-<uid>/<encoded-cwd>/tasks/<task_id>.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/<task_id>.output).
+async function listTaskFiles(): Promise<TaskFile[]> {
+  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<Announced[]> {
+  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<string | null> {
+  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=<id> | ?path=<abs> — 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<Uint8Array>({
+    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' },
+  });
+});
@@ -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`,
+  );
+}
@@ -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);
  });
@@ -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);