"""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")