"""Ptah Bridge - orchestrator connecting AsyncEventBus and MemoryTaskQueue. Ptah maps roadmap execution steps to registered agents and routes execution through the modular AgentRegistry/TaskDispatcher security boundary. Execution path: PtahBridge -> AgentRegistry -> TaskDispatcher identity -> scope -> policy -> approval -> audit -> Agent.execute() -> AuthorizedExecutionService re-verifies, does not re-ask -> ExecutionProvider A step that a human had to approve raises exactly one approval request. The dispatcher owns that decision and hands it to the agent's execution service as a grant bound to the one agent call it just authorized; the service re-checks scope, policy, identity, action and the payload digest, and only then proceeds without asking again. Both gates run on every step - the inner one is a verification, not a formality - but one operator decision authorises one step, which is what the documented single-approval flow promises. """ from __future__ import annotations import asyncio import logging import time from collections.abc import Iterable, Mapping from dataclasses import dataclass, field from enum import Enum from typing import Any from cyberbrain26.agents.orchestrator import ( AgentResult, BrowserAgent, CoderAgent, EvaluatorAgent, PlannerAgent, ResearchAgent, SecurityAgent, ) from cyberbrain26.control_plane import ( EXECUTION_ATTEMPTED_KEY, AsyncEventBus, MemoryTaskQueue, PersistentTaskStatus, SQLiteTaskQueue, ) from cyberbrain26.data_plane.sqlite_store import SQLiteDatabase from cyberbrain26.execution_plane import ( AuthorizedExecutionService, BubblewrapExecutionProvider, ExecutionRefused, ) from cyberbrain26.extension_layer.execution_provider import ExecutionProvider from cyberbrain26.extension_layer.registry_impl import ( InMemoryAgentRegistry, InMemoryTaskDispatcher, ) from cyberbrain26.security_plane.approval import ApprovalManager from cyberbrain26.security_plane.audit import AuditEventType, AuditLog from cyberbrain26.security_plane.identity import AgentIdentity from cyberbrain26.security_plane.permission import PermissionGateway, SecurityError from cyberbrain26.security_plane.policy import PolicyEngine logger = logging.getLogger(__name__) class RoadmapStatus(str, Enum): """Lifecycle status of a roadmap execution.""" PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" BLOCKED = "blocked" CANCELLED = "cancelled" class StepStatus(str, Enum): """Lifecycle status of a single roadmap step.""" PENDING = "pending" DISPATCHED = "dispatched" COMPLETED = "completed" FAILED = "failed" BLOCKED = "blocked" @dataclass class RoadmapStep: """A single execution step within a roadmap.""" 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()) __all__ = [ "AgentResult", "PtahBridge", "Roadmap", "RoadmapStatus", "RoadmapStep", "StepResult", ] class PtahBridge: """Orchestrator for roadmap-driven multi-agent execution. Ptah owns the default in-memory registry and dispatcher used by the local application. Every built-in Ptah agent receives an explicit AgentIdentity and shared PermissionGateway. The public roadmap/query API remains compatible with the previous implementation while execution is routed through the modular orchestration/security layer. """ def __init__( self, bus: AsyncEventBus | None = None, queue: MemoryTaskQueue | None = None, approval_manager: ApprovalManager | None = None, approval_timeout_seconds: float = 30.0, execution_provider: ExecutionProvider | None = None, database: SQLiteDatabase | None = None, *, identity_scopes: Mapping[str, Iterable[str]] | None = None, ) -> None: if isinstance(queue, SQLiteTaskQueue): if database is not None and queue.database.path != database.path: raise ValueError("PtahBridge and task queue must share the same SQLite database") database = queue.database elif database is not None and queue is not None: raise TypeError("pass a SQLiteTaskQueue when configuring PtahBridge with SQLiteDatabase") if ( database is not None and approval_manager is not None and ( approval_manager.database is None or approval_manager.database.path != database.path ) ): raise ValueError( "PtahBridge and ApprovalManager must share the same SQLite database" ) self.bus = bus or AsyncEventBus() self.database = database self.queue = queue or ( SQLiteTaskQueue(database) if database is not None else MemoryTaskQueue() ) if database is not None and isinstance(self.queue, SQLiteTaskQueue): self.queue.register_handler( "ptah.dispatch_step", self._dispatch_persistent_step, replace=True ) # Shared modular security/orchestration infrastructure. self.audit_log = AuditLog(database=database) self.policy_engine = PolicyEngine(audit_log=self.audit_log) self.permission_gateway = PermissionGateway( self.policy_engine, audit_log=self.audit_log ) effective_approval_manager = approval_manager or ( ApprovalManager(database=database) if database is not None else None ) self.approval_manager = effective_approval_manager self.execution_service = AuthorizedExecutionService( execution_provider or BubblewrapExecutionProvider(), self.permission_gateway, approval_manager=effective_approval_manager, audit_log=self.audit_log, approval_timeout_seconds=approval_timeout_seconds, ) self.registry = InMemoryAgentRegistry() self.dispatcher = InMemoryTaskDispatcher( registry=self.registry, policy_engine=self.policy_engine, audit_log=self.audit_log, approval_manager=effective_approval_manager, approval_timeout_seconds=approval_timeout_seconds, ) # Built-in Ptah agents. Every agent receives an explicit identity. self.planner = PlannerAgent( self.bus, self.queue, identity=AgentIdentity( agent_id="ptah-planner", role="planner", ), permission_gateway=self.permission_gateway, ) self.coder = CoderAgent( self.bus, self.queue, identity=AgentIdentity( agent_id="ptah-coder", role="coder", ), permission_gateway=self.permission_gateway, execution_service=self.execution_service, ) self.evaluator = EvaluatorAgent( self.bus, self.queue, identity=AgentIdentity( agent_id="ptah-evaluator", role="evaluator", ), permission_gateway=self.permission_gateway, ) self.security = SecurityAgent( self.bus, self.queue, identity=AgentIdentity( agent_id="ptah-security", role="security", ), permission_gateway=self.permission_gateway, ) self.research = ResearchAgent( self.bus, self.queue, identity=AgentIdentity( agent_id="ptah-research", role="research", ), permission_gateway=self.permission_gateway, ) self.browser = BrowserAgent( self.bus, self.queue, identity=AgentIdentity( agent_id="ptah-browser", role="browser", ), permission_gateway=self.permission_gateway, ) self._agents: dict[str, Any] = { "planner": self.planner, "coder": self.coder, "evaluator": self.evaluator, "security": self.security, "research": self.research, "browser": self.browser, } # Optional per-role identity scopes. The default (None) is deliberate and # must stay unchanged: built-in Ptah agents hold no scopes, so a step # carrying ``code`` is denied at the scope gate before policy and # approval are ever consulted. Callers that explicitly pass scopes are # widening the identity only; the merged scopes do not grant trust, # policy decisions, or approvals. Scopes are merged, never replaced, so # existing identity scopes survive. if identity_scopes is not None: for role, scopes in identity_scopes.items(): agent = self._agents.get(role) if agent is None: raise ValueError( f"identity_scopes references unknown Ptah role: {role!r}" ) agent.identity.scopes.update(str(scope) for scope in scopes) # Register all Ptah agents in the canonical registry. for agent in self._agents.values(): self.registry.register(agent.identity.agent_id, agent) 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. Validates that step IDs are unique. """ steps = steps or [] step_ids: set[str] = set() for step in steps: if step.id in step_ids: raise ValueError( f"duplicate step id '{step.id}' in roadmap '{roadmap_id}'" ) step_ids.add(step.id) roadmap = Roadmap(id=roadmap_id, name=name, steps=steps) 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 one roadmap step through the TaskDispatcher. The dispatcher performs the security boundary checks before the underlying agent's ``execute`` method is called. """ 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 # Include the roadmap/step identity in the dispatched payload without # changing the agent-facing task semantics. payload = dict(step.payload) payload.setdefault("task_id", step.id) payload.setdefault("roadmap_id", roadmap_id) step.status = StepStatus.DISPATCHED try: if self.database is not None and isinstance(self.queue, SQLiteTaskQueue): durable_id = f"ptah:{roadmap_id}:{step.id}" job = self.queue.get_job(durable_id) if job is None: self.queue.enqueue_job( "ptah.dispatch_step", { "task_id": step.id, "agent_id": agent.identity.agent_id, "payload": payload, }, task_id=durable_id, ) self.audit_log.record( AuditEventType.TASK_CREATED, agent_id=agent.identity.agent_id, task_id=step.id, action="dispatch_step", metadata={"durable_task_id": durable_id, "roadmap_id": roadmap_id}, ) job = self.queue.get_job(durable_id) if ( job is not None and job.status is PersistentTaskStatus.FAILED and self.queue.retry_failed(durable_id) ): # A previous attempt ended with an exception, typically a # denial raised by the dispatcher's authorization chain # before the agent body ever ran. Re-dispatch reruns that # whole chain from the top — identity scope, then policy, # then a freshly created approval request — so retrying # cannot skip any gate, and the approval an earlier attempt # obtained does not carry over: the grant that let the inner # gate skip a second request lived only inside that attempt. # It only makes a previously denied step retryable after the # operator changed something; it grants nothing by itself. # INTERRUPTED stays manual because there the side effects # may be partially applied. logger.warning( "re-dispatching previously failed durable step %s", durable_id ) self.audit_log.record( AuditEventType.ERROR, agent_id=agent.identity.agent_id, task_id=step.id, action="dispatch_step", result="failed_task_requeued", error=job.error or "", metadata={ "durable_task_id": durable_id, "roadmap_id": roadmap_id, }, ) job = self.queue.get_job(durable_id) if ( job is not None and job.status is PersistentTaskStatus.COMPLETED and self.queue.retry_blocked(durable_id) ): # The job is COMPLETED yet unusable: the agent caught the # refusal, so no code ran and there is nothing to protect # by refusing to retry. ``retry_blocked`` only re-opens a # record whose stored result structurally states # ``execution_attempted`` is false; a completed job whose # code actually ran keeps this path closed and is returned # as-is below. Re-dispatching reruns identity scope, # policy and approval from the top, so retrying cannot skip # a gate — it only re-offers the step for a fresh decision. logger.warning( "re-dispatching previously blocked durable step %s", durable_id ) self.audit_log.record( AuditEventType.ERROR, agent_id=agent.identity.agent_id, task_id=step.id, action="dispatch_step", result="blocked_task_requeued", error=str((job.result or {}).get("error") or ""), metadata={ "durable_task_id": durable_id, "roadmap_id": roadmap_id, }, ) job = self.queue.get_job(durable_id) if job is not None and job.status is PersistentTaskStatus.PENDING: job = await self.queue.dispatch_job(durable_id) if job is None: raise RuntimeError("durable roadmap task could not be loaded") if job.status is PersistentTaskStatus.COMPLETED and isinstance(job.result, dict): agent_result = AgentResult( agent_name=str(job.result.get("agent_name", agent.name)), task_id=step.id, success=bool(job.result.get("success", False)), data=job.result.get("data", {}), error=job.result.get("error"), # Missing marker (a legacy row) counts as attempted, so # an old record can never be mistaken for a blocked one. execution_attempted=( job.result.get(EXECUTION_ATTEMPTED_KEY) is not False ), ) elif job.status is PersistentTaskStatus.INTERRUPTED: raise RuntimeError( "step was interrupted after process restart; review its side effects " "before explicitly retrying the durable task" ) else: raise RuntimeError( job.error or f"durable task ended with status {job.status.value}" ) else: agent_result = await self.dispatcher.dispatch( task_id=step.id, agent_id=agent.identity.agent_id, payload=payload, ) except Exception as exc: if isinstance(exc, (SecurityError, ExecutionRefused)): logger.warning("step %s refused: %s", step.id, exc) else: logger.exception("step %s failed unexpectedly", 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 _dispatch_persistent_step(self, job: dict[str, Any]) -> dict[str, Any]: """Execute a persisted step payload and return JSON-safe result data.""" try: agent_result = await self.dispatcher.dispatch( task_id=str(job["task_id"]), agent_id=str(job["agent_id"]), payload=dict(job["payload"]), ) except (SecurityError, ExecutionRefused) as exc: logger.warning( "durable step %s refused: %s", str(job.get("task_id", "unknown")), exc ) return { "agent_name": "ptah-dispatcher", "success": False, "data": {}, "error": str(exc), EXECUTION_ATTEMPTED_KEY: False, } if isinstance(agent_result, AgentResult): return { "agent_name": agent_result.agent_name, "success": agent_result.success, "data": agent_result.data, "error": agent_result.error, # Persisted so a later run can tell "refused before execution" # from "the code ran and failed" without re-reading messages. EXECUTION_ATTEMPTED_KEY: agent_result.execution_attempted, } return { "agent_name": str(getattr(agent_result, "agent_name", "unknown")), "success": bool(getattr(agent_result, "success", True)), "data": getattr(agent_result, "data", {}) or {}, "error": getattr(agent_result, "error", None), # Unknown provenance: assume execution may have happened, which # keeps such a job non-retryable. EXECUTION_ATTEMPTED_KEY: True, } async def execute_roadmap(self, roadmap_id: str) -> Roadmap: """Execute roadmap steps while respecting dependencies. A roadmap with unresolved dependencies is explicitly marked BLOCKED instead of incorrectly being reported as COMPLETED. """ 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: # No executable step remains. The unresolved steps are # genuinely blocked by missing/failed dependencies. for step in pending: step.status = StepStatus.BLOCKED roadmap.status = RoadmapStatus.BLOCKED return roadmap 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.""" logger.debug( "PtahBridge shutdown for roadmap(s)=%s", list(self._roadmaps), )