canwu_sim/runtime/

settlement.rs

use super::event_payloads::{KnowledgePublished, RuntimeEventPayload};
use super::ingress::{PluginIngressCancellationProof, valid_ingress_cancellation_reason};
use super::validation::{
    EvidenceAvailability, RuntimeValidationContext, resolve_evidence_reference,
};
use super::{
    AssertUnwindSafe, BTreeMap, BTreeSet, BoundaryChange, BoundaryContext, BoundaryDirective,
    BoundaryEmission, BoundaryEmissionKind, BoundaryId, BoundaryIngressGeneration,
    BoundaryKnowledgeChange, BoundaryPhase, BoundaryProposal, BoundaryReceipt, BoundaryRecord,
    BoundaryRequest, BoundaryStateHashFormat, BoundarySystemContract,
    BoundaryTransactionCheckpoint, CanwuError, CauseRef, Command, CommandEnvelope, CommandIngress,
    CommandRequest, CommitmentDomains, DecisionAction, DecisionIngressRequest, DecisionMutation,
    DecisionOutcome, DecisionPolicyKind, DecisionRandomEvidence, DecisionStage, DomainRecord,
    DomainRecordChange, DomainRecordRef, DomainRecordVersionSource, EntityRef, ErrorCode,
    EventKind, EvidenceRef, GENESIS_BOUNDARY_HASH, HashSet, IngressCancellationAuthority,
    IngressPayload, KnowledgeHolderRef, KnowledgeRecord, KnowledgeRecordId, PluginComponentKey,
    PluginComponentRecord, PluginRegistry, PolicyDecision, RandomDrawAddress, RandomDrawOutcome,
    RandomOperationTarget, RefCell, ReservationAllocation, ReservationDisposition,
    ReservationOffer, ReservationOfferRecord, ReservationPoolKey, ReservationRef,
    ReservationRequest, ReservationRequestRecord, RunConfigurationSnapshot, RuntimeCurrentState,
    RuntimeState, ScheduleKey, ScheduledAction, SimTime, Simulation, SimulationView,
    SimulationViewState, StateKey, StateVisibility, SystemCadence, SystemDirective, canonical_hash,
    canonical_text, catch_unwind, claim_counter, component_key, compute_boundary_hash,
    invalid_snapshot_error, is_domain_record_state, proposal_entity_exists,
    proposal_entity_identity_exists, random, record_change_affected_entities, records,
    runtime_current_entity_exists, runtime_entity_exists,
    runtime_entity_exists_with_record_overlay, runtime_entity_identity_exists,
    validate_domain_dependents_with_records, validate_runtime_domain_dependents,
};

impl Simulation {
    pub fn settle_boundary(
        &mut self,
        request: BoundaryRequest,
    ) -> Result<BoundaryReceipt, CanwuError> {
        self.settle_boundary_with_state_hash_format(request, BoundaryStateHashFormat::CommitmentsV1)
    }

    pub(super) fn settle_boundary_with_state_hash_format(
        &mut self,
        mut request: BoundaryRequest,
        state_hash_format: BoundaryStateHashFormat,
    ) -> Result<BoundaryReceipt, CanwuError> {
        self.ensure_runtime_ready()?;
        if request.at < self.state.scheduler.now {
            return Err(CanwuError::new(
                ErrorCode::InvalidBoundary,
                "a settlement boundary cannot precede committed simulation time",
            ));
        }
        if self
            .state
            .scheduler
            .pending_ingress
            .first()
            .is_some_and(|key| key.due_at < request.at)
        {
            return Err(CanwuError::new(
                ErrorCode::InvalidBoundary,
                "a settlement boundary cannot step past earlier canonical ingress",
            ));
        }
        if request.cadences.contains(&SystemCadence::EventDriven) {
            return Err(CanwuError::new(
                ErrorCode::InvalidBoundary,
                "event-driven cadence is derived from admitted events, not caller supplied",
            ));
        }
        request.cadences.sort();
        request.cadences.dedup();

        let transaction = BoundaryTransactionCheckpoint::capture(&self.state);
        match self.settle_boundary_inner(request, state_hash_format) {
            Ok(receipt) => Ok(receipt),
            Err(error) => {
                transaction.restore(&mut self.state);
                Err(error)
            }
        }
    }

    fn settle_boundary_inner(
        &mut self,
        mut request: BoundaryRequest,
        state_hash_format: BoundaryStateHashFormat,
    ) -> Result<BoundaryReceipt, CanwuError> {
        self.advance_to_before_boundary(request.at)?;

        let admitted_ingress = self.take_due_ingress(request.at);
        let admitted_ingress_index: HashSet<_> = admitted_ingress.iter().copied().collect();
        let mut maintenance_changes = Vec::new();
        let mut maintenance_record_changes = Vec::new();
        for ingress_id in &admitted_ingress {
            let record = self
                .state
                .evidence
                .retained_ingress(*ingress_id)
                .cloned()
                .ok_or_else(|| {
                    CanwuError::new(
                        ErrorCode::InvalidSnapshot,
                        "pending ingress references an unknown record",
                    )
                })?;
            match record.payload {
                IngressPayload::Command { request: command } => {
                    let CommandRequest {
                        request_id,
                        expected_revision,
                        envelope,
                    } = *command;
                    self.admit_command(
                        Some(request_id),
                        Some(expected_revision),
                        envelope,
                        CommandIngress::LiveRequest,
                        None,
                        true,
                    )?;
                }
                IngressPayload::Calendar { cadences } => request.cadences.extend(cadences),
                IngressPayload::Plugin { .. } => {}
                IngressPayload::PluginCancellation { .. } => {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidSnapshot,
                        "a terminal ingress cancellation cannot be admitted",
                    ));
                }
                IngressPayload::Decision { request } => {
                    self.apply_decision_request(*request)?;
                }
                IngressPayload::Maintenance { request } => {
                    let (change, record_changes) = self.apply_maintenance_request(*request)?;
                    maintenance_changes.push(change);
                    maintenance_record_changes.extend(record_changes);
                }
            }
        }
        self.state
            .current
            .decisions
            .advance_time(request.at)
            .map_err(super::decision::decision_error)?;
        self.invalidate_commitments(CommitmentDomains::DECISIONS);
        self.execute_scheduled_at(request.at)?;
        request.cadences.sort();
        request.cadences.dedup();

        let admitted_attempt_count = self
            .state
            .evidence
            .archived
            .command_attempt_count
            .checked_add(
                u64::try_from(self.state.evidence.command_attempts.len()).map_err(|_| {
                    invalid_snapshot_error("attempt journal exceeds admission cursor range")
                })?,
            )
            .ok_or_else(|| invalid_snapshot_error("attempt journal cursor is exhausted"))?;
        let admitted_command_count = self
            .state
            .evidence
            .archived
            .command_count
            .checked_add(
                u64::try_from(self.state.evidence.commands.len()).map_err(|_| {
                    invalid_snapshot_error("command journal exceeds admission cursor range")
                })?,
            )
            .ok_or_else(|| invalid_snapshot_error("command journal cursor is exhausted"))?;
        let admitted_event_count = self
            .state
            .evidence
            .archived
            .event_count
            .checked_add(
                u64::try_from(self.state.evidence.events.len()).map_err(|_| {
                    invalid_snapshot_error("event journal exceeds admission cursor range")
                })?,
            )
            .ok_or_else(|| invalid_snapshot_error("event journal cursor is exhausted"))?;
        let admitted_attempt_start = self
            .state
            .counters
            .admitted_attempt_count
            .checked_sub(self.state.evidence.archived.command_attempt_count)
            .ok_or_else(|| {
                invalid_snapshot_error("runtime attempt admission cursor precedes live evidence")
            })?;
        let admitted_attempt_start = usize::try_from(admitted_attempt_start).map_err(|_| {
            invalid_snapshot_error("runtime attempt admission cursor exceeds platform range")
        })?;
        let admitted_command_start = self
            .state
            .counters
            .admitted_command_count
            .checked_sub(self.state.evidence.archived.command_count)
            .ok_or_else(|| {
                invalid_snapshot_error("runtime command admission cursor precedes live evidence")
            })?;
        let admitted_command_start = usize::try_from(admitted_command_start).map_err(|_| {
            invalid_snapshot_error("runtime command admission cursor exceeds platform range")
        })?;
        let admitted_event_start = self
            .state
            .counters
            .admitted_event_count
            .checked_sub(self.state.evidence.archived.event_count)
            .ok_or_else(|| {
                invalid_snapshot_error("runtime event admission cursor precedes live evidence")
            })?;
        let admitted_event_start = usize::try_from(admitted_event_start).map_err(|_| {
            invalid_snapshot_error("runtime event admission cursor exceeds platform range")
        })?;
        let admitted_attempts: Vec<_> = self
            .state
            .evidence
            .command_attempts
            .get(admitted_attempt_start..)
            .ok_or_else(|| {
                invalid_snapshot_error("runtime attempt admission cursor exceeds its journal")
            })?
            .iter()
            .map(|record| record.id)
            .collect();
        let admitted_commands: Vec<_> = self
            .state
            .evidence
            .commands
            .get(admitted_command_start..)
            .ok_or_else(|| {
                invalid_snapshot_error("runtime command admission cursor exceeds its journal")
            })?
            .iter()
            .map(|record| record.id)
            .collect();
        let admitted_events: Vec<_> = self
            .state
            .evidence
            .events
            .get(admitted_event_start..)
            .ok_or_else(|| {
                invalid_snapshot_error("runtime event admission cursor exceeds its journal")
            })?
            .iter()
            .map(|event| event.id)
            .collect();

        let (boundary_id_value, next_boundary_id) =
            claim_counter(self.state.counters.next_boundary_id, "boundary ID")?;
        let (correlation_id, next_correlation_id) = claim_counter(
            self.state.counters.next_correlation_id,
            "boundary correlation ID",
        )?;
        self.state.counters.next_boundary_id = next_boundary_id;
        self.state.counters.next_correlation_id = next_correlation_id;
        let boundary_id = BoundaryId::new(boundary_id_value);

        let boundary_snapshot = self.state.current.clone();
        let boundary_time = self.state.scheduler.now;
        let systems = self.plugins.boundary_systems.clone();
        let state_owners = self.plugins.state_owners.clone();
        let record_schemas = self.plugins.record_schemas.clone();
        let mut allocations = BTreeMap::new();
        let mut allocation_records = Vec::new();
        let mut reservation_offer_records = Vec::new();
        let mut reservation_request_records = Vec::new();
        let mut offers = Vec::new();
        let mut requests = Vec::new();
        let mut random_overlay = boundary_snapshot.random_streams.clone();
        let mut pending_random_draws = Vec::new();
        let mut keyed_random_draws = random::keyed_draws_with_reservations(
            &self.state.evidence.random_draws,
            &self.state.evidence.keyed_draw_reservations,
        )?;
        let mut visible_overlay = BTreeMap::new();
        let mut candidate_overlay = BTreeMap::new();
        let mut visible_record_overlay = BTreeMap::new();
        let mut candidate_record_overlay = BTreeMap::new();
        let mut visible_knowledge_overlay = BTreeMap::new();
        let mut pending_knowledge_changes = Vec::new();
        let mut knowledge_correlations = BTreeSet::new();
        let mut ordinary = Vec::new();
        let mut transitions = Vec::new();
        let mut deferred = Vec::new();
        let mut evidence = PendingBoundaryEvidence::default();
        let mut person_writes = super::persons::BoundaryPersonWrites::default();
        let evaluation_limits = self.state.metadata.run_configuration.evaluation_limits();
        let mut evaluation_traces = Vec::new();
        let mut transition_ledger = super::transitions::BoundaryTransitionLedger::new(
            boundary_id,
            &self.state.scheduler.transition_manifests,
        );
        for change in maintenance_record_changes {
            let change_index = u64::try_from(evidence.record_changes.len()).map_err(|_| {
                CanwuError::new(
                    ErrorCode::IdentifierExhausted,
                    "boundary record-change index exceeds the persistent identifier space",
                )
            })?;
            index_current_domain_record_version(
                &mut self.state,
                boundary_id,
                change_index,
                &change,
            );
            let event = self.append_event(
                EventKind::plugin(change.plugin.clone(), change.operation.event_type()),
                record_change_affected_entities(&change),
                change.summary.clone(),
                Some(CauseRef::Boundary(boundary_id)),
                correlation_id,
            )?;
            evidence.emissions.push(BoundaryEmission {
                plugin: change.plugin.clone(),
                system: change.system.clone(),
                event: event.id,
                kind: BoundaryEmissionKind::RecordChange { change_index },
            });
            evidence.record_changes.push(change);
        }
        for phase in BoundaryPhase::ALL {
            match phase {
                BoundaryPhase::AtomicDomainCommit => {
                    let (same_boundary, next_boundary) =
                        partition_boundary_visibility(std::mem::take(&mut ordinary));
                    self.apply_boundary_stage(
                        boundary_id,
                        correlation_id,
                        same_boundary,
                        &mut evidence,
                    )?;
                    deferred.extend(next_boundary);
                    visible_overlay.clear();
                    candidate_overlay.clear();
                    visible_record_overlay.clear();
                    candidate_record_overlay.clear();
                }
                BoundaryPhase::ConditionalTransitionCommit => {
                    // Audit ready transition manifests against the committed
                    // state phase 10 read, before the bundle commits.
                    transition_ledger.settle_ready(&self.state)?;
                    let (same_boundary, next_boundary) =
                        partition_boundary_visibility(std::mem::take(&mut transitions));
                    self.apply_boundary_stage(
                        boundary_id,
                        correlation_id,
                        same_boundary,
                        &mut evidence,
                    )?;
                    deferred.extend(next_boundary);
                    if transition_ledger.requires_post_check() {
                        self.check_transition_post_versions(
                            &mut transition_ledger,
                            &record_schemas,
                            &deferred,
                        )?;
                    }
                    visible_overlay.clear();
                    visible_record_overlay.clear();
                }
                _ => {}
            }

            let mut phase_directives = Vec::new();
            for registered in systems.iter().filter(|registered| {
                registered.contract.phase == phase
                    && boundary_system_due(
                        &registered.contract,
                        &request.cadences,
                        !admitted_events.is_empty() || !admitted_ingress.is_empty(),
                    )
            }) {
                let reader = format!("{}.{}", registered.plugin, registered.contract.name);
                let (view_current, view_now) = if phase <= BoundaryPhase::InvariantValidation {
                    (&boundary_snapshot, boundary_time)
                } else {
                    (&self.state.current, self.state.scheduler.now)
                };
                let random_session = random::RandomSession::new(
                    &random_overlay,
                    &registered.contract.random_streams,
                    boundary_snapshot.root_seed,
                    &registered.plugin,
                    &keyed_random_draws,
                )?;
                let proposal_evidence = proposal_evidence_refs(boundary_id, &evidence);
                let view = SimulationView {
                    state: SimulationViewState::Boundary {
                        current: view_current,
                        now: view_now,
                        runtime: &self.state,
                    },
                    state_owners: &state_owners,
                    reader: Some(&reader),
                    allowed_reads: Some(&registered.contract.reads),
                    allowed_ingress: Some(&admitted_ingress_index),
                    ingress_plugin: Some(&registered.plugin),
                    component_overlay: Some(&visible_overlay),
                    proposed_components: (phase == BoundaryPhase::InvariantValidation)
                        .then_some(&candidate_overlay),
                    record_overlay: Some(&visible_record_overlay),
                    proposed_records: (phase == BoundaryPhase::InvariantValidation)
                        .then_some(&candidate_record_overlay),
                    boundary_id: Some(boundary_id),
                    proposal_evidence: Some(&proposal_evidence),
                    knowledge_overlay: Some(&visible_knowledge_overlay),
                    allocations: Some(&allocations),
                    allowed_reservations: Some(&registered.contract.reservation_reads),
                    random_session: Some(RefCell::new(random_session)),
                    plugin_archive_provider: self.plugin_archive_provider.as_ref(),
                    transitions: Some(&transition_ledger),
                };
                let context = BoundaryContext {
                    boundary_id,
                    at: request.at,
                    phase,
                    plugin: registered.plugin.clone(),
                    system: registered.contract.name.clone(),
                    admitted_attempts: admitted_attempts.clone(),
                    admitted_commands: admitted_commands.clone(),
                    admitted_ingress: admitted_ingress.clone(),
                    admitted_events: admitted_events.clone(),
                    emitted_events: evidence
                        .emissions
                        .iter()
                        .map(|emission| emission.event)
                        .collect(),
                };
                let proposal =
                    catch_unwind(AssertUnwindSafe(|| (registered.handler)(&view, &context)))
                        .map_err(|_| {
                            CanwuError::new(
                                ErrorCode::PluginPanicked,
                                format!(
                                    "boundary system {}.{} panicked",
                                    registered.plugin, registered.contract.name
                                ),
                            )
                        })??;
                let random_execution = view
                    .finish_random_session()
                    .expect("boundary views always have a random session");
                // Registrations leave the proposal and staged transition
                // writes become ordinary directives of this system.
                let proposal = transition_ledger.admit(
                    &registered.plugin,
                    &registered.contract,
                    &self.plugins,
                    proposal,
                )?;
                super::evaluation::check_trace_budget(
                    evaluation_traces.len(),
                    &proposal.directives,
                    evaluation_limits,
                )?;
                validate_boundary_proposal(
                    &registered.plugin,
                    &registered.contract,
                    view_current,
                    &boundary_snapshot.person_availability,
                    view_now,
                    &self.state,
                    boundary_id,
                    &evidence,
                    &self.plugins,
                    &visible_record_overlay,
                    &visible_knowledge_overlay,
                    &proposal,
                    &random_execution.draws,
                )?;
                person_writes.stage(
                    &registered.plugin,
                    &registered.contract.name,
                    &proposal.directives,
                )?;
                random::extend_keyed_draws(&mut keyed_random_draws, &random_execution.draws)?;
                random_overlay.extend(random_execution.states);
                pending_random_draws.extend(random_execution.draws.into_iter().map(|draw| {
                    PendingBoundaryRandomDraw {
                        plugin: registered.plugin.clone(),
                        system: registered.contract.name.clone(),
                        draw,
                    }
                }));
                offers.extend(
                    proposal
                        .offers
                        .into_iter()
                        .map(|offer| PendingReservationOffer {
                            plugin: registered.plugin.clone(),
                            system: registered.contract.name.clone(),
                            offer,
                        }),
                );
                requests.extend(proposal.requests.into_iter().map(|request| {
                    PendingReservationRequest {
                        reservation: ReservationRef::new(
                            &registered.plugin,
                            &registered.contract.name,
                            &request.request,
                        ),
                        request,
                    }
                }));
                // Traces are evidence, not state: they leave the directive
                // stream here and are recorded on the boundary record.
                let mut state_directives = Vec::with_capacity(proposal.directives.len());
                let mut traces = Vec::new();
                for directive in proposal.directives {
                    match directive {
                        BoundaryDirective::RecordEvaluationTrace { trace } => traces.push(trace),
                        directive => state_directives.push(directive),
                    }
                }
                super::evaluation::stage_traces(
                    &mut evaluation_traces,
                    &registered.plugin,
                    &registered.contract,
                    traces,
                );
                phase_directives.extend(state_directives.into_iter().map(|directive| {
                    // Created persons commit at the end of the boundary and
                    // become visible to systems at the next boundary.
                    let visibility = if matches!(directive, BoundaryDirective::CreatePerson { .. })
                    {
                        StateVisibility::NextBoundary
                    } else {
                        registered.contract.visibility
                    };
                    StagedBoundaryDirective {
                        plugin: registered.plugin.clone(),
                        system: registered.contract.name.clone(),
                        phase,
                        visibility,
                        directive,
                    }
                }));
            }
            transition_ledger.close_phase();

            let (knowledge_directives, phase_directives) =
                partition_knowledge_directives(phase_directives);
            if !knowledge_directives.is_empty()
                && !matches!(
                    phase,
                    BoundaryPhase::PerceptionAndAttentionRefresh
                        | BoundaryPhase::PerspectiveAndReportMaterialization
                )
            {
                return Err(CanwuError::new(
                    ErrorCode::UndeclaredKnowledgeWrite,
                    "knowledge publication is allowed only in phases 4 and 13",
                ));
            }

            match phase {
                BoundaryPhase::PerceptionAndAttentionRefresh => {
                    if !phase_directives.is_empty() {
                        return Err(CanwuError::new(
                            ErrorCode::InvalidBoundary,
                            "phase 4 accepts knowledge publications but no ordinary directives",
                        ));
                    }
                    self.stage_knowledge_publications(
                        phase,
                        knowledge_directives,
                        &mut visible_knowledge_overlay,
                        &mut pending_knowledge_changes,
                        &mut knowledge_correlations,
                    )?;
                }
                BoundaryPhase::ReservationAndAllocation => {
                    let result = allocate_reservations(
                        std::mem::take(&mut offers),
                        std::mem::take(&mut requests),
                    )?;
                    allocations = result.by_reservation;
                    allocation_records = result.records;
                    reservation_offer_records = result.offers;
                    reservation_request_records = result.requests;
                }
                BoundaryPhase::DomainDeltaProposal => {
                    let record_context = BoundaryRecordOverlayContext {
                        current: &boundary_snapshot,
                        now: boundary_time,
                        scheduled_actions: &self.state.scheduler.actions,
                        run_configuration: &self.state.metadata.run_configuration,
                        schemas: &record_schemas,
                    };
                    extend_boundary_record_candidate_overlay(
                        &record_context,
                        &mut candidate_record_overlay,
                        &phase_directives,
                    )?;
                    extend_boundary_candidate_overlay(
                        &boundary_snapshot,
                        &candidate_record_overlay,
                        &mut candidate_overlay,
                        &phase_directives,
                    )?;
                    extend_boundary_record_overlay(
                        &record_context,
                        &mut visible_record_overlay,
                        &phase_directives,
                    )?;
                    extend_boundary_overlay(
                        &boundary_snapshot,
                        &visible_record_overlay,
                        &mut visible_overlay,
                        &phase_directives,
                    )?;
                    ordinary.extend(phase_directives);
                }
                BoundaryPhase::HistoricalCandidateEvaluation => {
                    let record_context = BoundaryRecordOverlayContext {
                        current: &self.state.current,
                        now: self.state.scheduler.now,
                        scheduled_actions: &self.state.scheduler.actions,
                        run_configuration: &self.state.metadata.run_configuration,
                        schemas: &record_schemas,
                    };
                    extend_boundary_record_overlay(
                        &record_context,
                        &mut visible_record_overlay,
                        &phase_directives,
                    )?;
                    extend_boundary_overlay(
                        &self.state.current,
                        &visible_record_overlay,
                        &mut visible_overlay,
                        &phase_directives,
                    )?;
                    transitions.extend(phase_directives);
                }
                BoundaryPhase::StrategicAggregation
                | BoundaryPhase::PerspectiveAndReportMaterialization => {
                    if phase == BoundaryPhase::PerspectiveAndReportMaterialization {
                        self.stage_knowledge_publications(
                            phase,
                            knowledge_directives,
                            &mut visible_knowledge_overlay,
                            &mut pending_knowledge_changes,
                            &mut knowledge_correlations,
                        )?;
                    }
                    let (same_boundary, next_boundary) =
                        partition_boundary_visibility(phase_directives);
                    self.apply_boundary_stage(
                        boundary_id,
                        correlation_id,
                        same_boundary,
                        &mut evidence,
                    )?;
                    deferred.extend(next_boundary);
                }
                _ if !phase_directives.is_empty() => {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidBoundary,
                        format!("boundary phase {phase:?} cannot produce state directives"),
                    ));
                }
                _ => {}
            }
        }

        self.apply_boundary_stage(boundary_id, correlation_id, deferred, &mut evidence)?;
        self.commit_knowledge_publications(
            boundary_id,
            correlation_id,
            &pending_knowledge_changes,
            &mut evidence.emissions,
        )?;
        let PendingBoundaryEvidence {
            changes,
            record_changes,
            emissions,
            mut generated_ingress,
            random_decisions,
            mut person_availability_changes,
            created_persons,
        } = evidence;
        self.invalidate_commitments(CommitmentDomains::RANDOM_STREAMS);
        self.state.current.random_streams = random_overlay;
        let mut random_outcomes = BTreeMap::new();
        for pending in &random_decisions {
            let ticket = self
                .state
                .current
                .decisions
                .ticket(pending.resolution.ticket_id)
                .ok_or_else(|| {
                    CanwuError::new(
                        ErrorCode::InvalidDecision,
                        "random decision ticket disappeared before boundary commit",
                    )
                })?;
            let option_id = random_resolution_selection(ticket, &pending.resolution)?;
            let previous = random_outcomes.insert(
                (
                    pending.resolution.sample.stream.clone(),
                    pending.resolution.sample.address.clone(),
                ),
                RandomDrawOutcome::DecisionSelection {
                    ticket_id: ticket.id,
                    ticket_version: ticket.version,
                    option_id,
                },
            );
            if previous.is_some() {
                return Err(CanwuError::new(
                    ErrorCode::InvalidRandomDraw,
                    "one random draw cannot resolve multiple decisions",
                ));
            }
        }
        let committed_random_draws = self.append_boundary_random_draws(
            boundary_id,
            correlation_id,
            pending_random_draws,
            random_outcomes,
        )?;
        self.materialize_boundary_random_decisions(
            boundary_id,
            &random_decisions,
            &committed_random_draws,
            &mut generated_ingress,
        )?;
        self.cancel_unavailable_person_tickets(&mut person_availability_changes)?;
        let transition_evidence = transition_ledger.finish();
        if !transition_evidence.registered.is_empty() || !transition_evidence.audits.is_empty() {
            self.invalidate_commitments(CommitmentDomains::SCHEDULER);
            self.state.scheduler.transition_manifests = transition_evidence.pending;
        }
        let random_draws = committed_random_draws
            .iter()
            .map(|draw| draw.id)
            .collect::<Vec<_>>();
        self.state.metadata.plugin_registration_closed = true;
        let state_hash = self.compute_boundary_state_hash_for(state_hash_format)?;
        let previous_hash = self
            .state
            .evidence
            .boundary_head_hash()
            .map_or_else(|| GENESIS_BOUNDARY_HASH.to_owned(), str::to_owned);
        let previous_maintenance_root = self
            .state
            .evidence
            .boundaries
            .last()
            .and_then(|boundary| boundary.maintenance_terminal_root.as_deref());
        let maintenance_terminal_root =
            if previous_maintenance_root.is_some() || !maintenance_changes.is_empty() {
                Some(canonical_hash(
                    "canwu.maintenance.terminal-root.v1",
                    &(
                        previous_maintenance_root.unwrap_or(GENESIS_BOUNDARY_HASH),
                        &maintenance_changes,
                    ),
                )?)
            } else {
                None
            };
        let mut record = BoundaryRecord {
            id: boundary_id,
            at: request.at,
            correlation_id,
            cadences: request.cadences,
            admitted_attempts,
            admitted_commands,
            admitted_ingress,
            generated_ingress: generated_ingress.clone(),
            admitted_events,
            reservation_offers: reservation_offer_records,
            reservation_requests: reservation_request_records,
            allocations: allocation_records.clone(),
            random_draws: random_draws.clone(),
            changes: changes.clone(),
            record_changes: record_changes.clone(),
            knowledge_changes: pending_knowledge_changes.clone(),
            maintenance_changes,
            maintenance_terminal_root,
            person_availability_changes,
            created_persons: created_persons.clone(),
            evaluation_traces,
            transition_manifests: transition_evidence.registered,
            transition_audits: transition_evidence.audits.clone(),
            emissions: emissions.clone(),
            state_hash: Some(state_hash),
            previous_hash,
            hash: String::new(),
        };
        record.hash = compute_boundary_hash(&record)?;
        let boundary_hash = record.hash.clone();
        self.state.evidence.boundaries.push(record);
        self.state.counters.admitted_attempt_count = admitted_attempt_count;
        self.state.counters.admitted_command_count = admitted_command_count;
        self.state.counters.admitted_event_count = admitted_event_count;
        self.advance_state_revision()?;
        self.refresh_checkpoint_hash()?;
        Ok(BoundaryReceipt {
            boundary_id,
            settled_at: request.at,
            emitted_events: emissions
                .into_iter()
                .map(|emission| emission.event)
                .collect(),
            generated_ingress: generated_ingress
                .into_iter()
                .map(|generation| generation.ingress)
                .collect(),
            random_draws,
            boundary_hash,
            change_count: changes.len(),
            record_change_count: record_changes.len(),
            knowledge_batch_count: pending_knowledge_changes.len(),
            knowledge_record_count: pending_knowledge_changes
                .iter()
                .map(|change| change.records.len())
                .sum(),
            allocations: allocation_records,
            created_persons: created_persons
                .iter()
                .map(super::CreatedPerson::from)
                .collect(),
            transition_audits: transition_evidence.audits,
        })
    }

    /// Checks the `expected_post` versions of the transition manifests
    /// committed in this boundary. Same-boundary phase-10 writes are already
    /// committed; the deferred next-boundary record writes of phases 7 and 10
    /// are applied to a candidate overlay in the order the end-of-boundary
    /// stage applies them. Phase-12 and phase-13 writes are not yet proposed,
    /// so they may still change a checked record afterwards.
    fn check_transition_post_versions(
        &self,
        ledger: &mut super::transitions::BoundaryTransitionLedger,
        record_schemas: &records::DomainRecordSchemas,
        deferred: &[StagedBoundaryDirective],
    ) -> Result<(), CanwuError> {
        let mut candidate = BTreeMap::new();
        extend_boundary_record_candidate_overlay(
            &BoundaryRecordOverlayContext {
                current: &self.state.current,
                now: self.state.scheduler.now,
                scheduled_actions: &self.state.scheduler.actions,
                run_configuration: &self.state.metadata.run_configuration,
                schemas: record_schemas,
            },
            &mut candidate,
            deferred,
        )?;
        ledger.check_expected_post(&|record| {
            candidate
                .get(record)
                .or_else(|| self.state.current.domain_records.get(record))
                .map(|record| record.version)
        })
    }

    fn apply_boundary_stage(
        &mut self,
        boundary_id: BoundaryId,
        correlation_id: u64,
        directives: Vec<StagedBoundaryDirective>,
        evidence: &mut PendingBoundaryEvidence,
    ) -> Result<(), CanwuError> {
        let random_decision_count = directives
            .iter()
            .filter(|staged| {
                matches!(
                    &staged.directive,
                    BoundaryDirective::ResolveDecisionRandomly { .. }
                )
            })
            .count();
        if evidence
            .random_decisions
            .len()
            .checked_add(random_decision_count)
            .is_none_or(|count| count > 1)
        {
            return Err(CanwuError::new(
                ErrorCode::InvalidDecision,
                "one boundary may generate at most one random decision resolution",
            ));
        }
        let changes = &mut evidence.changes;
        let record_changes = &mut evidence.record_changes;
        let emissions = &mut evidence.emissions;
        let generated_ingress = &mut evidence.generated_ingress;
        let mutation_requests: Vec<_> = directives
            .iter()
            .filter_map(|staged| match &staged.directive {
                BoundaryDirective::MutateRecord { mutation, summary } => {
                    Some(records::DomainMutationRequest {
                        plugin: &staged.plugin,
                        system: &staged.system,
                        visibility: staged.visibility,
                        mutation,
                        summary,
                    })
                }
                BoundaryDirective::SetComponent { .. }
                | BoundaryDirective::Emit { .. }
                | BoundaryDirective::ScheduleIngress { .. }
                | BoundaryDirective::SchedulePluginIngress { .. }
                | BoundaryDirective::ResolveDecisionRandomly { .. }
                | BoundaryDirective::PublishKnowledge { .. }
                | BoundaryDirective::SetPersonAvailability { .. }
                | BoundaryDirective::CreatePerson { .. }
                | BoundaryDirective::CancelPluginIngress { .. }
                | BoundaryDirective::RecordEvaluationTrace { .. }
                | BoundaryDirective::RegisterTransitionManifest { .. }
                | BoundaryDirective::StageTransitionWrite { .. } => None,
            })
            .collect();
        let mut stage_record_changes = BTreeMap::new();
        if !mutation_requests.is_empty() {
            let (next_records, applied) = records::apply_mutation_bundle_cow(
                &self.state.current.domain_records,
                &self.plugins.record_schemas,
                self.state.scheduler.now,
                &|entity| runtime_entity_exists(&self.state, entity),
                mutation_requests,
            )?;
            let first_index = record_changes.len();
            for (offset, change) in applied.iter().enumerate() {
                let index = first_index.checked_add(offset).ok_or_else(|| {
                    CanwuError::new(
                        ErrorCode::IdentifierExhausted,
                        "boundary record-change index exceeds the persistent identifier space",
                    )
                })?;
                let index = u64::try_from(index).map_err(|_| {
                    CanwuError::new(
                        ErrorCode::IdentifierExhausted,
                        "boundary record-change index exceeds the persistent identifier space",
                    )
                })?;
                index_current_domain_record_version(&mut self.state, boundary_id, index, change);
                stage_record_changes
                    .insert(change.current.reference.clone(), (index, change.clone()));
            }
            self.invalidate_commitments(CommitmentDomains::DOMAIN_RECORDS);
            self.state.current.domain_records = next_records;
            record_changes.extend(applied);
        }

        for staged in &directives {
            let unavailable = match &staged.directive {
                BoundaryDirective::SetComponent { entity, .. } => {
                    (!runtime_entity_exists(&self.state, entity)).then_some(entity)
                }
                BoundaryDirective::Emit { affected, .. } => affected
                    .iter()
                    .find(|entity| !runtime_entity_exists(&self.state, entity)),
                BoundaryDirective::ScheduleIngress { affected, .. } => affected
                    .iter()
                    .find(|entity| !runtime_entity_identity_exists(&self.state, entity)),
                BoundaryDirective::SchedulePluginIngress { affected, .. } => affected
                    .iter()
                    .find(|entity| !runtime_entity_identity_exists(&self.state, entity)),
                BoundaryDirective::MutateRecord { .. }
                | BoundaryDirective::ResolveDecisionRandomly { .. }
                | BoundaryDirective::PublishKnowledge { .. }
                | BoundaryDirective::SetPersonAvailability { .. }
                | BoundaryDirective::CreatePerson { .. }
                | BoundaryDirective::CancelPluginIngress { .. }
                | BoundaryDirective::RecordEvaluationTrace { .. }
                | BoundaryDirective::RegisterTransitionManifest { .. }
                | BoundaryDirective::StageTransitionWrite { .. } => None,
            };
            if let Some(entity) = unavailable {
                return Err(CanwuError::new(
                    ErrorCode::EntityNotFound,
                    format!(
                        "boundary stage {}.{} references unavailable entity {entity}",
                        staged.plugin, staged.system
                    ),
                )
                .with_entity(entity.clone()));
            }
        }

        for staged in directives {
            match staged.directive {
                BoundaryDirective::SetComponent {
                    state,
                    entity,
                    component,
                    value,
                    summary,
                } => {
                    let key = component_key(&staged.plugin, &state, &entity, &component);
                    self.invalidate_commitments(CommitmentDomains::PLUGIN_COMPONENTS);
                    let previous = self
                        .state
                        .current
                        .plugin_components
                        .get(&key)
                        .map(|record| record.value.clone());
                    self.state.current.plugin_components.insert(
                        key,
                        PluginComponentRecord {
                            plugin: staged.plugin.clone(),
                            state: state.clone(),
                            entity: entity.clone(),
                            component: component.clone(),
                            value: value.clone(),
                        },
                    );
                    let change_index = u64::try_from(changes.len()).map_err(|_| {
                        CanwuError::new(
                            ErrorCode::IdentifierExhausted,
                            "boundary change index exceeds the persistent identifier space",
                        )
                    })?;
                    changes.push(BoundaryChange {
                        plugin: staged.plugin.clone(),
                        system: staged.system.clone(),
                        state,
                        entity: entity.clone(),
                        component: component.clone(),
                        previous,
                        value,
                        visibility: staged.visibility,
                        summary: summary.clone(),
                    });
                    let event = self.append_event(
                        EventKind::plugin(staged.plugin.clone(), format!("{component}_changed")),
                        vec![entity],
                        summary,
                        Some(CauseRef::Boundary(boundary_id)),
                        correlation_id,
                    )?;
                    emissions.push(BoundaryEmission {
                        plugin: staged.plugin,
                        system: staged.system,
                        event: event.id,
                        kind: BoundaryEmissionKind::Change { change_index },
                    });
                }
                BoundaryDirective::MutateRecord { mutation, .. } => {
                    let Some((change_index, change)) = stage_record_changes.get(mutation.target())
                    else {
                        return Err(CanwuError::new(
                            ErrorCode::InvalidBoundary,
                            "record mutation is missing its committed change evidence",
                        ));
                    };
                    let event = self.append_event(
                        EventKind::plugin(staged.plugin.clone(), change.operation.event_type()),
                        record_change_affected_entities(change),
                        change.summary.clone(),
                        Some(CauseRef::Boundary(boundary_id)),
                        correlation_id,
                    )?;
                    emissions.push(BoundaryEmission {
                        plugin: staged.plugin,
                        system: staged.system,
                        event: event.id,
                        kind: BoundaryEmissionKind::RecordChange {
                            change_index: *change_index,
                        },
                    });
                }
                BoundaryDirective::Emit {
                    event_type,
                    summary,
                    affected,
                } => {
                    let event = self.append_event(
                        EventKind::plugin(staged.plugin.clone(), event_type),
                        affected,
                        summary,
                        Some(CauseRef::Boundary(boundary_id)),
                        correlation_id,
                    )?;
                    emissions.push(BoundaryEmission {
                        plugin: staged.plugin,
                        system: staged.system,
                        event: event.id,
                        kind: BoundaryEmissionKind::Explicit,
                    });
                }
                BoundaryDirective::PublishKnowledge { .. } => {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidBoundary,
                        "knowledge publication execution is not enabled in this runtime slice",
                    ));
                }
                BoundaryDirective::ScheduleIngress {
                    after,
                    packet_type,
                    priority,
                    payload,
                    mut affected,
                } => {
                    self.ensure_canonical_ingress_can_start()?;
                    let descriptor = self
                        .plugins
                        .ingress
                        .get(&(staged.plugin.clone(), packet_type.clone()))
                        .ok_or_else(|| {
                            CanwuError::new(
                                ErrorCode::InvalidPayload,
                                format!(
                                    "boundary system {}.{} scheduled undeclared ingress type {packet_type}",
                                    staged.plugin, staged.system
                                ),
                            )
                        })?
                        .clone();
                    descriptor.payload_schema.validate(&payload)?;
                    affected.sort();
                    affected.dedup();
                    let due_at = self.state.scheduler.now.checked_add(after).ok_or_else(|| {
                        CanwuError::new(
                            ErrorCode::InvalidDuration,
                            "boundary-generated ingress exceeds the supported time range",
                        )
                    })?;
                    let receipt = self.append_ingress(
                        due_at,
                        descriptor.class,
                        priority,
                        IngressPayload::Plugin {
                            plugin: staged.plugin.clone(),
                            packet_type,
                            payload,
                            affected_entities: affected,
                            archive_retention: Vec::new(),
                        },
                        Some(CauseRef::Boundary(boundary_id)),
                        true,
                    )?;
                    generated_ingress.push(BoundaryIngressGeneration {
                        ingress: receipt.ingress_id,
                        plugin: staged.plugin,
                        system: staged.system,
                        phase: staged.phase,
                        visibility: staged.visibility,
                    });
                }
                BoundaryDirective::SchedulePluginIngress {
                    target_plugin,
                    after,
                    packet_type,
                    priority,
                    payload,
                    mut affected,
                } => {
                    self.ensure_canonical_ingress_can_start()?;
                    let descriptor = self
                        .plugins
                        .ingress
                        .get(&(target_plugin.clone(), packet_type.clone()))
                        .ok_or_else(|| {
                            CanwuError::new(
                                ErrorCode::InvalidPayload,
                                format!(
                                    "boundary system {}.{} scheduled undeclared target ingress {}.{packet_type}",
                                    staged.plugin, staged.system, target_plugin
                                ),
                            )
                        })?
                        .clone();
                    descriptor.payload_schema.validate(&payload)?;
                    affected.sort();
                    affected.dedup();
                    let due_at = self.state.scheduler.now.checked_add(after).ok_or_else(|| {
                        CanwuError::new(
                            ErrorCode::InvalidDuration,
                            "boundary-generated cross-plugin ingress exceeds the supported time range",
                        )
                    })?;
                    let receipt = self.append_ingress(
                        due_at,
                        descriptor.class,
                        priority,
                        IngressPayload::Plugin {
                            plugin: target_plugin,
                            packet_type,
                            payload,
                            affected_entities: affected,
                            archive_retention: Vec::new(),
                        },
                        Some(CauseRef::Boundary(boundary_id)),
                        true,
                    )?;
                    generated_ingress.push(BoundaryIngressGeneration {
                        ingress: receipt.ingress_id,
                        plugin: staged.plugin,
                        system: staged.system,
                        phase: staged.phase,
                        visibility: staged.visibility,
                    });
                }
                BoundaryDirective::CancelPluginIngress { ingress_id, reason } => {
                    self.ensure_canonical_ingress_can_start()?;
                    let cancelled_in_this_boundary = generated_ingress.iter().any(|generation| {
                        self.state
                            .evidence
                            .retained_ingress(generation.ingress)
                            .is_some_and(|record| {
                                matches!(
                                    record.payload,
                                    IngressPayload::PluginCancellation { cancelled, .. }
                                        if cancelled == ingress_id
                                )
                            })
                    });
                    if cancelled_in_this_boundary {
                        return Err(CanwuError::new(
                            ErrorCode::InvalidBoundary,
                            format!(
                                "multiple boundary proposals cancel ingress {ingress_id}; boundary system {}.{} must not repeat a cancellation",
                                staged.plugin, staged.system
                            ),
                        ));
                    }
                    let target = self.plugin_ingress_cancellation_target(
                        ingress_id,
                        IngressCancellationAuthority::BoundarySystem,
                        PluginIngressCancellationProof {
                            permit: None,
                            replay: false,
                            boundary_plugin: Some(&staged.plugin),
                            current_generations: generated_ingress,
                        },
                        &reason,
                    )?;
                    let receipt = self.append_plugin_ingress_cancellation(
                        target,
                        IngressCancellationAuthority::BoundarySystem,
                        reason,
                        Some(CauseRef::Boundary(boundary_id)),
                        true,
                    )?;
                    generated_ingress.push(BoundaryIngressGeneration {
                        ingress: receipt.ingress_id,
                        plugin: staged.plugin,
                        system: staged.system,
                        phase: staged.phase,
                        visibility: staged.visibility,
                    });
                }
                BoundaryDirective::ResolveDecisionRandomly { resolution } => {
                    evidence
                        .random_decisions
                        .push(PendingRandomDecisionResolution {
                            plugin: staged.plugin,
                            system: staged.system,
                            phase: staged.phase,
                            visibility: staged.visibility,
                            resolution,
                        });
                }
                BoundaryDirective::SetPersonAvailability {
                    person,
                    availability,
                    summary,
                } => {
                    let change = self.apply_person_availability(
                        (
                            &staged.plugin,
                            &staged.system,
                            staged.phase,
                            staged.visibility,
                        ),
                        person,
                        availability,
                        summary,
                    )?;
                    evidence.person_availability_changes.push(change);
                }
                BoundaryDirective::CreatePerson {
                    draft,
                    correlation,
                    summary,
                } => {
                    let creation = self.apply_person_creation(
                        &staged.plugin,
                        &staged.system,
                        draft,
                        correlation,
                        summary,
                    )?;
                    evidence.created_persons.push(creation);
                }
                BoundaryDirective::RecordEvaluationTrace { .. } => {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidBoundary,
                        "evaluation traces are boundary evidence and never reach a commit stage",
                    ));
                }
                BoundaryDirective::RegisterTransitionManifest { .. }
                | BoundaryDirective::StageTransitionWrite { .. } => {
                    return Err(transition_directive_not_admitted(
                        &staged.plugin,
                        &staged.system,
                    ));
                }
            }
        }
        validate_runtime_domain_dependents(&self.state)?;
        Ok(())
    }

    fn materialize_boundary_random_decisions(
        &mut self,
        boundary_id: BoundaryId,
        pending: &[PendingRandomDecisionResolution],
        committed_draws: &[CommittedBoundaryRandomDraw],
        generated_ingress: &mut Vec<BoundaryIngressGeneration>,
    ) -> Result<(), CanwuError> {
        let expected_revision = self.revision().checked_add(1).ok_or_else(|| {
            CanwuError::new(
                ErrorCode::IdentifierExhausted,
                "random decision target revision is exhausted",
            )
        })?;
        let draws = committed_draws
            .iter()
            .map(|draw| ((draw.stream.clone(), draw.address.clone()), draw.id))
            .collect::<BTreeMap<_, _>>();
        for pending in pending {
            let resolution = &pending.resolution;
            let draw_id = draws
                .get(&(
                    resolution.sample.stream.clone(),
                    resolution.sample.address.clone(),
                ))
                .copied()
                .ok_or_else(|| {
                    CanwuError::new(
                        ErrorCode::InvalidRandomDraw,
                        "random decision draw was not committed by its source boundary",
                    )
                })?;
            let ticket = self
                .state
                .current
                .decisions
                .ticket(resolution.ticket_id)
                .cloned()
                .ok_or_else(|| {
                    CanwuError::new(
                        ErrorCode::InvalidDecision,
                        "random decision ticket disappeared before ingress generation",
                    )
                })?;
            let controller = self
                .state
                .current
                .decisions
                .controller(&resolution.controller_id)
                .cloned()
                .ok_or_else(|| {
                    CanwuError::new(
                        ErrorCode::InvalidDecision,
                        "random decision controller disappeared before ingress generation",
                    )
                })?;
            let option_id = random_resolution_selection(&ticket, resolution)?;
            let option = ticket.option(&option_id).ok_or_else(|| {
                CanwuError::new(
                    ErrorCode::InvalidDecision,
                    "random decision selected an unavailable option",
                )
            })?;
            let due_at = self.state.scheduler.now;
            let command = match &option.action {
                DecisionAction::Command { command } => {
                    let request_id = resolution.command_request_id.ok_or_else(|| {
                        CanwuError::new(
                            ErrorCode::InvalidDecision,
                            "random decision command option lacks a command request ID",
                        )
                    })?;
                    let command: Command =
                        serde_json::from_value(command.clone()).map_err(|error| {
                            CanwuError::new(
                                ErrorCode::InvalidDecision,
                                format!("decision option contains an invalid command: {error}"),
                            )
                        })?;
                    Some(CommandRequest::new(
                        request_id,
                        expected_revision,
                        CommandEnvelope::new(
                            super::decision::controller_issuer(&controller),
                            command,
                        )
                        .with_authority(super::decision::controller_authority(&controller))
                        .at_time(due_at),
                    ))
                }
                DecisionAction::None => None,
            };
            let random = Some(DecisionRandomEvidence {
                draw_id,
                value: resolution.sample.value,
                upper_exclusive: resolution.sample.upper_exclusive,
                option_weights: resolution.option_weights.clone(),
            });
            let outcome = DecisionOutcome::Selected {
                option_id: option_id.clone(),
            };
            let decision = match &resolution.tie_break {
                None => PolicyDecision {
                    outcome,
                    summary: random_policy_summary(&option_id),
                    evaluations: Vec::new(),
                    external: None,
                    random,
                    stage: None,
                    fired_guards: Vec::new(),
                },
                Some(pending) => PolicyDecision {
                    outcome,
                    summary: random_tie_break_summary(&option_id),
                    evaluations: pending.evaluations.clone(),
                    external: None,
                    random,
                    stage: Some(DecisionStage::Random),
                    fired_guards: pending.fired_guards.clone(),
                },
            };
            let mutation = DecisionMutation::Resolve {
                ticket_id: ticket.id,
                expected_version: ticket.version,
                controller_id: controller.id.clone(),
                policy: controller.policy.clone(),
                decision,
                command_request_id: resolution.command_request_id,
            };
            let mut request = DecisionIngressRequest::new(
                resolution.decision_request_id,
                expected_revision,
                mutation,
            );
            if let Some(command) = command {
                request = request.with_command(command);
            }
            let receipt = self.append_boundary_decision_ingress(
                boundary_id,
                due_at,
                resolution.priority,
                request,
            )?;
            generated_ingress.push(BoundaryIngressGeneration {
                ingress: receipt.ingress_id,
                plugin: pending.plugin.clone(),
                system: pending.system.clone(),
                phase: pending.phase,
                visibility: pending.visibility,
            });
        }
        Ok(())
    }

    fn stage_knowledge_publications(
        &mut self,
        phase: BoundaryPhase,
        directives: Vec<StagedBoundaryDirective>,
        visible_overlay: &mut BTreeMap<
            KnowledgeHolderRef,
            BTreeMap<KnowledgeRecordId, KnowledgeRecord>,
        >,
        pending: &mut Vec<BoundaryKnowledgeChange>,
        correlations: &mut BTreeSet<(String, String, String)>,
    ) -> Result<(), CanwuError> {
        let new_record_count = directives
            .iter()
            .map(|staged| match &staged.directive {
                BoundaryDirective::PublishKnowledge { records, .. } => records.len(),
                _ => 0,
            })
            .sum::<usize>();
        let total_records = pending
            .iter()
            .map(|change| change.records.len())
            .sum::<usize>()
            .checked_add(new_record_count)
            .ok_or_else(|| {
                CanwuError::new(
                    ErrorCode::KnowledgeLimitExceeded,
                    "boundary knowledge record count exceeds platform range",
                )
            })?;
        if total_records > crate::KnowledgeLimitsV1::CURRENT.records_per_boundary {
            return Err(CanwuError::new(
                ErrorCode::KnowledgeLimitExceeded,
                "boundary knowledge record limit exceeded",
            ));
        }
        for staged in directives {
            let BoundaryDirective::PublishKnowledge {
                holder,
                visibility,
                producer_correlation,
                records: drafts,
                summary,
            } = staged.directive
            else {
                return Err(CanwuError::new(
                    ErrorCode::InvalidBoundary,
                    "knowledge stage received an ordinary directive",
                ));
            };
            if let Some(value) = &producer_correlation
                && !correlations.insert((
                    staged.plugin.clone(),
                    staged.system.clone(),
                    value.clone(),
                ))
            {
                return Err(CanwuError::new(
                    ErrorCode::InvalidKnowledgeRecord,
                    "producer correlation is duplicated within one system and boundary",
                ));
            }
            let mut records = Vec::with_capacity(drafts.len());
            for draft in drafts {
                let (id, next_id) = claim_counter(
                    self.state.counters.next_knowledge_record_id,
                    "knowledge record ID",
                )?;
                self.state.counters.next_knowledge_record_id = next_id;
                let record = KnowledgeRecord {
                    id: KnowledgeRecordId::new(id),
                    holder: holder.clone(),
                    schema: draft.schema,
                    subjects: draft.subjects,
                    payload: draft.payload,
                    as_of: draft.as_of,
                    learned_at: self.state.scheduler.now,
                    confidence_per_mille: draft.confidence_per_mille,
                    origin: draft.origin,
                    supersedes: draft.supersedes,
                    contradicts: draft.contradicts,
                };
                if visibility == StateVisibility::SameBoundary {
                    visible_overlay
                        .entry(holder.clone())
                        .or_default()
                        .insert(record.id, record.clone());
                }
                records.push(record);
            }
            pending.push(BoundaryKnowledgeChange {
                plugin: staged.plugin,
                system: staged.system,
                phase,
                holder,
                producer_correlation,
                records,
                visibility,
                summary,
            });
        }
        Ok(())
    }

    fn commit_knowledge_publications(
        &mut self,
        boundary_id: BoundaryId,
        correlation_id: u64,
        changes: &[BoundaryKnowledgeChange],
        emissions: &mut Vec<BoundaryEmission>,
    ) -> Result<(), CanwuError> {
        if changes.is_empty() {
            return Ok(());
        }
        let mut ledger = self.state.current.knowledge.records.clone();
        for change in changes {
            let holder = ledger.entry(change.holder.clone()).or_default();
            for record in &change.records {
                if holder.insert(record.id, record.clone()).is_some() {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidKnowledgeRecord,
                        "knowledge publication attempted to reuse a global record ID",
                    ));
                }
            }
        }
        self.state.current.knowledge.records = ledger;
        self.invalidate_commitments(CommitmentDomains::KNOWLEDGE);
        for (index, change) in changes.iter().enumerate() {
            let record_count = u32::try_from(change.records.len()).map_err(|_| {
                CanwuError::new(
                    ErrorCode::KnowledgeLimitExceeded,
                    "knowledge publication event count exceeds u32",
                )
            })?;
            let affected = match &change.holder {
                KnowledgeHolderRef::Person(person) => vec![EntityRef::Person(*person)],
                KnowledgeHolderRef::Entity(entity) => vec![entity.clone()],
            };
            let event = self.append_event(
                KnowledgePublished {
                    holder: change.holder.clone(),
                    record_count,
                }
                .into_kind(),
                affected,
                change.summary.clone(),
                Some(CauseRef::Boundary(boundary_id)),
                correlation_id,
            )?;
            emissions.push(BoundaryEmission {
                plugin: change.plugin.clone(),
                system: change.system.clone(),
                event: event.id,
                kind: BoundaryEmissionKind::KnowledgeChange {
                    change_index: u64::try_from(index).map_err(|_| {
                        CanwuError::new(
                            ErrorCode::IdentifierExhausted,
                            "knowledge change index exceeds identifier space",
                        )
                    })?,
                },
            });
        }
        Ok(())
    }

    pub(super) fn apply_directives(
        &mut self,
        plugin: &str,
        directives: Vec<SystemDirective>,
        allowed_writes: &[StateKey],
        cause: &CauseRef,
        correlation_id: u64,
    ) -> Result<(), CanwuError> {
        for directive in directives {
            match directive {
                SystemDirective::SetComponent {
                    state,
                    entity,
                    component,
                    value,
                    summary,
                } => {
                    let key = component_key(plugin, &state, &entity, &component);
                    self.invalidate_commitments(CommitmentDomains::PLUGIN_COMPONENTS);
                    self.state.current.plugin_components.insert(
                        key,
                        PluginComponentRecord {
                            plugin: plugin.to_owned(),
                            state,
                            entity: entity.clone(),
                            component: component.clone(),
                            value,
                        },
                    );
                    self.emit(
                        EventKind::plugin(plugin, format!("{component}_changed")),
                        vec![entity],
                        summary,
                        Some(cause.clone()),
                        correlation_id,
                    )?;
                }
                SystemDirective::Emit {
                    event_type,
                    summary,
                    affected,
                } => {
                    self.emit(
                        EventKind::plugin(plugin, event_type),
                        affected,
                        summary,
                        Some(cause.clone()),
                        correlation_id,
                    )?;
                }
                SystemDirective::Schedule { after, directive } => {
                    let at = self.state.scheduler.now.checked_add(after).ok_or_else(|| {
                        CanwuError::new(
                            ErrorCode::InvalidDuration,
                            "plugin scheduled time exceeds the supported range",
                        )
                    })?;
                    self.schedule_at(
                        at,
                        ScheduledAction::PluginDirective {
                            plugin: plugin.to_owned(),
                            directive,
                            allowed_writes: allowed_writes.to_vec(),
                            cause: cause.clone(),
                            correlation_id,
                        },
                    )?;
                }
                SystemDirective::EnqueuePluginIngress {
                    after,
                    packet_type,
                    priority,
                    payload,
                    mut affected,
                } => {
                    self.ensure_canonical_ingress_can_start()?;
                    let descriptor = self
                        .plugins
                        .ingress
                        .get(&(plugin.to_owned(), packet_type.clone()))
                        .ok_or_else(|| {
                            CanwuError::new(
                                ErrorCode::InvalidPayload,
                                format!(
                                    "plugin command scheduled unregistered ingress type {plugin}.{packet_type}"
                                ),
                            )
                        })?
                        .clone();
                    descriptor.payload_schema.validate(&payload)?;
                    affected.sort();
                    affected.dedup();
                    if affected
                        .iter()
                        .any(|entity| !runtime_entity_identity_exists(&self.state, entity))
                    {
                        return Err(CanwuError::new(
                            ErrorCode::EntityNotFound,
                            "plugin command ingress references an unknown entity identity",
                        ));
                    }
                    let due_at = self.state.scheduler.now.checked_add(after).ok_or_else(|| {
                        CanwuError::new(
                            ErrorCode::InvalidDuration,
                            "plugin command ingress exceeds the supported time range",
                        )
                    })?;
                    self.append_ingress(
                        due_at,
                        descriptor.class,
                        priority,
                        IngressPayload::Plugin {
                            plugin: plugin.to_owned(),
                            packet_type,
                            payload,
                            affected_entities: affected,
                            archive_retention: Vec::new(),
                        },
                        Some(cause.clone()),
                        true,
                    )?;
                }
            }
        }
        Ok(())
    }
}

fn index_current_domain_record_version(
    state: &mut RuntimeState,
    boundary: BoundaryId,
    change_index: u64,
    change: &DomainRecordChange,
) {
    // Records change only while their boundary settles, when `now` is that
    // boundary's committed time.
    let established_at = state.scheduler.now;
    state.metadata.current_domain_record_versions.insert(
        change.current.reference.clone(),
        super::CurrentDomainRecordVersion {
            version: super::DomainRecordVersionRef {
                record: change.current.reference.clone(),
                version: change.current.version,
                established_by: super::DomainRecordVersionSource::BoundaryChange {
                    boundary,
                    change_index,
                },
            },
            established_at,
        },
    );
}

struct PendingReservationOffer {
    plugin: String,
    system: String,
    offer: ReservationOffer,
}

struct PendingReservationRequest {
    reservation: ReservationRef,
    request: ReservationRequest,
}

struct ReservationAllocationResult {
    by_reservation: BTreeMap<ReservationRef, ReservationAllocation>,
    offers: Vec<ReservationOfferRecord>,
    requests: Vec<ReservationRequestRecord>,
    records: Vec<ReservationAllocation>,
}

struct StagedBoundaryDirective {
    plugin: String,
    system: String,
    phase: BoundaryPhase,
    visibility: StateVisibility,
    directive: BoundaryDirective,
}

#[derive(Default)]
struct PendingBoundaryEvidence {
    changes: Vec<BoundaryChange>,
    record_changes: Vec<DomainRecordChange>,
    emissions: Vec<BoundaryEmission>,
    generated_ingress: Vec<BoundaryIngressGeneration>,
    random_decisions: Vec<PendingRandomDecisionResolution>,
    person_availability_changes: Vec<super::BoundaryPersonAvailabilityChange>,
    created_persons: Vec<super::BoundaryPersonCreation>,
}

struct PendingRandomDecisionResolution {
    plugin: String,
    system: String,
    phase: BoundaryPhase,
    visibility: StateVisibility,
    resolution: super::RandomDecisionResolution,
}

fn proposal_evidence_refs(
    boundary: BoundaryId,
    pending: &PendingBoundaryEvidence,
) -> BTreeSet<EvidenceRef> {
    let mut values = BTreeSet::new();
    for (index, change) in pending.record_changes.iter().enumerate() {
        if let Ok(change_index) = u64::try_from(index) {
            values.insert(EvidenceRef::DomainRecordVersion(
                super::DomainRecordVersionRef {
                    record: change.current.reference.clone(),
                    version: change.current.version,
                    established_by: DomainRecordVersionSource::BoundaryChange {
                        boundary,
                        change_index,
                    },
                },
            ));
        }
    }
    values.extend(
        pending
            .emissions
            .iter()
            .map(|emission| EvidenceRef::Event(emission.event)),
    );
    values
}

pub(super) struct PendingBoundaryRandomDraw {
    pub(super) plugin: String,
    pub(super) system: String,
    pub(super) draw: random::PendingRandomDraw,
}

pub(super) struct CommittedBoundaryRandomDraw {
    pub(super) id: super::RandomDrawId,
    pub(super) stream: super::RandomStreamKey,
    pub(super) address: RandomDrawAddress,
}

pub(super) fn boundary_system_due(
    contract: &BoundarySystemContract,
    cadences: &[SystemCadence],
    has_admitted_events: bool,
) -> bool {
    match contract.cadence {
        SystemCadence::EventDriven => has_admitted_events,
        _ => cadences.contains(&contract.cadence),
    }
}

pub(super) fn boundary_has_event_ingress(record: &BoundaryRecord) -> bool {
    !record.admitted_events.is_empty() || !record.admitted_ingress.is_empty()
}

#[allow(clippy::too_many_arguments)]
fn validate_boundary_proposal(
    plugin: &str,
    contract: &BoundarySystemContract,
    current: &RuntimeCurrentState,
    committed_availability: &BTreeMap<super::PersonId, super::PersonAvailability>,
    now: SimTime,
    runtime: &RuntimeState,
    boundary_id: BoundaryId,
    pending_evidence: &PendingBoundaryEvidence,
    plugins: &PluginRegistry,
    record_overlay: &BTreeMap<DomainRecordRef, DomainRecord>,
    knowledge_overlay: &BTreeMap<KnowledgeHolderRef, BTreeMap<KnowledgeRecordId, KnowledgeRecord>>,
    proposal: &BoundaryProposal,
    pending_random_draws: &[random::PendingRandomDraw],
) -> Result<(), CanwuError> {
    if contract.phase != BoundaryPhase::ReservationAndAllocation
        && (!proposal.offers.is_empty() || !proposal.requests.is_empty())
    {
        return Err(CanwuError::new(
            ErrorCode::InvalidBoundary,
            format!(
                "boundary system {plugin}.{} proposed reservations in phase {:?}",
                contract.name, contract.phase
            ),
        ));
    }

    let entity_exists = |entity: &EntityRef| {
        proposal_entity_exists(
            current,
            &plugins.record_schemas,
            record_overlay,
            proposal,
            entity,
        )
    };
    let mut offered_pools = BTreeSet::new();
    for offer in &proposal.offers {
        validate_reservation_pool(&offer.pool, &entity_exists)?;
        if !contract.reservation_offers.contains(&offer.pool.state)
            || plugins
                .state_owners
                .get(&offer.pool.state)
                .is_none_or(|owner| owner != plugin)
        {
            return Err(CanwuError::new(
                ErrorCode::InvalidBoundary,
                format!(
                    "boundary system {plugin}.{} offered undeclared state {}.{}",
                    contract.name, offer.pool.state.namespace, offer.pool.state.name
                ),
            ));
        }
        if !offered_pools.insert(&offer.pool) {
            return Err(CanwuError::new(
                ErrorCode::InvalidBoundary,
                format!(
                    "boundary system {plugin}.{} offered the same reservation pool twice",
                    contract.name
                ),
            ));
        }
    }

    let mut request_names = BTreeSet::new();
    for request in &proposal.requests {
        validate_reservation_pool(&request.pool, &entity_exists)?;
        if request.request.trim().is_empty()
            || request.request != request.request.trim()
            || request.tie_break.trim().is_empty()
            || request.tie_break != request.tie_break.trim()
            || request.quantity == 0
            || !request_names.insert(&request.request)
            || !contract.reservation_requests.contains(&request.pool.state)
        {
            return Err(CanwuError::new(
                ErrorCode::InvalidBoundary,
                format!(
                    "boundary system {plugin}.{} produced an invalid reservation request",
                    contract.name
                ),
            ));
        }
    }

    let publication_count = proposal
        .directives
        .iter()
        .filter(|directive| matches!(directive, BoundaryDirective::PublishKnowledge { .. }))
        .count();
    if publication_count > crate::KnowledgeLimitsV1::CURRENT.batches_per_system_boundary {
        return Err(CanwuError::new(
            ErrorCode::KnowledgeLimitExceeded,
            "system knowledge publication batch limit exceeded",
        ));
    }
    let mut component_keys = BTreeSet::new();
    let mut record_targets = BTreeSet::new();
    let mut producer_correlations = BTreeSet::new();
    let mut canonical_drafts = BTreeSet::new();
    let mut random_decision_samples = BTreeSet::new();
    let mut cancelled_ingress = BTreeSet::new();
    for directive in &proposal.directives {
        match directive {
            BoundaryDirective::SetComponent {
                state: state_key,
                entity,
                component,
                ..
            } => {
                if component.trim().is_empty()
                    || component != component.trim()
                    || !contract.writes.contains(state_key)
                    || plugins
                        .state_owners
                        .get(state_key)
                        .is_none_or(|owner| owner != plugin)
                    || is_domain_record_state(&plugins.record_schemas, state_key)
                {
                    return Err(CanwuError::new(
                        ErrorCode::UndeclaredStateWrite,
                        format!(
                            "boundary system {plugin}.{} produced an undeclared component write",
                            contract.name
                        ),
                    ));
                }
                if !entity_exists(entity) {
                    return Err(CanwuError::new(
                        ErrorCode::EntityNotFound,
                        format!(
                            "boundary system {plugin}.{} targeted missing entity {entity}",
                            contract.name
                        ),
                    )
                    .with_entity(entity.clone()));
                }
                let key = component_key(plugin, state_key, entity, component);
                if !component_keys.insert(key) {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidBoundary,
                        format!(
                            "boundary system {plugin}.{} wrote the same component twice",
                            contract.name
                        ),
                    ));
                }
            }
            BoundaryDirective::MutateRecord { mutation, summary } => {
                let target = mutation.target();
                let state_key = records::record_state_key(&target.kind);
                if !canonical_text(summary)
                    || !contract.writes.contains(&state_key)
                    || plugins
                        .state_owners
                        .get(&state_key)
                        .is_none_or(|owner| owner != plugin)
                    || plugins
                        .record_schemas
                        .get(&target.kind)
                        .is_none_or(|(owner, _)| owner != plugin)
                {
                    return Err(CanwuError::new(
                        ErrorCode::UndeclaredStateWrite,
                        format!(
                            "boundary system {plugin}.{} produced an undeclared record mutation",
                            contract.name
                        ),
                    ));
                }
                if !record_targets.insert(target.clone()) {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidBoundary,
                        format!(
                            "boundary system {plugin}.{} mutated the same record twice",
                            contract.name
                        ),
                    ));
                }
            }
            BoundaryDirective::Emit {
                event_type,
                affected,
                ..
            } => {
                if event_type.trim().is_empty()
                    || event_type != event_type.trim()
                    || !contract.emits.contains(event_type)
                {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidBoundary,
                        format!(
                            "boundary system {plugin}.{} emitted an undeclared event type",
                            contract.name
                        ),
                    ));
                }
                if affected.iter().any(|entity| !entity_exists(entity)) {
                    return Err(CanwuError::new(
                        ErrorCode::EntityNotFound,
                        format!(
                            "boundary system {plugin}.{} emitted an event for a missing entity",
                            contract.name
                        ),
                    ));
                }
            }
            BoundaryDirective::PublishKnowledge {
                holder,
                visibility,
                producer_correlation,
                records,
                summary,
            } => {
                if !matches!(
                    contract.phase,
                    BoundaryPhase::PerceptionAndAttentionRefresh
                        | BoundaryPhase::PerspectiveAndReportMaterialization
                ) {
                    return Err(CanwuError::new(
                        ErrorCode::UndeclaredKnowledgeWrite,
                        "knowledge publication is allowed only in phases 4 and 13",
                    ));
                }
                if records.is_empty()
                    || records.len() > crate::KnowledgeLimitsV1::CURRENT.records_per_batch
                {
                    return Err(CanwuError::new(
                        ErrorCode::KnowledgeLimitExceeded,
                        "knowledge publication batch is empty or exceeds its record limit",
                    ));
                }
                if !canonical_text(summary)
                    || summary.len() > crate::KnowledgeLimitsV1::CURRENT.text_bytes
                {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidKnowledgeRecord,
                        "knowledge publication summary is not canonical or exceeds its limit",
                    ));
                }
                if let Some(value) = producer_correlation
                    && (!canonical_text(value)
                        || value.len() > 256
                        || !producer_correlations.insert(value))
                {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidKnowledgeRecord,
                        "producer correlation is invalid or duplicated",
                    ));
                }
                for draft in records {
                    let Some(grant) = contract
                        .knowledge_writes
                        .iter()
                        .find(|grant| grant.schema == draft.schema)
                    else {
                        return Err(CanwuError::new(
                            ErrorCode::UndeclaredKnowledgeWrite,
                            format!(
                                "boundary system {plugin}.{} did not declare the knowledge schema",
                                contract.name
                            ),
                        ));
                    };
                    if !grant.visibilities.contains(visibility) {
                        return Err(CanwuError::new(
                            ErrorCode::UndeclaredKnowledgeWrite,
                            "knowledge publication visibility is not granted",
                        ));
                    }
                    let Some((owner, schema)) = plugins.knowledge_schemas.get(&draft.schema) else {
                        return Err(CanwuError::new(
                            ErrorCode::InvalidKnowledgeSchema,
                            "knowledge publication uses an unregistered schema",
                        ));
                    };
                    if owner != plugin || !schema.writable {
                        return Err(CanwuError::new(
                            ErrorCode::UndeclaredKnowledgeWrite,
                            "knowledge publication uses a foreign or read-only schema",
                        ));
                    }
                    super::knowledge::validate_draft(
                        draft,
                        schema,
                        holder,
                        current,
                        &plugins.record_schemas,
                    )?;
                    for reference in &draft.origin.evidence {
                        validate_proposal_evidence_reference(
                            runtime,
                            boundary_id,
                            pending_evidence,
                            reference,
                        )?;
                    }
                    let existing = current
                        .knowledge
                        .records
                        .get(holder)
                        .into_iter()
                        .flat_map(|records| records.iter())
                        .chain(
                            knowledge_overlay
                                .get(holder)
                                .into_iter()
                                .flat_map(|records| records.iter()),
                        )
                        .collect::<BTreeMap<_, _>>();
                    for related in draft.supersedes.iter().chain(&draft.contradicts) {
                        let Some(related_record) = existing.get(related) else {
                            return Err(CanwuError::new(
                                ErrorCode::KnowledgeRecordNotFound,
                                "knowledge relation does not resolve for the same holder at this cut",
                            ));
                        };
                        if related_record.schema.kind != draft.schema.kind {
                            return Err(CanwuError::new(
                                ErrorCode::InvalidKnowledgeRecord,
                                "knowledge supersession and contradiction cannot cross schema kinds",
                            ));
                        }
                    }
                    let encoded = serde_json::to_vec(&(holder, draft)).map_err(|error| {
                        CanwuError::new(
                            ErrorCode::InvalidKnowledgeRecord,
                            format!("holder-scoped knowledge draft could not be encoded: {error}"),
                        )
                    })?;
                    if !canonical_drafts.insert(encoded) {
                        return Err(CanwuError::new(
                            ErrorCode::InvalidKnowledgeRecord,
                            "one system proposal contains a duplicate canonical knowledge draft",
                        ));
                    }
                }
            }
            BoundaryDirective::ScheduleIngress {
                after,
                packet_type,
                payload,
                affected,
                ..
            } => {
                let descriptor = plugins
                    .ingress
                    .get(&(plugin.to_owned(), packet_type.clone()))
                    .ok_or_else(|| {
                        CanwuError::new(
                            ErrorCode::InvalidPayload,
                            format!(
                                "boundary system {plugin}.{} scheduled undeclared ingress type {packet_type}",
                                contract.name
                            ),
                        )
                    })?;
                if after.is_negative() || now.checked_add(*after).is_none() {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidDuration,
                        "boundary-generated ingress requires a nonnegative supported delay",
                    ));
                }
                descriptor.payload_schema.validate(payload)?;
                if affected.iter().any(|entity| {
                    !proposal_entity_identity_exists(
                        current,
                        &plugins.record_schemas,
                        proposal,
                        entity,
                    )
                }) {
                    return Err(CanwuError::new(
                        ErrorCode::EntityNotFound,
                        format!(
                            "boundary system {plugin}.{} scheduled ingress for an unknown entity identity",
                            contract.name
                        ),
                    ));
                }
            }
            BoundaryDirective::SchedulePluginIngress {
                target_plugin,
                after,
                packet_type,
                payload,
                affected,
                ..
            } => {
                let grant = super::PluginIngressTarget {
                    target_plugin: target_plugin.clone(),
                    packet_type: packet_type.clone(),
                };
                if !contract.plugin_ingress_targets.contains(&grant) {
                    return Err(CanwuError::new(
                        ErrorCode::UndeclaredStateWrite,
                        format!(
                            "boundary system {plugin}.{} did not declare target ingress {target_plugin}.{packet_type}",
                            contract.name
                        ),
                    ));
                }
                let descriptor = plugins
                    .ingress
                    .get(&(target_plugin.clone(), packet_type.clone()))
                    .ok_or_else(|| {
                        CanwuError::new(
                            ErrorCode::InvalidPayload,
                            format!(
                                "boundary system {plugin}.{} scheduled undeclared target ingress {target_plugin}.{packet_type}",
                                contract.name
                            ),
                        )
                    })?;
                if after.is_negative() || now.checked_add(*after).is_none() {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidDuration,
                        "boundary-generated cross-plugin ingress requires a nonnegative supported delay",
                    ));
                }
                descriptor.payload_schema.validate(payload)?;
                if affected.iter().any(|entity| {
                    !proposal_entity_identity_exists(
                        current,
                        &plugins.record_schemas,
                        proposal,
                        entity,
                    )
                }) {
                    return Err(CanwuError::new(
                        ErrorCode::EntityNotFound,
                        format!(
                            "boundary system {plugin}.{} scheduled cross-plugin ingress for an unknown entity identity",
                            contract.name
                        ),
                    ));
                }
            }
            BoundaryDirective::CancelPluginIngress { ingress_id, reason } => {
                if !valid_ingress_cancellation_reason(reason)
                    || !cancelled_ingress.insert(*ingress_id)
                {
                    return Err(CanwuError::new(
                        ErrorCode::InvalidPayload,
                        format!(
                            "boundary system {plugin}.{} proposed a duplicate or malformed ingress cancellation",
                            contract.name
                        ),
                    ));
                }
            }
            BoundaryDirective::ResolveDecisionRandomly { resolution } => {
                validate_random_decision_resolution(
                    plugin,
                    contract,
                    current,
                    committed_availability,
                    pending_random_draws,
                    &mut random_decision_samples,
                    resolution,
                )?;
            }
            BoundaryDirective::SetPersonAvailability {
                person,
                availability,
                summary,
            } => {
                super::persons::validate_availability_directive(
                    plugin,
                    contract,
                    now,
                    *person,
                    availability,
                    summary,
                    &entity_exists,
                )?;
            }
            BoundaryDirective::RecordEvaluationTrace { trace } => {
                super::evaluation::validate_trace_shape(
                    contract.phase,
                    trace,
                    boundary_id,
                    runtime.metadata.run_configuration.evaluation_limits(),
                )?;
                if !proposal_entity_identity_exists(
                    current,
                    &plugins.record_schemas,
                    proposal,
                    &trace.subject,
                ) {
                    return Err(CanwuError::new(
                        ErrorCode::EntityNotFound,
                        format!(
                            "boundary system {plugin}.{} traced an evaluation of unknown subject {}",
                            contract.name, trace.subject
                        ),
                    )
                    .with_entity(trace.subject.clone()));
                }
                for reference in trace.terms.iter().flat_map(|term| &term.evidence) {
                    validate_proposal_evidence_reference(
                        runtime,
                        boundary_id,
                        pending_evidence,
                        reference,
                    )?;
                }
            }
            BoundaryDirective::CreatePerson {
                draft,
                correlation,
                summary,
            } => {
                super::persons::validate_person_draft(
                    &super::persons::PersonDraftContext {
                        plugin,
                        contract,
                        now,
                        government_exists: &|id| current.governments.contains_key(&id),
                        territory_exists: &|id| current.territories.contains_key(&id),
                        entity_exists: &entity_exists,
                    },
                    draft,
                    correlation,
                    summary,
                )?;
                validate_proposal_evidence_reference(
                    runtime,
                    boundary_id,
                    pending_evidence,
                    &draft.provenance,
                )?;
            }
            BoundaryDirective::RegisterTransitionManifest { .. }
            | BoundaryDirective::StageTransitionWrite { .. } => {
                return Err(transition_directive_not_admitted(plugin, &contract.name));
            }
        }
    }
    Ok(())
}

/// Transition directives are resolved by the boundary transition ledger
/// before proposal validation, so reaching one later is a kernel fault.
fn transition_directive_not_admitted(plugin: &str, system: &str) -> CanwuError {
    CanwuError::new(
        ErrorCode::InvalidBoundary,
        format!(
            "boundary system {plugin}.{system} produced a transition directive outside the transition ledger"
        ),
    )
}

/// Validates one `ResolveDecisionRandomly` directive before its draw is
/// committed.
///
/// Availability is read from `committed_availability`, the state committed
/// before this boundary began: a resolution whose ticket has an unavailable
/// person decision maker (`DecisionMakerUnavailable`) or an assigned
/// controller with an unavailable authority person (`IssuerUnavailable`)
/// fails the boundary. The sweep and `Open` admission normally keep such
/// tickets closed, so this is a safeguard. Changes made in the same boundary
/// are deliberately not consulted; failing here would roll back that change
/// with the rest of the boundary and repeat on every retry. Such a draw is
/// committed and the end-of-boundary sweep then cancels the ticket, so the
/// generated resolution is rejected at its admission.
fn validate_random_decision_resolution(
    plugin: &str,
    contract: &BoundarySystemContract,
    current: &RuntimeCurrentState,
    committed_availability: &BTreeMap<super::PersonId, super::PersonAvailability>,
    pending_random_draws: &[random::PendingRandomDraw],
    used_samples: &mut BTreeSet<(super::RandomStreamKey, RandomDrawAddress)>,
    resolution: &super::RandomDecisionResolution,
) -> Result<(), CanwuError> {
    if resolution.decision_request_id.get() == 0
        || resolution
            .command_request_id
            .is_some_and(|request_id| request_id.get() == 0)
        || resolution.expected_version == 0
    {
        return Err(CanwuError::new(
            ErrorCode::InvalidDecision,
            "random decision resolution requires nonzero request IDs and ticket version",
        ));
    }
    let ticket = current
        .decisions
        .ticket(resolution.ticket_id)
        .ok_or_else(|| {
            CanwuError::new(
                ErrorCode::InvalidDecision,
                "random decision resolution references an unknown ticket",
            )
        })?;
    if !ticket.is_open()
        || ticket.version != resolution.expected_version
        || ticket.assigned_controller != resolution.controller_id
    {
        return Err(CanwuError::new(
            ErrorCode::InvalidDecision,
            "random decision resolution references a closed, stale, or differently controlled ticket",
        ));
    }
    let controller = current
        .decisions
        .controller(&resolution.controller_id)
        .ok_or_else(|| {
            CanwuError::new(
                ErrorCode::InvalidDecision,
                "random decision resolution references an unknown controller",
            )
        })?;
    super::persons::validate_decision_preparation(committed_availability, ticket, controller)?;
    match &resolution.tie_break {
        None if controller.policy.kind != DecisionPolicyKind::Random => {
            return Err(CanwuError::new(
                ErrorCode::InvalidDecision,
                "random decision resolution requires a controller with random policy identity",
            ));
        }
        None => {}
        Some(pending) => validate_random_tie_break(controller, ticket, resolution, pending)?,
    }
    if !contract.random_streams.contains(&resolution.sample.stream) {
        return Err(CanwuError::new(
            ErrorCode::UndeclaredRandomStream,
            format!(
                "boundary system {plugin}.{} did not declare the random decision stream",
                contract.name
            ),
        ));
    }
    let RandomDrawAddress::OperationV1(address) = &resolution.sample.address else {
        return Err(CanwuError::new(
            ErrorCode::InvalidRandomDraw,
            "random decisions require an operation-keyed draw",
        ));
    };
    if address.producer_plugin != plugin
        || address.target
            != (RandomOperationTarget::DecisionTicket {
                ticket_id: ticket.id,
                ticket_version: ticket.version,
            })
    {
        return Err(CanwuError::new(
            ErrorCode::InvalidRandomDraw,
            "random decision draw address does not bind the current ticket version",
        ));
    }
    let sample_key = (
        resolution.sample.stream.clone(),
        resolution.sample.address.clone(),
    );
    if !used_samples.insert(sample_key.clone()) {
        return Err(CanwuError::new(
            ErrorCode::InvalidRandomDraw,
            "one random draw cannot resolve more than one decision",
        ));
    }
    if !pending_random_draws.iter().any(|draw| {
        draw.stream == sample_key.0
            && draw.address == sample_key.1
            && draw.upper_exclusive == resolution.sample.upper_exclusive
            && draw.value == resolution.sample.value
    }) {
        return Err(CanwuError::new(
            ErrorCode::InvalidRandomDraw,
            "random decision resolution does not reference a draw produced by this proposal",
        ));
    }
    let total_weight = resolution
        .option_weights
        .iter()
        .try_fold(0_u64, |total, option| total.checked_add(option.weight))
        .ok_or_else(|| {
            CanwuError::new(
                ErrorCode::InvalidDecision,
                "random decision option weights overflow the supported range",
            )
        })?;
    if total_weight != resolution.sample.upper_exclusive {
        return Err(CanwuError::new(
            ErrorCode::InvalidDecision,
            "random decision option weights disagree with the draw bound",
        ));
    }
    let selected = random_resolution_selection(ticket, resolution)?;
    let action = &ticket
        .option(&selected)
        .expect("validated random decision selected an existing option")
        .action;
    if matches!(action, DecisionAction::Command { .. }) != resolution.command_request_id.is_some() {
        return Err(CanwuError::new(
            ErrorCode::InvalidDecision,
            "random decision command options require exactly one command request ID",
        ));
    }
    Ok(())
}

/// Validates the pending utility-policy decision a random tie-break resolves:
/// the draw covers exactly its near-equivalent candidates, and the generated
/// resolution can carry its evaluations and fired guards unchanged.
fn validate_random_tie_break(
    controller: &super::DecisionControllerBinding,
    ticket: &super::DecisionTicket,
    resolution: &super::RandomDecisionResolution,
    pending: &PolicyDecision,
) -> Result<(), CanwuError> {
    if controller.policy.kind != DecisionPolicyKind::Utility || !controller.random_tie_break {
        return Err(CanwuError::new(
            ErrorCode::InvalidDecision,
            "a random tie-break requires a utility-policy controller that permits tie-breaks",
        ));
    }
    let DecisionOutcome::PendingRandom { candidates } = &pending.outcome else {
        return Err(CanwuError::new(
            ErrorCode::InvalidDecision,
            "a random tie-break must carry a pending random policy decision",
        ));
    };
    if candidates != &resolution.option_weights
        || !pending.is_random_tie_break()
        || pending.external.is_some()
        || pending.random.is_some()
    {
        return Err(CanwuError::new(
            ErrorCode::InvalidDecision,
            "random tie-break weights must equal the pending candidates of an evidence-free random stage",
        ));
    }
    pending
        .validate(ticket)
        .map_err(|error| CanwuError::new(ErrorCode::InvalidDecision, error.to_string()))
}

/// Selects the option a random decision resolution draws: every available
/// option for a random-policy controller, only the pending candidates for a
/// utility-policy tie-break.
pub(super) fn random_resolution_selection(
    ticket: &super::DecisionTicket,
    resolution: &super::RandomDecisionResolution,
) -> Result<String, CanwuError> {
    if resolution.tie_break.is_some() {
        DecisionRandomEvidence::selected_candidate(
            ticket,
            &resolution.option_weights,
            resolution.sample.value,
        )
    } else {
        DecisionRandomEvidence::selected_option(
            ticket,
            &resolution.option_weights,
            resolution.sample.value,
        )
    }
    .map_err(|error| CanwuError::new(ErrorCode::InvalidDecision, error.to_string()))
}

pub(super) fn random_policy_summary(option_id: &str) -> String {
    format!("random policy selected {option_id}")
}

pub(super) fn random_tie_break_summary(option_id: &str) -> String {
    format!("random tie-break selected {option_id}")
}

fn validate_proposal_evidence_reference(
    runtime: &RuntimeState,
    boundary_id: BoundaryId,
    pending: &PendingBoundaryEvidence,
    reference: &EvidenceRef,
) -> Result<(), CanwuError> {
    if let EvidenceRef::DomainRecordVersion(version) = reference
        && let DomainRecordVersionSource::BoundaryChange {
            boundary,
            change_index,
        } = version.established_by
        && boundary == boundary_id
    {
        let resolved = usize::try_from(change_index)
            .ok()
            .and_then(|index| pending.record_changes.get(index))
            .is_some_and(|change| {
                change.current.reference == version.record
                    && change.current.version == version.version
            });
        return if resolved {
            Ok(())
        } else {
            Err(CanwuError::new(
                ErrorCode::EvidenceUnavailable,
                "knowledge origin references an unavailable current-boundary record version",
            ))
        };
    }

    if let EvidenceRef::Event(id) = reference
        && runtime
            .evidence
            .retained_event(*id)
            .is_some_and(|event| event.cause == Some(CauseRef::Boundary(boundary_id)))
    {
        if pending
            .emissions
            .iter()
            .any(|emission| emission.event == *id)
        {
            return Ok(());
        }
        return Err(CanwuError::new(
            ErrorCode::EvidenceUnavailable,
            "knowledge origin references an event outside the proposal-visible boundary cut",
        ));
    }

    if let EvidenceRef::Ingress(id) = reference
        && runtime
            .evidence
            .retained_ingress(*id)
            .is_some_and(|record| record.cause == Some(CauseRef::Boundary(boundary_id)))
    {
        return Err(CanwuError::new(
            ErrorCode::EvidenceUnavailable,
            "current-boundary generated ingress is not proposal-visible evidence",
        ));
    }

    match resolve_evidence_reference(&RuntimeValidationContext::new(runtime), reference) {
        EvidenceAvailability::Retained | EvidenceAvailability::Archived => Ok(()),
        EvidenceAvailability::Missing => Err(CanwuError::new(
            ErrorCode::EvidenceUnavailable,
            "knowledge origin references missing or wrong-version evidence",
        )),
    }
}

fn validate_reservation_pool(
    pool: &ReservationPoolKey,
    entity_exists: &dyn Fn(&EntityRef) -> bool,
) -> Result<(), CanwuError> {
    if pool.resource.trim().is_empty()
        || pool.resource != pool.resource.trim()
        || !entity_exists(&pool.entity)
    {
        return Err(CanwuError::new(
            ErrorCode::InvalidBoundary,
            "reservation pools require a canonical resource and an existing entity",
        ));
    }
    Ok(())
}

fn extend_boundary_overlay(
    current: &RuntimeCurrentState,
    record_overlay: &BTreeMap<DomainRecordRef, DomainRecord>,
    overlay: &mut BTreeMap<PluginComponentKey, PluginComponentRecord>,
    directives: &[StagedBoundaryDirective],
) -> Result<(), CanwuError> {
    extend_boundary_component_overlay(current, record_overlay, overlay, directives, false)
}

fn extend_boundary_candidate_overlay(
    current: &RuntimeCurrentState,
    record_overlay: &BTreeMap<DomainRecordRef, DomainRecord>,
    overlay: &mut BTreeMap<PluginComponentKey, PluginComponentRecord>,
    directives: &[StagedBoundaryDirective],
) -> Result<(), CanwuError> {
    extend_boundary_component_overlay(current, record_overlay, overlay, directives, true)
}

fn extend_boundary_component_overlay(
    current: &RuntimeCurrentState,
    record_overlay: &BTreeMap<DomainRecordRef, DomainRecord>,
    overlay: &mut BTreeMap<PluginComponentKey, PluginComponentRecord>,
    directives: &[StagedBoundaryDirective],
    include_next_boundary: bool,
) -> Result<(), CanwuError> {
    for staged in directives.iter().filter(|staged| {
        include_next_boundary || staged.visibility == StateVisibility::SameBoundary
    }) {
        if let BoundaryDirective::SetComponent {
            state: state_key,
            entity,
            component,
            value,
            ..
        } = &staged.directive
        {
            let key = component_key(&staged.plugin, state_key, entity, component);
            if overlay.contains_key(&key) {
                return Err(CanwuError::new(
                    ErrorCode::InvalidBoundary,
                    "multiple boundary proposals target the same component",
                ));
            }
            if !runtime_entity_exists_with_record_overlay(current, record_overlay, entity) {
                return Err(CanwuError::new(
                    ErrorCode::EntityNotFound,
                    format!("boundary proposal targeted missing entity {entity}"),
                ));
            }
            overlay.insert(
                key,
                PluginComponentRecord {
                    plugin: staged.plugin.clone(),
                    state: state_key.clone(),
                    entity: entity.clone(),
                    component: component.clone(),
                    value: value.clone(),
                },
            );
        }
    }
    Ok(())
}

fn extend_boundary_record_overlay(
    context: &BoundaryRecordOverlayContext<'_>,
    overlay: &mut BTreeMap<DomainRecordRef, DomainRecord>,
    directives: &[StagedBoundaryDirective],
) -> Result<(), CanwuError> {
    extend_boundary_domain_record_overlay(context, overlay, directives, false)
}

fn extend_boundary_record_candidate_overlay(
    context: &BoundaryRecordOverlayContext<'_>,
    overlay: &mut BTreeMap<DomainRecordRef, DomainRecord>,
    directives: &[StagedBoundaryDirective],
) -> Result<(), CanwuError> {
    extend_boundary_domain_record_overlay(context, overlay, directives, true)
}

struct BoundaryRecordOverlayContext<'a> {
    current: &'a RuntimeCurrentState,
    now: SimTime,
    scheduled_actions: &'a BTreeMap<ScheduleKey, ScheduledAction>,
    run_configuration: &'a RunConfigurationSnapshot,
    schemas: &'a records::DomainRecordSchemas,
}

fn extend_boundary_domain_record_overlay(
    context: &BoundaryRecordOverlayContext<'_>,
    overlay: &mut BTreeMap<DomainRecordRef, DomainRecord>,
    directives: &[StagedBoundaryDirective],
    include_next_boundary: bool,
) -> Result<(), CanwuError> {
    let requests: Vec<_> = directives
        .iter()
        .filter(|staged| {
            include_next_boundary || staged.visibility == StateVisibility::SameBoundary
        })
        .filter_map(|staged| match &staged.directive {
            BoundaryDirective::MutateRecord { mutation, summary } => {
                Some(records::DomainMutationRequest {
                    plugin: &staged.plugin,
                    system: &staged.system,
                    visibility: staged.visibility,
                    mutation,
                    summary,
                })
            }
            BoundaryDirective::SetComponent { .. }
            | BoundaryDirective::Emit { .. }
            | BoundaryDirective::ScheduleIngress { .. }
            | BoundaryDirective::SchedulePluginIngress { .. }
            | BoundaryDirective::ResolveDecisionRandomly { .. }
            | BoundaryDirective::PublishKnowledge { .. }
            | BoundaryDirective::SetPersonAvailability { .. }
            | BoundaryDirective::CreatePerson { .. }
            | BoundaryDirective::CancelPluginIngress { .. }
            | BoundaryDirective::RecordEvaluationTrace { .. }
            | BoundaryDirective::RegisterTransitionManifest { .. }
            | BoundaryDirective::StageTransitionWrite { .. } => None,
        })
        .collect();
    if requests.is_empty() {
        return Ok(());
    }
    let (next, changes) = records::apply_mutation_bundle_cow_with_overlay(
        &context.current.domain_records,
        overlay,
        context.schemas,
        context.now,
        &|entity| runtime_current_entity_exists(context.current, entity),
        requests,
    )?;
    validate_domain_dependents_with_records(
        &context.current.plugin_components,
        context.scheduled_actions,
        context.run_configuration,
        &next,
    )?;
    for change in changes {
        overlay.insert(change.current.reference.clone(), change.current);
    }
    Ok(())
}

fn partition_boundary_visibility(
    directives: Vec<StagedBoundaryDirective>,
) -> (Vec<StagedBoundaryDirective>, Vec<StagedBoundaryDirective>) {
    directives
        .into_iter()
        .partition(|staged| staged.visibility == StateVisibility::SameBoundary)
}

fn partition_knowledge_directives(
    directives: Vec<StagedBoundaryDirective>,
) -> (Vec<StagedBoundaryDirective>, Vec<StagedBoundaryDirective>) {
    directives
        .into_iter()
        .partition(|staged| matches!(staged.directive, BoundaryDirective::PublishKnowledge { .. }))
}

fn allocate_reservations(
    mut offers: Vec<PendingReservationOffer>,
    mut requests: Vec<PendingReservationRequest>,
) -> Result<ReservationAllocationResult, CanwuError> {
    offers.sort_by(|left, right| {
        left.offer
            .pool
            .cmp(&right.offer.pool)
            .then_with(|| left.plugin.cmp(&right.plugin))
            .then_with(|| left.system.cmp(&right.system))
    });
    let mut remaining = BTreeMap::new();
    let mut offer_records = Vec::new();
    for pending in offers {
        if remaining
            .insert(pending.offer.pool.clone(), pending.offer.capacity)
            .is_some()
        {
            return Err(CanwuError::new(
                ErrorCode::InvalidBoundary,
                format!(
                    "reservation pool was offered more than once, including by {}.{}",
                    pending.plugin, pending.system
                ),
            ));
        }
        offer_records.push(ReservationOfferRecord {
            plugin: pending.plugin,
            system: pending.system,
            offer: pending.offer,
        });
    }
    requests.sort_by(|left, right| {
        left.request
            .pool
            .cmp(&right.request.pool)
            .then_with(|| right.request.priority.cmp(&left.request.priority))
            .then_with(|| left.request.tie_break.cmp(&right.request.tie_break))
            .then_with(|| left.reservation.cmp(&right.reservation))
    });
    let mut seen = BTreeSet::new();
    let mut by_reservation = BTreeMap::new();
    let mut request_records = Vec::new();
    let mut records = Vec::new();
    for pending in requests {
        if !seen.insert(pending.reservation.clone()) {
            return Err(CanwuError::new(
                ErrorCode::InvalidBoundary,
                "reservation request identity is duplicated",
            ));
        }
        request_records.push(ReservationRequestRecord {
            reservation: pending.reservation.clone(),
            request: pending.request.clone(),
        });
        let available = remaining.entry(pending.request.pool.clone()).or_default();
        let granted = pending.request.quantity.min(*available);
        *available -= granted;
        let disposition = if granted == pending.request.quantity {
            ReservationDisposition::Fulfilled
        } else if granted == 0 {
            ReservationDisposition::Rejected
        } else {
            ReservationDisposition::Partial
        };
        let allocation = ReservationAllocation {
            reservation: pending.reservation.clone(),
            pool: pending.request.pool,
            requested: pending.request.quantity,
            granted,
            remaining_after: *available,
            disposition,
        };
        by_reservation.insert(pending.reservation, allocation.clone());
        records.push(allocation);
    }
    Ok(ReservationAllocationResult {
        by_reservation,
        offers: offer_records,
        requests: request_records,
        records,
    })
}