Add a per-thread raw event inspector

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);

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

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<string, unknown>) => ({ 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/);
+  });
+});