sync_session_skill_package.py

#!/usr/bin/env python3
"""Inspect and reconcile incoming offline Codex session packages."""

from __future__ import annotations

import argparse
import copy
import hashlib
import importlib.util
import json
import shutil
import sqlite3
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


def now_utc() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def normalize_cwd(path: str) -> str:
    if path.startswith("\\\\?\\"):
        return path[4:]
    return path


def read_jsonl(path: Path) -> list[dict]:
    rows: list[dict] = []
    if not path.exists():
        return rows
    for line in path.read_text(encoding="utf-8").splitlines():
        if line.strip():
            rows.append(json.loads(line))
    return rows


def write_jsonl(path: Path, rows: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), encoding="utf-8")


def load_manifest(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))


def save_manifest(path: Path, data: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def ensure_threads_schema(con: sqlite3.Connection) -> None:
    cur = con.cursor()
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS threads (
            id TEXT PRIMARY KEY,
            rollout_path TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            updated_at INTEGER NOT NULL,
            source TEXT NOT NULL,
            model_provider TEXT NOT NULL,
            cwd TEXT NOT NULL,
            title TEXT NOT NULL,
            sandbox_policy TEXT NOT NULL,
            approval_mode TEXT NOT NULL,
            tokens_used INTEGER NOT NULL DEFAULT 0,
            has_user_event INTEGER NOT NULL DEFAULT 0,
            archived INTEGER NOT NULL DEFAULT 0,
            archived_at INTEGER,
            git_sha TEXT,
            git_branch TEXT,
            git_origin_url TEXT,
            cli_version TEXT NOT NULL DEFAULT '',
            first_user_message TEXT NOT NULL DEFAULT '',
            agent_nickname TEXT,
            agent_role TEXT,
            memory_mode TEXT NOT NULL DEFAULT 'enabled',
            model TEXT,
            reasoning_effort TEXT,
            agent_path TEXT
        )
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS thread_dynamic_tools (
            thread_id TEXT NOT NULL,
            position INTEGER NOT NULL,
            name TEXT NOT NULL,
            description TEXT NOT NULL,
            input_schema TEXT NOT NULL,
            defer_loading INTEGER NOT NULL DEFAULT 0,
            PRIMARY KEY (thread_id, position)
        )
        """
    )
    con.commit()


def sha256_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def replace_embedded_cwd(text: str) -> str:
    if "<cwd>" not in text or "</cwd>" not in text:
        return text
    start = text.find("<cwd>") + len("<cwd>")
    end = text.find("</cwd>")
    if end <= start:
        return text
    return text[:start] + "<cwd-normalized>" + text[end:]


def canonicalize_event(obj: dict) -> dict:
    normalized = copy.deepcopy(obj)
    payload = normalized.get("payload")
    if not isinstance(payload, dict):
        return normalized

    if normalized.get("type") == "session_meta":
        payload.pop("id", None)
        payload.pop("timestamp", None)
        payload["cwd"] = "<cwd-normalized>"

    if normalized.get("type") == "turn_context":
        payload["cwd"] = "<cwd-normalized>"
        sandbox = payload.get("sandbox_policy")
        if isinstance(sandbox, dict):
            sandbox["writable_roots"] = ["<cwd-normalized>"]

    if (
        normalized.get("type") == "response_item"
        and payload.get("type") == "message"
        and payload.get("role") == "user"
    ):
        for item in payload.get("content", []):
            text_value = item.get("text")
            if isinstance(text_value, str):
                item["text"] = replace_embedded_cwd(text_value)

    return normalized


def canonical_lines(rows: list[dict]) -> list[str]:
    return [json.dumps(canonicalize_event(row), ensure_ascii=False, sort_keys=True, separators=(",", ":")) for row in rows]


def longest_common_prefix(left: list[str], right: list[str]) -> int:
    length = min(len(left), len(right))
    index = 0
    while index < length and left[index] == right[index]:
        index += 1
    return index


def make_session_fingerprint(rows: list[dict]) -> dict[str, Any]:
    lines = canonical_lines(rows)
    full_fingerprint = sha256_text("\n".join(lines))
    last_event_fingerprint = sha256_text(lines[-1]) if lines else ""
    return {
        "canonical_lines": lines,
        "session_fingerprint": f"sess_{full_fingerprint}",
        "event_count": len(lines),
        "last_event_fingerprint": f"evt_{last_event_fingerprint}" if last_event_fingerprint else "",
    }


def sync_root(codex_home: Path) -> Path:
    return codex_home / "session-sync"


def sync_transactions_dir(codex_home: Path) -> Path:
    return sync_root(codex_home) / "transactions"


def sidecar_path(codex_home: Path, name: str) -> Path:
    return sync_root(codex_home) / name


def load_sidecar_rows(codex_home: Path, name: str) -> list[dict]:
    return read_jsonl(sidecar_path(codex_home, name))


def upsert_sidecar_row(codex_home: Path, name: str, key_fields: list[str], new_row: dict) -> None:
    rows = load_sidecar_rows(codex_home, name)
    replaced = False
    for index, row in enumerate(rows):
        if all(row.get(field) == new_row.get(field) for field in key_fields):
            rows[index] = new_row
            replaced = True
            break
    if not replaced:
        rows.append(new_row)
    write_jsonl(sidecar_path(codex_home, name), rows)


def append_sidecar_row(codex_home: Path, name: str, row: dict) -> None:
    rows = load_sidecar_rows(codex_home, name)
    rows.append(row)
    write_jsonl(sidecar_path(codex_home, name), rows)


def replace_or_append_index_row(index_path: Path, session_id: str, title: str, updated_at: str) -> None:
    rows = read_jsonl(index_path)
    replaced = False
    for row in rows:
        if str(row.get("id")) == session_id:
            row["thread_name"] = title
            row["updated_at"] = updated_at
            replaced = True
            break
    if not replaced:
        rows.append({"id": session_id, "thread_name": title, "updated_at": updated_at})
    write_jsonl(index_path, rows)


def backup_file_if_exists(path: Path, backup_path: Path) -> None:
    if path.exists():
        backup_path.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(path, backup_path)


def sync_transaction_dir(codex_home: Path, transaction_id: str) -> Path:
    return sync_transactions_dir(codex_home) / transaction_id


def backup_sidecar_files(codex_home: Path, backups_dir: Path) -> None:
    for name in ["lineages.jsonl", "source-mappings.jsonl", "snapshots.jsonl", "imports.jsonl", "forks.jsonl"]:
        backup_file_if_exists(sidecar_path(codex_home, name), backups_dir / "session-sync" / name)


def remove_sidecar_rows(codex_home: Path, name: str, predicate) -> int:
    rows = load_sidecar_rows(codex_home, name)
    kept = [row for row in rows if not predicate(row)]
    removed = len(rows) - len(kept)
    if removed:
        write_jsonl(sidecar_path(codex_home, name), kept)
    return removed


def restore_file_or_delete(target: Path, backup: Path) -> None:
    if backup.exists():
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(backup, target)
    elif target.exists():
        if target.is_dir():
            shutil.rmtree(target)
        else:
            target.unlink()


def load_thread_row_from_backup_db(backup_db: Path, session_id: str) -> tuple[dict | None, list[dict]]:
    if not backup_db.exists():
        return None, []
    con = sqlite3.connect(backup_db)
    con.row_factory = sqlite3.Row
    row = con.execute("SELECT * FROM threads WHERE id = ?", (session_id,)).fetchone()
    tool_rows: list[dict] = []
    if row is not None:
        try:
            fetched = con.execute(
                "SELECT thread_id, position, name, description, input_schema, defer_loading FROM thread_dynamic_tools WHERE thread_id = ? ORDER BY position",
                (session_id,),
            ).fetchall()
            tool_rows = [dict(item) for item in fetched]
        except sqlite3.OperationalError:
            tool_rows = []
    con.close()
    return (dict(row) if row is not None else None), tool_rows


def restore_thread_row_and_tools(codex_home: Path, backup_db: Path, session_id: str) -> bool:
    row_dict, tool_rows = load_thread_row_from_backup_db(backup_db, session_id)
    if row_dict is None:
        return False
    con = sqlite3.connect(codex_home / "state_5.sqlite")
    ensure_threads_schema(con)
    columns = list(row_dict.keys())
    placeholders = ",".join("?" for _ in columns)
    con.execute(
        f"INSERT OR REPLACE INTO threads ({','.join(columns)}) VALUES ({placeholders})",
        [row_dict[column] for column in columns],
    )
    try:
        con.execute("DELETE FROM thread_dynamic_tools WHERE thread_id = ?", (session_id,))
    except sqlite3.OperationalError:
        pass
    for tool_row in tool_rows:
        con.execute(
            """
            INSERT INTO thread_dynamic_tools (thread_id, position, name, description, input_schema, defer_loading)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                tool_row["thread_id"],
                tool_row["position"],
                tool_row["name"],
                tool_row["description"],
                tool_row["input_schema"],
                tool_row["defer_loading"],
            ),
        )
    con.commit()
    con.close()
    return True


def load_index_row(index_path: Path, session_id: str) -> dict | None:
    for row in read_jsonl(index_path):
        if str(row.get("id")) == session_id:
            return row
    return None


def restore_index_row_from_backup(codex_home: Path, backup_index_path: Path, session_id: str) -> None:
    backup_row = load_index_row(backup_index_path, session_id)
    if backup_row is None:
        delete_session_index_row(codex_home / "session_index.jsonl", session_id)
        return
    replace_or_append_index_row(
        codex_home / "session_index.jsonl",
        session_id,
        str(backup_row.get("thread_name", "")),
        str(backup_row.get("updated_at", "")),
    )


def restore_sidecar_from_backup(codex_home: Path, backups: Path) -> None:
    restore_file_or_delete(sidecar_path(codex_home, "lineages.jsonl"), backups / "session-sync" / "lineages.jsonl")
    restore_file_or_delete(sidecar_path(codex_home, "source-mappings.jsonl"), backups / "session-sync" / "source-mappings.jsonl")
    restore_file_or_delete(sidecar_path(codex_home, "snapshots.jsonl"), backups / "session-sync" / "snapshots.jsonl")
    restore_file_or_delete(sidecar_path(codex_home, "imports.jsonl"), backups / "session-sync" / "imports.jsonl")
    restore_file_or_delete(sidecar_path(codex_home, "forks.jsonl"), backups / "session-sync" / "forks.jsonl")


def delete_session_index_row(index_path: Path, session_id: str) -> None:
    rows = read_jsonl(index_path)
    kept = [row for row in rows if str(row.get("id")) != session_id]
    if len(kept) != len(rows):
        write_jsonl(index_path, kept)


def prune_orphan_lineages(codex_home: Path) -> int:
    lineages = load_sidecar_rows(codex_home, "lineages.jsonl")
    mappings = load_sidecar_rows(codex_home, "source-mappings.jsonl")
    live_ids = {str(row.get("sync_lineage_id", "")) for row in mappings if row.get("sync_lineage_id")}
    kept = [row for row in lineages if str(row.get("sync_lineage_id", "")) in live_ids]
    removed = len(lineages) - len(kept)
    if removed:
        write_jsonl(sidecar_path(codex_home, "lineages.jsonl"), kept)
    return removed


def prune_invalid_source_mappings(codex_home: Path) -> int:
    return remove_sidecar_rows(
        codex_home,
        "source-mappings.jsonl",
        lambda row: not local_session_exists(codex_home, str(row.get("local_session_id", ""))),
    )


def read_sync_transaction_manifest(codex_home: Path, transaction_id: str) -> tuple[Path, dict]:
    tx_dir = sync_transaction_dir(codex_home, transaction_id)
    manifest_path = tx_dir / "manifest.json"
    if not manifest_path.exists():
        raise SystemExit(f"Sync transaction not found: {transaction_id}")
    return tx_dir, load_manifest(manifest_path)


def select_sync_transaction(codex_home: Path, transaction_id: str | None, latest: bool) -> tuple[Path, dict]:
    root = sync_transactions_dir(codex_home)
    if not root.exists():
        raise SystemExit("No session-sync transactions found.")
    if transaction_id:
        return read_sync_transaction_manifest(codex_home, transaction_id)
    if latest:
        manifests = []
        for tx in root.iterdir():
            manifest_path = tx / "manifest.json"
            if manifest_path.exists():
                manifest = load_manifest(manifest_path)
                if manifest.get("status") == "completed":
                    manifests.append((manifest.get("created_at", ""), tx, manifest))
        if not manifests:
            raise SystemExit("No completed sync transactions available for rollback.")
        manifests.sort(key=lambda item: item[0], reverse=True)
        _, tx_dir, manifest = manifests[0]
        return tx_dir, manifest
    raise SystemExit("Provide --transaction-id or --latest.")


def execute_fast_forward_imports(
    package_root: Path,
    codex_home: Path,
    source_env: str,
    source_package: str,
    decisions: list[SessionDecision],
    localized_threads: list[dict],
    index_rows: list[dict],
    original_threads: dict[str, dict],
    transfer: Any,
) -> dict:
    transaction_id = f"sync_{uuid.uuid7()}"
    tx_dir = sync_transaction_dir(codex_home, transaction_id)
    backups_dir = tx_dir / "backups"
    backups_dir.mkdir(parents=True, exist_ok=False)
    created_at = now_utc()

    backup_file_if_exists(codex_home / "state_5.sqlite", backups_dir / "codex-home" / "state_5.sqlite")
    backup_file_if_exists(codex_home / "state_5.sqlite-wal", backups_dir / "codex-home" / "state_5.sqlite-wal")
    backup_file_if_exists(codex_home / "state_5.sqlite-shm", backups_dir / "codex-home" / "state_5.sqlite-shm")
    backup_file_if_exists(codex_home / "session_index.jsonl", backups_dir / "codex-home" / "session_index.jsonl")
    backup_sidecar_files(codex_home, backups_dir)

    localized_by_id = {str(row["id"]): row for row in localized_threads}
    index_by_id = {str(row["id"]): row for row in index_rows}
    results = []

    con = sqlite3.connect(codex_home / "state_5.sqlite")
    transfer.ensure_threads_schema(con)

    for decision in decisions:
        if decision.decision != "fast_forward_import":
            continue
        session_id = str(decision.matched_local_session_id)
        localized_thread = localized_by_id[session_id]
        source_meta = original_threads[session_id]
        source_rollout = package_root / "sessions" / Path(str(source_meta["rollout_path"])).name
        source_text = source_rollout.read_text(encoding="utf-8")
        target_rollout = Path(str(localized_thread["rollout_path"]))
        backup_file_if_exists(target_rollout, backups_dir / "sessions" / target_rollout.name)
        target_rollout.parent.mkdir(parents=True, exist_ok=True)
        target_rollout.write_text(transfer.update_rollout_text(source_text, str(localized_thread["cwd"])), encoding="utf-8")

        thread_row = dict(localized_thread)
        thread_row["updated_at"] = int(datetime.now(timezone.utc).timestamp())
        columns = list(thread_row.keys())
        placeholders = ",".join("?" for _ in columns)
        con.execute(
            f"INSERT OR REPLACE INTO threads ({','.join(columns)}) VALUES ({placeholders})",
            [thread_row[column] for column in columns],
        )

        index_row = index_by_id.get(session_id)
        replace_or_append_index_row(
            codex_home / "session_index.jsonl",
            session_id,
            str(index_row["thread_name"]) if index_row and "thread_name" in index_row else str(localized_thread.get("title", "")),
            now_utc(),
        )

        upsert_sidecar_row(
            codex_home,
            "source-mappings.jsonl",
            ["source_env_fingerprint", "source_session_id", "local_session_id"],
            {
                "sync_lineage_id": decision.sync_lineage_id,
                "source_env_fingerprint": source_env,
                "source_package_fingerprint": source_package,
                "source_session_id": session_id,
                "local_session_id": session_id,
                "adoption_mode": "sync-fast-forward",
                "created_at": created_at,
                "updated_at": created_at,
                "is_current_tip_mapping": True,
            },
        )
        local_row, rollout_rows, _tools = load_local_session_rows(codex_home, session_id)
        upsert_sidecar_row(
            codex_home,
            "snapshots.jsonl",
            ["local_session_id"],
            snapshot_record(local_row, rollout_rows, str(decision.sync_lineage_id)),
        )
        append_sidecar_row(
            codex_home,
            "imports.jsonl",
            {
                "import_id": transaction_id,
                "created_at": created_at,
                "source_env_fingerprint": source_env,
                "source_package_fingerprint": source_package,
                "source_session_id": session_id,
                "matched_local_session_id": session_id,
                "sync_lineage_id": decision.sync_lineage_id,
                "decision": "fast_forward_import",
                "execute_mode": "execute",
                "notes": decision.reason,
            },
        )
        results.append(
            {
                "local_session_id": session_id,
                "rollout_path": str(target_rollout),
                "sync_lineage_id": decision.sync_lineage_id,
            }
        )

    con.commit()
    con.close()

    save_manifest(
        tx_dir / "manifest.json",
        {
            "transaction_id": transaction_id,
            "created_at": created_at,
            "status": "completed",
            "package_root": str(package_root),
            "codex_home": str(codex_home),
            "decision_counts": {"fast_forward_import": len(results)},
            "results": results,
        },
    )
    return {"transaction_id": transaction_id, "results": results}


def source_env_fingerprint(package_root: Path, manifest: dict) -> str:
    explicit = manifest.get("source_env_fingerprint")
    if isinstance(explicit, str) and explicit.strip():
        return explicit.strip()
    seed = json.dumps(
        {
            "format_version": manifest.get("format_version"),
            "package_kind": manifest.get("package_kind"),
            "cwd_filter": manifest.get("cwd_filter"),
            "preferred_installer_skill": manifest.get("preferred_installer_skill"),
            "skills_included": manifest.get("skills_included", []),
            "deferred_components": manifest.get("deferred_components", []),
        },
        ensure_ascii=False,
        sort_keys=True,
    )
    return f"env_{sha256_text(seed)[:16]}"


def source_package_fingerprint(package_root: Path, manifest: dict) -> str:
    seed = json.dumps({"package": str(package_root), "manifest": manifest}, ensure_ascii=False, sort_keys=True)
    return f"pkg_{sha256_text(seed)[:16]}"


def load_transfer_module() -> Any:
    script_path = Path(__file__).resolve().parents[2] / "codex-session-transfer" / "scripts" / "install_session_skill_package.py"
    spec = importlib.util.spec_from_file_location("codex_session_transfer_install", script_path)
    if spec is None or spec.loader is None:
        raise SystemExit(f"Unable to load transfer module from {script_path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def ensure_threads_row(con: sqlite3.Connection, session_id: str) -> sqlite3.Row:
    con.row_factory = sqlite3.Row
    row = con.execute("SELECT * FROM threads WHERE id = ?", (session_id,)).fetchone()
    if row is None:
        raise SystemExit(f"Local session not found: {session_id}")
    return row


def maybe_count_dynamic_tools(con: sqlite3.Connection, session_id: str) -> int:
    try:
        row = con.execute("SELECT COUNT(*) FROM thread_dynamic_tools WHERE thread_id = ?", (session_id,)).fetchone()
    except sqlite3.OperationalError:
        return 0
    return int(row[0])


def load_local_session_rows(codex_home: Path, session_id: str) -> tuple[sqlite3.Row, list[dict], int]:
    con = sqlite3.connect(codex_home / "state_5.sqlite")
    row = ensure_threads_row(con, session_id)
    tools_count = maybe_count_dynamic_tools(con, session_id)
    con.close()
    rollout_rows = read_jsonl(Path(row["rollout_path"]))
    return row, rollout_rows, tools_count


def snapshot_record(local_row: sqlite3.Row, rows: list[dict], sync_lineage_id: str) -> dict:
    fingerprint = make_session_fingerprint(rows)
    return {
        "local_session_id": str(local_row["id"]),
        "sync_lineage_id": sync_lineage_id,
        "rollout_path": str(local_row["rollout_path"]),
        "cwd": normalize_cwd(str(local_row["cwd"])),
        "session_fingerprint": fingerprint["session_fingerprint"],
        "prefix_fingerprint": fingerprint["session_fingerprint"],
        "event_count": fingerprint["event_count"],
        "message_count": sum(1 for row in rows if row.get("type") == "response_item"),
        "last_event_fingerprint": fingerprint["last_event_fingerprint"],
        "updated_at": now_utc(),
    }


def utc_iso_with_millis(dt: datetime) -> str:
    return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{int(dt.microsecond / 1000):03d}Z"


def utc_iso_seconds(dt: datetime) -> str:
    return dt.replace(microsecond=0).isoformat().replace("+00:00", "Z")


def create_local_session_identity(codex_home: Path) -> dict[str, Any]:
    new_uuid = uuid.uuid7()
    new_id = str(new_uuid)
    created_ms = new_uuid.time
    created_dt_utc = datetime.fromtimestamp(created_ms / 1000, tz=timezone.utc)
    rollout_name = f"rollout-{created_dt_utc.strftime('%Y-%m-%dT%H-%M-%S')}-{new_id}.jsonl"
    rollout_path = codex_home / "sessions" / created_dt_utc.strftime("%Y") / created_dt_utc.strftime("%m") / created_dt_utc.strftime("%d") / rollout_name
    return {
        "new_id": new_id,
        "created_at_epoch": int(created_dt_utc.timestamp()),
        "created_at_z": utc_iso_with_millis(created_dt_utc),
        "updated_at_z": utc_iso_seconds(created_dt_utc),
        "rollout_path": rollout_path,
    }


def rewrite_rollout_for_new_session(rows: list[dict], new_id: str, target_cwd: str, created_at_z: str) -> list[dict]:
    rewritten: list[dict] = []
    for obj in rows:
        cloned = copy.deepcopy(obj)
        payload = cloned.get("payload")

        if cloned.get("type") == "session_meta" and isinstance(payload, dict):
            payload["id"] = new_id
            payload["timestamp"] = created_at_z
            payload["cwd"] = target_cwd

        if cloned.get("type") == "turn_context" and isinstance(payload, dict):
            payload["cwd"] = target_cwd
            sandbox = payload.get("sandbox_policy")
            if isinstance(sandbox, dict):
                sandbox["writable_roots"] = [target_cwd]

        if (
            cloned.get("type") == "response_item"
            and isinstance(payload, dict)
            and payload.get("type") == "message"
            and payload.get("role") == "user"
        ):
            for item in payload.get("content", []):
                text_value = item.get("text")
                if isinstance(text_value, str):
                    item["text"] = replace_embedded_cwd(text_value).replace("<cwd-normalized>", target_cwd)

        rewritten.append(cloned)
    return rewritten


def clone_dynamic_tools(con: sqlite3.Connection, source_id: str, new_id: str) -> int:
    try:
        rows = con.execute(
            "SELECT position, name, description, input_schema, defer_loading FROM thread_dynamic_tools WHERE thread_id = ? ORDER BY position",
            (source_id,),
        ).fetchall()
    except sqlite3.OperationalError:
        return 0

    for row in rows:
        con.execute(
            """
            INSERT INTO thread_dynamic_tools (thread_id, position, name, description, input_schema, defer_loading)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (new_id, row[0], row[1], row[2], row[3], row[4]),
        )
    return len(rows)


def execute_diverged_imports(
    package_root: Path,
    codex_home: Path,
    source_env: str,
    source_package: str,
    decisions: list[SessionDecision],
    localized_threads: list[dict],
    original_threads: dict[str, dict],
) -> dict:
    transaction_id = f"sync_{uuid.uuid7()}"
    tx_dir = sync_transaction_dir(codex_home, transaction_id)
    backups_dir = tx_dir / "backups"
    backups_dir.mkdir(parents=True, exist_ok=False)
    created_at = now_utc()

    backup_file_if_exists(codex_home / "state_5.sqlite", backups_dir / "codex-home" / "state_5.sqlite")
    backup_file_if_exists(codex_home / "state_5.sqlite-wal", backups_dir / "codex-home" / "state_5.sqlite-wal")
    backup_file_if_exists(codex_home / "state_5.sqlite-shm", backups_dir / "codex-home" / "state_5.sqlite-shm")
    backup_file_if_exists(codex_home / "session_index.jsonl", backups_dir / "codex-home" / "session_index.jsonl")
    backup_sidecar_files(codex_home, backups_dir)

    localized_by_id = {str(row["id"]): row for row in localized_threads}
    results = []
    con = sqlite3.connect(codex_home / "state_5.sqlite")

    for decision in decisions:
        if decision.decision != "diverged_lineage":
            continue

        parent_session_id = str(decision.matched_local_session_id)
        localized_thread = localized_by_id[parent_session_id]
        source_meta = original_threads[parent_session_id]
        source_rollout = package_root / "sessions" / Path(str(source_meta["rollout_path"])).name
        incoming_rows = read_jsonl(source_rollout)
        new_identity = create_local_session_identity(codex_home)
        new_session_id = str(new_identity["new_id"])
        target_rollout = Path(str(new_identity["rollout_path"]))

        rewritten_rows = rewrite_rollout_for_new_session(
            incoming_rows,
            new_session_id,
            str(localized_thread["cwd"]),
            str(new_identity["created_at_z"]),
        )
        target_rollout.parent.mkdir(parents=True, exist_ok=True)
        write_jsonl(target_rollout, rewritten_rows)

        parent_row, _parent_rows, _tools = load_local_session_rows(codex_home, parent_session_id)
        row_dict = dict(parent_row)
        row_dict["id"] = new_session_id
        row_dict["rollout_path"] = str(target_rollout)
        row_dict["created_at"] = int(new_identity["created_at_epoch"])
        row_dict["updated_at"] = int(new_identity["created_at_epoch"])
        row_dict["cwd"] = str(localized_thread["cwd"])
        row_dict["title"] = f"{row_dict['title']} (sync fork)"
        if "sandbox_policy" in row_dict:
            row_dict["sandbox_policy"] = json.dumps(
                {
                    "type": "workspace-write",
                    "writable_roots": [str(localized_thread["cwd"])],
                    "network_access": False,
                    "exclude_tmpdir_env_var": False,
                    "exclude_slash_tmp": False,
                },
                ensure_ascii=False,
                separators=(",", ":"),
            )
        columns = list(row_dict.keys())
        placeholders = ",".join("?" for _ in columns)
        con.execute(f"INSERT INTO threads ({','.join(columns)}) VALUES ({placeholders})", [row_dict[column] for column in columns])
        clone_dynamic_tools(con, parent_session_id, new_session_id)

        replace_or_append_index_row(
            codex_home / "session_index.jsonl",
            new_session_id,
            str(row_dict["title"]),
            str(new_identity["updated_at_z"]),
        )

        upsert_sidecar_row(
            codex_home,
            "source-mappings.jsonl",
            ["source_env_fingerprint", "source_session_id", "local_session_id"],
            {
                "sync_lineage_id": decision.sync_lineage_id,
                "source_env_fingerprint": source_env,
                "source_package_fingerprint": source_package,
                "source_session_id": parent_session_id,
                "local_session_id": parent_session_id,
                "adoption_mode": "sync-parent",
                "created_at": created_at,
                "updated_at": created_at,
                "is_current_tip_mapping": False,
            },
        )
        upsert_sidecar_row(
            codex_home,
            "source-mappings.jsonl",
            ["source_env_fingerprint", "source_session_id", "local_session_id"],
            {
                "sync_lineage_id": decision.sync_lineage_id,
                "source_env_fingerprint": source_env,
                "source_package_fingerprint": source_package,
                "source_session_id": parent_session_id,
                "local_session_id": new_session_id,
                "adoption_mode": "sync-forked",
                "created_at": created_at,
                "updated_at": created_at,
                "is_current_tip_mapping": True,
            },
        )

        upsert_sidecar_row(
            codex_home,
            "snapshots.jsonl",
            ["local_session_id"],
            snapshot_record(row_dict, rewritten_rows, str(decision.sync_lineage_id)),
        )
        append_sidecar_row(
            codex_home,
            "forks.jsonl",
            {
                "fork_id": f"fork_{uuid.uuid7()}",
                "sync_lineage_id": decision.sync_lineage_id,
                "parent_local_session_id": parent_session_id,
                "child_local_session_id": new_session_id,
                "fork_point_event_index": int(decision.common_prefix_events),
                "fork_point_fingerprint": "",
                "created_at": created_at,
                "source": "incoming-package-divergence",
            },
        )
        append_sidecar_row(
            codex_home,
            "imports.jsonl",
            {
                "import_id": transaction_id,
                "created_at": created_at,
                "source_env_fingerprint": source_env,
                "source_package_fingerprint": source_package,
                "source_session_id": parent_session_id,
                "matched_local_session_id": new_session_id,
                "sync_lineage_id": decision.sync_lineage_id,
                "decision": "diverged_lineage",
                "execute_mode": "execute",
                "notes": decision.reason,
            },
        )
        results.append(
            {
                "parent_local_session_id": parent_session_id,
                "child_local_session_id": new_session_id,
                "rollout_path": str(target_rollout),
                "sync_lineage_id": decision.sync_lineage_id,
                "fork_point_event_index": decision.common_prefix_events,
            }
        )

    con.commit()
    con.close()
    save_manifest(
        tx_dir / "manifest.json",
        {
            "transaction_id": transaction_id,
            "created_at": created_at,
            "status": "completed",
            "package_root": str(package_root),
            "codex_home": str(codex_home),
            "decision_counts": {"diverged_lineage": len(results)},
            "results": results,
        },
    )
    return {"transaction_id": transaction_id, "results": results}


def existing_mapping_by_source(codex_home: Path) -> dict[tuple[str, str], dict]:
    rows = load_sidecar_rows(codex_home, "source-mappings.jsonl")
    result: dict[tuple[str, str], dict] = {}
    for row in rows:
        key = (str(row.get("source_env_fingerprint", "")), str(row.get("source_session_id", "")))
        if key != ("", ""):
            result[key] = row
    return result


def existing_mappings_for_session_id(codex_home: Path, source_session_id: str) -> list[dict]:
    rows = load_sidecar_rows(codex_home, "source-mappings.jsonl")
    return [row for row in rows if str(row.get("source_session_id", "")) == source_session_id]


def local_session_exists(codex_home: Path, session_id: str) -> bool:
    con = sqlite3.connect(codex_home / "state_5.sqlite")
    row = con.execute("SELECT 1 FROM threads WHERE id = ?", (session_id,)).fetchone()
    con.close()
    return row is not None


def resolve_known_mapping(codex_home: Path, source_env: str, source_session_id: str) -> tuple[dict | None, str | None]:
    mapping = existing_mapping_by_source(codex_home).get((source_env, source_session_id))
    if mapping is not None and local_session_exists(codex_home, str(mapping.get("local_session_id", ""))):
        return mapping, "exact-source-env"

    legacy_candidates = [
        row
        for row in existing_mappings_for_session_id(codex_home, source_session_id)
        if local_session_exists(codex_home, str(row.get("local_session_id", "")))
    ]
    if len(legacy_candidates) == 1:
        return legacy_candidates[0], "source-session-id-fallback"
    if len(legacy_candidates) > 1:
        current_tip = [row for row in legacy_candidates if bool(row.get("is_current_tip_mapping"))]
        if len(current_tip) == 1:
            return current_tip[0], "source-session-id-fallback-current-tip"

    return None, None


def existing_snapshot_by_local(codex_home: Path) -> dict[str, dict]:
    rows = load_sidecar_rows(codex_home, "snapshots.jsonl")
    return {str(row["local_session_id"]): row for row in rows if "local_session_id" in row}


def create_lineage_id(seed: str) -> str:
    return f"sl_{sha256_text(seed)[:16]}"


def adopt_local_session(codex_home: Path, local_session_id: str, source_env: str, source_session_id: str, adoption_mode: str) -> dict:
    local_row, rollout_rows, _ = load_local_session_rows(codex_home, local_session_id)
    existing_snapshots = existing_snapshot_by_local(codex_home)
    if local_session_id in existing_snapshots:
        return {
            "status": "already-adopted",
            "local_session_id": local_session_id,
            "sync_lineage_id": existing_snapshots[local_session_id]["sync_lineage_id"],
        }

    sync_lineage_id = create_lineage_id(f"{source_env}:{source_session_id}:{local_session_id}")
    created_at = now_utc()
    upsert_sidecar_row(
        codex_home,
        "lineages.jsonl",
        ["sync_lineage_id"],
        {
            "sync_lineage_id": sync_lineage_id,
            "lineage_version": 1,
            "created_at": created_at,
            "updated_at": created_at,
            "origin_hint": {
                "source_env_fingerprint": source_env,
                "first_source_session_id": source_session_id,
            },
            "status": "active",
        },
    )
    upsert_sidecar_row(
        codex_home,
        "source-mappings.jsonl",
        ["source_env_fingerprint", "source_session_id", "local_session_id"],
        {
            "sync_lineage_id": sync_lineage_id,
            "source_env_fingerprint": source_env,
            "source_package_fingerprint": "",
            "source_session_id": source_session_id,
            "local_session_id": local_session_id,
            "adoption_mode": adoption_mode,
            "created_at": created_at,
            "updated_at": created_at,
            "is_current_tip_mapping": True,
        },
    )
    upsert_sidecar_row(
        codex_home,
        "snapshots.jsonl",
        ["local_session_id"],
        snapshot_record(local_row, rollout_rows, sync_lineage_id),
    )
    return {
        "status": "adopted",
        "local_session_id": local_session_id,
        "sync_lineage_id": sync_lineage_id,
    }


@dataclass
class SessionDecision:
    source_session_id: str
    matched_local_session_id: str | None
    sync_lineage_id: str | None
    decision: str
    reason: str
    incoming_event_count: int
    local_event_count: int | None
    common_prefix_events: int
    proposed_local_session_id: str | None
    needs_adoption: bool


def classify_session(
    codex_home: Path,
    source_env: str,
    incoming_thread: dict,
    incoming_rows: list[dict],
) -> SessionDecision:
    source_session_id = str(incoming_thread["id"])
    incoming_fp = make_session_fingerprint(incoming_rows)
    mapping, mapping_mode = resolve_known_mapping(codex_home, source_env, source_session_id)
    local_candidate_id: str | None = None
    sync_lineage_id: str | None = None
    needs_adoption = False

    if mapping is not None:
        local_candidate_id = str(mapping["local_session_id"])
        sync_lineage_id = str(mapping["sync_lineage_id"])
    else:
        con = sqlite3.connect(codex_home / "state_5.sqlite")
        con.row_factory = sqlite3.Row
        row = con.execute("SELECT id FROM threads WHERE id = ?", (source_session_id,)).fetchone()
        con.close()
        if row is not None:
            local_candidate_id = str(row["id"])
            needs_adoption = True

    if local_candidate_id is None:
        return SessionDecision(
            source_session_id=source_session_id,
            matched_local_session_id=None,
            sync_lineage_id=None,
            decision="new_lineage",
            reason="No existing lineage mapping or local session id match was found.",
            incoming_event_count=incoming_fp["event_count"],
            local_event_count=None,
            common_prefix_events=0,
            proposed_local_session_id=None,
            needs_adoption=False,
        )

    local_row, local_rows, _ = load_local_session_rows(codex_home, local_candidate_id)
    local_fp = make_session_fingerprint(local_rows)
    common_prefix = longest_common_prefix(local_fp["canonical_lines"], incoming_fp["canonical_lines"])
    if incoming_fp["session_fingerprint"] == local_fp["session_fingerprint"]:
        return SessionDecision(
            source_session_id=source_session_id,
            matched_local_session_id=local_candidate_id,
            sync_lineage_id=sync_lineage_id,
            decision="unchanged",
            reason=(
                "Incoming transcript fingerprint matches the local session exactly."
                if mapping_mode != "source-session-id-fallback"
                else "Incoming transcript matches the local session exactly after legacy source mapping fallback."
            ),
            incoming_event_count=incoming_fp["event_count"],
            local_event_count=local_fp["event_count"],
            common_prefix_events=common_prefix,
            proposed_local_session_id=None,
            needs_adoption=needs_adoption,
        )
    if common_prefix == local_fp["event_count"] and incoming_fp["event_count"] > local_fp["event_count"]:
        return SessionDecision(
            source_session_id=source_session_id,
            matched_local_session_id=local_candidate_id,
            sync_lineage_id=sync_lineage_id,
            decision="fast_forward_import",
            reason=(
                "Incoming transcript is a strict superset of the local session."
                if mapping_mode != "source-session-id-fallback"
                else "Incoming transcript is a strict superset of the local session after legacy source mapping fallback."
            ),
            incoming_event_count=incoming_fp["event_count"],
            local_event_count=local_fp["event_count"],
            common_prefix_events=common_prefix,
            proposed_local_session_id=None,
            needs_adoption=needs_adoption,
        )
    if common_prefix == incoming_fp["event_count"] and local_fp["event_count"] > incoming_fp["event_count"]:
        return SessionDecision(
            source_session_id=source_session_id,
            matched_local_session_id=local_candidate_id,
            sync_lineage_id=sync_lineage_id,
            decision="incoming_subset",
            reason=(
                "Incoming transcript is an older subset of the local session."
                if mapping_mode != "source-session-id-fallback"
                else "Incoming transcript is an older subset of the local session after legacy source mapping fallback."
            ),
            incoming_event_count=incoming_fp["event_count"],
            local_event_count=local_fp["event_count"],
            common_prefix_events=common_prefix,
            proposed_local_session_id=None,
            needs_adoption=needs_adoption,
        )
    if common_prefix > 0:
        return SessionDecision(
            source_session_id=source_session_id,
            matched_local_session_id=local_candidate_id,
            sync_lineage_id=sync_lineage_id,
            decision="diverged_lineage",
            reason=(
                "Incoming and local transcripts share a prefix but diverge afterward."
                if mapping_mode != "source-session-id-fallback"
                else "Incoming and local transcripts share a prefix but diverge afterward after legacy source mapping fallback."
            ),
            incoming_event_count=incoming_fp["event_count"],
            local_event_count=local_fp["event_count"],
            common_prefix_events=common_prefix,
            proposed_local_session_id=str(uuid.uuid7()),
            needs_adoption=needs_adoption,
        )
    return SessionDecision(
        source_session_id=source_session_id,
        matched_local_session_id=str(local_row["id"]),
        sync_lineage_id=sync_lineage_id,
        decision="id_collision_unrelated",
        reason="The session id collides locally but transcript comparison does not support same-lineage matching.",
        incoming_event_count=incoming_fp["event_count"],
        local_event_count=local_fp["event_count"],
        common_prefix_events=common_prefix,
        proposed_local_session_id=str(uuid.uuid7()),
        needs_adoption=needs_adoption,
    )


def inspect_package(package_root: Path, codex_home: Path) -> dict:
    transfer = load_transfer_module()
    manifest = load_manifest(package_root / "package_manifest.json")
    source_env = source_env_fingerprint(package_root, manifest)
    source_package = source_package_fingerprint(package_root, manifest)
    threads = read_jsonl(package_root / "metadata" / "threads.jsonl")
    sessions = []
    for row in threads:
        mapping, mapping_mode = resolve_known_mapping(codex_home, source_env, str(row["id"]))
        sessions.append(
            {
                "source_session_id": str(row["id"]),
                "title": row.get("title", ""),
                "cwd": normalize_cwd(str(row.get("cwd", ""))),
                "known_mapping": mapping,
                "known_mapping_mode": mapping_mode,
                "source_rollout": str(package_root / "sessions" / Path(str(row["rollout_path"])).name),
            }
        )
    return {
        "mode": "inspect-package",
        "manifest": manifest,
        "preferred_installer_skill": manifest.get("preferred_installer_skill"),
        "source_env_fingerprint": source_env,
        "source_package_fingerprint": source_package,
        "session_count": len(threads),
        "sessions": sessions,
        "transfer_module_loaded": bool(transfer),
    }


def sync_package(package_root: Path, codex_home: Path, remote_home: str, execute: bool) -> dict:
    transfer = load_transfer_module()
    manifest, plan, localized_threads, index_rows, tool_rows = transfer.build_install_plan(package_root, codex_home, remote_home)
    source_env = source_env_fingerprint(package_root, manifest)
    source_package = source_package_fingerprint(package_root, manifest)
    original_threads = {str(row["id"]): row for row in read_jsonl(package_root / "metadata" / "threads.jsonl")}
    decisions: list[SessionDecision] = []
    adoption_actions = []

    for thread in localized_threads:
        source_meta = original_threads[str(thread["id"])]
        source_rollout = package_root / "sessions" / Path(str(source_meta["rollout_path"])).name
        incoming_rows = read_jsonl(source_rollout)
        decision = classify_session(codex_home, source_env, thread, incoming_rows)
        decisions.append(decision)
        if decision.needs_adoption and decision.matched_local_session_id is not None:
            adoption_actions.append(
                {
                    "local_session_id": decision.matched_local_session_id,
                    "source_session_id": decision.source_session_id,
                    "adoption_mode": "transfer-adopted",
                }
            )

    summary = {
        "mode": "execute" if execute else "dry-run",
        "manifest": manifest,
        "source_env_fingerprint": source_env,
        "source_package_fingerprint": source_package,
        "plan_summary": {
            "sessions": len(plan["sessions"]),
            "skills": len(plan["skills"]),
            "directories": len(plan["directories"]),
        },
        "adoption_actions": adoption_actions,
        "decisions": [
            {
                "source_session_id": item.source_session_id,
                "matched_local_session_id": item.matched_local_session_id,
                "sync_lineage_id": item.sync_lineage_id,
                "decision": item.decision,
                "reason": item.reason,
                "incoming_event_count": item.incoming_event_count,
                "local_event_count": item.local_event_count,
                "common_prefix_events": item.common_prefix_events,
                "proposed_local_session_id": item.proposed_local_session_id,
                "needs_adoption": item.needs_adoption,
            }
            for item in decisions
        ],
    }

    if not execute:
        return summary

    write_decisions = {item.decision for item in decisions if item.decision not in {"unchanged", "incoming_subset"}}
    if write_decisions == {"new_lineage"}:
        tx_dir, transaction, _operations = transfer.build_transaction(codex_home, package_root, remote_home, plan)
        transfer_result = transfer.execute_install(
            package_root,
            codex_home,
            plan,
            localized_threads,
            index_rows,
            tool_rows,
            tx_dir,
            transaction,
        )
        sync_transaction_id = f"sync_{transaction['transaction_id']}"
        created_at = now_utc()
        backup_sidecar_dir = sync_transactions_dir(codex_home) / sync_transaction_id / "backups"
        backup_sidecar_dir.mkdir(parents=True, exist_ok=True)
        backup_sidecar_files(codex_home, backup_sidecar_dir)
        save_manifest(
            sync_transactions_dir(codex_home) / sync_transaction_id / "manifest.json",
            {
                "transaction_id": sync_transaction_id,
                "created_at": created_at,
                "status": "completed",
                "package_root": str(package_root),
                "codex_home": str(codex_home),
                "remote_home": remote_home,
                "linked_transfer_transaction_id": transaction["transaction_id"],
                "decision_counts": {
                    "new_lineage": sum(1 for item in decisions if item.decision == "new_lineage"),
                },
            },
        )
        for thread in localized_threads:
            local_session_id = str(thread["id"])
            adoption_result = adopt_local_session(
                codex_home,
                local_session_id,
                source_env,
                local_session_id,
                adoption_mode="sync-imported",
            )
            sync_lineage_id = adoption_result["sync_lineage_id"]
            upsert_sidecar_row(
                codex_home,
                "source-mappings.jsonl",
                ["source_env_fingerprint", "source_session_id", "local_session_id"],
                {
                    "sync_lineage_id": sync_lineage_id,
                    "source_env_fingerprint": source_env,
                    "source_package_fingerprint": source_package,
                    "source_session_id": local_session_id,
                    "local_session_id": local_session_id,
                    "adoption_mode": "sync-imported",
                    "created_at": created_at,
                    "updated_at": created_at,
                    "is_current_tip_mapping": True,
                },
            )
            append_sidecar_row(
                codex_home,
                "imports.jsonl",
                {
                    "import_id": sync_transaction_id,
                    "created_at": created_at,
                    "source_env_fingerprint": source_env,
                    "source_package_fingerprint": source_package,
                    "source_session_id": local_session_id,
                    "matched_local_session_id": local_session_id,
                    "sync_lineage_id": sync_lineage_id,
                    "decision": "new_lineage",
                    "execute_mode": "execute",
                    "notes": "Installed via transfer-compatible new_lineage path.",
                },
            )
        summary["status"] = "completed"
        summary["transaction_id"] = sync_transaction_id
        summary["linked_transfer_transaction_id"] = transaction["transaction_id"]
        summary["transfer_result"] = transfer_result
        return summary

    if write_decisions == {"fast_forward_import"}:
        fast_forward_result = execute_fast_forward_imports(
            package_root,
            codex_home,
            source_env,
            source_package,
            decisions,
            localized_threads,
            index_rows,
            original_threads,
            transfer,
        )
        summary["status"] = "completed"
        summary["transaction_id"] = fast_forward_result["transaction_id"]
        summary["fast_forward_result"] = fast_forward_result
        return summary

    if write_decisions == {"diverged_lineage"}:
        diverged_result = execute_diverged_imports(
            package_root,
            codex_home,
            source_env,
            source_package,
            decisions,
            localized_threads,
            original_threads,
        )
        summary["status"] = "completed"
        summary["transaction_id"] = diverged_result["transaction_id"]
        summary["diverged_result"] = diverged_result
        return summary

    supported = {"unchanged", "incoming_subset"}
    unsupported = [item for item in decisions if item.decision not in supported]
    if unsupported:
        summary["status"] = "blocked"
        summary["message"] = (
            "Execute mode currently supports only pure new_lineage installs or fully non-writing sync plans. "
            "fast_forward_import, diverged_lineage, and id_collision_unrelated remain dry-run only."
        )
        return summary

    import_id = f"imp_{uuid.uuid7()}"
    transaction = {
        "transaction_id": import_id,
        "created_at": now_utc(),
        "status": "completed",
        "package_root": str(package_root),
        "codex_home": str(codex_home),
        "remote_home": remote_home,
        "decision_counts": {
            "unchanged": sum(1 for item in decisions if item.decision == "unchanged"),
            "incoming_subset": sum(1 for item in decisions if item.decision == "incoming_subset"),
        },
    }
    save_manifest(sync_transactions_dir(codex_home) / import_id / "manifest.json", transaction)
    for item in decisions:
        append_sidecar_row(
            codex_home,
            "imports.jsonl",
            {
                "import_id": import_id,
                "created_at": transaction["created_at"],
                "source_env_fingerprint": source_env,
                "source_package_fingerprint": source_package,
                "source_session_id": item.source_session_id,
                "matched_local_session_id": item.matched_local_session_id,
                "sync_lineage_id": item.sync_lineage_id,
                "decision": item.decision,
                "execute_mode": "execute",
                "notes": item.reason,
            },
        )
    summary["status"] = "completed"
    summary["transaction_id"] = import_id
    return summary


def list_transactions(codex_home: Path) -> dict:
    rows = []
    root = sync_transactions_dir(codex_home)
    if root.exists():
        for item in sorted(root.iterdir()):
            manifest_path = item / "manifest.json"
            if manifest_path.exists():
                rows.append(load_manifest(manifest_path))
    rows.sort(key=lambda row: row.get("created_at", ""), reverse=True)
    return {"transactions": rows}


def rollback_new_lineage_transaction(codex_home: Path, tx_dir: Path, manifest: dict, execute: bool) -> dict:
    plan = {
        "transaction_id": manifest["transaction_id"],
        "kind": "new_lineage",
        "linked_transfer_transaction_id": manifest.get("linked_transfer_transaction_id"),
    }
    if not execute:
        return {"mode": "dry-run-rollback", "plan": plan}

    transfer = load_transfer_module()
    transfer_tx_dir, transfer_manifest = transfer.select_transaction(
        codex_home, manifest.get("linked_transfer_transaction_id"), latest=False
    )
    transfer.rollback_transaction(codex_home, transfer_tx_dir, transfer_manifest, execute=True)

    impacted_imports = [row for row in load_sidecar_rows(codex_home, "imports.jsonl") if str(row.get("import_id")) == manifest["transaction_id"]]
    restore_file_or_delete(sidecar_path(codex_home, "lineages.jsonl"), tx_dir / "backups" / "session-sync" / "lineages.jsonl")
    restore_file_or_delete(sidecar_path(codex_home, "source-mappings.jsonl"), tx_dir / "backups" / "session-sync" / "source-mappings.jsonl")
    restore_file_or_delete(sidecar_path(codex_home, "snapshots.jsonl"), tx_dir / "backups" / "session-sync" / "snapshots.jsonl")
    restore_file_or_delete(sidecar_path(codex_home, "imports.jsonl"), tx_dir / "backups" / "session-sync" / "imports.jsonl")
    restore_file_or_delete(sidecar_path(codex_home, "forks.jsonl"), tx_dir / "backups" / "session-sync" / "forks.jsonl")
    if not impacted_imports:
        impacted_imports = [row for row in load_sidecar_rows(codex_home, "imports.jsonl") if str(row.get("import_id")) == manifest["transaction_id"]]
    remove_sidecar_rows(
        codex_home,
        "imports.jsonl",
        lambda row: str(row.get("import_id")) == manifest["transaction_id"],
    )
    impacted_local_ids = {str(row.get("matched_local_session_id", "")) for row in impacted_imports}
    impacted_lineage_ids = {str(row.get("sync_lineage_id", "")) for row in impacted_imports}
    remove_sidecar_rows(
        codex_home,
        "source-mappings.jsonl",
        lambda row: str(row.get("local_session_id", "")) in impacted_local_ids and str(row.get("sync_lineage_id", "")) in impacted_lineage_ids,
    )
    remove_sidecar_rows(
        codex_home,
        "snapshots.jsonl",
        lambda row: str(row.get("local_session_id", "")) in impacted_local_ids,
    )
    prune_invalid_source_mappings(codex_home)
    prune_orphan_lineages(codex_home)

    manifest["status"] = "rolled_back"
    manifest["rolled_back_at"] = now_utc()
    save_manifest(tx_dir / "manifest.json", manifest)
    return {"mode": "rollback", "transaction_id": manifest["transaction_id"], "kind": "new_lineage"}


def rollback_fast_forward_transaction(codex_home: Path, tx_dir: Path, manifest: dict, execute: bool) -> dict:
    plan = {
        "transaction_id": manifest["transaction_id"],
        "kind": "fast_forward_import",
        "results": manifest.get("results", []),
    }
    if not execute:
        return {"mode": "dry-run-rollback", "plan": plan}

    backups = tx_dir / "backups"
    backup_db = backups / "codex-home" / "state_5.sqlite"
    backup_index = backups / "codex-home" / "session_index.jsonl"

    for result in manifest.get("results", []):
        session_id = str(result.get("local_session_id", ""))
        rollout = Path(str(result.get("rollout_path", "")))
        restore_file_or_delete(rollout, backups / "sessions" / rollout.name)
        if session_id:
            restore_thread_row_and_tools(codex_home, backup_db, session_id)
            restore_index_row_from_backup(codex_home, backup_index, session_id)

    restore_sidecar_from_backup(codex_home, backups)

    manifest["status"] = "rolled_back"
    manifest["rolled_back_at"] = now_utc()
    save_manifest(tx_dir / "manifest.json", manifest)
    return {"mode": "rollback", "transaction_id": manifest["transaction_id"], "kind": "fast_forward_import"}


def rollback_diverged_transaction(codex_home: Path, tx_dir: Path, manifest: dict, execute: bool) -> dict:
    plan = {
        "transaction_id": manifest["transaction_id"],
        "kind": "diverged_lineage",
        "results": manifest.get("results", []),
    }
    if not execute:
        return {"mode": "dry-run-rollback", "plan": plan}

    backups = tx_dir / "backups"
    con = sqlite3.connect(codex_home / "state_5.sqlite")
    for result in manifest.get("results", []):
        child_session_id = str(result.get("child_local_session_id", ""))
        if child_session_id:
            try:
                con.execute("DELETE FROM thread_dynamic_tools WHERE thread_id = ?", (child_session_id,))
            except sqlite3.OperationalError:
                pass
            con.execute("DELETE FROM threads WHERE id = ?", (child_session_id,))
            delete_session_index_row(codex_home / "session_index.jsonl", child_session_id)
    con.commit()
    con.close()

    restore_sidecar_from_backup(codex_home, backups)

    for result in manifest.get("results", []):
        rollout = Path(str(result.get("rollout_path", "")))
        restore_file_or_delete(rollout, backups / "sessions" / rollout.name)

    manifest["status"] = "rolled_back"
    manifest["rolled_back_at"] = now_utc()
    save_manifest(tx_dir / "manifest.json", manifest)
    return {"mode": "rollback", "transaction_id": manifest["transaction_id"], "kind": "diverged_lineage"}


def rollback_transaction(codex_home: Path, transaction_id: str | None, latest: bool, execute: bool) -> dict:
    tx_dir, manifest = select_sync_transaction(codex_home, transaction_id, latest)
    kinds = [key for key, value in manifest.get("decision_counts", {}).items() if int(value) > 0]
    kind = kinds[0] if kinds else "non_writing"
    if kind == "new_lineage":
        return rollback_new_lineage_transaction(codex_home, tx_dir, manifest, execute)
    if kind == "fast_forward_import":
        return rollback_fast_forward_transaction(codex_home, tx_dir, manifest, execute)
    if kind == "diverged_lineage":
        return rollback_diverged_transaction(codex_home, tx_dir, manifest, execute)
    if not execute:
        return {"mode": "dry-run-rollback", "plan": {"transaction_id": manifest["transaction_id"], "kind": kind}}
    manifest["status"] = "rolled_back"
    manifest["rolled_back_at"] = now_utc()
    save_manifest(tx_dir / "manifest.json", manifest)
    return {"mode": "rollback", "transaction_id": manifest["transaction_id"], "kind": kind}


def prune_invalid_mappings_command(codex_home: Path) -> dict:
    removed = prune_invalid_source_mappings(codex_home)
    pruned_lineages = prune_orphan_lineages(codex_home)
    return {
        "mode": "prune-invalid-mappings",
        "removed_mappings": removed,
        "removed_orphan_lineages": pruned_lineages,
    }


def sidecar_health(codex_home: Path) -> dict:
    names = ["lineages.jsonl", "source-mappings.jsonl", "snapshots.jsonl", "imports.jsonl", "forks.jsonl"]
    files = []
    for name in names:
        path = sidecar_path(codex_home, name)
        files.append(
            {
                "name": name,
                "exists": path.exists(),
                "row_count": len(read_jsonl(path)) if path.exists() else 0,
            }
        )

    invalid_mappings = []
    for row in load_sidecar_rows(codex_home, "source-mappings.jsonl"):
        local_session_id = str(row.get("local_session_id", ""))
        if local_session_id and not local_session_exists(codex_home, local_session_id):
            invalid_mappings.append(row)

    return {
        "mode": "sidecar-health",
        "files": files,
        "invalid_mapping_count": len(invalid_mappings),
        "invalid_mappings": invalid_mappings,
    }


def rebuild_sidecar_from_sessions(codex_home: Path, session_ids: list[str], source_env: str) -> dict:
    adopted = []
    for session_id in session_ids:
        adopted.append(
            adopt_local_session(
                codex_home,
                session_id,
                source_env,
                session_id,
                adoption_mode="rebuild-adopt",
            )
        )
    return {
        "mode": "rebuild-sidecar",
        "source_env": source_env,
        "adopted_sessions": adopted,
    }


def replay_transactions_to_sidecar(codex_home: Path, include_rolled_back: bool = False) -> dict:
    root = sync_transactions_dir(codex_home)
    manifests: list[dict] = []
    if root.exists():
        for tx in sorted(root.iterdir()):
            manifest_path = tx / "manifest.json"
            if manifest_path.exists():
                manifest = load_manifest(manifest_path)
                if include_rolled_back or manifest.get("status") == "completed":
                    manifests.append(manifest)

    import_rows = load_sidecar_rows(codex_home, "imports.jsonl")
    fork_rows = load_sidecar_rows(codex_home, "forks.jsonl")
    existing_import_ids = {str(row.get("import_id", "")) for row in import_rows}
    existing_fork_keys = {
        (
            str(row.get("sync_lineage_id", "")),
            str(row.get("parent_local_session_id", "")),
            str(row.get("child_local_session_id", "")),
        )
        for row in fork_rows
    }

    added_imports = 0
    added_forks = 0

    for manifest in manifests:
        tx_id = str(manifest.get("transaction_id", ""))
        if not tx_id or tx_id in existing_import_ids:
            pass
        decision_counts = manifest.get("decision_counts", {})
        results = manifest.get("results", [])
        created_at = str(manifest.get("created_at", now_utc()))

        if "fast_forward_import" in decision_counts and int(decision_counts["fast_forward_import"]) > 0:
            for result in results:
                import_rows.append(
                    {
                        "import_id": tx_id,
                        "created_at": created_at,
                        "source_env_fingerprint": "",
                        "source_package_fingerprint": "",
                        "source_session_id": str(result.get("local_session_id", "")),
                        "matched_local_session_id": str(result.get("local_session_id", "")),
                        "sync_lineage_id": str(result.get("sync_lineage_id", "")),
                        "decision": "fast_forward_import",
                        "execute_mode": "execute",
                        "notes": "Replayed from transaction manifest.",
                    }
                )
                added_imports += 1
        if "diverged_lineage" in decision_counts and int(decision_counts["diverged_lineage"]) > 0:
            for result in results:
                child_id = str(result.get("child_local_session_id", ""))
                if child_id and not local_session_exists(codex_home, child_id) and not include_rolled_back:
                    continue
                import_rows.append(
                    {
                        "import_id": tx_id,
                        "created_at": created_at,
                        "source_env_fingerprint": "",
                        "source_package_fingerprint": "",
                        "source_session_id": str(result.get("parent_local_session_id", "")),
                        "matched_local_session_id": child_id,
                        "sync_lineage_id": str(result.get("sync_lineage_id", "")),
                        "decision": "diverged_lineage",
                        "execute_mode": "execute",
                        "notes": "Replayed from transaction manifest.",
                    }
                )
                added_imports += 1
                fork_key = (
                    str(result.get("sync_lineage_id", "")),
                    str(result.get("parent_local_session_id", "")),
                    child_id,
                )
                if fork_key not in existing_fork_keys:
                    fork_rows.append(
                        {
                            "fork_id": f"fork_replayed_{tx_id}",
                            "sync_lineage_id": fork_key[0],
                            "parent_local_session_id": fork_key[1],
                            "child_local_session_id": fork_key[2],
                            "fork_point_event_index": int(result.get("fork_point_event_index", 0)),
                            "fork_point_fingerprint": "",
                            "created_at": created_at,
                            "source": "replayed-transaction-manifest",
                        }
                    )
                    existing_fork_keys.add(fork_key)
                    added_forks += 1
        if "new_lineage" in decision_counts and int(decision_counts["new_lineage"]) > 0:
            linked = manifest.get("linked_transfer_transaction_id", "")
            import_rows.append(
                {
                    "import_id": tx_id,
                    "created_at": created_at,
                    "source_env_fingerprint": "",
                    "source_package_fingerprint": "",
                    "source_session_id": str(linked),
                    "matched_local_session_id": "",
                    "sync_lineage_id": "",
                    "decision": "new_lineage",
                    "execute_mode": "execute",
                    "notes": "Replayed from transaction manifest.",
                }
            )
            added_imports += 1
        existing_import_ids.add(tx_id)

    if import_rows:
        write_jsonl(sidecar_path(codex_home, "imports.jsonl"), import_rows)
    if fork_rows:
        write_jsonl(sidecar_path(codex_home, "forks.jsonl"), fork_rows)

    return {
        "mode": "replay-transactions-sidecar",
        "transactions_considered": len(manifests),
        "imports_added": added_imports,
        "forks_added": added_forks,
        "include_rolled_back": include_rolled_back,
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)

    inspect_cmd = sub.add_parser("inspect-package")
    inspect_cmd.add_argument("package_root")
    inspect_cmd.add_argument("--codex-home", required=True)

    sync_cmd = sub.add_parser("sync")
    sync_cmd.add_argument("package_root")
    sync_cmd.add_argument("--codex-home", required=True)
    sync_cmd.add_argument("--remote-home", required=True)
    sync_cmd.add_argument("--execute", action="store_true")

    adopt_cmd = sub.add_parser("adopt")
    adopt_cmd.add_argument("--local-session-id", required=True)
    adopt_cmd.add_argument("--codex-home", required=True)
    adopt_cmd.add_argument("--source-env", default="env_legacy_local")
    adopt_cmd.add_argument("--source-session-id")

    list_cmd = sub.add_parser("list-transactions")
    list_cmd.add_argument("--codex-home", required=True)

    rollback_cmd = sub.add_parser("rollback")
    rollback_cmd.add_argument("--codex-home", required=True)
    rollback_cmd.add_argument("--transaction-id")
    rollback_cmd.add_argument("--latest", action="store_true")
    rollback_cmd.add_argument("--execute", action="store_true")

    prune_cmd = sub.add_parser("prune-invalid-mappings")
    prune_cmd.add_argument("--codex-home", required=True)

    health_cmd = sub.add_parser("sidecar-health")
    health_cmd.add_argument("--codex-home", required=True)

    rebuild_cmd = sub.add_parser("rebuild-sidecar")
    rebuild_cmd.add_argument("--codex-home", required=True)
    rebuild_cmd.add_argument("--source-env", default="env_rebuilt_local")
    rebuild_cmd.add_argument("--local-session-id", action="append", required=True)

    replay_cmd = sub.add_parser("replay-transactions-sidecar")
    replay_cmd.add_argument("--codex-home", required=True)
    replay_cmd.add_argument("--include-rolled-back", action="store_true")

    return parser.parse_args()


def main() -> int:
    args = parse_args()

    if args.command == "inspect-package":
        result = inspect_package(Path(args.package_root).resolve(), Path(args.codex_home).resolve())
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0

    if args.command == "sync":
        result = sync_package(
            Path(args.package_root).resolve(),
            Path(args.codex_home).resolve(),
            str(Path(args.remote_home).resolve()),
            execute=args.execute,
        )
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0

    if args.command == "adopt":
        result = adopt_local_session(
            Path(args.codex_home).resolve(),
            args.local_session_id,
            args.source_env,
            args.source_session_id or args.local_session_id,
            adoption_mode="manual-adopt",
        )
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0

    if args.command == "list-transactions":
        print(json.dumps(list_transactions(Path(args.codex_home).resolve()), ensure_ascii=False, indent=2))
        return 0

    if args.command == "rollback":
        print(
            json.dumps(
                rollback_transaction(
                    Path(args.codex_home).resolve(),
                    args.transaction_id,
                    args.latest,
                    args.execute,
                ),
                ensure_ascii=False,
                indent=2,
            )
        )
        return 0

    if args.command == "prune-invalid-mappings":
        print(json.dumps(prune_invalid_mappings_command(Path(args.codex_home).resolve()), ensure_ascii=False, indent=2))
        return 0

    if args.command == "sidecar-health":
        print(json.dumps(sidecar_health(Path(args.codex_home).resolve()), ensure_ascii=False, indent=2))
        return 0

    if args.command == "rebuild-sidecar":
        print(
            json.dumps(
                rebuild_sidecar_from_sessions(
                    Path(args.codex_home).resolve(),
                    args.local_session_id,
                    args.source_env,
                ),
                ensure_ascii=False,
                indent=2,
            )
        )
        return 0

    if args.command == "replay-transactions-sidecar":
        print(
            json.dumps(
                replay_transactions_to_sidecar(
                    Path(args.codex_home).resolve(),
                    include_rolled_back=args.include_rolled_back,
                ),
                ensure_ascii=False,
                indent=2,
            )
        )
        return 0

    raise SystemExit("Unsupported command.")


if __name__ == "__main__":
    raise SystemExit(main())