Agentswarm/orchestrator/swarm_runtime.py

"""Agent Manager compatible swarm runtime bridge."""
import asyncio
from copy import deepcopy
import hashlib
import hmac
import json
import logging
import os
import time
import uuid
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

import httpx
from pydantic import BaseModel, Field

from .redis_client import redis_client

logger = logging.getLogger(__name__)


class CallbackConfig(BaseModel):
    """Callback settings provided by Agent Manager."""

    url: Optional[str] = None
    signing_secret_ref: Optional[str] = None
    subscribed_events: List[str] = Field(default_factory=list)


class SwarmRun(BaseModel):
    """Runtime-owned representation of a swarm run."""

    deployment_id: str
    swarm_id: str
    mode: str = "swarm"
    status: str
    objective: str
    manager_deployment_id: Optional[str] = None
    correlation_id: Optional[str] = None
    callback: CallbackConfig = Field(default_factory=CallbackConfig)
    task_ids: List[str] = Field(default_factory=list)
    approvals: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
    # Internal peer-communication telemetry (NOT a Manager event — kept off the HM event
    # registry on purpose). Feeds benchmark S_communication. correlation_ids are de-duplicated
    # so a retried/echoed message is not double counted.
    collaboration: Dict[str, Any] = Field(
        default_factory=lambda: {
            "request_correlations": [],   # correlation_ids of peer requests routed
            "reply_correlations": [],     # correlation_ids of peer replies routed
            "delivery_failures": 0,       # peer messages dropped (target not connected)
        }
    )
    # Benchmark Group B: quality grade of the run's generated code (fixture TestPassRate →
    # Q_quality). Populated only when ENABLE_QUALITY_EVAL is set and a fixture is bound. Feeds
    # benchmark reward(); empty dict means not graded (collector leaves reward NaN).
    quality: Dict[str, Any] = Field(default_factory=dict)
    # Benchmark Group A: per-assignment ACO decision telemetry ({task_id, agent_id, tau, eta,
    # p_norm, p_score, explored}). Internal state (NOT a Manager event). Appended only when
    # ENABLE_ACO_DISPATCH selected the assignment; empty → tau/eta/p_decision stay NaN.
    decisions: List[Dict[str, Any]] = Field(default_factory=list)
    # Issue #9: explainable dispatch-scoring decisions (`dispatch.decision_made` payloads:
    # candidate agents + per-dimension score breakdown + exclusion reasons). Internal state
    # (NOT a Manager event — a Manager-facing dispatch event needs event-schema registration,
    # cross-team). Appended only when ENABLE_DISPATCH_SCORE selected the assignment. Audit/
    # benchmark-replayable from here.
    dispatch_decisions: List[Dict[str, Any]] = Field(default_factory=list)
    metadata: Dict[str, Any] = Field(default_factory=dict)
    request_body: Dict[str, Any] = Field(default_factory=dict)
    created_at: float = Field(default_factory=time.time)
    updated_at: float = Field(default_factory=time.time)


class RuntimeValidationError(ValueError):
    """Validation error returned in the Agent Manager runtime envelope."""

    def __init__(self, message: str, code: str = "VALIDATION_ERROR"):
        super().__init__(message)
        self.message = message
        self.code = code


# Frozen client-facing event-type set (agent_swarm#15.2 / HM #28). These 13 types are the
# contract the desktop "task cockpit" renders against; HM (agent_callback.go) registers exactly
# these. The runtime emits other types too (deployment.status_changed, timeline.updated,
# budget.alert, task.heartbeat/blocked/retried/released) and many INTERNAL-only telemetry events
# that are NOT emitted to Manager (see assess_swarm_health / proposals / bids / reviews).
FROZEN_CLIENT_EVENT_TYPES = (
    "task.created",
    "task.claimed",
    "task.running",
    "task.completed",
    "task.failed",
    "handoff.created",
    "approval.requested",
    "approval.approved",
    "approval.rejected",
    "artifact.created",
    "swarm.completed",
    "swarm.failed",
    "swarm.stopped",
    # Review/rework timeline (agent_swarm#34) — emitted as REDACTED client projections from
    # run_cross_review (cross_review.*_client_payload): reviewer id/verdict/criteria/targets/
    # root_cause only, no evidence/summary/reason free text.
    "review.started",
    "review.decision_made",
    "rework.requested",
    "rework.completed",
)


class SwarmRuntime:
    """Stores swarm runs and emits Agent Manager callback events."""

    RUN_KEY_PREFIX = "swarm:"
    TASK_RUN_KEY_PREFIX = "swarm_task:"
    IDEMPOTENCY_KEY_PREFIX = "swarm_idempotency:"
    EVENT_KEY_PREFIX = "swarm_events:"
    EVENT_SEQ_KEY_PREFIX = "swarm_event_seq:"

    def __init__(self):
        self.callback_service_token = (
            os.getenv("AGENT_CALLBACK_SERVICE_TOKEN")
            or os.getenv("AGNET_CALLBACK_SERVICE_TOKEN")
        )
        self.callback_signing_secret = (
            os.getenv("AGENT_CALLBACK_SIGNING_SECRET")
            or os.getenv("AGNET_CALLBACK_SIGNING_SECRET")
        )
        self.runtime_source = os.getenv("SWARM_RUNTIME_SOURCE", "heicode-swarm-runtime")

    def multi_agent_workflow_enabled(self, body: Optional[Dict[str, Any]] = None) -> bool:
        """Return whether the DAG-style runtime workflow is enabled."""
        if os.getenv("ENABLE_SUBTASK_HANDOFF", "false").lower() not in {"1", "true", "yes"}:
            return False
        plan = (body or {}).get("orchestration_plan") or {}
        agents = plan.get("agents") or (body or {}).get("agents") or []
        return len(agents) > 0

    async def health(self) -> Dict[str, Any]:
        """Return Agent Manager compatible runtime health."""
        await redis_client.client.ping()
        return {
            "success": True,
            "data": {
                "status": "healthy",
                "service": "heicode-swarm-runtime",
                "version": "1.0.0",
                "runtime": os.getenv("SWARM_RUNTIME_PLATFORM", "aks"),
                "time": self._now_iso(),
                "capabilities": [
                    "swarm.create",
                    "task.flow",
                    "handoff.events",
                    "artifact.events",
                    "approval.pause_resume",
                    "deployment.stop",
                    "runtime.tasks.query",
                    "runtime.logs.query",
                    "runtime.events.query",
                    "runtime.metrics.query",
                    "runtime.workflow.query",
                    "runtime.diagnostics.query",
                ],
            },
        }

    def normalize_create_request(self, body: Dict[str, Any]) -> Dict[str, Any]:
        """Normalize the Manager swarm request into the runtime request shape."""
        normalized = deepcopy(body or {})
        metadata = dict(normalized.get("metadata") or {})
        callback = dict(normalized.get("callback") or {})
        requirement = dict(normalized.get("requirement") or {})
        model_selection = dict(normalized.get("model_selection") or {})
        existing_plan = dict(normalized.get("orchestration_plan") or {})

        uses_new_shape = normalized.get("mode") == "swarm" or bool(requirement)

        if uses_new_shape:
            primary_model = (
                model_selection.get("primary_model")
                or existing_plan.get("model_id")
                or normalized.get("model_id")
            )
            budget = deepcopy(
                normalized.get("budget")
                or existing_plan.get("budget")
                or {}
            )
            if not existing_plan.get("objective") and requirement.get("objective"):
                existing_plan["objective"] = requirement.get("objective")
            existing_plan.setdefault("sub_mode", "swarm")
            existing_plan.setdefault("risk_level", normalized.get("risk_level") or "medium")
            existing_plan.setdefault("budget", budget)
            existing_plan["model_id"] = primary_model
            existing_plan["requirement"] = self._redact_sensitive(requirement)
            existing_plan["attachments"] = self._redact_sensitive(
                requirement.get("attachments") or []
            )
            existing_plan["constraints"] = self._redact_sensitive(
                requirement.get("constraints") or []
            )
            existing_plan["acceptance_criteria"] = self._redact_sensitive(
                requirement.get("acceptance_criteria") or []
            )
            if not existing_plan.get("agents"):
                existing_plan["agents"] = normalized.get("agents") or []

            normalized["mode"] = "swarm"
            normalized["conversation_id"] = (
                normalized.get("conversation_id")
                or metadata.get("conversation_id")
            )
            normalized["requirement"] = requirement
            normalized["model_selection"] = model_selection
            normalized["model_id"] = primary_model
            normalized["orchestration_plan"] = existing_plan
            if normalized.get("conversation_id"):
                metadata.setdefault("conversation_id", normalized.get("conversation_id"))
            if requirement.get("objective"):
                metadata.setdefault("objective", requirement.get("objective"))
            if model_selection.get("type"):
                metadata.setdefault("model_selection_type", model_selection.get("type"))
        else:
            normalized["orchestration_plan"] = existing_plan

        normalized["metadata"] = metadata
        normalized["callback"] = callback
        return normalized

    def validate_create_request(self, body: Dict[str, Any]):
        """Validate the Manager create request before a runtime run is persisted."""
        normalized = self.normalize_create_request(body)
        plan = normalized.get("orchestration_plan")
        metadata = normalized.get("metadata")
        callback = normalized.get("callback")
        required_paths = {
            "orchestration_plan.objective": plan.get("objective") if isinstance(plan, dict) else None,
            "callback.url": callback.get("url") if isinstance(callback, dict) else None,
            "metadata.manager_deployment_id": metadata.get("manager_deployment_id") if isinstance(metadata, dict) else None,
        }
        missing = [path for path, value in required_paths.items() if value in (None, "", [], {})]
        if missing:
            raise RuntimeValidationError(f"Missing required field(s): {', '.join(missing)}")

        if normalized.get("mode") not in (None, "swarm"):
            raise RuntimeValidationError("mode must be 'swarm'")
        if normalized.get("model_selection"):
            selection_type = (normalized.get("model_selection") or {}).get("type")
            if selection_type not in (None, "primary"):
                raise RuntimeValidationError("model_selection.type must be 'primary' for swarm")

        billing_secret = (normalized.get("billing_context") or {}).get("secret_ref")
        if billing_secret and not self._is_azkv_ref(billing_secret):
            raise RuntimeValidationError("billing_context.secret_ref must use azkv://")

        self._validate_resource_grants(normalized.get("resource_grants") or [], "resource_grants")
        for index, agent in enumerate(plan.get("agents") or []):
            self._validate_resource_grants(
                agent.get("resource_grants") or [],
                f"orchestration_plan.agents[{index}].resource_grants",
            )
        for index, agent in enumerate(normalized.get("agents") or []):
            self._validate_resource_grants(
                agent.get("resource_grants") or [],
                f"agents[{index}].resource_grants",
            )

        for path in ("metadata", "resource_grants", "callback"):
            value = normalized.get(path)
            if value is not None:
                self._reject_plaintext_secrets(value, path)

    async def get_or_create_run(
        self,
        body: Dict[str, Any],
        idempotency_key: Optional[str],
        correlation_id: Optional[str],
    ) -> tuple[SwarmRun, bool]:
        """Create a swarm run or return the previous run for an idempotency key."""
        body = self.normalize_create_request(body)
        if idempotency_key:
            existing_swarm_id = await redis_client.get(
                f"{self.IDEMPOTENCY_KEY_PREFIX}{idempotency_key}"
            )
            if existing_swarm_id:
                existing_run = await self.get_run(existing_swarm_id)
                if existing_run:
                    return existing_run, False

        metadata = body.get("metadata") or {}
        plan = body.get("orchestration_plan") or {}
        callback = CallbackConfig.model_validate(body.get("callback") or {})

        swarm_id = f"swarm-{uuid.uuid4().hex[:12]}"
        deployment_id = f"runtime-dep-{uuid.uuid4().hex[:12]}"
        manager_deployment_id = (
            metadata.get("manager_deployment_id")
            or metadata.get("heicode_deployment_id")
        )
        objective = (
            plan.get("objective")
            or metadata.get("objective")
            or "Swarm runtime task"
        )

        run = SwarmRun(
            deployment_id=deployment_id,
            swarm_id=swarm_id,
            mode="swarm",
            status="running",
            objective=objective,
            manager_deployment_id=manager_deployment_id,
            correlation_id=correlation_id or metadata.get("correlation_id"),
            callback=callback,
            metadata=self._redact_sensitive(metadata),
            request_body=self._redact_sensitive(body),
        )

        if self._requires_approval(body):
            approval = self._build_approval(body, run)
            run.status = "waiting_approval"
            run.approvals[approval["approval_id"]] = approval

        await self.save_run(run)

        if idempotency_key:
            await redis_client.set(
                f"{self.IDEMPOTENCY_KEY_PREFIX}{idempotency_key}",
                run.swarm_id,
                ex=86400,
            )

        await self.emit_event(
            run,
            "deployment.status_changed",
            payload=self.status_payload(run, phase="Plan"),
        )

        for approval in run.approvals.values():
            await self.emit_event(run, "approval.requested", payload=approval)

        return run, True

    async def save_run(self, run: SwarmRun):
        """Persist a swarm run."""
        run.updated_at = time.time()
        await redis_client.set(
            f"{self.RUN_KEY_PREFIX}{run.swarm_id}",
            run.model_dump_json(),
        )

    async def get_run(self, swarm_id: str) -> Optional[SwarmRun]:
        """Get a swarm run by id."""
        data = await redis_client.get(f"{self.RUN_KEY_PREFIX}{swarm_id}")
        if not data:
            return None
        return SwarmRun.model_validate_json(data)

    async def get_run_by_identifier(self, identifier: str) -> Optional[SwarmRun]:
        """Find a run by swarm id, runtime deployment id, or manager deployment id."""
        direct = await self.get_run(identifier)
        if direct:
            return direct

        for key in await redis_client.keys(f"{self.RUN_KEY_PREFIX}*"):
            data = await redis_client.get(key)
            if not data:
                continue
            run = SwarmRun.model_validate_json(data)
            if identifier in {run.deployment_id, run.manager_deployment_id, run.swarm_id}:
                return run
        return None

    async def get_run_for_task(self, task_id: str) -> Optional[SwarmRun]:
        """Find the swarm run that owns a task."""
        swarm_id = await redis_client.get(f"{self.TASK_RUN_KEY_PREFIX}{task_id}")
        if not swarm_id:
            return None
        return await self.get_run(swarm_id)

    async def record_peer_message(
        self,
        run: SwarmRun,
        *,
        correlation_id: Optional[str],
        is_reply: bool,
        delivered: bool,
    ) -> None:
        """Record one routed peer-collaboration message for S_communication.

        Internal telemetry only: counters live on the run, not on the Manager event stream.
        Requests and replies are tracked by correlation_id so the collector can compute the
        share of requests that received an answer (a real communication-success rate).
        """
        collab = run.collaboration or {}
        requests = collab.setdefault("request_correlations", [])
        replies = collab.setdefault("reply_correlations", [])
        if correlation_id:
            bucket = replies if is_reply else requests
            if correlation_id not in bucket:
                bucket.append(correlation_id)
        if not delivered:
            collab["delivery_failures"] = int(collab.get("delivery_failures", 0)) + 1
        run.collaboration = collab
        await self.save_run(run)

    async def record_quality(self, run: SwarmRun, quality: Dict[str, Any]) -> None:
        """Store the benchmark quality grade for a completed run (Group B)."""
        run.quality = quality or {}
        await self.save_run(run)

    async def record_decision(self, run: SwarmRun, decision: Dict[str, Any]) -> None:
        """Append one ACO assignment decision (Group A telemetry). Bounded list."""
        run.decisions = (run.decisions or [])[-999:] + [decision]
        await self.save_run(run)

    async def record_dispatch_decision(self, run: SwarmRun, decision: Dict[str, Any]) -> None:
        """Append one explainable dispatch-scoring decision (#9). Internal, bounded list."""
        run.dispatch_decisions = (run.dispatch_decisions or [])[-999:] + [decision]
        await self.save_run(run)

    async def attach_task(self, run: SwarmRun, task_id: str):
        """Associate a queue task with a swarm run."""
        if task_id not in run.task_ids:
            run.task_ids.append(task_id)
        await redis_client.set(f"{self.TASK_RUN_KEY_PREFIX}{task_id}", run.swarm_id)
        await self.save_run(run)

    async def stop_run(self, deployment_id: str, reason: str = "") -> Optional[SwarmRun]:
        """Stop a run by runtime deployment id or swarm id."""
        run = await self.find_run_by_deployment_id(deployment_id)
        if not run:
            return None

        run.status = "stopped"
        await self.save_run(run)
        await self.emit_event(
            run,
            "deployment.status_changed",
            payload=self.status_payload(
                run,
                phase="Deliver",
                reason=reason or "Heicode Manager requested stop",
            ),
        )
        # Frozen terminal swarm event (agent_swarm#15.2): the client cockpit keys its terminal
        # banner off swarm.{completed|failed|stopped}, not deployment.status_changed.
        await self.emit_event(run, "swarm.stopped", payload={
            "status": "stopped",
            "reason": reason or "Heicode Manager requested stop",
        })
        return run

    async def find_run_by_deployment_id(self, deployment_id: str) -> Optional[SwarmRun]:
        """Find a run by runtime deployment id or swarm id."""
        return await self.get_run_by_identifier(deployment_id)

    async def record_approval_decision(
        self,
        swarm_id: str,
        approval_id: str,
        decision: Dict[str, Any],
    ) -> Optional[SwarmRun]:
        """Persist an approval decision and update run state."""
        run = await self.get_run_by_identifier(swarm_id)
        if not run:
            return None

        approval = run.approvals.get(approval_id, {"approval_id": approval_id})
        approval["decision"] = decision.get("decision")
        credential_lease = decision.get("credential_lease") or {}
        approval["credential_ref"] = (
            decision.get("credential_ref") or credential_lease.get("credential_ref")
        )
        approval["lease_id"] = decision.get("lease_id") or credential_lease.get("lease_id")
        approval["lease_expires_at"] = (
            decision.get("lease_expires_at") or credential_lease.get("expires_at")
        )
        approval["decided_at"] = self._now_iso()
        run.approvals[approval_id] = self._redact_sensitive(approval)

        if decision.get("decision") == "approved":
            run.status = "running"
        elif decision.get("decision") == "rejected":
            run.status = "blocked"

        await self.save_run(run)
        await self.emit_event(
            run,
            "deployment.status_changed",
            payload=self.status_payload(
                run,
                phase="Review",
                approval_id=approval_id,
                decision=decision.get("decision"),
                reason=decision.get("reason"),
            ),
        )
        # Frozen approval-outcome events (agent_swarm#15.2): the client renders an approval
        # timeline from approval.requested -> approval.approved|rejected.
        decision_value = decision.get("decision")
        if decision_value in ("approved", "rejected"):
            await self.emit_event(run, f"approval.{decision_value}", payload={
                "approval_id": approval_id,
                "decision": decision_value,
                "reason": decision.get("reason"),
            })
        if run.status == "blocked":
            await self.emit_event(run, "task.blocked", payload={
                "approval_id": approval_id,
                "reason": decision.get("reason") or "Approval rejected",
                "runtime_deployment_id": run.deployment_id,
            })
            await self.emit_event(run, "timeline.updated", payload={
                "summary": "Swarm blocked by approval rejection",
                "approval_id": approval_id,
            })
        return run

    async def emit_event(
        self,
        run: SwarmRun,
        event_type: str,
        task_id: Optional[str] = None,
        agent_instance_id: Optional[str] = None,
        payload: Optional[Dict[str, Any]] = None,
        artifact: Optional[Dict[str, Any]] = None,
    ):
        """Record and optionally emit a Manager callback event."""
        callback = run.callback
        occurred_at = self._now_iso()
        redacted_payload = self._redact_sensitive(payload or {})
        redacted_artifact = self._redact_sensitive(artifact) if artifact else None

        if redacted_artifact:
            # Freeze (agent_swarm#15.4): the client artifact view needs a stable flat shape
            # {uri, checksum, task_id, size_bytes, created_at}. Default created_at to the event time;
            # leave size_bytes absent when unknown (rule #9: do not fabricate a size).
            redacted_artifact.setdefault("created_at", occurred_at)
            if task_id and "task_id" not in redacted_artifact:
                redacted_artifact["task_id"] = task_id

        if redacted_artifact and event_type == "artifact.created":
            redacted_payload = {
                **redacted_artifact,
                **redacted_payload,
            }

        # Conform to HM event contract required fields (heicode agent_callback.go):
        # timeline.updated requires `title`; budget.alert requires `threshold_pct`.
        if event_type == "timeline.updated" and "title" not in redacted_payload:
            redacted_payload["title"] = redacted_payload.get("summary") or "timeline"
        if event_type == "budget.alert" and "threshold_pct" not in redacted_payload:
            threshold = redacted_payload.get("threshold")
            if isinstance(threshold, (int, float)):
                redacted_payload["threshold_pct"] = threshold * 100 if threshold <= 1 else threshold

        event_id = f"evt_{uuid.uuid4().hex}"
        # Per-swarm strictly-increasing sequence (agent_swarm#15.1): the client polls
        # `events?after=<sequence>` and dedups/orders by it. INCR is atomic so concurrent emits on
        # the same swarm get distinct, gap-free numbers starting at 1.
        sequence = await redis_client.incr(f"{self.EVENT_SEQ_KEY_PREFIX}{run.swarm_id}")
        body: Dict[str, Any] = {
            "event_id": event_id,
            "idempotency_key": event_id,
            "sequence": sequence,
            "event_type": event_type,
            "deployment_id": run.manager_deployment_id or run.deployment_id,
            "runtime_deployment_id": run.deployment_id,
            "swarm_id": run.swarm_id,
            "agent_instance_id": agent_instance_id,
            "task_id": task_id,
            "occurred_at": occurred_at,
            "correlation_id": run.correlation_id,
            "source": self.runtime_source,
            "payload": redacted_payload,
        }
        if redacted_artifact:
            body["artifact"] = redacted_artifact

        raw_body = json.dumps(body, ensure_ascii=False, separators=(",", ":"))
        await self._store_event(run.swarm_id, raw_body)

        if not callback.url:
            return
        if callback.subscribed_events and event_type not in callback.subscribed_events:
            return

        headers = {
            "Content-Type": "application/json",
            "X-Agent-Event-Id": event_id,
            "X-Agnet-Event-Id": event_id,
        }
        if run.correlation_id:
            headers["X-Correlation-ID"] = run.correlation_id
        if self.callback_service_token:
            headers["X-Agent-Service-Token"] = self.callback_service_token
            headers["X-Agnet-Service-Token"] = self.callback_service_token
        if self.callback_signing_secret:
            timestamp = str(int(time.time() * 1000))
            signature_payload = f"{timestamp}.{event_id}.{raw_body}"
            signature = hmac.new(
                self.callback_signing_secret.encode("utf-8"),
                signature_payload.encode("utf-8"),
                hashlib.sha256,
            ).hexdigest()
            headers["X-Agent-Timestamp"] = timestamp
            headers["X-Agent-Signature"] = f"sha256={signature}"
            headers["X-Agnet-Timestamp"] = timestamp
            headers["X-Agnet-Signature"] = f"sha256={signature}"

        if callback.url and (not callback.subscribed_events or event_type in callback.subscribed_events):
            await self._upsert_callback_attempt(
                run.swarm_id,
                event_id,
                {
                    "event_id": event_id,
                    "event_type": event_type,
                    "url": callback.url,
                    "status": "pending",
                    "attempted_at": self._now_iso(),
                },
            )
        asyncio.create_task(
            self._post_callback(run.swarm_id, callback.url, raw_body, headers, event_type, event_id)
        )

    async def list_events(
        self,
        swarm_id: str,
        limit: int = 100,
        cursor: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Return stored runtime events for a swarm."""
        start = int(cursor or 0)
        safe_limit = max(1, min(limit, 500))
        items = await redis_client.lrange(
            f"{self.EVENT_KEY_PREFIX}{swarm_id}",
            start,
            start + safe_limit - 1,
        )
        events = [json.loads(item) for item in items]
        next_cursor = str(start + safe_limit) if len(events) == safe_limit else None
        return {"events": events, "next_cursor": next_cursor}

    async def _store_event(self, swarm_id: str, raw_body: str):
        await redis_client.rpush(f"{self.EVENT_KEY_PREFIX}{swarm_id}", raw_body)

    async def _post_callback(
        self,
        swarm_id: str,
        url: str,
        raw_body: str,
        headers: Dict[str, str],
        event_type: str,
        event_id: str,
    ):
        """Send callback without blocking the agent WebSocket loop."""
        if not url:
            return
        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                response = await client.post(url, content=raw_body, headers=headers)
                await self._upsert_callback_attempt(
                    swarm_id,
                    event_id,
                    {
                        "event_id": event_id,
                        "event_type": event_type,
                        "url": url,
                        "status": "delivered" if response.status_code < 400 else "failed",
                        "response_status": response.status_code,
                        "attempted_at": self._now_iso(),
                    },
                )
                if response.status_code >= 400:
                    logger.warning(
                        "Manager callback %s failed with status %s: %s",
                        event_type,
                        response.status_code,
                        response.text[:500],
                    )
        except Exception as exc:
            await self._upsert_callback_attempt(
                swarm_id,
                event_id,
                {
                    "event_id": event_id,
                    "event_type": event_type,
                    "url": url,
                    "status": "failed",
                    "error": str(exc),
                    "attempted_at": self._now_iso(),
                },
            )
            logger.warning("Manager callback %s failed: %s", event_type, exc)

    def build_task_descriptions(self, body: Dict[str, Any]) -> List[Dict[str, Any]]:
        """Create a simple task graph from a Manager orchestration request."""
        body = self.normalize_create_request(body)
        plan = body.get("orchestration_plan") or {}
        agents = plan.get("agents") or body.get("agents") or []
        requirement = body.get("requirement") or {}
        objective = (
            requirement.get("objective")
            or plan.get("objective")
            or "Complete swarm objective"
        )
        multi_agent_enabled = self.multi_agent_workflow_enabled(body)
        context = {
            "orchestration_plan": self._redact_sensitive(plan),
            "resource_grants": self._redact_sensitive(body.get("resource_grants") or []),
            "sub_mode": body.get("sub_mode") or plan.get("sub_mode"),
            "conversation_id": body.get("conversation_id"),
            "mode": body.get("mode") or "swarm",
            "requirement": self._redact_sensitive(requirement),
        }

        if not agents or not multi_agent_enabled:
            return [{
                "task_id": "task-1",
                "title": "Swarm objective",
                "description": objective,
                "agent_role": "general",
                "required_capabilities": ["general"],
                "depends_on": [],
                "parent_task_id": None,
                "root_task_id": "task-1",
                "source": "runtime_bridge",
                "workflow_mode": "single_agent",
                "allow_handoff": False,
                "context": context,
            }]

        tasks = []
        for index, agent in enumerate(agents, start=1):
            role = agent.get("role") or f"agent-{index}"
            title = agent.get("title") or f"{role} task"
            description = agent.get("description") or f"[{role}] {objective}"
            task_id = agent.get("task_id") or f"task-{index}"
            required_capabilities = agent.get("required_capabilities") or [role]
            tasks.append({
                "task_id": task_id,
                "title": title,
                "description": description,
                "agent_role": role,
                "required_capabilities": required_capabilities,
                "depends_on": agent.get("depends_on") or [],
                "parent_task_id": None,
                "root_task_id": task_id,
                "source": "runtime_bridge",
                "workflow_mode": "multi_agent",
                "allow_handoff": True,
                "context": {
                    **context,
                    "agent_role": role,
                    "workflow_mode": "multi_agent",
                    "resource_grants": self._redact_sensitive(
                        agent.get("resource_grants") or []
                    ),
                },
            })
        return tasks

    def task_event_payload(self, task: Any, task_spec: Dict[str, Any]) -> Dict[str, Any]:
        """Return the minimum task graph payload required by Manager."""
        return {
            "task_id": task.task_id,
            "title": task.title or task_spec.get("title", task.description[:80]),
            "description": task.description,
            "agent_role": task.agent_role or task_spec.get("agent_role", "general"),
            "status": task.status.value if hasattr(task.status, "value") else task.status,
            "depends_on": task.depends_on or task_spec.get("depends_on") or [],
            "parent_task_id": task.parent_task_id or task_spec.get("parent_task_id"),
            "root_task_id": task.root_task_id or task_spec.get("root_task_id"),
            "required_capabilities": task.required_capabilities or task_spec.get("required_capabilities") or [],
            "source": task.source or task_spec.get("source", "runtime_bridge"),
            "attempt": getattr(task, "retry_count", 0),
        }

    def _requires_approval(self, body: Dict[str, Any]) -> bool:
        plan = body.get("orchestration_plan") or {}
        agile = plan.get("agile_context") or {}
        risk = str(plan.get("risk_level") or body.get("risk_level") or "").lower()
        return bool(agile.get("requires_user_approval")) or risk == "high"

    def status_payload(
        self,
        run: SwarmRun,
        phase: str,
        reason: Optional[str] = None,
        approval_id: Optional[str] = None,
        decision: Optional[str] = None,
        deliverable: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """Build the standard runtime status payload."""
        payload: Dict[str, Any] = {
            "status": run.status,
            "runtime_execution_status": run.status,
            "runtime_deployment_id": run.deployment_id,
            "manager_deployment_id": run.manager_deployment_id,
            "phase": phase,
        }
        if reason:
            payload["reason"] = reason
        if approval_id:
            payload["approval_id"] = approval_id
        if decision:
            payload["decision"] = decision
        if deliverable is not None:
            payload["deliverable"] = deliverable
        return payload

    async def _upsert_callback_attempt(
        self,
        swarm_id: str,
        event_id: str,
        attempt: Dict[str, Any],
    ):
        run = await self.get_run(swarm_id)
        if not run:
            return
        callback_attempts = list(run.metadata.get("callback_attempts") or [])
        filtered = [item for item in callback_attempts if item.get("event_id") != event_id]
        filtered.append(self._redact_sensitive(attempt))
        run.metadata["callback_attempts"] = filtered[-50:]
        await self.save_run(run)

    def _build_approval(self, body: Dict[str, Any], run: SwarmRun) -> Dict[str, Any]:
        grants = body.get("resource_grants") or []
        grant = grants[0] if grants else {}
        return {
            "approval_id": f"appr_{uuid.uuid4().hex[:12]}",
            "operation": "git.write",
            "resource_id": grant.get("resource_id", "repo-main"),
            "resource_type": grant.get("resource_type", "git"),
            "resource_scope": ",".join(grant.get("permission_scope") or []),
            "target_role": grant.get("target_role", "general"),
            "risk_level": "high",
            "requires_credential": True,
            "secret_ref": grant.get("secret_ref") or grant.get("ref"),
            "ttl_seconds": 900,
            "reason": "High-risk swarm run requires Manager approval",
            "runtime_deployment_id": run.deployment_id,
        }

    def _validate_resource_grants(self, grants: List[Dict[str, Any]], path: str):
        for index, grant in enumerate(grants):
            for key in ("secret_ref", "ref"):
                value = grant.get(key)
                if value and not self._is_azkv_ref(value):
                    raise RuntimeValidationError(f"{path}[{index}].{key} must use azkv://")

    def _reject_plaintext_secrets(self, value: Any, path: str):
        if isinstance(value, dict):
            for key, item in value.items():
                child_path = f"{path}.{key}"
                key_lower = key.lower()
                is_ref_key = key_lower.endswith("_ref") or key_lower == "ref"
                if self._looks_sensitive_key(key_lower) and not is_ref_key:
                    raise RuntimeValidationError(
                        f"Plaintext secret-like field is not allowed: {child_path}"
                    )
                if is_ref_key and isinstance(item, str) and item and not self._is_azkv_ref(item):
                    raise RuntimeValidationError(f"{child_path} must use azkv://")
                self._reject_plaintext_secrets(item, child_path)
        elif isinstance(value, list):
            for index, item in enumerate(value):
                self._reject_plaintext_secrets(item, f"{path}[{index}]")

    def _looks_sensitive_key(self, key: str) -> bool:
        return (
            any(part in key for part in ("token", "password", "passwd", "secret", "private_key", "api_key"))
            or key.endswith("_key")
        )

    def _is_azkv_ref(self, value: Any) -> bool:
        return isinstance(value, str) and value.startswith("azkv://")

    def _redact_sensitive(self, value: Any) -> Any:
        """Remove obvious plaintext secrets from callback-safe payloads."""
        sensitive_keys = {
            "password",
            "passwd",
            "token",
            "api_token",
            "access_token",
            "refresh_token",
            "private_key",
            "access_key",
            "secret",
            "client_secret",
            "connection_string",
        }
        if isinstance(value, dict):
            redacted = {}
            for key, item in value.items():
                key_lower = key.lower()
                if key_lower in {"secret_ref", "credential_ref", "signing_secret_ref"}:
                    redacted[key] = item
                elif key_lower in sensitive_keys or key_lower.endswith(("_token", "_secret", "_password", "_key")):
                    redacted[key] = "[redacted]"
                else:
                    redacted[key] = self._redact_sensitive(item)
            return redacted
        if isinstance(value, list):
            return [self._redact_sensitive(item) for item in value]
        return value

    def _now_iso(self) -> str:
        return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


swarm_runtime = SwarmRuntime()