multi-agent-mux/.agents/skills/lib_py/agents/adapters/codex.py

import os, json, glob, sqlite3, subprocess
from typing import Optional, Any, List
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext
from lib_py.verify_session import workspace_key

class CodexAgentAdapter(BaseAgentAdapter):
    @property
    def name(self) -> str:
        return 'codex'

    @property
    def own_key(self) -> str:
        return 'codex_session_id_own'

    @property
    def ready_tokens(self) -> str:
        return 'Codex|Ask Codex to do anything'

    @property
    def exit_key(self) -> str:
        return 'Ctrl+C'

    @property
    def delegate_agent_key(self) -> str:
        return 'codex-cli'

    @property
    def startup_dialogs(self) -> list:
        # #24: the trust prompt lists "1. Yes, continue" preselected, so Enter
        # accepts. Trust here also loads this project's local config, hooks and
        # exec policy, which is why --no-auto-trust exists.
        #
        # The 0.155.1 update notice (observed 2026-09-22) blocks the session the
        # same way; "2" is Skip, the choice that changes nothing.
        return [
            ('Do you trust the contents of this directory|1\\. Yes, continue',
             ['Enter'], 'trust'),
            ('1\\. Update now|Skip until next version', ['2'], 'notice'),
        ]

    @property
    def identity_cache_fields(self) -> tuple:
        return ('session_id',)

    @property
    def input_prompt(self) -> Optional[str]:
        # herdr 0.9.1 capture: one `› Ask Codex to do anything` line, no rules.
        return '›'

    @property
    def input_placeholder(self) -> Optional[str]:
        return 'Ask Codex to do anything'

    @property
    def input_stop_pattern(self) -> Optional[str]:
        # The status footer under the input line: `  <model> <effort> · <cwd> · …`
        # (the ` · ~/` or ` · /` of the cwd keeps a draft line with dots from matching)
        return r'^\s+\S.*\s\u00b7\s[~/]'

    @property
    def resident_children(self) -> tuple:
        # Kept alive while idle (measured 2026-09-25).
        return ('codex-code-mode-host',)

    def _codex_home(self, ctx: DiscoveryContext) -> str:
        return os.environ.get('CODEX_HOME') or f"{ctx.home_dir}/.codex"

    def _state_db(self, ctx: DiscoveryContext) -> Optional[str]:
        matches = glob.glob(f"{self._codex_home(ctx)}/state_*.sqlite")
        if not matches:
            return None
        return max(matches, key=os.path.getmtime)

    def _find_rollout_via_db(self, uuid: str, ctx: DiscoveryContext):
        """Returns (rollout_path, cwd) from the state DB, or (None, None)."""
        db = self._state_db(ctx)
        if not db or not os.path.exists(db):
            return None, None
        try:
            conn = sqlite3.connect(db)
            row = conn.execute("SELECT rollout_path, cwd FROM threads WHERE id=?", (uuid,)).fetchone()
            conn.close()
            if row and row[0] and os.path.exists(row[0]):
                return row[0], row[1]
        except Exception:
            pass
        return None, None

    def _find_rollout(self, uuid: str, ctx: DiscoveryContext) -> Optional[str]:
        path, _ = self._find_rollout_via_db(uuid, ctx)
        if path:
            return path
        matches = glob.glob(f"{self._codex_home(ctx)}/sessions/*/*/*/rollout-*-{uuid}.jsonl")
        return matches[0] if matches else None

    def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
        return self._find_rollout(uuid, ctx) or f"{self._codex_home(ctx)}/sessions/rollout-{uuid}.jsonl"

    def verify_artifact(self, uuid: str, ctx: DiscoveryContext) -> bool:
        db_path, db_cwd = self._find_rollout_via_db(uuid, ctx)
        path = db_path or self._find_rollout(uuid, ctx)
        if not path or not os.path.exists(path):
            return False
        if ctx.epoch and os.path.getmtime(path) < ctx.epoch:
            return False

        # Fail closed on invalid metadata or mismatched id / cwd (R4)
        try:
            with open(path) as f:
                first = f.readline().strip()
            meta = json.loads(first)
            if not isinstance(meta, dict):
                return False
            payload = meta.get('payload')
            if not isinstance(payload, dict):
                return False
            file_id = payload.get('id')
            file_cwd = payload.get('cwd')
            if not file_id or file_id != uuid:
                return False
            if not file_cwd:
                return False
            if db_cwd and workspace_key(db_cwd) != workspace_key(file_cwd):
                return False
            target_cwd = ctx.cwd or ctx.workspace
            if not target_cwd:
                return False
            if workspace_key(file_cwd) != workspace_key(target_cwd):
                return False
            return True
        except Exception:
            return False

    def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> List[str]:
        purged = []
        path = self._find_rollout(uuid, ctx)
        try:
            res = subprocess.run(['codex', 'delete', uuid, '--force'],
                                  capture_output=True, text=True, timeout=5)
            if res.returncode == 0:
                if path:
                    purged.append(path)
                return purged
        except Exception:
            pass
        if path and os.path.exists(path):
            os.remove(path)
            purged.append(path)
        db = self._state_db(ctx)
        if db and os.path.exists(db):
            try:
                conn = sqlite3.connect(db)
                conn.execute("DELETE FROM threads WHERE id=?", (uuid,))
                conn.commit()
                conn.close()
                purged.append(f"state db row for thread: {uuid}")
            except Exception:
                pass
        return purged

    def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
        return f"{binary} --dangerously-bypass-approvals-and-sandbox"

    def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
        if materialized and session_uuid:
            return f"{binary} resume --dangerously-bypass-approvals-and-sandbox {session_uuid}"
        return f"{binary} --dangerously-bypass-approvals-and-sandbox"

    def auth_ok(self, run_cmd: Optional[Any] = None) -> bool:
        home = os.environ.get('HOME_DIR') or os.environ.get('HOME') or os.path.expanduser('~')
        codex_home = os.environ.get('CODEX_HOME') or f"{home}/.codex"
        return os.path.exists(f"{codex_home}/auth.json") or bool(os.environ.get('OPENAI_API_KEY'))

    def discover(self, ctx: DiscoveryContext) -> List[str]:
        target_ws = ctx.workspace or ctx.cwd
        target_key = workspace_key(target_ws) if target_ws else ""

        db = self._state_db(ctx)
        candidates = []
        seen = set()

        if db and os.path.exists(db):
            try:
                conn = sqlite3.connect(db)
                cur = conn.cursor()
                cur.execute("PRAGMA table_info(threads)")
                cols = {r[1] for r in cur.fetchall()}
                order_col = "updated_at" if "updated_at" in cols else ("created_at" if "created_at" in cols else None)
                order_clause = f"ORDER BY {order_col} DESC" if order_col else ""

                # Query database with workspace scope before bounding (R3)
                if "cwd" in cols and target_ws:
                    direct_cwds = list({target_ws, os.path.realpath(target_ws)})
                    placeholders = ",".join("?" * len(direct_cwds))
                    cur.execute(f"SELECT id, cwd FROM threads WHERE cwd IN ({placeholders}) {order_clause} LIMIT 50", direct_cwds)
                    for uuid, cwd in cur.fetchall():
                        if uuid and uuid not in seen and self.verify_artifact(uuid, ctx):
                            candidates.append(uuid)
                            seen.add(uuid)

                # Page through batches if needed to catch symlink aliases / canonical paths
                if len(candidates) < 50:
                    cur.execute(f"SELECT id, cwd FROM threads {order_clause}")
                    while True:
                        batch = cur.fetchmany(100)
                        if not batch:
                            break
                        for uuid, cwd in batch:
                            if uuid in seen:
                                continue
                            if target_key and cwd and workspace_key(cwd) != target_key:
                                continue
                            if uuid and self.verify_artifact(uuid, ctx):
                                candidates.append(uuid)
                                seen.add(uuid)
                                if len(candidates) >= 50:
                                    break
                        if len(candidates) >= 50:
                            break
                conn.close()
            except Exception:
                pass

        # R2: If database yielded usable candidates, return them; otherwise fall through to transcript discovery
        if candidates:
            return candidates

        root = self._codex_home(ctx)
        paths = sorted(glob.glob(f"{root}/sessions/*/*/*/rollout-*.jsonl"), key=os.path.getmtime, reverse=True)
        for path in paths:
            base = os.path.basename(path)
            if base.endswith(".jsonl") and len(base) >= 42:
                uuid = base[:-6][-36:]
                if uuid and uuid not in seen and self.verify_artifact(uuid, ctx):
                    candidates.append(uuid)
                    seen.add(uuid)
                    if len(candidates) >= 50:
                        break
        return candidates