diff --git a/server/index.test.ts b/server/index.test.ts index 4cba6a9d7..eb856d2b9 100644 --- a/server/index.test.ts +++ b/server/index.test.ts @@ -756,6 +756,19 @@ describe("harness HTTP API", () => { } }); + it("validates the event inspector limit at the HTTP boundary", async () => { + const bot = (await api("GET", "/api/bots")).body.bots[0]; + for (const value of ["nope", "0", "-1", "1.5", "Infinity"]) { + const response = await api("GET", `/api/threads/${bot.threadId}/events?limit=${value}`); + expect(response.status).toBe(400); + expect(response.body.error).toContain("positive whole number"); + } + const ok = await api("GET", `/api/threads/${bot.threadId}/events?limit=1`); + expect(ok.status).toBe(200); + expect(Array.isArray(ok.body.entries)).toBe(true); + expect(ok.body.total).toEqual({ runtime: expect.any(Number), native: expect.any(Number) }); + }); + it("404s unknown routes with the route in the error", async () => { const res = await api("GET", "/api/definitely-not-a-route"); expect(res.status).toBe(404); diff --git a/server/index.ts b/server/index.ts index 6bdeb5134..cda2206ed 100644 --- a/server/index.ts +++ b/server/index.ts @@ -75,6 +75,7 @@ import { RepeatDetector, callKey } from "./repeat-detector.ts"; import { RoutineManager, type RoutineRunOn, type RoutineRunTrigger } from "./routines.ts"; import { fetchGithubTeam, fetchLibraryTeam, fetchTeamCatalog } from "./team-library.ts"; import { createTeamManifest, parseTeamManifest } from "./team-manifest.ts"; +import { readThreadEvents } from "./thread-events.ts"; import { listenWebhookIngress, webhookCredential, type WebhookIngress } from "./webhook-ingress.ts"; import { memberTurnSelection } from "./member-turn.ts"; import { WebhookManager } from "./webhooks.ts"; @@ -2951,6 +2952,25 @@ const server = createServer(async (req, res) => { return json(res, 200, { app: "openmausbot", pid: process.pid, static: Boolean(STATIC_DIR) }); } + // ── inspector: a thread's runtime events + native protocol tee ── + // Both logs already exist on disk; this only reads them back. Threads + // belong to bots or rooms — anything else is not a thread we know. + m = path.match(/^\/api\/threads\/([\w-]+)\/events$/); + if (m && method === "GET") { + const threadId = m[1]; + const known = + store.bots.some((b) => store.tasks(b.id).some((t) => t.threadId === threadId)) || + Boolean(store.groupByThread(threadId)); + if (!known) return json(res, 404, { error: "no such thread" }); + const rawLimit = url.searchParams.get("limit"); + const parsedLimit = rawLimit === null ? undefined : Number(rawLimit); + if (parsedLimit !== undefined && (!Number.isInteger(parsedLimit) || parsedLimit <= 0)) { + return json(res, 400, { error: "limit must be a positive whole number" }); + } + const limit = parsedLimit; + return json(res, 200, readThreadEvents({ eventsDir: EVENTS_DIR, nativeDir: NATIVE_DIR, threadId, limit })); + } + // ── provider instances (model picker) ── if (method === "GET" && path === "/api/instances") { // Rescan PATH first: this endpoint is how the app answers "what can I diff --git a/server/thread-events.test.ts b/server/thread-events.test.ts new file mode 100644 index 000000000..c43f97e1b --- /dev/null +++ b/server/thread-events.test.ts @@ -0,0 +1,141 @@ +import { appendFileSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; + +import { readThreadEvents } from "./thread-events.ts"; + +const dirs: string[] = []; +function tmp() { + const d = mkdtempSync(join(tmpdir(), "omb-thread-events-")); + dirs.push(d); + return d; +} +afterEach(() => { + for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }); +}); + +const line = (o: unknown) => JSON.stringify(o) + "\n"; +const runtime = (event: Record) => ({ provider: "test", threadId: "t1", ...event }); + +describe("readThreadEvents", () => { + it("returns an empty page when neither log exists", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + expect(readThreadEvents({ eventsDir, nativeDir, threadId: "t1" })).toEqual({ + entries: [], + total: { runtime: 0, native: 0 }, + }); + }); + + it("merges runtime and native lines by time, tagging their source", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + writeFileSync( + join(eventsDir, "t1.ndjson"), + line(runtime({ eventId: "e1", type: "turn.started", createdAt: "2026-08-17T10:00:00.000Z" })) + + line(runtime({ eventId: "e2", type: "turn.completed", createdAt: "2026-08-17T10:00:02.000Z", ok: true })), + ); + writeFileSync( + join(nativeDir, "t1.ndjson"), + line({ at: "2026-08-17T10:00:01.000Z", dir: "out", source: "claude.sdk.message", msg: { type: "user" } }), + ); + const page = readThreadEvents({ eventsDir, nativeDir, threadId: "t1" }); + expect(page.total).toEqual({ runtime: 2, native: 1 }); + expect(page.entries.map((e) => [e.kind, e.at])).toEqual([ + ["runtime", "2026-08-17T10:00:00.000Z"], + ["native", "2026-08-17T10:00:01.000Z"], + ["runtime", "2026-08-17T10:00:02.000Z"], + ]); + // each entry keeps its original record whole under `data` + expect(page.entries[1]).toMatchObject({ kind: "native", data: { dir: "out", msg: { type: "user" } } }); + expect(page.entries[0]).toMatchObject({ kind: "runtime", data: { eventId: "e1" } }); + }); + + it("caps each log to its most recent `limit` lines and reports what it skipped", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + let body = ""; + for (let i = 0; i < 10; i++) { + body += line(runtime({ eventId: `e${i}`, type: "content.delta", createdAt: `2026-08-17T10:00:${String(i).padStart(2, "0")}.000Z`, streamKind: "assistant_text", delta: String(i) })); + } + writeFileSync(join(eventsDir, "t1.ndjson"), body); + const page = readThreadEvents({ eventsDir, nativeDir, threadId: "t1", limit: 3 }); + expect(page.entries.map((e) => (e.data as { eventId: string }).eventId)).toEqual(["e7", "e8", "e9"]); + expect(page.total.runtime).toBe(10); + }); + + it("skips a corrupt line rather than failing the whole read", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + writeFileSync( + join(eventsDir, "t1.ndjson"), + line(runtime({ eventId: "e1", type: "turn.started", createdAt: "2026-08-17T10:00:00.000Z" })) + + "{not json\n" + + line(runtime({ eventId: "e2", type: "turn.completed", createdAt: "2026-08-17T10:00:02.000Z", ok: true })), + ); + const page = readThreadEvents({ eventsDir, nativeDir, threadId: "t1" }); + expect(page.entries).toHaveLength(2); + // `total` is the number of non-empty log lines; malformed records are + // counted but deliberately absent from the returned entries. + expect(page.total.runtime).toBe(3); + }); + + it("discards JSON-valid records that do not satisfy the inspector wire contract", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + writeFileSync( + join(eventsDir, "t1.ndjson"), + line(null) + + line({ eventId: "incomplete", provider: "claude", threadId: "t1", createdAt: "1", type: "content.delta", streamKind: "assistant_text" }) + + line({ eventId: "valid", provider: "claude", threadId: "t1", createdAt: "2", type: "content.delta", streamKind: "assistant_text", delta: "ok" }), + ); + writeFileSync(join(nativeDir, "t1.ndjson"), line(null) + line({ at: "2", dir: "in", source: "claude", msg: {} })); + const page = readThreadEvents({ eventsDir, nativeDir, threadId: "t1" }); + expect(page.entries.map((entry) => [entry.kind, (entry.data as { eventId?: string }).eventId])).toEqual([ + ["runtime", "valid"], + ["native", undefined], + ]); + expect(page.total).toEqual({ runtime: 3, native: 2 }); + }); + + it("keeps walking backward when a corrupt tail record would otherwise consume the limit", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + const body = Array.from({ length: 20 }, (_, i) => + line(runtime({ eventId: `e${i}`, type: "content.delta", createdAt: `2026-08-17T10:00:${String(i).padStart(2, "0")}.000Z`, streamKind: "assistant_text", delta: String(i) })), + ).join(""); + writeFileSync(join(eventsDir, "t1.ndjson"), body + "{broken}\n"); + const page = readThreadEvents({ eventsDir, nativeDir, threadId: "t1", limit: 3 }); + expect(page.entries.map((e) => (e.data as { eventId: string }).eventId)).toEqual(["e17", "e18", "e19"]); + expect(page.total.runtime).toBe(21); + }); + + it("normalizes non-finite and fractional limits at the helper boundary", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + writeFileSync(join(eventsDir, "t1.ndjson"), line(runtime({ eventId: "e1", createdAt: "1", type: "turn.started" })) + line(runtime({ eventId: "e2", createdAt: "2", type: "turn.started" }))); + expect(readThreadEvents({ eventsDir, nativeDir, threadId: "t1", limit: Number.NaN }).entries).toHaveLength(2); + expect(readThreadEvents({ eventsDir, nativeDir, threadId: "t1", limit: 1.9 }).entries).toHaveLength(1); + }); + + it("updates cached totals from appended bytes and preserves multibyte records across read chunks", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + const file = join(eventsDir, "t1.ndjson"); + writeFileSync(file, line(runtime({ eventId: "large", createdAt: "1", type: "turn.started", text: "🐭".repeat(40_000) }))); + expect(readThreadEvents({ eventsDir, nativeDir, threadId: "t1", limit: 2 }).total.runtime).toBe(1); + + appendFileSync(file, line(runtime({ eventId: "latest", createdAt: "2", type: "turn.started" }))); + const page = readThreadEvents({ eventsDir, nativeDir, threadId: "t1", limit: 2 }); + expect(page.total.runtime).toBe(2); + expect(page.entries.map((entry) => (entry.data as { eventId: string }).eventId)).toEqual(["large", "latest"]); + expect((page.entries[0]!.data as { text: string }).text.startsWith("🐭🐭")).toBe(true); + }); + + it("refuses a thread id that could escape the log directory", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + expect(() => readThreadEvents({ eventsDir, nativeDir, threadId: "../bots" })).toThrow(/thread id/); + }); +}); diff --git a/server/thread-events.ts b/server/thread-events.ts new file mode 100644 index 000000000..ee7e8c6ab --- /dev/null +++ b/server/thread-events.ts @@ -0,0 +1,264 @@ +// The inspector's data: what a thread's turn actually looked like on the +// wire. Nothing new is captured here — the harness already tees two logs +// per thread, and this just reads them back: +// +// events/.ndjson — the normalized RuntimeEvent stream the bus +// publishes (server/harness/bus.ts) +// native/.ndjson — the provider's own protocol messages, +// verbatim and secret-redacted +// (server/drivers/native.ts) +// +// Merged by timestamp so a tool call and the raw message behind it sit +// next to each other. Newest-`limit` only: a long-lived thread has +// thousands of native lines and the panel wants the recent ones first. +import { closeSync, fstatSync, openSync, readSync, type Stats } from "node:fs"; +import { join } from "node:path"; +import type { RuntimeEvent } from "./contracts.ts"; + +/** One line of native/.ndjson (server/drivers/native.ts). */ +export interface NativeRecord { + at: string; + dir: "in" | "out"; + source: string; + msg: unknown; +} + +export type InspectorEntry = + | { kind: "runtime"; at: string; data: RuntimeEvent } + | { kind: "native"; at: string; data: NativeRecord }; + +export interface InspectorPage { + entries: InspectorEntry[]; + /** line counts before the cap, so the UI can say "showing 200 of 1,687" */ + total: { runtime: number; native: number }; +} + +const DEFAULT_LIMIT = 300; +const MAX_LIMIT = 2000; +const READ_CHUNK = 64 * 1024; +const MAX_TAIL_BYTES = 8 * 1024 * 1024; + +interface LineCount { + dev: number; + ino: number; + size: number; + mtimeMs: number; + complete: number; + trailing: boolean; +} +type FileStat = Pick; + +// Counts are incremental per append-only log. The first request scans bytes +// once (without decoding or parsing every JSON record); later requests inspect +// only bytes appended since the cached size. Keep this bounded across threads. +const lineCounts = new Map(); +const LINE_COUNT_CACHE_MAX = 256; + +/** Thread ids are uuids the harness minted; anything else is not a file we + * should be reading. */ +function assertThreadId(threadId: string) { + if (!/^[\w-]+$/.test(threadId)) throw new Error("invalid thread id"); +} + +function countLines(fd: number, file: string, stat: FileStat): number { + const previous = lineCounts.get(file); + const appended = + previous && + previous.dev === stat.dev && + previous.ino === stat.ino && + stat.size >= previous.size && + (stat.size > previous.size || stat.mtimeMs === previous.mtimeMs); + if (appended && stat.size === previous.size) return previous.complete + Number(previous.trailing); + + let offset = appended ? previous.size : 0; + let complete = appended ? previous.complete : 0; + let trailing = appended ? previous.trailing : false; + while (offset < stat.size) { + const length = Math.min(READ_CHUNK, stat.size - offset); + const chunk = Buffer.allocUnsafe(length); + const read = readSync(fd, chunk, 0, length, offset); + if (read <= 0) break; + for (let i = 0; i < read; i++) { + if (chunk[i] === 0x0a) { + if (trailing) complete++; + trailing = false; + } else if (chunk[i] !== 0x0d) { + trailing = true; + } + } + offset += read; + } + const next = { dev: stat.dev, ino: stat.ino, size: stat.size, mtimeMs: stat.mtimeMs, complete, trailing }; + lineCounts.delete(file); + lineCounts.set(file, next); + while (lineCounts.size > LINE_COUNT_CACHE_MAX) lineCounts.delete(lineCounts.keys().next().value!); + return complete + Number(trailing); +} + +type RecordGuard = (value: unknown) => value is T; + +function parseRecent(text: string, includeFirst: boolean, limit: number, valid: RecordGuard): T[] { + const lines = text.split("\n"); + if (!includeFirst) lines.shift(); + const out: T[] = []; + for (const raw of lines) { + if (!raw) continue; + try { + const value: unknown = JSON.parse(raw); + if (valid(value)) out.push(value); + } catch { + // A torn line during a write, or a hand-edited record. Keep looking + // farther back until we still have `limit` valid recent entries. + } + } + return out.slice(-limit); +} + +function readRecentLines(file: string, limit: number, valid: RecordGuard): { lines: T[]; total: number } { + let fd: number; + try { + fd = openSync(file, "r"); + } catch { + return { lines: [], total: 0 }; + } + try { + const stat = fstatSync(fd); + const total = countLines(fd, file, stat); + let position = stat.size; + let bytes = Buffer.alloc(0); + let lines: T[] = []; + while (position > 0 && bytes.length < MAX_TAIL_BYTES) { + const remaining = MAX_TAIL_BYTES - bytes.length; + const start = Math.max(0, position - Math.min(READ_CHUNK, remaining)); + const length = position - start; + const chunk = Buffer.allocUnsafe(length); + const read = readSync(fd, chunk, 0, length, start); + if (read <= 0) break; + bytes = Buffer.concat([chunk.subarray(0, read), bytes]); + position = start; + // The first line is partial until we reach byte zero. Parse only once + // enough complete candidates exist; corrupt candidates make us keep + // walking backwards rather than returning fewer valid rows. + const text = bytes.toString("utf8"); + if (position === 0 || text.split("\n").length - 1 >= limit) { + lines = parseRecent(text, position === 0, limit, valid); + if (lines.length >= limit || position === 0) break; + } + } + if (lines.length === 0 && bytes.length > 0) { + lines = parseRecent(bytes.toString("utf8"), position === 0, limit, valid); + } + return { lines, total }; + } finally { + closeSync(fd); + } +} + +const isRecord = (value: unknown): value is Record => typeof value === "object" && value !== null && !Array.isArray(value); +const stringOrMissing = (value: unknown) => value === undefined || typeof value === "string"; +const stringOrNullOrMissing = (value: unknown) => value === undefined || value === null || typeof value === "string"; +const numberOrNullOrMissing = (value: unknown) => value === undefined || value === null || typeof value === "number"; +const stringsOrMissing = (value: unknown) => value === undefined || (Array.isArray(value) && value.every((item) => typeof item === "string")); + +function isRuntimeEvent(value: unknown): value is RuntimeEvent { + if ( + !isRecord(value) || + typeof value.eventId !== "string" || + typeof value.provider !== "string" || + typeof value.threadId !== "string" || + typeof value.createdAt !== "string" || + typeof value.type !== "string" || + !stringOrMissing(value.providerInstanceId) || + !stringOrMissing(value.turnId) || + !stringOrMissing(value.itemId) || + !stringOrMissing(value.requestId) + ) return false; + switch (value.type) { + case "session.started": + return (value.sessionId === null || typeof value.sessionId === "string") && stringOrNullOrMissing(value.model); + case "session.exited": + return stringOrMissing(value.reason); + case "turn.started": + return true; + case "turn.completed": + return ( + typeof value.ok === "boolean" && + stringOrNullOrMissing(value.stopReason) && + numberOrNullOrMissing(value.cost) && + stringsOrMissing(value.denials) && + (value.usage === undefined || + (isRecord(value.usage) && typeof value.usage.input === "number" && typeof value.usage.output === "number")) + ); + case "item.started": + return (value.itemType === "tool" || value.itemType === "reasoning") && stringOrMissing(value.title); + case "item.updated": + return (value.itemType === "tool" || value.itemType === "reasoning") && numberOrNullOrMissing(value.tokens); + case "item.completed": + return value.itemType === "assistant_text" ? typeof value.text === "string" : value.itemType === "tool" && typeof value.ok === "boolean"; + case "content.delta": + return (value.streamKind === "assistant_text" || value.streamKind === "reasoning_text") && typeof value.delta === "string"; + case "request.opened": + return ( + (value.requestType === "permission" || value.requestType === "question") && + typeof value.tool === "string" && + typeof value.summary === "string" && + stringsOrMissing(value.choices) + ); + case "request.resolved": + return ( + (value.behavior === "allow" || value.behavior === "deny" || value.behavior === "answer") && + (value.source === "user" || + value.source === "auto" || + value.source === "timeout" || + value.source === "system" || + value.source === "unavailable" || + value.source === "peer") + ); + case "thread.token-usage.updated": + return typeof value.input === "number" && typeof value.output === "number"; + case "runtime.error": + return typeof value.message === "string" && (value.setup === undefined || typeof value.setup === "boolean"); + default: + return false; + } +} + +function isNativeRecord(value: unknown): value is NativeRecord { + return ( + isRecord(value) && + typeof value.at === "string" && + (value.dir === "in" || value.dir === "out") && + typeof value.source === "string" && + Object.hasOwn(value, "msg") + ); +} + +export function readThreadEvents(input: { + eventsDir: string; + nativeDir: string; + threadId: string; + limit?: number; +}): InspectorPage { + const { eventsDir, nativeDir, threadId } = input; + assertThreadId(threadId); + const requested = input.limit ?? DEFAULT_LIMIT; + const limit = Number.isFinite(requested) ? Math.max(1, Math.min(Math.trunc(requested), MAX_LIMIT)) : DEFAULT_LIMIT; + + const runtime = readRecentLines(join(eventsDir, `${threadId}.ndjson`), limit, isRuntimeEvent); + const native = readRecentLines(join(nativeDir, `${threadId}.ndjson`), limit, isNativeRecord); + + // cap each log on its own, then merge: the native tee is several times + // chattier than the runtime stream, and one shared cap would leave the + // Events lens with a handful of rows behind hundreds of raw ones + const merged: InspectorEntry[] = [ + ...runtime.lines.map((data): InspectorEntry => ({ kind: "runtime", at: data.createdAt, data })), + ...native.lines.map((data): InspectorEntry => ({ kind: "native", at: data.at, data })), + ]; + // stable sort: ties keep file order, which is emit order + merged.sort((a, b) => (a.at < b.at ? -1 : a.at > b.at ? 1 : 0)); + + return { + entries: merged, + total: { runtime: runtime.total, native: native.total }, + }; +} diff --git a/src/App.tsx b/src/App.tsx index c7cc4d3b2..7b959a8aa 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -9,6 +9,7 @@ import { GroupView } from "@/components/GroupView"; import { SettingsPanel } from "@/components/SettingsPanel"; import { PluginsPanel } from "@/components/PluginsPanel"; import { ComputerPanel } from "@/components/ComputerPanel"; +import { InspectorPanel } from "@/components/InspectorPanel"; import { SettingsModal } from "@/components/SettingsModal"; import { UpdateBanner } from "@/components/UpdateBanner"; import { DesktopCapabilitiesProvider } from "@/components/DesktopCapabilities"; @@ -128,6 +129,7 @@ function Shell() { )} {state.settingsOpen && bot && } {state.computerOpen && bot && } + {state.inspectorOpen && bot && } {state.appSettingsOpen && } {state.pluginsOpen && } {/* mounted after the modals: same z-50 tier, so DOM order keeps the diff --git a/src/components/ChatView.tsx b/src/components/ChatView.tsx index 4ed42768a..3870e6b3b 100644 --- a/src/components/ChatView.tsx +++ b/src/components/ChatView.tsx @@ -7,6 +7,7 @@ import { ChevronDown, ChevronLeft, ChevronRight, + Bug, Clock, Copy, Crown, @@ -859,6 +860,18 @@ export function ChatView({ bot }: { bot: Bot }) { > + diff --git a/src/components/InspectorPanel.tsx b/src/components/InspectorPanel.tsx new file mode 100644 index 000000000..48475b946 --- /dev/null +++ b/src/components/InspectorPanel.tsx @@ -0,0 +1,204 @@ +// The raw event inspector: what a thread's turns actually looked like on +// the wire, for the moment a bot misbehaves and the chat view can't say +// why. Two lenses over the same thread: +// +// Events — the harness's normalized RuntimeEvent stream: turns, tool +// items, requests, token usage, errors. Follows live over SSE. +// Raw — the provider's own protocol messages, verbatim (the native +// tee). Read from disk; refreshed when a turn settles. +// +// Nothing here is captured for the panel's sake — both logs already exist +// under ~/.openmausbot (server/harness/bus.ts, server/drivers/native.ts). +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { Bug, ChevronDown, ChevronRight, RefreshCw, X } from "lucide-react"; +import { useStore, type Bot } from "@/state/store"; +import { cn } from "@/lib/cn"; +import { formatTime, toRows, type InspectorEntry, type InspectorPage, type InspectorRow } from "@/lib/inspector"; +import type { RuntimeEvent } from "../../server/contracts.ts"; + +type Lens = "events" | "raw"; + +export function InspectorPanel({ bot }: { bot: Bot }) { + const { dispatch } = useStore(); + const threadId = bot.threadId; + const [lens, setLens] = useState("events"); + const [page, setPage] = useState(null); + const [error, setError] = useState(null); + const [expanded, setExpanded] = useState>(() => new Set()); + const listRef = useRef(null); + const stickToBottom = useRef(true); + const loadAbort = useRef(null); + + const load = useCallback(async () => { + loadAbort.current?.abort(); + const controller = new AbortController(); + loadAbort.current = controller; + try { + const res = await fetch(`/api/threads/${threadId}/events?limit=400`, { signal: controller.signal }); + if (!res.ok) throw new Error(`${res.status}`); + const next = (await res.json()) as InspectorPage; + if (controller.signal.aborted) return; + setPage(next); + setError(null); + } catch (e) { + if (controller.signal.aborted) return; + setError(e instanceof Error ? e.message : String(e)); + } finally { + if (loadAbort.current === controller) loadAbort.current = null; + } + }, [threadId]); + + // history from disk on open / thread change + useEffect(() => { + setPage(null); + setExpanded(new Set()); + stickToBottom.current = true; + void load(); + return () => loadAbort.current?.abort(); + }, [load]); + + // live: append this thread's runtime events as they stream, and re-read + // the disk when a turn settles so the native tee (not on the SSE) catches + // up. Own EventSource on purpose: the store folds runtime events into + // chat state and does not re-emit them. + useEffect(() => { + const es = new EventSource("/api/events?screens=off"); + let settle: ReturnType | null = null; + es.onmessage = (raw) => { + let frame: { kind?: string; event?: RuntimeEvent }; + try { + frame = JSON.parse(raw.data); + } catch { + return; + } + if (frame.kind !== "runtime" || !frame.event || frame.event.threadId !== threadId) return; + const event = frame.event; + setPage((prev) => { + const entry: InspectorEntry = { kind: "runtime", at: event.createdAt, data: event }; + if (!prev) return { entries: [entry], total: { runtime: 1, native: 0 } }; + return { entries: [...prev.entries, entry], total: { ...prev.total, runtime: prev.total.runtime + 1 } }; + }); + if (event.type === "turn.completed" || event.type === "runtime.error") { + if (settle) clearTimeout(settle); + settle = setTimeout(() => void load(), 400); + } + }; + return () => { + es.close(); + if (settle) clearTimeout(settle); + }; + }, [threadId, load]); + + const entries = useMemo( + () => (page ? page.entries.filter((e) => (lens === "raw" ? e.kind === "native" : e.kind === "runtime")) : []), + [page, lens], + ); + const rows = useMemo(() => toRows(entries), [entries]); + + // follow the tail unless the user has scrolled up to read + useEffect(() => { + const el = listRef.current; + if (el && stickToBottom.current) el.scrollTop = el.scrollHeight; + }, [page?.entries.length, rows.length, lens]); + const onScroll = () => { + const el = listRef.current; + if (!el) return; + stickToBottom.current = el.scrollHeight - el.scrollTop - el.clientHeight < 40; + }; + + const toggle = (key: string) => + setExpanded((prev) => { + const next = new Set(prev); + if (next.has(key)) next.delete(key); + else next.add(key); + return next; + }); + + const shown = entries.length; + const total = lens === "raw" ? (page?.total.native ?? 0) : (page?.total.runtime ?? 0); + + return ( + + ); +} + +function Row({ row, open, onToggle }: { row: InspectorRow; open: boolean; onToggle: () => void }) { + return ( +
+ + {open && ( + 
+          {JSON.stringify(row.data, null, 2)}
+
+ )} +
+ ); +} diff --git a/src/lib/inspector.test.ts b/src/lib/inspector.test.ts new file mode 100644 index 000000000..fd69971c5 --- /dev/null +++ b/src/lib/inspector.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from "vitest"; + +import { summarizeNative, summarizeRuntime, toRows, type InspectorEntry } from "./inspector"; + +const base = { eventId: "e", provider: "claudeAgent" as const, threadId: "t", createdAt: "2026-08-17T10:00:00.000Z" }; + +describe("summarizeRuntime", () => { + it("labels turn boundaries and failures by tone", () => { + expect(summarizeRuntime({ ...base, type: "turn.started", turnId: "abcdef12-rest" })).toEqual({ + summary: "turn started · abcdef12", + tone: "boundary", + }); + expect(summarizeRuntime({ ...base, type: "turn.completed", ok: true, stopReason: "end_turn", cost: 0.01234 })).toEqual({ + summary: "turn ok · end_turn · $0.0123", + tone: "boundary", + }); + expect(summarizeRuntime({ ...base, type: "turn.completed", ok: false }).tone).toBe("error"); + expect(summarizeRuntime({ ...base, type: "runtime.error", message: "boom", setup: true }).summary).toBe("setup: boom"); + }); + + it("clips long assistant text to one line", () => { + const text = "line one\nline two ".repeat(30); + const { summary } = summarizeRuntime({ ...base, type: "item.completed", itemType: "assistant_text", text }); + expect(summary.startsWith("assistant: line one line two")).toBe(true); + expect(summary.length).toBeLessThanOrEqual("assistant: ".length + 120); + expect(summary).not.toContain("\n"); + }); +}); + +describe("summarizeNative", () => { + it("names JSON-RPC methods and claude stream-json messages", () => { + expect(summarizeNative({ at: "", dir: "out", source: "acp", msg: { jsonrpc: "2.0", id: 3, method: "session/prompt" } })).toBe( + "session/prompt #3", + ); + expect(summarizeNative({ at: "", dir: "in", source: "claude", msg: { type: "assistant", message: { role: "assistant" } } })).toBe( + "assistant · assistant", + ); + expect(summarizeNative({ at: "", dir: "in", source: "acp", msg: { jsonrpc: "2.0", id: 3, result: {} } })).toBe("result #3"); + expect(summarizeNative({ at: "", dir: "in", source: "agy.stream", msg: { event: "result", result: { status: "SUCCESS" } } })).toBe( + "result · SUCCESS", + ); + expect(summarizeNative({ at: "", dir: "in", source: "agy.stream", msg: { event: "step_update", step: {} } })).toBe("step_update"); + }); +}); + +describe("toRows", () => { + it("folds a run of content.delta on one stream into one row", () => { + const entries: InspectorEntry[] = [ + { kind: "runtime", at: "1", data: { ...base, eventId: "a", type: "turn.started" } }, + { kind: "runtime", at: "2", data: { ...base, eventId: "b", type: "content.delta", streamKind: "assistant_text", delta: "Hel" } }, + { kind: "runtime", at: "3", data: { ...base, eventId: "c", type: "content.delta", streamKind: "assistant_text", delta: "lo" } }, + { kind: "runtime", at: "4", data: { ...base, eventId: "d", type: "content.delta", streamKind: "reasoning_text", delta: "hmm" } }, + { kind: "native", at: "5", data: { at: "5", dir: "in", source: "claude", msg: { type: "result" } } }, + ]; + const rows = toRows(entries); + expect(rows.map((r) => [r.tag, r.count, r.summary])).toEqual([ + ["turn.started", 1, "turn started"], + ["content.delta", 2, "assistant_text: Hello"], + ["content.delta", 1, "reasoning_text: hmm"], + ["← in", 1, "claude · result"], + ]); + }); + + it("builds a bounded folded preview without rejoining the full delta history", () => { + const entries: InspectorEntry[] = Array.from({ length: 500 }, (_, i) => ({ + kind: "runtime" as const, + at: String(i), + data: { ...base, eventId: `d${i}`, type: "content.delta" as const, streamKind: "assistant_text" as const, delta: `word${i} ` }, + })); + const [row] = toRows(entries); + expect(row.count).toBe(500); + expect(row.summary).toMatch(/^assistant_text: word0 word1/); + expect(row.summary.endsWith("…")).toBe(true); + expect(row.summary.length).toBeLessThanOrEqual("assistant_text: ".length + 120); + expect((row.data as unknown[])).toHaveLength(500); + }); +}); diff --git a/src/lib/inspector.ts b/src/lib/inspector.ts new file mode 100644 index 000000000..667701604 --- /dev/null +++ b/src/lib/inspector.ts @@ -0,0 +1,185 @@ +// Turning the inspector's two record shapes into one-line summaries. Pure +// so the panel stays a thin renderer and the labels can be tested. +import type { RuntimeEvent } from "../../server/contracts.ts"; +import type { InspectorEntry, NativeRecord } from "../../server/thread-events.ts"; +export type { InspectorEntry, InspectorPage, NativeRecord } from "../../server/thread-events.ts"; + +interface FoldPreview { + text: string; + pendingSpace: boolean; + overflow: boolean; +} + +/** A row in the panel: adjacent content.delta events fold into one so a + * streamed paragraph is one line, not three hundred. */ +export interface InspectorRow { + key: string; + kind: "runtime" | "native"; + at: string; + /** short badge — the event type, or in/out for native */ + tag: string; + /** what happened, in one line */ + summary: string; + /** visual weight: a turn boundary, a failure, or plain */ + tone: "boundary" | "error" | "plain"; + /** the record(s) behind the row, for the expanded view */ + data: unknown; + /** > 1 when deltas were folded */ + count: number; + /** bounded, incrementally normalized preview for a folded delta run */ + preview?: FoldPreview; +} + +const clip = (s: string, n = 120) => (s.length > n ? `${s.slice(0, n - 1)}…` : s); +const oneLine = (s: string) => s.replace(/\s+/g, " ").trim(); +const FOLD_PREVIEW_CHARS = 119; + +function appendPreview(previous: FoldPreview | undefined, delta: string): FoldPreview { + const next: FoldPreview = previous ? { ...previous } : { text: "", pendingSpace: false, overflow: false }; + if (next.overflow) return next; + for (const char of delta) { + if (/\s/.test(char)) { + if (next.text) next.pendingSpace = true; + continue; + } + if (next.pendingSpace) { + if (next.text.length >= FOLD_PREVIEW_CHARS) { + next.overflow = true; + break; + } + next.text += " "; + next.pendingSpace = false; + } + if (next.text.length >= FOLD_PREVIEW_CHARS) { + next.overflow = true; + break; + } + next.text += char; + } + return next; +} + +const previewText = (preview: FoldPreview) => `${preview.text}${preview.overflow ? "…" : ""}`; + +export function summarizeRuntime(e: RuntimeEvent): { summary: string; tone: InspectorRow["tone"] } { + switch (e.type) { + case "session.started": + return { summary: `session ${e.sessionId ?? "(none)"}${e.model ? ` · ${e.model}` : ""}`, tone: "plain" }; + case "session.exited": + return { summary: `session exited${e.reason ? ` · ${e.reason}` : ""}`, tone: "plain" }; + case "turn.started": + return { summary: `turn started${e.turnId ? ` · ${e.turnId.slice(0, 8)}` : ""}`, tone: "boundary" }; + case "turn.completed": { + const parts = [e.ok ? "turn ok" : "turn failed"]; + if (e.stopReason) parts.push(e.stopReason); + if (typeof e.cost === "number") parts.push(`$${e.cost.toFixed(4)}`); + if (e.denials?.length) parts.push(`${e.denials.length} denied`); + return { summary: parts.join(" · "), tone: e.ok ? "boundary" : "error" }; + } + case "item.started": + return { summary: `${e.itemType}${e.title ? `: ${clip(oneLine(e.title))}` : " started"}`, tone: "plain" }; + case "item.updated": + return { summary: `${e.itemType} updated${typeof e.tokens === "number" ? ` · ${e.tokens} tok` : ""}`, tone: "plain" }; + case "item.completed": + if (e.itemType === "assistant_text") return { summary: `assistant: ${clip(oneLine(e.text))}`, tone: "plain" }; + return { summary: `tool ${e.ok ? "ok" : "failed"}`, tone: e.ok ? "plain" : "error" }; + case "content.delta": + return { summary: `${e.streamKind}: ${clip(oneLine(e.delta))}`, tone: "plain" }; + case "request.opened": + return { summary: `${e.requestType}: ${e.tool} — ${clip(oneLine(e.summary))}`, tone: "plain" }; + case "request.resolved": + return { summary: `resolved ${e.behavior} · ${e.source}`, tone: "plain" }; + case "thread.token-usage.updated": + return { summary: `tokens in ${e.input} · out ${e.output}`, tone: "plain" }; + case "runtime.error": + return { summary: `${e.setup ? "setup: " : ""}${clip(oneLine(e.message))}`, tone: "error" }; + default: + return { summary: (e as { type: string }).type, tone: "plain" }; + } +} + +export function summarizeNative(r: NativeRecord): string { + const msg = r.msg as Record | null; + if (!msg || typeof msg !== "object") return String(r.msg); + const method = typeof msg.method === "string" ? msg.method : undefined; + const type = typeof msg.type === "string" ? msg.type : undefined; + const id = msg.id !== undefined ? ` #${String(msg.id)}` : ""; + if (method) return `${method}${id}`; + // antigravity's stream keys on `event`, with the outcome under result.status + if (typeof msg.event === "string") { + const status = (msg.result as Record | undefined)?.status; + return typeof status === "string" ? `${msg.event} · ${status}` : msg.event; + } + if (type) { + // claude stream-json: surface the role/subtype so a user turn and an + // assistant chunk don't both read as "message" + const inner = msg.message as Record | undefined; + const role = typeof inner?.role === "string" ? ` · ${inner.role}` : ""; + const subtype = typeof msg.subtype === "string" ? ` · ${msg.subtype}` : ""; + return `${type}${subtype}${role}`; + } + if (msg.result !== undefined) return `result${id}`; + if (msg.error !== undefined) return `error${id}`; + return clip(oneLine(JSON.stringify(msg))); +} + +/** Entries → rows, folding runs of content.delta on the same stream. */ +export function toRows(entries: InspectorEntry[]): InspectorRow[] { + const rows: InspectorRow[] = []; + for (const [i, entry] of entries.entries()) { + if (entry.kind === "native") { + rows.push({ + key: `n${i}`, + kind: "native", + at: entry.at, + tag: entry.data.dir === "out" ? "→ out" : "← in", + summary: `${entry.data.source} · ${summarizeNative(entry.data)}`, + tone: "plain", + data: entry.data, + count: 1, + }); + continue; + } + const e = entry.data; + const last = rows.at(-1); + if ( + e.type === "content.delta" && + last?.kind === "runtime" && + last.tag === "content.delta" && + (last.data as RuntimeEvent[])[0]?.type === "content.delta" && + ((last.data as RuntimeEvent[])[0] as { streamKind: string }).streamKind === e.streamKind + ) { + const list = last.data as RuntimeEvent[]; + list.push(e); + last.count = list.length; + last.preview = appendPreview(last.preview, e.delta); + last.summary = `${e.streamKind}: ${previewText(last.preview)}`; + continue; + } + const { summary, tone } = summarizeRuntime(e); + const preview = e.type === "content.delta" ? appendPreview(undefined, e.delta) : undefined; + const streamKind = e.type === "content.delta" ? e.streamKind : undefined; + rows.push({ + key: e.eventId || `r${i}`, + kind: "runtime", + at: entry.at, + tag: e.type, + summary: preview && streamKind ? `${streamKind}: ${previewText(preview)}` : summary, + tone, + data: e.type === "content.delta" ? [e] : e, + count: 1, + preview, + }); + } + return rows; +} + +export function formatTime(iso: string): string { + const d = new Date(iso); + if (Number.isNaN(d.getTime())) return "--:--:--"; + const hh = String(d.getHours()).padStart(2, "0"); + const mm = String(d.getMinutes()).padStart(2, "0"); + const ss = String(d.getSeconds()).padStart(2, "0"); + const ms = String(d.getMilliseconds()).padStart(3, "0"); + return `${hh}:${mm}:${ss}.${ms}`; +} diff --git a/src/state/store.tsx b/src/state/store.tsx index afb3bae44..2d395ac00 100644 --- a/src/state/store.tsx +++ b/src/state/store.tsx @@ -274,6 +274,8 @@ export interface AppState { settingsOpen: boolean; pluginsOpen: boolean; computerOpen: boolean; + /** the per-thread event inspector (runtime stream + native protocol tee) */ + inspectorOpen: boolean; appSettingsOpen: boolean; appSettingsSection: AppSettingsSection; /** latest live frame of a bot's computer, per botId */ @@ -363,6 +365,7 @@ export type Action = | { type: "toggleSettings"; open?: boolean } | { type: "togglePlugins"; open?: boolean } | { type: "toggleComputer"; open?: boolean } + | { type: "toggleInspector"; open?: boolean } | { type: "focusMessage"; threadId: string; messageId: string } | { type: "focusMessageConsumed"; nonce: number } | { type: "toggleAppSettings"; open?: boolean; section?: AppSettingsSection } @@ -434,6 +437,7 @@ function reducer(state: AppState, action: Action): AppState { activeView: "routines", settingsOpen: false, computerOpen: false, + inspectorOpen: false, appSettingsOpen: false, pluginsOpen: false, }; @@ -685,6 +689,7 @@ function reducer(state: AppState, action: Action): AppState { ...state, settingsOpen: open, computerOpen: open ? false : state.computerOpen, + inspectorOpen: open ? false : state.inspectorOpen, appSettingsOpen: open ? false : state.appSettingsOpen, }; } @@ -709,6 +714,17 @@ function reducer(state: AppState, action: Action): AppState { ...state, computerOpen: open, settingsOpen: open ? false : state.settingsOpen, + inspectorOpen: open ? false : state.inspectorOpen, + appSettingsOpen: open ? false : state.appSettingsOpen, + }; + } + case "toggleInspector": { + const open = action.open ?? !state.inspectorOpen; + return { + ...state, + inspectorOpen: open, + settingsOpen: open ? false : state.settingsOpen, + computerOpen: open ? false : state.computerOpen, appSettingsOpen: open ? false : state.appSettingsOpen, }; } @@ -720,6 +736,7 @@ function reducer(state: AppState, action: Action): AppState { appSettingsSection: action.section ?? state.appSettingsSection, settingsOpen: open ? false : state.settingsOpen, computerOpen: open ? false : state.computerOpen, + inspectorOpen: open ? false : state.inspectorOpen, pluginsOpen: open ? false : state.pluginsOpen, }; } @@ -830,6 +847,7 @@ const initialState: AppState = { settingsOpen: false, pluginsOpen: false, computerOpen: false, + inspectorOpen: false, appSettingsOpen: false, appSettingsSection: "general", screens: {},