42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
# base.py — BaseAgentAdapter abstract base class, SpawnSpec, and DiscoveryContext
|
|
|
|
import os, json, sqlite3
|
|
from typing import Optional, Dict, Any, List
|
|
from lib_py.paths import resolve_home
|
|
from lib_py.verify_session import verify_session_uuid, workspace_key
|
|
|
|
class SpawnSpec:
|
|
def __init__(self, agent_name: str, session_name: str, command: str, env: Optional[Dict[str, str]] = None):
|
|
self.agent_name = agent_name
|
|
self.session_name = session_name
|
|
self.command = command
|
|
self.env = env or {}
|
|
|
|
class DiscoveryContext:
|
|
def __init__(self, workspace: str, agent_name: str, home_dir: Optional[str] = None):
|
|
self.workspace = workspace
|
|
self.agent_name = agent_name
|
|
self.home_dir = resolve_home(home_dir)
|
|
|
|
class BaseAgentAdapter:
|
|
@property
|
|
def name(self) -> str:
|
|
raise NotImplementedError
|
|
|
|
@property
|
|
def own_key(self) -> str:
|
|
raise NotImplementedError
|
|
|
|
def derive_session_name(self, slug: str, role: str = "creator") -> str:
|
|
r = role.lower()
|
|
return f"{slug}-{r}-{self.name}"
|
|
|
|
def matches_session_name(self, session_name: str) -> bool:
|
|
for r in ('creator', 'planner', 'reviewer'):
|
|
if session_name.endswith(f"-{r}-{self.name}"):
|
|
return True
|
|
return session_name.endswith(f"-{self.name}")
|
|
|
|
def verify_session(self, ws: str, uuid: str, row: Optional[Dict[str, Any]] = None, mode: str = "discover") -> bool:
|
|
return verify_session_uuid(ws, self.name, uuid, row=row, mode=mode)
|