from __future__ import annotations import asyncio import logging from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any, Protocol from uuid import uuid4 from cyberbrain26.control_plane import AsyncEventBus, MemoryTaskQueue from cyberbrain26.execution_plane.secure_gateway import ( AuthorizedExecutionService, ExecutionRefused, ) from cyberbrain26.security_plane.identity import AgentIdentity from cyberbrain26.security_plane.permission import PermissionGateway, SecurityError logger = logging.getLogger(__name__) PLANNER_TASK_REQUEST = "planner.task_request" PLANNER_TASK_COMPLETED = "planner.task_completed" CODER_TASK_REQUEST = "coder.task_request" CODER_TASK_COMPLETED = "coder.task_completed" EVALUATOR_TASK_REQUEST = "evaluator.task_request" EVALUATOR_TASK_COMPLETED = "evaluator.task_completed" SECURITY_TASK_REQUEST = "security.task_request" SECURITY_TASK_COMPLETED = "security.task_completed" RESEARCH_TASK_REQUEST = "research.task_request" RESEARCH_TASK_COMPLETED = "research.task_completed" BROWSER_TASK_REQUEST = "browser.task_request" BROWSER_TASK_COMPLETED = "browser.task_completed" BUS_TASK_ACTION = "execute" DEFAULT_PLAN_STEPS = 3 MAX_PLAN_STEPS = 50 BROWSER_CAPABILITY_UNAVAILABLE_REASON = ( "BrowserAgent has no browsing backend; the capability is not implemented " "and no page was fetched or inspected." ) @dataclass class AgentResult: """Structured outcome of one agent run. ``execution_attempted`` records whether the sandbox may already have run code for this result. It defaults to True on purpose: a consumer that cannot prove the code never started must assume side effects are possible and therefore must not re-run the step. Only an agent that structurally knows it was refused *before* the provider was invoked may set it to False, which is what makes a blocked step retryable. """ agent_name: str task_id: str | None success: bool data: dict[str, Any] = field(default_factory=dict) error: str | None = None execution_attempted: bool = True class TaskAuthorizer(Protocol): """Authorization collaborator required before bus-driven agent work. Security rationale: the event bus is a shared, publicly reachable channel (``PtahBridge.bus`` is exposed to callers), so a published ``*_TASK_REQUEST`` is untrusted input. An agent therefore never calls ``execute`` for a bus task on its own: it first hands the payload to an injected authorizer that must run the canonical chain (identity -> scope -> policy -> approval -> audit). Contract for implementations: * return normally when the request may proceed; * raise (``SecurityError`` for a policy refusal, or any other exception when the decision could not be made) when it may not. Only "returned without raising" counts as authorization. A return value is deliberately not inspected, so an implementation cannot accidentally turn a falsy return into an approval. ``BaseAgent`` treats every raise as a refusal and treats a missing authorizer as a refusal too, which keeps the failure mode closed when wiring is incomplete. """ async def __call__(self, task_data: dict[str, Any]) -> None: ... class DispatcherAuthorizerPort(Protocol): """Structural view of ``InMemoryTaskDispatcher._authorize``. Declared structurally rather than imported so the agents module keeps no import-time dependency on the extension layer, and so the extension layer can rename its internals without a silent behaviour change here. """ async def _authorize( self, task_id: str, registered_agent_id: str, agent: Any, action: str, resource: str | None = ..., ) -> None: ... class DispatcherAuthorizer: """Adapter binding a bus task to a dispatcher's authorization decision. ``InMemoryTaskDispatcher._authorize`` already implements the canonical chain (identity match -> policy -> approval -> audit) and already fails closed by raising ``SecurityError``. This adapter forwards to it so the bus path and the dispatcher path share one decision instead of duplicating - and eventually weakening - policy evaluation inside the agents. Construction order matters: the bridge must build the agents first and then assign the adapter, because the adapter needs both the agent and its identity:: agent = PlannerAgent(bridge.bus, bridge.queue, identity=..., permission_gateway=...) agent.authorizer = DispatcherAuthorizer(bridge.dispatcher, agent.identity.agent_id, agent) """ def __init__( self, dispatcher: DispatcherAuthorizerPort, agent_id: str, agent: Any, *, action: str = BUS_TASK_ACTION, resource: str | None = None, ) -> None: self._dispatcher = dispatcher self._agent_id = agent_id self._agent = agent self._action = action self._resource = resource async def __call__(self, task_data: dict[str, Any]) -> None: raw_task_id = task_data.get("task_id") task_id = str(raw_task_id) if raw_task_id else f"bus-{uuid4().hex}" await self._dispatcher._authorize( task_id, self._agent_id, self._agent, self._action, self._resource, ) class BaseAgent(ABC): def __init__( self, bus: AsyncEventBus, queue: MemoryTaskQueue, identity: AgentIdentity | None = None, permission_gateway: PermissionGateway | None = None, authorizer: TaskAuthorizer | None = None, ) -> None: """Wire the agent onto the bus. ``authorizer`` gates every bus-delivered task request. It defaults to ``None``, which means "no authorization collaborator was injected" and therefore **deny**: the agent refuses bus-driven work instead of executing it. The default is deliberately fail-closed so that a construction site which has not been migrated to the new contract loses the unauthenticated entry point rather than keeping it. Callers that legitimately drive agents over the bus must inject a ``TaskAuthorizer`` (see ``DispatcherAuthorizer``), either in this constructor or by assigning ``agent.authorizer`` afterwards. """ self.bus = bus self.queue = queue self.results: list[AgentResult] = [] self._task_request_event: str | None = None self._task_completed_event: str | None = None self.identity = identity self.permission_gateway = permission_gateway self.authorizer = authorizer @property @abstractmethod def name(self) -> str: ... @property @abstractmethod def role(self) -> str: ... def _subscribe(self) -> None: if self._task_request_event: self.bus.subscribe(self._task_request_event, self._on_task_request) async def _on_task_request(self, event: Any) -> AgentResult | None: """Handle a bus-delivered task request. Security rationale: this handler used to call ``execute`` directly, so anyone holding a reference to the event bus could run an agent body with an arbitrary payload and no identity, scope, policy or approval check at all - the bus was a full bypass of the authorization chain. Every bus task must now clear an injected ``TaskAuthorizer`` first; a missing or denying authorizer produces a recorded refusal instead of work. """ payload = getattr(event, "payload", None) task_data: dict[str, Any] = payload if isinstance(payload, dict) else {} refusal = await self._authorize_task_request(task_data) result = refusal if refusal is not None else await self.execute(task_data) if result is not None and self._task_completed_event: await self.bus.publish(self._task_completed_event, { "agent": self.name, "result": {"success": result.success, "data": result.data, "error": result.error}, "task_id": result.task_id, }) return result @abstractmethod async def execute(self, task_data: dict[str, Any]) -> AgentResult | None: ... async def _authorize_task_request( self, task_data: dict[str, Any] ) -> AgentResult | None: """Authorize one bus task request before the agent body may run. Returns ``None`` when the request is authorized (the caller then runs ``execute``), or an ``AgentResult`` describing a refusal. Fail-closed: no authorizer, a raising authorizer, or any unexpected exception from the authorizer all mean "refuse". A refusal is recorded with ``execution_attempted=False`` because it is structurally known to have happened before the agent body started, which keeps the step retryable once the wiring is fixed. """ if self.authorizer is None: return self._refuse_task_request( task_data, "bus task requests require an injected authorizer; none is configured", ) try: await self.authorizer(task_data) except SecurityError as exc: return self._refuse_task_request(task_data, str(exc)) except Exception as exc: # noqa: BLE001 - fail closed: any unusable # authorization decision (bug, timeout, unavailable gateway) # must deny, never fall through to execute. logger.warning( "%s bus task authorization could not be completed, refusing: %s", self.name, exc, ) return self._refuse_task_request( task_data, f"authorization failed closed: {exc}" ) return None def _refuse_task_request(self, task_data: dict[str, Any], reason: str) -> AgentResult: """Record and return a refusal for an unauthorized bus task request.""" logger.warning("%s refused a bus task request: %s", self.name, reason) return self.publish_result( self._task_id_from(task_data), False, {"role": self.role, "source": "event_bus"}, reason, execution_attempted=False, ) def _task_id_from(self, task_data: dict[str, Any]) -> str | None: """Best-effort task id for a refusal or unavailable-capability result.""" raw = task_data.get("task_id") return None if raw is None else str(raw) def publish_result( self, task_id: str | None, success: bool, data: dict[str, Any] | None = None, error: str | None = None, *, execution_attempted: bool = True, ) -> AgentResult: """Record an outcome and mark whether execution was attempted.""" result = AgentResult( agent_name=self.name, task_id=task_id, success=success, data=data or {}, error=error, execution_attempted=execution_attempted, ) self.results.append(result) return result def check_permission(self, action: str, resource: str | None = None) -> bool: """Check whether this agent is permitted to perform an action. Returns True if allowed, False if denied or requires approval. """ if self.identity is None or self.permission_gateway is None: return False result = self.permission_gateway.evaluate(self.identity, action, resource) return result.decision.value == "allow" class PlannerAgent(BaseAgent): def __init__( self, bus: AsyncEventBus, queue: MemoryTaskQueue, identity: AgentIdentity | None = None, permission_gateway: PermissionGateway | None = None, *, authorizer: TaskAuthorizer | None = None, ) -> None: super().__init__(bus, queue, identity, permission_gateway, authorizer) self._task_request_event = PLANNER_TASK_REQUEST self._task_completed_event = PLANNER_TASK_COMPLETED self._plan: list[dict[str, Any]] = [] self._subscribe() @property def name(self) -> str: return "PlannerAgent" @property def role(self) -> str: return "planner" async def execute(self, task_data: dict[str, Any]) -> AgentResult | None: """Build an in-memory plan, honouring ``max_steps`` from the payload. ``max_steps`` is what ``cyberbrain plan --steps/-n`` sends, and it used to be ignored: the plan was hard-coded to three steps, so the CLI flag silently had no effect. """ goal = task_data.get("goal", "unknown") steps, error = self._resolve_step_count(task_data.get("max_steps")) if error is not None: # Rejected on payload shape: no plan was produced and nothing was # enqueued, so the caller can retry with a corrected max_steps. return self.publish_result( None, False, {"goal": goal}, error, execution_attempted=False, ) sub_tasks = [ {"step": i, "description": f"{goal} - step {i}"} for i in range(1, steps + 1) ] self._plan = sub_tasks task_id = self.queue.enqueue(lambda: asyncio.sleep(0)) result = self.publish_result( task_id, True, {"goal": goal, "plan": sub_tasks, "max_steps": steps} ) return result @staticmethod def _resolve_step_count(requested: Any) -> tuple[int, str | None]: """Validate ``max_steps`` and return ``(steps, error)``. Rules, chosen because ``max_steps`` arrives over the event bus and is therefore untrusted input: * absent (``None``) -> ``DEFAULT_PLAN_STEPS`` (3); * ``bool`` or non-integer -> refused, never coerced, because ``True`` and ``"3"`` are ambiguous input, not a step count; * ``<= 0`` -> refused, because an empty plan is not a plan; * ``> MAX_PLAN_STEPS`` (50) -> capped, so one request cannot inflate the plan and the resulting roadmap without bound. The cap is reported back as the effective ``max_steps`` so the caller sees what was used. """ if requested is None: return DEFAULT_PLAN_STEPS, None if isinstance(requested, bool) or not isinstance(requested, int): return DEFAULT_PLAN_STEPS, f"max_steps must be a positive integer, got {requested!r}" if requested <= 0: return DEFAULT_PLAN_STEPS, f"max_steps must be greater than zero, got {requested}" return min(requested, MAX_PLAN_STEPS), None def get_plan(self) -> list[dict[str, Any]]: return list(self._plan) class CoderAgent(BaseAgent): def __init__( self, bus: AsyncEventBus, queue: MemoryTaskQueue, identity: AgentIdentity | None = None, permission_gateway: PermissionGateway | None = None, execution_service: AuthorizedExecutionService | None = None, *, authorizer: TaskAuthorizer | None = None, ) -> None: super().__init__(bus, queue, identity, permission_gateway, authorizer) self.execution_service = execution_service self._task_request_event = CODER_TASK_REQUEST self._task_completed_event = CODER_TASK_COMPLETED self._implemented: list[str] = [] self._subscribe() @property def name(self) -> str: return "CoderAgent" @property def role(self) -> str: return "coder" async def execute(self, task_data: dict[str, Any]) -> AgentResult | None: if "code" in task_data: description = task_data.get("description", "code execution") code = task_data.get("code") language = task_data.get("language") task_id = str(task_data.get("task_id") or f"coder-{uuid4().hex}") if not isinstance(code, str) or not isinstance(language, str): # Rejected on payload shape, the provider was never reached. return self.publish_result( task_id, False, {"description": description}, "code and language must be strings", execution_attempted=False, ) if self.execution_service is None or self.identity is None: # Nothing was authorized and nothing ran; the step is blocked, # not broken, so it may be retried once the bridge is wired. return self.publish_result( task_id, False, {"description": description}, "authorized execution service and agent identity are required", execution_attempted=False, ) try: execution = await self.execution_service.execute( self.identity, code, language, task_id=task_id, timeout_seconds=task_data.get("timeout_seconds", 30), workspace=task_data.get("workspace"), ) except ExecutionRefused as exc: # The authorization chain refused the request before the # provider was invoked, so no code ran and re-running cannot # duplicate anything. Say so structurally instead of leaving # the consumer to guess from the error text. return self.publish_result( task_id, False, {"description": description}, str(exc), execution_attempted=False, ) except (SecurityError, RuntimeError, ValueError) as exc: # Anything else is unknown: the sandbox may already have run # the code, so the step stays non-retryable by default. return self.publish_result( task_id, False, {"description": description}, str(exc), ) success = execution.get("return_code") == 0 and not execution.get("timed_out", False) if success: self._implemented.append(str(description)) error = None if not success: error = execution.get("stderr") or ( "execution timed out" if execution.get("timed_out", False) else f"execution exited with code {execution.get('return_code')}" ) return self.publish_result( task_id, success, {"description": description, "execution": execution}, error, ) task_id = self.queue.enqueue(self._do_code) try: record = self.queue.get(task_id) assert record is not None await self.queue.dispatch(record) except (KeyError, RuntimeError, asyncio.CancelledError) as exc: logger.warning("CoderAgent queue dispatch failed: %s", exc) except SecurityError: logger.warning("CoderAgent permission denied during execution") raise except Exception: logger.exception("CoderAgent unexpected error during queue dispatch") raise description = task_data.get("description", "unknown") self._implemented.append(description) result = self.publish_result(task_id, True, {"description": description, "status": "implemented"}) return result async def _do_code(self) -> str: return "code_complete" def get_implemented(self) -> list[str]: return list(self._implemented) class EvaluatorAgent(BaseAgent): def __init__( self, bus: AsyncEventBus, queue: MemoryTaskQueue, identity: AgentIdentity | None = None, permission_gateway: PermissionGateway | None = None, *, authorizer: TaskAuthorizer | None = None, ) -> None: super().__init__(bus, queue, identity, permission_gateway, authorizer) self._task_request_event = EVALUATOR_TASK_REQUEST self._task_completed_event = EVALUATOR_TASK_COMPLETED self._assessments: list[dict[str, Any]] = [] self._subscribe() @property def name(self) -> str: return "EvaluatorAgent" @property def role(self) -> str: return "evaluator" async def execute(self, task_data: dict[str, Any]) -> AgentResult | None: task_id = self.queue.enqueue(self._do_evaluate) try: record = self.queue.get(task_id) assert record is not None await self.queue.dispatch(record) except (KeyError, RuntimeError, asyncio.CancelledError) as exc: logger.warning("EvaluatorAgent queue dispatch failed: %s", exc) except SecurityError: logger.warning("EvaluatorAgent permission denied during execution") raise except Exception: logger.exception("EvaluatorAgent unexpected error during queue dispatch") raise score = task_data.get("score", 0.0) passed = score >= 0.5 assessment = {"passed": passed, "score": score} self._assessments.append(assessment) result = self.publish_result(task_id, passed, assessment) return result async def _do_evaluate(self) -> dict[str, Any]: return {"status": "evaluated"} def get_assessments(self) -> list[dict[str, Any]]: return list(self._assessments) class SecurityAgent(BaseAgent): def __init__( self, bus: AsyncEventBus, queue: MemoryTaskQueue, identity: AgentIdentity | None = None, permission_gateway: PermissionGateway | None = None, *, authorizer: TaskAuthorizer | None = None, ) -> None: super().__init__(bus, queue, identity, permission_gateway, authorizer) self._task_request_event = SECURITY_TASK_REQUEST self._task_completed_event = SECURITY_TASK_COMPLETED self._findings: list[dict[str, Any]] = [] self._subscribe() @property def name(self) -> str: return "SecurityAgent" @property def role(self) -> str: return "security" async def execute(self, task_data: dict[str, Any]) -> AgentResult | None: task_id = self.queue.enqueue(self._do_security_check) try: record = self.queue.get(task_id) assert record is not None await self.queue.dispatch(record) except (KeyError, RuntimeError, asyncio.CancelledError) as exc: logger.warning("SecurityAgent queue dispatch failed: %s", exc) except SecurityError: logger.warning("SecurityAgent permission denied during execution") raise except Exception: logger.exception("SecurityAgent unexpected error during queue dispatch") raise requires_review = task_data.get("requires_review", False) finding = {"requires_review": requires_review, "severity": "high" if requires_review else "low"} self._findings.append(finding) result = self.publish_result(task_id, not requires_review, finding) return result async def _do_security_check(self) -> dict[str, Any]: return {"status": "checked"} def get_findings(self) -> list[dict[str, Any]]: return list(self._findings) class ResearchAgent(BaseAgent): def __init__( self, bus: AsyncEventBus, queue: MemoryTaskQueue, identity: AgentIdentity | None = None, permission_gateway: PermissionGateway | None = None, *, authorizer: TaskAuthorizer | None = None, ) -> None: super().__init__(bus, queue, identity, permission_gateway, authorizer) self._task_request_event = RESEARCH_TASK_REQUEST self._task_completed_event = RESEARCH_TASK_COMPLETED self._research: list[dict[str, Any]] = [] self._subscribe() @property def name(self) -> str: return "ResearchAgent" @property def role(self) -> str: return "research" async def execute(self, task_data: dict[str, Any]) -> AgentResult | None: task_id = self.queue.enqueue(self._do_research) try: record = self.queue.get(task_id) assert record is not None await self.queue.dispatch(record) except (KeyError, RuntimeError, asyncio.CancelledError) as exc: logger.warning("ResearchAgent queue dispatch failed: %s", exc) except SecurityError: logger.warning("ResearchAgent permission denied during execution") raise except Exception: logger.exception("ResearchAgent unexpected error during queue dispatch") raise topic = task_data.get("topic", "unknown") self._research.append({"topic": topic}) result = self.publish_result(task_id, True, {"topic": topic, "summary": f"Research on {topic}"}) return result async def _do_research(self) -> dict[str, Any]: return {"status": "researched"} def get_research(self) -> list[dict[str, Any]]: return list(self._research) class BrowserAgent(BaseAgent): def __init__( self, bus: AsyncEventBus, queue: MemoryTaskQueue, identity: AgentIdentity | None = None, permission_gateway: PermissionGateway | None = None, *, authorizer: TaskAuthorizer | None = None, ) -> None: super().__init__(bus, queue, identity, permission_gateway, authorizer) self._task_request_event = BROWSER_TASK_REQUEST self._task_completed_event = BROWSER_TASK_COMPLETED self._browsed: list[dict[str, Any]] = [] self._subscribe() @property def name(self) -> str: return "BrowserAgent" @property def role(self) -> str: return "browser" async def execute(self, task_data: dict[str, Any]) -> AgentResult | None: """Report the browsing capability as unavailable. Security rationale: this agent has no browsing backend. It used to enqueue a no-op coroutine that returned ``{"status": "browsed"}``, then published ``success=True`` with a ``"Browsed "`` summary - a fake success report in a product whose whole point is trustworthy verdicts. A consumer had no way to tell "inspected the page" from "did nothing", and a negative finding that never happened looks exactly like a clean one. The agent now refuses every request with an explicit reason, records ``execution_attempted=False`` (nothing was attempted), and never appends to ``self._browsed``, so absence of evidence can no longer be read as evidence of absence. The role stays registered so the six-role Ptah topology is unchanged; it is a capability stub, not a working browser agent. """ return self.publish_result( self._task_id_from(task_data), False, { "url": task_data.get("url"), "status": "capability_unavailable", "capability": "browsing", "reason": BROWSER_CAPABILITY_UNAVAILABLE_REASON, }, BROWSER_CAPABILITY_UNAVAILABLE_REASON, execution_attempted=False, ) def get_browsed(self) -> list[dict[str, Any]]: """Pages actually fetched. Always empty: ``BrowserAgent`` has no backend, so nothing is ever recorded. Kept so existing callers keep working. """ return list(self._browsed)