Godopu
76f10b2096
- Implement CodexAdapter in lib_py/agents/adapters/codex.py with safe flags and TUI readiness - Extend create_session, resume_session, stop_session, and reconcile for codex CLI - Support codex agent delegation and loop integration across all multi-agent skills - Add unit, component, and contract test suites for codex integration - Verify all 462 tests passing cleanly
60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
# registry.py — Static agent adapter registry and resolution functions (N4 & N5)
|
|
|
|
from typing import Optional, Dict, Any
|
|
from lib_py.agents.base import BaseAgentAdapter
|
|
from lib_py.agents.adapters.claude import ClaudeAgentAdapter
|
|
from lib_py.agents.adapters.agy import AgyAgentAdapter
|
|
from lib_py.agents.adapters.hermes import HermesAgentAdapter
|
|
from lib_py.agents.adapters.grok import GrokAgentAdapter
|
|
from lib_py.agents.adapters.opencode import OpenCodeAgentAdapter
|
|
from lib_py.agents.adapters.codex import CodexAgentAdapter
|
|
|
|
_ADAPTERS: Dict[str, BaseAgentAdapter] = {
|
|
'claude': ClaudeAgentAdapter(),
|
|
'agy': AgyAgentAdapter(),
|
|
'hermes': HermesAgentAdapter(),
|
|
'grok': GrokAgentAdapter(),
|
|
'opencode': OpenCodeAgentAdapter(),
|
|
'codex': CodexAgentAdapter(),
|
|
}
|
|
|
|
def get_adapter(agent_name: str) -> Optional[BaseAgentAdapter]:
|
|
if not agent_name:
|
|
return None
|
|
return _ADAPTERS.get(agent_name.lower())
|
|
|
|
def own_key(agent_name: str) -> Optional[str]:
|
|
adapter = get_adapter(agent_name)
|
|
return adapter.own_key if adapter else None
|
|
|
|
def agent_of_row(row: Dict[str, Any], session_name: str = "", match_cmd: bool = True) -> Optional[str]:
|
|
"""
|
|
Resolves agent for a given registry row or session name using strict priority order (N5):
|
|
1. Explicit 'agent' field in row dictionary
|
|
2. Session name suffix matching (*-{creator,planner,reviewer}-<agent> or *-<agent>)
|
|
3. pane.cmd exact match (if match_cmd is True, e.g. for non-adoption lookup)
|
|
4. None (resolution failure)
|
|
"""
|
|
if isinstance(row, dict):
|
|
explicit = row.get('agent')
|
|
if explicit and str(explicit).lower() in _ADAPTERS:
|
|
return str(explicit).lower()
|
|
if not session_name:
|
|
session_name = row.get('name', '')
|
|
|
|
name = session_name or (row.get('name', '') if isinstance(row, dict) else '')
|
|
if name:
|
|
for agent_name, adapter in _ADAPTERS.items():
|
|
if adapter.matches_session_name(name):
|
|
return agent_name
|
|
|
|
if match_cmd and isinstance(row, dict):
|
|
pane = row.get('pane') or {}
|
|
cmd = pane.get('cmd', '') if isinstance(pane, dict) else ''
|
|
if cmd:
|
|
for agent_name in _ADAPTERS:
|
|
if cmd == agent_name or cmd.endswith(f"/{agent_name}"):
|
|
return agent_name
|
|
|
|
return None
|