import os, sys, sqlite3, shutil from typing import Optional, Any, List from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext 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' @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]: return '─{10,}' def _resolve_db(self, ctx: DiscoveryContext) -> Optional[str]: # 1. Try opencode CLI "opencode db path" try: import subprocess bin_path = shutil.which("opencode") if not 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): bin_path = candidate_bin break if bin_path: 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() res = subprocess.run( [bin_path, "db", "path"], capture_output=True, text=True, timeout=2, env=env, cwd=cwd if os.path.exists(cwd) else None ) if res.returncode == 0: out = res.stdout.strip() if out and os.path.exists(out): return out except Exception: pass # 2. Filesystem fallbacks base_dir = f"{ctx.home_dir}/.local/share/opencode" candidates = [ f"{base_dir}/opencode.db", f"{base_dir}/storage.db", ] for cand in candidates: if os.path.exists(cand): return cand if os.path.isdir(f"{base_dir}/project"): for root, _, files in os.walk(f"{base_dir}/project"): for f in files: if f.endswith(".db"): return os.path.join(root, f) return candidates[0] 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: conn = sqlite3.connect(db) cursor = conn.cursor() cursor.execute("PRAGMA table_info(session)") cols = [col[1] for col in cursor.fetchall()] if not cols: conn.close() return False cwd_col = "directory" if "directory" in cols else ("cwd" if "cwd" in cols else None) time_col = "time_created" if "time_created" in cols else ("created_at" if "created_at" in cols else ("started_at" if "started_at" in cols else None)) query_cols = [] if cwd_col: query_cols.append(cwd_col) if time_col: query_cols.append(time_col) query = "SELECT " + (", ".join(query_cols) if query_cols else "id") + " FROM session WHERE id=?" r = cursor.execute(query, (uuid,)).fetchone() conn.close() if not r: return False found_cwd = None started_at = None idx = 0 if cwd_col: found_cwd = r[idx] idx += 1 if time_col: started_at = r[idx] idx += 1 if ctx.epoch and started_at is not None: try: val = float(started_at) if time_col == "time_created" or val > 1e11: val = val / 1000.0 if val < 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): 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() 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 spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str: return f"{binary} --auto --agent build" 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 --agent build" return f"{binary} --auto --agent build" 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: conn = sqlite3.connect(db) cursor = conn.cursor() cursor.execute("PRAGMA table_info(session)") cols = [col[1] for col in cursor.fetchall()] if not cols: conn.close() return [] cwd_col = "directory" if "directory" in cols else ("cwd" if "cwd" in cols else None) time_col = "time_created" if "time_created" in cols else ("created_at" if "created_at" in cols else ("started_at" if "started_at" in cols else None)) query = "SELECT id FROM session" params = [] where_clauses = [] if cwd_col and ctx.workspace: where_clauses.append(f"{cwd_col} = ?") params.append(ctx.workspace) if time_col and ctx.epoch: if time_col == "time_created": where_clauses.append(f"{time_col} >= ?") params.append(int(float(ctx.epoch) * 1000)) else: where_clauses.append(f"{time_col} >= ?") params.append(ctx.epoch) if where_clauses: query += " WHERE " + " AND ".join(where_clauses) if time_col: query += f" ORDER BY {time_col} DESC" query += " LIMIT 20" rows = cursor.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 []