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>
This commit is contained in:
@@ -42,6 +42,7 @@ export function App() {
|
|||||||
<Route path="/files" element={<Dashboard.FilesScreen />} />
|
<Route path="/files" element={<Dashboard.FilesScreen />} />
|
||||||
<Route path="/music" element={<Dashboard.MusicScreen />} />
|
<Route path="/music" element={<Dashboard.MusicScreen />} />
|
||||||
<Route path="/system-monitor" element={<Dashboard.SystemMonitorScreen />} />
|
<Route path="/system-monitor" element={<Dashboard.SystemMonitorScreen />} />
|
||||||
|
<Route path="/activity" element={<Dashboard.ActivityScreen />} />
|
||||||
|
|
||||||
<Route path="/code-editor" element={<Dashboard.CodeEditor />} />
|
<Route path="/code-editor" element={<Dashboard.CodeEditor />} />
|
||||||
<Route path="/skills" element={<Dashboard.Skills />} />
|
<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,
|
Workflow,
|
||||||
Music,
|
Music,
|
||||||
Activity,
|
Activity,
|
||||||
|
Radio,
|
||||||
} from 'lucide-react';
|
} from 'lucide-react';
|
||||||
|
|
||||||
export const ALL_DOCK_ITEMS: DockItem[] = [
|
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: 'Browser', to: '/browser', icon: Globe, color: '#06b6d4' },
|
||||||
{ label: 'Desktop', to: '/desktop', icon: MonitorSmartphone, color: '#ec4899' },
|
{ label: 'Desktop', to: '/desktop', icon: MonitorSmartphone, color: '#ec4899' },
|
||||||
{ label: 'Monitor', to: '/system-monitor', icon: Activity, color: '#0ea5e9' },
|
{ 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' },
|
{ label: 'Dashboards', to: '/dashboards', icon: LayoutGrid, color: '#8b5cf6' },
|
||||||
];
|
];
|
||||||
|
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ export * from './Tasks';
|
|||||||
export * from './Files';
|
export * from './Files';
|
||||||
export * from './Music';
|
export * from './Music';
|
||||||
export * from './SystemMonitor';
|
export * from './SystemMonitor';
|
||||||
|
export * from './Activity';
|
||||||
export * from './CodeEditor';
|
export * from './CodeEditor';
|
||||||
export * from './ChatHistory';
|
export * from './ChatHistory';
|
||||||
export * from './Dashboards';
|
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 { startDiscordBotIfConfigured } from './channels/discord/bot';
|
||||||
import { startTelegramBotIfConfigured } from './channels/telegram/bot';
|
import { startTelegramBotIfConfigured } from './channels/telegram/bot';
|
||||||
import { startWhatsAppBotIfConfigured } from './channels/whatsapp/bot';
|
import { startWhatsAppBotIfConfigured } from './channels/whatsapp/bot';
|
||||||
|
import { startChatEventRetention } from './api/chat/retention';
|
||||||
|
|
||||||
mkdirSync(DATA_PATH, { recursive: true });
|
mkdirSync(DATA_PATH, { recursive: true });
|
||||||
ensureItemDirs();
|
ensureItemDirs();
|
||||||
@@ -13,6 +14,8 @@ ensureItemDirs();
|
|||||||
ensureToolLoader();
|
ensureToolLoader();
|
||||||
// Queue is initialized by the sidecar process
|
// Queue is initialized by the sidecar process
|
||||||
|
|
||||||
|
startChatEventRetention();
|
||||||
|
|
||||||
await startDiscordBotIfConfigured().catch((err) => {
|
await startDiscordBotIfConfigured().catch((err) => {
|
||||||
console.error('[channels] Failed to start Discord bot:', 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 { router as fileBrowserRouter } from './api/file-browser/router';
|
||||||
import { musicRouter } from './api/music/router';
|
import { musicRouter } from './api/music/router';
|
||||||
import { systemMonitorRouter } from './api/system-monitor/system-monitor';
|
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 './api/music/sidecar-server'; // side-effect: capture the officer-music audio server port
|
||||||
import { devServerRouter, devServerProxyRouter } from './api/dev-server/router';
|
import { devServerRouter, devServerProxyRouter } from './api/dev-server/router';
|
||||||
import { dockRouter } from './api/dock/dock';
|
import { dockRouter } from './api/dock/dock';
|
||||||
@@ -95,6 +96,7 @@ protectedRouter.route('/task-logs', taskLogsRouter);
|
|||||||
protectedRouter.route('/file-browser', fileBrowserRouter);
|
protectedRouter.route('/file-browser', fileBrowserRouter);
|
||||||
protectedRouter.route('/music', musicRouter);
|
protectedRouter.route('/music', musicRouter);
|
||||||
protectedRouter.route('/system-monitor', systemMonitorRouter);
|
protectedRouter.route('/system-monitor', systemMonitorRouter);
|
||||||
|
protectedRouter.route('/activity', activityRouter);
|
||||||
protectedRouter.route('/dev-server', devServerRouter);
|
protectedRouter.route('/dev-server', devServerRouter);
|
||||||
protectedRouter.route('/dock', dockRouter);
|
protectedRouter.route('/dock', dockRouter);
|
||||||
protectedRouter.route('/integrations', integrationsRouter);
|
protectedRouter.route('/integrations', integrationsRouter);
|
||||||
|
|||||||
Reference in New Issue
Block a user