Add deterministic five-task mission runtime

src/auction_coordination/runtime.py

@@ -0,0 +1,591 @@
+"""Deterministic end-to-end mission runtime for Agent 11.
+
+The runtime advances the synthetic diligence task graph and records protocol
+traffic. It never chooses a worker. Peers decide whether to compete, the
+auction protocol admits responses, and deterministic settlement selects each
+winner.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from enum import Enum
+from statistics import mean
+
+from .auction import AuctionRoundProtocol
+from .bidding import BidAction, LocalBidPolicy, PeerBidSettings
+from .execution import DeterministicTaskExecutor, ExecutionError
+from .models import FailureClass, TaskAnnouncement, TaskAward, TaskFailure, WorkProduct
+from .reauction import BoundedReauctionPolicy, RecoveryAction
+from .registry import PeerRegistry
+from .scenario import TaskBlueprint
+from .settlement import DeterministicSettlement, SettlementResult
+
+
+class MissionEventType(str, Enum):
+    """Protocol-level traffic recorded by the deterministic mission runtime."""
+
+    TASK_ANNOUNCEMENT = "task_announcement"
+    REAUCTION_ANNOUNCEMENT = "reauction_announcement"
+    BID = "bid"
+    BID_ABSTENTION = "bid_abstention"
+    AUCTION_CLOSED = "auction_closed"
+    TASK_AWARD = "task_award"
+    TASK_RESULT = "task_result"
+    TASK_FAILURE = "task_failure"
+    ESCALATION = "escalation"
+
+
+class TaskRunStatus(str, Enum):
+    COMPLETED = "completed"
+    ESCALATED = "escalated"
+
+
+@dataclass(frozen=True)
+class MissionEvent:
+    sequence: int
+    event_type: MissionEventType
+    task_id: str
+    auction_round: int
+    occurred_at: datetime
+    actor_id: str | None
+    accepted: bool | None
+    detail: str
+
+
+@dataclass(frozen=True)
+class TaskMetrics:
+    task_id: str
+    capability: str
+    status: TaskRunStatus
+    eligible_peers: int
+    bids: int
+    abstentions: int
+    rejected_responses: int
+    auction_rounds: int
+    reauction_count: int
+    no_valid_bid_rounds: int
+    execution_failures: int
+    messages: int
+    winning_peer: str | None
+    winning_score: float | None
+    second_place_score: float | None
+    winning_margin: float | None
+    winning_estimated_cost: float | None
+    execution_cost: float
+    allocation_efficiency: float | None
+
+
+@dataclass(frozen=True)
+class TaskRunResult:
+    blueprint: TaskBlueprint
+    status: TaskRunStatus
+    final_announcement: TaskAnnouncement
+    final_award: TaskAward | None
+    work_product: WorkProduct | None
+    failures: tuple[TaskFailure, ...]
+    metrics: TaskMetrics
+
+
+@dataclass(frozen=True)
+class MissionMetrics:
+    mission_success: bool
+    tasks_attempted: int
+    tasks_completed: int
+    tasks_reauctioned: int
+    escalated_tasks: int
+    failed_auctions: int
+    execution_failures: int
+    total_bids: int
+    total_abstentions: int
+    rejected_responses: int
+    total_messages: int
+    average_bids_per_task: float
+    average_messages_per_task: float
+    total_synthetic_execution_cost: float
+    allocation_efficiency: float | None
+
+
+@dataclass(frozen=True)
+class MissionResult:
+    task_results: tuple[TaskRunResult, ...]
+    work_products: tuple[WorkProduct, ...]
+    events: tuple[MissionEvent, ...]
+    metrics: MissionMetrics
+
+
+class DeterministicMissionRuntime:
+    """Run a bounded five-task distributed-auction mission deterministically."""
+
+    def __init__(
+        self,
+        *,
+        registry: PeerRegistry,
+        bid_settings: dict[str, PeerBidSettings],
+        task_blueprints: tuple[TaskBlueprint, ...],
+        executor: DeterministicTaskExecutor | None = None,
+        failure_injections: frozenset[tuple[str, int]] = frozenset(),
+    ) -> None:
+        self._registry = registry
+        self._bid_settings = dict(bid_settings)
+        self._task_blueprints = task_blueprints
+        self._executor = executor or DeterministicTaskExecutor(registry)
+        self._failure_injections = failure_injections
+        self._bid_policy = LocalBidPolicy(registry)
+        self._settlement = DeterministicSettlement(registry)
+        self._reauction = BoundedReauctionPolicy()
+        self._validate_configuration()
+
+    def run(self, *, started_at: datetime) -> MissionResult:
+        """Execute the configured mission in deterministic dependency order."""
+
+        if started_at.tzinfo is None or started_at.utcoffset() is None:
+            raise ValueError("started_at must include timezone information")
+
+        sequence = 0
+        clock_offset = 0
+        events: list[MissionEvent] = []
+        task_results: list[TaskRunResult] = []
+        completed: dict[str, WorkProduct] = {}
+
+        def tick() -> datetime:
+            nonlocal clock_offset
+            value = started_at + timedelta(seconds=clock_offset)
+            clock_offset += 1
+            return value
+
+        def emit(
+            event_type: MissionEventType,
+            *,
+            task_id: str,
+            auction_round: int,
+            detail: str,
+            actor_id: str | None = None,
+            accepted: bool | None = None,
+        ) -> None:
+            nonlocal sequence
+            sequence += 1
+            events.append(
+                MissionEvent(
+                    sequence=sequence,
+                    event_type=event_type,
+                    task_id=task_id,
+                    auction_round=auction_round,
+                    occurred_at=tick(),
+                    actor_id=actor_id,
+                    accepted=accepted,
+                    detail=detail,
+                )
+            )
+
+        for blueprint in self._task_blueprints:
+            required_input_ids = tuple(
+                completed[task_id].work_product_id
+                for task_id in blueprint.dependency_task_ids
+            )
+            announcement = TaskAnnouncement(
+                task_id=blueprint.task_id,
+                auction_id=blueprint.auction_id,
+                auction_round=1,
+                requested_capability=blueprint.capability,
+                summary=blueprint.summary,
+                minimum_capability_score=blueprint.minimum_capability_score,
+                maximum_cost=blueprint.maximum_cost,
+                required_input_ids=required_input_ids,
+                max_auction_rounds=blueprint.max_auction_rounds,
+                opened_at=tick(),
+            )
+
+            result = self._run_task(
+                blueprint=blueprint,
+                initial_announcement=announcement,
+                available_inputs=tuple(completed.values()),
+                tick=tick,
+                emit=emit,
+                task_event_start=len(events),
+                events=events,
+            )
+            task_results.append(result)
+
+            if result.work_product is not None:
+                completed[blueprint.task_id] = result.work_product
+            if result.status is TaskRunStatus.ESCALATED:
+                break
+
+        metrics = self._mission_metrics(tuple(task_results))
+        work_products = tuple(
+            completed[blueprint.task_id]
+            for blueprint in self._task_blueprints
+            if blueprint.task_id in completed
+        )
+        return MissionResult(
+            task_results=tuple(task_results),
+            work_products=work_products,
+            events=tuple(events),
+            metrics=metrics,
+        )
+
+    def _run_task(
+        self,
+        *,
+        blueprint: TaskBlueprint,
+        initial_announcement: TaskAnnouncement,
+        available_inputs: tuple[WorkProduct, ...],
+        tick,
+        emit,
+        task_event_start: int,
+        events: list[MissionEvent],
+    ) -> TaskRunResult:
+        announcement = initial_announcement
+        protocol = AuctionRoundProtocol(announcement, self._registry)
+        failures: list[TaskFailure] = []
+        bids = 0
+        abstentions = 0
+        rejected = 0
+        rounds = 0
+        reauctions = 0
+        no_valid_bid_rounds = 0
+        execution_failures = 0
+        execution_cost = 0.0
+        efficiencies: list[float] = []
+        final_settlement: SettlementResult | None = None
+        final_award: TaskAward | None = None
+
+        while True:
+            rounds += 1
+            announce_type = (
+                MissionEventType.TASK_ANNOUNCEMENT
+                if announcement.auction_round == 1
+                else MissionEventType.REAUCTION_ANNOUNCEMENT
+            )
+            emit(
+                announce_type,
+                task_id=announcement.task_id,
+                auction_round=announcement.auction_round,
+                detail=(
+                    f"opened auction round {announcement.auction_round} for "
+                    f"{announcement.requested_capability}"
+                ),
+            )
+
+            for agent_id in sorted(self._bid_settings):
+                settings = self._bid_settings[agent_id]
+                decision = self._bid_policy.evaluate(
+                    announcement=announcement,
+                    settings=settings,
+                    submitted_at=tick(),
+                )
+                if decision.action is BidAction.BID:
+                    assert decision.bid is not None
+                    receipt = protocol.submit_bid(decision.bid)
+                    event_type = MissionEventType.BID
+                else:
+                    assert decision.abstention is not None
+                    receipt = protocol.submit_abstention(decision.abstention)
+                    event_type = MissionEventType.BID_ABSTENTION
+
+                emit(
+                    event_type,
+                    task_id=announcement.task_id,
+                    auction_round=announcement.auction_round,
+                    actor_id=agent_id,
+                    accepted=receipt.accepted,
+                    detail=f"{receipt.code.value}: {receipt.detail}",
+                )
+                if receipt.accepted and event_type is MissionEventType.BID:
+                    bids += 1
+                elif receipt.accepted:
+                    abstentions += 1
+                else:
+                    rejected += 1
+
+            protocol.close()
+            emit(
+                MissionEventType.AUCTION_CLOSED,
+                task_id=announcement.task_id,
+                auction_round=announcement.auction_round,
+                detail=(
+                    f"closed with {len(protocol.accepted_bids())} admitted bids and "
+                    f"{len(protocol.accepted_abstentions())} admitted abstentions"
+                ),
+            )
+
+            if not protocol.accepted_bids():
+                no_valid_bid_rounds += 1
+                emit(
+                    MissionEventType.ESCALATION,
+                    task_id=announcement.task_id,
+                    auction_round=announcement.auction_round,
+                    detail="no valid bids remained after protocol admission",
+                )
+                return self._task_result(
+                    blueprint=blueprint,
+                    status=TaskRunStatus.ESCALATED,
+                    announcement=announcement,
+                    final_award=None,
+                    work_product=None,
+                    failures=tuple(failures),
+                    bids=bids,
+                    abstentions=abstentions,
+                    rejected=rejected,
+                    rounds=rounds,
+                    reauctions=reauctions,
+                    no_valid_bid_rounds=no_valid_bid_rounds,
+                    execution_failures=execution_failures,
+                    execution_cost=execution_cost,
+                    efficiencies=efficiencies,
+                    final_settlement=None,
+                    message_count=len(events) - task_event_start,
+                )
+
+            settlement = self._settlement.settle(protocol=protocol, awarded_at=tick())
+            final_settlement = settlement
+            final_award = settlement.award
+            winner = settlement.winning_scorecard
+            execution_cost += winner.estimated_cost
+            best_valid_utility = max(card.total_score for card in settlement.ranked_scorecards)
+            efficiencies.append(
+                1.0 if best_valid_utility == 0.0 else winner.total_score / best_valid_utility
+            )
+            emit(
+                MissionEventType.TASK_AWARD,
+                task_id=announcement.task_id,
+                auction_round=announcement.auction_round,
+                actor_id=settlement.award.awarded_agent_id,
+                detail=(
+                    f"awarded at score {settlement.award.winning_score:.4f} and synthetic "
+                    f"cost {winner.estimated_cost:.2f}"
+                ),
+            )
+
+            try:
+                if (announcement.task_id, announcement.auction_round) in self._failure_injections:
+                    raise ExecutionError(
+                        "injected deterministic execution failure",
+                        failure_class=FailureClass.EXECUTION_ERROR,
+                    )
+                product = self._executor.execute(
+                    announcement=announcement,
+                    award=settlement.award,
+                    completed_at=tick(),
+                    available_inputs=available_inputs,
+                )
+            except ExecutionError as exc:
+                execution_failures += 1
+                failure = self._reauction.record_failure(
+                    award=settlement.award,
+                    failure_class=exc.failure_class,
+                    detail=str(exc),
+                    failed_at=tick(),
+                )
+                failures.append(failure)
+                emit(
+                    MissionEventType.TASK_FAILURE,
+                    task_id=announcement.task_id,
+                    auction_round=announcement.auction_round,
+                    actor_id=settlement.award.awarded_agent_id,
+                    detail=f"{failure.failure_class.value}: {failure.detail}",
+                )
+                plan = self._reauction.plan(
+                    announcement=announcement,
+                    award=settlement.award,
+                    failure=failure,
+                    reopened_at=tick(),
+                )
+                if plan.action is RecoveryAction.ESCALATE:
+                    emit(
+                        MissionEventType.ESCALATION,
+                        task_id=announcement.task_id,
+                        auction_round=announcement.auction_round,
+                        detail=plan.reason,
+                    )
+                    return self._task_result(
+                        blueprint=blueprint,
+                        status=TaskRunStatus.ESCALATED,
+                        announcement=announcement,
+                        final_award=final_award,
+                        work_product=None,
+                        failures=tuple(failures),
+                        bids=bids,
+                        abstentions=abstentions,
+                        rejected=rejected,
+                        rounds=rounds,
+                        reauctions=reauctions,
+                        no_valid_bid_rounds=no_valid_bid_rounds,
+                        execution_failures=execution_failures,
+                        execution_cost=execution_cost,
+                        efficiencies=efficiencies,
+                        final_settlement=final_settlement,
+                        message_count=len(events) - task_event_start,
+                    )
+
+                reauctions += 1
+                protocol = self._reauction.open_reauction_round(
+                    plan=plan,
+                    registry=self._registry,
+                )
+                assert plan.next_announcement is not None
+                announcement = plan.next_announcement
+                continue
+
+            emit(
+                MissionEventType.TASK_RESULT,
+                task_id=announcement.task_id,
+                auction_round=announcement.auction_round,
+                actor_id=product.agent_id,
+                detail=f"validated work product {product.work_product_id}",
+            )
+            return self._task_result(
+                blueprint=blueprint,
+                status=TaskRunStatus.COMPLETED,
+                announcement=announcement,
+                final_award=final_award,
+                work_product=product,
+                failures=tuple(failures),
+                bids=bids,
+                abstentions=abstentions,
+                rejected=rejected,
+                rounds=rounds,
+                reauctions=reauctions,
+                no_valid_bid_rounds=no_valid_bid_rounds,
+                execution_failures=execution_failures,
+                execution_cost=execution_cost,
+                efficiencies=efficiencies,
+                final_settlement=final_settlement,
+                message_count=len(events) - task_event_start,
+            )
+
+    def _task_result(
+        self,
+        *,
+        blueprint: TaskBlueprint,
+        status: TaskRunStatus,
+        announcement: TaskAnnouncement,
+        final_award: TaskAward | None,
+        work_product: WorkProduct | None,
+        failures: tuple[TaskFailure, ...],
+        bids: int,
+        abstentions: int,
+        rejected: int,
+        rounds: int,
+        reauctions: int,
+        no_valid_bid_rounds: int,
+        execution_failures: int,
+        execution_cost: float,
+        efficiencies: list[float],
+        final_settlement: SettlementResult | None,
+        message_count: int,
+    ) -> TaskRunResult:
+        winning_score = None
+        second_place_score = None
+        winning_margin = None
+        winning_cost = None
+        winning_peer = None
+        if final_settlement is not None:
+            winning_score = final_settlement.award.winning_score
+            second_place_score = final_settlement.second_place_score
+            winning_peer = final_settlement.award.awarded_agent_id
+            winning_cost = final_settlement.winning_scorecard.estimated_cost
+            if second_place_score is not None:
+                winning_margin = winning_score - second_place_score
+
+        metrics = TaskMetrics(
+            task_id=blueprint.task_id,
+            capability=blueprint.capability,
+            status=status,
+            eligible_peers=len(
+                self._registry.eligible_profiles(
+                    blueprint.capability,
+                    blueprint.minimum_capability_score,
+                )
+            ),
+            bids=bids,
+            abstentions=abstentions,
+            rejected_responses=rejected,
+            auction_rounds=rounds,
+            reauction_count=reauctions,
+            no_valid_bid_rounds=no_valid_bid_rounds,
+            execution_failures=execution_failures,
+            messages=message_count,
+            winning_peer=winning_peer,
+            winning_score=winning_score,
+            second_place_score=second_place_score,
+            winning_margin=winning_margin,
+            winning_estimated_cost=winning_cost,
+            execution_cost=round(execution_cost, 12),
+            allocation_efficiency=(
+                round(mean(efficiencies), 12) if efficiencies else None
+            ),
+        )
+        return TaskRunResult(
+            blueprint=blueprint,
+            status=status,
+            final_announcement=announcement,
+            final_award=final_award,
+            work_product=work_product,
+            failures=failures,
+            metrics=metrics,
+        )
+
+    def _mission_metrics(self, results: tuple[TaskRunResult, ...]) -> MissionMetrics:
+        attempted = len(results)
+        completed = sum(result.status is TaskRunStatus.COMPLETED for result in results)
+        efficiencies = [
+            result.metrics.allocation_efficiency
+            for result in results
+            if result.metrics.allocation_efficiency is not None
+        ]
+        mission_success = attempted == len(self._task_blueprints) and completed == attempted
+        return MissionMetrics(
+            mission_success=mission_success,
+            tasks_attempted=attempted,
+            tasks_completed=completed,
+            tasks_reauctioned=sum(result.metrics.reauction_count > 0 for result in results),
+            escalated_tasks=sum(result.status is TaskRunStatus.ESCALATED for result in results),
+            failed_auctions=sum(result.metrics.no_valid_bid_rounds for result in results),
+            execution_failures=sum(result.metrics.execution_failures for result in results),
+            total_bids=sum(result.metrics.bids for result in results),
+            total_abstentions=sum(result.metrics.abstentions for result in results),
+            rejected_responses=sum(result.metrics.rejected_responses for result in results),
+            total_messages=sum(result.metrics.messages for result in results),
+            average_bids_per_task=(
+                round(sum(result.metrics.bids for result in results) / attempted, 12)
+                if attempted
+                else 0.0
+            ),
+            average_messages_per_task=(
+                round(sum(result.metrics.messages for result in results) / attempted, 12)
+                if attempted
+                else 0.0
+            ),
+            total_synthetic_execution_cost=round(
+                sum(result.metrics.execution_cost for result in results), 12
+            ),
+            allocation_efficiency=(round(mean(efficiencies), 12) if efficiencies else None),
+        )
+
+    def _validate_configuration(self) -> None:
+        if not self._task_blueprints:
+            raise ValueError("mission requires at least one task blueprint")
+        task_ids = [blueprint.task_id for blueprint in self._task_blueprints]
+        if len(task_ids) != len(set(task_ids)):
+            raise ValueError("task blueprint IDs must be unique")
+
+        seen: set[str] = set()
+        for blueprint in self._task_blueprints:
+            missing = [
+                dependency
+                for dependency in blueprint.dependency_task_ids
+                if dependency not in seen
+            ]
+            if missing:
+                raise ValueError(
+                    f"task {blueprint.task_id} depends on unavailable prior tasks: "
+                    f"{', '.join(missing)}"
+                )
+            seen.add(blueprint.task_id)
+
+        registered_ids = {profile.agent_id for profile in self._registry.all_profiles()}
+        if set(self._bid_settings) != registered_ids:
+            raise ValueError("bid settings must cover exactly the registered peer identities")