Godopu
b4821fafa8
- Option B / C-3b: Deprecate isolation.root and remove its 4 legacy consumers in lib.sh, workspace_uuid.py, verify_session.py, and stop_session.sh. - M2~M7 Migration: Implement artifact_path, verify_artifact, purge_artifacts, spawn_spec, resume_spec, auth_ok, discover, ready_tokens, exit_key, and delegate_agent_key across BaseAgentAdapter and all 4 adapters (Claude, Agy, Hermes, Cline). - Migrate shell duplications in lib.sh, create_session.sh, resume_session.sh, stop_session.sh, and reconcile.sh to python -m lib_py.agents facts. - Hardening & Corrections: Fix Cline resume_spec flag to '-i --id', apply shlex.quote across all facts variables with MAM_ prefix, add safe fallback in wait_for_tui_ready. - Add comprehensive contract tests in tests/test_a4_adapter_contract.py and sync IMPROVEMENTS.md / LOG.md. - Verified: 100% PASS across unit, component, contract, and integration tests.
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
import os, sys, sqlite3
|
|
from typing import Optional, Any
|
|
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext
|
|
from lib_py.verify_session import workspace_key
|
|
|
|
class HermesAgentAdapter(BaseAgentAdapter):
|
|
@property
|
|
def name(self) -> str:
|
|
return 'hermes'
|
|
|
|
@property
|
|
def own_key(self) -> str:
|
|
return 'hermes_conversation_id_own'
|
|
|
|
@property
|
|
def ready_tokens(self) -> str:
|
|
return 'Hermes'
|
|
|
|
@property
|
|
def exit_key(self) -> str:
|
|
return '/exit'
|
|
|
|
@property
|
|
def delegate_agent_key(self) -> str:
|
|
return 'hermes-agent'
|
|
|
|
@property
|
|
def identity_cache_fields(self) -> tuple:
|
|
return ('session_id',)
|
|
|
|
def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
|
|
return f"{ctx.home_dir}/.hermes/sessions/session_{uuid}.json"
|
|
|
|
def verify_artifact(self, uuid: str, ctx: DiscoveryContext) -> bool:
|
|
hdb = f"{ctx.home_dir}/.hermes/state.db"
|
|
if not os.path.exists(hdb):
|
|
return False
|
|
if ctx.epoch and os.path.getmtime(hdb) < ctx.epoch:
|
|
return False
|
|
try:
|
|
conn = sqlite3.connect(hdb)
|
|
r = conn.execute("SELECT cwd FROM sessions WHERE id=?", (uuid,)).fetchone()
|
|
conn.close()
|
|
if not r:
|
|
return False
|
|
found_cwd = r[0]
|
|
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 = []
|
|
json_file = self.artifact_path(uuid, ctx)
|
|
if os.path.exists(json_file):
|
|
os.remove(json_file)
|
|
purged.append(json_file)
|
|
hdb = f"{ctx.home_dir}/.hermes/state.db"
|
|
if os.path.exists(hdb):
|
|
try:
|
|
conn = sqlite3.connect(hdb)
|
|
conn.execute("DELETE FROM sessions WHERE id=?", (uuid,))
|
|
conn.execute("DELETE FROM messages WHERE session_id=?", (uuid,))
|
|
conn.commit()
|
|
conn.close()
|
|
purged.append(f"db records for session: {uuid}")
|
|
except Exception as e:
|
|
sys.stderr.write(f"WARN: purge hermes db records failed: {e}\n")
|
|
return purged
|
|
|
|
def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
|
|
return binary
|
|
|
|
def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
|
|
if materialized and session_uuid:
|
|
return f"{binary} --resume {session_uuid}"
|
|
return binary
|
|
|
|
def auth_ok(self, run_cmd: Optional[Any] = None) -> bool:
|
|
return True
|
|
|
|
def discover(self, ctx: DiscoveryContext) -> list:
|
|
hdb = f"{ctx.home_dir}/.hermes/state.db"
|
|
if os.path.exists(hdb):
|
|
try:
|
|
conn = sqlite3.connect(hdb)
|
|
r = conn.execute("SELECT id FROM sessions WHERE cwd=? ORDER BY started_at DESC LIMIT 1", (ctx.workspace,)).fetchone()
|
|
conn.close()
|
|
if r:
|
|
cand = r[0]
|
|
if cand and self.verify_artifact(cand, ctx):
|
|
return [cand]
|
|
except Exception:
|
|
pass
|
|
return []
|