Repository navigation
Add a per-thread raw event inspector #182
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit
Hold shift + click to select a range
9c15395
Add a per-thread raw event inspector
aivsomkar 4bb775f
Merge remote-tracking branch 'origin/main' into feat/harness-1.2-even…
aivsomkar 9b49c96
Merge remote-tracking branch 'origin/main' into codex/pr-182-review
milind-soni 759e1e6
Validate inspector log records at the wire boundary
milind-soni 2b03415
Merge remote-tracking branch 'origin/main' into codex/pr-182-review
milind-soni File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| import { 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"; | ||
|
|
||
| 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({ eventId: "e1", type: "turn.started", threadId: "t1", createdAt: "2026-08-17T10:00:00.000Z" }) + | ||
| line({ eventId: "e2", type: "turn.completed", threadId: "t1", 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({ eventId: `e${i}`, type: "content.delta", threadId: "t1", createdAt: `2026-08-17T10:00:${String(i).padStart(2, "0")}.000Z` }); | ||
| } | ||
| 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({ eventId: "e1", type: "turn.started", threadId: "t1", createdAt: "2026-08-17T10:00:00.000Z" }) + | ||
| "{not json\n" + | ||
| line({ eventId: "e2", type: "turn.completed", threadId: "t1", createdAt: "2026-08-17T10:00:02.000Z", ok: true }), | ||
| ); | ||
| const page = readThreadEvents({ eventsDir, nativeDir, threadId: "t1" }); | ||
| expect(page.entries).toHaveLength(2); | ||
| expect(page.total.runtime).toBe(2); | ||
| }); | ||
|
|
||
| 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/); | ||
| }); | ||
| }); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,89 @@ | ||
| // 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/<threadId>.ndjson — the normalized RuntimeEvent stream the bus | ||
| // publishes (server/harness/bus.ts) | ||
| // native/<threadId>.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 { readFileSync } from "node:fs"; | ||
| import { join } from "node:path"; | ||
|
|
||
| export type InspectorEntry = | ||
| | { kind: "runtime"; at: string; data: unknown } | ||
| | { kind: "native"; at: string; data: unknown }; | ||
|
|
||
| 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; | ||
|
|
||
| /** 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 readLines(file: string): unknown[] { | ||
| let text: string; | ||
| try { | ||
| text = readFileSync(file, "utf8"); | ||
| } catch { | ||
| return []; | ||
| } | ||
| const out: unknown[] = []; | ||
| for (const raw of text.split("\n")) { | ||
| if (!raw) continue; | ||
| try { | ||
| out.push(JSON.parse(raw)); | ||
| } catch { | ||
| // a torn last line during a write, or a hand-edited file — the rest | ||
| // of the log is still worth showing | ||
| } | ||
| } | ||
| return out; | ||
| } | ||
|
|
||
| const timeOf = (value: unknown, key: "createdAt" | "at"): string => { | ||
| const record = value as Record<string, unknown> | null; | ||
| const at = record?.[key]; | ||
| return typeof at === "string" ? at : ""; | ||
| }; | ||
|
|
||
| export function readThreadEvents(input: { | ||
| eventsDir: string; | ||
| nativeDir: string; | ||
| threadId: string; | ||
| limit?: number; | ||
| }): InspectorPage { | ||
| const { eventsDir, nativeDir, threadId } = input; | ||
| assertThreadId(threadId); | ||
| const limit = Math.max(1, Math.min(input.limit ?? DEFAULT_LIMIT, MAX_LIMIT)); | ||
|
|
||
| const runtime = readLines(join(eventsDir, `${threadId}.ndjson`)); | ||
| const native = readLines(join(nativeDir, `${threadId}.ndjson`)); | ||
|
|
||
| // 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.slice(-limit).map((data): InspectorEntry => ({ kind: "runtime", at: timeOf(data, "createdAt"), data })), | ||
| ...native.slice(-limit).map((data): InspectorEntry => ({ kind: "native", at: timeOf(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.length, native: native.length }, | ||
| }; | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.