a3s-code-core 8.2.0

//! Typed, cooperative control for an in-flight Code run.
//!
//! A run is deliberately controlled through a small per-run inbox instead of
//! mutating loop state from the host thread.  This keeps the execution loop as
//! the sole owner of its transcript while still allowing an embedding host to
//! steer or interrupt work at well-defined safe points.  The wire types in
//! this module are also used by the language SDKs and can be persisted by a
//! host without depending on Rust internals.

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use thiserror::Error;
use tokio::sync::{Mutex, Notify};
use tokio_util::sync::CancellationToken;

use crate::hooks::{HookExecutor, HookOutcome};

/// Schema carried by a run-control request.
pub const RUN_CONTROL_REQUEST_SCHEMA_V1: &str = "a3s.code.run-control-request.v1";
/// Schema carried by a run-control receipt.
pub const RUN_CONTROL_RECEIPT_SCHEMA_V1: &str = "a3s.code.run-control-receipt.v1";
/// Maximum UTF-8 bytes accepted for a single steer message.
pub const RUN_CONTROL_MAX_INPUT_BYTES: usize = 128 * 1024;
/// Maximum UTF-8 bytes accepted for an interrupt reason.
pub const RUN_CONTROL_MAX_REASON_BYTES: usize = 4 * 1024;
/// Maximum number of controls waiting at one run safe point.
pub const RUN_CONTROL_MAX_QUEUE: usize = 64;
/// Number of request receipts retained for idempotent retries.
pub const RUN_CONTROL_MAX_SEEN_REQUESTS: usize = 256;
/// Maximum size of an externally supplied identifier.
pub const RUN_CONTROL_MAX_ID_BYTES: usize = 512;

fn default_request_schema() -> String {
    RUN_CONTROL_REQUEST_SCHEMA_V1.to_string()
}

fn default_receipt_schema() -> String {
    RUN_CONTROL_RECEIPT_SCHEMA_V1.to_string()
}

/// The operation requested by a host.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunControlOperation {
    /// Append a user-directed steering message at the next loop safe point.
    Steer,
    /// Cooperatively stop the active run.
    Interrupt,
}

/// A typed control command.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum RunControlCommand {
    /// Add a new user direction without starting a second turn.
    Steer { input: String },
    /// Stop the current run. `force` is advisory: the runtime remains
    /// cooperative and will never skip cleanup or governance boundaries.
    Interrupt {
        reason: Option<String>,
        #[serde(default)]
        force: bool,
    },
}

impl RunControlCommand {
    pub fn operation(&self) -> RunControlOperation {
        match self {
            Self::Steer { .. } => RunControlOperation::Steer,
            Self::Interrupt { .. } => RunControlOperation::Interrupt,
        }
    }
}

/// Versioned request accepted by a run-control inbox.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RunControlRequest {
    /// Protocol schema identifier.
    #[serde(default = "default_request_schema")]
    pub schema: String,
    /// Idempotency key generated by the host.
    pub request_id: String,
    /// Optional session binding. When present it must match the target run.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// Immutable target run id.
    pub run_id: String,
    /// Expected logical turn. A mismatch is rejected as stale input.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_turn_id: Option<String>,
    /// Monotonic control revision observed by the host.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_turn_revision: Option<u64>,
    /// Control payload.
    pub command: RunControlCommand,
    /// Optional host deadline in Unix milliseconds.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub deadline_ms: Option<u64>,
}

impl RunControlRequest {
    /// Build a request with a fresh id. The target run is filled by the
    /// session convenience APIs when omitted here.
    pub fn new(run_id: impl Into<String>, command: RunControlCommand) -> Self {
        Self {
            schema: RUN_CONTROL_REQUEST_SCHEMA_V1.to_string(),
            request_id: uuid::Uuid::new_v4().to_string(),
            session_id: None,
            run_id: run_id.into(),
            expected_turn_id: None,
            expected_turn_revision: None,
            command,
            deadline_ms: None,
        }
    }

    pub fn steer(run_id: impl Into<String>, input: impl Into<String>) -> Self {
        Self::new(
            run_id,
            RunControlCommand::Steer {
                input: input.into(),
            },
        )
    }

    pub fn interrupt(run_id: impl Into<String>) -> Self {
        Self::new(
            run_id,
            RunControlCommand::Interrupt {
                reason: None,
                force: false,
            },
        )
    }

    pub fn with_session_id(mut self, session_id: impl Into<String>) -> Self {
        self.session_id = Some(session_id.into());
        self
    }

    pub fn with_expected_turn(mut self, turn_id: impl Into<String>, revision: u64) -> Self {
        self.expected_turn_id = Some(turn_id.into());
        self.expected_turn_revision = Some(revision);
        self
    }

    pub fn with_deadline_ms(mut self, deadline_ms: u64) -> Self {
        self.deadline_ms = Some(deadline_ms);
        self
    }

    /// Validate bounded input and protocol identity before enqueueing.
    pub fn validate(&self) -> Result<(), RunControlError> {
        if self.schema != RUN_CONTROL_REQUEST_SCHEMA_V1 {
            return Err(RunControlError::InvalidRequest(format!(
                "unsupported schema `{}`",
                self.schema
            )));
        }
        validate_id("request_id", &self.request_id)?;
        validate_id("run_id", &self.run_id)?;
        if let Some(session_id) = &self.session_id {
            validate_id("session_id", session_id)?;
        }
        if let Some(turn_id) = &self.expected_turn_id {
            validate_id("expected_turn_id", turn_id)?;
        }
        match &self.command {
            RunControlCommand::Steer { input } => {
                if input.trim().is_empty() {
                    return Err(RunControlError::InvalidRequest(
                        "steer input must not be empty".to_string(),
                    ));
                }
                if input.len() > RUN_CONTROL_MAX_INPUT_BYTES {
                    return Err(RunControlError::InvalidRequest(format!(
                        "steer input exceeds {} bytes",
                        RUN_CONTROL_MAX_INPUT_BYTES
                    )));
                }
            }
            RunControlCommand::Interrupt { reason, .. } => {
                if let Some(reason) = reason {
                    if reason.len() > RUN_CONTROL_MAX_REASON_BYTES {
                        return Err(RunControlError::InvalidRequest(format!(
                            "interrupt reason exceeds {} bytes",
                            RUN_CONTROL_MAX_REASON_BYTES
                        )));
                    }
                }
            }
        }
        Ok(())
    }
}

fn validate_id(name: &str, value: &str) -> Result<(), RunControlError> {
    if value.trim().is_empty() {
        return Err(RunControlError::InvalidRequest(format!(
            "{name} must not be empty"
        )));
    }
    if value.len() > RUN_CONTROL_MAX_ID_BYTES {
        return Err(RunControlError::InvalidRequest(format!(
            "{name} exceeds {RUN_CONTROL_MAX_ID_BYTES} bytes"
        )));
    }
    if value
        .chars()
        .any(|character| character == '\0' || character == '\r' || character == '\n')
    {
        return Err(RunControlError::InvalidRequest(format!(
            "{name} contains a control character"
        )));
    }
    Ok(())
}

/// Convenience input for [`crate::AgentSession::steer`](crate::AgentSession::steer).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SteerRequest {
    pub input: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub request_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub run_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_turn_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_turn_revision: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub deadline_ms: Option<u64>,
}

impl SteerRequest {
    pub fn new(input: impl Into<String>) -> Self {
        Self {
            input: input.into(),
            request_id: None,
            run_id: None,
            expected_turn_id: None,
            expected_turn_revision: None,
            deadline_ms: None,
        }
    }

    pub fn with_run_id(mut self, run_id: impl Into<String>) -> Self {
        self.run_id = Some(run_id.into());
        self
    }

    pub fn with_expected_turn(mut self, turn_id: impl Into<String>, revision: u64) -> Self {
        self.expected_turn_id = Some(turn_id.into());
        self.expected_turn_revision = Some(revision);
        self
    }

    pub(crate) fn into_protocol(self, session_id: &str, active_run_id: &str) -> RunControlRequest {
        RunControlRequest {
            schema: RUN_CONTROL_REQUEST_SCHEMA_V1.to_string(),
            request_id: self
                .request_id
                .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
            session_id: Some(session_id.to_string()),
            run_id: self.run_id.unwrap_or_else(|| active_run_id.to_string()),
            expected_turn_id: self.expected_turn_id,
            expected_turn_revision: self.expected_turn_revision,
            command: RunControlCommand::Steer { input: self.input },
            deadline_ms: self.deadline_ms,
        }
    }
}

/// Convenience input for [`crate::AgentSession::interrupt`](crate::AgentSession::interrupt).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InterruptRequest {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    #[serde(default)]
    pub force: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub request_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub run_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_turn_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expected_turn_revision: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub deadline_ms: Option<u64>,
}

impl InterruptRequest {
    pub fn new() -> Self {
        Self {
            reason: None,
            force: false,
            request_id: None,
            run_id: None,
            expected_turn_id: None,
            expected_turn_revision: None,
            deadline_ms: None,
        }
    }

    pub fn with_reason(mut self, reason: impl Into<String>) -> Self {
        self.reason = Some(reason.into());
        self
    }

    pub fn with_run_id(mut self, run_id: impl Into<String>) -> Self {
        self.run_id = Some(run_id.into());
        self
    }

    pub fn with_expected_turn(mut self, turn_id: impl Into<String>, revision: u64) -> Self {
        self.expected_turn_id = Some(turn_id.into());
        self.expected_turn_revision = Some(revision);
        self
    }

    pub(crate) fn into_protocol(self, session_id: &str, active_run_id: &str) -> RunControlRequest {
        RunControlRequest {
            schema: RUN_CONTROL_REQUEST_SCHEMA_V1.to_string(),
            request_id: self
                .request_id
                .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
            session_id: Some(session_id.to_string()),
            run_id: self.run_id.unwrap_or_else(|| active_run_id.to_string()),
            expected_turn_id: self.expected_turn_id,
            expected_turn_revision: self.expected_turn_revision,
            command: RunControlCommand::Interrupt {
                reason: self.reason,
                force: self.force,
            },
            deadline_ms: self.deadline_ms,
        }
    }
}

impl Default for InterruptRequest {
    fn default() -> Self {
        Self::new()
    }
}

/// Receipt state returned by the host boundary.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunControlReceiptState {
    /// The request passed validation and is waiting for a safe point.
    Accepted,
    /// The execution loop consumed the request.
    Applied,
    /// The request was rejected before it entered the inbox.
    Rejected,
    /// The run ended before an accepted request could be applied.
    Settled,
}

/// Machine-readable rejection/settlement detail.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunControlErrorInfo {
    pub code: String,
    pub message: String,
}

/// Durable, idempotent acknowledgement for a control request.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RunControlReceipt {
    #[serde(default = "default_receipt_schema")]
    pub schema: String,
    pub request_id: String,
    pub session_id: String,
    pub run_id: String,
    pub operation: RunControlOperation,
    pub state: RunControlReceiptState,
    pub sequence: u64,
    pub turn_id: Option<String>,
    pub turn_revision: u64,
    pub accepted_at_ms: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub applied_at_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<RunControlErrorInfo>,
}

/// Read-only state exposed to SDK/UI clients for optimistic concurrency.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunControlSnapshot {
    pub session_id: String,
    pub run_id: String,
    pub active: bool,
    pub turn_id: Option<String>,
    pub turn_revision: u64,
    pub queued_controls: usize,
    pub interrupt_requested: bool,
    pub last_sequence: u64,
}

/// Errors returned before a control is accepted.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum RunControlError {
    #[error("invalid run-control request: {0}")]
    InvalidRequest(String),
    #[error("run-control target session does not match the active session")]
    SessionMismatch,
    #[error("run-control target run `{run_id}` is not the active run")]
    RunMismatch { run_id: String },
    #[error("there is no active run to control")]
    NoActiveRun,
    #[error(
        "stale run-control request: expected turn={expected_turn_id:?}, revision={expected_revision:?}; current turn={actual_turn_id:?}, revision={actual_revision}"
    )]
    StaleTurn {
        expected_turn_id: Option<String>,
        expected_revision: Option<u64>,
        actual_turn_id: Option<String>,
        actual_revision: u64,
    },
    #[error("run-control request deadline has expired")]
    DeadlineExceeded,
    #[error("run-control inbox is full")]
    QueueFull,
    #[error("run-control inbox is closed")]
    Closed,
    #[error("request id `{request_id}` was already used for a different command")]
    DuplicateRequest { request_id: String },
    #[error("run-control request was denied by a hook: {reason}")]
    HookDenied { reason: String },
    #[error("run-control request must be retried after {retry_after_ms} ms: {reason}")]
    HookRetry { reason: String, retry_after_ms: u64 },
}

impl RunControlError {
    pub const fn code(&self) -> &'static str {
        match self {
            Self::InvalidRequest(_) => "INVALID_REQUEST",
            Self::SessionMismatch => "SESSION_MISMATCH",
            Self::RunMismatch { .. } => "RUN_MISMATCH",
            Self::NoActiveRun => "NO_ACTIVE_RUN",
            Self::StaleTurn { .. } => "STALE_TURN",
            Self::DeadlineExceeded => "DEADLINE_EXCEEDED",
            Self::QueueFull => "QUEUE_FULL",
            Self::Closed => "CLOSED",
            Self::DuplicateRequest { .. } => "DUPLICATE_REQUEST",
            Self::HookDenied { .. } => "HOOK_DENIED",
            Self::HookRetry { .. } => "HOOK_RETRY",
        }
    }
}

#[derive(Debug, Clone)]
pub(crate) struct PendingRunControl {
    pub(crate) request: RunControlRequest,
    pub(crate) receipt: RunControlReceipt,
}

#[derive(Debug)]
struct SeenRequest {
    fingerprint: String,
    /// Keep the immutable request alongside its receipt so shutdown can
    /// settle controls that have already been drained by the loop but have
    /// not reached `mark_applied` yet.  Without this, a close racing a safe
    /// point could leave an `Accepted` receipt forever and never emit the
    /// terminal post-hook event.
    request: RunControlRequest,
    receipt: RunControlReceipt,
}

#[derive(Debug)]
struct InboxState {
    session_id: String,
    run_id: String,
    active: bool,
    closed: bool,
    turn_id: Option<String>,
    turn_revision: u64,
    queue: VecDeque<PendingRunControl>,
    /// Controls removed from `queue` by the loop and currently being applied.
    /// They remain owned by the inbox until the loop acknowledges the
    /// transition, which gives close a complete set to settle.
    in_flight: HashMap<String, PendingRunControl>,
    seen: HashMap<String, SeenRequest>,
    seen_order: VecDeque<String>,
    last_sequence: u64,
    interrupt_requested: bool,
}

/// Internal per-run inbox. It is shared by the public session facade and the
/// execution loop, but only the loop consumes pending steering messages.
#[derive(Debug)]
pub(crate) struct RunControlInbox {
    run_id: String,
    state: Mutex<InboxState>,
    /// Serializes admission and receipt transitions with the loop's safe
    /// point.  Without this gate two identical concurrent requests could both
    /// execute a policy Hook, or the loop could emit `applied` before the
    /// caller-visible `accepted` observation had been recorded.
    admission: Mutex<()>,
    notify: Notify,
    cancellation: CancellationToken,
    /// Run-frozen governance executor. Keeping this on the inbox prevents a
    /// later Session hook mutation from changing the authority of an active
    /// run-control request.
    hook_executor: Option<Arc<dyn HookExecutor>>,
}

impl RunControlInbox {
    #[cfg(test)]
    pub(crate) fn new(
        session_id: impl Into<String>,
        run_id: impl Into<String>,
        cancellation: CancellationToken,
    ) -> Arc<Self> {
        Self::new_with_hook_executor(session_id, run_id, cancellation, None)
    }

    pub(crate) fn new_with_hook_executor(
        session_id: impl Into<String>,
        run_id: impl Into<String>,
        cancellation: CancellationToken,
        hook_executor: Option<Arc<dyn HookExecutor>>,
    ) -> Arc<Self> {
        let run_id = run_id.into();
        Arc::new(Self {
            run_id: run_id.clone(),
            state: Mutex::new(InboxState {
                session_id: session_id.into(),
                run_id,
                active: true,
                closed: false,
                turn_id: None,
                turn_revision: 0,
                queue: VecDeque::new(),
                in_flight: HashMap::new(),
                seen: HashMap::new(),
                seen_order: VecDeque::new(),
                last_sequence: 0,
                interrupt_requested: false,
            }),
            admission: Mutex::new(()),
            notify: Notify::new(),
            cancellation,
            hook_executor,
        })
    }

    pub(crate) fn is_cancelled(&self) -> bool {
        self.cancellation.is_cancelled()
    }

    #[cfg(test)]
    pub(crate) fn cancellation(&self) -> CancellationToken {
        self.cancellation.clone()
    }

    /// Fast identity accessor used while a session slot is being swapped.
    /// The run id is immutable after construction, so this avoids exposing
    /// the mutable inbox state to callers.
    pub(crate) fn snapshot_run_id(&self) -> String {
        self.run_id.clone()
    }

    pub(crate) async fn update_turn(&self, turn: usize) -> RunControlSnapshot {
        self.update_turn_id(format!("turn-{turn}")).await
    }

    pub(crate) async fn update_turn_id(&self, turn_id: String) -> RunControlSnapshot {
        let _admission = self.admission.lock().await;
        let mut state = self.state.lock().await;
        if state.turn_id.as_deref() != Some(turn_id.as_str()) {
            state.turn_id = Some(turn_id);
            state.turn_revision = state.turn_revision.saturating_add(1);
        }
        snapshot(&state)
    }

    pub(crate) async fn snapshot(&self) -> RunControlSnapshot {
        let state = self.state.lock().await;
        snapshot(&state)
    }

    #[cfg(test)]
    pub(crate) async fn submit(
        &self,
        request: RunControlRequest,
        now_ms: u64,
    ) -> Result<RunControlReceipt, RunControlError> {
        request.validate()?;
        let _admission = self.admission.lock().await;
        let (receipt, cancel) = self.submit_locked(request, now_ms).await?;
        drop(_admission);
        if cancel {
            // Cancellation is fired only after the request has been durably
            // accepted in the inbox, so a caller can safely retry by id.
            self.cancellation.cancel();
        }
        self.notify.notify_waiters();
        Ok(receipt)
    }

    /// Admit one request while the caller holds the transition gate.  Keeping
    /// the state mutation in one helper lets hook-backed admission publish its
    /// `accepted` observation before the loop can drain the queue.
    async fn submit_locked(
        &self,
        request: RunControlRequest,
        now_ms: u64,
    ) -> Result<(RunControlReceipt, bool), RunControlError> {
        let fingerprint = request_fingerprint(&request)?;
        let mut state = self.state.lock().await;

        if let Some(previous) = state.seen.get(&request.request_id) {
            if previous.fingerprint == fingerprint {
                return Ok((previous.receipt.clone(), false));
            }
            return Err(RunControlError::DuplicateRequest {
                request_id: request.request_id,
            });
        }
        if request
            .session_id
            .as_deref()
            .is_some_and(|id| id != state.session_id)
        {
            return Err(RunControlError::SessionMismatch);
        }
        if request.run_id != state.run_id {
            return Err(RunControlError::RunMismatch {
                run_id: request.run_id,
            });
        }
        if state.closed || !state.active || self.is_cancelled() {
            return Err(if state.closed {
                RunControlError::Closed
            } else {
                RunControlError::NoActiveRun
            });
        }
        if request
            .deadline_ms
            .is_some_and(|deadline| now_ms > deadline)
        {
            return Err(RunControlError::DeadlineExceeded);
        }
        if (request.expected_turn_id.is_some() && request.expected_turn_id != state.turn_id)
            || request
                .expected_turn_revision
                .is_some_and(|revision| revision != state.turn_revision)
        {
            return Err(RunControlError::StaleTurn {
                expected_turn_id: request.expected_turn_id,
                expected_revision: request.expected_turn_revision,
                actual_turn_id: state.turn_id.clone(),
                actual_revision: state.turn_revision,
            });
        }
        if state.queue.len() >= RUN_CONTROL_MAX_QUEUE {
            return Err(RunControlError::QueueFull);
        }

        state.last_sequence = state.last_sequence.saturating_add(1);
        let receipt = RunControlReceipt {
            schema: RUN_CONTROL_RECEIPT_SCHEMA_V1.to_string(),
            request_id: request.request_id.clone(),
            session_id: state.session_id.clone(),
            run_id: state.run_id.clone(),
            operation: request.command.operation(),
            state: RunControlReceiptState::Accepted,
            sequence: state.last_sequence,
            turn_id: state.turn_id.clone(),
            turn_revision: state.turn_revision,
            accepted_at_ms: now_ms,
            applied_at_ms: None,
            error: None,
        };
        let cancel = matches!(request.command, RunControlCommand::Interrupt { .. });
        if cancel {
            state.interrupt_requested = true;
        }
        state.queue.push_back(PendingRunControl {
            request: request.clone(),
            receipt: receipt.clone(),
        });
        state.seen.insert(
            receipt.request_id.clone(),
            SeenRequest {
                fingerprint,
                request,
                receipt: receipt.clone(),
            },
        );
        state.seen_order.push_back(receipt.request_id.clone());
        while state.seen_order.len() > RUN_CONTROL_MAX_SEEN_REQUESTS {
            if let Some(expired) = state.seen_order.pop_front() {
                state.seen.remove(&expired);
            }
        }
        Ok((receipt, cancel))
    }

    pub(crate) async fn drain(&self) -> Vec<PendingRunControl> {
        let _admission = self.admission.lock().await;
        let mut state = self.state.lock().await;
        let pending: Vec<_> = state.queue.drain(..).collect();
        for item in &pending {
            state
                .in_flight
                .insert(item.receipt.request_id.clone(), item.clone());
        }
        pending
    }

    pub(crate) async fn mark_applied(
        &self,
        pending: &PendingRunControl,
        turn_id: Option<String>,
        turn_revision: u64,
        now_ms: u64,
    ) -> RunControlReceipt {
        let _admission = self.admission.lock().await;
        let receipt = {
            let mut state = self.state.lock().await;
            // Shutdown may win the race after `drain` and settle this item.
            // Preserve that terminal state and avoid emitting a second post
            // receipt or applying a control after the run has ended.
            let Some(seen) = state.seen.get_mut(&pending.receipt.request_id) else {
                return pending.receipt.clone();
            };
            if seen.receipt.state != RunControlReceiptState::Accepted {
                return seen.receipt.clone();
            }
            let mut receipt = seen.receipt.clone();
            receipt.state = RunControlReceiptState::Applied;
            receipt.turn_id = turn_id;
            receipt.turn_revision = turn_revision;
            receipt.applied_at_ms = Some(now_ms);
            seen.receipt = receipt.clone();
            state.in_flight.remove(&receipt.request_id);
            receipt
        };
        drop(_admission);
        if receipt.state == RunControlReceiptState::Applied {
            self.record_receipt(&pending.request, &receipt).await;
        }
        receipt
    }

    /// Mark accepted-but-unconsumed controls as settled when a run exits.
    pub(crate) async fn close(&self, now_ms: u64) {
        let _admission = self.admission.lock().await;
        let settled = {
            let mut state = self.state.lock().await;
            state.active = false;
            state.closed = true;
            let mut settled = Vec::new();
            state.queue.clear();
            state.in_flight.clear();
            // Iterate the retained receipts rather than only the queue: the
            // loop may have drained a control immediately before close.
            for seen in state.seen.values_mut() {
                if seen.receipt.state == RunControlReceiptState::Accepted {
                    seen.receipt.state = RunControlReceiptState::Settled;
                    seen.receipt.applied_at_ms = Some(now_ms);
                    seen.receipt.error = Some(RunControlErrorInfo {
                        code: "RUN_ENDED".to_string(),
                        message: "run ended before the control reached a safe point".to_string(),
                    });
                    settled.push((seen.request.clone(), seen.receipt.clone()));
                }
            }
            settled
        };
        drop(_admission);
        self.notify.notify_waiters();
        for (request, receipt) in settled {
            self.record_receipt(&request, &receipt).await;
        }
    }

    pub(crate) async fn deactivate(&self, now_ms: u64) {
        self.close(now_ms).await;
    }

    /// Submit a control through the run-frozen governance boundary. The hook
    /// decision is made before queue admission; accepted, applied, and
    /// settled receipts are then emitted as observational post events.
    pub(crate) async fn submit_with_hooks(
        &self,
        request: RunControlRequest,
        now_ms: u64,
    ) -> Result<RunControlReceipt, RunControlError> {
        request.validate()?;

        // Serialize the policy decision, queue admission, and accepted
        // observation.  This is required for both concurrent idempotency and
        // the accepted-before-applied event ordering guarantee.
        let _admission = self.admission.lock().await;

        // Idempotent retries must not execute a policy callback twice. This is
        // especially important for host hooks that charge a budget or create
        // an approval record as a side effect.
        if let Some(receipt) = self.known_receipt(&request).await? {
            return Ok(receipt);
        }

        if let Some(executor) = &self.hook_executor {
            match executor.before_run_control(&request).await {
                HookOutcome::Continue(_) | HookOutcome::Skip => {}
                outcome => {
                    let error = hook_outcome_error(outcome);
                    let rejected =
                        rejected_receipt(&request, &self.snapshot().await, now_ms, &error);
                    executor.record_run_control(&request, &rejected).await;
                    return Err(error);
                }
            }
        }

        let (receipt, cancel) = self.submit_locked(request.clone(), now_ms).await?;
        self.record_receipt(&request, &receipt).await;
        drop(_admission);
        if cancel {
            self.cancellation.cancel();
        }
        self.notify.notify_waiters();
        Ok(receipt)
    }

    async fn known_receipt(
        &self,
        request: &RunControlRequest,
    ) -> Result<Option<RunControlReceipt>, RunControlError> {
        let fingerprint = request_fingerprint(request)?;
        let state = self.state.lock().await;
        match state.seen.get(&request.request_id) {
            None => Ok(None),
            Some(previous) if previous.fingerprint == fingerprint => {
                Ok(Some(previous.receipt.clone()))
            }
            Some(_) => Err(RunControlError::DuplicateRequest {
                request_id: request.request_id.clone(),
            }),
        }
    }

    async fn record_receipt(&self, request: &RunControlRequest, receipt: &RunControlReceipt) {
        if let Some(executor) = &self.hook_executor {
            executor.record_run_control(request, receipt).await;
        }
    }
}

fn request_fingerprint(request: &RunControlRequest) -> Result<String, RunControlError> {
    let encoded = serde_json::to_vec(request).map_err(|error| {
        RunControlError::InvalidRequest(format!("could not encode request: {error}"))
    })?;
    Ok(format!("sha256:{:x}", Sha256::digest(encoded)))
}

fn hook_outcome_error(outcome: HookOutcome) -> RunControlError {
    match outcome {
        HookOutcome::Block { reason } => RunControlError::HookDenied { reason },
        HookOutcome::Retry {
            reason,
            retry_after_ms,
        } => RunControlError::HookRetry {
            reason,
            retry_after_ms,
        },
        HookOutcome::Escalate { reason, target } => RunControlError::HookDenied {
            reason: target
                .map(|target| format!("{reason} (escalate to {target})"))
                .unwrap_or(reason),
        },
        HookOutcome::Continue(_) | HookOutcome::Skip => RunControlError::InvalidRequest(
            "unexpected non-terminal run-control hook outcome".to_string(),
        ),
    }
}

fn rejected_receipt(
    request: &RunControlRequest,
    snapshot: &RunControlSnapshot,
    now_ms: u64,
    error: &RunControlError,
) -> RunControlReceipt {
    RunControlReceipt {
        schema: RUN_CONTROL_RECEIPT_SCHEMA_V1.to_string(),
        request_id: request.request_id.clone(),
        session_id: snapshot.session_id.clone(),
        run_id: request.run_id.clone(),
        operation: request.command.operation(),
        state: RunControlReceiptState::Rejected,
        sequence: 0,
        turn_id: snapshot.turn_id.clone(),
        turn_revision: snapshot.turn_revision,
        accepted_at_ms: now_ms,
        applied_at_ms: Some(now_ms),
        error: Some(RunControlErrorInfo {
            code: error.code().to_string(),
            message: error.to_string(),
        }),
    }
}

fn snapshot(state: &InboxState) -> RunControlSnapshot {
    RunControlSnapshot {
        session_id: state.session_id.clone(),
        run_id: state.run_id.clone(),
        active: state.active && !state.closed,
        turn_id: state.turn_id.clone(),
        turn_revision: state.turn_revision,
        queued_controls: state.queue.len(),
        interrupt_requested: state.interrupt_requested,
        last_sequence: state.last_sequence,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hooks::{HookEvent, HookResult};
    use async_trait::async_trait;
    use std::sync::Mutex as StdMutex;

    fn inbox() -> Arc<RunControlInbox> {
        RunControlInbox::new("session-1", "run-1", CancellationToken::new())
    }

    #[tokio::test]
    async fn steer_is_idempotent_and_applies_at_safe_point() {
        let inbox = inbox();
        let turn = inbox.update_turn(1).await;
        let mut request = RunControlRequest::steer("run-1", "focus on tests")
            .with_session_id("session-1")
            .with_expected_turn("turn-1", turn.turn_revision);
        request.request_id = "req-1".to_string();

        let accepted = inbox.submit(request.clone(), 10).await.unwrap();
        assert_eq!(accepted.state, RunControlReceiptState::Accepted);
        assert_eq!(inbox.submit(request, 11).await.unwrap(), accepted);

        let pending = inbox.drain().await;
        assert_eq!(pending.len(), 1);
        let applied = inbox
            .mark_applied(
                &pending[0],
                Some("turn-1".to_string()),
                turn.turn_revision,
                12,
            )
            .await;
        assert_eq!(applied.state, RunControlReceiptState::Applied);
        assert_eq!(inbox.snapshot().await.queued_controls, 0);
    }

    #[tokio::test]
    async fn stale_turn_and_duplicate_conflict_are_rejected() {
        let inbox = inbox();
        let turn = inbox.update_turn(1).await;
        let mut request = RunControlRequest::steer("run-1", "first")
            .with_expected_turn("turn-1", turn.turn_revision);
        request.request_id = "same-id".to_string();
        inbox.submit(request.clone(), 1).await.unwrap();

        let mut conflicting = request.clone();
        conflicting.command = RunControlCommand::Steer {
            input: "different".to_string(),
        };
        assert!(matches!(
            inbox.submit(conflicting, 2).await,
            Err(RunControlError::DuplicateRequest { .. })
        ));

        inbox.update_turn(2).await;
        let stale = RunControlRequest::steer("run-1", "late")
            .with_expected_turn("turn-1", turn.turn_revision);
        assert!(matches!(
            inbox.submit(stale, 3).await,
            Err(RunControlError::StaleTurn { .. })
        ));
    }

    #[tokio::test]
    async fn interrupt_is_accepted_before_cancellation_fires() {
        let inbox = inbox();
        let request = RunControlRequest::interrupt("run-1");
        let receipt = inbox.submit(request, 1).await.unwrap();
        assert_eq!(receipt.state, RunControlReceiptState::Accepted);
        assert!(inbox.cancellation().is_cancelled());
        assert!(inbox.snapshot().await.interrupt_requested);
    }

    #[tokio::test]
    async fn close_settles_pending_requests() {
        let inbox = inbox();
        let request = RunControlRequest::steer("run-1", "not applied");
        let accepted = inbox.submit(request, 1).await.unwrap();
        inbox.close(2).await;
        assert!(!inbox.snapshot().await.active);
        let retry = RunControlRequest {
            request_id: accepted.request_id.clone(),
            ..RunControlRequest::steer("run-1", "not applied")
        };
        let settled = inbox.submit(retry, 3).await.unwrap();
        assert_eq!(settled.state, RunControlReceiptState::Settled);
        assert_eq!(settled.error.unwrap().code, "RUN_ENDED");
    }

    #[tokio::test]
    async fn close_settles_a_control_already_drained_by_the_loop() {
        let inbox = inbox();
        let request = RunControlRequest::steer("run-1", "close race");
        let accepted = inbox.submit(request, 1).await.unwrap();
        let pending = inbox.drain().await;
        assert_eq!(pending.len(), 1);

        // The loop has taken ownership of the queue item, but has not yet
        // acknowledged application. Closing must still produce a terminal
        // receipt rather than leaving the request permanently Accepted.
        inbox.close(2).await;
        let retry = RunControlRequest {
            request_id: accepted.request_id.clone(),
            ..RunControlRequest::steer("run-1", "close race")
        };
        let settled = inbox.submit(retry, 3).await.unwrap();
        assert_eq!(settled.state, RunControlReceiptState::Settled);

        // A late safe-point acknowledgement cannot resurrect the settled
        // request or emit another transition.
        let late = inbox
            .mark_applied(&pending[0], Some("turn-1".into()), 1, 4)
            .await;
        assert_eq!(late.state, RunControlReceiptState::Settled);
    }

    #[derive(Debug, Default)]
    struct RecordingHook {
        events: StdMutex<Vec<HookEvent>>,
        deny: bool,
    }

    #[async_trait]
    impl HookExecutor for RecordingHook {
        async fn fire(&self, event: &HookEvent) -> HookResult {
            self.events.lock().unwrap().push(event.clone());
            if self.deny && matches!(event, HookEvent::PreRunControl(_)) {
                HookResult::block("host policy denied control")
            } else {
                HookResult::continue_()
            }
        }
    }

    #[tokio::test]
    async fn governance_hooks_observe_each_receipt_transition_once() {
        let hooks = Arc::new(RecordingHook::default());
        let inbox = RunControlInbox::new_with_hook_executor(
            "session-1",
            "run-1",
            CancellationToken::new(),
            Some(hooks.clone()),
        );
        let request = RunControlRequest::steer("run-1", "keep the answer concise")
            .with_session_id("session-1");
        let accepted = inbox.submit_with_hooks(request.clone(), 10).await.unwrap();
        let pending = inbox.drain().await;
        let _applied = inbox
            .mark_applied(&pending[0], Some("turn-1".to_string()), 1, 11)
            .await;
        inbox.close(12).await;

        let events = hooks.events.lock().unwrap();
        assert_eq!(
            events
                .iter()
                .filter(|event| matches!(event, HookEvent::PreRunControl(_)))
                .count(),
            1,
        );
        assert_eq!(
            events
                .iter()
                .filter(|event| matches!(event, HookEvent::PostRunControl(_)))
                .count(),
            2,
            "accepted and applied receipts must both be observable",
        );
        assert_eq!(accepted.state, RunControlReceiptState::Accepted);
    }

    #[tokio::test]
    async fn concurrent_duplicate_submission_runs_governance_once() {
        let hooks = Arc::new(RecordingHook::default());
        let inbox = RunControlInbox::new_with_hook_executor(
            "session-1",
            "run-1",
            CancellationToken::new(),
            Some(hooks.clone()),
        );
        let mut request = RunControlRequest::steer("run-1", "one admission");
        request.request_id = "concurrent-request".to_string();

        let attempts = (0..16).map(|_| {
            let inbox = Arc::clone(&inbox);
            let request = request.clone();
            async move { inbox.submit_with_hooks(request, 10).await }
        });
        let receipts = futures::future::join_all(attempts).await;
        let first = receipts[0].as_ref().expect("submission should succeed");
        assert!(receipts.iter().all(|result| result.as_ref() == Ok(first)));

        let events = hooks.events.lock().unwrap();
        assert_eq!(
            events
                .iter()
                .filter(|event| matches!(event, HookEvent::PreRunControl(_)))
                .count(),
            1,
            "a concurrent idempotent retry must not re-run the policy hook",
        );
        assert_eq!(
            events
                .iter()
                .filter(|event| matches!(event, HookEvent::PostRunControl(_)))
                .count(),
            1,
            "only the first admission emits an accepted observation",
        );
    }

    #[tokio::test]
    async fn denied_control_never_enters_the_inbox() {
        let hooks = Arc::new(RecordingHook {
            deny: true,
            ..Default::default()
        });
        let inbox = RunControlInbox::new_with_hook_executor(
            "session-1",
            "run-1",
            CancellationToken::new(),
            Some(hooks.clone()),
        );
        let error = inbox
            .submit_with_hooks(RunControlRequest::interrupt("run-1"), 10)
            .await
            .unwrap_err();
        assert!(matches!(error, RunControlError::HookDenied { .. }));
        assert_eq!(inbox.snapshot().await.queued_controls, 0);
        let events = hooks.events.lock().unwrap();
        assert!(events.iter().any(|event| matches!(
            event,
            HookEvent::PostRunControl(crate::hooks::PostRunControlEvent {
                state: RunControlReceiptState::Rejected,
                ..
            })
        )));
    }
}