import os, sys, json, shutil, subprocess from typing import Optional, Any, Dict, List from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, _ro_connect, capture_result, conversation_in_use from lib_py.verify_session import verify_session_uuid, workspace_key def _holder_pids(paths: List[str]) -> Optional[List[int]]: """PIDs holding any of `paths` open (lsof -t), [] when none does, None when lsof cannot answer (missing, failed, timed out).""" existing = [p for p in paths if p and os.path.exists(p)] if not existing: return [] lsof = shutil.which("lsof") if not lsof: return None try: r = subprocess.run([lsof, "-t"] + existing, capture_output=True, text=True, timeout=10) except Exception: return None pids = sorted({int(t) for t in r.stdout.split() if t.isdigit()}) # lsof exits 1 both when nothing holds the files and on errors. if r.returncode not in (0, 1) or (r.returncode == 1 and not pids and r.stderr.strip()): return None return pids def _process_parents() -> Optional[Dict[int, int]]: """pid -> ppid of every running process (ps), None when ps fails.""" try: out = subprocess.run(["ps", "-eo", "pid=,ppid="], capture_output=True, text=True, timeout=10) except Exception: return None if out.returncode != 0: return None parents: Dict[int, int] = {} for line in out.stdout.splitlines(): parts = line.split() if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit(): parents[int(parts[0])] = int(parts[1]) return parents or None 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": # Issue #40: the cache maps each workspace to the conversation last # opened there, so an id it maps to another workspace is that # workspace's; and an id another row records is never this row's, # whether or not the cache names it for this cwd. here = workspace_key(ctx.cwd) if any(v == uuid and workspace_key(k) != here for k, v in self._last_conversations(ctx).items()): return False if uuid in (ctx.row.get("_sibling_claimed_uuids") or []): return False try: conn = _ro_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 _last_conversations(self, ctx: DiscoveryContext) -> dict: """cache/last_conversations.json: {cwd: id of the conversation last opened there}.""" try: with open(f"{self._cli_root(ctx)}/cache/last_conversations.json") as f: data = json.load(f) return data if isinstance(data, dict) else {} except Exception: return {} def capture_active_session_id(self, ctx: DiscoveryContext) -> dict: """Issue #40: the cache only proposes an id - it holds one id per cwd, rewritten whenever any agy opens a conversation there, so it cannot tell two sessions in one cwd apart. The id is this session's only when its presence lock (or DB) is held open by the row's pane process or one of its descendants. Anything unprovable is not pinned (no cache-only fallback).""" here = workspace_key(ctx.cwd) proposed = sorted({v for k, v in self._last_conversations(ctx).items() if v and workspace_key(k) == here}) if not proposed: return capture_result("no-candidate", reason="last_conversations.json names no conversation for this cwd") if len(proposed) > 1: return capture_result("ambiguous", reason="last_conversations.json maps this cwd to several conversations", candidates=proposed) uuid = proposed[0] if uuid in (ctx.row.get("_sibling_claimed_uuids") or []): return capture_result("sibling-claimed", reason=f"{uuid} is recorded by another session", candidates=[uuid]) if not verify_session_uuid(ctx.cwd, self.name, uuid, row=ctx.row, home_dir=ctx.home_dir, mode="discover"): return capture_result("rejected", reason=f"{uuid} failed verification (missing, older than the session, or empty)", candidates=[uuid]) status, reason = self._owner_check(uuid, ctx) if status != "ok": return capture_result(status, reason=reason, candidates=[uuid]) return capture_result("captured", session_id=uuid, candidates=[uuid]) def _owner_check(self, uuid: str, ctx: DiscoveryContext): """('ok', '') when a holder of the conversation's presence lock or DB is the row's pane process or its descendant; otherwise ('unverifiable-owner' | 'owner-mismatch', reason).""" try: pane_pid = int((ctx.row.get("pane") or {}).get("pid") or 0) except (TypeError, ValueError): pane_pid = 0 if pane_pid <= 1: return "unverifiable-owner", "the row records no pane pid" db = self.artifact_path(uuid, ctx) paths = [f"{self._cli_root(ctx)}/presence/{uuid}.lock", db, f"{db}-wal", f"{db}-shm"] holders = _holder_pids(paths) if holders is None: return "unverifiable-owner", "lsof is unavailable or failed" if not holders: return "unverifiable-owner", f"no process holds {uuid}'s presence lock or DB" if pane_pid in holders: return "ok", "" parents = _process_parents() if parents is None or pane_pid not in parents: return "unverifiable-owner", f"pane process {pane_pid} cannot be found" for pid in holders: seen = set() while pid > 1 and pid not in seen: if pid == pane_pid: return "ok", "" seen.add(pid) pid = parents.get(pid, 0) return "owner-mismatch", f"{uuid} is held by pid(s) {holders}, none in pane process {pane_pid}'s tree" 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 []