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

import os, sys, json, sqlite3, shutil
from typing import Optional, Any
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, conversation_in_use

class AgyAgentAdapter(BaseAgentAdapter):
    @property
    def name(self) -> str:
        return 'agy'

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

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

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

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

    @property
    def startup_dialogs(self) -> list:
        # #24: "Do you trust the contents of this project?" with
        # "> Yes, I trust this folder" preselected - Enter accepts.
        return [
            ('Do you trust the contents of this project|Yes, I trust this folder',
             ['Enter'], 'trust'),
        ]

    @property
    def identity_cache_fields(self) -> tuple:
        return ('conversation_id', 'conversation_db', 'conversation_brain_dir')

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

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

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

    def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
        return f"{ctx.home_dir}/.gemini/antigravity-cli/conversations/{uuid}.db"

    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
        if ctx.mode == "discover":
            lc = f"{ctx.home_dir}/.gemini/antigravity-cli/cache/last_conversations.json"
            cache_match = False
            if os.path.exists(lc):
                try:
                    with open(lc) as f:
                        lc_data = json.load(f)
                    cache_match = (lc_data.get(ctx.cwd) == uuid)
                except Exception:
                    cache_match = False
            if not cache_match:
                if uuid in (ctx.row.get("_sibling_claimed_uuids") or []):
                    return False
        try:
            conn = sqlite3.connect(path)
            r = conn.execute("SELECT count(*) FROM steps").fetchone()
            conn.close()
            if not r or r[0] < 1:
                return False
        except Exception:
            return False
        return True

    def _cli_root(self, ctx: DiscoveryContext) -> str:
        return f"{ctx.home_dir}/.gemini/antigravity-cli"

    def _artifact_paths(self, uuid: str, ctx: DiscoveryContext) -> list:
        """Every on-disk piece of one conversation, DB first: the SQLite
        sidecars, the brain dir, the title annotation and the presence lock."""
        db = self.artifact_path(uuid, ctx)
        root = self._cli_root(ctx)
        return ([db] + [f"{db}{suffix}" for suffix in ("-wal", "-shm", "-journal")]
                + [f"{root}/brain/{uuid}", f"{root}/annotations/{uuid}.pbtxt",
                   f"{root}/presence/{uuid}.lock"])

    def has_artifacts(self, uuid: str, ctx: DiscoveryContext) -> bool:
        # Any piece counts, not only the DB: a purge that was interrupted (or
        # the pre-#10 purge) can leave a WAL, annotation or lock behind with
        # the DB already gone, and those orphans must still be reachable.
        return any(os.path.exists(p) for p in self._artifact_paths(uuid, ctx))

    def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool:
        # The holder may keep the presence lock (or a sidecar) open rather
        # than the DB, and may name the uuid nowhere; the lsof leg of the
        # check must see every path of the conversation.
        return conversation_in_use(uuid, self._artifact_paths(uuid, ctx))

    def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
        # Issue #10: a purge must leave nothing resumable behind. Besides the
        # conversation DB and the brain dir, agy keeps SQLite sidecars (-wal /
        # -shm / -journal, created whenever the DB is opened - the stop path
        # itself does so while capturing the id, and an unflushed WAL can still
        # hold conversation content), the title annotation and a presence lock.
        purged = []
        db = self.artifact_path(uuid, ctx)
        for path in [db] + [f"{db}{suffix}" for suffix in ("-wal", "-shm", "-journal")]:
            if os.path.exists(path):
                os.remove(path)
                purged.append(path)
        root = self._cli_root(ctx)
        brain = f"{root}/brain/{uuid}"
        if os.path.isdir(brain):
            shutil.rmtree(brain, ignore_errors=True)
            purged.append(brain)
        annotation = f"{root}/annotations/{uuid}.pbtxt"
        if os.path.exists(annotation):
            os.remove(annotation)
            purged.append(annotation)
        # The presence lock marks a conversation some process has open; only
        # remove it once nothing references the conversation any more.
        lock = f"{root}/presence/{uuid}.lock"
        if os.path.exists(lock):
            if self.conversation_in_use(uuid, ctx):
                print(f"WARN: presence lock kept, conversation {uuid} is still in use: {lock}", file=sys.stderr)
            else:
                os.remove(lock)
                purged.append(lock)
        return purged

    def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
        """Anything under the agy CLI tree still carrying this conversation id."""
        root = self._cli_root(ctx)
        if not os.path.isdir(root):
            return []
        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))
            # Do not descend into a matched directory; it is reported as a whole.
            dirnames[:] = [d for d in dirnames if uuid not in d]
        return sorted(found)

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

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

    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("~")
        return os.path.exists(f"{home}/.gemini/oauth_creds.json") or os.path.exists(f"{home}/.gemini/antigravity-cli/antigravity-oauth-token")

    def discover(self, ctx: DiscoveryContext) -> list:
        lc = f"{ctx.home_dir}/.gemini/antigravity-cli/cache/last_conversations.json"
        if os.path.exists(lc):
            try:
                with open(lc) as f:
                    cand = json.load(f).get(ctx.workspace)
                if cand and self.verify_artifact(cand, ctx):
                    return [cand]
            except Exception:
                pass
        return []