multi-agent-mux/.agents/skills/lib_py/agents/base.py

# base.py — BaseAgentAdapter abstract base class, SpawnSpec, and DiscoveryContext

import os, json, sqlite3, shutil, subprocess
from typing import Optional, Dict, Any, List
from urllib.parse import quote
from lib_py.paths import resolve_home
from lib_py.verify_session import verify_session_uuid, workspace_key


# Session-id variables the agent CLIs export to their tool shells (the list
# orc_onboard.sh's auto-detection reads).
AGENT_SESSION_ENV = ("CLAUDE_CODE_SESSION_ID", "ANTIGRAVITY_CONVERSATION_ID",
                     "HERMES_SESSION_ID", "GROK_SESSION_ID", "OPENCODE_SESSION_ID")
# Process basenames that are agent CLIs (orc_onboard.sh detect_nearest_agent).
AGENT_CLI_NAMES = ("claude", "agy", "hermes", "grok", "opencode", "codex")


def _is_agent_cli(command: str) -> bool:
    """An agent CLI process: some token before the first option has an agent
    basename (`claude …`, `node …/claude …`, `/bin/sh …/agy …`) - the same
    reading orc_onboard.sh's detect_nearest_agent applies."""
    for tok in command.split():
        if tok.startswith("-"):
            break
        if os.path.basename(tok) in AGENT_CLI_NAMES:
            return True
    return False


def conversation_in_use(uuid: str, paths=()) -> bool:
    """A live process still references the conversation (Issues #10, #13).

    A purge must not touch a conversation that is in use, which is any of:

    - the caller runs inside that conversation's own agent session: one of its
      session-id variables (AGENT_SESSION_ENV) equals the uuid. An orchestrator
      purging itself from its own tool shell hits exactly this, and claude
      keeps no transcript open, so neither check below would see it;
    - a process outside this purge's own helper chain names the uuid on its
      command line (`claude -r <uuid>`, `agy --conversation <uuid>`, ...). The
      helper chain - this python and the shells/scripts that launched it -
      names the uuid too and is excluded, but only up to the nearest agent CLI
      ancestor: that ancestor is the session the purge runs in, and if its
      command line carries the uuid it is the conversation's live process;
    - one of `paths` is held open (lsof, where available).
    """
    for var in AGENT_SESSION_ENV:
        if os.environ.get(var, "").strip() == uuid:
            return True
    try:
        out = subprocess.run(["ps", "-eo", "pid=,ppid=,args="],
                             capture_output=True, text=True, timeout=10).stdout
    except Exception:
        out = ""
    parents: Dict[int, int] = {}
    commands: Dict[int, str] = {}
    holders: List[int] = []
    for line in out.splitlines():
        parts = line.split(None, 2)
        if len(parts) < 2 or not (parts[0].isdigit() and parts[1].isdigit()):
            continue
        pid, ppid = int(parts[0]), int(parts[1])
        parents[pid] = ppid
        commands[pid] = parts[2] if len(parts) > 2 else ""
        if uuid in commands[pid]:
            holders.append(pid)
    if holders:
        me = os.getpid()
        helper_chain = set()
        p = me
        while p and p not in helper_chain:
            if p != me and _is_agent_cli(commands.get(p, "")):
                break
            helper_chain.add(p)
            p = parents.get(p, 0)
        if any(h not in helper_chain for h in holders):
            return True
    existing = [p for p in paths if p and os.path.exists(p)]
    if existing and shutil.which("lsof"):
        try:
            r = subprocess.run(["lsof", "-t"] + existing, capture_output=True, text=True, timeout=10)
            if r.stdout.strip():
                return True
        except Exception:
            pass
    return False


def _plausible_id(uuid: str) -> bool:
    """An id safe to build a path from and to search for: not empty, not
    made of dots alone ("." names the parent dir itself), no "/", ".." or "*"."""
    return isinstance(uuid, str) and bool(uuid.strip(".")) and not any(s in uuid for s in ("/", "..", "*"))


def scan_names(root: str, uuid: str, skip=()) -> list:
    """Paths under `root` whose name carries `uuid` (Issue #31). A matched
    directory is reported whole and not descended; `skip` dirs are not walked."""
    if not _plausible_id(uuid) or not os.path.isdir(root):
        return []
    skip_real = {os.path.realpath(s) for s in skip if s}
    found = []
    for dirpath, dirnames, filenames in os.walk(root):
        for name in dirnames + filenames:
            if uuid in name:
                found.append(os.path.join(dirpath, name))
        dirnames[:] = [d for d in dirnames if uuid not in d
                       and not (skip_real and os.path.realpath(os.path.join(dirpath, d)) in skip_real)]
    return sorted(found)


def db_rows_naming(db: str, uuid: str, columns=(), id_tables=()) -> list:
    """"<db> (<table>.<col>: <n> rows)" for every table holding rows of this
    conversation in one of `columns`, or in the id column of an `id_tables`
    (table, column) pair (Issue #31). Read-only; unreadable tables are skipped."""
    if not _plausible_id(uuid) or not db or not os.path.isfile(db):
        return []
    try:
        conn = sqlite3.connect(f"file:{quote(os.path.abspath(db))}?mode=ro", uri=True)
    except Exception:
        return []
    found = []
    try:
        tables = [r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")]
        for table in tables:
            qt = table.replace('"', '""')
            try:
                cols = {r[1] for r in conn.execute(f'PRAGMA table_info("{qt}")')}
            except Exception:
                continue
            wanted = [c for c in columns if c in cols]
            wanted += [c for t, c in id_tables if t == table and c in cols and c not in wanted]
            for col in wanted:
                try:
                    n = conn.execute(f'SELECT count(*) FROM "{qt}" WHERE "{col}"=?', (uuid,)).fetchone()[0]
                except Exception:
                    continue
                if n:
                    found.append(f"{db} ({table}.{col}: {n} rows)")
    except Exception:
        pass
    finally:
        conn.close()
    return found

class SpawnSpec:
    def __init__(self, agent_name: str, session_name: str, command: str, env: Optional[Dict[str, str]] = None):
        self.agent_name = agent_name
        self.session_name = session_name
        self.command = command
        self.env = env or {}

class DiscoveryContext:
    def __init__(self, workspace: str, agent_name: str = "", home_dir: Optional[str] = None,
                 claude_dir: Optional[str] = None, epoch: int = 0,
                 row: Optional[Dict[str, Any]] = None, mode: str = "discover"):
        self.workspace = workspace
        self.agent_name = agent_name
        self.home_dir = resolve_home(home_dir)
        # projects/ lives under claude's config root, which CLAUDE_CONFIG_DIR moves.
        config_root = os.environ.get("CLAUDE_CONFIG_DIR") or f"{self.home_dir}/.claude"
        self.claude_dir = claude_dir or os.environ.get("CLAUDE_PROJECT_DIR", f"{config_root}/projects")
        self.epoch = epoch
        self.row = row or {}
        self.mode = mode

    @property
    def ws_key(self) -> str:
        return workspace_key(self.workspace)

    @property
    def cwd(self) -> str:
        return (self.row.get("pane", {}).get("cwd", "") if isinstance(self.row, dict) else "") or self.workspace

class BaseAgentAdapter:
    @property
    def name(self) -> str:
        raise NotImplementedError

    @property
    def own_key(self) -> str:
        raise NotImplementedError

    @property
    def ready_tokens(self) -> str:
        raise NotImplementedError

    @property
    def exit_key(self) -> str:
        raise NotImplementedError

    @property
    def delegate_agent_key(self) -> str:
        raise NotImplementedError

    @property
    def identity_cache_fields(self) -> tuple:
        """agent_identities 캐시의 에이전트별 필드명.
        B-10(Option A)로 캐시 읽기 경로가 제거되어 현재 생산 소비자는 0건이지만,
        캐시 쓰기 경로가 도입되면 즉시 필요한 유일한 스키마 기술이므로 존치한다.
        임의 삭제 금지 — 삭제 시 4개 어댑터에 필드명을 다시 흩뿌려야 한다."""
        raise NotImplementedError

    @property
    def input_prompt(self) -> Optional[str]:
        return None

    @property
    def input_placeholder(self) -> Optional[str]:
        return None

    @property
    def input_placeholder_pattern(self) -> Optional[str]:
        """Regex the input box's first line matches in full while it shows
        only a placeholder whose text varies (#60); input_placeholder stays
        the literal lib.sh greps for."""
        return None

    @property
    def input_rule_pattern(self) -> Optional[str]:
        return None

    @property
    def input_frame_chars(self) -> str:
        """Box-border characters stripped from both ends of an input line (#30)."""
        return ''

    @property
    def input_block_prefix(self) -> Optional[str]:
        """Line prefix of an input box drawn as a block instead of between rules (#30)."""
        return None

    @property
    def input_stop_pattern(self) -> Optional[str]:
        """Regex of the status line that ends the input area (#30): inside a
        block, the first matching line; below a prompt line (no rules), only
        the last non-blank line of the pane."""
        return None

    @property
    def resident_children(self) -> tuple:
        """Process names this CLI keeps as children while idle (#30). The stall
        watcher does not count them as background work."""
        return ()

    @property
    def agent_status_reliable(self) -> bool:
        """Whether herdr's agent_status for this CLI can drive the idle judgement
        (#30). On herdr 0.9.1 claude, agy, grok, opencode and codex all report
        working/idle transitions; a False here turns the idle nudge off."""
        return True

    @property
    def modal_tokens(self) -> Optional[str]:
        return None

    @property
    def startup_dialogs(self) -> List[tuple]:
        """Blocking modals this CLI can show before its TUI is usable, most
        specific first: (pattern, keys, kind).

        `pattern` is an extended-regex signature matched against the pane tail,
        `keys` are herdr send-keys arguments sent in order, and `kind` is
        'trust' (a directory-trust prompt, gated by --auto-trust) or 'notice'
        (anything else that blocks the start, always answered).

        Issue #24: without this the start-up trust prompts of agy, grok and
        codex burn the whole readiness timeout and onboarding is skipped.
        """
        return []

    @property
    def strong_ready_tokens(self) -> str:
        return self.ready_tokens

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

    def derive_session_name(self, slug: str, role: str = "creator") -> str:
        r = role.lower()
        return f"{slug}-{r}-{self.name}"

    def matches_session_name(self, session_name: str) -> bool:
        for r in ('creator', 'planner', 'reviewer'):
            if session_name.endswith(f"-{r}-{self.name}"):
                return True
        return session_name.endswith(f"-{self.name}")

    def verify_session(self, ws: str, uuid: str, row: Optional[Dict[str, Any]] = None, mode: str = "discover") -> bool:
        return verify_session_uuid(ws, self.name, uuid, row=row, mode=mode)

    def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
        raise NotImplementedError

    def verify_artifact(self, uuid: str, ctx: DiscoveryContext) -> bool:
        raise NotImplementedError

    def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
        raise NotImplementedError

    def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
        """Paths still carrying `uuid` after a purge. Adapters that know their
        on-disk tree override this; the default has nothing to scan."""
        return []

    def has_artifacts(self, uuid: str, ctx: DiscoveryContext) -> bool:
        """Anything of this conversation is on disk (a purge target exists
        even when `verify_artifact` would not vouch for it). The default only
        trusts an artifact path that embeds the uuid - a shared store (one DB
        for every conversation) exists for any uuid and proves nothing."""
        path = self.artifact_path(uuid, ctx)
        return uuid in path and os.path.exists(path)

    def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool:
        """True while a live process still references the conversation."""
        return conversation_in_use(uuid, [self.artifact_path(uuid, ctx)])

    def in_use_hints(self, uuid: str, ctx: DiscoveryContext) -> list:
        """Where to look for what holds the conversation, shown when a purge
        is refused because it is in use."""
        return [f"any process with it on its command line (an editor, a grep, ...): ps -axww | grep {uuid}"]

    def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
        raise NotImplementedError

    def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
        raise NotImplementedError

    def auth_ok(self, run_cmd: Optional[Any] = None) -> bool:
        raise NotImplementedError

    def discover(self, ctx: DiscoveryContext) -> list:
        raise NotImplementedError