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

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

class ClaudeAgentAdapter(BaseAgentAdapter):
    @property
    def name(self) -> str:
        return 'claude'

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

    @property
    def ready_tokens(self) -> str:
        return 'Anthropic|Assistant|Chat|Welcome|Claude Code|Opus|Sonnet|Haiku'

    @property
    def exit_key(self) -> str:
        return '/exit'

    @property
    def delegate_agent_key(self) -> str:
        return 'claude-code'

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

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

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

    @property
    def input_rule_pattern(self) -> str:
        return '─{10,}'

    @property
    def modal_tokens(self) -> Optional[str]:
        return 'Try the new fullscreen renderer\\?'

    @property
    def startup_dialogs(self) -> list:
        # The ladder handle_startup_dialogs used to hardcode, unchanged.
        return [
            ('Do you trust the files|Yes, I trust this folder|Quick safety check', ['Enter'], 'trust'),
            # Fullscreen-renderer upsell, not the idle /tui tip: Enter would
            # accept and restart the session without the permission flags.
            ('Yes, try it', ['Escape'], 'notice'),
            ('Yes, proceed', ['Down', 'Enter'], 'trust'),
            ('Resuming the full session', ['Enter'], 'notice'),
            ('Press Enter to continue', ['Enter'], 'notice'),
        ]

    def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
        return f"{ctx.claude_dir}/{ctx.ws_key}/{uuid}.jsonl"

    def verify_artifact(self, uuid: str, ctx: DiscoveryContext) -> bool:
        path = self.artifact_path(uuid, ctx)
        if not os.path.exists(path):
            return False
        if ctx.epoch and os.path.getmtime(path) < ctx.epoch:
            return False
        try:
            valid_session = False
            found_cwd = None
            with open(path) as f:
                for _ in range(50):
                    line = f.readline()
                    if not line:
                        break
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        payload = json.loads(line)
                        if payload.get("sessionId") == uuid:
                            valid_session = True
                        if payload.get("cwd"):
                            found_cwd = payload.get("cwd")
                            break
                    except Exception:
                        pass
            if not valid_session:
                return False
            if found_cwd and workspace_key(found_cwd) != workspace_key(ctx.cwd):
                return False
        except Exception:
            return False
        return True

    def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
        jsonl = self.artifact_path(uuid, ctx)
        if os.path.exists(jsonl):
            os.remove(jsonl)
            return [jsonl]
        return []

    def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
        if use_wrapper or not session_uuid:
            return f"{binary} --dangerously-skip-permissions"
        return f"{binary} --dangerously-skip-permissions --session-id {session_uuid}"

    def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
        if materialized and session_uuid:
            return f"{binary} --dangerously-skip-permissions -r {session_uuid}"
        elif session_uuid:
            return f"{binary} --dangerously-skip-permissions --session-id {session_uuid}"
        return f"{binary} --dangerously-skip-permissions"

    def auth_ok(self, run_cmd: Optional[Any] = None) -> bool:
        if run_cmd is not None:
            try:
                rc, stdout, stderr = run_cmd(['claude', 'auth', 'status'])
                return rc == 0 and ('"loggedIn":true' in (stdout or '').replace(' ', ''))
            except Exception:
                return False
        try:
            res = subprocess.run(['claude', 'auth', 'status'], capture_output=True, text=True)
            return res.returncode == 0 and ('"loggedIn":true' in (res.stdout or '').replace(' ', ''))
        except Exception:
            return False

    def discover(self, ctx: DiscoveryContext) -> list:
        key = workspace_key(ctx.workspace)
        proj = f"{ctx.claude_dir}/{key}"
        candidates = []
        if os.path.isdir(proj):
            for j in sorted(glob.glob(f"{proj}/*.jsonl"), key=os.path.getmtime, reverse=True):
                cand = os.path.basename(j)[:-6]
                if cand and self.verify_artifact(cand, ctx):
                    candidates.append(cand)
        return candidates