diff --git a/src/lib/ingest/index.ts b/src/lib/ingest/index.ts index 3dd1854..254fb8d 100644 --- a/src/lib/ingest/index.ts +++ b/src/lib/ingest/index.ts @@ -9,6 +9,7 @@ import { saveParsedSession, saveStagedParsedSession } from "../../db/sessions.js import { getFileState, setFileState, updateIngestionStats } from "../../db/ingestion.js"; import { registerMachine, recomputeMachineCounts } from "../../db/machines.js"; import { getSessionsDir } from "../paths.js"; +import { ingestionStateMtime, type SessionParser } from "./types.js"; // Register the built-in parsers on import. registerParser(new ClaudeParser()); @@ -48,8 +49,8 @@ interface IngestLockInfo { } interface FileSnapshot { - mtime: string; size: number; + stateMtime: string; } function ingestLockPath(): string { @@ -119,17 +120,19 @@ function withIngestLock(fn: () => T): T { } } -function snapshotFile(file: string): FileSnapshot | null { +function snapshotFile(parser: SessionParser, file: string): FileSnapshot | null { try { const st = statSync(file); - return { mtime: st.mtime.toISOString(), size: st.size }; + const mtime = st.mtime.toISOString(); + const auxiliarySignature = parser.auxiliaryIngestionSignature?.(file) ?? null; + return { size: st.size, stateMtime: ingestionStateMtime(mtime, auxiliarySignature) }; } catch { return null; } } function sameSnapshot(a: FileSnapshot, b: FileSnapshot): boolean { - return a.mtime === b.mtime && a.size === b.size; + return a.stateMtime === b.stateMtime && a.size === b.size; } function ingestSourceUnlocked(source: string, opts: IngestOptions = {}): IngestResult { @@ -142,7 +145,7 @@ function ingestSourceUnlocked(source: string, opts: IngestOptions = {}): IngestR for (const file of files) { result.scanned++; - const before = snapshotFile(file); + const before = snapshotFile(parser, file); if (!before) { // File vanished between listing and stat — skip. continue; @@ -150,7 +153,7 @@ function ingestSourceUnlocked(source: string, opts: IngestOptions = {}): IngestR if (!opts.force) { const state = getFileState(source, file); - if (state && state.status === "ok" && state.file_mtime === before.mtime && state.file_size === before.size) { + if (state && state.status === "ok" && state.file_mtime === before.stateMtime && state.file_size === before.size) { result.skipped++; continue; } @@ -158,20 +161,20 @@ function ingestSourceUnlocked(source: string, opts: IngestOptions = {}): IngestR try { const parsed = parser.parseFileResult?.(file, { preferStaging: true }) ?? { sessions: parser.parseFile(file) }; - const after = snapshotFile(file); + const after = snapshotFile(parser, file); try { if (!after) { - setFileState(source, file, before.mtime, before.size, "pending", "file vanished after parsing"); + setFileState(source, file, before.stateMtime, before.size, "pending", "file vanished after parsing"); opts.onProgress?.(`[${source}] deferred ${file}: file vanished after parsing`); continue; } if (parsed.incompleteTrailingRecord) { - setFileState(source, file, after.mtime, after.size, "pending", "incomplete trailing JSONL record"); + setFileState(source, file, after.stateMtime, after.size, "pending", "incomplete trailing JSONL record"); opts.onProgress?.(`[${source}] deferred ${file}: incomplete trailing JSONL record`); continue; } if (!sameSnapshot(before, after)) { - setFileState(source, file, after.mtime, after.size, "pending", "file changed during parsing"); + setFileState(source, file, after.stateMtime, after.size, "pending", "file changed during parsing"); opts.onProgress?.(`[${source}] deferred ${file}: file changed during parsing`); continue; } @@ -189,7 +192,7 @@ function ingestSourceUnlocked(source: string, opts: IngestOptions = {}): IngestR result.sessions++; fileSessions++; } - setFileState(source, file, after.mtime, after.size, "ok"); + setFileState(source, file, after.stateMtime, after.size, "ok"); result.ingested++; opts.onProgress?.(`[${source}] ingested ${file} (${fileSessions} session${fileSessions === 1 ? "" : "s"})`); } finally { @@ -200,7 +203,7 @@ function ingestSourceUnlocked(source: string, opts: IngestOptions = {}): IngestR } catch (err) { const error = err as Error; result.errors++; - setFileState(source, file, before.mtime, before.size, "error", error.message); + setFileState(source, file, before.stateMtime, before.size, "error", error.message); opts.onProgress?.(`[${source}] ERROR ${file}: ${error.message}`); opts.onError?.(error); } diff --git a/src/lib/ingest/openai-rollout.ts b/src/lib/ingest/openai-rollout.ts index b962235..f8a610d 100644 --- a/src/lib/ingest/openai-rollout.ts +++ b/src/lib/ingest/openai-rollout.ts @@ -55,6 +55,16 @@ export class OpenAiRolloutParser implements SessionParser { return out.sort(); } + auxiliaryIngestionSignature(_filePath: string): string | null { + if (!this.sessionIndexPath) return null; + try { + const stat = statSync(this.sessionIndexPath()); + return `${stat.mtimeMs}:${stat.size}`; + } catch { + return null; + } + } + parseFile(filePath: string): ParsedSession[] { return this.parseFileResult(filePath).sessions; } @@ -76,7 +86,9 @@ export class OpenAiRolloutParser implements SessionParser { let title: string | undefined; let parentThreadId: string | undefined; let forkedFromId: string | undefined; + let originator: string | undefined; let threadSource: string | undefined; + let subagentDepth: number | undefined; let agentNickname: string | undefined; let agentRole: string | undefined; let agentPath: string | undefined; @@ -94,14 +106,23 @@ export class OpenAiRolloutParser implements SessionParser { if (typeof payload.model_provider === "string") modelProvider = payload.model_provider; if (typeof payload.parent_thread_id === "string") parentThreadId = payload.parent_thread_id; if (typeof payload.forked_from_id === "string") forkedFromId = payload.forked_from_id; + if (typeof payload.originator === "string") originator = payload.originator; if (typeof payload.thread_source === "string") threadSource = payload.thread_source; if (typeof payload.agent_nickname === "string") agentNickname = payload.agent_nickname; if (typeof payload.agent_role === "string") agentRole = payload.agent_role; if (typeof payload.agent_path === "string") agentPath = payload.agent_path; + const structuredSource = parseSessionSource(payload.source); + parentThreadId = structuredSource.parentThreadId ?? parentThreadId; + threadSource = structuredSource.threadSource ?? threadSource; + subagentDepth = structuredSource.subagentDepth ?? subagentDepth; + agentNickname = structuredSource.agentNickname ?? agentNickname; + agentRole = structuredSource.agentRole ?? agentRole; + agentPath = structuredSource.agentPath ?? agentPath; isSubagent = Boolean(parentThreadId) || threadSource === "subagent" || - isStructuredSubagentSource(payload.source); + threadSource?.startsWith("subagent:") || + structuredSource.isSubagent; const metaTimestamp = typeof payload.timestamp === "string" ? payload.timestamp : ts; if (metaTimestamp && !firstTs) firstTs = metaTimestamp; @@ -224,8 +245,10 @@ export class OpenAiRolloutParser implements SessionParser { ended_at: lastTs ?? null, source_modified_at: mtime, metadata: compactMetadata({ + originator, forked_from_id: forkedFromId, thread_source: threadSource, + subagent_depth: subagentDepth, agent_nickname: agentNickname, agent_role: agentRole, agent_path: agentPath, @@ -279,20 +302,69 @@ export class OpenAiRolloutParser implements SessionParser { } } -function isStructuredSubagentSource(value: unknown): boolean { - if (typeof value === "string") return value.toLowerCase() === "subagent"; - if (!value || typeof value !== "object") return false; - return Object.keys(value as Record).some((key) => - key.replaceAll("_", "").toLowerCase().includes("subagent"), - ); +interface SessionSourceMetadata { + threadSource?: string; + parentThreadId?: string; + subagentDepth?: number; + agentNickname?: string; + agentRole?: string; + agentPath?: string; + isSubagent: boolean; +} + +function parseSessionSource(value: unknown): SessionSourceMetadata { + if (typeof value === "string") { + return { threadSource: value, isSubagent: value.toLowerCase() === "subagent" }; + } + + const source = asRecord(value); + if (!source) return { isSubagent: false }; + const subagent = asRecord(source.subagent); + if (!subagent) { + return { + isSubagent: Object.keys(source).some((key) => + key.replaceAll("_", "").toLowerCase().includes("subagent"), + ), + }; + } + + const threadSpawn = asRecord(subagent.thread_spawn); + return { + threadSource: threadSpawn ? "subagent:thread_spawn" : "subagent", + parentThreadId: stringField(threadSpawn, "parent_thread_id") ?? stringField(subagent, "parent_thread_id"), + subagentDepth: numberField(threadSpawn, "depth") ?? numberField(subagent, "depth"), + agentNickname: stringField(threadSpawn, "agent_nickname") ?? stringField(subagent, "agent_nickname"), + agentRole: stringField(threadSpawn, "agent_role") ?? stringField(subagent, "agent_role"), + agentPath: stringField(threadSpawn, "agent_path") ?? stringField(subagent, "agent_path"), + isSubagent: true, + }; +} + +function asRecord(value: unknown): Record | undefined { + return value != null && typeof value === "object" + ? value as Record + : undefined; +} + +function stringField(value: Record | undefined, key: string): string | undefined { + const field = value?.[key]; + return typeof field === "string" ? field : undefined; +} + +function numberField(value: Record | undefined, key: string): number | undefined { + const field = value?.[key]; + return typeof field === "number" && Number.isFinite(field) ? field : undefined; } function compactMetadata( - values: Record, + values: Record, ): Record { - return Object.fromEntries( - Object.entries(values).filter((entry): entry is [string, string] => Boolean(entry[1])), - ); + const metadata: Record = {}; + for (const [key, value] of Object.entries(values)) { + if (typeof value === "string" && value) metadata[key] = value; + if (typeof value === "number" && Number.isFinite(value)) metadata[key] = value; + } + return metadata; } interface RolloutSink { diff --git a/src/lib/ingest/types.ts b/src/lib/ingest/types.ts index afbeee8..b093467 100644 --- a/src/lib/ingest/types.ts +++ b/src/lib/ingest/types.ts @@ -33,12 +33,21 @@ export interface SessionParser { sessionRoots(): string[]; /** Enumerate absolute paths of session files under the roots. */ listSessionFiles(): string[]; + /** Signature of auxiliary parser input that must invalidate stored file state. */ + auxiliaryIngestionSignature?(filePath: string): string | null; /** Parse a session file into normalized sessions. Most providers yield one per file; some (gemini logs.json) yield many. Returns [] if none. */ parseFile(filePath: string): ParsedSession[]; /** Parse a session file and return parser state useful to safe ingestion. */ parseFileResult?(filePath: string, opts?: ParseFileOptions): ParseFileResult; } +/** Preserve legacy mtime state unless a parser has an auxiliary ingestion input. */ +export function ingestionStateMtime(fileMtime: string, auxiliarySignature: string | null): string { + return auxiliarySignature === null + ? fileMtime + : JSON.stringify([fileMtime, auxiliarySignature]); +} + /** Flatten a Claude/Codex content value (string or array of blocks) into plain text. */ export function flattenContent(content: unknown): string { if (content == null) return ""; diff --git a/src/lib/watch.ts b/src/lib/watch.ts index 8ae91c5..eb612d5 100644 --- a/src/lib/watch.ts +++ b/src/lib/watch.ts @@ -3,6 +3,7 @@ import { dirname, join } from "node:path"; import { listFileStates } from "../db/ingestion.js"; import { getSessionsDbPath } from "./paths.js"; import { listParsers, ingestSource, type IngestResult } from "./ingest/index.js"; +import { ingestionStateMtime } from "./ingest/types.js"; export interface WatchOptions { /** Restrict watching to these provider sources. Defaults to every parser. */ @@ -115,7 +116,13 @@ function sourceLagSeconds(source: string): number | null { try { const stat = statSync(file); const state = fileStates.get(file); - if (state?.status === "ok" && state.file_mtime === stat.mtime.toISOString() && state.file_size === stat.size) continue; + const mtime = stat.mtime.toISOString(); + const auxiliarySignature = parser.auxiliaryIngestionSignature?.(file) ?? null; + if ( + state?.status === "ok" && + state.file_mtime === ingestionStateMtime(mtime, auxiliarySignature) && + state.file_size === stat.size + ) continue; newestPendingMtime = Math.max(newestPendingMtime, stat.mtimeMs); } catch { continue; diff --git a/test/fixtures/codewith/rollout-root.jsonl b/test/fixtures/codewith/rollout-root.jsonl new file mode 100644 index 0000000..cf3a180 --- /dev/null +++ b/test/fixtures/codewith/rollout-root.jsonl @@ -0,0 +1,4 @@ +{"timestamp":"2026-06-01T10:00:00.000Z","type":"session_meta","payload":{"id":"codewith-fixture-root","timestamp":"2026-06-01T10:00:00.000Z","cwd":"/workspace/synthetic-root-project","cli_version":"0.0.0-fixture","model_provider":"openai","originator":"codewith"}} +{"timestamp":"2026-06-01T10:00:01.000Z","type":"turn_context","payload":{"cwd":"/workspace/synthetic-root-project","model":"gpt-fixture-root","model_provider":"openai"}} +{"timestamp":"2026-06-01T10:00:02.000Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"Generic fallback root title"}]}} +{"timestamp":"2026-06-01T10:00:03.000Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Synthetic root response"}]}} diff --git a/test/fixtures/codewith/rollout-subagent.jsonl b/test/fixtures/codewith/rollout-subagent.jsonl new file mode 100644 index 0000000..255370f --- /dev/null +++ b/test/fixtures/codewith/rollout-subagent.jsonl @@ -0,0 +1,4 @@ +{"timestamp":"2026-06-01T10:01:00.000Z","type":"session_meta","payload":{"id":"codewith-fixture-subagent","timestamp":"2026-06-01T10:01:00.000Z","cwd":"/workspace/synthetic-root-project","cli_version":"0.0.0-fixture","model_provider":"openai","originator":"codewith","source":{"subagent":{"thread_spawn":{"parent_thread_id":"codewith-fixture-root","depth":1,"agent_nickname":"fixture-worker","agent_role":"worker","agent_path":"/root/fixture-worker"}}}}} +{"timestamp":"2026-06-01T10:01:01.000Z","type":"turn_context","payload":{"cwd":"/workspace/synthetic-root-project","model":"gpt-fixture-subagent","model_provider":"openai"}} +{"timestamp":"2026-06-01T10:01:02.000Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"Generic fallback subagent title"}]}} +{"timestamp":"2026-06-01T10:01:03.000Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Synthetic subagent response"}]}} diff --git a/test/fixtures/codewith/session_index.jsonl b/test/fixtures/codewith/session_index.jsonl new file mode 100644 index 0000000..c5130dc --- /dev/null +++ b/test/fixtures/codewith/session_index.jsonl @@ -0,0 +1,2 @@ +{"id":"codewith-fixture-root","thread_name":"Indexed Codewith root title","auth_profile":"synthetic-profile-label","credential_metadata":{"kind":"synthetic-marker"}} +{"id":"codewith-fixture-subagent","thread_name":"Indexed Codewith subagent title","auth_profile":"synthetic-profile-label","credential_metadata":{"kind":"synthetic-marker"}} diff --git a/test/ingest.test.ts b/test/ingest.test.ts index 9c33fff..84d9204 100644 --- a/test/ingest.test.ts +++ b/test/ingest.test.ts @@ -74,6 +74,7 @@ const sharedRolloutLines = (cwd: string) => ].join("\n"); const rolloutFixture = (name: string) => join(import.meta.dir, "fixtures", "rollouts", name); +const codewithFixture = (name: string) => join(import.meta.dir, "fixtures", "codewith", name); function storedRolloutSnapshot(sourceId: string): string { const session = getSessionBySource("codex", sourceId); @@ -188,6 +189,72 @@ describe("ingestSource", () => { expect(codewithStats?.session_count).toBe(1); }); + it("round-trips indexed Codewith root and subagent provenance without auth metadata", () => { + const codewithRoot = join(root, "codewith"); + const sessionDir = join(codewithRoot, "sessions", "2026", "06", "01"); + const rootRollout = join(sessionDir, "rollout-2026-06-01T10-00-00-codewith-fixture-root.jsonl"); + const subagentRollout = join(sessionDir, "rollout-2026-06-01T10-01-00-codewith-fixture-subagent.jsonl"); + const sessionIndex = join(codewithRoot, "session_index.jsonl"); + mkdirSync(sessionDir, { recursive: true }); + copyFileSync(codewithFixture("rollout-root.jsonl"), rootRollout); + copyFileSync(codewithFixture("rollout-subagent.jsonl"), subagentRollout); + copyFileSync(codewithFixture("session_index.jsonl"), sessionIndex); + + expect(ingestSource("codewith")).toMatchObject({ + source: "codewith", + scanned: 2, + ingested: 2, + sessions: 2, + errors: 0, + }); + + const rootSession = getSessionBySource("codewith", "codewith-fixture-root"); + const subagentSession = getSessionBySource("codewith", "codewith-fixture-subagent"); + expect(rootSession).toMatchObject({ + source_path: rootRollout, + title: "Indexed Codewith root title", + project_path: "/workspace/synthetic-root-project", + project_name: "synthetic-root-project", + model: "gpt-fixture-root", + model_provider: "openai", + is_subagent: false, + parent_session_id: null, + started_at: "2026-06-01T10:00:00.000Z", + ended_at: "2026-06-01T10:00:03.000Z", + metadata: { originator: "codewith" }, + }); + expect(rootSession?.machine).toBeTruthy(); + expect(subagentSession).toMatchObject({ + source_path: subagentRollout, + title: "Indexed Codewith subagent title", + project_path: "/workspace/synthetic-root-project", + project_name: "synthetic-root-project", + model: "gpt-fixture-subagent", + model_provider: "openai", + is_subagent: true, + parent_session_id: "codewith-fixture-root", + started_at: "2026-06-01T10:01:00.000Z", + ended_at: "2026-06-01T10:01:03.000Z", + metadata: { + originator: "codewith", + thread_source: "subagent:thread_spawn", + subagent_depth: 1, + agent_nickname: "fixture-worker", + agent_role: "worker", + agent_path: "/root/fixture-worker", + }, + }); + expect(subagentSession?.machine).toBe(rootSession?.machine); + + appendFileSync( + sessionIndex, + '\n{"id":"codewith-fixture-root","thread_name":"Updated indexed root title","auth_profile":"synthetic-profile-label"}\n', + ); + const refreshed = ingestSource("codewith"); + expect(refreshed).toMatchObject({ scanned: 2, ingested: 2, skipped: 0, sessions: 2, errors: 0 }); + expect(getSessionBySource("codewith", "codewith-fixture-root")?.title).toBe("Updated indexed root title"); + }); + it("throws for an unknown source", () => { expect(() => ingestSource("nope")).toThrow(/No parser registered/); });