| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
| from datetime import UTC, datetime |
| import os |
| from pathlib import Path |
| import shutil |
| import sqlite3 |
|
|
| from backend.codex_adapter import find_newest_state_db, load_codex_sessions |
|
|
|
|
| FIXTURES_DIR = Path(__file__).parent / "fixtures" / "codex" |
| FIXED_NOW = datetime(2026, 4, 24, 12, 0, tzinfo=UTC) |
|
|
|
|
| @dataclass(slots=True) |
| class FakeProcessRecord: |
| pid: int |
| ppid: int |
| tty: str | None |
| elapsed_seconds: int |
| cwd: str | None |
| terminal_program: str | None |
| harness_hint: str | None |
| command: str | None = None |
|
|
|
|
| def test_find_newest_state_db_uses_latest_mtime(tmp_path): |
| codex_home = tmp_path / ".codex" |
| codex_home.mkdir() |
|
|
| older_db = codex_home / "state_11.sqlite" |
| newer_db = codex_home / "state_2.sqlite" |
| _create_state_db(older_db) |
| _create_state_db(newer_db) |
|
|
| older_db_mtime = datetime(2026, 4, 24, 10, 0, tzinfo=UTC).timestamp() |
| newer_db_mtime = datetime(2026, 4, 24, 11, 0, tzinfo=UTC).timestamp() |
|
|
| os.utime(older_db, (older_db_mtime, older_db_mtime)) |
| os.utime(newer_db, (newer_db_mtime, newer_db_mtime)) |
|
|
| assert find_newest_state_db(codex_home) == newer_db |
|
|
|
|
| def test_load_codex_sessions_reads_threads_children_and_statuses(tmp_path): |
| codex_home = tmp_path / ".codex" |
| codex_home.mkdir() |
| state_db = codex_home / "state_5.sqlite" |
| _create_state_db(state_db) |
|
|
| connection = sqlite3.connect(state_db) |
| try: |
| connection.executemany( |
| """ |
| INSERT INTO threads ( |
| id, |
| cwd, |
| title, |
| first_user_message, |
| created_at, |
| updated_at, |
| created_at_ms, |
| updated_at_ms, |
| archived |
| ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) |
| """, |
| [ |
| ( |
| "thread-live", |
| "/workspace/demo", |
| "Implement dashboard adapter", |
| "ignored", |
| 1_777_009_000, |
| 1_777_010_500, |
| None, |
| 1_777_010_500_000, |
| 0, |
| ), |
| ( |
| "thread-old", |
| "/workspace/legacy", |
| "", |
| "Clean up old experiment logs for the dashboard", |
| 1_776_900_000, |
| 1_776_900_100, |
| None, |
| 1_776_900_100_000, |
| 0, |
| ), |
| ], |
| ) |
| connection.executemany( |
| "INSERT INTO thread_spawn_edges (parent_thread_id, child_thread_id, status) VALUES (?, ?, ?)", |
| [ |
| ("thread-live", "child-a", "open"), |
| ("thread-live", "child-b", "open"), |
| ("unrelated", "child-c", "closed"), |
| ], |
| ) |
| connection.commit() |
| finally: |
| connection.close() |
|
|
| process_records = [ |
| FakeProcessRecord( |
| pid=101, |
| ppid=1, |
| tty="pts/5", |
| elapsed_seconds=15, |
| cwd="/workspace/demo", |
| terminal_program="kitty", |
| harness_hint="codex", |
| command="codex", |
| ) |
| ] |
|
|
| rows = load_codex_sessions( |
| process_records=process_records, |
| codex_home=codex_home, |
| now=FIXED_NOW, |
| ) |
|
|
| assert [row.id for row in rows] == ["thread-live", "thread-old"] |
|
|
| live_row = rows[0] |
| assert live_row.source_kind == "codex_db" |
| assert live_row.title == "Implement dashboard adapter" |
| assert live_row.project_name == "demo" |
| assert live_row.children_count == 2 |
| assert live_row.resume_command == "codex resume thread-live" |
| assert live_row.status == "working" |
| assert live_row.status_reason == "Matched live codex process by cwd." |
| assert live_row.confidence == 0.98 |
| assert live_row.source_refs == ["codex_db:thread-live"] |
|
|
| old_row = rows[1] |
| assert old_row.title == "Clean up old experiment logs for the dashboard" |
| assert old_row.project_name == "legacy" |
| assert old_row.children_count == 0 |
| assert old_row.status == "stale" |
| assert "stale" in old_row.status_reason |
| assert old_row.resume_command == "codex resume thread-old" |
|
|
|
|
| def test_load_codex_sessions_uses_session_fallback_when_db_is_missing(tmp_path): |
| codex_home = tmp_path / ".codex" |
| sessions_dir = codex_home / "sessions" / "2026" / "04" / "24" |
| sessions_dir.mkdir(parents=True) |
|
|
| shutil.copy2(FIXTURES_DIR / "fallback-parent.jsonl", sessions_dir / "parent.jsonl") |
| shutil.copy2(FIXTURES_DIR / "fallback-child.jsonl", sessions_dir / "child.jsonl") |
|
|
| rows = load_codex_sessions(codex_home=codex_home, now=FIXED_NOW) |
|
|
| assert [row.id for row in rows] == ["child-thread", "parent-thread"] |
|
|
| child_row = rows[0] |
| assert child_row.source_kind == "codex_sessions" |
| assert child_row.title == "Implement the GTK slice with live reload." |
| assert child_row.children_count == 0 |
| assert child_row.resume_command == "codex resume child-thread" |
| assert child_row.status == "waiting" |
|
|
| parent_row = rows[1] |
| assert parent_row.project_name == "HarnessDashboard" |
| assert parent_row.children_count == 1 |
| assert parent_row.resume_command == "codex resume parent-thread" |
| assert parent_row.source_refs == ["codex_sessions:parent-thread"] |
|
|
|
|
| def test_load_codex_sessions_returns_empty_list_for_empty_tables(tmp_path): |
| codex_home = tmp_path / ".codex" |
| codex_home.mkdir() |
| state_db = codex_home / "state_1.sqlite" |
| _create_state_db(state_db) |
|
|
| assert load_codex_sessions(codex_home=codex_home, now=FIXED_NOW) == [] |
|
|
|
|
| def _create_state_db(path: Path) -> None: |
| connection = sqlite3.connect(path) |
| try: |
| connection.executescript((FIXTURES_DIR / "state_schema.sql").read_text(encoding="utf-8")) |
| connection.commit() |
| finally: |
| connection.close() |
|
|
