diff --git a/Sensor/README.md b/Sensor/README.md index bd81ecf..aad6194 100644 --- a/Sensor/README.md +++ b/Sensor/README.md @@ -16,7 +16,7 @@ ADR Sensor is a Python library that collects telemetry from AI coding agents to | **Cursor IDE** | `cursor` | SQLite (`state.vscdb`) | macOS, Linux, Windows | | **Cline (Claude Dev)** | `cline` | JSON task files | macOS, Linux, Windows | | **Claude Desktop** | `claude_desktop` | JSONL audit logs | macOS, Windows | -| **OpenAI Codex CLI** | `codex` | JSONL (`~/.codex/sessions/`) | macOS, Linux, Windows | +| **OpenAI Codex CLI** | `codex` | JSONL + SQLite path catalogs | macOS, Linux, Windows | | **Warp Terminal** | `warp` | SQLite (`warp.sqlite`) | macOS, Windows | | **opencode** | `opencode` | SQLite (`opencode.db`) or JSON tree | macOS, Linux | @@ -33,6 +33,23 @@ Both emit `source: "claude_desktop"`. Dispatch sessions get a distinct `session_context`, so detection rules can treat unattended runs differently from interactive ones. Interactive session ids are unchanged. +### OpenAI Codex CLI + +The Codex parser reads JSONL rollout files from `$CODEX_HOME/sessions/`. It also +opens every `$CODEX_HOME/state_*.sqlite` catalog in read-only mode and supplements +filesystem discovery with rollout paths from compatible `threads` tables. A table +must have `id` and `rollout_path` columns; `updated_at` and `updated_at_ms` are +optional. Relative rollout paths are resolved from `CODEX_HOME`, and only existing +regular `.jsonl` files are accepted. Catalog and filesystem paths are deduplicated. + +If `CODEX_HOME` is unset or empty, it defaults to `~/.codex`. Rollouts use a 14-day +lookback by default. Filtering uses the newer of the file modification time and any +valid catalog update timestamp, so a recently updated catalog entry can retain an +older file. Corrupt, locked, or incompatible catalogs are skipped without affecting +files found under `sessions/`; malformed timestamps fall back to file modification +time. Pass `max_age_days` to `CodexParser` or `AgentObserver` to customize the +lookback; values less than or equal to zero disable age filtering for `CodexParser`. + ### opencode [opencode](https://github.com/sst/opencode) uses the XDG layout on every platform, @@ -298,6 +315,7 @@ cannot run on the current platform are skipped rather than failing. | Variable | Read by | Effect | | ----------------- | -------------------------- | ----------------------------------------------------------------- | +| `CODEX_HOME` | Codex parser | Codex data root containing `sessions/` and optional `state_*.sqlite` catalogs (default `~/.codex`) | | `XDG_CACHE_HOME` | `AgentObserver` | Base for `--save-sessions` output (`$XDG_CACHE_HOME/adr_sensor`, default `~/.cache/adr_sensor`) | | `XDG_DATA_HOME` | opencode parser | Overrides the opencode data directory (default `~/.local/share/opencode`) | | `OPENCODE_DB` | opencode parser | Overrides the opencode SQLite filename or path (`:memory:` is ignored) | diff --git a/Sensor/adr_sensor/observer.py b/Sensor/adr_sensor/observer.py index 97e2109..1424d38 100644 --- a/Sensor/adr_sensor/observer.py +++ b/Sensor/adr_sensor/observer.py @@ -71,7 +71,7 @@ def __init__(self, output_dir: Optional[Path] = None, max_age_days: Optional[int self.claude_desktop_parser = ( ClaudeDesktopParser(max_age_days=max_age_days) if max_age_days is not None else ClaudeDesktopParser() ) - self.codex_parser = CodexParser() + self.codex_parser = CodexParser(max_age_days=max_age_days) if max_age_days is not None else CodexParser() self.cline_parser = ClineParser() self.warp_parser = WarpParser(max_age_days=max_age_days) if max_age_days is not None else WarpParser() self.opencode_parser = ( diff --git a/Sensor/adr_sensor/parsers/codex_parser.py b/Sensor/adr_sensor/parsers/codex_parser.py index 4987c95..6ba7a57 100644 --- a/Sensor/adr_sensor/parsers/codex_parser.py +++ b/Sensor/adr_sensor/parsers/codex_parser.py @@ -1,11 +1,17 @@ """ Parser for OpenAI Codex CLI logs. -Reads JSONL files from ~/.codex/sessions/ +Discovers JSONL files from $CODEX_HOME/sessions/ and read-only state catalogs. + +Performance-optimized: Skips rollout files older than 2 weeks by default. """ import json +import math +import os +import sqlite3 +import stat import traceback -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Dict, List, Optional @@ -14,25 +20,47 @@ from ..utils.timestamp_utils import normalize_timestamp from .base_parser import BaseParser +MAX_LOG_AGE_DAYS = 14 + class CodexParser(BaseParser): """Parser for OpenAI Codex CLI JSONL log files.""" - def __init__(self): - self.base_path = Path.home() / ".codex/sessions" + def __init__(self, max_age_days: int = MAX_LOG_AGE_DAYS): + codex_home_env = os.environ.get("CODEX_HOME") + self.codex_home = Path(codex_home_env).expanduser() if codex_home_env else Path.home() / ".codex" + self.base_path = self.codex_home / "sessions" + self.max_age_days = max_age_days def parse_all(self) -> List[AgentEvent]: """Parse all available Codex logs.""" entries = [] - if not self.base_path.exists(): - print(f"[CODEX] No logs found at {self.base_path}") + rollout_candidates = self._discover_rollout_files() + if not rollout_candidates: + print(f"[CODEX] No logs found under {self.codex_home}") return entries - jsonl_files = list(self.base_path.glob("**/*.jsonl")) - print(f"[CODEX] Found {len(jsonl_files)} JSONL files") + print(f"[CODEX] Found {len(rollout_candidates)} JSONL files") + + rollout_files = list(rollout_candidates) + if self.max_age_days > 0: + cutoff_time = datetime.now(timezone.utc) - timedelta(days=self.max_age_days) + rollout_files = [] + skipped_count = 0 + + for jsonl_file, activity_time in rollout_candidates.items(): + if activity_time >= cutoff_time: + rollout_files.append(jsonl_file) + else: + skipped_count += 1 + + if skipped_count > 0: + print(f"[CODEX] Skipped {skipped_count} files older than {self.max_age_days} days") + + print(f"[CODEX] Processing {len(rollout_files)} files") - for jsonl_file in jsonl_files: + for jsonl_file in rollout_files: try: entry = self.parse_jsonl_file(jsonl_file) if entry and entry.has_meaningful_content(): @@ -42,6 +70,134 @@ def parse_all(self) -> List[AgentEvent]: return entries + def _discover_rollout_files(self) -> Dict[Path, datetime]: + """Return valid rollout paths and their latest known activity times.""" + candidates: Dict[Path, datetime] = {} + + try: + for rollout_path in self.base_path.glob("**/*.jsonl"): + self._add_rollout_candidate(candidates, rollout_path) + except OSError as e: + # Keep any files yielded before an inaccessible directory interrupted discovery. + print(f"[CODEX] Error discovering logs under {self.base_path}: {e}") + + try: + for catalog_path in self.codex_home.glob("state_*.sqlite"): + self._add_catalog_rollouts(candidates, catalog_path) + except OSError as e: + print(f"[CODEX] Error discovering state catalogs under {self.codex_home}: {e}") + + return candidates + + @staticmethod + def _add_rollout_candidate( + candidates: Dict[Path, datetime], + rollout_path: Path, + catalog_timestamp: Optional[datetime] = None, + ) -> None: + """Add one existing regular JSONL file, merging duplicate activity times.""" + try: + if rollout_path.suffix != ".jsonl": + return + + resolved_path = rollout_path.resolve() + file_stat = resolved_path.stat() + if not stat.S_ISREG(file_stat.st_mode): + return + file_mtime = datetime.fromtimestamp(file_stat.st_mtime, tz=timezone.utc) + except (OSError, RuntimeError, ValueError, OverflowError): + return + + activity_time = file_mtime + if catalog_timestamp is not None and catalog_timestamp > activity_time: + activity_time = catalog_timestamp + + previous_activity = candidates.get(resolved_path) + if previous_activity is None or activity_time > previous_activity: + candidates[resolved_path] = activity_time + + def _add_catalog_rollouts(self, candidates: Dict[Path, datetime], catalog_path: Path) -> None: + """Read rollout paths from one compatible Codex state catalog.""" + connection = None + try: + if not catalog_path.is_file(): + return + + catalog_uri = f"{catalog_path.resolve().as_uri()}?mode=ro" + connection = sqlite3.connect(catalog_uri, uri=True, timeout=0) + + columns = {str(row[1]).lower() for row in connection.execute("PRAGMA table_info(threads)")} + if not {"id", "rollout_path"}.issubset(columns): + return + + timestamp_columns = [name for name in ("updated_at", "updated_at_ms") if name in columns] + selected_columns = ['"id"', '"rollout_path"', *(f'"{name}"' for name in timestamp_columns)] + query = f'SELECT {", ".join(selected_columns)} FROM "threads"' + + for row in connection.execute(query): + raw_rollout_path = row[1] + if not isinstance(raw_rollout_path, str) or not raw_rollout_path: + continue + + rollout_path = Path(raw_rollout_path) + if not rollout_path.is_absolute(): + rollout_path = self.codex_home / rollout_path + + catalog_timestamp = None + for column_name, value in zip(timestamp_columns, row[2:]): + timestamp = self._parse_catalog_timestamp(value, milliseconds=column_name == "updated_at_ms") + if timestamp is not None and (catalog_timestamp is None or timestamp > catalog_timestamp): + catalog_timestamp = timestamp + + self._add_rollout_candidate(candidates, rollout_path, catalog_timestamp) + except (OSError, sqlite3.Error, ValueError) as e: + print(f"[CODEX] Error reading state catalog {catalog_path}: {e}") + finally: + if connection is not None: + try: + connection.close() + except sqlite3.Error: + pass + + @staticmethod + def _parse_catalog_timestamp(value: Any, milliseconds: bool = False) -> Optional[datetime]: + """Parse a catalog timestamp, returning None when the value is malformed.""" + if value is None or isinstance(value, bool): + return None + + if isinstance(value, str): + raw_value = value.strip() + if not raw_value: + return None + try: + numeric_value = float(raw_value) + except ValueError: + try: + iso_value = raw_value[:-1] + "+00:00" if raw_value.endswith(("Z", "z")) else raw_value + parsed = datetime.fromisoformat(iso_value) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + except (OSError, ValueError, OverflowError): + return None + elif isinstance(value, (int, float)): + try: + numeric_value = float(value) + except (OverflowError, ValueError): + return None + else: + return None + + if not math.isfinite(numeric_value): + return None + if milliseconds or abs(numeric_value) >= 1e12: + numeric_value /= 1000 + + try: + return datetime.fromtimestamp(numeric_value, tz=timezone.utc) + except (OSError, ValueError, OverflowError): + return None + def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: """Parse a single JSONL file.""" try: diff --git a/Sensor/tests/test_observer.py b/Sensor/tests/test_observer.py index 2854b0f..2b4cd1c 100644 --- a/Sensor/tests/test_observer.py +++ b/Sensor/tests/test_observer.py @@ -17,6 +17,11 @@ def test_init_default(self, tmp_path): observer = AgentObserver(output_dir=tmp_path) assert observer.output_dir == tmp_path + def test_init_passes_max_age_days_to_codex(self, tmp_path): + observer = AgentObserver(output_dir=tmp_path, max_age_days=0) + + assert observer.codex_parser.max_age_days == 0 + def test_display_summary_empty(self, tmp_path, capsys): """Test display summary with no data.""" observer = AgentObserver(output_dir=tmp_path) diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index 4112a14..aa3a572 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -157,6 +157,54 @@ def test_parse_no_directory(self): class TestCodexParser: + @staticmethod + def _write_rollout(file_path: Path, session_id: str) -> None: + """Write a minimal synthetic Codex rollout with meaningful content.""" + file_path.parent.mkdir(parents=True, exist_ok=True) + events = [ + { + "type": "session_meta", + "payload": { + "id": session_id, + "timestamp": "2025-01-15T10:00:00Z", + "cwd": "/tmp/example-project", + }, + }, + { + "type": "response_item", + "payload": { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "Review this example"}], + }, + }, + ] + file_path.write_text("".join(f"{json.dumps(event)}\n" for event in events), encoding="utf-8") + + @staticmethod + def _write_catalog(catalog_path: Path, rows: list, timestamp_columns: tuple = ()) -> None: + """Write a minimal Codex-shaped state catalog.""" + catalog_path.parent.mkdir(parents=True, exist_ok=True) + columns = ["id TEXT", "rollout_path TEXT", *(f"{name}" for name in timestamp_columns)] + connection = sqlite3.connect(str(catalog_path)) + try: + connection.execute(f"CREATE TABLE threads ({', '.join(columns)})") + placeholders = ", ".join("?" for _ in columns) + for row in rows: + values = [row.get("id"), row.get("rollout_path")] + values.extend(row.get(name) for name in timestamp_columns) + connection.execute(f"INSERT INTO threads VALUES ({placeholders})", values) + connection.commit() + finally: + connection.close() + + @staticmethod + def _parser_for_home(codex_home: Path, max_age_days: int = 14) -> CodexParser: + parser = CodexParser(max_age_days=max_age_days) + parser.codex_home = codex_home + parser.base_path = codex_home / "sessions" + return parser + def test_parse_jsonl_file(self, tmp_path): """Test parsing a Codex CLI JSONL file.""" jsonl_file = tmp_path / "rollout-001.jsonl" @@ -427,13 +475,315 @@ def test_event_msg_records_are_ignored(self, tmp_path): assert len(entry.chat_history) == 2 # user message + assistant tool turn assert len([t for m in entry.chat_history for t in m.tools]) == 1 - def test_parse_no_directory(self): + def test_parse_no_directory(self, tmp_path): """Test parse_all when directory doesn't exist.""" parser = CodexParser() - parser.base_path = Path("/nonexistent/path") + parser.codex_home = tmp_path / "missing-codex-home" + parser.base_path = parser.codex_home / "sessions" entries = parser.parse_all() assert entries == [] + def test_default_max_age_days(self): + assert CodexParser().max_age_days == 14 + + def test_uses_codex_home(self, tmp_path): + codex_home = tmp_path / "custom-codex-home" + + with patch.dict(os.environ, {"CODEX_HOME": str(codex_home)}): + parser = CodexParser() + + assert parser.codex_home == codex_home + assert parser.base_path == codex_home / "sessions" + + def test_defaults_to_dot_codex_under_home(self, tmp_path): + with patch.dict(os.environ, {"CODEX_HOME": ""}): + with patch("adr_sensor.parsers.codex_parser.Path.home", return_value=tmp_path): + parser = CodexParser() + + assert parser.codex_home == tmp_path / ".codex" + assert parser.base_path == tmp_path / ".codex/sessions" + + def test_parse_all_filters_old_rollouts_by_mtime(self, tmp_path): + recent_rollout = tmp_path / "2025/01/15/rollout-recent.jsonl" + old_rollout = tmp_path / "2024/01/15/rollout-old.jsonl" + self._write_rollout(recent_rollout, "recent") + self._write_rollout(old_rollout, "old") + + now = datetime.now(timezone.utc).timestamp() + os.utime(recent_rollout, (now, now)) + old_mtime = now - timedelta(days=30).total_seconds() + os.utime(old_rollout, (old_mtime, old_mtime)) + + parser = CodexParser(max_age_days=14) + parser.codex_home = tmp_path + parser.base_path = tmp_path + + entries = parser.parse_all() + + assert [entry.session_id for entry in entries] == ["codex_recent"] + + @pytest.mark.parametrize("max_age_days", [0, -1]) + def test_non_positive_max_age_days_includes_all_history(self, tmp_path, max_age_days): + rollout = tmp_path / "rollout-history.jsonl" + self._write_rollout(rollout, "history") + old_mtime = datetime.now(timezone.utc).timestamp() - timedelta(days=365).total_seconds() + os.utime(rollout, (old_mtime, old_mtime)) + + parser = CodexParser(max_age_days=max_age_days) + parser.codex_home = tmp_path + parser.base_path = tmp_path + + assert [entry.session_id for entry in parser.parse_all()] == ["codex_history"] + + def test_stat_failure_skips_only_inaccessible_rollout(self, tmp_path): + inaccessible_rollout = tmp_path / "rollout-inaccessible.jsonl" + readable_rollout = tmp_path / "rollout-readable.jsonl" + self._write_rollout(inaccessible_rollout, "inaccessible") + self._write_rollout(readable_rollout, "readable") + + parser = CodexParser() + parser.codex_home = tmp_path + parser.base_path = tmp_path + original_stat = Path.stat + + def selective_stat(path, *args, **kwargs): + if path == inaccessible_rollout: + raise PermissionError("metadata unavailable") + return original_stat(path, *args, **kwargs) + + with patch.object(Path, "stat", selective_stat): + entries = parser.parse_all() + + assert [entry.session_id for entry in entries] == ["codex_readable"] + + def test_glob_failure_keeps_already_discovered_rollouts(self, tmp_path): + rollout = tmp_path / "rollout-discovered.jsonl" + self._write_rollout(rollout, "discovered") + + parser = CodexParser() + parser.codex_home = tmp_path + parser.base_path = tmp_path + + def interrupted_glob(path, pattern): + if pattern == "state_*.sqlite": + assert path == tmp_path + return + assert path == tmp_path + assert pattern == "**/*.jsonl" + yield rollout + raise OSError("directory scan interrupted") + + with patch.object(Path, "glob", interrupted_glob): + entries = parser.parse_all() + + assert [entry.session_id for entry in entries] == ["codex_discovered"] + + def test_discovers_every_compatible_catalog_without_modifying_them(self, tmp_path): + codex_home = tmp_path / "codex home #1" + relative_rollout = codex_home / "archived" / "relative.jsonl" + absolute_rollout = tmp_path / "external" / "absolute.jsonl" + ignored_rollout = codex_home / "archived" / "ignored.jsonl" + self._write_rollout(relative_rollout, "relative") + self._write_rollout(absolute_rollout, "absolute") + self._write_rollout(ignored_rollout, "ignored") + + first_catalog = codex_home / "state_1.sqlite" + second_catalog = codex_home / "state_2.sqlite" + ignored_catalog = codex_home / "other.sqlite" + self._write_catalog( + first_catalog, + [{"id": "thread-relative", "rollout_path": str(relative_rollout.relative_to(codex_home))}], + ) + self._write_catalog( + second_catalog, + [{"id": "thread-absolute", "rollout_path": str(absolute_rollout)}], + ) + self._write_catalog( + ignored_catalog, + [{"id": "thread-ignored", "rollout_path": str(ignored_rollout)}], + ) + catalog_snapshots = { + path: (path.read_bytes(), path.stat().st_mtime_ns) for path in (first_catalog, second_catalog) + } + + entries = self._parser_for_home(codex_home, max_age_days=0).parse_all() + + assert {entry.session_id for entry in entries} == {"codex_relative", "codex_absolute"} + for path, (content, mtime_ns) in catalog_snapshots.items(): + assert path.read_bytes() == content + assert path.stat().st_mtime_ns == mtime_ns + + @pytest.mark.parametrize(("timestamp_column", "multiplier"), [("updated_at", 1), ("updated_at_ms", 1000)]) + def test_catalog_timestamp_can_keep_old_rollout_in_lookback(self, tmp_path, timestamp_column, multiplier): + codex_home = tmp_path / "codex-home" + rollout = codex_home / "archived" / f"{timestamp_column}.jsonl" + self._write_rollout(rollout, timestamp_column) + + now = datetime.now(timezone.utc).timestamp() + old_timestamp = now - timedelta(days=30).total_seconds() + os.utime(rollout, (old_timestamp, old_timestamp)) + self._write_catalog( + codex_home / f"state_{timestamp_column}.sqlite", + [ + { + "id": f"thread-{timestamp_column}", + "rollout_path": str(rollout.relative_to(codex_home)), + timestamp_column: int(now * multiplier), + } + ], + timestamp_columns=(timestamp_column,), + ) + + entries = self._parser_for_home(codex_home).parse_all() + + assert [entry.session_id for entry in entries] == [f"codex_{timestamp_column}"] + + def test_lookback_uses_newest_timestamp_and_deduplicates_paths(self, tmp_path): + codex_home = tmp_path / "codex-home" + catalog_recent = codex_home / "sessions" / "catalog-recent.jsonl" + file_recent = codex_home / "sessions" / "file-recent.jsonl" + both_old = codex_home / "sessions" / "both-old.jsonl" + malformed_recent = codex_home / "sessions" / "malformed-recent.jsonl" + malformed_old = codex_home / "archived" / "malformed-old.jsonl" + + for path, session_id in ( + (catalog_recent, "catalog-recent"), + (file_recent, "file-recent"), + (both_old, "both-old"), + (malformed_recent, "malformed-recent"), + (malformed_old, "malformed-old"), + ): + self._write_rollout(path, session_id) + + now = datetime.now(timezone.utc).timestamp() + old_timestamp = now - timedelta(days=30).total_seconds() + for path in (catalog_recent, both_old, malformed_old): + os.utime(path, (old_timestamp, old_timestamp)) + + self._write_catalog( + codex_home / "state_seconds.sqlite", + [ + { + "id": "catalog-recent-old-copy", + "rollout_path": str(catalog_recent.relative_to(codex_home)), + "updated_at": int(old_timestamp), + }, + { + "id": "file-recent", + "rollout_path": str(file_recent.relative_to(codex_home)), + "updated_at": int(old_timestamp), + }, + { + "id": "both-old", + "rollout_path": str(both_old.relative_to(codex_home)), + "updated_at": int(old_timestamp), + }, + { + "id": "malformed-recent", + "rollout_path": str(malformed_recent.relative_to(codex_home)), + "updated_at": "not-a-timestamp", + }, + { + "id": "malformed-old", + "rollout_path": str(malformed_old.relative_to(codex_home)), + "updated_at": "not-a-timestamp", + }, + ], + timestamp_columns=("updated_at",), + ) + self._write_catalog( + codex_home / "state_milliseconds.sqlite", + [ + { + "id": "catalog-recent-new-copy", + "rollout_path": str(catalog_recent.resolve()), + "updated_at_ms": int(now * 1000), + } + ], + timestamp_columns=("updated_at_ms",), + ) + + entries = self._parser_for_home(codex_home).parse_all() + + assert {entry.session_id for entry in entries} == { + "codex_catalog-recent", + "codex_file-recent", + "codex_malformed-recent", + } + assert len(entries) == 3 + + def test_catalog_rejects_incompatible_schemas_and_non_jsonl_files(self, tmp_path): + codex_home = tmp_path / "codex-home" + valid_rollout = codex_home / "archived" / "valid.jsonl" + wrong_extension = codex_home / "archived" / "wrong.txt" + directory_path = codex_home / "archived" / "directory.jsonl" + self._write_rollout(valid_rollout, "valid") + self._write_rollout(wrong_extension, "wrong-extension") + directory_path.mkdir(parents=True) + + self._write_catalog( + codex_home / "state_valid.sqlite", + [ + {"id": "valid", "rollout_path": str(valid_rollout)}, + {"id": "wrong-extension", "rollout_path": str(wrong_extension)}, + {"id": "directory", "rollout_path": str(directory_path)}, + {"id": "missing", "rollout_path": str(codex_home / "missing.jsonl")}, + {"id": "empty", "rollout_path": ""}, + {"id": "wrong-type", "rollout_path": 42}, + ], + ) + + incompatible_schemas = { + "state_missing_id.sqlite": "CREATE TABLE threads (rollout_path TEXT)", + "state_missing_path.sqlite": "CREATE TABLE threads (id TEXT)", + "state_no_threads.sqlite": "CREATE TABLE metadata (id TEXT, rollout_path TEXT)", + } + for name, statement in incompatible_schemas.items(): + connection = sqlite3.connect(str(codex_home / name)) + try: + connection.execute(statement) + connection.commit() + finally: + connection.close() + + entries = self._parser_for_home(codex_home, max_age_days=0).parse_all() + + assert [entry.session_id for entry in entries] == ["codex_valid"] + + def test_corrupt_and_locked_catalogs_do_not_hide_other_candidates(self, tmp_path): + codex_home = tmp_path / "codex-home" + filesystem_rollout = codex_home / "sessions" / "filesystem.jsonl" + valid_catalog_rollout = codex_home / "archived" / "valid-catalog.jsonl" + locked_catalog_rollout = codex_home / "archived" / "locked-catalog.jsonl" + self._write_rollout(filesystem_rollout, "filesystem") + self._write_rollout(valid_catalog_rollout, "valid-catalog") + self._write_rollout(locked_catalog_rollout, "locked-catalog") + + corrupt_catalog = codex_home / "state_corrupt.sqlite" + corrupt_catalog.write_bytes(b"not a sqlite database") + corrupt_snapshot = (corrupt_catalog.read_bytes(), corrupt_catalog.stat().st_mtime_ns) + self._write_catalog( + codex_home / "state_valid.sqlite", + [{"id": "valid", "rollout_path": str(valid_catalog_rollout)}], + ) + locked_catalog = codex_home / "state_locked.sqlite" + self._write_catalog( + locked_catalog, + [{"id": "locked", "rollout_path": str(locked_catalog_rollout)}], + ) + + locking_connection = sqlite3.connect(str(locked_catalog), timeout=0) + locking_connection.execute("BEGIN EXCLUSIVE") + try: + entries = self._parser_for_home(codex_home, max_age_days=0).parse_all() + finally: + locking_connection.rollback() + locking_connection.close() + + assert {entry.session_id for entry in entries} == {"codex_filesystem", "codex_valid-catalog"} + assert corrupt_catalog.read_bytes() == corrupt_snapshot[0] + assert corrupt_catalog.stat().st_mtime_ns == corrupt_snapshot[1] + def _build_warp_db(db_path: Path, conversations: list) -> None: """Create a synthetic warp.sqlite matching the schema WarpParser queries.