From 74ed10b6e96bf699a56669c8d6c40484f960b970 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=BE=E5=B1=B1=E8=B2=B4=E8=87=B3?= Date: Mon, 1 Jun 2026 18:03:57 +0900 Subject: [PATCH] feat(import): add codex adapter for native rollout logs Add `basou import codex`, the second import adapter, deriving Basou sessions from OpenAI Codex native rollout logs (~/.codex/sessions//rollout-*.jsonl). Each rollout becomes one imported session carrying session_started / session_ended and command_executed from exec_command calls (bash -c ""), with the real exit_code and duration_ms parsed from the paired command output. Codex stores rollouts by date rather than per project, so discovery walks the tree and matches each rollout's recorded working directory against --project (default: the repository root); only sessions started in that project are imported. File changes and decisions are not derived: Codex applies edits inside exec_command and has no structured question / answer record, so neither has a clean signal to map. The claude-code and codex commands now share a vendor-neutral commit loop (importDerivedSessions): discovery builds the candidates, the shared core dedups, validates, force-replaces, and writes. Adds codex-import to the session source.kind enum. Verified on real data: 38 sessions / 1451 events imported from 45 basou-workspace rollouts (7 skipped as no-action), with exit codes and working-directory sanitization intact. --- CHANGELOG.md | 15 + docs/spec/schemas.md | 2 +- packages/cli/src/commands/import.test.ts | 192 +++++++++++- packages/cli/src/commands/import.ts | 285 +++++++++++++++-- packages/core/src/adapters/codex/index.ts | 2 + .../adapters/codex/rollout-importer.test.ts | 217 +++++++++++++ .../src/adapters/codex/rollout-importer.ts | 289 ++++++++++++++++++ packages/core/src/index.ts | 5 + packages/core/src/schemas/session.schema.ts | 3 + 9 files changed, 978 insertions(+), 32 deletions(-) create mode 100644 packages/core/src/adapters/codex/index.ts create mode 100644 packages/core/src/adapters/codex/rollout-importer.test.ts create mode 100644 packages/core/src/adapters/codex/rollout-importer.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a06ca095..fb36ba8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,21 @@ All notable changes to **basou** are recorded here. The project follows - `session.source.external_id` (additive, optional) — records the originating session id in the source tool's namespace (e.g. the Claude Code session UUID), the key that makes re-imports idempotent. +- `basou import codex` — derive Basou sessions from OpenAI Codex native + rollout logs (`~/.codex/sessions//rollout-*.jsonl`) after the fact. + Each rollout becomes one imported session carrying `session_started` / + `session_ended` and `command_executed` (from `exec_command` calls, recorded + as `bash -c ""`) with the real `exit_code` and `duration_ms` parsed + from the paired command output. Because Codex stores rollouts by date rather + than per project, discovery walks the tree and matches each rollout's + recorded working directory against `--project` (default: the current + repository root); only sessions started in that project are imported. + `--all`, `--session `, `--force`, `--dry-run`, and `--json` behave as + for `import claude-code`. File changes and decisions are not derived: Codex + applies edits inside `exec_command` (no dedicated edit tool) and has no + structured question / answer record, so neither has a clean signal to map. +- `codex-import` session `source.kind` (additive enum value) to distinguish + Codex-rollout-derived sessions from Claude imports and live adapter runs. ## 0.3.1 — 2026-05-21 diff --git a/docs/spec/schemas.md b/docs/spec/schemas.md index 83dae3d0..6deed6a7 100644 --- a/docs/spec/schemas.md +++ b/docs/spec/schemas.md @@ -66,7 +66,7 @@ session: workspace_id: "ws_01HXWS..." source: - kind: "claude-code-adapter" # or "human", "import", "terminal" + kind: "claude-code-adapter" # or "claude-code-import", "codex-import", "human", "import", "terminal" version: "0.1.0" started_at: "2026-05-04T09:00:00+09:00" diff --git a/packages/cli/src/commands/import.test.ts b/packages/cli/src/commands/import.test.ts index 188299b0..9b6d3912 100644 --- a/packages/cli/src/commands/import.test.ts +++ b/packages/cli/src/commands/import.test.ts @@ -13,7 +13,7 @@ import { writeYamlFile, } from "@basou/core"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { doRunImportClaudeCode } from "./import.js"; +import { doRunImportClaudeCode, doRunImportCodex } from "./import.js"; const execFileAsync = promisify(execFile); @@ -28,10 +28,12 @@ const FIXED_DATE = new Date("2026-05-09T03:00:00.000Z"); let tmpRepo: string | undefined; let projectsRoot: string | undefined; +let codexRoot: string | undefined; beforeEach(async () => { tmpRepo = await mkdtemp(join(tmpdir(), "basou-import-cli-test-")); projectsRoot = await mkdtemp(join(tmpdir(), "basou-import-projects-")); + codexRoot = await mkdtemp(join(tmpdir(), "basou-import-codex-")); await execFileAsync("git", ["-c", "init.defaultBranch=main", "init"], { cwd: tmpRepo, env: ENV }); await execFileAsync("git", ["config", "user.email", "test@example.com"], { cwd: tmpRepo, @@ -41,11 +43,12 @@ beforeEach(async () => { }); afterEach(async () => { - for (const dir of [tmpRepo, projectsRoot]) { + for (const dir of [tmpRepo, projectsRoot, codexRoot]) { if (dir !== undefined) await rm(dir, { recursive: true, force: true }); } tmpRepo = undefined; projectsRoot = undefined; + codexRoot = undefined; process.exitCode = 0; vi.restoreAllMocks(); }); @@ -55,6 +58,11 @@ function getProjectsRoot(): string { return projectsRoot; } +function getCodexRoot(): string { + if (codexRoot === undefined) throw new Error("codexRoot not initialized"); + return codexRoot; +} + async function setupInitedRepo(): Promise { const repo = await realpath(tmpRepo as string); const paths = await ensureBasouDirectory(repo); @@ -274,3 +282,183 @@ describe("basou import claude-code", () => { expect(await listSessionDirs(repo)).toEqual([firstId]); }); }); + +/** Write a synthetic Codex rollout under a date directory in the sessions root. */ +async function writeRollout( + sessionId: string, + records: Array>, + subdir = join("2026", "05", "10"), +): Promise { + const dir = join(getCodexRoot(), subdir); + await mkdir(dir, { recursive: true }); + const body = records.map((r) => JSON.stringify(r)).join("\n"); + await writeFile(join(dir, `rollout-${sessionId}.jsonl`), body); +} + +function codexActionRollout(cwd: string, sessionId: string): Array> { + return [ + { + type: "session_meta", + timestamp: "2026-05-10T00:00:00.000Z", + payload: { id: sessionId, cwd, timestamp: "2026-05-10T00:00:00.000Z" }, + }, + { + type: "response_item", + timestamp: "2026-05-10T00:00:01.000Z", + payload: { + type: "function_call", + name: "exec_command", + arguments: JSON.stringify({ cmd: "npm test", workdir: cwd }), + call_id: "call_1", + }, + }, + { + type: "response_item", + timestamp: "2026-05-10T00:00:02.000Z", + payload: { + type: "function_call_output", + call_id: "call_1", + output: "Wall time: 0.5000 seconds\nProcess exited with code 0\nOutput:\nok", + }, + }, + ]; +} + +describe("basou import codex", () => { + it("--all imports a rollout whose session cwd matches the project", async () => { + const repo = await setupInitedRepo(); + await writeRollout("codex-1", codexActionRollout(repo, "codex-1")); + + await doRunImportCodex({ all: true }, { cwd: repo, codexSessionsDir: getCodexRoot() }); + + const dirs = await listSessionDirs(repo); + expect(dirs).toHaveLength(1); + + const sessionDir = join(basouPaths(repo).sessions, dirs[0] as string); + const session = SessionSchema.parse(await readYamlFile(join(sessionDir, "session.yaml"))); + expect(session.session.source.kind).toBe("codex-import"); + expect(session.session.source.external_id).toBe("codex-1"); + expect(session.session.status).toBe("imported"); + + const eventsBody = await readFile(join(sessionDir, "events.jsonl"), "utf8"); + const types = eventsBody + .split("\n") + .filter((l) => l.trim().length > 0) + .map((l) => (JSON.parse(l) as { type: string }).type); + expect(types).toContain("command_executed"); + expect(types[0]).toBe("session_started"); + expect(types[types.length - 1]).toBe("session_ended"); + }); + + it("imports only rollouts started in the requested project", async () => { + const repo = await setupInitedRepo(); + await writeRollout("codex-here", codexActionRollout(repo, "codex-here")); + // A rollout from a different project must not be imported. + await writeRollout( + "codex-other", + codexActionRollout("/some/other/project", "codex-other"), + join("2026", "05", "11"), + ); + + await doRunImportCodex({ all: true }, { cwd: repo, codexSessionsDir: getCodexRoot() }); + + const dirs = await listSessionDirs(repo); + expect(dirs).toHaveLength(1); + const session = SessionSchema.parse( + await readYamlFile(join(basouPaths(repo).sessions, dirs[0] as string, "session.yaml")), + ); + expect(session.session.source.external_id).toBe("codex-here"); + }); + + it("--session imports a single rollout by its Codex session id", async () => { + const repo = await setupInitedRepo(); + await writeRollout("codex-1", codexActionRollout(repo, "codex-1")); + await writeRollout("codex-2", codexActionRollout(repo, "codex-2"), join("2026", "05", "11")); + + await doRunImportCodex({ session: "codex-2" }, { cwd: repo, codexSessionsDir: getCodexRoot() }); + + const dirs = await listSessionDirs(repo); + expect(dirs).toHaveLength(1); + const session = SessionSchema.parse( + await readYamlFile(join(basouPaths(repo).sessions, dirs[0] as string, "session.yaml")), + ); + expect(session.session.source.external_id).toBe("codex-2"); + }); + + it("is idempotent: re-import skips an already-imported rollout", async () => { + const repo = await setupInitedRepo(); + await writeRollout("codex-1", codexActionRollout(repo, "codex-1")); + + await doRunImportCodex({ all: true }, { cwd: repo, codexSessionsDir: getCodexRoot() }); + expect(await listSessionDirs(repo)).toHaveLength(1); + + await doRunImportCodex({ all: true }, { cwd: repo, codexSessionsDir: getCodexRoot() }); + expect(await listSessionDirs(repo)).toHaveLength(1); + }); + + it("--force replaces an already-imported rollout instead of skipping", async () => { + const repo = await setupInitedRepo(); + await writeRollout("codex-1", codexActionRollout(repo, "codex-1")); + + await doRunImportCodex({ all: true }, { cwd: repo, codexSessionsDir: getCodexRoot() }); + const firstDirs = await listSessionDirs(repo); + expect(firstDirs).toHaveLength(1); + const firstId = firstDirs[0] as string; + + await doRunImportCodex( + { all: true, force: true }, + { cwd: repo, codexSessionsDir: getCodexRoot() }, + ); + const secondDirs = await listSessionDirs(repo); + expect(secondDirs).toHaveLength(1); + expect(secondDirs[0]).not.toBe(firstId); + }); + + it("--dry-run writes nothing to disk", async () => { + const repo = await setupInitedRepo(); + await writeRollout("codex-1", codexActionRollout(repo, "codex-1")); + + await doRunImportCodex( + { all: true, dryRun: true }, + { cwd: repo, codexSessionsDir: getCodexRoot() }, + ); + + expect(await listSessionDirs(repo)).toHaveLength(0); + }); + + it("skips rollouts with no exec_command", async () => { + const repo = await setupInitedRepo(); + await writeRollout("codex-empty", [ + { + type: "session_meta", + timestamp: "2026-05-10T00:00:00.000Z", + payload: { id: "codex-empty", cwd: repo, timestamp: "2026-05-10T00:00:00.000Z" }, + }, + { + type: "response_item", + timestamp: "2026-05-10T00:00:01.000Z", + payload: { type: "reasoning", summary: [] }, + }, + ]); + + await doRunImportCodex({ all: true }, { cwd: repo, codexSessionsDir: getCodexRoot() }); + expect(await listSessionDirs(repo)).toHaveLength(0); + }); + + it("requires --session or --all", async () => { + const repo = await setupInitedRepo(); + await expect( + doRunImportCodex({}, { cwd: repo, codexSessionsDir: getCodexRoot() }), + ).rejects.toThrow("Specify --session or --all"); + }); + + it("errors when the Codex sessions directory is absent", async () => { + const repo = await setupInitedRepo(); + await expect( + doRunImportCodex( + { all: true }, + { cwd: repo, codexSessionsDir: join(getCodexRoot(), "does-not-exist") }, + ), + ).rejects.toThrow("Codex sessions directory not found"); + }); +}); diff --git a/packages/cli/src/commands/import.ts b/packages/cli/src/commands/import.ts index ab3f3b9e..3fc2cc3b 100644 --- a/packages/cli/src/commands/import.ts +++ b/packages/cli/src/commands/import.ts @@ -1,20 +1,26 @@ +import { createReadStream, type Dirent } from "node:fs"; import { readdir, readFile, rm } from "node:fs/promises"; import { homedir } from "node:os"; import { basename, join } from "node:path"; +import { createInterface } from "node:readline"; import { assertBasouRootSafe, type BasouPaths, basouPaths, type ClaudeTranscriptRecord, + type CodexRolloutRecord, claudeTranscriptToImportPayload, + codexRolloutToImportPayload, enumerateSessionDirs, findErrorCode, type ImportSessionResult, importSessionFromJson, + type Manifest, readManifest, readSessionYaml, resolveRepositoryRoot, type Session, + type SessionImportPayload, SessionImportPayloadSchema, } from "@basou/core"; import type { Command } from "commander"; @@ -23,7 +29,8 @@ import { isVerbose, renderCliError } from "../lib/error-render.js"; const SES_PREFIX = "ses_"; const SHORT_ID_LEN = 6; -export type ImportClaudeCodeOptions = { +/** Options shared by every `basou import ` subcommand. */ +export type ImportOptions = { project?: string; session?: string; all?: boolean; @@ -33,20 +40,39 @@ export type ImportClaudeCodeOptions = { verbose?: boolean; }; +export type ImportClaudeCodeOptions = ImportOptions; +export type ImportCodexOptions = ImportOptions; + export type ImportContext = { /** Defaults to `process.cwd()`. Injectable for tests. */ cwd?: string; /** - * Root that holds per-project transcript directories. Defaults to + * Root that holds per-project Claude transcript directories. Defaults to * `~/.claude/projects`. Injectable for tests so no real home dir is touched. */ claudeProjectsDir?: string; + /** + * Root that holds Codex rollout logs (`///rollout-*.jsonl`). + * Defaults to `~/.codex/sessions`. Injectable for tests. + */ + codexSessionsDir?: string; +}; + +/** + * A single source session ready to be derived and imported. The dedup key + * (`externalId`) is known up front from discovery; `toPayload` reads and + * transforms the source log lazily, so a session that is skipped (already + * imported) is never read. + */ +type ImportCandidate = { + externalId: string; + toPayload: () => Promise; }; /** - * Wire `basou import claude-code` onto `program`. The `import` group mirrors - * the `session` group layout so future adapters (`import codex`, ...) slot in - * without changing the visible surface. + * Wire the `basou import` command group onto `program`. Each adapter + * (`claude-code`, `codex`, ...) is a subcommand sharing the same flags, so + * future adapters slot in without changing the visible surface. */ export function registerImportCommand(program: Command): void { const importCmd = program @@ -72,6 +98,26 @@ export function registerImportCommand(program: Command): void { .action(async (options: ImportClaudeCodeOptions) => { await runImportClaudeCode(options); }); + + importCmd + .command("codex") + .description("Derive Basou sessions from OpenAI Codex native rollout logs (~/.codex/sessions)") + .option( + "--project ", + "Source project path whose rollouts to import (defaults to the current repository root)", + ) + .option("--session ", "Import a single rollout by its Codex session id") + .option("--all", "Import every rollout found for the project") + .option( + "--force", + "Re-import sessions already imported: delete and replace them instead of skipping", + ) + .option("--dry-run", "Validate and preview only; do not write to disk") + .option("--json", "Output the result as JSON") + .option("-v, --verbose", "Show error causes") + .action(async (options: ImportCodexOptions) => { + await runImportCodex(options); + }); } /** @@ -90,28 +136,106 @@ export async function runImportClaudeCode( } } +/** + * Programmatic entry for `basou import codex`. Owns process exit state; + * tests should prefer {@link doRunImportCodex}. + */ +export async function runImportCodex( + options: ImportCodexOptions, + ctx: ImportContext = {}, +): Promise { + try { + await doRunImportCodex(options, ctx); + } catch (error: unknown) { + renderCliError(error, { verbose: isVerbose(options) }); + process.exitCode = 1; + } +} + export async function doRunImportClaudeCode( options: ImportClaudeCodeOptions, ctx: ImportContext, ): Promise { + assertSelector(options); + const { repositoryRoot, paths, manifest } = await resolveImportTarget(ctx); + + const projectPath = options.project ?? repositoryRoot; + const projectsRoot = ctx.claudeProjectsDir ?? join(homedir(), ".claude", "projects"); + const transcriptDir = join(projectsRoot, encodeProjectDir(projectPath)); + + const files = await selectTranscriptFiles(transcriptDir, options); + const candidates: ImportCandidate[] = files.map((file) => { + // The transcript filename is the Claude session id; it is both the dedup + // key and the source external_id. + const externalId = basename(file, ".jsonl"); + return { + externalId, + toPayload: async () => + claudeTranscriptToImportPayload(await readJsonlRecords(file), { + workspaceId: manifest.workspace.id, + externalId, + }), + }; + }); + + await importDerivedSessions(paths, manifest, options, candidates); +} + +export async function doRunImportCodex( + options: ImportCodexOptions, + ctx: ImportContext, +): Promise { + assertSelector(options); + const { repositoryRoot, paths, manifest } = await resolveImportTarget(ctx); + + const projectPath = options.project ?? repositoryRoot; + const sessionsRoot = ctx.codexSessionsDir ?? join(homedir(), ".codex", "sessions"); + + const rollouts = await discoverCodexRollouts(sessionsRoot, projectPath, options); + const candidates: ImportCandidate[] = rollouts.map(({ file, externalId }) => ({ + externalId, + toPayload: async () => + codexRolloutToImportPayload((await readJsonlRecords(file)) as CodexRolloutRecord[], { + workspaceId: manifest.workspace.id, + externalId, + }), + })); + + await importDerivedSessions(paths, manifest, options, candidates); +} + +function assertSelector(options: ImportOptions): void { if (options.session === undefined && options.all !== true) { throw new Error("Specify --session or --all"); } +} +async function resolveImportTarget( + ctx: ImportContext, +): Promise<{ repositoryRoot: string; paths: BasouPaths; manifest: Manifest }> { const cwd = ctx.cwd ?? process.cwd(); const repositoryRoot = await resolveRepositoryRootForImport(cwd); const paths = basouPaths(repositoryRoot); await assertWorkspaceInitialized(paths.root); const manifest = await readManifest(paths); + return { repositoryRoot, paths, manifest }; +} - const projectPath = options.project ?? repositoryRoot; - const projectsRoot = ctx.claudeProjectsDir ?? join(homedir(), ".claude", "projects"); - const transcriptDir = join(projectsRoot, encodeProjectDir(projectPath)); - - const files = await selectTranscriptFiles(transcriptDir, options); +/** + * The vendor-neutral import core: dedup, derive, validate, and write each + * candidate, then report. Every `basou import ` funnels its + * discovered candidates through here, so dedup / `--force` / `--dry-run` + * semantics stay identical across adapters. + */ +async function importDerivedSessions( + paths: BasouPaths, + manifest: Manifest, + options: ImportOptions, + candidates: ReadonlyArray, +): Promise { const existingByExternalId = await loadExistingByExternalId(paths); - // Session ids imported earlier in THIS run, so two transcript files that map - // to one Claude session id never double-import within a single invocation. + // Session ids imported earlier in THIS run, so two source files that map to + // one session id never double-import within a single invocation. const seenThisRun = new Set(); const results: ImportSessionResult[] = []; @@ -119,9 +243,7 @@ export async function doRunImportClaudeCode( let skippedExisting = 0; let replaced = 0; let sanitizedPaths = 0; - for (const file of files) { - // The transcript filename is the Claude session id; dedup on it. - const externalId = basename(file, ".jsonl"); + for (const { externalId, toPayload } of candidates) { if (seenThisRun.has(externalId)) { skippedExisting++; continue; @@ -134,11 +256,7 @@ export async function doRunImportClaudeCode( continue; } - const records = await readTranscript(file); - const payload = claudeTranscriptToImportPayload(records, { - workspaceId: manifest.workspace.id, - externalId, - }); + const payload = await toPayload(); if (payload === null) { skippedNoAction++; continue; @@ -254,18 +372,124 @@ async function selectTranscriptFiles( .map((name) => join(transcriptDir, name)); } -async function readTranscript(file: string): Promise { +/** + * Discover the Codex rollouts that belong to `projectPath`. Codex stores + * rollouts under date directories (not per-project like Claude), so the whole + * tree is walked and each rollout's `session_meta.cwd` is matched against the + * project. The exact-match is also the safety boundary: only sessions started + * in the requested project are ever imported. `--session` narrows to a single + * rollout by its Codex session id within that project. + */ +async function discoverCodexRollouts( + sessionsRoot: string, + projectPath: string, + options: ImportCodexOptions, +): Promise> { + const files = await findRolloutFiles(sessionsRoot); + const matched: Array<{ file: string; externalId: string }> = []; + for (const file of files) { + const meta = await readRolloutMeta(file); + if (meta === undefined) continue; + if (meta.cwd !== projectPath) continue; + if (options.session !== undefined && meta.id !== options.session) continue; + matched.push({ file, externalId: meta.id }); + } + return matched; +} + +/** Recursively collect every `rollout-*.jsonl` under the Codex sessions root. */ +async function findRolloutFiles(sessionsRoot: string): Promise { + const found: string[] = []; + const walk = async (dir: string, isRoot: boolean): Promise => { + let entries: Dirent[]; + try { + entries = await readdir(dir, { withFileTypes: true }); + } catch (error: unknown) { + if (findErrorCode(error, "ENOENT")) { + if (isRoot) { + throw new Error("Codex sessions directory not found", { cause: error }); + } + return; // a subdir vanished mid-walk; ignore + } + throw new Error("Failed to read Codex sessions directory", { cause: error }); + } + for (const entry of entries) { + const full = join(dir, entry.name); + if (entry.isDirectory()) { + await walk(full, false); + } else if ( + entry.isFile() && + entry.name.startsWith("rollout-") && + entry.name.endsWith(".jsonl") + ) { + found.push(full); + } + } + }; + await walk(sessionsRoot, true); + return found.sort(); +} + +/** + * Read just the `session_meta` (first record) of a rollout to learn its + * project cwd and session id, without parsing the whole — usually large — log. + * Returns `undefined` for any file whose first record is not a usable + * `session_meta`, so the caller can skip it. + */ +async function readRolloutMeta(file: string): Promise<{ id: string; cwd: string } | undefined> { + const firstLine = await readFirstLine(file); + if (firstLine === undefined) return undefined; + let parsed: unknown; + try { + parsed = JSON.parse(firstLine); + } catch { + return undefined; + } + if (!isObject(parsed) || parsed.type !== "session_meta") return undefined; + const payload = isObject(parsed.payload) ? parsed.payload : undefined; + if (payload === undefined) return undefined; + const id = payload.id; + const cwd = payload.cwd; + if (typeof id !== "string" || id.length === 0) return undefined; + if (typeof cwd !== "string" || cwd.length === 0) return undefined; + return { id, cwd }; +} + +/** Read the first non-empty line of a file, streaming so large files are cheap. */ +async function readFirstLine(file: string): Promise { + const stream = createReadStream(file, { encoding: "utf8" }); + const rl = createInterface({ input: stream, crlfDelay: Number.POSITIVE_INFINITY }); + try { + for await (const line of rl) { + const trimmed = line.trim(); + if (trimmed.length > 0) return trimmed; + } + return undefined; + } catch { + return undefined; + } finally { + rl.close(); + stream.destroy(); + } +} + +/** + * Read a JSONL native log into an array of records. A malformed line is + * skipped rather than failing the whole file, so partial native logs still + * yield best-effort provenance. + */ +async function readJsonlRecords(file: string): Promise { let body: string; try { body = await readFile(file, "utf8"); } catch (error: unknown) { if (findErrorCode(error, "ENOENT")) { - throw new Error("Transcript not found", { cause: error }); + throw new Error("Source log not found", { cause: error }); } if (findErrorCode(error, "EISDIR")) { - throw new Error("Transcript path is not a file", { cause: error }); + throw new Error("Source log path is not a file", { cause: error }); } - throw new Error("Failed to read transcript", { cause: error }); + throw new Error("Failed to read source log", { cause: error }); } const records: ClaudeTranscriptRecord[] = []; @@ -274,19 +498,22 @@ async function readTranscript(file: string): Promise { if (trimmed.length === 0) continue; try { const parsed: unknown = JSON.parse(trimmed); - if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) { - records.push(parsed as ClaudeTranscriptRecord); + if (isObject(parsed)) { + records.push(parsed); } } catch { - // A malformed line is skipped rather than failing the whole transcript; - // partial native logs should still yield best-effort provenance. + // A malformed line is skipped rather than failing the whole file. } } return records; } +function isObject(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + function printImportResult( - options: ImportClaudeCodeOptions, + options: ImportOptions, results: ImportSessionResult[], counts: { skippedNoAction: number; skippedExisting: number; replaced: number }, ): void { diff --git a/packages/core/src/adapters/codex/index.ts b/packages/core/src/adapters/codex/index.ts new file mode 100644 index 00000000..b28985de --- /dev/null +++ b/packages/core/src/adapters/codex/index.ts @@ -0,0 +1,2 @@ +export type { CodexRolloutRecord, CodexRolloutToPayloadOptions } from "./rollout-importer.js"; +export { CODEX_IMPORT_SOURCE, codexRolloutToImportPayload } from "./rollout-importer.js"; diff --git a/packages/core/src/adapters/codex/rollout-importer.test.ts b/packages/core/src/adapters/codex/rollout-importer.test.ts new file mode 100644 index 00000000..49184fd5 --- /dev/null +++ b/packages/core/src/adapters/codex/rollout-importer.test.ts @@ -0,0 +1,217 @@ +import { describe, expect, it } from "vitest"; +import { SessionImportPayloadSchema } from "../../schemas/session-import.schema.js"; +import { + CODEX_IMPORT_SOURCE, + type CodexRolloutRecord, + codexRolloutToImportPayload, +} from "./rollout-importer.js"; + +const WS_ID = "ws_01HXABCDEF1234567890ABCDEF"; +const CWD = "/Users/x/projects/foo"; +const SESSION_ID = "019df266-a7bf-77f3-bee4-4d7d27c9b847"; + +function sessionMeta(ts: string, cwd = CWD): CodexRolloutRecord { + return { type: "session_meta", timestamp: ts, payload: { id: SESSION_ID, cwd, timestamp: ts } }; +} + +function execCall(ts: string, callId: string, cmd: string, workdir?: string): CodexRolloutRecord { + const args: Record = { cmd, yield_time_ms: 1000 }; + if (workdir !== undefined) args.workdir = workdir; + return { + type: "response_item", + timestamp: ts, + payload: { + type: "function_call", + name: "exec_command", + arguments: JSON.stringify(args), + call_id: callId, + }, + }; +} + +function execOutput(ts: string, callId: string, output: string): CodexRolloutRecord { + return { + type: "response_item", + timestamp: ts, + payload: { type: "function_call_output", call_id: callId, output }, + }; +} + +function transform(records: CodexRolloutRecord[]) { + return codexRolloutToImportPayload(records, { workspaceId: WS_ID }); +} + +describe("codexRolloutToImportPayload", () => { + it("derives session lifecycle + command_executed from exec_command calls", () => { + const records: CodexRolloutRecord[] = [ + sessionMeta("2026-05-10T00:00:00.000Z"), + execCall("2026-05-10T00:00:01.000Z", "call_1", "npm test", `${CWD}/pkg`), + execOutput( + "2026-05-10T00:00:02.000Z", + "call_1", + "Wall time: 1.5000 seconds\nProcess exited with code 0\nOutput:\nok", + ), + // A non-exec function call must be ignored (no clean command signal). + { + type: "response_item", + timestamp: "2026-05-10T00:00:03.000Z", + payload: { type: "function_call", name: "update_plan", arguments: "{}", call_id: "call_2" }, + }, + ]; + + const payload = transform(records); + expect(payload).not.toBeNull(); + if (payload === null) return; + + expect(SessionImportPayloadSchema.safeParse(payload).success).toBe(true); + expect(payload.session.source.kind).toBe(CODEX_IMPORT_SOURCE); + expect(payload.session.started_at).toBe("2026-05-10T00:00:00.000Z"); + // The last record's timestamp ends the session, even when it is not a command. + expect(payload.session.ended_at).toBe("2026-05-10T00:00:03.000Z"); + expect(payload.session.working_directory).toBe(CWD); + expect(payload.session.workspace_id).toBe(WS_ID); + expect(payload.session.source.external_id).toBe(SESSION_ID); + expect(payload.session.invocation.command).toBe("codex"); + // File changes are deferred for Codex, so related_files stays empty. + expect(payload.session.related_files).toEqual([]); + expect(payload.session.label).toBe("codex 2026-05-10: 1 command"); + + expect(payload.events.map((e) => e.type)).toEqual([ + "session_started", + "command_executed", + "session_ended", + ]); + + const command = payload.events[1]; + if (command?.type !== "command_executed") throw new Error("expected command_executed"); + expect(command.command).toBe("bash"); + expect(command.args).toEqual(["-c", "npm test"]); + // workdir from the call arguments wins over the session cwd. + expect(command.cwd).toBe(`${CWD}/pkg`); + expect(command.exit_code).toBe(0); + expect(command.duration_ms).toBe(1500); + }); + + it("parses a negative (signal) exit code and falls back to the session cwd", () => { + const records: CodexRolloutRecord[] = [ + sessionMeta("2026-05-10T00:00:00.000Z"), + // No workdir in arguments -> command_executed.cwd falls back to session cwd. + execCall("2026-05-10T00:00:01.000Z", "call_1", "sleep 100"), + execOutput( + "2026-05-10T00:00:02.000Z", + "call_1", + "Wall time: 0.0000 seconds\nProcess exited with code -1\nOutput:\n", + ), + ]; + const payload = transform(records); + expect(payload).not.toBeNull(); + if (payload === null) return; + const command = payload.events[1]; + if (command?.type !== "command_executed") throw new Error("expected command_executed"); + expect(command.exit_code).toBe(-1); + expect(command.cwd).toBe(CWD); + expect(command.duration_ms).toBe(0); + }); + + it("records a null exit code when the output has no completion line", () => { + const records: CodexRolloutRecord[] = [ + sessionMeta("2026-05-10T00:00:00.000Z"), + // A command that yielded before completing: no paired output at all. + execCall("2026-05-10T00:00:01.000Z", "call_1", "tail -f log"), + ]; + const payload = transform(records); + expect(payload).not.toBeNull(); + if (payload === null) return; + const command = payload.events[1]; + if (command?.type !== "command_executed") throw new Error("expected command_executed"); + expect(command.exit_code).toBeNull(); + expect(command.duration_ms).toBe(0); + }); + + it("orders output even when records are not timestamp-sorted on disk", () => { + const records: CodexRolloutRecord[] = [ + sessionMeta("2026-05-10T00:00:00.000Z"), + execCall("2026-05-10T00:00:05.000Z", "call_2", "second"), + execCall("2026-05-10T00:00:01.000Z", "call_1", "first"), + ]; + const payload = transform(records); + expect(payload).not.toBeNull(); + if (payload === null) return; + expect(SessionImportPayloadSchema.safeParse(payload).success).toBe(true); + expect(payload.session.started_at).toBe("2026-05-10T00:00:00.000Z"); + expect(payload.session.ended_at).toBe("2026-05-10T00:00:05.000Z"); + for (let i = 1; i < payload.events.length; i++) { + const prevEvent = payload.events[i - 1]; + const currEvent = payload.events[i]; + if (prevEvent === undefined || currEvent === undefined) continue; + expect(Date.parse(currEvent.occurred_at)).toBeGreaterThanOrEqual( + Date.parse(prevEvent.occurred_at), + ); + } + const commands = payload.events.filter((e) => e.type === "command_executed"); + expect(commands).toHaveLength(2); + if (commands[0]?.type === "command_executed") { + expect(commands[0].args).toEqual(["-c", "first"]); + } + }); + + it("prefers the provided externalId over the session_meta id", () => { + const records: CodexRolloutRecord[] = [ + sessionMeta("2026-05-10T00:00:00.000Z"), + execCall("2026-05-10T00:00:01.000Z", "call_1", "ls"), + ]; + const payload = codexRolloutToImportPayload(records, { + workspaceId: WS_ID, + externalId: "from-option", + }); + expect(payload).not.toBeNull(); + if (payload === null) return; + expect(payload.session.source.external_id).toBe("from-option"); + expect(payload.session.label).not.toContain("from-option"); + expect(payload.session.label).toMatch(/^codex \d{4}-\d{2}-\d{2}: \d+ command/); + }); + + it("returns null when no exec_command exists", () => { + const records: CodexRolloutRecord[] = [ + sessionMeta("2026-05-10T00:00:00.000Z"), + { + type: "response_item", + timestamp: "2026-05-10T00:00:01.000Z", + payload: { type: "reasoning", summary: [] }, + }, + { + type: "response_item", + timestamp: "2026-05-10T00:00:02.000Z", + payload: { type: "function_call", name: "update_plan", arguments: "{}", call_id: "c" }, + }, + ]; + expect(transform(records)).toBeNull(); + }); + + it("returns null for an empty rollout", () => { + expect(transform([])).toBeNull(); + }); + + it("skips malformed-shaped records without throwing", () => { + const records: CodexRolloutRecord[] = [ + sessionMeta("2026-05-10T00:00:00.000Z"), + { type: "response_item", timestamp: "2026-05-10T00:00:01.000Z", payload: "not-an-object" }, + // exec_command with unparseable arguments is skipped. + { + type: "response_item", + timestamp: "2026-05-10T00:00:02.000Z", + payload: { type: "function_call", name: "exec_command", arguments: "{bad", call_id: "c0" }, + }, + execCall("2026-05-10T00:00:03.000Z", "call_1", "echo hi"), + ]; + const payload = transform(records); + expect(payload).not.toBeNull(); + if (payload === null) return; + expect(payload.events.map((e) => e.type)).toEqual([ + "session_started", + "command_executed", + "session_ended", + ]); + expect(payload.session.working_directory).toBe(CWD); + }); +}); diff --git a/packages/core/src/adapters/codex/rollout-importer.ts b/packages/core/src/adapters/codex/rollout-importer.ts new file mode 100644 index 00000000..aaf3fe90 --- /dev/null +++ b/packages/core/src/adapters/codex/rollout-importer.ts @@ -0,0 +1,289 @@ +import { type PrefixedId, prefixedUlid } from "../../ids/ulid.js"; +import type { Event } from "../../schemas/event.schema.js"; +import type { Manifest } from "../../schemas/manifest.schema.js"; +import type { SessionImportPayload } from "../../schemas/session-import.schema.js"; + +/** + * The `source` string stamped on every event derived from an OpenAI Codex + * native rollout log, and the matching session `source.kind`. + */ +export const CODEX_IMPORT_SOURCE = "codex-import"; + +/** + * One parsed line of a Codex rollout log + * (`~/.codex/sessions///
/rollout-*.jsonl`). Each line is an + * envelope `{ type, timestamp, payload }` where `payload` shape depends on + * `type`. As with the Claude importer the format is the vendor's internal + * log, not Basou's schema, so every field is read defensively — unknown + * record / payload types and missing fields are skipped rather than rejected. + */ +export type CodexRolloutRecord = Record; + +/** Options for {@link codexRolloutToImportPayload}. */ +export type CodexRolloutToPayloadOptions = { + /** Workspace id of the target Basou workspace (from its manifest). */ + workspaceId: Manifest["workspace"]["id"]; + /** + * Codex session id (`session_meta.payload.id`). Stored as + * `session.source.external_id` so re-imports can be deduplicated. Falls back + * to the id read from the rollout's `session_meta` record when omitted. + */ + externalId?: string; +}; + +/** + * Transform a Codex native rollout log into a Basou {@link SessionImportPayload}, + * ready to hand to `importSessionFromJson`. + * + * This is a pure function: no disk or environment access. It DERIVES Basou's + * provenance-level events from the rollout's message-level records: + * + * - `session_started` / `session_ended` from the first / last timestamped record. + * - `command_executed` from each `exec_command` function call, recorded as + * `bash -c ""`. The shell line and working directory come from the + * call's JSON `arguments` (`{ cmd, workdir }`); the exit code and duration + * are parsed from the paired `function_call_output` (matched by `call_id`), + * whose text carries `Process exited with code N` and `Wall time: X seconds`. + * + * Unlike the Claude importer this derives no `file_changed`: Codex has no + * dedicated edit tool and applies edits inside `exec_command` (e.g. + * `apply_patch`), so there is no clean file-change signal to map. Decisions + * and approvals are likewise not derivable — Codex records an `approval_policy` + * (a policy, not a per-action approval) and has no structured question/answer + * record. Both are deferred. + * + * Returns `null` when the rollout has no timestamped records or no observable + * `exec_command` — such sessions carry no provenance worth importing and are + * skipped by the caller. + * + * Event `id` / `session_id` are placeholders; `importSessionFromJson` mints + * fresh ids on the way in. They are valid-by-construction so the payload still + * passes `SessionImportPayloadSchema` validation upstream. + */ +export function codexRolloutToImportPayload( + records: ReadonlyArray, + options: CodexRolloutToPayloadOptions, +): SessionImportPayload | null { + const placeholderSessionId = prefixedUlid("ses"); + // A command's exit code and duration live on its `function_call_output`, + // which arrives after the originating `function_call`; pre-index outputs by + // call_id so commands can be completed in the single forward pass below. + const outputsByCallId = indexOutputs(records); + const derived: Event[] = []; + // Real rollouts are written in arrival order, but track the earliest / + // latest timestamp explicitly (rather than trusting first / last line) and + // order the derived events by occurred_at below, mirroring the Claude path. + let minTs: string | undefined; + let maxTs: string | undefined; + let workingDir: string | undefined; + let codexSessionId: string | undefined; + + for (const record of records) { + const ts = readString(record.timestamp); + if (ts === undefined) continue; + if (minTs === undefined || Date.parse(ts) < Date.parse(minTs)) minTs = ts; + if (maxTs === undefined || Date.parse(ts) > Date.parse(maxTs)) maxTs = ts; + + const payload = isObject(record.payload) ? record.payload : undefined; + if (payload === undefined) continue; + + if (readString(record.type) === "session_meta") { + // The session-level cwd and id are the most reliable working directory + // and dedup key; take the first occurrence and keep it. + if (workingDir === undefined) workingDir = readString(payload.cwd); + if (codexSessionId === undefined) codexSessionId = readString(payload.id); + continue; + } + + if (readString(record.type) !== "response_item") continue; + if (readString(payload.type) !== "function_call") continue; + if (readString(payload.name) !== "exec_command") continue; + + const command = readExecCommand(payload.arguments); + if (command === undefined) continue; + const cwd = command.workdir ?? workingDir ?? "."; + const output = readCallId(payload.call_id, outputsByCallId); + derived.push( + commandExecutedEvent(ts, placeholderSessionId, command.cmd, cwd, { + exitCode: parseExitCode(output), + durationMs: parseWallTimeMs(output), + }), + ); + } + + if (minTs === undefined || maxTs === undefined) return null; + if (derived.length === 0) return null; + + // Order derived events by occurred_at so the assembled stream is + // non-decreasing — importSessionFromJson rejects out-of-order events. + // Array.prototype.sort is stable, so same-timestamp events keep their + // rollout order. + derived.sort((a, b) => Date.parse(a.occurred_at) - Date.parse(b.occurred_at)); + + const events: Event[] = [ + sessionStartedEvent(minTs, placeholderSessionId), + ...derived, + sessionEndedEvent(maxTs, placeholderSessionId), + ]; + + const externalId = options.externalId ?? codexSessionId; + // Human-readable label: when + how much, so the session reads as content in + // `basou session list` / handoff rather than an opaque id. The source id is + // kept structurally in `source.external_id` (not the label), and paths are + // deliberately excluded — the label is NOT path-sanitized downstream, so a + // raw file path here would leak an operator-private prefix. + const commandCount = derived.length; + const date = minTs.slice(0, 10); + const label = `codex ${date}: ${commandCount} ${commandCount === 1 ? "command" : "commands"}`; + + const payload: SessionImportPayload = { + schema_version: "0.1.0", + session: { + label, + workspace_id: options.workspaceId, + source: { + kind: CODEX_IMPORT_SOURCE, + version: "0.1.0", + ...(externalId !== undefined ? { external_id: externalId } : {}), + }, + started_at: minTs, + ended_at: maxTs, + // Validated against the canonical enum here; importSessionFromJson + // overwrites it with the literal "imported" regardless. + status: "imported", + working_directory: workingDir ?? ".", + invocation: { command: "codex", args: [], exit_code: null }, + related_files: [], + summary: null, + }, + events, + }; + return payload; +} + +// --- event builders ------------------------------------------------------- + +function baseEvent( + occurredAt: string, + sessionId: PrefixedId<"ses">, +): { + schema_version: "0.1.0"; + id: PrefixedId<"evt">; + session_id: PrefixedId<"ses">; + occurred_at: string; + source: string; +} { + return { + schema_version: "0.1.0", + id: prefixedUlid("evt"), + session_id: sessionId, + occurred_at: occurredAt, + source: CODEX_IMPORT_SOURCE, + }; +} + +function sessionStartedEvent(occurredAt: string, sessionId: PrefixedId<"ses">): Event { + return { ...baseEvent(occurredAt, sessionId), type: "session_started" }; +} + +function sessionEndedEvent(occurredAt: string, sessionId: PrefixedId<"ses">): Event { + return { ...baseEvent(occurredAt, sessionId), type: "session_ended" }; +} + +function commandExecutedEvent( + occurredAt: string, + sessionId: PrefixedId<"ses">, + command: string, + cwd: string, + outcome: { exitCode: number | null; durationMs: number }, +): Event { + return { + ...baseEvent(occurredAt, sessionId), + type: "command_executed", + command: "bash", + args: ["-c", command], + cwd, + exit_code: outcome.exitCode, + duration_ms: outcome.durationMs, + }; +} + +// --- defensive readers ---------------------------------------------------- + +function readString(value: unknown): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function isObject(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +/** + * Parse an `exec_command` call's JSON `arguments` string into its shell line + * and optional working directory. Returns `undefined` when the arguments are + * not parseable or carry no `cmd`, so the caller can skip the call. + */ +function readExecCommand(value: unknown): { cmd: string; workdir: string | undefined } | undefined { + const raw = readString(value); + if (raw === undefined) return undefined; + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return undefined; + } + if (!isObject(parsed)) return undefined; + const cmd = readString(parsed.cmd); + if (cmd === undefined) return undefined; + return { cmd, workdir: readString(parsed.workdir) }; +} + +function readCallId(value: unknown, outputs: ReadonlyMap): string | undefined { + const callId = readString(value); + return callId !== undefined ? outputs.get(callId) : undefined; +} + +/** + * Codex's `exec_command` output text reports the child's exit code as + * `Process exited with code N` (N may be negative for signal termination). + * Returns `null` when the line is absent — the command may have yielded before + * completing or the session was cut off mid-command. + */ +function parseExitCode(output: string | undefined): number | null { + if (output === undefined) return null; + const match = output.match(/Process exited with code (-?\d+)/); + return match?.[1] !== undefined ? Number.parseInt(match[1], 10) : null; +} + +/** + * Codex's `exec_command` output text reports wall-clock duration as + * `Wall time: X seconds`. Returns `0` (the schema floor) when absent or + * non-finite, matching the Claude importer's missing-duration default. + */ +function parseWallTimeMs(output: string | undefined): number { + if (output === undefined) return 0; + const match = output.match(/Wall time:\s*([\d.]+)\s*seconds/); + if (match?.[1] === undefined) return 0; + const seconds = Number.parseFloat(match[1]); + return Number.isFinite(seconds) ? Math.round(seconds * 1000) : 0; +} + +/** + * Index every `function_call_output`'s text by its `call_id`, so a command's + * exit code and duration can be looked up at the originating `function_call`. + * Only string outputs are kept — image / structured tool results are arrays + * and carry no command outcome. + */ +function indexOutputs(records: ReadonlyArray): Map { + const byId = new Map(); + for (const record of records) { + if (readString(record.type) !== "response_item") continue; + const payload = isObject(record.payload) ? record.payload : undefined; + if (payload === undefined) continue; + if (readString(payload.type) !== "function_call_output") continue; + const callId = readString(payload.call_id); + const output = readString(payload.output); + if (callId !== undefined && output !== undefined) byId.set(callId, output); + } + return byId; +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 5f3aba9f..19c22de0 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -16,6 +16,11 @@ export { resolveClaudeCodeCommand, summarizeAdapterOutput, } from "./adapters/claude-code/index.js"; +export type { + CodexRolloutRecord, + CodexRolloutToPayloadOptions, +} from "./adapters/codex/index.js"; +export { CODEX_IMPORT_SOURCE, codexRolloutToImportPayload } from "./adapters/codex/index.js"; export type { ApprovalLocation, LoadedApproval } from "./approval/index.js"; export { enumerateApprovals, isLazyExpired, loadApproval } from "./approval/index.js"; export type { DecisionsRendererInput, DecisionsRendererResult } from "./decisions/index.js"; diff --git a/packages/core/src/schemas/session.schema.ts b/packages/core/src/schemas/session.schema.ts index 9b43182c..6479c1d2 100644 --- a/packages/core/src/schemas/session.schema.ts +++ b/packages/core/src/schemas/session.schema.ts @@ -27,12 +27,15 @@ export type SessionStatus = z.infer; * - `claude-code-adapter` — a live `basou run claude-code` process wrap. * - `claude-code-import` — derived after the fact from a Claude Code native * transcript (`~/.claude/projects/*.jsonl`) by `basou import claude-code`. + * - `codex-import` — derived after the fact from an OpenAI Codex native + * rollout log (date-partitioned `~/.codex/sessions`) by `basou import codex`. * - `import` — a round-trip of a Basou-format export (`basou session import`). * - `human` / `terminal` — manually-authored / terminal-recorded sessions. */ export const SessionSourceKindSchema = z.enum([ "claude-code-adapter", "claude-code-import", + "codex-import", "human", "import", "terminal",