import os, sys, sqlite3, shutil, subprocess from typing import Optional, Any, List from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, scan_names, db_rows_naming from lib_py.verify_session import workspace_key _PROVIDER_ENV_VARS = ( 'ANTHROPIC_API_KEY', 'OPENAI_API_KEY', 'GEMINI_API_KEY', 'GROQ_API_KEY' ) class OpenCodeAgentAdapter(BaseAgentAdapter): @property def name(self) -> str: return 'opencode' @property def own_key(self) -> str: return 'opencode_session_id_own' @property def ready_tokens(self) -> str: return 'OpenCode|Chat|Ask anything|tab agents|ctrl\\+p|Build auto|commands' @property def exit_key(self) -> str: return '/exit' @property def delegate_agent_key(self) -> str: return 'opencode-cli' @property def identity_cache_fields(self) -> tuple: return ('session_id',) @property def input_prompt(self) -> Optional[str]: return '❯' @property def input_placeholder(self) -> Optional[str]: return '' @property def input_rule_pattern(self) -> Optional[str]: # Kept for lib.sh's readiness corroboration; the input box itself is a # ┃ block (input_block_prefix), which the extractor reads first. return '─{10,}' @property def input_block_prefix(self) -> Optional[str]: return '┃' @property def input_stop_pattern(self) -> Optional[str]: # The input area ends at the right-aligned cwd line (a long run of # blanks after the ┃) or the `Build auto · ` status line. return r'^\s{20,}\S|^\s*(Build|Plan)\b.*·' def _bin(self, ctx: DiscoveryContext) -> Optional[str]: bin_path = shutil.which("opencode") if bin_path: return bin_path for candidate_bin in [ f"{ctx.home_dir}/.opencode/bin/opencode", os.path.expanduser("~/.opencode/bin/opencode"), ]: if os.path.exists(candidate_bin): return candidate_bin return None def _run(self, ctx: DiscoveryContext, args: list, timeout: int): """Run the opencode CLI against ctx's home (its data dir follows HOME and XDG_DATA_HOME); None when there is no binary.""" bin_path = self._bin(ctx) if not bin_path: return None env = dict(os.environ) if ctx.home_dir: env["HOME"] = ctx.home_dir env["HOME_DIR"] = ctx.home_dir env["XDG_DATA_HOME"] = os.path.join(ctx.home_dir, ".local/share") cwd = ctx.workspace or ctx.cwd or os.getcwd() return subprocess.run([bin_path] + args, capture_output=True, text=True, timeout=timeout, env=env, cwd=cwd if os.path.exists(cwd) else None) def _resolve_db(self, ctx: DiscoveryContext) -> Optional[str]: # opencode 2.x prints its paths with `debug paths` (the 1.x `db path` # is gone: it prints the help and exits 0). Fallback: the default. try: res = self._run(ctx, ["debug", "paths"], timeout=5) if res is not None and res.returncode == 0: for line in res.stdout.splitlines(): parts = line.split(None, 1) if len(parts) == 2 and parts[0] == "db": return parts[1].strip() except Exception: pass return f"{ctx.home_dir}/.local/share/opencode/opencode.db" def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str: db = self._resolve_db(ctx) return db or f"{ctx.home_dir}/.local/share/opencode/opencode.db" def verify_artifact(self, uuid: str, ctx: DiscoveryContext) -> bool: db = self._resolve_db(ctx) if not db or not os.path.exists(db): return False try: # opencode 2.x keeps sessions in session_v2 (a 1.x `session` table # survives a migration but is no longer written). conn = sqlite3.connect(db) r = conn.execute("SELECT directory, time_created FROM session_v2 WHERE id=?", (uuid,)).fetchone() conn.close() if not r: return False found_cwd, started_at = r if ctx.epoch and started_at is not None: try: if float(started_at) / 1000.0 < float(ctx.epoch): return False except (ValueError, TypeError): pass if found_cwd and ctx.cwd and workspace_key(found_cwd) != workspace_key(ctx.cwd): return False return True except Exception: return False def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list: purged = [] db = self._resolve_db(ctx) if db and os.path.exists(db) and self._has_session(db, uuid): # opencode deletes a 2.x session with its child sessions and every # row it keeps for them; --standalone so no background service is # started for it. try: res = self._run(ctx, ["session", "delete", "--standalone", "--", uuid], timeout=60) if res is None: sys.stderr.write(f"WARN: purge opencode: no opencode binary to delete session {uuid}\n") elif res.returncode == 0: purged.append(f"opencode session (with its child sessions): {uuid}") else: sys.stderr.write(f"WARN: purge opencode: session delete {uuid} exited {res.returncode}: " f"{(res.stderr or res.stdout).strip()}\n") except Exception as e: sys.stderr.write(f"WARN: purge opencode: session delete {uuid} failed: {e}\n") if db and os.path.exists(db): # Rows a 1.x opencode left in the tables a migration keeps. try: conn = sqlite3.connect(db) cursor = conn.cursor() cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = {row[0] for row in cursor.fetchall()} if "message" in tables: conn.execute("DELETE FROM message WHERE session_id=?", (uuid,)) if "messages" in tables: conn.execute("DELETE FROM messages WHERE session_id=?", (uuid,)) if "part" in tables: conn.execute("DELETE FROM part WHERE session_id=?", (uuid,)) if "parts" in tables: conn.execute("DELETE FROM parts WHERE session_id=?", (uuid,)) if "session" in tables: conn.execute("DELETE FROM session WHERE id=?", (uuid,)) if "sessions" in tables: conn.execute("DELETE FROM sessions WHERE id=?", (uuid,)) conn.commit() conn.close() if tables & {"message", "messages", "part", "parts", "session", "sessions"}: purged.append(f"sqlite rows for session: {uuid}") except Exception as e: sys.stderr.write(f"WARN: purge opencode db records failed: {e}\n") return purged def _has_session(self, db: str, uuid: str) -> bool: try: conn = sqlite3.connect(db) try: return conn.execute("SELECT 1 FROM session_v2 WHERE id=?", (uuid,)).fetchone() is not None finally: conn.close() except Exception: return False def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list: # Issue #31: whatever still names the session (event-sourced rows a # 1.x opencode left, a failed `session delete`) is reported. found = scan_names(f"{ctx.home_dir}/.local/share/opencode", uuid) return found + db_rows_naming(self._resolve_db(ctx), uuid, ("session_id", "aggregate_id"), id_tables=(("session", "id"), ("session_v2", "id"))) # opencode 2.x: the default agent is build (`--agent` is gone from the # TUI), and --standalone keeps the server a child of the pane's TUI # instead of the per-user background service. def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str: return f"{binary} --auto --standalone" def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str: if materialized and session_uuid: return f"{binary} --session {session_uuid} --auto --standalone" return f"{binary} --auto --standalone" def auth_ok(self, run_cmd: Optional[Any] = None) -> bool: if any(os.environ.get(v) for v in _PROVIDER_ENV_VARS): return True home = os.environ.get("HOME_DIR") or os.environ.get("HOME") or os.path.expanduser("~") auth_file = f"{home}/.local/share/opencode/auth.json" return os.path.exists(auth_file) def discover(self, ctx: DiscoveryContext) -> list: db = self._resolve_db(ctx) if not db or not os.path.exists(db): return [] try: # Top-level sessions only: a subagent's session (parent_id set) # shares the directory and is newer than the one the TUI shows. query = "SELECT id FROM session_v2 WHERE parent_id IS NULL" params = [] if ctx.workspace: query += " AND directory = ?" params.append(ctx.workspace) if ctx.epoch: query += " AND time_created >= ?" params.append(int(float(ctx.epoch) * 1000)) query += " ORDER BY time_created DESC LIMIT 20" conn = sqlite3.connect(db) rows = conn.execute(query, params).fetchall() conn.close() candidates = [] for (cand,) in rows: if cand and self.verify_artifact(cand, ctx): candidates.append(cand) return candidates except Exception: return []