"""Ptah Bridge - orchestrator connecting AsyncEventBus and MemoryTaskQueue. Maps roadmap execution steps to agent requests (Planner, Coder, Evaluator, Security, Research) and exposes an interface for Ptah Desktop. """ from __future__ import annotations import asyncio import logging import time from dataclasses import dataclass, field from enum import Enum from typing import Any from cyberbrain26.agents.orchestrator import ( AgentResult, CoderAgent, EvaluatorAgent, PlannerAgent, ResearchAgent, SecurityAgent, ) from cyberbrain26.control_plane import AsyncEventBus, MemoryTaskQueue logger = logging.getLogger(__name__) class RoadmapStatus(str, Enum): """Lifecycle status of a roadmap execution.""" PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" class StepStatus(str, Enum): """Lifecycle status of a single roadmap step.""" PENDING = "pending" DISPATCHED = "dispatched" COMPLETED = "completed" FAILED = "failed" @dataclass class RoadmapStep: """A single execution step within a roadmap. Each step is mapped to an agent role (planner, coder, evaluator, security, research) and dispatched through the orchestrator. """ id: str role: str description: str payload: dict[str, Any] = field(default_factory=dict) depends_on: list[str] = field(default_factory=list) status: StepStatus = StepStatus.PENDING result: StepResult | None = None @dataclass class StepResult: """Outcome of executing a single roadmap step.""" step_id: str role: str success: bool data: dict[str, Any] = field(default_factory=dict) error: str | None = None @dataclass class Roadmap: """A collection of steps executed as a unit by the bridge.""" id: str name: str steps: list[RoadmapStep] = field(default_factory=list) status: RoadmapStatus = RoadmapStatus.PENDING created_at: float = field(default_factory=lambda: time.time()) # Re-export AgentResult for convenience. __all__ = [ "AgentResult", "PtahBridge", "Roadmap", "RoadmapStatus", "RoadmapStep", "StepResult", ] class PtahBridge: """Orchestrator that maps roadmap execution steps to agent requests. Connects to an :class:`AsyncEventBus` and :class:`MemoryTaskQueue`, instantiates the five agent roles, and exposes an interface for Ptah Desktop to drive roadmap execution. """ def __init__(self, bus: AsyncEventBus | None = None, queue: MemoryTaskQueue | None = None) -> None: self.bus = bus or AsyncEventBus() self.queue = queue or MemoryTaskQueue() self.planner = PlannerAgent(self.bus, self.queue) self.coder = CoderAgent(self.bus, self.queue) self.evaluator = EvaluatorAgent(self.bus, self.queue) self.security = SecurityAgent(self.bus, self.queue) self.research = ResearchAgent(self.bus, self.queue) self._agents: dict[str, Any] = { "planner": self.planner, "coder": self.coder, "evaluator": self.evaluator, "security": self.security, "research": self.research, } self._roadmaps: dict[str, Roadmap] = {} self._step_results: dict[str, StepResult] = {} self._lock = asyncio.Lock() # ------------------------------------------------------------------ # Roadmap management # ------------------------------------------------------------------ def create_roadmap(self, roadmap_id: str, name: str, steps: list[RoadmapStep] | None = None) -> Roadmap: """Register a new roadmap and return it.""" roadmap = Roadmap(id=roadmap_id, name=name, steps=steps or []) self._roadmaps[roadmap_id] = roadmap return roadmap def get_roadmap(self, roadmap_id: str) -> Roadmap | None: """Return a roadmap by id, or None if it does not exist.""" return self._roadmaps.get(roadmap_id) def list_roadmaps(self) -> list[Roadmap]: """Return all registered roadmaps.""" return list(self._roadmaps.values()) # ------------------------------------------------------------------ # Step execution # ------------------------------------------------------------------ async def execute_step(self, roadmap_id: str, step: RoadmapStep) -> StepResult: """Dispatch a single roadmap step to the appropriate agent. The step's ``role`` selects the agent; the step's ``payload`` is forwarded as the task data. Results are recorded and returned. """ agent = self._agents.get(step.role) if agent is None: result = StepResult( step_id=step.id, role=step.role, success=False, error=f"unknown role: {step.role}", ) step.status = StepStatus.FAILED step.result = result self._step_results[step.id] = result return result step.status = StepStatus.DISPATCHED try: agent_result = await agent.execute(step.payload) except Exception as exc: logger.exception("step %s failed", step.id) result = StepResult( step_id=step.id, role=step.role, success=False, error=str(exc), ) step.status = StepStatus.FAILED step.result = result self._step_results[step.id] = result return result success = bool(agent_result.success) if agent_result else False data = agent_result.data if agent_result else {} error = agent_result.error if agent_result else None result = StepResult( step_id=step.id, role=step.role, success=success, data=data, error=error, ) step.status = StepStatus.COMPLETED if success else StepStatus.FAILED step.result = result self._step_results[step.id] = result return result async def execute_roadmap(self, roadmap_id: str) -> Roadmap: """Execute every step of a roadmap in order, respecting dependencies. Steps whose ``depends_on`` list is satisfied are dispatched; the roadmap status transitions to RUNNING then COMPLETED or FAILED. """ roadmap = self._roadmaps.get(roadmap_id) if roadmap is None: raise KeyError(f"unknown roadmap: {roadmap_id}") async with self._lock: roadmap.status = RoadmapStatus.RUNNING completed: set[str] = set() failed = False pending = list(roadmap.steps) while pending: progress = False for step in list(pending): deps = set(step.depends_on) if deps and not deps.issubset(completed): continue pending.remove(step) await self.execute_step(roadmap_id, step) if step.result is not None: if step.result.success: completed.add(step.id) else: failed = True progress = True if not progress: # Remaining steps are blocked on missing dependencies. break roadmap.status = RoadmapStatus.FAILED if failed else RoadmapStatus.COMPLETED return roadmap # ------------------------------------------------------------------ # Query helpers (Ptah Desktop interface) # ------------------------------------------------------------------ def step_result(self, step_id: str) -> StepResult | None: """Return the recorded result for a step, if any.""" return self._step_results.get(step_id) def roadmap_status(self, roadmap_id: str) -> RoadmapStatus | None: """Return the current status of a roadmap, or None if unknown.""" roadmap = self._roadmaps.get(roadmap_id) return roadmap.status if roadmap else None def agent_results(self, role: str) -> list[AgentResult]: """Return the accumulated results for a given agent role.""" agent = self._agents.get(role) return list(agent.results) if agent else [] async def shutdown(self) -> None: """Best-effort cleanup of the bridge (no persistent state to close).""" logger.debug("PtahBridge shutdown for roadmap(s)=%s", list(self._roadmaps))