enact_core/kernel/

kernel.rs

//! ExecutionKernel - The core execution engine
//!
//! The kernel is the single point of execution. It:
//! - Owns the Execution state
//! - Applies actions through the reducer
//! - Emits events for observers
//! - Enforces invariants
//!
//! All execution MUST go through the kernel.
//!
//! ## ⚠️ CODE OWNERSHIP & FORBIDDEN PATTERNS
//!
//! **This module is the SINGLE SOURCE OF TRUTH for execution orchestration.**
//!
//! ### Code Ownership
//! - Only `kernel::kernel` (ExecutionKernel) may orchestrate execution
//! - ExecutionKernel owns the Execution state
//! - All state transitions MUST go through `kernel::reducer::reduce()`
//!
//! ### Explicitly Forbidden Patterns
//!
//! These patterns are **forbidden forever**. If any of these happen, Enact loses its "Now" guarantee.
//!
//! 1. **Kernel calling providers directly** – Providers are resolved before kernel execution.
//!    - The kernel receives resolved providers via `ExecutionRequest`, not provider names or registry lookups
//!    - No global registries or dynamic discovery in kernel
//!    - Provider resolution happens outside kernel (in runner/control plane)
//!
//! 2. **Streaming mutating state** – Streaming only subscribes and delivers events, never mutates execution state.
//!    - Streaming is a read-only observer
//!    - EventEmitter must not have access to ExecutionKernel, Reducer, or ExecutionState for mutation
//!
//! 3. **Signals driving execution** – Signals are hints only, never drive state transitions.
//!    - SignalBus implementations must not have access to ExecutionKernel, Reducer, or ExecutionState
//!    - Signals cannot directly trigger state changes
//!
//! 4. **Tools bypassing ToolPolicy** – All tool execution MUST go through ToolExecutor.
//!    - ToolExecutor enforces ToolPolicy before every invocation
//!    - No direct tool calls from kernel
//!
//! 5. **Context being optional** – TenantContext is REQUIRED for all executions.
//!    - There is no "system" execution without a tenant
//!    - All execution methods must require TenantContext
//!
//! 6. **IDs being redefined outside kernel** – Kernel is the ONLY source of truth for IDs.
//!    - No other module may define ExecutionId, StepId, or other execution identifiers
//!    - All IDs must come from `kernel::ids`
//!
//! ### Invariants Enforced
//!
//! - **Single source of truth**: Only the kernel may mutate `Execution` or `Step` state via `kernel::reducer`
//! - **Policy-first enforcement**: All policy checks run before external providers; decisions are recorded as events
//! - **Execution Service Parity**: Enact must always run as a decoupled execution service (HTTP/gRPC)
//! - **Observable decision points**: Branching decisions emit `Decision` events with evidence artifacts
//!
//! @see docs/TECHNICAL/04-KERNEL_INVARIANTS.md
//!
//! ## Error Handling (feat-02)
//!
//! All failures use `ExecutionError` which provides:
//! - Deterministic retry decisions
//! - Structured error categories
//! - Backoff hints
//! - Idempotency tracking

use super::artifact::{ArtifactStore, ArtifactType, PutArtifactRequest};
use super::enforcement::{
    EnforcementMiddleware, EnforcementResult, ExecutionUsage, LongRunningExecutionPolicy,
};
use super::error::ExecutionError;
use super::execution_model::Execution;
use super::execution_state::{ExecutionState, StepState, WaitReason};
use super::ids::{ArtifactId, ExecutionId, SpawnMode, StepId, StepType};
use super::persistence::{ExecutionSnapshot, StateStore};
use super::reducer::{reduce, ExecutionAction, ReducerError};
use crate::context::TenantContext;
use crate::graph::{CompiledGraph, NodeState};
use crate::inbox::{ControlAction, InboxMessage, InboxStore};
use crate::signal::SignalBus;
use crate::streaming::{EventEmitter, ProtectedEventEmitter, StreamEvent};
use std::sync::Arc;
use std::time::Instant;
use tokio_util::sync::CancellationToken;

/// Action to take after processing inbox messages
#[derive(Debug, Clone)]
enum InboxAction {
    /// Continue execution normally
    Continue,
    /// Pause execution
    Pause,
    /// Cancel execution with reason
    Cancel(String),
}

/// ExecutionKernel - the core execution engine
///
/// This is THE place where execution happens. Runner wires things up,
/// but kernel does the work.
///
/// ## Invariant: TenantContext is REQUIRED
///
/// Every execution MUST have a TenantContext. This ensures:
/// - Multi-tenant isolation
/// - Resource limit enforcement
/// - Billing attribution
/// - Audit compliance
pub struct ExecutionKernel {
    /// Current execution state
    execution: Execution,
    /// Tenant context (REQUIRED - enforces multi-tenant isolation)
    tenant_context: TenantContext,
    /// Event emitter for streaming
    emitter: EventEmitter,
    /// Protected event emitter for sensitive content (feat-guardrails)
    ///
    /// When configured, events with potentially sensitive content (step outputs,
    /// tool results) are emitted through this emitter which applies protection
    /// processors (PII masking, content filtering, etc.) before streaming.
    protected_emitter: Option<ProtectedEventEmitter>,
    /// Cancellation token for async cancellation (proper cooperative cancellation)
    cancellation_token: CancellationToken,
    /// Inbox store for mid-execution guidance (INV-INBOX-*)
    inbox: Option<Arc<dyn InboxStore>>,
    /// Optional mutable snapshot store (best-effort cache, non-authoritative)
    state_store: Option<Arc<dyn StateStore>>,
    /// Optional signal bus for non-authoritative lifecycle hints/notifications
    signal_bus: Option<Arc<dyn SignalBus>>,
    /// Artifact store for storing execution artifacts (feat-04)
    artifact_store: Option<Arc<dyn ArtifactStore>>,
    /// Enforcement middleware for resource limits (feat-03)
    ///
    /// Tracks usage (steps, tokens, cost, discovery depth) and enforces limits
    /// before each step execution. Integrated with long-running execution controls.
    enforcement: Arc<EnforcementMiddleware>,
    /// Long-running execution policy (agentic DAG controls)
    ///
    /// Controls discovery depth, discovered step limits, cost thresholds,
    /// and idle timeout for long-running agentic executions.
    long_running_policy: LongRunningExecutionPolicy,
    /// Execution usage tracker (registered with enforcement middleware)
    usage: Option<Arc<ExecutionUsage>>,
    /// SpawnMode - how this execution was spawned (for inbox routing)
    ///
    /// Controls inbox message routing:
    /// - Inline: shares parent's inbox
    /// - Child { inherit_inbox: true }: checks both parent and own inbox
    /// - Child { inherit_inbox: false }: isolated inbox
    ///
    /// @see docs/TECHNICAL/32-SPAWN-MODE.md
    spawn_mode: Option<SpawnMode>,
    /// Parent execution ID (if spawned as child)
    ///
    /// Used for inbox inheritance when spawn_mode is Child with inherit_inbox=true
    parent_execution_id: Option<ExecutionId>,
}

impl ExecutionKernel {
    /// Create a new kernel with a fresh execution
    ///
    /// ## Arguments
    /// * `tenant_context` - REQUIRED tenant context for multi-tenant isolation
    ///
    /// ## Invariant
    /// TenantContext is REQUIRED for all executions. There is no "system" execution
    /// without a tenant - this is enforced at compile time.
    pub fn new(tenant_context: TenantContext) -> Self {
        let mut execution = Execution::new();
        execution.tenant_id = Some(tenant_context.tenant_id.clone());
        Self {
            execution,
            tenant_context,
            emitter: EventEmitter::new(),
            protected_emitter: None,
            cancellation_token: CancellationToken::new(),
            inbox: None,
            state_store: None,
            signal_bus: None,
            artifact_store: None,
            enforcement: Arc::new(EnforcementMiddleware::new()),
            long_running_policy: LongRunningExecutionPolicy::standard(),
            usage: None,
            spawn_mode: None,
            parent_execution_id: None,
        }
    }

    /// Create a kernel with an existing execution (for replay)
    ///
    /// ## Arguments
    /// * `execution` - Existing execution state to resume
    /// * `tenant_context` - REQUIRED tenant context (must match execution's tenant)
    pub fn with_execution(execution: Execution, tenant_context: TenantContext) -> Self {
        // Note: In production, we should verify execution.tenant_id matches tenant_context.tenant_id
        Self {
            execution,
            tenant_context,
            emitter: EventEmitter::new(),
            protected_emitter: None,
            cancellation_token: CancellationToken::new(),
            inbox: None,
            state_store: None,
            signal_bus: None,
            artifact_store: None,
            enforcement: Arc::new(EnforcementMiddleware::new()),
            long_running_policy: LongRunningExecutionPolicy::standard(),
            usage: None,
            spawn_mode: None,
            parent_execution_id: None,
        }
    }

    /// Set the protected event emitter for content protection
    ///
    /// When set, events with potentially sensitive content (step outputs,
    /// tool results, etc.) are emitted through this protected emitter which
    /// applies protection processors before streaming.
    ///
    /// ## Usage
    /// ```ignore
    /// use enact_core::streaming::{ProtectedEventEmitter, PiiProtectionProcessor};
    ///
    /// let protected_emitter = ProtectedEventEmitter::new()
    ///     .with_processor(Arc::new(PiiProtectionProcessor::new()));
    ///
    /// let kernel = ExecutionKernel::new(tenant_context)
    ///     .with_protected_emitter(protected_emitter);
    /// ```
    pub fn with_protected_emitter(mut self, emitter: ProtectedEventEmitter) -> Self {
        self.protected_emitter = Some(emitter);
        self
    }

    /// Set the inbox store for mid-execution guidance
    ///
    /// When set, the kernel will check the inbox after every step (INV-INBOX-001)
    /// and process messages in priority order (INV-INBOX-002).
    pub fn with_inbox(mut self, inbox: Arc<dyn InboxStore>) -> Self {
        self.inbox = Some(inbox);
        self
    }

    /// Get the inbox store (if set)
    pub fn inbox(&self) -> Option<&Arc<dyn InboxStore>> {
        self.inbox.as_ref()
    }

    /// Set the state store for snapshot persistence (best-effort cache)
    pub fn with_state_store(mut self, state_store: Arc<dyn StateStore>) -> Self {
        self.state_store = Some(state_store);
        self
    }

    /// Get state store (if configured)
    pub fn state_store(&self) -> Option<&Arc<dyn StateStore>> {
        self.state_store.as_ref()
    }

    /// Set the signal bus for optional execution lifecycle signaling
    pub fn with_signal_bus(mut self, signal_bus: Arc<dyn SignalBus>) -> Self {
        self.signal_bus = Some(signal_bus);
        self
    }

    /// Get signal bus (if configured)
    pub fn signal_bus(&self) -> Option<&Arc<dyn SignalBus>> {
        self.signal_bus.as_ref()
    }

    /// Set the artifact store for storing execution artifacts
    ///
    /// When set, the kernel can store artifacts produced by steps.
    /// Artifacts are emitted as events for audit trail.
    pub fn with_artifact_store(mut self, store: Arc<dyn ArtifactStore>) -> Self {
        self.artifact_store = Some(store);
        self
    }

    /// Get the artifact store (if set)
    pub fn artifact_store(&self) -> Option<&Arc<dyn ArtifactStore>> {
        self.artifact_store.as_ref()
    }

    /// Set the enforcement middleware for resource limits
    ///
    /// When set, the kernel uses this enforcement middleware to track usage
    /// and check limits before each step execution.
    pub fn with_enforcement(mut self, enforcement: Arc<EnforcementMiddleware>) -> Self {
        self.enforcement = enforcement;
        self
    }

    /// Get the enforcement middleware
    pub fn enforcement(&self) -> &Arc<EnforcementMiddleware> {
        &self.enforcement
    }

    /// Set the long-running execution policy
    ///
    /// Controls discovery depth, discovered step limits, cost thresholds,
    /// and idle timeout for long-running agentic executions.
    pub fn with_long_running_policy(mut self, policy: LongRunningExecutionPolicy) -> Self {
        self.long_running_policy = policy;
        self
    }

    /// Get the long-running execution policy
    pub fn long_running_policy(&self) -> &LongRunningExecutionPolicy {
        &self.long_running_policy
    }

    /// Set the spawn mode for this execution
    ///
    /// Controls inbox message routing:
    /// - Inline: shares parent's inbox (same ExecutionId)
    /// - Child { inherit_inbox: true }: checks both parent and own inbox
    /// - Child { inherit_inbox: false }: isolated inbox
    ///
    /// @see docs/TECHNICAL/32-SPAWN-MODE.md
    pub fn with_spawn_mode(mut self, spawn_mode: SpawnMode) -> Self {
        self.spawn_mode = Some(spawn_mode);
        self
    }

    /// Get the spawn mode (if set)
    pub fn spawn_mode(&self) -> Option<&SpawnMode> {
        self.spawn_mode.as_ref()
    }

    /// Set the parent execution ID for child executions
    ///
    /// Used for inbox inheritance when spawn_mode is Child with inherit_inbox=true.
    /// Must be set when spawning child executions that need to inherit parent's inbox.
    pub fn with_parent_execution_id(mut self, parent_id: ExecutionId) -> Self {
        self.parent_execution_id = Some(parent_id);
        self
    }

    /// Get the parent execution ID (if set)
    pub fn parent_execution_id(&self) -> Option<&ExecutionId> {
        self.parent_execution_id.as_ref()
    }

    /// Get current usage snapshot for this execution
    pub fn usage_snapshot(&self) -> Option<super::enforcement::UsageSnapshot> {
        self.usage
            .as_ref()
            .map(|u| super::enforcement::UsageSnapshot::from(u.as_ref()))
    }

    /// Register this execution with the enforcement middleware
    ///
    /// Call this at the start of execution to begin tracking usage.
    /// Returns the usage tracker for this execution.
    pub async fn register_for_enforcement(&mut self) -> Arc<ExecutionUsage> {
        let usage = self
            .enforcement
            .register_execution(
                self.execution.id.clone(),
                self.tenant_context.tenant_id.clone(),
            )
            .await;
        self.usage = Some(Arc::clone(&usage));
        usage
    }

    /// Unregister this execution from the enforcement middleware
    ///
    /// Call this at the end of execution to stop tracking usage.
    pub async fn unregister_from_enforcement(&self) {
        self.enforcement
            .unregister_execution(&self.execution.id)
            .await;
    }

    /// Check all resource limits before executing a step
    ///
    /// This checks:
    /// - Basic limits (steps, tokens, wall time)
    /// - Long-running limits (discovery depth, discovered steps, cost, idle)
    ///
    /// Returns an error if any limit is exceeded.
    pub async fn check_limits_before_step(&self) -> Result<(), ExecutionError> {
        // Check basic resource limits
        let basic_result = self
            .enforcement
            .check_all_limits(&self.execution.id, &self.tenant_context.limits)
            .await;

        if let EnforcementResult::Blocked(violation) = basic_result {
            return Err(violation.to_error());
        }

        // Check long-running execution limits
        let long_running_result = self
            .enforcement
            .check_long_running_limits(&self.execution.id, &self.long_running_policy)
            .await;

        if let EnforcementResult::Blocked(violation) = long_running_result {
            return Err(violation.to_error());
        }

        // Emit warning events if approaching limits
        if self.enforcement.emit_warning_events_enabled() {
            if let Some(warning) = match (&basic_result, &long_running_result) {
                (EnforcementResult::Warning(w), _) => Some(w),
                (_, EnforcementResult::Warning(w)) => Some(w),
                _ => None,
            } {
                tracing::warn!(
                    execution_id = %self.execution.id,
                    warning_type = ?warning.warning_type,
                    usage_percent = warning.usage_percent,
                    message = %warning.message,
                    "Enforcement warning"
                );
                self.emitter.emit(StreamEvent::policy_decision_warn(
                    &self.execution.id,
                    None,
                    "enforcement",
                    warning.message.clone(),
                ));
            }
        }
        Ok(())
    }

    /// Record step completion with the enforcement middleware
    pub async fn record_step_completed(&self) {
        self.enforcement.record_step(&self.execution.id).await;
    }

    /// Record token usage with the enforcement middleware
    pub async fn record_token_usage(&self, input_tokens: u32, output_tokens: u32) {
        self.enforcement
            .record_tokens(&self.execution.id, input_tokens, output_tokens)
            .await;
    }

    /// Record cost with the enforcement middleware
    pub async fn record_cost(&self, cost_usd: f64) {
        self.enforcement
            .record_cost(&self.execution.id, cost_usd)
            .await;
    }

    /// Record a discovered step with the enforcement middleware
    pub async fn record_discovered_step(&self) {
        self.enforcement
            .record_discovered_step(&self.execution.id)
            .await;
    }

    /// Push discovery depth (entering a sub-agent execution)
    pub async fn push_discovery_depth(&self) {
        self.enforcement
            .push_discovery_depth(&self.execution.id)
            .await;
    }

    /// Pop discovery depth (exiting a sub-agent execution)
    pub async fn pop_discovery_depth(&self) {
        self.enforcement
            .pop_discovery_depth(&self.execution.id)
            .await;
    }

    /// Store an artifact produced by a step
    ///
    /// This method:
    /// 1. Stores the artifact in the artifact store
    /// 2. Emits an ArtifactCreated event for audit trail
    /// 3. Returns the artifact ID
    ///
    /// ## Arguments
    /// * `step_id` - The step that produced this artifact
    /// * `name` - Name of the artifact
    /// * `artifact_type` - Type of artifact
    /// * `content` - Raw content bytes
    ///
    /// ## Returns
    /// The generated ArtifactId, or None if no artifact store is configured
    pub async fn store_artifact(
        &self,
        step_id: &StepId,
        name: impl Into<String>,
        artifact_type: ArtifactType,
        content: Vec<u8>,
    ) -> Option<ArtifactId> {
        let store = self.artifact_store.as_ref()?;

        let request = PutArtifactRequest::new(
            self.execution.id.clone(),
            step_id.clone(),
            name,
            artifact_type,
            content,
        );

        match store.put(request).await {
            Ok(response) => {
                // Emit ArtifactCreated event for audit trail
                let artifact_type_str = format!("{:?}", artifact_type);
                self.emitter.emit(StreamEvent::artifact_created(
                    &self.execution.id,
                    step_id,
                    &response.artifact_id,
                    artifact_type_str,
                ));

                tracing::debug!(
                    execution_id = %self.execution.id,
                    step_id = %step_id,
                    artifact_id = %response.artifact_id,
                    "Artifact stored"
                );

                Some(response.artifact_id)
            }
            Err(e) => {
                tracing::warn!(
                    execution_id = %self.execution.id,
                    step_id = %step_id,
                    error = %e,
                    "Failed to store artifact"
                );
                None
            }
        }
    }

    /// Store a text artifact (convenience method)
    pub async fn store_text_artifact(
        &self,
        step_id: &StepId,
        name: impl Into<String>,
        content: impl Into<String>,
    ) -> Option<ArtifactId> {
        self.store_artifact(
            step_id,
            name,
            ArtifactType::Text,
            content.into().into_bytes(),
        )
        .await
    }

    /// Store a JSON artifact (convenience method)
    pub async fn store_json_artifact(
        &self,
        step_id: &StepId,
        name: impl Into<String>,
        value: &serde_json::Value,
    ) -> Option<ArtifactId> {
        let content = serde_json::to_vec_pretty(value).ok()?;
        self.store_artifact(step_id, name, ArtifactType::Json, content)
            .await
    }

    /// Get the tenant context
    pub fn tenant_context(&self) -> &TenantContext {
        &self.tenant_context
    }

    /// Get the execution ID
    pub fn execution_id(&self) -> &ExecutionId {
        &self.execution.id
    }

    /// Get the current execution state
    pub fn state(&self) -> ExecutionState {
        self.execution.state
    }

    /// Get the event emitter
    pub fn emitter(&self) -> &EventEmitter {
        &self.emitter
    }

    /// Get the execution reference
    pub fn execution(&self) -> &Execution {
        &self.execution
    }

    /// Check if cancelled
    ///
    /// Uses CancellationToken for proper async cancellation support.
    pub fn is_cancelled(&self) -> bool {
        self.cancellation_token.is_cancelled()
    }

    /// Cancel the execution
    ///
    /// This triggers cooperative cancellation of all async operations.
    /// The actual state transition happens through dispatch.
    pub fn cancel(&self, _reason: impl Into<String>) {
        self.cancellation_token.cancel();
        // Note: actual state transition happens through dispatch
    }

    /// Get a child cancellation token
    ///
    /// Child tokens are cancelled when the parent is cancelled,
    /// but cancelling a child doesn't affect the parent.
    pub fn child_cancellation_token(&self) -> CancellationToken {
        self.cancellation_token.child_token()
    }

    /// Get the cancellation token for use in async operations
    ///
    /// Use this with `tokio::select!` to make async operations cancellable:
    /// ```ignore
    /// tokio::select! {
    ///     _ = token.cancelled() => { /* handle cancellation */ }
    ///     result = some_async_operation() => { /* handle result */ }
    /// }
    /// ```
    pub fn cancellation_token(&self) -> &CancellationToken {
        &self.cancellation_token
    }

    /// Dispatch an action to the reducer
    ///
    /// This is the ONLY way to change execution state.
    pub fn dispatch(&mut self, action: ExecutionAction) -> Result<(), ReducerError> {
        // Apply through reducer
        reduce(&mut self.execution, action.clone())?;

        // Emit corresponding event
        self.emit_event_for_action(&action);
        // Persist mutable snapshot cache (best-effort)
        self.persist_snapshot_best_effort();
        // Emit non-authoritative lifecycle signal (best-effort)
        self.emit_signal_best_effort(&action);

        Ok(())
    }

    /// Start execution
    pub fn start(&mut self) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::Start)
    }

    /// Begin a step
    pub fn begin_step(
        &mut self,
        step_type: StepType,
        name: impl Into<String>,
        parent_step_id: Option<StepId>,
    ) -> Result<StepId, ReducerError> {
        let step_id = StepId::new();
        self.dispatch(ExecutionAction::StepStarted {
            step_id: step_id.clone(),
            parent_step_id,
            step_type,
            name: name.into(),
            source: None,
        })?;
        Ok(step_id)
    }

    /// Complete a step
    pub fn complete_step(
        &mut self,
        step_id: StepId,
        output: Option<String>,
        duration_ms: u64,
    ) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::StepCompleted {
            step_id,
            output,
            duration_ms,
        })
    }

    /// Fail a step with a structured error (feat-02)
    pub fn fail_step(
        &mut self,
        step_id: StepId,
        error: ExecutionError,
    ) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::StepFailed { step_id, error })
    }

    /// Fail a step with a simple message (creates a KernelInternal error)
    pub fn fail_step_with_message(
        &mut self,
        step_id: StepId,
        message: impl Into<String>,
    ) -> Result<(), ReducerError> {
        self.fail_step(step_id, ExecutionError::kernel_internal(message))
    }

    /// Pause execution
    pub fn pause(&mut self, reason: impl Into<String>) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::Pause {
            reason: reason.into(),
        })
    }

    /// Resume execution
    pub fn resume(&mut self) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::Resume)
    }

    /// Enter waiting state
    pub fn wait_for(&mut self, reason: WaitReason) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::Wait { reason })
    }

    /// Signal that external input was received
    pub fn input_received(&mut self) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::InputReceived)
    }

    /// Complete execution
    pub fn complete(&mut self, output: Option<String>) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::Complete { output })
    }

    /// Fail execution with a structured error (feat-02)
    pub fn fail(&mut self, error: ExecutionError) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::Fail { error })
    }

    /// Fail execution with a simple message (creates a KernelInternal error)
    pub fn fail_with_message(&mut self, message: impl Into<String>) -> Result<(), ReducerError> {
        self.fail(ExecutionError::kernel_internal(message))
    }

    /// Cancel execution
    pub fn cancel_execution(&mut self, reason: impl Into<String>) -> Result<(), ReducerError> {
        self.dispatch(ExecutionAction::Cancel {
            reason: reason.into(),
        })
    }

    /// Execute a compiled graph
    ///
    /// ## Invariants
    /// - INV-INBOX-001: Inbox is checked after every step
    /// - INV-INBOX-002: Control messages are processed first (via priority_order)
    /// - INV-INBOX-003: Inbox events are emitted for audit trail
    ///
    /// ## Async Cancellation
    /// Uses tokio::select! with CancellationToken for cooperative cancellation.
    /// Node execution can be interrupted cleanly if cancellation is requested.
    pub async fn execute_graph(
        &mut self,
        graph: &CompiledGraph,
        input: &str,
    ) -> anyhow::Result<NodeState> {
        // Start execution
        self.start()?;

        let mut state = NodeState::from_string(input);
        let mut current_node = graph.entry_point().to_string();

        // Get cancellation token for use in select!
        let cancel_token = self.cancellation_token.clone();

        loop {
            // Check for cancellation before each step
            if self.is_cancelled() {
                self.cancel_execution("Cancelled by user")?;
                anyhow::bail!("Execution cancelled");
            }

            // Get the node
            let node = graph
                .get_node(&current_node)
                .ok_or_else(|| anyhow::anyhow!("Node '{}' not found", current_node))?;

            // Begin step
            let step_start = Instant::now();
            let step_id = self.begin_step(StepType::FunctionNode, current_node.clone(), None)?;

            // Execute node with cancellation support using tokio::select!
            // This allows long-running operations to be interrupted
            let node_future = node.execute(state.clone());
            let result = tokio::select! {
                biased;  // Check cancellation first

                _ = cancel_token.cancelled() => {
                    // Cancellation requested during node execution
                    let error = ExecutionError::kernel_internal("Cancelled during step execution")
                        .with_step_id(step_id.clone());
                    self.fail_step(step_id, error)?;
                    self.cancel_execution("Cancelled during step execution")?;
                    anyhow::bail!("Execution cancelled during step");
                }
                result = node_future => result,
            };

            let duration_ms = step_start.elapsed().as_millis() as u64;

            match result {
                Ok(new_state) => {
                    state = new_state;
                    self.complete_step(
                        step_id,
                        Some(state.as_str().unwrap_or_default().to_string()),
                        duration_ms,
                    )?;
                }
                Err(e) => {
                    let error = ExecutionError::kernel_internal(e.to_string())
                        .with_step_id(step_id.clone());
                    self.fail_step(step_id.clone(), error.clone())?;
                    self.fail(error)?;
                    return Err(e);
                }
            }

            // INV-INBOX-001: Check inbox after every step
            if let Some(action) = self.check_inbox()? {
                match action {
                    InboxAction::Pause => {
                        // Pause has already been applied via dispatch in check_inbox()
                        // In a full implementation, we would suspend here and wait for resume
                        // For now, we log and continue - the state machine is already in Paused
                        tracing::info!(
                            execution_id = %self.execution.id,
                            "Execution paused via inbox, continuing in paused state"
                        );
                    }
                    InboxAction::Cancel(reason) => {
                        self.cancel_execution(&reason)?;
                        anyhow::bail!("Execution cancelled via inbox: {}", reason);
                    }
                    InboxAction::Continue => {
                        // Continue execution normally
                    }
                }
            }

            // Get next nodes
            let output = state.as_str().unwrap_or_default();
            let next = graph.get_next(&current_node, output);

            if next.is_empty() {
                break;
            }

            match &next[0] {
                crate::graph::EdgeTarget::End => break,
                crate::graph::EdgeTarget::Node(n) => {
                    current_node = n.clone();
                }
            }
        }

        // Complete execution
        self.complete(Some(state.as_str().unwrap_or_default().to_string()))?;

        Ok(state)
    }

    /// Check inbox for messages and process them
    ///
    /// ## Invariants
    /// - INV-INBOX-001: Called after every step
    /// - INV-INBOX-002: Control messages processed first (via drain_messages sorting)
    /// - INV-INBOX-003: All messages emit events for audit trail
    /// - INV-SPAWN-002: Inbox inheritance based on SpawnMode
    ///
    /// ## SpawnMode Routing (@see docs/TECHNICAL/32-SPAWN-MODE.md)
    ///
    /// - Inline mode (default): Check current execution's inbox
    /// - Child { inherit_inbox: true }: Check both parent and own inbox
    /// - Child { inherit_inbox: false }: Check only own inbox (isolated)
    fn check_inbox(&mut self) -> Result<Option<InboxAction>, ReducerError> {
        let inbox = match &self.inbox {
            Some(inbox) => inbox.clone(),
            None => return Ok(None),
        };

        // Determine which execution IDs to check based on SpawnMode
        let execution_ids_to_check = self.get_inbox_execution_ids();

        // Fast path: check if any inbox has messages
        let has_messages = execution_ids_to_check
            .iter()
            .any(|id| inbox.has_control_messages(id) || !inbox.is_empty(id));

        if !has_messages {
            return Ok(None);
        }

        // Drain messages from all applicable inboxes, sorted by priority (INV-INBOX-002)
        let messages: Vec<InboxMessage> = execution_ids_to_check
            .iter()
            .flat_map(|id| inbox.drain_messages(id))
            .collect();

        let mut action = InboxAction::Continue;

        for message in messages {
            // INV-INBOX-003: Emit event for audit trail
            self.emit_inbox_event(&message);

            match message {
                InboxMessage::Control(ctrl) => {
                    match ctrl.action {
                        ControlAction::Pause => {
                            tracing::info!(
                                execution_id = %self.execution.id,
                                actor = %ctrl.actor,
                                reason = ?ctrl.reason,
                                "Inbox: Pause requested"
                            );
                            self.pause(
                                ctrl.reason
                                    .unwrap_or_else(|| "Paused via inbox".to_string()),
                            )?;
                            action = InboxAction::Pause;
                        }
                        ControlAction::Resume => {
                            tracing::info!(
                                execution_id = %self.execution.id,
                                actor = %ctrl.actor,
                                "Inbox: Resume requested"
                            );
                            self.resume()?;
                            action = InboxAction::Continue;
                        }
                        ControlAction::Cancel => {
                            let reason = ctrl
                                .reason
                                .unwrap_or_else(|| "Cancelled via inbox".to_string());
                            tracing::info!(
                                execution_id = %self.execution.id,
                                actor = %ctrl.actor,
                                reason = %reason,
                                "Inbox: Cancel requested"
                            );
                            return Ok(Some(InboxAction::Cancel(reason)));
                        }
                        ControlAction::Checkpoint => {
                            tracing::info!(
                                execution_id = %self.execution.id,
                                "Inbox: Checkpoint requested"
                            );
                            // Checkpoint is handled by the runner, not the kernel
                        }
                        ControlAction::Compact => {
                            tracing::info!(
                                execution_id = %self.execution.id,
                                "Inbox: Compact requested"
                            );
                            // Compact is handled by the runner, not the kernel
                        }
                    }
                }
                InboxMessage::Guidance(guidance) => {
                    tracing::info!(
                        execution_id = %self.execution.id,
                        from = ?guidance.from,
                        priority = ?guidance.priority,
                        content = %guidance.content,
                        "Inbox: Guidance received"
                    );
                    // Guidance is logged but not acted upon in the kernel
                    // The agent/LLM layer processes guidance
                }
                InboxMessage::Evidence(evidence) => {
                    tracing::info!(
                        execution_id = %self.execution.id,
                        source = ?evidence.source,
                        impact = ?evidence.impact,
                        title = %evidence.title,
                        "Inbox: Evidence received"
                    );
                    // Evidence is logged but not acted upon in the kernel
                    // The agent/LLM layer processes evidence
                }
                InboxMessage::A2a(a2a) => {
                    tracing::debug!(
                        execution_id = %self.execution.id,
                        from_agent = %a2a.from_agent,
                        message_type = %a2a.message_type,
                        "Inbox: A2A message received"
                    );
                    // A2A messages are logged but not acted upon in the kernel
                    // The agent/LLM layer processes A2A messages
                }
            }
        }

        Ok(Some(action))
    }

    /// Emit an event for an inbox message (INV-INBOX-003)
    fn emit_inbox_event(&self, message: &InboxMessage) {
        let event =
            StreamEvent::inbox_message(&self.execution.id, message.id(), message.message_type());
        self.emitter.emit(event);
    }

    /// Determine which execution IDs to check for inbox messages based on SpawnMode
    ///
    /// ## SpawnMode Routing Rules (INV-SPAWN-002)
    ///
    /// - Inline mode: Check current execution's inbox (same as default)
    /// - Child { inherit_inbox: true }: Check both parent and own inbox
    /// - Child { inherit_inbox: false }: Check only own inbox (isolated)
    /// - No spawn_mode: Default to current execution only
    ///
    /// @see docs/TECHNICAL/32-SPAWN-MODE.md
    #[cfg_attr(test, allow(dead_code))]
    pub(crate) fn get_inbox_execution_ids(&self) -> Vec<ExecutionId> {
        match &self.spawn_mode {
            // Inline mode: use current execution only (inline shares parent's ExecutionId anyway)
            Some(SpawnMode::Inline) => {
                vec![self.execution.id.clone()]
            }
            // Child mode with inherit_inbox: check both parent and own inbox
            Some(SpawnMode::Child {
                inherit_inbox: true,
                ..
            }) => {
                let mut ids = vec![self.execution.id.clone()];
                if let Some(parent_id) = &self.parent_execution_id {
                    ids.push(parent_id.clone());
                    tracing::debug!(
                        execution_id = %self.execution.id,
                        parent_id = %parent_id,
                        "Checking both parent and own inbox (inherit_inbox=true)"
                    );
                }
                ids
            }
            // Child mode without inherit_inbox: isolated inbox
            Some(SpawnMode::Child {
                inherit_inbox: false,
                ..
            }) => {
                tracing::debug!(
                    execution_id = %self.execution.id,
                    "Using isolated inbox (inherit_inbox=false)"
                );
                vec![self.execution.id.clone()]
            }
            // No spawn mode set: default to current execution
            None => {
                vec![self.execution.id.clone()]
            }
        }
    }

    /// Emit a stream event corresponding to an action
    fn emit_event_for_action(&self, action: &ExecutionAction) {
        let event = match action {
            ExecutionAction::Start => StreamEvent::execution_start(&self.execution.id),
            ExecutionAction::StepStarted {
                step_id,
                step_type,
                name,
                ..
            } => StreamEvent::step_start(
                &self.execution.id,
                step_id,
                step_type.clone(),
                name.clone(),
            ),
            ExecutionAction::StepCompleted {
                step_id,
                output,
                duration_ms,
            } => StreamEvent::step_end(&self.execution.id, step_id, output.clone(), *duration_ms),
            ExecutionAction::StepFailed { step_id, error } => {
                StreamEvent::step_failed(&self.execution.id, step_id, error.clone())
            }
            ExecutionAction::Pause { reason } => {
                StreamEvent::execution_paused(&self.execution.id, reason.clone())
            }
            ExecutionAction::Resume => StreamEvent::execution_resumed(&self.execution.id),
            ExecutionAction::Complete { output } => {
                let duration = self.execution.duration_ms().unwrap_or(0);
                StreamEvent::execution_end(&self.execution.id, output.clone(), duration)
            }
            ExecutionAction::Fail { error } => {
                StreamEvent::execution_failed(&self.execution.id, error.clone())
            }
            ExecutionAction::Cancel { reason } => {
                StreamEvent::execution_cancelled(&self.execution.id, reason.clone())
            }
            ExecutionAction::Wait { .. } | ExecutionAction::InputReceived => {
                // No specific stream events for these yet
                return;
            }
        };

        self.emitter.emit(event);
    }

    fn persist_snapshot_best_effort(&self) {
        let Some(store) = self.state_store.as_ref() else {
            return;
        };

        let current_step_id = self.execution.step_order.iter().rev().find_map(|id| {
            self.execution
                .steps
                .get(id)
                .and_then(|s| (s.state == StepState::Running).then_some(id.clone()))
        });

        let step_outputs = self
            .execution
            .steps
            .iter()
            .filter_map(|(step_id, step)| {
                step.output
                    .as_ref()
                    .map(|output| (step_id.clone(), serde_json::Value::String(output.clone())))
            })
            .collect();

        let mut snapshot = ExecutionSnapshot::with_user(
            self.execution.id.clone(),
            self.tenant_context.tenant_id.clone(),
            self.tenant_context.user_id.clone(),
            self.execution.state,
            self.execution.step_order.len() as u64,
        );
        snapshot.current_step_id = current_step_id;
        snapshot.step_outputs = step_outputs;

        let store = Arc::clone(store);
        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            handle.spawn(async move {
                if let Err(e) = store.save_snapshot(snapshot).await {
                    tracing::debug!("State snapshot persistence failed: {}", e);
                }
            });
        }
    }

    fn emit_signal_best_effort(&self, action: &ExecutionAction) {
        let Some(bus) = self.signal_bus.as_ref() else {
            return;
        };

        let action_name = match action {
            ExecutionAction::Start => "start",
            ExecutionAction::StepStarted { .. } => "step_started",
            ExecutionAction::StepCompleted { .. } => "step_completed",
            ExecutionAction::StepFailed { .. } => "step_failed",
            ExecutionAction::Pause { .. } => "paused",
            ExecutionAction::Resume => "resumed",
            ExecutionAction::Complete { .. } => "completed",
            ExecutionAction::Fail { .. } => "failed",
            ExecutionAction::Cancel { .. } => "cancelled",
            ExecutionAction::Wait { .. } => "waiting",
            ExecutionAction::InputReceived => "input_received",
        };

        let signal = serde_json::json!({
            "execution_id": self.execution.id.to_string(),
            "tenant_id": self.tenant_context.tenant_id.to_string(),
            "action": action_name,
            "state": format!("{:?}", self.execution.state),
        });

        let signal_bytes = match serde_json::to_vec(&signal) {
            Ok(bytes) => bytes,
            Err(_) => return,
        };

        let bus = Arc::clone(bus);
        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            handle.spawn(async move {
                if let Err(e) = bus.emit("execution.lifecycle", &signal_bytes).await {
                    tracing::debug!("Signal emit failed: {}", e);
                }
            });
        }
    }

    // =========================================================================
    // Protected Event Emission (P2 #2: ProtectedEventEmitter Integration)
    // =========================================================================

    /// Emit an event with protection processing
    ///
    /// If a protected emitter is configured, the event passes through the
    /// protection pipeline before being emitted. Otherwise, falls back to
    /// the regular emitter.
    ///
    /// Use this for events that may contain sensitive content (step outputs,
    /// tool results, etc.).
    pub async fn emit_protected(&self, event: StreamEvent) -> anyhow::Result<()> {
        if let Some(protected) = &self.protected_emitter {
            protected.emit(event).await?;
        } else {
            self.emitter.emit(event);
        }
        Ok(())
    }

    /// Emit an event without protection (control events, etc.)
    ///
    /// Use for events that are guaranteed safe (control signals, execution
    /// lifecycle events without content).
    pub fn emit_unprotected(&self, event: StreamEvent) {
        if let Some(protected) = &self.protected_emitter {
            protected.emit_unprotected(event);
        } else {
            self.emitter.emit(event);
        }
    }

    /// Check if protected emitter is configured
    pub fn has_protected_emitter(&self) -> bool {
        self.protected_emitter.is_some()
    }

    /// Get the protected emitter (if configured)
    pub fn protected_emitter(&self) -> Option<&ProtectedEventEmitter> {
        self.protected_emitter.as_ref()
    }
}

// Note: ExecutionKernel does NOT implement Default because TenantContext is REQUIRED.
// This is intentional - every execution must have a tenant for multi-tenant isolation.

#[cfg(test)]
mod tests {
    use super::*;
    use crate::context::ResourceLimits;
    use crate::TenantId;

    #[tokio::test]
    async fn emits_warning_event_when_limits_near_threshold() {
        let limits = ResourceLimits {
            max_steps: 5,
            ..Default::default()
        };
        let tenant = TenantContext::new(TenantId::new()).with_limits(limits);

        let mut kernel = ExecutionKernel::new(tenant);
        kernel.register_for_enforcement().await;

        // Record progress to reach warning threshold (80% at next step)
        kernel.record_step_completed().await;
        kernel.record_step_completed().await;
        kernel.record_step_completed().await;

        kernel.check_limits_before_step().await.unwrap();

        let events = kernel.emitter.drain();
        assert!(
            events.iter().any(|e| {
                matches!(
                    e,
                    StreamEvent::PolicyDecision {
                        decision,
                        tool_name,
                        ..
                    } if decision == "warn" && tool_name == "enforcement"
                )
            }),
            "expected enforcement warning event"
        );
    }

    // =========================================================================
    // SpawnMode Builder Tests
    // =========================================================================

    #[test]
    fn test_kernel_with_spawn_mode_inline() {
        let tenant = TenantContext::new(TenantId::new());
        let kernel = ExecutionKernel::new(tenant).with_spawn_mode(SpawnMode::Inline);

        assert!(kernel.spawn_mode().is_some());
        assert_eq!(*kernel.spawn_mode().unwrap(), SpawnMode::Inline);
    }

    #[test]
    fn test_kernel_with_spawn_mode_child() {
        let tenant = TenantContext::new(TenantId::new());
        let kernel = ExecutionKernel::new(tenant).with_spawn_mode(SpawnMode::Child {
            background: true,
            inherit_inbox: true,
            policies: None,
        });

        assert!(kernel.spawn_mode().is_some());
        if let Some(SpawnMode::Child {
            background,
            inherit_inbox,
            ..
        }) = kernel.spawn_mode()
        {
            assert!(*background);
            assert!(*inherit_inbox);
        } else {
            panic!("Expected SpawnMode::Child");
        }
    }

    #[test]
    fn test_kernel_with_parent_execution_id() {
        let tenant = TenantContext::new(TenantId::new());
        let parent_id = ExecutionId::from_string("exec_parent_123");
        let kernel = ExecutionKernel::new(tenant).with_parent_execution_id(parent_id.clone());

        assert!(kernel.parent_execution_id().is_some());
        assert_eq!(*kernel.parent_execution_id().unwrap(), parent_id);
    }

    #[test]
    fn test_kernel_default_no_spawn_mode() {
        let tenant = TenantContext::new(TenantId::new());
        let kernel = ExecutionKernel::new(tenant);

        assert!(kernel.spawn_mode().is_none());
        assert!(kernel.parent_execution_id().is_none());
    }

    // =========================================================================
    // Inbox Routing by SpawnMode Tests
    // =========================================================================

    #[test]
    fn test_get_inbox_execution_ids_no_spawn_mode() {
        let tenant = TenantContext::new(TenantId::new());
        let kernel = ExecutionKernel::new(tenant);

        let ids = kernel.get_inbox_execution_ids();
        assert_eq!(ids.len(), 1, "Should return only current execution ID");
        assert_eq!(ids[0], *kernel.execution_id());
    }

    #[test]
    fn test_get_inbox_execution_ids_inline_mode() {
        let tenant = TenantContext::new(TenantId::new());
        let kernel = ExecutionKernel::new(tenant).with_spawn_mode(SpawnMode::Inline);

        let ids = kernel.get_inbox_execution_ids();
        assert_eq!(
            ids.len(),
            1,
            "Inline mode should check current execution only"
        );
        assert_eq!(ids[0], *kernel.execution_id());
    }

    #[test]
    fn test_get_inbox_execution_ids_child_isolated() {
        let tenant = TenantContext::new(TenantId::new());
        let parent_id = ExecutionId::from_string("exec_parent");
        let kernel = ExecutionKernel::new(tenant)
            .with_spawn_mode(SpawnMode::Child {
                background: false,
                inherit_inbox: false,
                policies: None,
            })
            .with_parent_execution_id(parent_id);

        let ids = kernel.get_inbox_execution_ids();
        assert_eq!(
            ids.len(),
            1,
            "Child with inherit_inbox=false should be isolated"
        );
        assert_eq!(
            ids[0],
            *kernel.execution_id(),
            "Should only check own inbox"
        );
    }

    #[test]
    fn test_get_inbox_execution_ids_child_inherit() {
        let tenant = TenantContext::new(TenantId::new());
        let parent_id = ExecutionId::from_string("exec_parent_inherit");
        let kernel = ExecutionKernel::new(tenant)
            .with_spawn_mode(SpawnMode::Child {
                background: false,
                inherit_inbox: true,
                policies: None,
            })
            .with_parent_execution_id(parent_id.clone());

        let ids = kernel.get_inbox_execution_ids();
        assert_eq!(
            ids.len(),
            2,
            "Child with inherit_inbox=true should check both inboxes"
        );
        assert!(
            ids.contains(kernel.execution_id()),
            "Should include own execution ID"
        );
        assert!(
            ids.contains(&parent_id),
            "Should include parent execution ID"
        );
    }

    #[test]
    fn test_get_inbox_execution_ids_child_inherit_no_parent_id() {
        let tenant = TenantContext::new(TenantId::new());
        // Child with inherit_inbox=true but no parent_execution_id set
        let kernel = ExecutionKernel::new(tenant).with_spawn_mode(SpawnMode::Child {
            background: false,
            inherit_inbox: true,
            policies: None,
        });

        let ids = kernel.get_inbox_execution_ids();
        // Should gracefully handle missing parent_execution_id
        assert_eq!(
            ids.len(),
            1,
            "Without parent_execution_id, should only return own ID"
        );
        assert_eq!(ids[0], *kernel.execution_id());
    }

    #[test]
    fn test_get_inbox_execution_ids_child_background_isolated() {
        let tenant = TenantContext::new(TenantId::new());
        let kernel = ExecutionKernel::new(tenant).with_spawn_mode(SpawnMode::Child {
            background: true,     // Background child
            inherit_inbox: false, // Isolated
            policies: None,
        });

        let ids = kernel.get_inbox_execution_ids();
        assert_eq!(ids.len(), 1, "Background child with isolated inbox");
    }

    #[test]
    fn test_get_inbox_execution_ids_child_background_inherit() {
        let tenant = TenantContext::new(TenantId::new());
        let parent_id = ExecutionId::from_string("exec_background_parent");
        let kernel = ExecutionKernel::new(tenant)
            .with_spawn_mode(SpawnMode::Child {
                background: true,    // Background child
                inherit_inbox: true, // Inherits parent inbox
                policies: None,
            })
            .with_parent_execution_id(parent_id.clone());

        let ids = kernel.get_inbox_execution_ids();
        assert_eq!(ids.len(), 2, "Background child can still inherit inbox");
        assert!(ids.contains(&parent_id));
    }
}