""" Operator control surface for Lazarus Pit v2.0. Provides the command grammar and API for: summon, invite, team, status, close, destroy """ import json import uuid from datetime import datetime, timedelta from typing import Dict, List, Optional from .cell import Cell, CellMember, CellPackager, CellRegistry, CellRole, CellState from .backends.process_backend import ProcessBackend class OperatorCtl: """The operator's command surface for the Lazarus Pit.""" def __init__(self, registry: Optional[CellRegistry] = None, packager: Optional[CellPackager] = None): self.registry = registry or CellRegistry() self.packager = packager or CellPackager() self.backend = ProcessBackend() # ------------------------------------------------------------------ # Commands # ------------------------------------------------------------------ def summon(self, agent: str, project: str, owner: str = "human:Alexander", backend: str = "process", ttl_minutes: int = 60) -> dict: """Summon a single agent into a fresh resurrection cell.""" cell_id = f"laz-{uuid.uuid4().hex[:8]}" cell = self.packager.create(cell_id, backend=backend) cell.project = project cell.owner = owner cell.ttl_minutes = ttl_minutes cell.state = CellState.ACTIVE cell.members.append(CellMember( identity=f"agent:{agent}", role=CellRole.EXECUTOR, joined_at=datetime.now().astimezone().isoformat(), )) # Spawn a simple keep-alive process (real agent bootstrap would go here) res = self.backend.spawn( cell_id=cell_id, hermes_home=cell.hermes_home, command=["python3", "-c", "import time; time.sleep(86400)"], ) self.registry.save(cell) return { "success": res.success, "cell_id": cell_id, "project": project, "agent": agent, "backend": backend, "pid": res.pid, "message": res.message, } def invite(self, cell_id: str, guest: str, role: str, invited_by: str = "human:Alexander") -> dict: """Invite a guest bot or human into an active cell.""" cell = self.registry.load(cell_id) if not cell: return {"success": False, "message": f"Cell {cell_id} not found"} if cell.state not in (CellState.PROPOSED, CellState.ACTIVE, CellState.IDLE): return {"success": False, "message": f"Cell {cell_id} is not accepting invites (state={cell.state.value})"} role_enum = CellRole(role) cell.members.append(CellMember( identity=guest, role=role_enum, joined_at=datetime.now().astimezone().isoformat(), )) cell.state = CellState.ACTIVE self.registry.save(cell) return { "success": True, "cell_id": cell_id, "guest": guest, "role": role_enum.value, "invited_by": invited_by, "message": f"Invited {guest} as {role_enum.value} to {cell_id}", } def team(self, agents: List[str], project: str, owner: str = "human:Alexander", backend: str = "process", ttl_minutes: int = 120) -> dict: """Form a multi-agent team in one cell against a project.""" cell_id = f"laz-{uuid.uuid4().hex[:8]}" cell = self.packager.create(cell_id, backend=backend) cell.project = project cell.owner = owner cell.ttl_minutes = ttl_minutes cell.state = CellState.ACTIVE for agent in agents: cell.members.append(CellMember( identity=f"agent:{agent}", role=CellRole.EXECUTOR, joined_at=datetime.now().astimezone().isoformat(), )) res = self.backend.spawn( cell_id=cell_id, hermes_home=cell.hermes_home, command=["python3", "-c", "import time; time.sleep(86400)"], ) self.registry.save(cell) return { "success": res.success, "cell_id": cell_id, "project": project, "agents": agents, "backend": backend, "pid": res.pid, "message": res.message, } def status(self, cell_id: Optional[str] = None) -> dict: """Show status of one cell or all active cells.""" if cell_id: cell = self.registry.load(cell_id) if not cell: return {"success": False, "message": f"Cell {cell_id} not found"} probe = self.backend.probe(cell_id) return { "success": True, "cell": cell.to_dict(), "probe": probe.message, "expired": cell.is_expired(), } active = self.registry.list_active() cells = [] for cell in active: probe = self.backend.probe(cell.cell_id) cells.append({ "cell_id": cell.cell_id, "project": cell.project, "backend": cell.backend, "state": cell.state.value, "members": [m.identity for m in cell.members], "probe": probe.message, "expired": cell.is_expired(), }) return {"success": True, "active_cells": cells} def close(self, cell_id: str, closed_by: str = "human:Alexander") -> dict: """Gracefully close a cell.""" cell = self.registry.load(cell_id) if not cell: return {"success": False, "message": f"Cell {cell_id} not found"} res = self.backend.close(cell_id) cell.state = CellState.CLOSING self.registry.save(cell) return { "success": res.success, "cell_id": cell_id, "message": f"Closed {cell_id} by {closed_by}", } def destroy(self, cell_id: str, destroyed_by: str = "human:Alexander") -> dict: """Forcefully destroy a cell and all runtime state.""" cell = self.registry.load(cell_id) if not cell: return {"success": False, "message": f"Cell {cell_id} not found"} res = self.backend.destroy(cell_id) self.packager.destroy(cell_id) cell.state = CellState.DESTROYED cell.destroyed_at = datetime.now().astimezone().isoformat() self.registry.save(cell) return { "success": True, "cell_id": cell_id, "message": f"Destroyed {cell_id} by {destroyed_by}", } # ------------------------------------------------------------------ # CLI helpers # ------------------------------------------------------------------ def run_cli(self, args: List[str]) -> dict: if not args: return self.status() cmd = args[0] if cmd == "summon" and len(args) >= 3: return self.summon(agent=args[1], project=args[2]) if cmd == "invite" and len(args) >= 4: return self.invite(cell_id=args[1], guest=args[2], role=args[3]) if cmd == "team" and len(args) >= 3: agents = args[1].split("+") project = args[2] return self.team(agents=agents, project=project) if cmd == "status": return self.status(cell_id=args[1] if len(args) > 1 else None) if cmd == "close" and len(args) >= 2: return self.close(cell_id=args[1]) if cmd == "destroy" and len(args) >= 2: return self.destroy(cell_id=args[1]) return { "success": False, "message": ( "Unknown command. Usage:\n" " lazarus summon \n" " lazarus invite \n" " lazarus team \n" " lazarus status [cell_id]\n" " lazarus close \n" " lazarus destroy \n" ) }