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

import os, json, glob, shutil
from typing import Optional, Any
from urllib.parse import quote
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext
from lib_py.verify_session import workspace_key


class GrokAgentAdapter(BaseAgentAdapter):
    @property
    def name(self) -> str:
        return 'grok'

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

    @property
    def ready_tokens(self) -> str:
        return 'Grok|xAI|Assistant|❯|>>>'

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

    @property
    def delegate_agent_key(self) -> str:
        return 'grok-build'

    @property
    def startup_dialogs(self) -> list:
        # #24: "Do you trust the contents of this directory?" answered with y.
        return [
            ('Do you trust the contents of this directory', ['y'], 'trust'),
        ]

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

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

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

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

    def _ws_dir(self, ctx: DiscoveryContext) -> str:
        return quote(os.path.realpath(ctx.cwd), safe='')

    def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
        return f"{ctx.home_dir}/.grok/sessions/{self._ws_dir(ctx)}/{uuid}/chat_history.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("session_id") == uuid or payload.get("sessionId") == uuid:
                            valid_session = True
                        if payload.get("cwd"):
                            found_cwd = payload.get("cwd")
                            break
                    except Exception:
                        pass
            if not valid_session and found_cwd is None:
                # Directory path itself encodes the workspace and session UUID
                valid_session = os.path.isdir(os.path.dirname(path))
            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:
        purged = []
        sess_dir = os.path.dirname(self.artifact_path(uuid, ctx))
        if os.path.isdir(sess_dir):
            shutil.rmtree(sess_dir, ignore_errors=True)
            purged.append(sess_dir)
        elif os.path.exists(self.artifact_path(uuid, ctx)):
            os.remove(self.artifact_path(uuid, ctx))
            purged.append(self.artifact_path(uuid, ctx))
        return purged

    def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
        if session_uuid:
            return f"{binary} --session-id {session_uuid} --permission-mode bypassPermissions"
        return f"{binary} --permission-mode bypassPermissions"

    def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
        if materialized and session_uuid:
            return f"{binary} --resume {session_uuid} --permission-mode bypassPermissions"
        if session_uuid:
            return f"{binary} --session-id {session_uuid} --permission-mode bypassPermissions"
        return f"{binary} --permission-mode bypassPermissions"

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

    def discover(self, ctx: DiscoveryContext) -> list:
        root = f"{ctx.home_dir}/.grok/sessions/{self._ws_dir(ctx)}"
        if not os.path.isdir(root):
            return []
        entries = [d for d in glob.glob(f"{root}/*") if os.path.isdir(d)]
        entries.sort(key=os.path.getmtime, reverse=True)
        candidates = []
        for d in entries:
            cand = os.path.basename(d)
            if cand and self.verify_artifact(cand, ctx):
                candidates.append(cand)
        return candidates