AFA-LIBERTAS Integration Mapping¶

┌─────────────────────────────────────────────────────────────────────────────┐
│                     AFA ↔ LIBERTAS INTEGRATION                              │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  AFA Data Plane                    LIBERTAS                   AFA Control   │
│  ┌─────────────┐                   ┌─────────────┐            Plane         │
│  │  Security   │                   │  Workflow   │            ┌───────────┐ │
│  │  Analyzer   │───────┐           │  Engine     │◀──────────▶│ RepoGuard │ │
│  ├─────────────┤       │           ├─────────────┤            ├───────────┤ │
│  │ Performance │       │           │   Actors    │            │  Quality  │ │
│  │  Analyzer   │       │           │  - AI       │            │  Metrics  │ │
│  ├─────────────┤       │           │  - Human    │            ├───────────┤ │
│  │Maintainabili│       └──────────▶│  - Hybrid   │◀──────────▶│   Cost    │ │
│  │ty Expert    │                   ├─────────────┤            │  Limiter  │ │
│  ├─────────────┤                   │  Handoffs   │            ├───────────┤ │
│  │Documentation│                   │  - Review   │            │ Security  │ │
│  │   Agent     │                   │  - Override │            │   Gate    │ │
│  ├─────────────┤                   │  - Escalate │            ├───────────┤ │
│  │Optimization │                   └─────────────┘            │ Learning  │ │
│  │ Generator   │                                              │ Feedback  │ │
│  └─────────────┘                                              └───────────┘ │
│        │                                                            │        │
│        │              pcw_decide()                                  │        │
│        └────────────────────────────────────────────────────────────┘        │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

async def pcw_decide(
    candidates: List[CodeProposal],
    context: EnhancementContext
) -> CodeProposal:
    """
    Primary Confidence-Weighted Decision function.
    Selects the optimal code proposal from candidates.
    """
    # Evaluate each candidate through gates
    scored = []
    for candidate in candidates:
        security_score = await security_gate.evaluate(candidate)
        if not security_score.passes:
            continue  # Hard stop

        quality_score = await quality_metrics.evaluate(candidate)
        cost_estimate = await cost_limiter.estimate(candidate)

        combined = combine_scores(
            security=security_score,
            quality=quality_score,
            cost=cost_estimate,
            weights=context.weights
        )
        scored.append((candidate, combined))

    # Select highest scoring proposal
    if not scored:
        return None

    return max(scored, key=lambda x: x[1])[0]

from libertas_opus import WorkflowEngine, WorkflowContext, Actor
from libertas_opus.protocols import HandoffProtocol

async def pcw_decide(
    candidates: List[CodeProposal],
    context: AEGISContext
) -> DecisionResult:
    """
    Confidence-Weighted Decision integrated with LIBERTAS orchestration.

    This function bridges AFA's candidate generation with LIBERTAS
    workflow-based evaluation and approval.
    """

    # Step 1: Create LIBERTAS workflow context
    workflow_context = WorkflowContext(
        workflow_id="aegis_proposal_evaluation",
        initial_data={
            "candidates": [c.to_dict() for c in candidates],
            "aegis_context": context.to_dict(),
            "timestamp": datetime.utcnow().isoformat(),
            "request_id": str(uuid.uuid4())
        },
        metadata={
            "source": "afa_data_plane",
            "version": "3.0-RO"
        }
    )

    # Step 2: Load workflow definition
    workflow = WorkflowRegistry.get("aegis_proposal_evaluation")

    # Step 3: Initialize engine with actors
    engine = WorkflowEngine(
        workflow=workflow,
        context=workflow_context,
        actors={
            "ai_analyzer": AIAnalyzerActor(context.model_config),
            "human_reviewer": HumanReviewerActor(context.review_channel),
            "governance_pair": GovernancePairActor(context.governance_config)
        }
    )

    # Step 4: Execute workflow
    result = await engine.run()

    # Step 5: Translate result
    if result.status == WorkflowStatus.COMPLETED:
        outputs = result.outputs

        if "approved_proposal" in outputs:
            return DecisionResult(
                decision=Decision.APPROVE,
                proposal=CodeProposal.from_dict(outputs["approved_proposal"]),
                confidence=outputs.get("confidence_score", 0.0),
                approval_path=outputs.get("approval_path"),
                audit_trail=result.audit_entries
            )
        else:
            return DecisionResult(
                decision=Decision.REJECT,
                reason=outputs.get("rejection_reason", "Unknown"),
                gate_results=outputs.get("gate_results"),
                audit_trail=result.audit_entries
            )

    elif result.status == WorkflowStatus.FAILED:
        return DecisionResult(
            decision=Decision.ERROR,
            reason=f"Workflow failed: {result.error}",
            audit_trail=result.audit_entries
        )

    else:
        return DecisionResult(
            decision=Decision.PENDING,
            reason="Workflow in progress",
            workflow_id=result.workflow_id
        )

# /src/workflows/templates/aegis_proposal_evaluation.yaml

workflow:
  id: aegis_proposal_evaluation
  version: "1.0.0"
  description: "End-to-end AEGIS proposal evaluation pipeline"

  config:
    max_concurrent_tasks: 5
    timeout_seconds: 300
    retry_policy:
      max_attempts: 3
      backoff_multiplier: 2

  actors:
    - id: ai_analyzer
      type: AI
      capabilities:
        - analyze
        - score
        - recommend
        - generate
      config:
        model: "gpt-4-turbo"
        temperature: 0.1
        max_tokens: 4096

    - id: human_reviewer
      type: HUMAN
      capabilities:
        - review
        - approve
        - reject
        - override_request
      config:
        notification_channel: slack
        timeout_hours: 24

    - id: governance_pair
      type: GOVERNANCE
      capabilities:
        - two_key_override
        - emergency_halt
      config:
        required_roles:
          - risk_lead
          - security_lead
        signature_format: Hybrid-PQ (Ed25519+ML-DSA-44) or BIP-322
        key_storage: ML-KEM-768 encrypted

  tasks:
    # ===== Layer 0: Invariants =====
    - id: security_gate
      name: "Security Invariant Check"
      layer: 0
      actor: ai_analyzer
      type: gate
      inputs:
        - candidates
      outputs:
        - security_results
        - passed_candidates
      config:
        sast_enabled: true
        sca_enabled: true
        slsa_level: 3
      on_failure:
        action: terminate
        reason: "Security invariant violation"

    # ===== Layer 1: Policy =====
    - id: policy_evaluation
      name: "DOS Policy Evaluation"
      layer: 1
      actor: ai_analyzer
      type: computation
      depends_on:
        - security_gate
      inputs:
        - passed_candidates
        - aegis_context
      outputs:
        - utility_scores
        - decision_paths
        - three_point_estimates
        - complexity_breakdown

    # ===== Layer 2: Gates =====
    - id: quantitative_gates
      name: "Guardrail Gate Evaluation"
      layer: 2
      actor: ai_analyzer
      type: gate
      depends_on:
        - policy_evaluation
      inputs:
        - passed_candidates
        - utility_scores
        - complexity_breakdown
      outputs:
        - gate_results
        - confidence_scores
        - best_candidate
      config:
        gates:
          - risk_gate
          - profit_gate
          - novelty_gate
          - complexity_floor
          - quality_gate
          - utility_lcb

    # ===== Decision Routing =====
    - id: decision_routing
      name: "Route to Approval Path"
      type: decision
      depends_on:
        - quantitative_gates
      inputs:
        - gate_results
        - confidence_scores
        - best_candidate
      branches:
        - condition: "all_gates_passed AND min_confidence >= 0.95"
          next: auto_approve
          outputs:
            approval_path: "AUTO"

        - condition: "any_gate_failed AND is_override_eligible"
          next: human_review
          outputs:
            approval_path: "HUMAN_REVIEW"

        - condition: "any_gate_failed AND NOT is_override_eligible"
          next: auto_reject
          outputs:
            approval_path: "REJECTED"

    # ===== Layer 3: Human Review Path =====
    - id: human_review
      name: "Human Review"
      layer: 3
      actor: human_reviewer
      type: review
      handoff:
        protocol: AI_TO_HUMAN_REVIEW
        context_fields:
          - best_candidate
          - gate_results
          - utility_scores
          - rejection_reasons
      outputs:
        - review_decision
        - reviewer_notes
      branches:
        - condition: "review_decision == APPROVE"
          next: approved_by_human

        - condition: "review_decision == REQUEST_OVERRIDE"
          next: two_key_override

        - condition: "review_decision == REJECT"
          next: rejected_by_human

    # ===== Two-Key Override =====
    - id: two_key_override
      name: "Two-Key Override"
      layer: 3
      actor: governance_pair
      type: override
      handoff:
        protocol: TWO_KEY_OVERRIDE
        context_fields:
          - best_candidate
          - gate_results
          - review_decision
          - override_rationale
      config:
        dual_signature: true
        signature_format: Hybrid-PQ (Ed25519+ML-DSA-44) or BIP-322
        key_storage: ML-KEM-768 encrypted (KeyStoreRepository)
        required_roles:
          - risk_lead
          - security_lead
        audit_immutable: true
        pq_resistant: true
      outputs:
        - override_decision
        - signatures
        - audit_entry
      branches:
        - condition: "override_decision == APPROVE"
          next: approved_by_override

        - condition: "override_decision == REJECT"
          next: rejected_by_override

    # ===== Terminal States =====
    - id: auto_approve
      name: "Automatic Approval"
      type: terminal
      outputs:
        - approved_proposal: "best_candidate"
        - confidence_score: "min_confidence"
        - approval_path: "AUTO"

    - id: approved_by_human
      name: "Approved by Human Review"
      type: terminal
      outputs:
        - approved_proposal: "best_candidate"
        - confidence_score: "review_confidence"
        - approval_path: "HUMAN_REVIEW"
        - reviewer_notes: "reviewer_notes"

    - id: approved_by_override
      name: "Approved by Two-Key Override"
      type: terminal
      outputs:
        - approved_proposal: "best_candidate"
        - approval_path: "TWO_KEY_OVERRIDE"
        - override_signatures: "signatures"
        - override_rationale: "override_rationale"

    - id: auto_reject
      name: "Automatic Rejection"
      type: terminal
      outputs:
        - rejection_reason: "gate_failure_summary"
        - gate_results: "gate_results"

    - id: rejected_by_human
      name: "Rejected by Human Review"
      type: terminal
      outputs:
        - rejection_reason: "review_rejection_reason"
        - reviewer_notes: "reviewer_notes"

    - id: rejected_by_override
      name: "Override Rejected"
      type: terminal
      outputs:
        - rejection_reason: "override_rejection_reason"
        - audit_entry: "audit_entry"

  telemetry:
    emit_on:
      - task_start
      - task_complete
      - decision_branch
      - handoff
      - terminal
    fields:
      - workflow_id
      - task_id
      - actor_id
      - timestamp
      - duration_ms
      - inputs_hash
      - outputs_hash

# /src/workflows/templates/aegis_override.yaml

workflow:
  id: aegis_two_key_override
  version: "1.0.0"
  description: "BIP-322 dual-signature override for gate failures"

  config:
    timeout_seconds: 3600  # 1 hour
    audit_level: maximum

  actors:
    - id: risk_lead
      type: GOVERNANCE
      role: risk_lead
      capabilities:
        - first_key_approval
        - risk_assessment

    - id: security_lead
      type: GOVERNANCE
      role: security_lead
      capabilities:
        - second_key_approval
        - security_assessment

  tasks:
    - id: validate_request
      name: "Validate Override Request"
      type: validation
      inputs:
        - proposal
        - gate_results
        - override_rationale
      outputs:
        - validation_result
        - override_eligible
      config:
        checks:
          - has_rationale
          - not_security_invariant  # Cannot override Layer 0
          - within_budget

    - id: first_key_approval
      name: "Risk Lead Approval"
      actor: risk_lead
      type: approval
      depends_on:
        - validate_request
      condition: "override_eligible == true"
      handoff:
        protocol: GOVERNANCE_REVIEW
        required_fields:
          - risk_assessment
          - mitigation_plan
      outputs:
        - first_signature
        - first_key_timestamp

    - id: second_key_approval
      name: "Security Lead Approval"
      actor: security_lead
      type: approval
      depends_on:
        - first_key_approval
      condition: "first_signature != null"
      handoff:
        protocol: GOVERNANCE_REVIEW
        required_fields:
          - security_assessment
          - compliance_check
      outputs:
        - second_signature
        - second_key_timestamp

    - id: verify_signatures
      name: "Verify BIP-322 Signatures"
      type: cryptographic
      depends_on:
        - second_key_approval
      inputs:
        - first_signature
        - second_signature
        - proposal_hash
      outputs:
        - signatures_valid
        - verification_proof
      config:
        signature_format: BIP-322
        hash_algorithm: SHA-256

    - id: create_audit_entry
      name: "Create Immutable Audit Entry"
      type: audit
      depends_on:
        - verify_signatures
      condition: "signatures_valid == true"
      inputs:
        - proposal
        - first_signature
        - second_signature
        - risk_assessment
        - security_assessment
        - override_rationale
      outputs:
        - audit_entry_id
        - merkle_proof
      config:
        storage: immutable
        replication: cross_region

    - id: override_approved
      name: "Override Approved"
      type: terminal
      depends_on:
        - create_audit_entry
      outputs:
        - override_decision: "APPROVE"
        - signatures:
            first: "first_signature"
            second: "second_signature"
        - audit_entry: "audit_entry_id"

    - id: override_rejected
      name: "Override Rejected"
      type: terminal
      trigger: any_failure
      outputs:
        - override_decision: "REJECT"
        - rejection_reason: "failure_reason"

# /src/actors/ai_analyzer.py

from libertas_opus import Actor, ActorType, TaskContext
from libertas_opus.capabilities import Capability
from typing import Dict, Any, List

class AIAnalyzerActor(Actor):
    """AI-powered analysis and scoring actor."""

    actor_type = ActorType.AI
    capabilities = [
        Capability.ANALYZE,
        Capability.SCORE,
        Capability.RECOMMEND,
        Capability.GENERATE
    ]

    def __init__(self, model_config: Dict[str, Any]):
        super().__init__()
        self.model = model_config.get("model", "gpt-4-turbo")
        self.temperature = model_config.get("temperature", 0.1)
        self.client = self._init_client(model_config)

    async def execute_task(
        self,
        task: TaskContext
    ) -> Dict[str, Any]:
        """Execute task based on task type."""

        if task.type == "gate":
            return await self._execute_gate(task)
        elif task.type == "computation":
            return await self._execute_computation(task)
        else:
            raise ValueError(f"Unknown task type: {task.type}")

    async def _execute_gate(
        self,
        task: TaskContext
    ) -> Dict[str, Any]:
        """Execute security or quantitative gate."""

        candidates = task.inputs.get("candidates", [])
        results = []

        for candidate in candidates:
            # Security analysis
            if task.id == "security_gate":
                sast = await self._run_sast(candidate)
                sca = await self._run_sca(candidate)
                slsa = await self._verify_slsa(candidate)

                passed = (
                    sast.critical_count == 0 and
                    sca.vulnerability_count == 0 and
                    slsa.level >= task.config.get("slsa_level", 3)
                )

                results.append({
                    "candidate_id": candidate["id"],
                    "passed": passed,
                    "sast_findings": sast.findings,
                    "sca_findings": sca.findings,
                    "slsa_level": slsa.level
                })

            # Quantitative gates
            elif task.id == "quantitative_gates":
                gate_results = {}
                utility_scores = task.inputs.get("utility_scores", {})

                for gate in task.config.get("gates", []):
                    gate_result = await self._evaluate_gate(
                        gate, candidate, utility_scores
                    )
                    gate_results[gate] = gate_result

                all_passed = all(g["passed"] for g in gate_results.values())
                min_confidence = min(
                    g.get("confidence", 1.0) for g in gate_results.values()
                )

                results.append({
                    "candidate_id": candidate["id"],
                    "gates": gate_results,
                    "all_passed": all_passed,
                    "min_confidence": min_confidence
                })

        # Return best candidate
        passed_candidates = [
            r for r in results
            if r.get("passed", r.get("all_passed", False))
        ]

        return {
            "security_results": results,
            "passed_candidates": passed_candidates,
            "gate_results": results,
            "confidence_scores": {
                r["candidate_id"]: r.get("min_confidence", 0.0)
                for r in results
            },
            "best_candidate": passed_candidates[0] if passed_candidates else None
        }

    async def _execute_computation(
        self,
        task: TaskContext
    ) -> Dict[str, Any]:
        """Execute DOS policy computation."""

        candidates = task.inputs.get("passed_candidates", [])
        context = task.inputs.get("aegis_context", {})

        utility_scores = {}
        decision_paths = {}
        three_point_estimates = {}
        complexity_breakdown = {}

        for candidate in candidates:
            # Three-point estimation
            estimates = await self._compute_three_point(candidate, context)
            three_point_estimates[candidate["id"]] = estimates

            # Complexity decomposition
            complexity = await self._compute_complexity(candidate)
            complexity_breakdown[candidate["id"]] = complexity

            # Decision path routing
            path = "REFACTORING" if complexity["delta"] < 0 else "INVESTMENT"
            decision_paths[candidate["id"]] = path

            # Utility calculation
            utility = await self._compute_utility(
                candidate, estimates, complexity, path, context
            )
            utility_scores[candidate["id"]] = utility

        return {
            "utility_scores": utility_scores,
            "decision_paths": decision_paths,
            "three_point_estimates": three_point_estimates,
            "complexity_breakdown": complexity_breakdown
        }

    async def _compute_utility(
        self,
        candidate: Dict,
        estimates: Dict,
        complexity: Dict,
        path: str,
        context: Dict
    ) -> Dict[str, float]:
        """Compute vectorized utility per Rubric v2.1."""

        # Extract values
        delta_P_H = estimates["profit"]["expected"]
        delta_V_L = estimates.get("value_low_conf", 0)
        delta_R = estimates["risk"]["expected"]
        delta_C_S = complexity["static_delta"]
        delta_C_D = complexity["dynamic_delta"]
        delta_OPEX = estimates.get("opex_delta", 0)

        # Exchange rates from context
        gamma = context.get("gamma", 0.3)
        kappa = context.get("kappa", 1.0) if delta_R < 0 else 0
        # Complexity tax rates from schema/interface-contract.yaml (Single Source of Truth)
        phi_S = context.get("phi_S", 100)   # Static complexity cost ($/month/kLOC)
        phi_D = context.get("phi_D", 2000)  # Dynamic complexity cost ($/month/service)

        # Utility formula
        U = (
            (delta_P_H + gamma * delta_V_L) +
            kappa * delta_R -
            (phi_S * delta_C_S + phi_D * delta_C_D) -
            delta_OPEX
        )

        # Variance for LCB
        variance = (
            estimates["profit"]["variance"] +
            (gamma ** 2) * estimates.get("value_variance", 0) +
            (kappa ** 2) * estimates["risk"]["variance"]
        )

        # Lower confidence bound
        z_alpha = 1.645  # 95% confidence
        lcb = U - z_alpha * math.sqrt(variance)

        return {
            "raw": U,
            "variance": variance,
            "lcb": lcb,
            "components": {
                "profit_high": delta_P_H,
                "value_low": gamma * delta_V_L,
                "risk": kappa * delta_R,
                "complexity_tax": phi_S * delta_C_S + phi_D * delta_C_D,
                "opex": delta_OPEX
            }
        }

# /src/actors/governance.py

from libertas_opus import Actor, ActorType, HandoffContext
from libertas_opus.protocols import HandoffProtocol, HandoffProtocolType
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
import hashlib
import json

class GovernancePairActor(Actor):
    """Two-key governance actor for override approval."""

    actor_type = ActorType.GOVERNANCE
    capabilities = [
        Capability.TWO_KEY_OVERRIDE,
        Capability.EMERGENCY_HALT
    ]

    def __init__(self, governance_config: Dict[str, Any]):
        super().__init__()
        self.required_roles = governance_config.get("required_roles", [
            "risk_lead", "security_lead"
        ])
        self.notification_service = NotificationService(governance_config)
        self.audit_store = ImmutableAuditStore(governance_config)

    async def execute_task(
        self,
        task: TaskContext
    ) -> Dict[str, Any]:
        """Execute two-key override flow."""

        # Create override request
        override_request = OverrideRequest(
            proposal=task.inputs["proposal"],
            gate_results=task.inputs["gate_results"],
            rationale=task.inputs.get("override_rationale", ""),
            requester=task.context.get("requester_id")
        )

        # Compute proposal hash
        proposal_hash = self._compute_hash(override_request.proposal)

        # Request first key (Risk Lead)
        first_key_result = await self._request_approval(
            role="risk_lead",
            request=override_request,
            proposal_hash=proposal_hash
        )

        if not first_key_result.approved:
            return {
                "override_decision": "REJECT",
                "rejection_reason": first_key_result.reason,
                "rejecting_role": "risk_lead"
            }

        # Request second key (Security Lead)
        second_key_result = await self._request_approval(
            role="security_lead",
            request=override_request,
            proposal_hash=proposal_hash,
            first_key_signature=first_key_result.signature
        )

        if not second_key_result.approved:
            return {
                "override_decision": "REJECT",
                "rejection_reason": second_key_result.reason,
                "rejecting_role": "security_lead"
            }

        # Verify both signatures
        signatures_valid = self._verify_dual_signatures(
            proposal_hash=proposal_hash,
            first_signature=first_key_result.signature,
            second_signature=second_key_result.signature
        )

        if not signatures_valid:
            return {
                "override_decision": "REJECT",
                "rejection_reason": "Signature verification failed"
            }

        # Create immutable audit entry
        audit_entry = await self._create_audit_entry(
            override_request=override_request,
            first_key_result=first_key_result,
            second_key_result=second_key_result,
            proposal_hash=proposal_hash
        )

        return {
            "override_decision": "APPROVE",
            "signatures": {
                "first": first_key_result.signature,
                "second": second_key_result.signature
            },
            "audit_entry": audit_entry.id,
            "merkle_proof": audit_entry.merkle_proof
        }

    def _compute_hash(self, proposal: Dict) -> str:
        """Compute SHA-256 hash of proposal."""
        canonical = json.dumps(proposal, sort_keys=True)
        return hashlib.sha256(canonical.encode()).hexdigest()

    def _verify_dual_signatures(
        self,
        proposal_hash: str,
        first_signature: str,
        second_signature: str
    ) -> bool:
        """Verify BIP-322 dual signatures."""
        # BIP-322 signature verification
        first_valid = self._verify_bip322(
            message=proposal_hash,
            signature=first_signature,
            role="risk_lead"
        )
        second_valid = self._verify_bip322(
            message=proposal_hash,
            signature=second_signature,
            role="security_lead"
        )
        return first_valid and second_valid

    async def _create_audit_entry(
        self,
        override_request: OverrideRequest,
        first_key_result: ApprovalResult,
        second_key_result: ApprovalResult,
        proposal_hash: str
    ) -> AuditEntry:
        """Create immutable audit entry with Merkle proof."""

        entry = AuditEntry(
            event_type="TWO_KEY_OVERRIDE",
            timestamp=datetime.utcnow().isoformat(),
            proposal_hash=proposal_hash,
            proposal_id=override_request.proposal["id"],
            rationale=override_request.rationale,
            approvers=[
                {
                    "role": "risk_lead",
                    "user_id": first_key_result.approver_id,
                    "signature": first_key_result.signature,
                    "timestamp": first_key_result.timestamp,
                    "assessment": first_key_result.assessment
                },
                {
                    "role": "security_lead",
                    "user_id": second_key_result.approver_id,
                    "signature": second_key_result.signature,
                    "timestamp": second_key_result.timestamp,
                    "assessment": second_key_result.assessment
                }
            ],
            gate_results=override_request.gate_results
        )

        # Store with Merkle proof
        stored = await self.audit_store.append_immutable(entry)
        entry.id = stored.id
        entry.merkle_proof = stored.merkle_proof

        return entry

# Extension to libertas_opus/models.py

class ActorType(str, Enum):
    """Extended actor types for AEGIS integration."""

    # Original types
    HUMAN = "human"
    AI = "ai"
    HYBRID = "hybrid"

    # AEGIS extensions
    GOVERNANCE = "governance"    # Two-key override authority
    CALIBRATOR = "calibrator"    # Statistical threshold tuning
    AUDITOR = "auditor"          # Read-only audit access

# Extension to libertas_opus/protocols.py

class HandoffProtocolType(str, Enum):
    """Extended handoff protocols for AEGIS."""

    # Original protocols
    AI_TO_HUMAN_FULL = "ai_to_human_full"
    AI_TO_HUMAN_REVIEW = "ai_to_human_review"
    HUMAN_TO_AI = "human_to_ai"
    AI_ASSISTS_HUMAN = "ai_assists_human"
    COLLABORATIVE_ITERATION = "collaborative_iteration"

    # AEGIS extensions
    TWO_KEY_OVERRIDE = "two_key_override"
    GOVERNANCE_REVIEW = "governance_review"
    CALIBRATION_REVIEW = "calibration_review"
    EMERGENCY_HALT = "emergency_halt"

class TwoKeyOverrideProtocol(HandoffProtocol):
    """BIP-322 dual-signature handoff protocol."""

    protocol_type = HandoffProtocolType.TWO_KEY_OVERRIDE

    def __init__(self):
        self.required_roles = ["risk_lead", "security_lead"]
        self.signature_format = "BIP-322"

    def validate_handoff(self, context: HandoffContext) -> bool:
        """Validate two-key handoff requirements."""

        # Require exactly 2 governance actors
        if len(context.approvers) != 2:
            return False

        # Verify distinct roles
        roles = {a.role for a in context.approvers}
        if roles != set(self.required_roles):
            return False

        # Verify signatures
        for approver in context.approvers:
            if not self._verify_signature(
                approver.signature,
                context.proposal_hash,
                approver.public_key
            ):
                return False

        return True

    def execute_handoff(
        self,
        context: HandoffContext
    ) -> HandoffResult:
        """Execute two-key handoff with audit."""

        # Create audit entry
        audit_entry = self._create_audit_entry(context)

        return HandoffResult(
            success=True,
            new_owner=context.execution_actor,
            audit_id=audit_entry.id,
            metadata={
                "signatures": [a.signature for a in context.approvers],
                "roles": [a.role for a in context.approvers]
            }
        )

# Extension to libertas_opus/engine.py

class WorkflowEngine:
    """Extended workflow engine for AEGIS."""

    async def run(self) -> WorkflowResult:
        """Execute workflow with AEGIS integrations."""

        # Original implementation...

        # AEGIS extension: Emit telemetry on each task
        for task in self.workflow.tasks:
            await self._emit_telemetry(task, "task_start")

            result = await self._execute_task(task)

            await self._emit_telemetry(task, "task_complete", result)

            # AEGIS extension: Check for emergency halt
            if await self._check_emergency_halt():
                return WorkflowResult(
                    status=WorkflowStatus.HALTED,
                    reason="Emergency halt triggered"
                )

        return result

    async def _emit_telemetry(
        self,
        task: Task,
        event: str,
        result: Optional[Dict] = None
    ):
        """Emit AEGIS-compliant telemetry."""

        telemetry = {
            "workflow_id": self.context.workflow_id,
            "task_id": task.id,
            "actor_id": task.actor,
            "timestamp": datetime.utcnow().isoformat(),
            "event": event,
            "duration_ms": self._get_duration(task),
        }

        if result:
            telemetry["outputs_hash"] = self._hash_outputs(result)

        await self.telemetry_collector.emit(telemetry)

┌────────────┐     ┌────────────┐     ┌────────────┐     ┌────────────┐
│    AFA     │     │  LIBERTAS  │     │   Gates    │     │  Decision  │
│ Data Plane │────▶│  Workflow  │────▶│ Evaluation │────▶│  Routing   │
└────────────┘     └────────────┘     └────────────┘     └────────────┘
      │                  │                  │                  │
      │                  │                  │                  │
      ▼                  ▼                  ▼                  ▼
┌────────────┐     ┌────────────┐     ┌────────────┐     ┌────────────┐
│ Candidates │     │   Actors   │     │   Gate     │     │  Terminal  │
│ Generated  │     │  Assigned  │     │  Results   │     │   State    │
└────────────┘     └────────────┘     └────────────┘     └────────────┘

┌────────────┐     ┌────────────┐     ┌────────────┐     ┌────────────┐
│   Human    │     │  Override  │     │  Risk Lead │     │  Security  │
│  Reviewer  │────▶│  Request   │────▶│  Approval  │────▶│    Lead    │
└────────────┘     └────────────┘     └────────────┘     └────────────┘
                         │                  │                  │
                         │                  │                  │
                         ▼                  ▼                  ▼
                   ┌────────────┐     ┌────────────┐     ┌────────────┐
                   │  Validate  │     │ 1st Key    │     │ 2nd Key    │
                   │  Request   │     │ Signature  │     │ Signature  │
                   └────────────┘     └────────────┘     └────────────┘
                                            │                  │
                                            └────────┬─────────┘
                                                     ▼
                                            ┌────────────────┐
                                            │  Verify Both   │
                                            │   Signatures   │
                                            └────────────────┘
                                                     │
                                                     ▼
                                            ┌────────────────┐
                                            │ Immutable Audit│
                                            │     Entry      │
                                            └────────────────┘