deepstrike_core/runtime/kernel/

runtime.rs

use super::*;
use std::collections::{HashMap, HashSet, VecDeque};
use std::ops::{Deref, DerefMut};

pub(super) const EVENT_REPLAY_WINDOW_CAPACITY: usize = 256;
const COMPLETED_EFFECT_REPLAY_WINDOW_CAPACITY: usize = 256;
const SNAPSHOT_INPUT_LIMIT: usize = 10_000;
const MAX_INPUT_BYTES: usize = 16 * 1024 * 1024;
const SNAPSHOT_JOURNAL_BYTES_LIMIT: usize = 64 * 1024 * 1024;

#[derive(Clone)]
struct RecordedTransition {
    fingerprint: Vec<u8>,
    step: KernelStep,
}

#[derive(serde::Serialize)]
struct KernelSnapshotRefV2<'a> {
    snapshot_version: u32,
    abi_version: u32,
    initial_policy: KernelSnapshotPolicy,
    lifecycle: KernelLifecycle,
    #[serde(skip_serializing_if = "Option::is_none")]
    operation_id: Option<&'a str>,
    next_step_seq: u64,
    snapshot_input_limit: usize,
    max_input_bytes: usize,
    snapshot_journal_bytes_limit: usize,
    accepted_input_bytes: usize,
    accepted_inputs: &'a [KernelInput],
    #[serde(skip_serializing_if = "Option::is_none")]
    last_step: Option<&'a KernelStep>,
}

#[derive(Clone)]
struct AcceptedCancellation {
    reason: CancellationReason,
    pending_call_ids: Vec<String>,
    step: KernelStep,
}

struct ReplayWindow {
    entries: HashMap<String, RecordedTransition>,
    order: VecDeque<String>,
    capacity: usize,
}

impl ReplayWindow {
    fn new(capacity: usize) -> Self {
        Self {
            entries: HashMap::with_capacity(capacity),
            order: VecDeque::with_capacity(capacity),
            capacity,
        }
    }

    fn get(&self, identity: &str) -> Option<&RecordedTransition> {
        self.entries.get(identity)
    }

    fn insert(&mut self, identity: String, transition: RecordedTransition) {
        if self.entries.len() == self.capacity {
            if let Some(expired_identity) = self.order.pop_front() {
                self.entries.remove(&expired_identity);
            }
        }
        self.order.push_back(identity.clone());
        self.entries.insert(identity, transition);
    }

    fn set_capacity(&mut self, capacity: usize) {
        self.capacity = capacity;
        while self.entries.len() > capacity {
            if let Some(expired_identity) = self.order.pop_front() {
                self.entries.remove(&expired_identity);
            }
        }
        self.entries.shrink_to(capacity);
        self.order.shrink_to(capacity);
    }

    #[cfg(test)]
    fn len(&self) -> usize {
        self.entries.len()
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PendingEffectKind {
    Provider,
    Tool,
    Milestone,
    Approval,
    WorkflowSpawn,
    Preempt,
    MemoryPersist,
    MemoryQuery,
    LargeResultSpool,
    PageOutArchive,
}

#[derive(Clone, Copy)]
enum LifecycleTransition {
    Stay,
    Configure,
    Start,
    Resume,
}

#[derive(Clone, Copy, PartialEq, Eq)]
enum StepMode {
    Live,
    SnapshotReplay,
}

struct StepIdentity {
    operation_id: String,
    input_event_id: String,
    step_seq: u64,
}

impl StepIdentity {
    fn empty(&self, observations: Vec<KernelObservation>) -> KernelStep {
        KernelStep::empty(
            self.operation_id.clone(),
            self.input_event_id.clone(),
            self.step_seq,
            observations,
        )
    }

    fn single(self, action: LoopAction, observations: Vec<KernelObservation>) -> KernelStep {
        KernelStep::single(
            self.operation_id,
            self.input_event_id,
            self.step_seq,
            action,
            observations,
        )
    }
}

#[doc(hidden)]
pub struct KernelRuntimeState {
    sm: LoopStateMachine,
    initial_policy: SchedulerBudget,
    lifecycle: KernelLifecycle,
    operation_id: Option<String>,
    next_step_seq: u64,
    recorded_events: ReplayWindow,
    pending_effects: HashMap<String, PendingEffectKind>,
    completed_effects: ReplayWindow,
    memory_records: crate::mm::memory::MemoryRecordStore,
    pending_memory_write: Option<crate::mm::memory::MemoryRecord>,
    pending_memory_store: Option<crate::mm::memory::MemoryRecordStore>,
    pending_memory_query: Option<(crate::mm::memory::MemoryQuery, usize)>,
    budget_usage_reported: bool,
    accepted_cancellation: Option<AcceptedCancellation>,
    accepted_inputs: Vec<KernelInput>,
    accepted_input_count: usize,
    accepted_input_bytes: usize,
    snapshot_input_limit: usize,
    max_input_bytes: usize,
    snapshot_journal_bytes_limit: usize,
    snapshot_overflowed: bool,
    last_step: Option<KernelStep>,
}

struct PreparedCandidate {
    token: String,
    base_generation: u64,
    state: Box<KernelRuntimeState>,
    step: KernelStep,
    accepted_inputs_before: usize,
}

/// Pure kernel runtime wrapper. SDKs should migrate toward feeding
/// `KernelInput` values here instead of directly driving `LoopStateMachine`.
pub struct KernelRuntime {
    state: Option<Box<KernelRuntimeState>>,
    prepared: Option<PreparedCandidate>,
    generation: u64,
    poisoned: bool,
    next_prepare_token: u64,
}

impl Deref for KernelRuntime {
    type Target = KernelRuntimeState;

    fn deref(&self) -> &Self::Target {
        self.state
            .as_deref()
            .expect("kernel runtime state is staged; commit or abort the prepared transition")
    }
}

impl DerefMut for KernelRuntime {
    fn deref_mut(&mut self) -> &mut Self::Target {
        self.state
            .as_deref_mut()
            .expect("kernel runtime state is staged; commit or abort the prepared transition")
    }
}

impl KernelRuntime {
    pub fn new(policy: SchedulerBudget) -> Self {
        Self {
            state: Some(Box::new(KernelRuntimeState {
                sm: LoopStateMachine::new(policy.clone()),
                initial_policy: policy,
                lifecycle: KernelLifecycle::Created,
                operation_id: None,
                next_step_seq: 1,
                recorded_events: ReplayWindow::new(EVENT_REPLAY_WINDOW_CAPACITY),
                pending_effects: HashMap::new(),
                completed_effects: ReplayWindow::new(COMPLETED_EFFECT_REPLAY_WINDOW_CAPACITY),
                memory_records: crate::mm::memory::MemoryRecordStore::default(),
                pending_memory_write: None,
                pending_memory_store: None,
                pending_memory_query: None,
                budget_usage_reported: false,
                accepted_cancellation: None,
                accepted_inputs: Vec::new(),
                accepted_input_count: 0,
                accepted_input_bytes: 0,
                snapshot_input_limit: SNAPSHOT_INPUT_LIMIT,
                max_input_bytes: MAX_INPUT_BYTES,
                snapshot_journal_bytes_limit: SNAPSHOT_JOURNAL_BYTES_LIMIT,
                snapshot_overflowed: false,
                last_step: None,
            })),
            prepared: None,
            generation: 0,
            poisoned: false,
            next_prepare_token: 1,
        }
    }

    /// Capture a portable ABI-v2 checkpoint without exposing private scheduler representation.
    pub fn snapshot(&self) -> Result<KernelSnapshot, KernelFault> {
        self.ensure_snapshot_available()?;
        Ok(KernelSnapshot {
            snapshot_version: KERNEL_SNAPSHOT_VERSION,
            abi_version: KERNEL_ABI_VERSION,
            initial_policy: KernelSnapshotPolicy::from(&self.initial_policy),
            lifecycle: self.lifecycle,
            operation_id: self.operation_id.clone(),
            next_step_seq: self.next_step_seq,
            snapshot_input_limit: self.snapshot_input_limit,
            max_input_bytes: self.max_input_bytes,
            snapshot_journal_bytes_limit: self.snapshot_journal_bytes_limit,
            accepted_input_bytes: self.accepted_input_bytes,
            accepted_inputs: self.accepted_inputs.clone(),
            last_step: self.last_step.clone(),
        })
    }

    pub fn snapshot_json(&self) -> Result<String, KernelFault> {
        self.ensure_snapshot_available()?;
        let snapshot = KernelSnapshotRefV2 {
            snapshot_version: KERNEL_SNAPSHOT_VERSION,
            abi_version: KERNEL_ABI_VERSION,
            initial_policy: KernelSnapshotPolicy::from(&self.initial_policy),
            lifecycle: self.lifecycle,
            operation_id: self.operation_id.as_deref(),
            next_step_seq: self.next_step_seq,
            snapshot_input_limit: self.snapshot_input_limit,
            max_input_bytes: self.max_input_bytes,
            snapshot_journal_bytes_limit: self.snapshot_journal_bytes_limit,
            accepted_input_bytes: self.accepted_input_bytes,
            accepted_inputs: &self.accepted_inputs,
            last_step: self.last_step.as_ref(),
        };
        serde_json::to_string(&snapshot).map_err(|error| {
            snapshot_fault(
                self.operation_id.clone(),
                format!("failed to encode kernel snapshot: {error}"),
            )
        })
    }

    /// Restore by deterministically replaying accepted public ABI transactions, then fail closed
    /// if any checkpoint metadata differs from the rebuilt runtime.
    pub fn restore_snapshot(snapshot: KernelSnapshot) -> Result<Self, KernelFault> {
        if snapshot.snapshot_version != KERNEL_SNAPSHOT_VERSION
            || snapshot.abi_version != KERNEL_ABI_VERSION
            || !(1..=100_000).contains(&snapshot.snapshot_input_limit)
            || !(256..=64 * 1024 * 1024).contains(&snapshot.max_input_bytes)
            || !(256..=1024 * 1024 * 1024usize).contains(&snapshot.snapshot_journal_bytes_limit)
            || snapshot.accepted_inputs.len() > snapshot.snapshot_input_limit
            || snapshot.accepted_input_bytes > snapshot.snapshot_journal_bytes_limit
        {
            return Err(snapshot_fault(
                snapshot.operation_id,
                "incompatible kernel snapshot version or bounds".to_string(),
            ));
        }

        let initial_policy = SchedulerBudget::try_from(&snapshot.initial_policy)
            .map_err(|message| snapshot_fault(snapshot.operation_id.clone(), message))?;
        if KernelSnapshotPolicy::from(&initial_policy) != snapshot.initial_policy {
            return Err(snapshot_fault(
                snapshot.operation_id,
                "kernel snapshot policy is not canonically encoded".to_string(),
            ));
        }
        let mut runtime = Self::new(initial_policy);
        runtime.snapshot_input_limit = snapshot.snapshot_input_limit;
        runtime.max_input_bytes = snapshot.max_input_bytes;
        runtime.snapshot_journal_bytes_limit = snapshot.snapshot_journal_bytes_limit;
        let mut event_ids = HashSet::with_capacity(snapshot.accepted_inputs.len());
        for input in snapshot.accepted_inputs.iter().cloned() {
            if !event_ids.insert(input.event_id.clone()) {
                return Err(snapshot_fault(
                    snapshot.operation_id.clone(),
                    format!(
                        "kernel snapshot journal repeats accepted event_id {}",
                        input.event_id
                    ),
                ));
            }
            let step = runtime.step_internal(input, StepMode::SnapshotReplay);
            if let Some(fault) = step.faults.first() {
                return Err(snapshot_fault(
                    snapshot.operation_id.clone(),
                    format!(
                        "snapshot replay rejected an accepted input: {}",
                        fault.message
                    ),
                ));
            }
        }
        let rebuilt_last = serde_json::to_value(&runtime.last_step).ok();
        let expected_last = serde_json::to_value(&snapshot.last_step).ok();
        if runtime.lifecycle != snapshot.lifecycle
            || runtime.operation_id != snapshot.operation_id
            || runtime.next_step_seq != snapshot.next_step_seq
            || runtime.snapshot_input_limit != snapshot.snapshot_input_limit
            || runtime.max_input_bytes != snapshot.max_input_bytes
            || runtime.snapshot_journal_bytes_limit != snapshot.snapshot_journal_bytes_limit
            || runtime.accepted_input_count != snapshot.accepted_inputs.len()
            || runtime.accepted_input_bytes != snapshot.accepted_input_bytes
            || runtime.snapshot_overflowed
            || rebuilt_last != expected_last
        {
            return Err(snapshot_fault(
                snapshot.operation_id,
                "kernel snapshot metadata does not match deterministic replay".to_string(),
            ));
        }
        runtime.accepted_inputs = snapshot.accepted_inputs;
        runtime.generation = runtime.accepted_input_count as u64;
        Ok(runtime)
    }

    pub fn restore_snapshot_json(snapshot_json: &str) -> Result<Self, KernelFault> {
        let snapshot = serde_json::from_str(snapshot_json).map_err(|error| {
            snapshot_fault(None, format!("invalid kernel snapshot JSON: {error}"))
        })?;
        Self::restore_snapshot(snapshot)
    }

    #[cfg(test)]
    pub(super) fn state_machine(&self) -> &LoopStateMachine {
        &self.sm
    }

    #[cfg(test)]
    pub(super) fn clear_test_observations(&mut self) {
        self.sm.take_observations();
    }

    #[cfg(test)]
    pub(super) fn push_test_history(&mut self, message: Message, tokens: u32) {
        self.sm.ctx.push_history(message, tokens);
    }

    #[cfg(test)]
    pub(super) fn accepted_snapshot_input_count(&self) -> usize {
        self.accepted_inputs.len()
    }

    pub fn is_terminal(&self) -> bool {
        self.lifecycle.is_terminal()
    }

    pub fn diagnostics(&self) -> KernelDiagnostics {
        KernelDiagnostics {
            lifecycle: self.lifecycle,
            next_step_seq: self.next_step_seq,
            accepted_input_count: self.accepted_input_count,
            accepted_input_bytes: self.accepted_input_bytes,
            snapshot_input_limit: self.snapshot_input_limit,
            snapshot_journal_bytes_limit: self.snapshot_journal_bytes_limit,
            max_input_bytes: self.max_input_bytes,
            snapshot_overflowed: self.snapshot_overflowed,
            recorded_event_count: self.recorded_events.entries.len(),
            completed_effect_count: self.completed_effects.entries.len(),
            pending_effect_count: self.pending_effects.len(),
        }
    }

    pub fn lifecycle(&self) -> KernelLifecycle {
        self.lifecycle
    }

    pub fn turn(&self) -> u32 {
        self.sm.turn
    }

    pub fn recovery_content_bytes(&self) -> usize {
        let tokens = self
            .sm
            .ctx
            .config
            .recovery_content_tokens(self.sm.ctx.max_tokens);
        self.sm.ctx.engine.token_budget_to_bytes(tokens)
    }

    pub fn render(&self) -> RenderedContext {
        self.sm.ctx.render()
    }

    pub fn drain_new_messages(&mut self) -> Vec<Message> {
        self.sm.drain_new_messages()
    }

    pub fn preserved_refs(&self) -> Vec<String> {
        self.sm.ctx.partitions.task_state.preserved_refs.clone()
    }

    pub fn count_tokens(&self, text: &str) -> u32 {
        self.sm.ctx.engine.count(text)
    }

    /// L1 (RunGroup): this vehicle's cumulative sub-agent spawns this run, read back by the SDK at run
    /// end to charge the group ledger (so the next member's cumulative spawn cap is seeded correctly).
    pub fn local_subagents_spawned(&self) -> u32 {
        self.sm.local_subagents_spawned()
    }

    #[cfg(test)]
    pub(super) fn pending_provider_effect_id(&self) -> String {
        self.pending_effect_id(PendingEffectKind::Provider)
    }

    #[cfg(test)]
    pub(super) fn pending_tool_effect_id(&self) -> String {
        self.pending_effect_id(PendingEffectKind::Tool)
    }

    #[cfg(test)]
    fn pending_effect_id(&self, kind: PendingEffectKind) -> String {
        let matches = self
            .pending_effects
            .iter()
            .filter(|(_, pending_kind)| **pending_kind == kind)
            .collect::<Vec<_>>();
        assert_eq!(
            matches.len(),
            1,
            "test transition must have one {kind:?} effect"
        );
        matches[0].0.clone()
    }

    #[cfg(test)]
    pub(super) fn recorded_event_count(&self) -> usize {
        self.recorded_events.len()
    }

    fn boundary_max_input_bytes(&self) -> usize {
        self.state
            .as_deref()
            .or_else(|| {
                self.prepared
                    .as_ref()
                    .map(|candidate| candidate.state.as_ref())
            })
            .map_or(MAX_INPUT_BYTES, |state| state.max_input_bytes)
    }

    fn boundary_step_seq(&self) -> u64 {
        self.state
            .as_deref()
            .map(|state| state.next_step_seq)
            .or_else(|| {
                self.prepared
                    .as_ref()
                    .map(|candidate| candidate.step.step_seq)
            })
            .unwrap_or(1)
    }

    /// Stage one transition without publishing it as committed runtime state. A newly accepted
    /// transition returns an opaque one-shot token; exact event replay and rejected inputs return no
    /// token because neither requires a new durable transaction.
    pub fn prepare_step(&mut self, mut input: KernelInput) -> KernelPreparedStep {
        normalize_input(&mut input);
        let base_generation = self.generation;
        let operation_id = input.operation_id.clone();
        let event_id = input.event_id.clone();

        if self.poisoned {
            return KernelPreparedStep {
                status: KernelPreparationStatus::Rejected,
                base_generation,
                prepare_token: None,
                input,
                step: transaction_fault_step(
                    operation_id,
                    event_id,
                    self.boundary_step_seq(),
                    KernelFaultCode::TransactionConflict,
                    "kernel runtime was invalidated by a transaction consistency error".to_string(),
                ),
            };
        }

        if let Some(candidate) = &self.prepared {
            return KernelPreparedStep {
                status: KernelPreparationStatus::Rejected,
                base_generation,
                prepare_token: None,
                input,
                step: transaction_fault_step(
                    operation_id,
                    event_id,
                    candidate.step.step_seq,
                    KernelFaultCode::TransactionConflict,
                    "another kernel transition is already prepared".to_string(),
                ),
            };
        }

        if self.snapshot_overflowed {
            return KernelPreparedStep {
                status: KernelPreparationStatus::Rejected,
                base_generation,
                prepare_token: None,
                input,
                step: transaction_fault_step(
                    operation_id,
                    event_id,
                    self.next_step_seq,
                    KernelFaultCode::ResourceLimitExceeded,
                    "kernel journal cannot roll back a prepared transition after overflow"
                        .to_string(),
                ),
            };
        }

        let fingerprint = serde_json::to_vec(&input)
            .expect("KernelInput serialization must succeed after typed construction");
        let was_exact_replay = self
            .recorded_events
            .get(&event_id)
            .is_some_and(|recorded| recorded.fingerprint == fingerprint);
        let accepted_inputs_before = self.accepted_inputs.len();
        let step = self.step_internal(input.clone(), StepMode::Live);
        let candidate_state = self
            .state
            .take()
            .expect("step execution must leave a runtime state");

        if !step.faults.is_empty() {
            self.state = Some(candidate_state);
            return KernelPreparedStep {
                status: KernelPreparationStatus::Rejected,
                base_generation,
                prepare_token: None,
                input,
                step,
            };
        }

        if was_exact_replay {
            self.state = Some(candidate_state);
            return KernelPreparedStep {
                status: KernelPreparationStatus::Replayed,
                base_generation,
                prepare_token: None,
                input,
                step,
            };
        }

        let token = format!("kernel-prepare-{}", self.next_prepare_token);
        self.next_prepare_token = self.next_prepare_token.saturating_add(1);
        self.prepared = Some(PreparedCandidate {
            token: token.clone(),
            base_generation,
            state: candidate_state,
            step: step.clone(),
            accepted_inputs_before,
        });
        KernelPreparedStep {
            status: KernelPreparationStatus::Prepared,
            base_generation,
            prepare_token: Some(token),
            input,
            step,
        }
    }

    /// Publish the candidate selected by `prepare_step`. A token mismatch invalidates this runtime;
    /// the host must discard it and rebuild from the authoritative committed stream.
    pub fn commit_prepared(&mut self, token: &str) -> Result<KernelStep, KernelFault> {
        if self.poisoned {
            return Err(transaction_conflict_fault(
                None,
                "kernel runtime was invalidated by a transaction consistency error".to_string(),
            ));
        }
        let Some(candidate) = self.prepared.as_ref() else {
            return Err(transaction_conflict_fault(
                None,
                "no kernel transition is prepared".to_string(),
            ));
        };
        if candidate.token != token {
            let operation_id = candidate.state.operation_id.clone();
            self.poisoned = true;
            return Err(transaction_conflict_fault(
                operation_id,
                "prepare token does not match the staged transition".to_string(),
            ));
        }
        if candidate.base_generation != self.generation {
            let operation_id = candidate.state.operation_id.clone();
            self.poisoned = true;
            return Err(transaction_conflict_fault(
                operation_id,
                "prepared transition no longer matches the runtime generation".to_string(),
            ));
        }

        let candidate = self
            .prepared
            .take()
            .expect("prepared candidate was validated above");
        let step = candidate.step.clone();
        self.state = Some(candidate.state);
        self.generation = self.generation.saturating_add(1);
        Ok(step)
    }

    /// Abort a staged transition after durable append failure. Rollback is intentionally paid only
    /// on this failure path: the committed prefix is deterministically rebuilt from its journal,
    /// while the normal prepare/commit path moves the candidate state without cloning history.
    pub fn abort_prepared(&mut self, token: &str) -> Result<(), KernelFault> {
        if self.poisoned {
            return Err(transaction_conflict_fault(
                None,
                "kernel runtime was invalidated by a transaction consistency error".to_string(),
            ));
        }
        let Some(candidate) = self.prepared.as_ref() else {
            return Err(transaction_conflict_fault(
                None,
                "no kernel transition is prepared".to_string(),
            ));
        };
        if candidate.token != token {
            let operation_id = candidate.state.operation_id.clone();
            self.poisoned = true;
            return Err(transaction_conflict_fault(
                operation_id,
                "prepare token does not match the staged transition".to_string(),
            ));
        }

        let candidate = self
            .prepared
            .take()
            .expect("prepared candidate was validated above");
        let mut committed_inputs = candidate.state.accepted_inputs.clone();
        committed_inputs.truncate(candidate.accepted_inputs_before);
        match Self::rebuild_accepted_inputs(
            candidate.state.initial_policy.clone(),
            committed_inputs,
        ) {
            Ok(mut restored) => {
                self.state = restored.state.take();
                Ok(())
            }
            Err(fault) => {
                self.prepared = Some(candidate);
                Err(fault)
            }
        }
    }

    fn rebuild_accepted_inputs(
        initial_policy: SchedulerBudget,
        accepted_inputs: Vec<KernelInput>,
    ) -> Result<Self, KernelFault> {
        let mut runtime = Self::new(initial_policy);
        for input in accepted_inputs.iter().cloned() {
            let step = runtime.step_internal(input, StepMode::SnapshotReplay);
            if let Some(fault) = step.faults.first() {
                return Err(snapshot_fault(
                    runtime.operation_id.clone(),
                    format!(
                        "committed transaction replay rejected an accepted input: {}",
                        fault.message
                    ),
                ));
            }
        }
        runtime.accepted_inputs = accepted_inputs;
        runtime.generation = runtime.accepted_input_count as u64;
        Ok(runtime)
    }

    /// Decode and execute one wire input. The version is probed before decoding
    /// the v2 envelope so a v1 payload receives a structured kernel fault rather
    /// than a host-language deserialization error.
    pub fn step_json(&mut self, input_json: &str) -> Result<KernelStep, serde_json::Error> {
        let max_input_bytes = self.boundary_max_input_bytes();
        if input_json.len() > max_input_bytes {
            return Ok(transaction_fault_step(
                String::new(),
                String::new(),
                self.boundary_step_seq(),
                KernelFaultCode::ResourceLimitExceeded,
                format!(
                    "kernel input is {} bytes; configured maximum is {} bytes",
                    input_json.len(),
                    max_input_bytes
                ),
            ));
        }
        let value: serde_json::Value = serde_json::from_str(input_json)?;
        if value.get("version").and_then(serde_json::Value::as_u64)
            != Some(KERNEL_ABI_VERSION as u64)
        {
            let operation_id = value
                .get("operation_id")
                .and_then(serde_json::Value::as_str)
                .unwrap_or_default()
                .to_string();
            let event_id = value
                .get("event_id")
                .and_then(serde_json::Value::as_str)
                .unwrap_or_default()
                .to_string();
            let received_version = value
                .get("version")
                .and_then(serde_json::Value::as_u64)
                .map_or_else(|| "missing".to_string(), |version| version.to_string());
            return Ok(transaction_fault_step(
                operation_id,
                event_id,
                self.boundary_step_seq(),
                KernelFaultCode::VersionMismatch,
                format!(
                    "kernel ABI version mismatch: input v{received_version}, kernel v{KERNEL_ABI_VERSION}"
                ),
            ));
        }

        serde_json::from_value(value).map(|input| self.step(input))
    }

    /// Decode and stage one wire input for a host-controlled durable commit boundary.
    pub fn prepare_step_json(
        &mut self,
        input_json: &str,
    ) -> Result<KernelPreparedStep, serde_json::Error> {
        let value: serde_json::Value = serde_json::from_str(input_json)?;
        let input: KernelInput = serde_json::from_value(value)?;
        let max_input_bytes = self.boundary_max_input_bytes();
        if input_json.len() > max_input_bytes {
            return Ok(KernelPreparedStep {
                status: KernelPreparationStatus::Rejected,
                base_generation: self.generation,
                prepare_token: None,
                step: transaction_fault_step(
                    input.operation_id.clone(),
                    input.event_id.clone(),
                    self.boundary_step_seq(),
                    KernelFaultCode::ResourceLimitExceeded,
                    format!(
                        "kernel input is {} bytes; configured maximum is {} bytes",
                        input_json.len(),
                        max_input_bytes
                    ),
                ),
                input,
            });
        }
        Ok(self.prepare_step(input))
    }

    pub fn step(&mut self, input: KernelInput) -> KernelStep {
        if self.poisoned {
            return transaction_fault_step(
                input.operation_id,
                input.event_id,
                self.boundary_step_seq(),
                KernelFaultCode::TransactionConflict,
                "kernel runtime was invalidated by a transaction consistency error".to_string(),
            );
        }
        if let Some(candidate) = &self.prepared {
            return transaction_fault_step(
                input.operation_id,
                input.event_id,
                candidate.step.step_seq,
                KernelFaultCode::TransactionConflict,
                "another kernel transition is already prepared".to_string(),
            );
        }
        let accepted_input_count = self.accepted_input_count;
        let step = self.step_internal(input, StepMode::Live);
        if self.accepted_input_count > accepted_input_count {
            self.generation = self.generation.saturating_add(1);
        }
        step
    }

    fn step_internal(&mut self, mut input: KernelInput, mode: StepMode) -> KernelStep {
        normalize_input(&mut input);
        let operation_id = input.operation_id.clone();
        let event_id = input.event_id.clone();

        if input.version != KERNEL_ABI_VERSION {
            return self.fault_step(
                operation_id,
                event_id,
                KernelFaultCode::VersionMismatch,
                format!(
                    "kernel ABI version mismatch: input v{}, kernel v{}",
                    input.version, KERNEL_ABI_VERSION
                ),
                None,
            );
        }

        if operation_id.is_empty() || event_id.is_empty() {
            return self.fault_step(
                operation_id,
                event_id,
                KernelFaultCode::InvalidConfig,
                "operation_id and event_id must be non-empty".to_string(),
                None,
            );
        }

        let fingerprint = serde_json::to_vec(&input)
            .expect("KernelInput serialization must succeed after typed construction");
        let input_bytes = fingerprint.len();
        if mode == StepMode::Live && input_bytes > self.max_input_bytes {
            return self.fault_step(
                operation_id,
                event_id,
                KernelFaultCode::ResourceLimitExceeded,
                format!(
                    "kernel input is {input_bytes} bytes; configured maximum is {} bytes",
                    self.max_input_bytes
                ),
                None,
            );
        }
        if let Some(recorded) = self.recorded_events.get(&event_id) {
            if recorded.fingerprint == fingerprint {
                return recorded.step.clone();
            }
            return self.fault_step(
                operation_id,
                event_id,
                KernelFaultCode::DuplicateEventConflict,
                "event_id was already accepted with a different payload".to_string(),
                None,
            );
        }

        if let Some(bound_operation_id) = &self.operation_id {
            if bound_operation_id != &operation_id {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::OperationMismatch,
                    format!("input operation does not match bound operation {bound_operation_id}"),
                    None,
                );
            }
        }

        if let KernelInputEvent::CancelOperation {
            operation_id: cancelled_operation_id,
            reason,
            pending_call_ids,
        } = &input.event
        {
            if cancelled_operation_id != &operation_id {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::OperationMismatch,
                    "cancel_operation identity does not match the input envelope".to_string(),
                    None,
                );
            }
            if cancelled_operation_id.is_empty()
                || pending_call_ids.iter().any(|call_id| call_id.is_empty())
            {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidConfig,
                    "cancel_operation requires non-empty operation and pending call identities"
                        .to_string(),
                    None,
                );
            }
            if let Some(accepted) = &self.accepted_cancellation {
                if accepted.reason == *reason && accepted.pending_call_ids == *pending_call_ids {
                    let step = accepted.step.clone();
                    self.recorded_events.insert(
                        event_id,
                        RecordedTransition {
                            fingerprint,
                            step: step.clone(),
                        },
                    );
                    self.record_accepted_input(
                        (mode == StepMode::Live).then_some(input),
                        input_bytes,
                    );
                    self.last_step = Some(step.clone());
                    return step;
                }
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::DuplicateEventConflict,
                    "cancel_operation conflicts with the committed cancellation".to_string(),
                    None,
                );
            }
        }

        let result_fingerprint = result_effect(&input.event).map(|(effect_id, _)| {
            (
                effect_id.to_string(),
                serde_json::to_vec(&input.event)
                    .expect("KernelInputEvent serialization must succeed after typed construction"),
            )
        });
        if let Some((effect_id, result_fingerprint)) = &result_fingerprint {
            if let Some(completed) = self.completed_effects.get(effect_id) {
                if completed.fingerprint == *result_fingerprint {
                    let step = completed.step.clone();
                    self.recorded_events.insert(
                        event_id,
                        RecordedTransition {
                            fingerprint,
                            step: step.clone(),
                        },
                    );
                    self.record_accepted_input(
                        (mode == StepMode::Live).then_some(input),
                        input_bytes,
                    );
                    self.last_step = Some(step.clone());
                    return step;
                }
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::UnexpectedEffectResult,
                    format!("effect result conflicts with the completed result: {effect_id}"),
                    Some(effect_id.clone()),
                );
            }
        }

        let lifecycle_transition = match self.lifecycle_transition(&input.event) {
            Ok(transition) => transition,
            Err(message) => {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidLifecycle,
                    message,
                    None,
                );
            }
        };
        if let KernelInputEvent::ConfigureRun { config } = &input.event {
            if let Err(message) = validate_run_config(config, self.sm.ctx.max_tokens) {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidConfig,
                    message,
                    None,
                );
            }
            if config
                .reliability
                .as_ref()
                .and_then(|reliability| reliability.snapshot_input_limit)
                .is_some_and(|limit| limit <= self.accepted_input_count)
            {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidConfig,
                    "snapshot_input_limit must leave room for the configure transaction"
                        .to_string(),
                    None,
                );
            }
            if config
                .reliability
                .as_ref()
                .and_then(|reliability| reliability.snapshot_journal_bytes_limit)
                .is_some_and(|limit| limit < self.accepted_input_bytes.saturating_add(input_bytes))
            {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidConfig,
                    "snapshot_journal_bytes_limit must leave room for the configure transaction"
                        .to_string(),
                    None,
                );
            }
            if config
                .reliability
                .as_ref()
                .and_then(|reliability| reliability.max_input_bytes)
                .is_some_and(|limit| limit < input_bytes)
            {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidConfig,
                    "max_input_bytes must admit the configure transaction".to_string(),
                    None,
                );
            }
        }
        if let KernelInputEvent::DeliverSignal {
            delivery_id,
            attempt,
            ..
        } = &input.event
        {
            if delivery_id.is_empty() || *attempt == 0 {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidConfig,
                    "deliver_signal requires a non-empty delivery_id and attempt >= 1".to_string(),
                    None,
                );
            }
        }
        if let KernelInputEvent::SetSignalPolicy { policy } = &input.event {
            if let Err(message) = validate_signal_policy(policy) {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidConfig,
                    message,
                    None,
                );
            }
        }
        if let KernelInputEvent::QueryMemory { query } = &input.event {
            if let Err(message) = query.validate() {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::InvalidConfig,
                    message,
                    None,
                );
            }
        }
        if let KernelInputEvent::MemoryQueryResult {
            effect_id,
            hits,
            error,
        } = &input.event
        {
            if error.is_none() {
                let Some((query, requested_k)) = self.pending_memory_query.as_ref() else {
                    return self.fault_step(
                        operation_id,
                        event_id,
                        KernelFaultCode::UnexpectedEffectResult,
                        "memory query result has no pending query".to_string(),
                        Some(effect_id.clone()),
                    );
                };
                if let Err(message) = query.validate_hits(hits, *requested_k) {
                    return self.fault_step(
                        operation_id,
                        event_id,
                        KernelFaultCode::UnexpectedEffectResult,
                        message,
                        Some(effect_id.clone()),
                    );
                }
            }
        }
        if let KernelInputEvent::WorkflowSpawnResult {
            effect_id,
            started_agent_ids,
            failures,
            error,
        } = &input.event
        {
            if let Err(message) = self.sm.validate_workflow_spawn_result(
                started_agent_ids,
                failures,
                error.as_deref(),
            ) {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::UnexpectedEffectResult,
                    message,
                    Some(effect_id.clone()),
                );
            }
        }
        if let KernelInputEvent::LargeResultSpoolResult {
            effect_id,
            spool_ref,
            error,
        } = &input.event
        {
            if error.is_none() && spool_ref.as_deref().map_or(true, str::is_empty) {
                return self.fault_step(
                    operation_id,
                    event_id,
                    KernelFaultCode::UnexpectedEffectResult,
                    "successful spool result requires a non-empty spool_ref".to_string(),
                    Some(effect_id.clone()),
                );
            }
        }

        if let Some((effect_id, expected_kind)) = result_effect(&input.event) {
            match self.pending_effects.get(effect_id) {
                Some(actual_kind) if actual_kind == &expected_kind => {
                    self.pending_effects.remove(effect_id);
                }
                _ => {
                    return self.fault_step(
                        operation_id,
                        event_id,
                        KernelFaultCode::UnexpectedEffectResult,
                        format!("effect result does not match a pending effect: {effect_id}"),
                        Some(effect_id.to_string()),
                    );
                }
            }
        }

        if self.operation_id.is_none() {
            self.operation_id = Some(operation_id.clone());
        }

        if input.observed_at_ms > 0 {
            self.sm.set_observed_time(input.observed_at_ms);
        }

        let accepted_input = (mode == StepMode::Live).then(|| input.clone());
        let step_seq = self.allocate_step_seq();
        let cancellation_identity = match &input.event {
            KernelInputEvent::CancelOperation {
                reason,
                pending_call_ids,
                ..
            } => Some((*reason, pending_call_ids.clone())),
            _ => None,
        };
        let mut step = self.dispatch(
            StepIdentity {
                operation_id: operation_id.clone(),
                input_event_id: event_id.clone(),
                step_seq,
            },
            input.event,
        );
        let reservation_id = self
            .sm
            .budget_grant()
            .map(|grant| grant.reservation_id.clone());
        for observation in &mut step.observations {
            if let KernelObservation::BudgetExceeded {
                operation_id: observed_operation_id,
                reservation_id: observed_reservation_id,
                ..
            } = observation
            {
                *observed_operation_id = operation_id.clone();
                if observed_reservation_id.is_none() {
                    *observed_reservation_id = reservation_id.clone();
                }
            }
        }
        let is_terminal = step
            .actions
            .iter()
            .any(|action| matches!(action.effect, KernelEffect::Done { .. }));
        if is_terminal && !self.budget_usage_reported {
            if let Some(reservation_id) = reservation_id {
                let (tokens, subagents, rounds) = self.sm.local_budget_usage();
                step.observations
                    .push(KernelObservation::BudgetUsageReported {
                        operation_id: operation_id.clone(),
                        reservation_id,
                        tokens,
                        subagents,
                        rounds,
                    });
                self.budget_usage_reported = true;
            }
        }
        self.advance_lifecycle(lifecycle_transition, &step);
        if is_terminal {
            self.pending_effects.clear();
        } else {
            for action in &step.actions {
                if let Some(kind) = pending_effect_kind(&action.effect) {
                    self.pending_effects
                        .retain(|_, pending_kind| *pending_kind != kind);
                    self.pending_effects.insert(action.effect_id.clone(), kind);
                }
            }
        }
        if let Some((effect_id, result_fingerprint)) = result_fingerprint {
            self.completed_effects.insert(
                effect_id,
                RecordedTransition {
                    fingerprint: result_fingerprint,
                    step: step.clone(),
                },
            );
        }
        if let Some((reason, pending_call_ids)) = cancellation_identity {
            self.accepted_cancellation = Some(AcceptedCancellation {
                reason,
                pending_call_ids,
                step: step.clone(),
            });
        }
        self.recorded_events.insert(
            event_id,
            RecordedTransition {
                fingerprint,
                step: step.clone(),
            },
        );
        self.record_accepted_input(accepted_input, input_bytes);
        self.last_step = Some(step.clone());
        step
    }

    fn ensure_snapshot_available(&self) -> Result<(), KernelFault> {
        if self.poisoned {
            return Err(transaction_conflict_fault(
                None,
                "kernel runtime was invalidated by a transaction consistency error".to_string(),
            ));
        }
        if let Some(candidate) = &self.prepared {
            return Err(transaction_conflict_fault(
                candidate.state.operation_id.clone(),
                "cannot snapshot an uncommitted prepared transition".to_string(),
            ));
        }
        if self.snapshot_overflowed
            || self.accepted_input_count > self.snapshot_input_limit
            || self.accepted_input_bytes > self.snapshot_journal_bytes_limit
        {
            return Err(snapshot_fault(
                self.operation_id.clone(),
                format!(
                    "snapshot input journal exceeded configured limits: {} inputs / {} bytes",
                    self.snapshot_input_limit, self.snapshot_journal_bytes_limit
                ),
            ));
        }
        Ok(())
    }

    fn record_accepted_input(&mut self, input: Option<KernelInput>, input_bytes: usize) {
        if self.accepted_input_count < self.snapshot_input_limit
            && self.accepted_input_bytes.saturating_add(input_bytes)
                <= self.snapshot_journal_bytes_limit
        {
            self.accepted_input_count += 1;
            self.accepted_input_bytes += input_bytes;
            if let Some(input) = input {
                self.accepted_inputs.push(input);
            }
        } else {
            self.snapshot_overflowed = true;
        }
    }

    fn dispatch(&mut self, identity: StepIdentity, event: KernelInputEvent) -> KernelStep {
        let action = match event {
            KernelInputEvent::SetTools { tools } => {
                self.sm.tools = tools;
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetAvailableSkills { skills } => {
                self.sm.ctx.set_available_skills(skills);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SkillActivated { name, lease_turns } => {
                // B1: record the activation (B2 reads it in emit_call_llm to narrow tools).
                // The returned `changed` flag is the epoch boundary for D's cache re-anchor.
                // K3: a lease converts to an absolute expiry turn here (the manager is turn-blind).
                let expires_at_turn = lease_turns.map(|n| self.sm.turn.saturating_add(n));
                self.sm.ctx.activate_skill_leased(name, expires_at_turn);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SkillDeactivated { name } => {
                self.sm.ctx.deactivate_skill(&name);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetStableCoreTools { tool_ids } => {
                self.sm
                    .ctx
                    .set_stable_core_tools(tool_ids.into_iter().map(Into::into));
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetMemoryEnabled { enabled } => {
                self.sm.ctx.set_memory_enabled(enabled);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetKnowledgeEnabled { enabled } => {
                self.sm.ctx.set_knowledge_enabled(enabled);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetPlanToolEnabled { enabled } => {
                self.sm.ctx.set_plan_tool_enabled(enabled);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetTokenizer { .. } => {
                // Local BPE tokenisers are no longer used — accuracy comes from
                // observed_input_tokens reported by the provider API (P0-1 Step 2).
                // char_approx is always used for pre-flight truncation estimates.
                self.sm.ctx.engine = ContextTokenEngine::char_approx();
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::AddSystemMessage { content, tokens } => {
                self.sm
                    .ctx
                    .partitions
                    .system
                    .push(Message::system(content), tokens.max(1));
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::AddKnowledgeMessage {
                content,
                tokens,
                key,
                pinned,
            } => {
                // P1-B2 cache contract: the knowledge partition renders into the cached system[1]
                // block. Appending here is the right home for *stable* reference material (skill
                // defs, durable artifacts) — it's append-only, so the existing prefix stays
                // byte-stable, and a fresh append costs only a one-time system[1] re-cache. Do NOT
                // route *per-turn* retrievals (a memory/knowledge lookup that changes every turn)
                // through here: each would rewrite the cached block and invalidate it plus the
                // history cache every turn. Volatile per-turn context belongs on the signal/tail
                // path (`push_signal` → state_turn), which is uncached *and* high-attention (P1-F).
                //
                // K1: a `key` gives the entry identity — a same-key push stages a boundary-deferred
                // upsert instead of appending a duplicate (the cache contract above is why the swap
                // waits for the next compaction/renewal boundary).
                self.sm.ctx.push_knowledge_entry(
                    key.map(compact_str::CompactString::from),
                    Message::system(content),
                    tokens.max(1),
                    pinned,
                );
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::RemoveKnowledge { key } => {
                self.sm.ctx.remove_knowledge(&key);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::AddHistoryMessage { message, tokens } => {
                let tokens = tokens.unwrap_or_else(|| self.sm.ctx.engine.count_message(&message));
                self.sm.ctx.push_history(message, tokens.max(1));
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::PreloadHistory { messages } => {
                self.sm.preload_history(messages);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::MountCapability { capability } => {
                self.sm.mount_capability(capability, None, None);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::UnmountCapability {
                capability_kind,
                id,
            } => {
                self.sm.unmount_capability(capability_kind, &id);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::LoadMilestoneContract { contract } => {
                self.sm.load_milestone_contract(contract);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::LoadGovernancePolicy {
                default_action,
                rules,
                vetoed_tools,
                rate_limits,
                constraints,
            } => {
                self.sm.set_governance(build_governance_pipeline(
                    default_action,
                    rules,
                    vetoed_tools,
                    rate_limits,
                    constraints,
                ));
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::ConfigureRun { config } => {
                // K2: apply a bundle of run-setup config in one event (tools / governance / attention /
                // quota / scheduler / toggles), replacing the ~10 separate `Set*` / `Load*` events the
                // SDK used to fire one-by-one. Each field is optional; an absent field is left untouched.
                // The individual events remain for runtime mutation (skill mount, mid-run budget change).
                // Each branch delegates to exactly the method its granular event uses, so the two paths
                // can never diverge.
                let RunConfig {
                    tools,
                    available_skills,
                    stable_core_tools,
                    memory_enabled,
                    knowledge_enabled,
                    plan_tool_enabled,
                    tokenizer,
                    governance,
                    signal_policy,
                    prompt_budget,
                    context_policy,
                    scheduler_policy,
                    resource_quota,
                    budget_grant,
                    repeat_fuse,
                    criteria_gate,
                    knowledge_budget_ratio,
                    entropy_watch,
                    reliability,
                } = config;
                if let Some(tools) = tools {
                    self.sm.tools = tools;
                }
                if let Some(skills) = available_skills {
                    self.sm.ctx.set_available_skills(skills);
                }
                if let Some(ids) = stable_core_tools {
                    self.sm
                        .ctx
                        .set_stable_core_tools(ids.into_iter().map(Into::into));
                }
                if let Some(enabled) = memory_enabled {
                    self.sm.ctx.set_memory_enabled(enabled);
                }
                if let Some(enabled) = knowledge_enabled {
                    self.sm.ctx.set_knowledge_enabled(enabled);
                }
                if let Some(enabled) = plan_tool_enabled {
                    self.sm.ctx.set_plan_tool_enabled(enabled);
                }
                if tokenizer.is_some() {
                    self.sm.ctx.engine = ContextTokenEngine::char_approx();
                }
                if let Some(g) = governance {
                    self.sm.set_governance(build_governance_pipeline(
                        g.default_action,
                        g.rules,
                        g.vetoed_tools,
                        g.rate_limits,
                        g.constraints,
                    ));
                }
                if let Some(policy) = signal_policy {
                    self.sm.set_signal_policy(
                        policy.queue_max as usize,
                        policy.ttl_ms,
                        policy.deadline_escalation.unwrap_or(false),
                    );
                }
                if let Some(prompt_budget) = prompt_budget {
                    self.sm.ctx.set_prompt_budget(prompt_budget);
                }
                if let Some(context_policy) = context_policy {
                    self.sm.ctx.apply_context_policy(&context_policy);
                }
                if let Some(policy) = scheduler_policy {
                    self.sm.set_scheduler_policy(policy);
                }
                if let Some(quota) = resource_quota {
                    self.sm.set_resource_quota(quota);
                }
                if let Some(grant) = budget_grant {
                    self.sm.set_budget_grant(grant);
                }
                if let Some(fuse) = repeat_fuse {
                    self.sm.set_repeat_fuse(fuse);
                }
                if let Some(enabled) = criteria_gate {
                    self.sm.set_criteria_gate(enabled);
                }
                if let Some(ratio) = knowledge_budget_ratio {
                    self.sm.ctx.config.knowledge_budget_ratio = ratio;
                }
                if let Some(watch) = entropy_watch {
                    self.sm.set_entropy_watch(watch);
                }
                if let Some(reliability) = reliability {
                    if let Some(capacity) = reliability.event_replay_capacity {
                        self.recorded_events.set_capacity(capacity);
                    }
                    if let Some(capacity) = reliability.completed_effect_replay_capacity {
                        self.completed_effects.set_capacity(capacity);
                    }
                    if let Some(limit) = reliability.snapshot_input_limit {
                        self.snapshot_input_limit = limit;
                    }
                    if let Some(limit) = reliability.max_input_bytes {
                        self.max_input_bytes = limit;
                    }
                    if let Some(limit) = reliability.snapshot_journal_bytes_limit {
                        self.snapshot_journal_bytes_limit = limit;
                    }
                    self.sm.set_reliability_config(&reliability);
                }
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetSignalPolicy { policy } => {
                self.sm.set_signal_policy(
                    policy.queue_max as usize,
                    policy.ttl_ms,
                    policy.deadline_escalation.unwrap_or(false),
                );
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::PageIn { entries } => {
                self.sm.apply_page_in(&entries);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::ForceCompact => {
                self.sm.force_compact();
                self.sm
                    .externalize_pending_host_effect(LoopAction::AwaitingResume)
            }
            KernelInputEvent::UpdateTask { update } => {
                self.sm.ctx.update_task(update);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::StartRun { task, run_spec } => {
                self.sm.run_spec = run_spec;
                self.sm.start(task)
            }
            KernelInputEvent::CapabilityCommand { command } => {
                self.sm.execute_capability_command(command);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::Resume => self.sm.resume_after_preload(),
            KernelInputEvent::ApprovalResult {
                effect_id: _,
                approved_calls,
                denied_calls,
                error,
            } => match error {
                Some(error) => self.sm.retry_approval(error),
                None => self.sm.resolve_approval(approved_calls, denied_calls),
            },
            KernelInputEvent::WorkflowSpawnResult {
                effect_id: _,
                started_agent_ids,
                failures,
                error,
            } => match error {
                Some(error) => self.sm.retry_workflow_spawn(error),
                None => self.sm.resolve_workflow_spawn(started_agent_ids, failures),
            },
            KernelInputEvent::PreemptResult {
                effect_id: _,
                error,
            } => match error {
                Some(error) => self.sm.retry_preempt(error),
                None => self.sm.resolve_preempt(),
            },
            KernelInputEvent::MemoryPersistResult {
                effect_id: _,
                error,
            } => {
                let memory = self
                    .pending_memory_write
                    .take()
                    .expect("validated memory result requires pending write");
                let staged_store = self
                    .pending_memory_store
                    .take()
                    .expect("validated memory result requires staged scoped upsert");
                let turn = self.sm.turn;
                match error {
                    Some(error) => {
                        self.sm
                            .observations
                            .push(KernelObservation::MemoryWriteFailed {
                                turn,
                                record_id: memory.record_id,
                                error,
                            })
                    }
                    None => {
                        self.memory_records = staged_store;
                        self.sm.observations.push(KernelObservation::MemoryWritten {
                            turn,
                            record_id: memory.record_id,
                            scope: memory.scope,
                            memory_kind: memory.kind,
                            name: memory.name,
                            size_bytes: memory.content.len() as u32,
                        });
                    }
                }
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::MemoryQueryResult {
                effect_id: _,
                hits,
                error,
            } => {
                let (query, requested_k) = self
                    .pending_memory_query
                    .take()
                    .expect("validated memory result requires pending query");
                let turn = self.sm.turn;
                let continuation = match error {
                    Some(error) => {
                        let continuation = self.sm.resume_after_preload();
                        self.sm
                            .observations
                            .push(KernelObservation::MemoryQueryFailed {
                                turn,
                                scope: query.scope,
                                query: query.query,
                                error,
                            });
                        continuation
                    }
                    None => {
                        // M3/M4: journal recall lifecycle derived from the routed hits. Each hit
                        // carries the host store's current recall_count, so the incremented count is
                        // computed statelessly here (the durable ledger lives host-side) and mirrored
                        // back via observation. Recall time is the current turn — deterministic and
                        // replay-safe; the kernel owns no wall clock. Promotion is edge-triggered on
                        // the threshold crossing so it never nags.
                        let promotion_threshold = self
                            .sm
                            .memory_policy()
                            .and_then(|policy| policy.promotion_recall_threshold);
                        let mut recalls = Vec::new();
                        let mut promotions = Vec::new();
                        for hit in &hits {
                            let trust = match hit.record.provenance.trust {
                                crate::mm::memory::MemoryTrustLevel::Untrusted => "untrusted",
                                crate::mm::memory::MemoryTrustLevel::UserAsserted => {
                                    "user_asserted"
                                }
                                crate::mm::memory::MemoryTrustLevel::HostVerified => {
                                    "host_verified"
                                }
                            };
                            let content = format!(
                                "[MEMORY record_id={} trust={} score={:.3}] {}",
                                hit.record.record_id, trust, hit.score, hit.record.content
                            );
                            let tokens = self.sm.ctx.engine.count(&content).max(1);
                            self.sm.ctx.push_history(
                                crate::types::message::Message::user(content),
                                tokens,
                            );

                            let record_id = hit.record.record_id.clone();
                            let before = hit.record.recall_count;
                            let after = before.saturating_add(1);
                            recalls.push(crate::mm::memory::MemoryRecallLifecycle {
                                record_id: record_id.clone(),
                                recall_count: after,
                                last_recalled_at: turn as u64,
                            });
                            if let Some(threshold) = promotion_threshold {
                                if before < threshold && after >= threshold {
                                    promotions.push((record_id, after));
                                }
                            }
                        }
                        let continuation = self.sm.resume_after_preload();
                        self.sm.observations.push(KernelObservation::MemoryQueried {
                            turn,
                            scope: query.scope.clone(),
                            query: query.query,
                            requested_k,
                            requires_async_response: false,
                        });
                        if !recalls.is_empty() {
                            self.sm.observations.push(KernelObservation::MemoryRecalled {
                                turn,
                                scope: query.scope,
                                recalls,
                            });
                        }
                        for (record_id, recall_count) in promotions {
                            self.sm
                                .observations
                                .push(KernelObservation::PromotionSuggested {
                                    turn,
                                    record_id,
                                    recall_count,
                                });
                        }
                        continuation
                    }
                };
                // Recall is a deferred host effect in the reasoning path. Re-render after
                // committing the hits so the provider continuation observes the recalled
                // history instead of leaving the loop parked on an empty step.
                return identity.single(continuation, self.sm.take_observations());
            }
            KernelInputEvent::LargeResultSpoolResult {
                effect_id: _,
                spool_ref,
                error,
            } => self.sm.resolve_large_result_spool(spool_ref, error),
            KernelInputEvent::PageOutArchiveResult {
                effect_id: _,
                archive_ref,
                error,
            } => self.sm.resolve_page_out_archive(archive_ref, error),
            KernelInputEvent::SetSchedulerBudget { max_wall_ms } => {
                self.sm.set_wall_budget(max_wall_ms);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetResourceQuota { quota } => {
                self.sm.set_resource_quota(quota);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetCriteriaGate { enabled } => {
                self.sm.set_criteria_gate(enabled);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetKnowledgeBudget { ratio } => {
                self.sm.ctx.config.knowledge_budget_ratio = ratio;
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetRepeatFuse {
                enabled,
                deny_after,
                terminate_after,
            } => {
                let mut cfg = self.sm.repeat_fuse_config();
                if let Some(e) = enabled {
                    cfg.enabled = e;
                }
                if let Some(d) = deny_after {
                    cfg.deny_after = d;
                }
                if let Some(t) = terminate_after {
                    cfg.terminate_after = t;
                }
                self.sm.set_repeat_fuse(cfg);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::SetEntropyWatch {
                enabled,
                threshold,
                hysteresis,
                cooldown_turns,
                notify_model,
            } => {
                let mut cfg = self.sm.entropy_watch_config();
                if let Some(e) = enabled {
                    cfg.enabled = e;
                }
                if let Some(t) = threshold {
                    cfg.threshold = t;
                }
                if let Some(h) = hysteresis {
                    cfg.hysteresis = h;
                }
                if let Some(c) = cooldown_turns {
                    cfg.cooldown_turns = c;
                }
                if let Some(n) = notify_model {
                    cfg.notify_model = n;
                }
                self.sm.set_entropy_watch(cfg);
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::ProviderResult {
                effect_id: _,
                message,
                observed_input_tokens,
                observed_output_tokens: _,
                now_ms,
                stop_reason,
            } => {
                if let Some(tokens) = observed_input_tokens {
                    self.sm.ctx.set_observed_prompt_tokens(tokens);
                }
                // Feed the clock before the governance gate fires inside `feed`, so the
                // rate limiter sees a real timestamp (no-op when no policy is loaded).
                if let Some(ms) = now_ms {
                    self.sm.set_observed_time(ms);
                }
                // Stash stop_reason so `feed` can detect an output-cap truncation and drive recovery.
                self.sm.set_pending_stop_reason(stop_reason);
                self.sm.feed(LoopEvent::LLMResponse { message })
            }
            KernelInputEvent::ToolResults {
                effect_id: _,
                results,
            } => self.sm.feed(LoopEvent::ToolResults { results }),
            KernelInputEvent::ProviderError {
                effect_id: _,
                message,
            } => {
                // Reactive recovery is a kernel decision: classify + bounded compact-and-retry,
                // returning the next action (retry or honest terminal) through the common tail.
                self.sm.recover_from_provider_error(&message)
            }
            KernelInputEvent::DeliverSignal {
                delivery_id,
                attempt,
                signal,
            } => match self.sm.signal_event(
                identity.operation_id.clone(),
                delivery_id,
                attempt,
                signal,
            ) {
                Some(action) => action,
                // Non-actionable disposition (queued / observed / ignored / dropped):
                // no provider call this step, just the correlated delivery disposition.
                None => {
                    return identity.empty(self.sm.take_observations());
                }
            },
            KernelInputEvent::MilestoneResult {
                effect_id: _,
                result,
            } => self.sm.feed(LoopEvent::MilestoneResult { result }),
            KernelInputEvent::SpawnSubAgent {
                spec,
                parent_session_id,
            } => self.sm.spawn_sub_agent(spec, &parent_session_id),
            KernelInputEvent::LoadWorkflow {
                spec,
                parent_session_id,
                resumed_submissions,
                resumed_submission_bases,
                resumed_outcomes,
            } => {
                if resumed_outcomes.is_empty() && resumed_submissions.is_empty() {
                    self.sm.load_workflow(spec, &parent_session_id)
                } else {
                    self.sm.load_workflow_resumed(
                        spec,
                        &parent_session_id,
                        &resumed_submissions,
                        &resumed_submission_bases,
                        &resumed_outcomes,
                    )
                }
            }
            KernelInputEvent::SubAgentCompleted { result } => {
                self.sm.feed(LoopEvent::SubAgentCompleted { result })
            }
            KernelInputEvent::SubmitWorkflowNodes {
                nodes,
                submitter_agent_id,
            } => self
                .sm
                .submit_workflow_nodes(nodes, submitter_agent_id.as_deref()),
            KernelInputEvent::SubmitWorkflow {
                spec,
                parent_session_id,
                submitter_agent_id,
            } => self
                .sm
                .submit_workflow(spec, &parent_session_id, submitter_agent_id.as_deref()),
            KernelInputEvent::SetMemoryPolicy {
                memory_path,
                stale_warning_days,
                retrieval_top_k,
                validation_enabled,
                max_content_bytes,
                max_name_length,
                promotion_recall_threshold,
            } => {
                // Phase 7: install the memory policy. The kernel enforces validation_enabled +
                // retrieval_top_k + size/name overrides at the WriteMemory/QueryMemory traps and
                // (M4) promotion_recall_threshold on recall; memory_path / stale_warning_days are
                // carried for the SDK's recall I/O. Capacity eviction is host-side (the host store
                // owns the full cross-session record set).
                self.sm.set_memory_policy(crate::mm::memory::MemoryPolicy {
                    memory_path,
                    stale_warning_days,
                    retrieval_top_k,
                    validation_enabled,
                    max_content_bytes,
                    max_name_length,
                    promotion_recall_threshold,
                });
                return identity.empty(self.sm.take_observations());
            }
            KernelInputEvent::WriteMemory { memory } => {
                // Phase 7: Validate memory write request.
                // Kernel validates; SDK performs I/O.
                use crate::mm::memory::validate_memory_write;
                let turn = self.sm.turn;
                // M2: route the write through the syscall trap so the resource quota (write-rate
                // limit) applies. A rate-limited / denied write surfaces as a validation failure
                // (the write does not happen) and short-circuits before validation.
                let disposition = self
                    .sm
                    .gate_syscall(&crate::syscall::Syscall::WriteMemory(memory.clone()));
                if !disposition.is_allowed() {
                    let error = match disposition {
                        crate::syscall::Disposition::RateLimited { retry_after_ms } => {
                            format!("memory write rate limited; retry after {retry_after_ms}ms")
                        }
                        crate::syscall::Disposition::Deny { reason, .. } => {
                            format!("memory write denied: {reason}")
                        }
                        _ => "memory write not permitted".to_string(),
                    };
                    self.sm
                        .observations
                        .push(KernelObservation::MemoryValidationFailed {
                            turn,
                            record_id: memory.record_id.clone(),
                            error,
                        });
                    return identity.empty(self.sm.take_observations());
                }
                // Validate honoring any installed memory policy: a policy with validation disabled
                // admits the write outright; a policy with size/name overrides validates against
                // those; no policy uses the default rules (pre-policy behavior).
                let validation_result = match self.sm.memory_policy() {
                    Some(p) if !p.validation_enabled => Ok(()),
                    Some(p) => p.validation().validate(&memory),
                    None => validate_memory_write(&memory),
                };
                match validation_result {
                    Ok(()) => {
                        let key = memory.key();
                        let mut staged_store = self.memory_records.clone();
                        if let Err(error) = staged_store.upsert(memory) {
                            self.sm
                                .observations
                                .push(KernelObservation::MemoryValidationFailed {
                                    turn,
                                    record_id: error.record_id().to_string(),
                                    error: format!("record id conflicts with another scoped key"),
                                });
                            return identity.empty(self.sm.take_observations());
                        }
                        let canonical = staged_store
                            .get(&key.scope, key.kind, &key.name)
                            .expect("scoped upsert must retain its canonical record")
                            .clone();
                        self.pending_memory_write = Some(canonical.clone());
                        self.pending_memory_store = Some(staged_store);
                        LoopAction::PersistMemory { memory: canonical }
                    }
                    Err(err) => {
                        // Emit validation error observation
                        use crate::mm::memory::MemoryValidationError;
                        let error_msg = match err {
                            MemoryValidationError::MissingRequiredField { field } => {
                                format!("Missing required field: {}", field)
                            }
                            MemoryValidationError::ContentTooLarge { size, limit } => {
                                format!("Content too large: {} bytes (limit: {})", size, limit)
                            }
                            MemoryValidationError::ForbiddenPattern { pattern, reason } => {
                                format!("Forbidden pattern '{}': {}", pattern, reason)
                            }
                            MemoryValidationError::InvalidKind { kind } => {
                                format!("Invalid kind: {}", kind)
                            }
                            MemoryValidationError::NameTooLong { length, limit } => {
                                format!("Name too long: {} chars (limit: {})", length, limit)
                            }
                        };
                        self.sm
                            .observations
                            .push(KernelObservation::MemoryValidationFailed {
                                turn,
                                record_id: memory.record_id.clone(),
                                error: error_msg,
                            });
                        return identity.empty(self.sm.take_observations());
                    }
                }
            }
            KernelInputEvent::QueryMemory { query } => {
                // Phase 7: Query memory for context.
                // Kernel emits observation; SDK responds asynchronously.
                // An installed policy caps retrieval breadth: requested_k = min(query.top_k, policy).
                let requested_k = match self.sm.memory_policy() {
                    Some(p) => p.clamp_top_k(query.top_k),
                    None => query.top_k,
                };
                self.pending_memory_query = Some((query.clone(), requested_k));
                LoopAction::QueryMemory { query, requested_k }
            }
            KernelInputEvent::CompleteRun => self.sm.feed(LoopEvent::Complete),
            KernelInputEvent::CancelOperation {
                operation_id,
                reason,
                pending_call_ids,
            } => {
                self.pending_memory_write = None;
                self.pending_memory_store = None;
                self.pending_memory_query = None;
                self.sm
                    .cancel_operation(operation_id, reason, pending_call_ids)
            }
        };
        let action = self.sm.externalize_pending_host_effect(action);
        if matches!(action, LoopAction::AwaitingResume) {
            return identity.empty(self.sm.take_observations());
        }
        identity.single(action, self.sm.take_observations())
    }

    fn lifecycle_transition(
        &self,
        event: &KernelInputEvent,
    ) -> Result<LifecycleTransition, String> {
        if self.lifecycle.is_terminal() && !matches!(event, KernelInputEvent::DeliverSignal { .. })
        {
            return Err(format!(
                "kernel is terminal in lifecycle {:?}",
                self.lifecycle
            ));
        }

        match event {
            KernelInputEvent::ConfigureRun { .. } => match self.lifecycle {
                KernelLifecycle::Created | KernelLifecycle::Configured => {
                    Ok(LifecycleTransition::Configure)
                }
                _ => Err(format!(
                    "configure_run is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::SetSignalPolicy { .. } => match self.lifecycle {
                KernelLifecycle::Created | KernelLifecycle::Configured => {
                    Ok(LifecycleTransition::Configure)
                }
                _ => Err(format!(
                    "set_signal_policy is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::StartRun { .. } => match self.lifecycle {
                KernelLifecycle::Created | KernelLifecycle::Configured => {
                    Ok(LifecycleTransition::Start)
                }
                _ => Err(format!(
                    "start_run is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::Resume => match self.lifecycle {
                KernelLifecycle::Configured => Ok(LifecycleTransition::Resume),
                _ => Err(format!(
                    "resume is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::ApprovalResult { .. }
            | KernelInputEvent::WorkflowSpawnResult { .. }
            | KernelInputEvent::PreemptResult { .. } => match self.lifecycle {
                KernelLifecycle::Suspended => Ok(LifecycleTransition::Resume),
                _ => Err(format!(
                    "effect result is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::MemoryPersistResult { .. }
            | KernelInputEvent::LargeResultSpoolResult { .. }
            | KernelInputEvent::PageOutArchiveResult { .. } => match self.lifecycle {
                KernelLifecycle::Configured | KernelLifecycle::Running => {
                    Ok(LifecycleTransition::Stay)
                }
                _ => Err(format!(
                    "memory effect result is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::MemoryQueryResult { .. } => match self.lifecycle {
                KernelLifecycle::Configured | KernelLifecycle::Running => {
                    Ok(LifecycleTransition::Resume)
                }
                _ => Err(format!(
                    "memory query result is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::ProviderResult { .. }
            | KernelInputEvent::ProviderError { .. }
            | KernelInputEvent::ToolResults { .. }
            | KernelInputEvent::MilestoneResult { .. }
            | KernelInputEvent::LoadWorkflow { .. }
            | KernelInputEvent::SpawnSubAgent { .. } => match self.lifecycle {
                KernelLifecycle::Running => Ok(LifecycleTransition::Stay),
                _ => Err(format!(
                    "execution input is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::SubmitWorkflow { .. }
            | KernelInputEvent::SubmitWorkflowNodes { .. }
            | KernelInputEvent::CompleteRun
            | KernelInputEvent::CancelOperation { .. } => match self.lifecycle {
                KernelLifecycle::Running | KernelLifecycle::Suspended => {
                    Ok(LifecycleTransition::Stay)
                }
                _ => Err(format!(
                    "execution input is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::DeliverSignal { .. } => match self.lifecycle {
                KernelLifecycle::Running
                | KernelLifecycle::Suspended
                | KernelLifecycle::Completed
                | KernelLifecycle::Failed
                | KernelLifecycle::Cancelled => Ok(LifecycleTransition::Stay),
                _ => Err(format!(
                    "deliver_signal is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            KernelInputEvent::SubAgentCompleted { .. } => match self.lifecycle {
                KernelLifecycle::Running | KernelLifecycle::Suspended => {
                    Ok(LifecycleTransition::Resume)
                }
                _ => Err(format!(
                    "sub_agent_completed is not valid in lifecycle {:?}",
                    self.lifecycle
                )),
            },
            _ => match self.lifecycle {
                KernelLifecycle::Suspended => Err(
                    "only resume, sub_agent_completed, or cancellation is valid while suspended"
                        .to_string(),
                ),
                KernelLifecycle::Created => Ok(LifecycleTransition::Configure),
                _ => Ok(LifecycleTransition::Stay),
            },
        }
    }

    fn advance_lifecycle(&mut self, transition: LifecycleTransition, step: &KernelStep) {
        if let Some(result) = step.actions.iter().find_map(|action| match &action.effect {
            KernelEffect::Done { result } => Some(result),
            _ => None,
        }) {
            self.lifecycle = match result.termination {
                crate::types::result::TerminationReason::Completed => KernelLifecycle::Completed,
                crate::types::result::TerminationReason::UserAbort => KernelLifecycle::Cancelled,
                _ => KernelLifecycle::Failed,
            };
            return;
        }

        if self.sm.is_suspended() {
            self.lifecycle = KernelLifecycle::Suspended;
            return;
        }

        self.lifecycle = match transition {
            LifecycleTransition::Configure if self.lifecycle == KernelLifecycle::Created => {
                KernelLifecycle::Configured
            }
            LifecycleTransition::Start | LifecycleTransition::Resume => KernelLifecycle::Running,
            _ => self.lifecycle,
        };
    }

    fn allocate_step_seq(&mut self) -> u64 {
        let step_seq = self.next_step_seq;
        self.next_step_seq = self.next_step_seq.saturating_add(1);
        step_seq
    }

    fn fault_step(
        &mut self,
        operation_id: String,
        event_id: String,
        code: KernelFaultCode,
        message: String,
        effect_id: Option<String>,
    ) -> KernelStep {
        // Rejected transactions do not consume a committed step identity. This keeps effect IDs
        // deterministic across retries and lets snapshots rebuild from accepted inputs only.
        let step_seq = self.next_step_seq;
        KernelStep::fault(
            operation_id.clone(),
            event_id.clone(),
            step_seq,
            KernelFault {
                code,
                message,
                operation_id: Some(operation_id),
                event_id: Some(event_id),
                effect_id,
            },
        )
    }
}

fn normalize_input(input: &mut KernelInput) {
    if let KernelInputEvent::CancelOperation {
        pending_call_ids, ..
    } = &mut input.event
    {
        pending_call_ids.sort();
        pending_call_ids.dedup();
    }
}

fn transaction_fault_step(
    operation_id: String,
    event_id: String,
    step_seq: u64,
    code: KernelFaultCode,
    message: String,
) -> KernelStep {
    KernelStep::fault(
        operation_id.clone(),
        event_id.clone(),
        step_seq,
        KernelFault {
            code,
            message,
            operation_id: Some(operation_id),
            event_id: Some(event_id),
            effect_id: None,
        },
    )
}

fn transaction_conflict_fault(operation_id: Option<String>, message: String) -> KernelFault {
    KernelFault {
        code: KernelFaultCode::TransactionConflict,
        message,
        operation_id,
        event_id: None,
        effect_id: None,
    }
}

fn snapshot_fault(operation_id: Option<String>, message: String) -> KernelFault {
    KernelFault {
        code: KernelFaultCode::SnapshotIncompatible,
        message,
        operation_id,
        event_id: None,
        effect_id: None,
    }
}

fn result_effect(event: &KernelInputEvent) -> Option<(&str, PendingEffectKind)> {
    match event {
        KernelInputEvent::ProviderResult { effect_id, .. }
        | KernelInputEvent::ProviderError { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::Provider))
        }
        KernelInputEvent::ToolResults { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::Tool))
        }
        KernelInputEvent::MilestoneResult { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::Milestone))
        }
        KernelInputEvent::ApprovalResult { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::Approval))
        }
        KernelInputEvent::WorkflowSpawnResult { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::WorkflowSpawn))
        }
        KernelInputEvent::PreemptResult { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::Preempt))
        }
        KernelInputEvent::MemoryPersistResult { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::MemoryPersist))
        }
        KernelInputEvent::MemoryQueryResult { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::MemoryQuery))
        }
        KernelInputEvent::LargeResultSpoolResult { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::LargeResultSpool))
        }
        KernelInputEvent::PageOutArchiveResult { effect_id, .. } => {
            Some((effect_id, PendingEffectKind::PageOutArchive))
        }
        _ => None,
    }
}

fn pending_effect_kind(effect: &KernelEffect) -> Option<PendingEffectKind> {
    match effect {
        KernelEffect::CallProvider { .. } => Some(PendingEffectKind::Provider),
        KernelEffect::ExecuteTool { .. } => Some(PendingEffectKind::Tool),
        KernelEffect::EvaluateMilestone { .. } => Some(PendingEffectKind::Milestone),
        KernelEffect::RequestApproval { .. } => Some(PendingEffectKind::Approval),
        KernelEffect::SpawnWorkflow { .. } => Some(PendingEffectKind::WorkflowSpawn),
        KernelEffect::PreemptSubAgents { .. } => Some(PendingEffectKind::Preempt),
        KernelEffect::PersistMemory { .. } => Some(PendingEffectKind::MemoryPersist),
        KernelEffect::QueryMemory { .. } => Some(PendingEffectKind::MemoryQuery),
        KernelEffect::SpoolLargeResult { .. } => Some(PendingEffectKind::LargeResultSpool),
        KernelEffect::ArchivePageOut { .. } => Some(PendingEffectKind::PageOutArchive),
        KernelEffect::Done { .. } => None,
    }
}

fn validate_run_config(config: &RunConfig, max_tokens: u32) -> Result<(), String> {
    if config
        .budget_grant
        .as_ref()
        .is_some_and(|grant| grant.reservation_id.is_empty())
    {
        return Err("budget_grant reservation_id must be non-empty".to_string());
    }
    if let Some(reliability) = &config.reliability {
        for (name, capacity) in [
            ("event_replay_capacity", reliability.event_replay_capacity),
            (
                "completed_effect_replay_capacity",
                reliability.completed_effect_replay_capacity,
            ),
        ] {
            if let Some(capacity) = capacity {
                if !(1..=65_536).contains(&capacity) {
                    return Err(format!("{name} must be between 1 and 65536"));
                }
            }
        }
        if reliability
            .snapshot_input_limit
            .is_some_and(|value| !(1..=100_000).contains(&value))
        {
            return Err("snapshot_input_limit must be between 1 and 100000".to_string());
        }
        if reliability
            .max_input_bytes
            .is_some_and(|value| !(256..=64 * 1024 * 1024).contains(&value))
        {
            return Err("max_input_bytes must be between 256 and 67108864".to_string());
        }
        if reliability
            .snapshot_journal_bytes_limit
            .is_some_and(|value| !(256..=1024 * 1024 * 1024).contains(&value))
        {
            return Err(
                "snapshot_journal_bytes_limit must be between 256 and 1073741824".to_string(),
            );
        }
        if reliability
            .provider_recovery_attempts
            .is_some_and(|value| value > 16)
        {
            return Err("provider_recovery_attempts must be at most 16".to_string());
        }
        if reliability
            .output_recovery_attempts
            .is_some_and(|value| value > 16)
        {
            return Err("output_recovery_attempts must be at most 16".to_string());
        }
        if reliability
            .host_effect_retry_attempts
            .is_some_and(|value| value > 16)
        {
            return Err("host_effect_retry_attempts must be at most 16".to_string());
        }
        let threshold = reliability.spool_threshold_bytes.unwrap_or(50 * 1024);
        let preview = reliability.spool_preview_bytes.unwrap_or(2 * 1024);
        if threshold == 0 {
            return Err("spool_threshold_bytes must be greater than zero".to_string());
        }
        if preview == 0 || preview > threshold {
            return Err(
                "spool_preview_bytes must be greater than zero and no larger than spool_threshold_bytes"
                    .to_string(),
            );
        }
    }
    if let Some(policy) = &config.signal_policy {
        validate_signal_policy(policy)?;
    }
    if config
        .prompt_budget
        .is_some_and(|budget| budget.reserved_tokens() >= max_tokens)
    {
        return Err(
            "prompt_budget reserves must leave at least one token for provider input".to_string(),
        );
    }
    if let Some(policy) = &config.context_policy {
        policy.validate()?;
    }
    if let Some(policy) = config.scheduler_policy {
        policy.validate()?;
    }
    if let Some(ratio) = config.knowledge_budget_ratio {
        if !ratio.is_finite() || !(0.0..=1.0).contains(&ratio) {
            return Err("knowledge_budget_ratio must be finite and between 0 and 1".to_string());
        }
    }
    if let Some(fuse) = config.repeat_fuse {
        if fuse.enabled
            && fuse.deny_after > 0
            && fuse.terminate_after > 0
            && fuse.deny_after >= fuse.terminate_after
        {
            return Err("repeat_fuse terminate_after must be greater than deny_after".to_string());
        }
    }
    if let Some(watch) = config.entropy_watch {
        if !watch.threshold.is_finite() || !(0.0..=1.0).contains(&watch.threshold) {
            return Err("entropy_watch threshold must be finite and between 0 and 1".to_string());
        }
        if !watch.hysteresis.is_finite()
            || watch.hysteresis < 0.0
            || watch.hysteresis > watch.threshold
        {
            return Err(
                "entropy_watch hysteresis must be finite and no greater than threshold".to_string(),
            );
        }
    }
    if let Some(quota) = &config.resource_quota {
        if matches!(quota.memory_writes_per_window, Some((_, 0))) {
            return Err("memory write quota window must be greater than zero".to_string());
        }
    }
    if let Some(governance) = &config.governance {
        if governance
            .rate_limits
            .iter()
            .any(|limit| limit.window_ms == 0)
        {
            return Err("governance rate-limit windows must be greater than zero".to_string());
        }
        for constraint in &governance.constraints {
            if let ConstraintSpec::Range {
                min: Some(min),
                max: Some(max),
                ..
            } = constraint
            {
                if !min.is_finite() || !max.is_finite() || min > max {
                    return Err(
                        "governance range constraints require finite min <= max".to_string()
                    );
                }
            }
        }
    }
    Ok(())
}

fn validate_signal_policy(policy: &SignalPolicyConfig) -> Result<(), String> {
    if policy.version != SIGNAL_POLICY_VERSION {
        return Err(format!(
            "signal_policy version must be {SIGNAL_POLICY_VERSION}"
        ));
    }
    if policy.queue_max == 0 {
        return Err("signal_policy.queue_max must be greater than zero".to_string());
    }
    if matches!(policy.ttl_ms, Some(0)) {
        return Err("signal_policy.ttl_ms must be greater than zero when present".to_string());
    }
    Ok(())
}