harness_codex/

app_server_runtime.rs

// SPDX-License-Identifier: AGPL-3.0-or-later

//! Pure, injected JSONL state machine for the Codex whole-agent app-server seam.
//!
//! This module does not own process creation or authentication. Callers supply
//! complete transcript lines and transport the returned request lines.

use crate::app_server::CURRENT_STANDALONE_CLI_VERSION;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::path::{Component, Path};
use thiserror::Error;

const MAX_LINE_BYTES: usize = 256 * 1024;
const MAX_TEXT_BYTES: usize = 64 * 1024;
const MAX_OUTPUT_BYTES: usize = 256 * 1024;
const MAX_ID_BYTES: usize = 128;
const MODEL_PAGE_LIMIT: usize = 50;
const MAX_MODELS: usize = 256;
const MAX_PENDING: usize = 64;
const MAX_THREADS: usize = 64;
const MAX_TURNS: usize = 64;
const MAX_ITEMS_PER_TURN: usize = 256;
const MAX_THREAD_PATH_BYTES: usize = 4 * 1024;

const TOKEN_KEYS: [&str; 9] = [
    "access_token",
    "accesstoken",
    "refresh_token",
    "refreshtoken",
    "id_token",
    "idtoken",
    "authorization",
    "api_key",
    "apikey",
];

#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ReasoningEffort {
    Minimal,
    Low,
    Medium,
    High,
    Xhigh,
    Max,
    Ultra,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum CodexRole {
    Luna,
    Terra,
    Sol,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct CodexRoleProjection {
    pub role: CodexRole,
    pub task_class: &'static str,
    pub escalates_to: Option<CodexRole>,
    pub escalation_reason: Option<&'static str>,
}

pub const OPENAI_ROLE_PROJECTIONS: [CodexRoleProjection; 3] = [
    CodexRoleProjection {
        role: CodexRole::Luna,
        task_class: "routine",
        escalates_to: Some(CodexRole::Terra),
        escalation_reason: Some("qualifying correctness, format, or capability failure"),
    },
    CodexRoleProjection {
        role: CodexRole::Terra,
        task_class: "standard",
        escalates_to: Some(CodexRole::Sol),
        escalation_reason: Some("qualifying correctness, format, or capability failure"),
    },
    CodexRoleProjection {
        role: CodexRole::Sol,
        task_class: "complex_or_high_risk",
        escalates_to: None,
        escalation_reason: None,
    },
];

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct ManagedAccountProjection {
    pub plan_type: Option<String>,
    pub requires_openai_auth: bool,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct ModelProjection {
    pub id: String,
    pub display_name: String,
    pub supported_efforts: Vec<ReasoningEffort>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct ModelPageProjection {
    pub models: Vec<ModelProjection>,
    pub next_cursor: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct RateLimitWindowProjection {
    pub used_percent: u8,
    pub window_duration_minutes: Option<u64>,
    pub resets_at_unix_seconds: Option<u64>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct RateLimitsProjection {
    pub primary: Option<RateLimitWindowProjection>,
    pub secondary: Option<RateLimitWindowProjection>,
    pub reached_type: Option<String>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReplayDisposition {
    EligibleBeforeVisibility,
    DeniedVisibleOutput,
    DeniedToolOrSideEffect,
    DeniedModelReroute,
}

/// Explicit handling policy for whole-agent tool lifecycle events.
///
/// The production-safe default rejects every tool or side effect. Canonical
/// MCP lifecycle handling must be selected before the runtime is used so a
/// caller cannot widen a live turn after it begins.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum CodexToolPolicy {
    #[default]
    RejectAll,
    CanonicalMcpLifecycle,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ToolCallStatus {
    Completed,
    Failed,
}

/// Content-free mapping of canonical `turn/completed` failure metadata.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TerminalFailureClass {
    ContextWindow,
    UsageLimit,
    Transport,
    InvalidRequest,
    Unauthorized,
    Sandbox,
    Internal,
    Other,
    Unclassified,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TerminalStatus {
    Completed,
    Failed,
    Interrupted,
}

#[derive(Clone, PartialEq, Eq)]
pub struct TurnCompletionReceipt {
    pub thread_id: String,
    pub turn_id: String,
    pub selected_model: String,
    pub selected_effort: ReasoningEffort,
    pub status: TerminalStatus,
    pub output: Option<String>,
    pub replay: ReplayDisposition,
    pub failure: Option<TerminalFailureClass>,
    /// Authoritative input tokens for this turn only. Successful turns always
    /// contain this value; failed or interrupted turns may not have emitted it.
    pub input_tokens: Option<u64>,
    /// Authoritative output tokens for this turn only. Successful turns always
    /// contain this value; failed or interrupted turns may not have emitted it.
    pub output_tokens: Option<u64>,
}

impl fmt::Debug for TurnCompletionReceipt {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("TurnCompletionReceipt")
            .field("thread_id", &self.thread_id)
            .field("turn_id", &self.turn_id)
            .field("selected_model", &self.selected_model)
            .field("selected_effort", &self.selected_effort)
            .field("status", &self.status)
            .field("output", &self.output.as_ref().map(|_| "[REDACTED]"))
            .field("replay", &self.replay)
            .field("failure", &self.failure)
            .field("input_tokens", &self.input_tokens)
            .field("output_tokens", &self.output_tokens)
            .finish()
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RuntimeEvent {
    InitializeAccepted,
    /// A strictly validated, content-free compatibility marker. The process
    /// client consumes this marker immediately and never buffers it.
    NonSemanticNotificationIgnored,
    ManagedAccount(ManagedAccountProjection),
    ModelPage(ModelPageProjection),
    RateLimits(RateLimitsProjection),
    ThreadStarted {
        thread_id: String,
    },
    TurnAccepted {
        thread_id: String,
        turn_id: String,
    },
    TurnStarted {
        thread_id: String,
        turn_id: String,
    },
    ItemStarted {
        thread_id: String,
        turn_id: String,
        item_id: String,
    },
    ItemDelta {
        thread_id: String,
        turn_id: String,
        item_id: String,
    },
    ItemCompleted {
        thread_id: String,
        turn_id: String,
        item_id: String,
    },
    ToolCallStarted {
        thread_id: String,
        turn_id: String,
        item_id: String,
    },
    ToolCallCompleted {
        thread_id: String,
        turn_id: String,
        item_id: String,
        status: ToolCallStatus,
    },
    ModelReroutedFailure {
        thread_id: String,
        turn_id: String,
        from_model: String,
        to_model: String,
    },
    InterruptAcknowledged {
        thread_id: String,
        turn_id: String,
    },
    TurnCompleted(TurnCompletionReceipt),
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum HandshakeState {
    New,
    InitializePending,
    InitializeAccepted,
    Ready,
}

#[derive(Clone, Debug)]
enum PendingRequest {
    Initialize,
    AccountRead,
    ModelList,
    RateLimitsRead,
    ThreadStart {
        model: String,
        effort: ReasoningEffort,
    },
    TurnStart {
        thread_id: String,
        model: String,
        effort: ReasoningEffort,
    },
    Interrupt {
        thread_id: String,
        turn_id: String,
    },
}

#[derive(Clone, Debug)]
struct ThreadSelection {
    model: String,
    effort: ReasoningEffort,
}

/// Notification-first correlation keyed by the still-pending `turn/start`
/// JSON-RPC request ID. Values are content-free and consumed by the response.
#[derive(Clone, Debug, PartialEq, Eq)]
struct PendingTurnStarted {
    thread_id: String,
    turn_id: String,
}

#[derive(Clone, Debug)]
enum ItemKind {
    AgentMessage,
    Reasoning,
    McpToolCall(ToolLifecycleBinding),
}

#[derive(Clone, Debug)]
struct ToolLifecycleBinding {
    item_id: String,
    server: String,
    tool: String,
    arguments_digest: [u8; 32],
}

#[derive(Clone, Debug)]
struct ItemState {
    kind: ItemKind,
    delta_bytes: usize,
    completed: Option<String>,
    lifecycle_complete: bool,
}

#[derive(Clone, Debug)]
struct TurnState {
    model: String,
    effort: ReasoningEffort,
    items: BTreeMap<String, ItemState>,
    final_item: Option<String>,
    visible_output: bool,
    tool_or_side_effect: bool,
    model_rerouted: bool,
    interrupt_requested: bool,
    lifecycle_started: bool,
    output_bytes: usize,
    usage: Option<TurnUsageCounts>,
    // The app-server's `total` projection is a private, monotonic admission
    // baseline. Receipt accounting remains bound to the latest `last`
    // projection: `TurnCompletionReceipt` is explicitly per-turn, while
    // `total` is cumulative across the app-server's accounting scope.
    total_usage: Option<CumulativeTurnUsageCounts>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct TurnUsageCounts {
    input_tokens: u64,
    output_tokens: u64,
}

#[derive(Clone, Copy, PartialEq, Eq)]
struct CumulativeTurnUsageCounts {
    cached_input_tokens: u64,
    cache_write_input_tokens: Option<u64>,
    input_tokens: u64,
    output_tokens: u64,
    reasoning_output_tokens: u64,
    total_tokens: u64,
}

impl std::fmt::Debug for CumulativeTurnUsageCounts {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("CumulativeTurnUsageCounts(<redacted>)")
    }
}

impl TurnState {
    fn new(model: String, effort: ReasoningEffort, lifecycle_started: bool) -> Self {
        Self {
            model,
            effort,
            items: BTreeMap::new(),
            final_item: None,
            visible_output: false,
            tool_or_side_effect: false,
            model_rerouted: false,
            interrupt_requested: false,
            lifecycle_started,
            output_bytes: 0,
            usage: None,
            total_usage: None,
        }
    }

    fn replay(&self) -> ReplayDisposition {
        if self.model_rerouted {
            ReplayDisposition::DeniedModelReroute
        } else if self.tool_or_side_effect {
            ReplayDisposition::DeniedToolOrSideEffect
        } else if self.visible_output {
            ReplayDisposition::DeniedVisibleOutput
        } else {
            ReplayDisposition::EligibleBeforeVisibility
        }
    }
}

#[derive(Debug, Error, PartialEq, Eq)]
pub enum RuntimeError {
    #[error("app-server runtime input is invalid")]
    InvalidInput,
    #[error("app-server runtime input contained forbidden sensitive material")]
    SensitiveInputRejected,
    #[error("app-server handshake state is invalid")]
    InvalidHandshake,
    #[error("app-server runtime is not initialized")]
    NotInitialized,
    #[error("app-server pending request capacity is exhausted")]
    PendingCapacityExceeded,
    #[error("app-server response correlation is unknown")]
    UnknownCorrelation,
    #[error("app-server response was a protocol failure")]
    ProtocolFailure,
    #[error("app-server returned a JSON-RPC error response")]
    JsonRpcRejected,
    #[error("unsupported app-server notification method: {0}")]
    UnsupportedNotificationMethod(String),
    #[error("the account is not managed ChatGPT")]
    AccountNotManagedChatgpt,
    #[error("the app-server model page exceeds configured bounds")]
    ModelPageLimitExceeded,
    #[error("the requested model is not present in the bounded catalog")]
    UnknownModel,
    #[error("the requested reasoning effort is not advertised by the model")]
    UnsupportedEffort,
    #[error("the app-server event does not match an active thread or turn")]
    CorrelationMismatch,
    #[error("a tool or side-effecting item was rejected")]
    SideEffectingItemRejected,
    #[error("the app-server output exceeds configured bounds")]
    OutputLimitExceeded,
}

pub struct CodexAgentRuntime {
    tool_policy: CodexToolPolicy,
    handshake: HandshakeState,
    next_id: u64,
    pending: BTreeMap<u64, PendingRequest>,
    /// At most `MAX_TURNS` notification-first markers may coexist with active
    /// turns. A marker never authorizes replay and cannot outlive this runtime.
    pending_turn_started: BTreeMap<u64, PendingTurnStarted>,
    models: BTreeMap<String, ModelProjection>,
    threads: BTreeMap<String, ThreadSelection>,
    turns: BTreeMap<(String, String), TurnState>,
    initialized_server_version: Option<String>,
}

impl fmt::Debug for CodexAgentRuntime {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("CodexAgentRuntime")
            .field("tool_policy", &self.tool_policy)
            .field("handshake", &self.handshake)
            .field("pending_count", &self.pending.len())
            .field(
                "pending_turn_started_count",
                &self.pending_turn_started.len(),
            )
            .field("model_count", &self.models.len())
            .field("thread_count", &self.threads.len())
            .field("turn_count", &self.turns.len())
            .field("transcript", &"[REDACTED]")
            .finish()
    }
}

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

impl CodexAgentRuntime {
    pub fn new() -> Self {
        Self::with_tool_policy(CodexToolPolicy::RejectAll)
    }

    pub fn with_tool_policy(tool_policy: CodexToolPolicy) -> Self {
        Self {
            tool_policy,
            handshake: HandshakeState::New,
            next_id: 1,
            pending: BTreeMap::new(),
            pending_turn_started: BTreeMap::new(),
            models: BTreeMap::new(),
            threads: BTreeMap::new(),
            turns: BTreeMap::new(),
            initialized_server_version: None,
        }
    }

    /// Version attested by the exact app-server child that answered initialize.
    pub fn initialized_server_version(&self) -> Option<&str> {
        self.initialized_server_version.as_deref()
    }

    pub fn initialize_request(&mut self) -> Result<String, RuntimeError> {
        if self.handshake != HandshakeState::New {
            return Err(RuntimeError::InvalidHandshake);
        }
        let line = self.request(
            "initialize",
            json!({
                "clientInfo":{"name":"steadymux","title":"SteadyMux","version":env!("CARGO_PKG_VERSION")},
                "capabilities":{"experimentalApi":false}
            }),
            PendingRequest::Initialize,
        )?;
        self.handshake = HandshakeState::InitializePending;
        Ok(line)
    }

    pub fn initialized_notification(&mut self) -> Result<String, RuntimeError> {
        if self.handshake != HandshakeState::InitializeAccepted {
            return Err(RuntimeError::InvalidHandshake);
        }
        self.handshake = HandshakeState::Ready;
        encode_line(&json!({"jsonrpc":"2.0","method":"initialized"}))
    }

    pub fn account_read_request(&mut self) -> Result<String, RuntimeError> {
        self.ready_request(
            "account/read",
            json!({"refreshToken":false}),
            PendingRequest::AccountRead,
        )
    }

    pub fn model_list_request(&mut self, cursor: Option<&str>) -> Result<String, RuntimeError> {
        let cursor = cursor.map(validated_id).transpose()?;
        self.ready_request(
            "model/list",
            json!({"cursor":cursor,"limit":MODEL_PAGE_LIMIT,"includeHidden":false}),
            PendingRequest::ModelList,
        )
    }

    pub fn rate_limits_read_request(&mut self) -> Result<String, RuntimeError> {
        self.ready_request(
            "account/rateLimits/read",
            json!({}),
            PendingRequest::RateLimitsRead,
        )
    }

    pub fn thread_start_request(
        &mut self,
        model: &str,
        effort: ReasoningEffort,
    ) -> Result<String, RuntimeError> {
        self.thread_start_request_with_instructions(model, effort, None, None)
    }

    pub fn thread_start_request_with_instructions(
        &mut self,
        model: &str,
        effort: ReasoningEffort,
        base_instructions: Option<&str>,
        developer_instructions: Option<&str>,
    ) -> Result<String, RuntimeError> {
        self.require_model_effort(model, effort)?;
        let base_instructions = validated_instruction(base_instructions)?;
        let developer_instructions = validated_instruction(developer_instructions)?;
        let mut params = json!({
            "model":model,
            "modelProvider":"openai",
            "approvalPolicy":"never",
            "sandbox":"read-only",
            "ephemeral":true
        });
        let params_object = params.as_object_mut().ok_or(RuntimeError::InvalidInput)?;
        if let Some(base_instructions) = base_instructions {
            params_object.insert(
                "baseInstructions".to_owned(),
                Value::String(base_instructions),
            );
        }
        if let Some(developer_instructions) = developer_instructions {
            params_object.insert(
                "developerInstructions".to_owned(),
                Value::String(developer_instructions),
            );
        }
        self.ready_request(
            "thread/start",
            params,
            PendingRequest::ThreadStart {
                model: model.to_owned(),
                effort,
            },
        )
    }

    pub fn turn_start_request(
        &mut self,
        thread_id: &str,
        client_message_id: &str,
        text: &str,
        model: &str,
        effort: ReasoningEffort,
    ) -> Result<String, RuntimeError> {
        let thread_id = validated_id(thread_id)?;
        let client_message_id = validated_id(client_message_id)?;
        if text.is_empty() || text.len() > MAX_TEXT_BYTES || text.contains('\0') {
            return Err(RuntimeError::InvalidInput);
        }
        self.require_model_effort(model, effort)?;
        let selection = self
            .threads
            .get(&thread_id)
            .ok_or(RuntimeError::CorrelationMismatch)?;
        if selection.model != model || selection.effort != effort {
            return Err(RuntimeError::CorrelationMismatch);
        }
        self.ready_request(
            "turn/start",
            json!({
                "threadId":thread_id,
                "clientUserMessageId":client_message_id,
                "input":[{"type":"text","text":text}],
                "model":model,
                "effort":effort,
                "approvalPolicy":"never",
                "sandboxPolicy":{"type":"readOnly"}
            }),
            PendingRequest::TurnStart {
                thread_id,
                model: model.to_owned(),
                effort,
            },
        )
    }

    pub fn turn_interrupt_request(
        &mut self,
        thread_id: &str,
        turn_id: &str,
    ) -> Result<String, RuntimeError> {
        let thread_id = validated_id(thread_id)?;
        let turn_id = validated_id(turn_id)?;
        let turn = self
            .turns
            .get(&(thread_id.clone(), turn_id.clone()))
            .ok_or(RuntimeError::CorrelationMismatch)?;
        if turn.interrupt_requested {
            return Err(RuntimeError::CorrelationMismatch);
        }
        let line = self.ready_request(
            "turn/interrupt",
            json!({"threadId":thread_id,"turnId":turn_id}),
            PendingRequest::Interrupt {
                thread_id: thread_id.clone(),
                turn_id: turn_id.clone(),
            },
        )?;
        self.turns
            .get_mut(&(thread_id, turn_id))
            .ok_or(RuntimeError::CorrelationMismatch)?
            .interrupt_requested = true;
        Ok(line)
    }

    pub fn ingest_line(&mut self, line: &[u8]) -> Result<RuntimeEvent, RuntimeError> {
        if line.is_empty()
            || line.len() > MAX_LINE_BYTES
            || line.contains(&b'\n')
            || line.contains(&b'\r')
        {
            return Err(RuntimeError::InvalidInput);
        }
        let value: Value = serde_json::from_slice(line).map_err(|_| RuntimeError::InvalidInput)?;
        if let Some(id) = value.get("id").and_then(Value::as_u64) {
            if value.get("error").is_none() && contains_forbidden_sensitive_material(&value) {
                return Err(RuntimeError::SensitiveInputRejected);
            }
            let result = self.ingest_response(id, &value);
            if matches!(result, Err(RuntimeError::ProtocolFailure)) {
                eprintln!("steadymux-owned-child: runtime-protocol-source=response");
            } else if matches!(result, Err(RuntimeError::CorrelationMismatch)) {
                eprintln!("{}", correlation_mismatch_response_marker());
            }
            return result;
        }
        if contains_forbidden_sensitive_material(&value) {
            return Err(RuntimeError::SensitiveInputRejected);
        }
        let method = value
            .get("method")
            .and_then(Value::as_str)
            .ok_or(RuntimeError::InvalidInput)?;
        let params = value.get("params").ok_or(RuntimeError::InvalidInput)?;
        let result = self.ingest_notification(method, params);
        if matches!(result, Err(RuntimeError::ProtocolFailure)) {
            let method_sha256 = Sha256::digest(method.as_bytes())
                .iter()
                .map(|byte| format!("{byte:02x}"))
                .collect::<String>();
            eprintln!(
                "steadymux-owned-child: runtime-protocol-source=notification method-sha256={method_sha256}"
            );
        } else if matches!(result, Err(RuntimeError::CorrelationMismatch)) {
            eprintln!("{}", correlation_mismatch_notification_marker(method));
        }
        result
    }

    fn request(
        &mut self,
        method: &str,
        params: Value,
        pending: PendingRequest,
    ) -> Result<String, RuntimeError> {
        if self.pending.len() >= MAX_PENDING {
            return Err(RuntimeError::PendingCapacityExceeded);
        }
        let id = self.next_id;
        self.next_id = self
            .next_id
            .checked_add(1)
            .ok_or(RuntimeError::PendingCapacityExceeded)?;
        self.pending.insert(id, pending);
        encode_line(&json!({"jsonrpc":"2.0","id":id,"method":method,"params":params}))
    }

    fn ready_request(
        &mut self,
        method: &str,
        params: Value,
        pending: PendingRequest,
    ) -> Result<String, RuntimeError> {
        if self.handshake != HandshakeState::Ready {
            return Err(RuntimeError::NotInitialized);
        }
        self.request(method, params, pending)
    }

    fn require_model_effort(
        &self,
        model: &str,
        effort: ReasoningEffort,
    ) -> Result<(), RuntimeError> {
        let model = self.models.get(model).ok_or(RuntimeError::UnknownModel)?;
        if !model.supported_efforts.contains(&effort) {
            return Err(RuntimeError::UnsupportedEffort);
        }
        Ok(())
    }

    fn ingest_response(&mut self, id: u64, value: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let pending = self
            .pending
            .remove(&id)
            .ok_or(RuntimeError::UnknownCorrelation)?;
        let pending_turn_started = self.pending_turn_started.remove(&id);
        if pending_turn_started.is_some() && !matches!(&pending, PendingRequest::TurnStart { .. }) {
            self.discard_notification_first_turn(pending_turn_started.as_ref());
            return Err(RuntimeError::CorrelationMismatch);
        }
        if value.get("error").is_some() {
            self.discard_notification_first_turn(pending_turn_started.as_ref());
            return if valid_json_rpc_error_response(value) {
                Err(RuntimeError::JsonRpcRejected)
            } else {
                Err(RuntimeError::ProtocolFailure)
            };
        }
        let result = match value.get("result") {
            Some(result) => result,
            None => {
                self.discard_notification_first_turn(pending_turn_started.as_ref());
                return Err(RuntimeError::InvalidInput);
            }
        };
        match pending {
            PendingRequest::Initialize => {
                if self.handshake != HandshakeState::InitializePending || !result.is_object() {
                    return Err(RuntimeError::InvalidHandshake);
                }
                let user_agent = string_at(result, &["userAgent"])?;
                self.initialized_server_version =
                    app_server_version_from_user_agent(user_agent).map(str::to_owned);
                self.handshake = HandshakeState::InitializeAccepted;
                Ok(RuntimeEvent::InitializeAccepted)
            }
            PendingRequest::AccountRead => self.ingest_account(result),
            PendingRequest::ModelList => self.ingest_models(result),
            PendingRequest::RateLimitsRead => self.ingest_rate_limits(result),
            PendingRequest::ThreadStart { model, effort } => {
                let thread_id = validated_id(string_at(result, &["thread", "id"])?)?;
                if self.threads.len() >= MAX_THREADS || self.threads.contains_key(&thread_id) {
                    return Err(RuntimeError::CorrelationMismatch);
                }
                self.threads
                    .insert(thread_id.clone(), ThreadSelection { model, effort });
                Ok(RuntimeEvent::ThreadStarted { thread_id })
            }
            PendingRequest::TurnStart {
                thread_id,
                model,
                effort,
            } => {
                let parsed_response = (|| {
                    let turn_id = validated_id(string_at(result, &["turn", "id"])?)?;
                    if string_at(result, &["turn", "status"])? != "inProgress" {
                        return Err(RuntimeError::InvalidInput);
                    }
                    Ok(turn_id)
                })();
                let turn_id = match parsed_response {
                    Ok(turn_id) => turn_id,
                    Err(error) => {
                        self.discard_notification_first_turn(pending_turn_started.as_ref());
                        return Err(error);
                    }
                };
                let key = (thread_id.clone(), turn_id.clone());
                match pending_turn_started {
                    Some(started) => {
                        let exact_match = started.thread_id == thread_id
                            && started.turn_id == turn_id
                            && self.turns.get(&key).is_some_and(|turn| {
                                turn.model == model
                                    && turn.effort == effort
                                    && turn.lifecycle_started
                            });
                        if !exact_match {
                            self.discard_notification_first_turn(Some(&started));
                            return Err(RuntimeError::CorrelationMismatch);
                        }
                    }
                    None => {
                        if self.turns.len() >= MAX_TURNS || self.turns.contains_key(&key) {
                            return Err(RuntimeError::CorrelationMismatch);
                        }
                        self.turns.insert(key, TurnState::new(model, effort, false));
                    }
                }
                Ok(RuntimeEvent::TurnAccepted { thread_id, turn_id })
            }
            PendingRequest::Interrupt { thread_id, turn_id } => {
                if result.as_object().is_none_or(|object| !object.is_empty()) {
                    return Err(RuntimeError::InvalidInput);
                }
                Ok(RuntimeEvent::InterruptAcknowledged { thread_id, turn_id })
            }
        }
    }

    fn ingest_account(&self, result: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let account = result
            .get("account")
            .ok_or(RuntimeError::AccountNotManagedChatgpt)?;
        if account.get("type").and_then(Value::as_str) != Some("chatgpt") {
            return Err(RuntimeError::AccountNotManagedChatgpt);
        }
        let plan_type = account
            .get("planType")
            .and_then(Value::as_str)
            .map(bounded_text)
            .transpose()?;
        let requires_openai_auth = result
            .get("requiresOpenaiAuth")
            .and_then(Value::as_bool)
            .ok_or(RuntimeError::InvalidInput)?;
        Ok(RuntimeEvent::ManagedAccount(ManagedAccountProjection {
            plan_type,
            requires_openai_auth,
        }))
    }

    fn ingest_models(&mut self, result: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let data = result
            .get("data")
            .and_then(Value::as_array)
            .ok_or(RuntimeError::InvalidInput)?;
        if data.len() > MODEL_PAGE_LIMIT
            || self.models.len().saturating_add(data.len()) > MAX_MODELS
        {
            return Err(RuntimeError::ModelPageLimitExceeded);
        }
        let mut page = Vec::with_capacity(data.len());
        let mut page_ids = BTreeSet::new();
        for raw in data {
            if raw.get("hidden").and_then(Value::as_bool).unwrap_or(false) {
                continue;
            }
            let id = validated_id(
                raw.get("id")
                    .and_then(Value::as_str)
                    .ok_or(RuntimeError::InvalidInput)?,
            )?;
            if self.models.contains_key(&id) || !page_ids.insert(id.clone()) {
                return Err(RuntimeError::InvalidInput);
            }
            let display_name = bounded_text(
                raw.get("displayName")
                    .and_then(Value::as_str)
                    .ok_or(RuntimeError::InvalidInput)?,
            )?;
            let effort_values = raw
                .get("supportedReasoningEfforts")
                .and_then(Value::as_array)
                .ok_or(RuntimeError::InvalidInput)?;
            if effort_values.is_empty() || effort_values.len() > 8 {
                return Err(RuntimeError::InvalidInput);
            }
            let mut seen = BTreeSet::new();
            let mut efforts = Vec::with_capacity(effort_values.len());
            for effort in effort_values {
                let effort_value = effort
                    .get("reasoningEffort")
                    .and_then(Value::as_str)
                    .ok_or(RuntimeError::InvalidInput)?;
                let effort = parse_effort(effort_value)?;
                if seen.insert(effort) {
                    efforts.push(effort);
                }
            }
            let projection = ModelProjection {
                id: id.clone(),
                display_name,
                supported_efforts: efforts,
            };
            page.push(projection);
        }
        let next_cursor = result
            .get("nextCursor")
            .and_then(Value::as_str)
            .map(validated_id)
            .transpose()?;
        for projection in &page {
            self.models
                .insert(projection.id.clone(), projection.clone());
        }
        Ok(RuntimeEvent::ModelPage(ModelPageProjection {
            models: page,
            next_cursor,
        }))
    }

    fn ingest_rate_limits(&self, result: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let limits = result.get("rateLimits").ok_or(RuntimeError::InvalidInput)?;
        let primary = parse_window(limits.get("primary"))?;
        let secondary = parse_window(limits.get("secondary"))?;
        let reached_type = limits
            .get("rateLimitReachedType")
            .and_then(Value::as_str)
            .map(bounded_text)
            .transpose()?;
        Ok(RuntimeEvent::RateLimits(RateLimitsProjection {
            primary,
            secondary,
            reached_type,
        }))
    }

    fn ingest_notification(
        &mut self,
        method: &str,
        params: &Value,
    ) -> Result<RuntimeEvent, RuntimeError> {
        match method {
            "remoteControl/status/changed" => {
                let object = params.as_object().ok_or(RuntimeError::ProtocolFailure)?;
                let exact_shape = object.len() == 4
                    && object.get("status").and_then(Value::as_str) == Some("disabled")
                    && valid_opaque_control_value(object.get("serverName"))
                    && valid_opaque_control_value(object.get("installationId"))
                    && object.get("environmentId").is_some_and(Value::is_null);
                if !exact_shape {
                    return Err(RuntimeError::ProtocolFailure);
                }
                Ok(RuntimeEvent::NonSemanticNotificationIgnored)
            }
            "mcpServer/startupStatus/updated" => self.ingest_mcp_startup_status(params),
            "warning" => self.ingest_warning(params),
            "hook/started" | "hook/completed" => Err(RuntimeError::ProtocolFailure),
            "thread/started" => self.ingest_thread_started(params),
            "thread/status/changed" => self.ingest_thread_status_changed(params),
            "thread/tokenUsage/updated" => self.ingest_thread_token_usage_updated(params),
            "account/rateLimits/updated" => self.ingest_rate_limits_updated(params),
            "item/reasoning/summaryPartAdded" => self.ingest_reasoning_part_added(params),
            "item/reasoning/summaryTextDelta" => {
                self.ingest_reasoning_delta(params, "summaryIndex")
            }
            "item/reasoning/textDelta" => self.ingest_reasoning_delta(params, "contentIndex"),
            "turn/started" => self.ingest_turn_started(params),
            "item/started" => self.ingest_item_started(params),
            "item/agentMessage/delta" => self.ingest_delta(params),
            "item/completed" => self.ingest_item_completed(params),
            "model/rerouted" => self.ingest_rerouted(params),
            "turn/completed" => self.ingest_turn_completed(params),
            _ => match validated_notification_method(method) {
                Ok(method) => Err(RuntimeError::UnsupportedNotificationMethod(method)),
                Err(_) => Err(RuntimeError::ProtocolFailure),
            },
        }
    }

    fn ingest_mcp_startup_status(&self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        const ALLOWED: [&str; 5] = ["threadId", "name", "status", "error", "failureReason"];
        let object = params.as_object().ok_or(RuntimeError::ProtocolFailure)?;
        let refusal = if !object.contains_key("name")
            || !object.contains_key("status")
            || !object.keys().all(|key| ALLOWED.contains(&key.as_str()))
        {
            Some("keys")
        } else if !valid_opaque_control_value(object.get("name")) {
            Some("name")
        } else if !matches!(
            object.get("status").and_then(Value::as_str),
            Some("starting" | "ready" | "failed" | "cancelled")
        ) {
            Some("status")
        } else if object.get("error").is_some_and(|error| {
            !error.is_null()
                && error.as_str().is_none_or(|error| {
                    error.is_empty()
                        || error.len() > MAX_TEXT_BYTES
                        || error.chars().any(char::is_control)
                })
        }) {
            Some("error")
        } else if object.get("failureReason").is_some_and(|reason| {
            !reason.is_null() && reason.as_str() != Some("reauthenticationRequired")
        }) {
            Some("reason")
        } else if object
            .get("threadId")
            .is_some_and(|thread_id| !thread_id.is_null() && !thread_id.is_string())
        {
            Some("thread")
        } else {
            None
        };
        if let Some(class) = refusal {
            eprintln!("steadymux-owned-child: mcp-startup-refused={class}");
            return Err(RuntimeError::ProtocolFailure);
        }
        if let Some(thread_id) = object.get("threadId").and_then(Value::as_str) {
            let thread_id = validated_id(thread_id)?;
            if !self.threads.contains_key(&thread_id) {
                return Err(RuntimeError::CorrelationMismatch);
            }
        }
        Ok(RuntimeEvent::NonSemanticNotificationIgnored)
    }

    fn ingest_warning(&self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let object = exact_object(params, &["threadId", "message"])?;
        let thread_id = object
            .get("threadId")
            .and_then(Value::as_str)
            .filter(|value| {
                !value.is_empty() && value.len() <= MAX_ID_BYTES && !value.contains('\0')
            })
            .ok_or(RuntimeError::ProtocolFailure)?;
        if !self.threads.contains_key(thread_id) {
            return Err(RuntimeError::CorrelationMismatch);
        }
        if object
            .get("message")
            .and_then(Value::as_str)
            .is_none_or(|message| {
                message.is_empty() || message.len() > MAX_TEXT_BYTES || message.contains('\0')
            })
        {
            return Err(RuntimeError::ProtocolFailure);
        }
        Ok(RuntimeEvent::NonSemanticNotificationIgnored)
    }

    fn ingest_rate_limits_updated(&self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let object = exact_object(params, &["rateLimits"])?;
        let limits = exact_object(
            object
                .get("rateLimits")
                .ok_or(RuntimeError::ProtocolFailure)?,
            &[
                "limitId",
                "limitName",
                "primary",
                "secondary",
                "credits",
                "individualLimit",
                "spendControlReached",
                "planType",
                "rateLimitReachedType",
            ],
        )?;
        let primary = exact_object(
            limits.get("primary").ok_or(RuntimeError::ProtocolFailure)?,
            &["usedPercent", "windowDurationMins", "resetsAt"],
        )?;
        let credits = exact_object(
            limits.get("credits").ok_or(RuntimeError::ProtocolFailure)?,
            &["hasCredits", "unlimited", "balance"],
        )?;
        let valid = valid_opaque_control_value(limits.get("limitId"))
            && limits.get("limitName").is_some_and(Value::is_null)
            && limits.get("secondary").is_some_and(Value::is_null)
            && limits.get("individualLimit").is_some_and(Value::is_null)
            && limits
                .get("spendControlReached")
                .is_some_and(|value| value.is_null() || value.is_boolean())
            && limits.get("rateLimitReachedType").is_some_and(|value| {
                value.is_null()
                    || value
                        .as_str()
                        .is_some_and(|text| bounded_text(text).is_ok())
            })
            && valid_opaque_control_value(limits.get("planType"))
            && primary
                .get("usedPercent")
                .and_then(Value::as_u64)
                .is_some_and(|percent| percent <= 100)
            && primary
                .get("windowDurationMins")
                .and_then(Value::as_u64)
                .is_some()
            && primary.get("resetsAt").and_then(Value::as_u64).is_some()
            && credits.get("hasCredits").and_then(Value::as_bool).is_some()
            && credits.get("unlimited").and_then(Value::as_bool).is_some()
            && credits
                .get("balance")
                .and_then(Value::as_str)
                .is_some_and(|balance| {
                    !balance.is_empty()
                        && balance.len() <= MAX_ID_BYTES
                        && balance
                            .bytes()
                            .all(|byte| byte.is_ascii_digit() || byte == b'.')
                });
        if !valid {
            return Err(RuntimeError::ProtocolFailure);
        }
        Ok(RuntimeEvent::NonSemanticNotificationIgnored)
    }

    fn ingest_thread_started(&self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let object = exact_object(params, &["thread"])?;
        let thread_id = validate_ephemeral_empty_thread(
            object.get("thread").ok_or(RuntimeError::ProtocolFailure)?,
        )?;
        if !self.threads.contains_key(&thread_id) {
            return Err(RuntimeError::CorrelationMismatch);
        }
        Ok(RuntimeEvent::NonSemanticNotificationIgnored)
    }

    fn ingest_turn_started(&mut self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let thread_id = validated_id(string_at(params, &["threadId"])?)?;
        let turn_id = validated_id(string_at(params, &["turn", "id"])?)?;
        if string_at(params, &["turn", "status"])? != "inProgress" {
            return Err(RuntimeError::ProtocolFailure);
        }
        if let Ok(turn) = self.turn_mut(&thread_id, &turn_id) {
            if turn.lifecycle_started {
                return Err(RuntimeError::CorrelationMismatch);
            }
            turn.lifecycle_started = true;
            return Ok(RuntimeEvent::TurnStarted { thread_id, turn_id });
        }

        let mut matching_requests =
            self.pending
                .iter()
                .filter_map(|(request_id, request)| match request {
                    PendingRequest::TurnStart {
                        thread_id: pending_thread_id,
                        model,
                        effort,
                    } if pending_thread_id == &thread_id => {
                        Some((*request_id, model.clone(), *effort))
                    }
                    _ => None,
                });
        let (request_id, model, effort) = matching_requests
            .next()
            .ok_or(RuntimeError::CorrelationMismatch)?;
        if matching_requests.next().is_some() || self.pending_turn_started.contains_key(&request_id)
        {
            return Err(RuntimeError::CorrelationMismatch);
        }
        if self.turns.len() >= MAX_TURNS {
            return Err(RuntimeError::PendingCapacityExceeded);
        }
        self.pending_turn_started.insert(
            request_id,
            PendingTurnStarted {
                thread_id: thread_id.clone(),
                turn_id: turn_id.clone(),
            },
        );
        self.turns.insert(
            (thread_id.clone(), turn_id.clone()),
            TurnState::new(model, effort, true),
        );
        Ok(RuntimeEvent::TurnStarted { thread_id, turn_id })
    }

    fn ingest_thread_status_changed(&self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let object = exact_object(params, &["threadId", "status"])?;
        let thread_id = validated_id(
            object
                .get("threadId")
                .and_then(Value::as_str)
                .ok_or(RuntimeError::ProtocolFailure)?,
        )?;
        if !self.threads.contains_key(&thread_id) {
            return Err(RuntimeError::CorrelationMismatch);
        }
        let status = object.get("status").ok_or(RuntimeError::ProtocolFailure)?;
        let status = status.as_object().ok_or(RuntimeError::ProtocolFailure)?;
        let allowed = match status.get("type").and_then(Value::as_str) {
            Some("idle") => status.len() == 1,
            Some("active") => {
                status.len() == 2
                    && status
                        .get("activeFlags")
                        .and_then(Value::as_array)
                        .is_some_and(|flags| {
                            flags.len() <= 2
                                && flags.iter().all(|flag| {
                                    matches!(
                                        flag.as_str(),
                                        Some("waitingOnApproval" | "waitingOnUserInput")
                                    )
                                })
                        })
            }
            _ => false,
        };
        if !allowed {
            return Err(RuntimeError::ProtocolFailure);
        }
        Ok(RuntimeEvent::NonSemanticNotificationIgnored)
    }

    fn ingest_thread_token_usage_updated(
        &mut self,
        params: &Value,
    ) -> Result<RuntimeEvent, RuntimeError> {
        let object = exact_object(params, &["threadId", "turnId", "tokenUsage"])
            .map_err(|_| token_usage_refusal("params-shape"))?;
        let thread_id = object
            .get("threadId")
            .and_then(Value::as_str)
            .ok_or_else(|| token_usage_refusal("ids"))
            .and_then(|id| validated_id(id).map_err(|_| token_usage_refusal("ids")))?;
        let turn_id = object
            .get("turnId")
            .and_then(Value::as_str)
            .ok_or_else(|| token_usage_refusal("ids"))
            .and_then(|id| validated_id(id).map_err(|_| token_usage_refusal("ids")))?;
        let usage = validated_last_turn_usage(object.get("tokenUsage"))?;
        let total_usage = validated_total_turn_usage(object.get("tokenUsage"))?;
        let key = (thread_id.clone(), turn_id.clone());
        let correlation_refusal = match self.turns.get(&key) {
            None => Some(TokenUsageCorrelationRefusal::MissingPair),
            Some(turn) if turn.model_rerouted => Some(TokenUsageCorrelationRefusal::Rerouted),
            Some(turn) if turn.usage.is_some() && turn.total_usage.is_none() => {
                Some(TokenUsageCorrelationRefusal::PriorUsage)
            }
            Some(_) => None,
        };
        if let Some(reason) = correlation_refusal {
            emit_token_usage_correlation_marker(reason, &thread_id, &turn_id, &self.turns);
            return Err(token_usage_refusal("turn-state/correlation"));
        }
        if let Some(turn) = self.turns.get(&key)
            && let Some(previous_total) = turn.total_usage
        {
            let total_direction = cumulative_usage_direction(previous_total, total_usage);
            if total_direction == PriorUsageDirection::DecreaseOrMixed {
                emit_prior_usage_direction_marker(
                    turn.usage
                        .map_or(PriorUsageDirection::Unavailable, |previous| {
                            prior_usage_direction(previous, usage)
                        }),
                );
                emit_prior_usage_projection_marker(
                    turn.usage.map_or(TotalUsageRelativeToLast::Equal, |last| {
                        total_usage_relative_to_last(last, total_usage)
                    }),
                    total_direction,
                );
                emit_token_usage_correlation_marker(
                    TokenUsageCorrelationRefusal::PriorUsage,
                    &thread_id,
                    &turn_id,
                    &self.turns,
                );
                return Err(token_usage_refusal("turn-state/correlation"));
            }
        }
        let turn = self
            .turn_mut(&thread_id, &turn_id)
            .map_err(|_| token_usage_refusal("turn-state/correlation"))?;
        // Some app-server versions deliver this authoritative usage update
        // after the accepted turn-start response but before `turn/started`.
        // The response has already created the exact turn record, so this is
        // not notification-first admission.  Keep every other boundary
        // closed: notifications never create turns, reroutes cannot report
        // usage, and every later update must prove a nondecreasing complete
        // cumulative projection. `last` is replaceable protocol state, but
        // only after that cumulative proof succeeds.
        turn.usage = Some(usage);
        turn.total_usage = Some(total_usage);
        Ok(RuntimeEvent::NonSemanticNotificationIgnored)
    }

    fn ingest_reasoning_part_added(&self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let object = exact_object(params, &["threadId", "turnId", "itemId", "summaryIndex"])?;
        let (thread_id, turn_id, item_id) = correlated_reasoning_ids(object, "summaryIndex")?;
        self.require_active_reasoning_item(&thread_id, &turn_id, &item_id)?;
        Ok(RuntimeEvent::NonSemanticNotificationIgnored)
    }

    fn ingest_reasoning_delta(
        &mut self,
        params: &Value,
        index_key: &str,
    ) -> Result<RuntimeEvent, RuntimeError> {
        let object = exact_object(
            params,
            &["threadId", "turnId", "itemId", index_key, "delta"],
        )?;
        let (thread_id, turn_id, item_id) = correlated_reasoning_ids(object, index_key)?;
        let delta = object
            .get("delta")
            .and_then(Value::as_str)
            .filter(|delta| delta.len() <= MAX_TEXT_BYTES)
            .ok_or(RuntimeError::ProtocolFailure)?;
        let turn = self.turn_mut(&thread_id, &turn_id)?;
        if turn.model_rerouted || !turn.lifecycle_started {
            return Err(RuntimeError::ProtocolFailure);
        }
        let item = turn
            .items
            .get_mut(&item_id)
            .filter(|item| matches!(item.kind, ItemKind::Reasoning) && !item.lifecycle_complete)
            .ok_or(RuntimeError::CorrelationMismatch)?;
        if item.delta_bytes.saturating_add(delta.len()) > MAX_OUTPUT_BYTES
            || turn.output_bytes.saturating_add(delta.len()) > MAX_OUTPUT_BYTES
        {
            return Err(RuntimeError::OutputLimitExceeded);
        }
        item.delta_bytes += delta.len();
        turn.output_bytes += delta.len();
        Ok(RuntimeEvent::NonSemanticNotificationIgnored)
    }

    fn require_active_reasoning_item(
        &self,
        thread_id: &str,
        turn_id: &str,
        item_id: &str,
    ) -> Result<(), RuntimeError> {
        let turn = self
            .turns
            .get(&(thread_id.to_owned(), turn_id.to_owned()))
            .ok_or(RuntimeError::CorrelationMismatch)?;
        if turn.model_rerouted || !turn.lifecycle_started {
            return Err(RuntimeError::ProtocolFailure);
        }
        turn.items
            .get(item_id)
            .filter(|item| matches!(item.kind, ItemKind::Reasoning) && !item.lifecycle_complete)
            .map(|_| ())
            .ok_or(RuntimeError::CorrelationMismatch)
    }

    fn ingest_item_started(&mut self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let thread_id = validated_id(string_at(params, &["threadId"])?)?;
        let turn_id = validated_id(string_at(params, &["turnId"])?)?;
        let item = params.get("item").ok_or(RuntimeError::InvalidInput)?;
        let item_type = string_at(item, &["type"])?;
        if item_type == "userMessage" {
            self.validate_user_message_notification(params, "startedAtMs")?;
            return Ok(RuntimeEvent::NonSemanticNotificationIgnored);
        }
        let item_id = validated_id(string_at(item, &["id"])?)?;
        let item_kind = match item_type {
            "agentMessage" => ItemKind::AgentMessage,
            "reasoning" => ItemKind::Reasoning,
            "mcpToolCall" if self.tool_policy == CodexToolPolicy::CanonicalMcpLifecycle => {
                if string_at(item, &["status"])? != "inProgress"
                    || item.get("arguments").is_none()
                    || !null_or_absent(item, "result")
                    || !null_or_absent(item, "error")
                {
                    return Err(RuntimeError::ProtocolFailure);
                }
                ItemKind::McpToolCall(ToolLifecycleBinding {
                    item_id: item_id.clone(),
                    server: validated_tool_name(string_at(item, &["server"])?)?,
                    tool: validated_tool_name(string_at(item, &["tool"])?)?,
                    arguments_digest: digest_json(
                        item.get("arguments").ok_or(RuntimeError::ProtocolFailure)?,
                    )?,
                })
            }
            "mcpToolCall" => {
                emit_mcp_tool_policy_refusal_marker(item, McpToolLifecycle::Started);
                if let Ok(turn) = self.turn_mut(&thread_id, &turn_id) {
                    turn.tool_or_side_effect = true;
                }
                return Err(RuntimeError::SideEffectingItemRejected);
            }
            _ => {
                emit_side_effecting_item_type_marker(item_type);
                if let Ok(turn) = self.turn_mut(&thread_id, &turn_id) {
                    turn.tool_or_side_effect = true;
                }
                return Err(RuntimeError::SideEffectingItemRejected);
            }
        };
        let tool_call = matches!(item_kind, ItemKind::McpToolCall(_));
        let turn = self.turn_mut(&thread_id, &turn_id)?;
        if turn.model_rerouted || !turn.lifecycle_started {
            return Err(RuntimeError::ProtocolFailure);
        }
        if turn.items.len() >= MAX_ITEMS_PER_TURN || turn.items.contains_key(&item_id) {
            return Err(RuntimeError::CorrelationMismatch);
        }
        if tool_call {
            turn.tool_or_side_effect = true;
        }
        turn.items.insert(
            item_id.clone(),
            ItemState {
                kind: item_kind,
                delta_bytes: 0,
                completed: None,
                lifecycle_complete: false,
            },
        );
        if tool_call {
            Ok(RuntimeEvent::ToolCallStarted {
                thread_id,
                turn_id,
                item_id,
            })
        } else {
            Ok(RuntimeEvent::ItemStarted {
                thread_id,
                turn_id,
                item_id,
            })
        }
    }

    fn ingest_delta(&mut self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let thread_id = validated_id(string_at(params, &["threadId"])?)?;
        let turn_id = validated_id(string_at(params, &["turnId"])?)?;
        let item_id = validated_id(string_at(params, &["itemId"])?)?;
        let delta = string_at(params, &["delta"])?;
        let turn = self.turn_mut(&thread_id, &turn_id)?;
        if turn.model_rerouted || !turn.lifecycle_started {
            return Err(RuntimeError::ProtocolFailure);
        }
        let item = turn
            .items
            .get_mut(&item_id)
            .ok_or(RuntimeError::CorrelationMismatch)?;
        if !matches!(item.kind, ItemKind::AgentMessage) {
            return Err(RuntimeError::CorrelationMismatch);
        }
        if item.lifecycle_complete
            || item.delta_bytes.saturating_add(delta.len()) > MAX_OUTPUT_BYTES
            || turn.output_bytes.saturating_add(delta.len()) > MAX_OUTPUT_BYTES
        {
            return Err(RuntimeError::OutputLimitExceeded);
        }
        item.delta_bytes += delta.len();
        turn.output_bytes += delta.len();
        if !delta.is_empty() {
            turn.visible_output = true;
        }
        Ok(RuntimeEvent::ItemDelta {
            thread_id,
            turn_id,
            item_id,
        })
    }

    fn ingest_item_completed(&mut self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let thread_id = validated_id(string_at(params, &["threadId"])?)?;
        let turn_id = validated_id(string_at(params, &["turnId"])?)?;
        let item = params.get("item").ok_or(RuntimeError::InvalidInput)?;
        let item_type = string_at(item, &["type"])?;
        if item_type == "userMessage" {
            self.validate_user_message_notification(params, "completedAtMs")?;
            return Ok(RuntimeEvent::NonSemanticNotificationIgnored);
        }
        let item_id = validated_id(string_at(item, &["id"])?)?;
        if item_type == "mcpToolCall" {
            if self.tool_policy != CodexToolPolicy::CanonicalMcpLifecycle {
                emit_mcp_tool_policy_refusal_marker(item, McpToolLifecycle::Completed);
                let turn = self.turn_mut(&thread_id, &turn_id)?;
                turn.tool_or_side_effect = true;
                return Err(RuntimeError::SideEffectingItemRejected);
            }
            let server = validated_tool_name(string_at(item, &["server"])?)?;
            let tool = validated_tool_name(string_at(item, &["tool"])?)?;
            let arguments_digest =
                digest_json(item.get("arguments").ok_or(RuntimeError::ProtocolFailure)?)?;
            let status = match string_at(item, &["status"])? {
                "completed" if non_null(item, "result") && null_or_absent(item, "error") => {
                    ToolCallStatus::Completed
                }
                "failed" if non_null(item, "result") || non_null(item, "error") => {
                    ToolCallStatus::Failed
                }
                _ => return Err(RuntimeError::ProtocolFailure),
            };
            let turn = self.turn_mut(&thread_id, &turn_id)?;
            if turn.model_rerouted || !turn.lifecycle_started {
                return Err(RuntimeError::ProtocolFailure);
            }
            let state = turn
                .items
                .get_mut(&item_id)
                .ok_or(RuntimeError::CorrelationMismatch)?;
            match &state.kind {
                ItemKind::McpToolCall(binding)
                    if binding.item_id == item_id
                        && binding.server == server
                        && binding.tool == tool
                        && binding.arguments_digest == arguments_digest => {}
                _ => return Err(RuntimeError::CorrelationMismatch),
            }
            if state.lifecycle_complete {
                return Err(RuntimeError::CorrelationMismatch);
            }
            state.lifecycle_complete = true;
            return Ok(RuntimeEvent::ToolCallCompleted {
                thread_id,
                turn_id,
                item_id,
                status,
            });
        }
        if !matches!(item_type, "agentMessage" | "reasoning") {
            emit_side_effecting_item_type_marker(item_type);
            if let Ok(turn) = self.turn_mut(&thread_id, &turn_id) {
                turn.tool_or_side_effect = true;
            }
            return Err(RuntimeError::SideEffectingItemRejected);
        }
        let text = if item_type == "agentMessage" {
            Some(bounded_output(string_at(item, &["text"])?)?)
        } else {
            None
        };
        let turn = self.turn_mut(&thread_id, &turn_id)?;
        if turn.model_rerouted || !turn.lifecycle_started {
            return Err(RuntimeError::ProtocolFailure);
        }
        let item = turn
            .items
            .get_mut(&item_id)
            .ok_or(RuntimeError::CorrelationMismatch)?;
        let expected_kind = if item_type == "agentMessage" {
            matches!(item.kind, ItemKind::AgentMessage)
        } else {
            matches!(item.kind, ItemKind::Reasoning)
        };
        if !expected_kind || item.lifecycle_complete {
            return Err(RuntimeError::CorrelationMismatch);
        }
        let text_len = text.as_ref().map_or(0, String::len);
        if turn.output_bytes.saturating_add(text_len) > MAX_OUTPUT_BYTES {
            return Err(RuntimeError::OutputLimitExceeded);
        }
        turn.output_bytes += text_len;
        item.completed = text;
        item.lifecycle_complete = true;
        if item_type == "agentMessage" {
            turn.final_item = Some(item_id.clone());
            turn.visible_output = true;
        }
        Ok(RuntimeEvent::ItemCompleted {
            thread_id,
            turn_id,
            item_id,
        })
    }

    fn validate_user_message_notification(
        &self,
        params: &Value,
        timestamp_key: &str,
    ) -> Result<(), RuntimeError> {
        let object = exact_object(params, &["item", "threadId", "turnId", timestamp_key])?;
        let thread_id = validated_id(
            object
                .get("threadId")
                .and_then(Value::as_str)
                .ok_or(RuntimeError::ProtocolFailure)?,
        )?;
        let turn_id = validated_id(
            object
                .get("turnId")
                .and_then(Value::as_str)
                .ok_or(RuntimeError::ProtocolFailure)?,
        )?;
        if object.get(timestamp_key).and_then(Value::as_u64).is_none() {
            return Err(RuntimeError::ProtocolFailure);
        }
        let item = exact_object(
            object.get("item").ok_or(RuntimeError::ProtocolFailure)?,
            &["type", "id", "clientId", "content"],
        )?;
        if item.get("type").and_then(Value::as_str) != Some("userMessage") {
            return Err(RuntimeError::ProtocolFailure);
        }
        validated_id(
            item.get("id")
                .and_then(Value::as_str)
                .ok_or(RuntimeError::ProtocolFailure)?,
        )?;
        validated_id(
            item.get("clientId")
                .and_then(Value::as_str)
                .ok_or(RuntimeError::ProtocolFailure)?,
        )?;
        let content = item
            .get("content")
            .and_then(Value::as_array)
            .filter(|content| content.len() == 1)
            .ok_or(RuntimeError::ProtocolFailure)?;
        let text = exact_object(&content[0], &["type", "text", "text_elements"])?;
        if text.get("type").and_then(Value::as_str) != Some("text")
            || text
                .get("text")
                .and_then(Value::as_str)
                .is_none_or(|text| text.len() > MAX_TEXT_BYTES || text.contains('\0'))
            || text
                .get("text_elements")
                .and_then(Value::as_array)
                .is_none_or(|elements| !elements.is_empty())
        {
            return Err(RuntimeError::ProtocolFailure);
        }
        let turn = self
            .turns
            .get(&(thread_id, turn_id))
            .ok_or(RuntimeError::CorrelationMismatch)?;
        if !turn.lifecycle_started || turn.model_rerouted {
            return Err(RuntimeError::ProtocolFailure);
        }
        Ok(())
    }

    fn ingest_rerouted(&mut self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let thread_id = validated_id(string_at(params, &["threadId"])?)?;
        let turn_id = validated_id(string_at(params, &["turnId"])?)?;
        let from_model = validated_id(string_at(params, &["fromModel"])?)?;
        let to_model = validated_id(string_at(params, &["toModel"])?)?;
        let turn = self.turn_mut(&thread_id, &turn_id)?;
        if turn.model != from_model {
            return Err(RuntimeError::CorrelationMismatch);
        }
        turn.model_rerouted = true;
        Ok(RuntimeEvent::ModelReroutedFailure {
            thread_id,
            turn_id,
            from_model,
            to_model,
        })
    }

    fn ingest_turn_completed(&mut self, params: &Value) -> Result<RuntimeEvent, RuntimeError> {
        let thread_id = validated_id(string_at(params, &["threadId"])?)?;
        let turn_id = validated_id(string_at(params, &["turn", "id"])?)?;
        let turn_value = params.get("turn").ok_or(RuntimeError::InvalidInput)?;
        let (status, failure) = match string_at(turn_value, &["status"])? {
            "completed" if null_or_absent(turn_value, "error") => (TerminalStatus::Completed, None),
            "failed" => {
                let error = turn_value
                    .get("error")
                    .filter(|error| error.is_object())
                    .ok_or(RuntimeError::ProtocolFailure)?;
                (TerminalStatus::Failed, Some(parse_terminal_failure(error)))
            }
            "interrupted" if null_or_absent(turn_value, "error") => {
                (TerminalStatus::Interrupted, None)
            }
            _ => return Err(RuntimeError::InvalidInput),
        };
        let turn = self
            .turns
            .remove(&(thread_id.clone(), turn_id.clone()))
            .ok_or(RuntimeError::CorrelationMismatch)?;
        if !turn.lifecycle_started || turn.items.values().any(|item| !item.lifecycle_complete) {
            return Err(RuntimeError::ProtocolFailure);
        }
        if turn.model_rerouted && status != TerminalStatus::Failed {
            return Err(RuntimeError::ProtocolFailure);
        }
        if status == TerminalStatus::Interrupted && !turn.interrupt_requested {
            return Err(RuntimeError::CorrelationMismatch);
        }
        let usage = match status {
            TerminalStatus::Completed => {
                // Completion is only valid after a fully validated cumulative
                // baseline and its latest per-turn projection were accepted.
                // The receipt intentionally projects the latter because its
                // public contract is per-turn rather than app-wide cumulative.
                turn.total_usage.ok_or(RuntimeError::ProtocolFailure)?;
                Some(turn.usage.ok_or(RuntimeError::ProtocolFailure)?)
            }
            TerminalStatus::Failed | TerminalStatus::Interrupted => turn.usage,
        };
        let output = turn
            .final_item
            .as_ref()
            .and_then(|item_id| turn.items.get(item_id))
            .and_then(|item| item.completed.clone());
        Ok(RuntimeEvent::TurnCompleted(TurnCompletionReceipt {
            thread_id,
            turn_id,
            selected_model: turn.model.clone(),
            selected_effort: turn.effort,
            status,
            output,
            replay: turn.replay(),
            failure,
            input_tokens: usage.map(|usage| usage.input_tokens),
            output_tokens: usage.map(|usage| usage.output_tokens),
        }))
    }

    fn turn_mut(&mut self, thread_id: &str, turn_id: &str) -> Result<&mut TurnState, RuntimeError> {
        self.turns
            .get_mut(&(thread_id.to_owned(), turn_id.to_owned()))
            .ok_or(RuntimeError::CorrelationMismatch)
    }

    fn discard_notification_first_turn(&mut self, started: Option<&PendingTurnStarted>) {
        if let Some(started) = started {
            self.turns
                .remove(&(started.thread_id.clone(), started.turn_id.clone()));
        }
    }
}

fn side_effecting_item_type_marker(item_type: &str) -> String {
    let item_type_sha256 = Sha256::digest(item_type.as_bytes())
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<String>();
    format!("steadymux-owned-child: side-effecting-item-type-sha256={item_type_sha256}")
}

fn emit_side_effecting_item_type_marker(item_type: &str) {
    eprintln!("{}", side_effecting_item_type_marker(item_type));
}

#[derive(Clone, Copy)]
enum McpToolLifecycle {
    Started,
    Completed,
}

fn mcp_tool_policy_refusal_marker(item: &Value, lifecycle: McpToolLifecycle) -> String {
    let lifecycle = match lifecycle {
        McpToolLifecycle::Started => "started",
        McpToolLifecycle::Completed => "completed",
    };
    let status = match item.get("status").and_then(Value::as_str) {
        Some("inProgress") => "in-progress",
        Some("completed") => "completed",
        Some("failed") => "failed",
        _ => "unknown",
    };
    let server = mcp_tool_identity_commitment(item.get("server").and_then(Value::as_str));
    let tool = mcp_tool_identity_commitment(item.get("tool").and_then(Value::as_str));
    format!(
        "steadymux-owned-child: side-effecting-item-refused=mcp-tool-policy \
         lifecycle={lifecycle} status={status} server-sha256={server} tool-sha256={tool}"
    )
}

fn mcp_tool_identity_commitment(value: Option<&str>) -> String {
    let Some(value) = value else {
        return "missing-or-invalid".to_owned();
    };
    if validated_tool_name(value).is_err() {
        return "missing-or-invalid".to_owned();
    }
    Sha256::digest(value.as_bytes())
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

fn emit_mcp_tool_policy_refusal_marker(item: &Value, lifecycle: McpToolLifecycle) {
    eprintln!("{}", mcp_tool_policy_refusal_marker(item, lifecycle));
}

fn encode_line(value: &Value) -> Result<String, RuntimeError> {
    let mut encoded = serde_json::to_string(value).map_err(|_| RuntimeError::InvalidInput)?;
    encoded.push('\n');
    Ok(encoded)
}

fn validated_id(value: &str) -> Result<String, RuntimeError> {
    if value.is_empty()
        || value.len() > MAX_ID_BYTES
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
    {
        return Err(RuntimeError::InvalidInput);
    }
    Ok(value.to_owned())
}

fn bounded_text(value: &str) -> Result<String, RuntimeError> {
    if value.len() > MAX_ID_BYTES || value.chars().any(char::is_control) {
        return Err(RuntimeError::InvalidInput);
    }
    Ok(value.to_owned())
}

fn bounded_output(value: &str) -> Result<String, RuntimeError> {
    if value.len() > MAX_OUTPUT_BYTES {
        return Err(RuntimeError::OutputLimitExceeded);
    }
    Ok(value.to_owned())
}

fn validated_instruction(value: Option<&str>) -> Result<Option<String>, RuntimeError> {
    let Some(value) = value else {
        return Ok(None);
    };
    if value.len() > MAX_TEXT_BYTES || value.contains('\0') {
        return Err(RuntimeError::InvalidInput);
    }
    if looks_like_secret(value) {
        return Err(RuntimeError::SensitiveInputRejected);
    }
    Ok(Some(value.to_owned()))
}

fn validated_tool_name(value: &str) -> Result<String, RuntimeError> {
    if value.is_empty() || value.len() > MAX_ID_BYTES || value.chars().any(char::is_control) {
        return Err(RuntimeError::InvalidInput);
    }
    Ok(value.to_owned())
}

fn valid_opaque_control_value(value: Option<&Value>) -> bool {
    value.and_then(Value::as_str).is_some_and(|value| {
        !value.is_empty() && value.len() <= MAX_ID_BYTES && !value.chars().any(char::is_control)
    })
}

fn validate_ephemeral_empty_thread(value: &Value) -> Result<String, RuntimeError> {
    const REQUIRED: [&str; 12] = [
        "cliVersion",
        "createdAt",
        "cwd",
        "ephemeral",
        "id",
        "modelProvider",
        "preview",
        "sessionId",
        "source",
        "status",
        "turns",
        "updatedAt",
    ];
    const OPTIONAL: [&str; 14] = [
        "agentNickname",
        "agentRole",
        "canAcceptDirectInput",
        "extra",
        "forkedFromId",
        "gitInfo",
        "historyMode",
        "name",
        "parentThreadId",
        "path",
        "recencyAt",
        "section",
        "sectionEnteredAt",
        "threadSource",
    ];
    const NULL_ONLY: [&str; 11] = [
        "agentNickname",
        "agentRole",
        "extra",
        "forkedFromId",
        "gitInfo",
        "name",
        "parentThreadId",
        "path",
        "section",
        "sectionEnteredAt",
        "threadSource",
    ];

    let thread = value.as_object().ok_or(RuntimeError::ProtocolFailure)?;
    let missing_keys = REQUIRED
        .iter()
        .filter(|key| !thread.contains_key(**key))
        .copied()
        .collect::<Vec<_>>();
    let unknown_keys = thread
        .keys()
        .filter(|key| !REQUIRED.contains(&key.as_str()) && !OPTIONAL.contains(&key.as_str()))
        .map(String::as_str)
        .collect::<Vec<_>>();
    let refusal = if !missing_keys.is_empty() || !unknown_keys.is_empty() {
        for key in missing_keys.iter().chain(unknown_keys.iter()) {
            let key_sha256 = Sha256::digest(key.as_bytes())
                .iter()
                .map(|byte| format!("{byte:02x}"))
                .collect::<String>();
            eprintln!("steadymux-owned-child: thread-started-key-sha256={key_sha256}");
        }
        eprintln!(
            "steadymux-owned-child: thread-started-key-counts=missing-{}-unknown-{}",
            missing_keys.len(),
            unknown_keys.len()
        );
        Some("keys")
    } else if NULL_ONLY
        .iter()
        .any(|key| thread.get(*key).is_some_and(|value| !value.is_null()))
    {
        Some("null-only")
    } else if thread.get("ephemeral").and_then(Value::as_bool) != Some(true)
        || thread.get("modelProvider").and_then(Value::as_str) != Some("openai")
        || !matches!(
            thread.get("source").and_then(Value::as_str),
            Some("appServer" | "vscode")
        )
    {
        Some("core")
    } else if thread.get("cliVersion").and_then(Value::as_str)
        != Some(CURRENT_STANDALONE_CLI_VERSION)
    {
        Some("version")
    } else if !valid_idle_thread_status(thread.get("status")) {
        Some("status")
    } else if thread
        .get("turns")
        .and_then(Value::as_array)
        .is_none_or(|turns| !turns.is_empty())
    {
        Some("turns")
    } else if thread
        .get("preview")
        .and_then(Value::as_str)
        .is_none_or(|preview| preview.len() > MAX_TEXT_BYTES)
    {
        Some("preview")
    } else if !valid_thread_path(thread.get("cwd")) {
        Some("cwd")
    } else if thread.get("recencyAt").is_some_and(|value| {
        !value.is_null() && value.as_i64().is_none_or(|timestamp| timestamp < 0)
    }) || thread
        .get("canAcceptDirectInput")
        .is_some_and(|value| value.as_bool() != Some(true))
        || thread
            .get("historyMode")
            .is_some_and(|value| !matches!(value.as_str(), Some("legacy" | "paginated")))
    {
        Some("optional")
    } else {
        None
    };
    if let Some(class) = refusal {
        eprintln!("steadymux-owned-child: thread-started-refused={class}");
        return Err(RuntimeError::ProtocolFailure);
    }

    let created_at = thread
        .get("createdAt")
        .and_then(Value::as_i64)
        .filter(|timestamp| *timestamp >= 0)
        .ok_or_else(|| {
            eprintln!("steadymux-owned-child: thread-started-refused=timestamps");
            RuntimeError::ProtocolFailure
        })?;
    thread
        .get("updatedAt")
        .and_then(Value::as_i64)
        .filter(|timestamp| *timestamp >= created_at)
        .ok_or_else(|| {
            eprintln!("steadymux-owned-child: thread-started-refused=timestamps");
            RuntimeError::ProtocolFailure
        })?;
    validated_id(
        thread
            .get("sessionId")
            .and_then(Value::as_str)
            .ok_or_else(|| {
                eprintln!("steadymux-owned-child: thread-started-refused=ids");
                RuntimeError::ProtocolFailure
            })?,
    )?;
    validated_id(thread.get("id").and_then(Value::as_str).ok_or_else(|| {
        eprintln!("steadymux-owned-child: thread-started-refused=ids");
        RuntimeError::ProtocolFailure
    })?)
}

fn valid_idle_thread_status(value: Option<&Value>) -> bool {
    value.and_then(Value::as_object).is_some_and(|status| {
        status.len() == 1 && status.get("type").and_then(Value::as_str) == Some("idle")
    })
}

fn valid_thread_path(value: Option<&Value>) -> bool {
    let Some(value) = value.and_then(Value::as_str) else {
        return false;
    };
    if value.is_empty()
        || value.len() > MAX_THREAD_PATH_BYTES
        || value.chars().any(char::is_control)
    {
        return false;
    }
    let path = Path::new(value);
    path.is_absolute()
        && path
            .components()
            .all(|component| matches!(component, Component::RootDir | Component::Normal(_)))
}

fn valid_json_rpc_error_response(value: &Value) -> bool {
    let Some(response) = value.as_object() else {
        return false;
    };
    if !(response.len() == 2 || response.len() == 3)
        || !response.contains_key("id")
        || !response.contains_key("error")
        || !response
            .keys()
            .all(|key| matches!(key.as_str(), "id" | "error" | "jsonrpc"))
        || response
            .get("jsonrpc")
            .is_some_and(|version| version.as_str() != Some("2.0"))
    {
        return false;
    }
    let Some(error) = response.get("error").and_then(Value::as_object) else {
        return false;
    };
    (error.len() == 2 || error.len() == 3)
        && error.get("code").and_then(Value::as_i64).is_some()
        && error.get("message").and_then(Value::as_str).is_some()
        && error
            .keys()
            .all(|key| matches!(key.as_str(), "code" | "message" | "data"))
}

fn exact_object<'a>(
    value: &'a Value,
    expected_keys: &[&str],
) -> Result<&'a serde_json::Map<String, Value>, RuntimeError> {
    let object = value.as_object().ok_or(RuntimeError::ProtocolFailure)?;
    if object.len() != expected_keys.len()
        || expected_keys.iter().any(|key| !object.contains_key(*key))
    {
        return Err(RuntimeError::ProtocolFailure);
    }
    Ok(object)
}

fn correlated_reasoning_ids(
    object: &serde_json::Map<String, Value>,
    index_key: &str,
) -> Result<(String, String, String), RuntimeError> {
    let thread_id = validated_id(
        object
            .get("threadId")
            .and_then(Value::as_str)
            .ok_or(RuntimeError::ProtocolFailure)?,
    )?;
    let turn_id = validated_id(
        object
            .get("turnId")
            .and_then(Value::as_str)
            .ok_or(RuntimeError::ProtocolFailure)?,
    )?;
    let item_id = validated_id(
        object
            .get("itemId")
            .and_then(Value::as_str)
            .ok_or(RuntimeError::ProtocolFailure)?,
    )?;
    let valid_index = object
        .get(index_key)
        .and_then(Value::as_u64)
        .is_some_and(|index| index < MAX_ITEMS_PER_TURN as u64);
    if !valid_index {
        return Err(RuntimeError::ProtocolFailure);
    }
    Ok((thread_id, turn_id, item_id))
}

fn token_usage_refusal(class: &'static str) -> RuntimeError {
    eprintln!("steadymux-owned-child: token-usage-refused={class}");
    RuntimeError::ProtocolFailure
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum TokenUsageCorrelationRefusal {
    MissingPair,
    Rerouted,
    PriorUsage,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum PriorUsageDirection {
    Unavailable,
    Same,
    MonotonicNondecreasing,
    DecreaseOrMixed,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum TotalUsageRelativeToLast {
    Equal,
    TotalNondecreasing,
    TotalDecreaseOrMixed,
}

impl TotalUsageRelativeToLast {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Equal => "equal",
            Self::TotalNondecreasing => "total-nondecreasing",
            Self::TotalDecreaseOrMixed => "total-decrease-or-mixed",
        }
    }
}

impl PriorUsageDirection {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Unavailable => "unavailable",
            Self::Same => "same",
            Self::MonotonicNondecreasing => "monotonic-nondecreasing",
            Self::DecreaseOrMixed => "decrease-or-mixed",
        }
    }
}

fn prior_usage_direction(
    previous: TurnUsageCounts,
    incoming: TurnUsageCounts,
) -> PriorUsageDirection {
    if previous == incoming {
        PriorUsageDirection::Same
    } else if incoming.input_tokens >= previous.input_tokens
        && incoming.output_tokens >= previous.output_tokens
    {
        PriorUsageDirection::MonotonicNondecreasing
    } else {
        PriorUsageDirection::DecreaseOrMixed
    }
}

fn prior_usage_direction_marker(direction: PriorUsageDirection) -> String {
    format!(
        "steadymux-owned-child: token-usage-prior-direction={}",
        direction.as_str()
    )
}

fn emit_prior_usage_direction_marker(direction: PriorUsageDirection) {
    eprintln!("{}", prior_usage_direction_marker(direction));
}

fn total_usage_relative_to_last(
    last: TurnUsageCounts,
    total: CumulativeTurnUsageCounts,
) -> TotalUsageRelativeToLast {
    if total.input_tokens == last.input_tokens && total.output_tokens == last.output_tokens {
        TotalUsageRelativeToLast::Equal
    } else if total.input_tokens >= last.input_tokens && total.output_tokens >= last.output_tokens {
        TotalUsageRelativeToLast::TotalNondecreasing
    } else {
        TotalUsageRelativeToLast::TotalDecreaseOrMixed
    }
}

fn cumulative_usage_direction(
    previous: CumulativeTurnUsageCounts,
    incoming: CumulativeTurnUsageCounts,
) -> PriorUsageDirection {
    if previous == incoming {
        PriorUsageDirection::Same
    } else if incoming.cached_input_tokens >= previous.cached_input_tokens
        && cache_write_non_decreasing(
            previous.cache_write_input_tokens,
            incoming.cache_write_input_tokens,
        )
        && incoming.input_tokens >= previous.input_tokens
        && incoming.output_tokens >= previous.output_tokens
        && incoming.reasoning_output_tokens >= previous.reasoning_output_tokens
        && incoming.total_tokens >= previous.total_tokens
    {
        PriorUsageDirection::MonotonicNondecreasing
    } else {
        PriorUsageDirection::DecreaseOrMixed
    }
}

/// `cacheWriteInputTokens` is an optional cumulative field. It is compared
/// only when both frames report it; an omitted value is not treated as a
/// regression against a previously reported positive count.
fn cache_write_non_decreasing(previous: Option<u64>, incoming: Option<u64>) -> bool {
    match (previous, incoming) {
        (Some(previous), Some(incoming)) => incoming >= previous,
        _ => true,
    }
}

fn prior_usage_projection_marker(
    total_relative_to_last: TotalUsageRelativeToLast,
    total_direction: PriorUsageDirection,
) -> String {
    format!(
        "steadymux-owned-child: token-usage-prior-projections total-vs-last={} total-vs-prior={}",
        total_relative_to_last.as_str(),
        total_direction.as_str(),
    )
}

fn emit_prior_usage_projection_marker(
    total_relative_to_last: TotalUsageRelativeToLast,
    total_direction: PriorUsageDirection,
) {
    eprintln!(
        "{}",
        prior_usage_projection_marker(total_relative_to_last, total_direction)
    );
}

impl TokenUsageCorrelationRefusal {
    const fn as_str(self) -> &'static str {
        match self {
            Self::MissingPair => "missing-pair",
            Self::Rerouted => "rerouted",
            Self::PriorUsage => "prior-usage",
        }
    }
}

fn emit_token_usage_correlation_marker(
    reason: TokenUsageCorrelationRefusal,
    incoming_thread_id: &str,
    incoming_turn_id: &str,
    turns: &BTreeMap<(String, String), TurnState>,
) {
    eprintln!(
        "{}",
        token_usage_correlation_marker(reason, incoming_thread_id, incoming_turn_id, turns)
    );
}

fn token_usage_correlation_marker(
    reason: TokenUsageCorrelationRefusal,
    incoming_thread_id: &str,
    incoming_turn_id: &str,
    turns: &BTreeMap<(String, String), TurnState>,
) -> String {
    fn sha256_hex(value: &[u8]) -> String {
        Sha256::digest(value)
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect()
    }

    fn update_id(hasher: &mut Sha256, id: &str) {
        hasher.update((id.len() as u64).to_be_bytes());
        hasher.update(id.as_bytes());
    }

    fn id_commitment(domain: &[u8], ids: impl IntoIterator<Item = String>) -> String {
        let mut digest = Sha256::new();
        digest.update(domain);
        for id in ids {
            update_id(&mut digest, &id);
        }
        digest
            .finalize()
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect()
    }

    let mut known_pairs = Sha256::new();
    known_pairs.update(b"steadymux.token-usage-known-pairs.v1\0");
    for (thread_id, turn_id) in turns.keys() {
        update_id(&mut known_pairs, thread_id);
        update_id(&mut known_pairs, turn_id);
    }
    let known_pairs_sha256 = known_pairs
        .finalize()
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<String>();
    let known_active_threads_sha256 = id_commitment(
        b"steadymux.token-usage-known-active-threads.v1\0",
        turns
            .keys()
            .map(|(thread_id, _)| thread_id.clone())
            .collect::<BTreeSet<_>>(),
    );
    let known_active_turns_sha256 = id_commitment(
        b"steadymux.token-usage-known-active-turns.v1\0",
        turns
            .keys()
            .map(|(_, turn_id)| turn_id.clone())
            .collect::<BTreeSet<_>>(),
    );

    format!(
        "steadymux-owned-child: token-usage-correlation reason={} incoming-thread-sha256={} incoming-turn-sha256={} known-active-threads-sha256={known_active_threads_sha256} known-active-turns-sha256={known_active_turns_sha256} known-active-pairs-sha256={known_pairs_sha256} known-active-pair-count={}",
        reason.as_str(),
        sha256_hex(incoming_thread_id.as_bytes()),
        sha256_hex(incoming_turn_id.as_bytes()),
        turns.len(),
    )
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum TokenBreakdownValidation {
    Valid,
    Shape,
    Keys,
    Counts,
}

fn validated_last_turn_usage(value: Option<&Value>) -> Result<TurnUsageCounts, RuntimeError> {
    let Some(usage) = value.and_then(Value::as_object) else {
        return Err(token_usage_refusal("usage-keys"));
    };
    if !(usage.len() == 2 || usage.len() == 3)
        || !usage.contains_key("last")
        || !usage.contains_key("total")
        || !usage
            .keys()
            .all(|key| matches!(key.as_str(), "last" | "total" | "modelContextWindow"))
    {
        return Err(token_usage_refusal("usage-keys"));
    }
    match valid_token_breakdown(usage.get("last")) {
        TokenBreakdownValidation::Valid => {}
        TokenBreakdownValidation::Shape => return Err(token_usage_refusal("last-shape")),
        TokenBreakdownValidation::Keys => {
            emit_token_breakdown_key_diagnostics(usage.get("last"));
            return Err(token_usage_refusal("last-counts"));
        }
        TokenBreakdownValidation::Counts => return Err(token_usage_refusal("last-counts")),
    }
    match valid_token_breakdown(usage.get("total")) {
        TokenBreakdownValidation::Valid => {}
        TokenBreakdownValidation::Keys => {
            emit_token_breakdown_key_diagnostics(usage.get("total"));
            return Err(token_usage_refusal("total-shape"));
        }
        TokenBreakdownValidation::Shape | TokenBreakdownValidation::Counts => {
            return Err(token_usage_refusal("total-shape"));
        }
    }
    if !usage
        .get("modelContextWindow")
        .is_none_or(|value| value.is_null() || value.as_i64().is_some_and(|window| window >= 0))
    {
        return Err(token_usage_refusal("context-window"));
    }
    let last = usage
        .get("last")
        .and_then(Value::as_object)
        .ok_or_else(|| token_usage_refusal("last-shape"))?;
    let input_tokens = last
        .get("inputTokens")
        .and_then(Value::as_i64)
        .and_then(|tokens| u64::try_from(tokens).ok())
        .ok_or_else(|| token_usage_refusal("last-counts"))?;
    let output_tokens = last
        .get("outputTokens")
        .and_then(Value::as_i64)
        .and_then(|tokens| u64::try_from(tokens).ok())
        .ok_or_else(|| token_usage_refusal("last-counts"))?;
    Ok(TurnUsageCounts {
        input_tokens,
        output_tokens,
    })
}

/// Extracts the complete validated cumulative projection. This remains private
/// runtime state: completion receipts deliberately expose only the latest
/// per-turn `last` projection.
fn validated_total_turn_usage(
    value: Option<&Value>,
) -> Result<CumulativeTurnUsageCounts, RuntimeError> {
    let total = value
        .and_then(Value::as_object)
        .and_then(|usage| usage.get("total"))
        .and_then(Value::as_object)
        .ok_or_else(|| token_usage_refusal("total-shape"))?;
    let count = |key| {
        total
            .get(key)
            .and_then(Value::as_i64)
            .and_then(|tokens| u64::try_from(tokens).ok())
            .ok_or_else(|| token_usage_refusal("total-shape"))
    };
    Ok(CumulativeTurnUsageCounts {
        cached_input_tokens: count("cachedInputTokens")?,
        cache_write_input_tokens: total
            .get("cacheWriteInputTokens")
            .map(|value| {
                value
                    .as_i64()
                    .and_then(|tokens| u64::try_from(tokens).ok())
                    .ok_or_else(|| token_usage_refusal("total-shape"))
            })
            .transpose()?,
        input_tokens: count("inputTokens")?,
        output_tokens: count("outputTokens")?,
        reasoning_output_tokens: count("reasoningOutputTokens")?,
        total_tokens: count("totalTokens")?,
    })
}

fn valid_token_breakdown(value: Option<&Value>) -> TokenBreakdownValidation {
    const REQUIRED: [&str; 5] = [
        "cachedInputTokens",
        "inputTokens",
        "outputTokens",
        "reasoningOutputTokens",
        "totalTokens",
    ];
    let Some(breakdown) = value.and_then(Value::as_object) else {
        return TokenBreakdownValidation::Shape;
    };
    if !(breakdown.len() == REQUIRED.len() || breakdown.len() == REQUIRED.len() + 1) {
        return TokenBreakdownValidation::Keys;
    }
    if !breakdown
        .keys()
        .all(|key| REQUIRED.contains(&key.as_str()) || key.as_str() == "cacheWriteInputTokens")
    {
        return TokenBreakdownValidation::Keys;
    }
    if REQUIRED.iter().any(|key| {
        breakdown
            .get(*key)
            .and_then(Value::as_i64)
            .is_none_or(|tokens| tokens < 0)
    }) {
        return TokenBreakdownValidation::Counts;
    }
    if !breakdown
        .get("cacheWriteInputTokens")
        .is_none_or(|value| value.as_i64().is_some_and(|tokens| tokens >= 0))
    {
        return TokenBreakdownValidation::Counts;
    }
    TokenBreakdownValidation::Valid
}

fn emit_token_breakdown_key_diagnostics(value: Option<&Value>) {
    const REQUIRED: [&str; 5] = [
        "cachedInputTokens",
        "inputTokens",
        "outputTokens",
        "reasoningOutputTokens",
        "totalTokens",
    ];
    let Some(breakdown) = value.and_then(Value::as_object) else {
        return;
    };
    let required = REQUIRED.into_iter().collect::<BTreeSet<_>>();
    let allowed = required
        .iter()
        .copied()
        .chain(std::iter::once("cacheWriteInputTokens"))
        .collect::<BTreeSet<_>>();
    let actual = breakdown
        .keys()
        .map(String::as_str)
        .collect::<BTreeSet<_>>();
    let missing = required.difference(&actual).copied().collect::<Vec<_>>();
    let unknown = actual.difference(&allowed).copied().collect::<Vec<_>>();
    for key in missing.iter().chain(unknown.iter()) {
        let key_sha256 = Sha256::digest(key.as_bytes())
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect::<String>();
        eprintln!("steadymux-owned-child: token-usage-breakdown-key-sha256={key_sha256}");
    }
    eprintln!(
        "steadymux-owned-child: token-usage-breakdown-key-counts=missing-{}-unknown-{}",
        missing.len(),
        unknown.len()
    );
}

fn validated_notification_method(value: &str) -> Result<String, RuntimeError> {
    if value.is_empty()
        || value.len() > MAX_ID_BYTES
        || !value.bytes().all(|byte| {
            byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'_' | b'.' | b':')
        })
    {
        return Err(RuntimeError::ProtocolFailure);
    }
    Ok(value.to_owned())
}

/// Return a fixed, content-free provenance marker for a response whose pending
/// request did not correlate. The response ID and payload intentionally never
/// enter this diagnostic surface.
fn correlation_mismatch_response_marker() -> &'static str {
    "steadymux-owned-child: correlation-mismatch source=response class=correlation"
}

/// Return a fixed-class, content-free provenance marker for a notification
/// whose state transition did not correlate. The app-server method itself is
/// committed only by SHA-256 so logs cannot disclose server protocol details.
fn correlation_mismatch_notification_marker(method: &str) -> String {
    let handler = match method {
        "thread/tokenUsage/updated" => "usage",
        "thread/started" | "thread/status/changed" => "thread",
        "turn/started" | "turn/completed" => "turn",
        method if method.starts_with("item/") => "item",
        method if method.starts_with("hook/") => "hook",
        method if method.starts_with("mcpServer/") => "mcp",
        _ => "control",
    };
    let method_sha256 = Sha256::digest(method.as_bytes())
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<String>();
    format!(
        "steadymux-owned-child: correlation-mismatch source=notification handler={handler} mismatch=correlation method-sha256={method_sha256}"
    )
}

fn digest_json(value: &Value) -> Result<[u8; 32], RuntimeError> {
    let encoded = serde_json::to_vec(value).map_err(|_| RuntimeError::InvalidInput)?;
    Ok(Sha256::digest(encoded).into())
}

fn app_server_version_from_user_agent(user_agent: &str) -> Option<&str> {
    let identity = user_agent.split_ascii_whitespace().next()?;
    let (originator, version) = identity.split_once('/')?;
    if originator != "steadymux"
        || version.is_empty()
        || version.len() > MAX_ID_BYTES
        || version.chars().any(char::is_control)
    {
        return None;
    }
    Some(version)
}

fn null_or_absent(value: &Value, key: &str) -> bool {
    value.get(key).is_none_or(Value::is_null)
}

fn non_null(value: &Value, key: &str) -> bool {
    value.get(key).is_some_and(|field| !field.is_null())
}

fn parse_terminal_failure(error: &Value) -> TerminalFailureClass {
    let Some(info) = error.get("codexErrorInfo") else {
        return TerminalFailureClass::Unclassified;
    };
    let discriminator = match info {
        Value::String(value) => Some(value.as_str()),
        Value::Object(object) => object
            .get("type")
            .and_then(Value::as_str)
            .or_else(|| object.keys().next().map(String::as_str)),
        _ => None,
    };
    let Some(discriminator) = discriminator else {
        return TerminalFailureClass::Other;
    };
    let normalized = discriminator.to_ascii_lowercase();
    if normalized.contains("contextwindow") {
        TerminalFailureClass::ContextWindow
    } else if normalized.contains("usage")
        || normalized.contains("budget")
        || normalized.contains("ratelimit")
    {
        TerminalFailureClass::UsageLimit
    } else if normalized.contains("connection")
        || normalized.contains("stream")
        || normalized.contains("http")
    {
        TerminalFailureClass::Transport
    } else if normalized.contains("badrequest") || normalized.contains("invalidrequest") {
        TerminalFailureClass::InvalidRequest
    } else if normalized.contains("unauthorized") || normalized.contains("authentication") {
        TerminalFailureClass::Unauthorized
    } else if normalized.contains("sandbox") || normalized.contains("permission") {
        TerminalFailureClass::Sandbox
    } else if normalized.contains("internal") {
        TerminalFailureClass::Internal
    } else {
        TerminalFailureClass::Other
    }
}

fn string_at<'a>(value: &'a Value, path: &[&str]) -> Result<&'a str, RuntimeError> {
    let mut cursor = value;
    for key in path {
        cursor = cursor.get(*key).ok_or(RuntimeError::InvalidInput)?;
    }
    cursor.as_str().ok_or(RuntimeError::InvalidInput)
}

fn parse_effort(value: &str) -> Result<ReasoningEffort, RuntimeError> {
    match value {
        "minimal" => Ok(ReasoningEffort::Minimal),
        "low" => Ok(ReasoningEffort::Low),
        "medium" => Ok(ReasoningEffort::Medium),
        "high" => Ok(ReasoningEffort::High),
        "xhigh" => Ok(ReasoningEffort::Xhigh),
        "max" => Ok(ReasoningEffort::Max),
        "ultra" => Ok(ReasoningEffort::Ultra),
        _ => Err(RuntimeError::InvalidInput),
    }
}

fn parse_window(value: Option<&Value>) -> Result<Option<RateLimitWindowProjection>, RuntimeError> {
    let Some(value) = value else { return Ok(None) };
    if value.is_null() {
        return Ok(None);
    }
    let used_percent = value
        .get("usedPercent")
        .and_then(Value::as_u64)
        .ok_or(RuntimeError::InvalidInput)?;
    if used_percent > 100 {
        return Err(RuntimeError::InvalidInput);
    }
    Ok(Some(RateLimitWindowProjection {
        used_percent: used_percent as u8,
        window_duration_minutes: value.get("windowDurationMins").and_then(Value::as_u64),
        resets_at_unix_seconds: value.get("resetsAt").and_then(Value::as_u64),
    }))
}

fn contains_forbidden_sensitive_material(value: &Value) -> bool {
    match value {
        Value::Object(object) => object.iter().any(|(key, value)| {
            let normalized = key.to_ascii_lowercase();
            TOKEN_KEYS
                .iter()
                .any(|forbidden| normalized.contains(forbidden))
                || contains_forbidden_sensitive_material(value)
        }),
        Value::Array(values) => values.iter().any(contains_forbidden_sensitive_material),
        Value::String(value) => looks_like_secret(value),
        _ => false,
    }
}

fn looks_like_secret(value: &str) -> bool {
    let lower = value.to_ascii_lowercase();
    lower.contains("bearer ")
        || lower.contains("-----begin private key-----")
        || lower.contains("-----begin rsa private key-----")
        || lower.contains("github_pat_")
        || lower.contains("xoxb-")
        || lower.contains("xoxp-")
        || value
            .split(|character: char| character.is_whitespace() || matches!(character, '"' | '\''))
            .any(|token| {
                (token.starts_with("sk-") && token.len() >= 20)
                    || (token.starts_with("AKIA")
                        && token.len() == 20
                        && token
                            .bytes()
                            .all(|byte| byte.is_ascii_uppercase() || byte.is_ascii_digit()))
            })
}

#[cfg(test)]
mod notification_tests {
    use super::*;
    use serde_json::json;

    fn ingest(runtime: &mut CodexAgentRuntime, value: Value) -> Result<RuntimeEvent, RuntimeError> {
        runtime.ingest_line(
            &serde_json::to_vec(&value).expect("serialize app-server notification fixture"),
        )
    }

    fn active_reasoning_runtime() -> CodexAgentRuntime {
        let thread_id = "thr-notification".to_owned();
        let turn_id = "turn-notification".to_owned();
        let item_id = "reasoning-notification".to_owned();
        let mut threads = BTreeMap::new();
        threads.insert(
            thread_id.clone(),
            ThreadSelection {
                model: "gpt-5.6-luna".to_owned(),
                effort: ReasoningEffort::Low,
            },
        );
        let mut items = BTreeMap::new();
        items.insert(
            item_id,
            ItemState {
                kind: ItemKind::Reasoning,
                delta_bytes: 0,
                completed: None,
                lifecycle_complete: false,
            },
        );
        let mut turns = BTreeMap::new();
        turns.insert(
            (thread_id, turn_id),
            TurnState {
                model: "gpt-5.6-luna".to_owned(),
                effort: ReasoningEffort::Low,
                items,
                final_item: None,
                visible_output: false,
                tool_or_side_effect: false,
                model_rerouted: false,
                interrupt_requested: false,
                lifecycle_started: true,
                output_bytes: 0,
                usage: None,
                total_usage: None,
            },
        );
        CodexAgentRuntime {
            tool_policy: CodexToolPolicy::RejectAll,
            handshake: HandshakeState::Ready,
            next_id: 1,
            pending: BTreeMap::new(),
            pending_turn_started: BTreeMap::new(),
            models: BTreeMap::new(),
            threads,
            turns,
            initialized_server_version: None,
        }
    }

    fn pending_turn_runtime() -> CodexAgentRuntime {
        let mut runtime = active_reasoning_runtime();
        runtime.pending.insert(
            1,
            PendingRequest::TurnStart {
                thread_id: "thr-notification".to_owned(),
                model: "gpt-5.6-luna".to_owned(),
                effort: ReasoningEffort::Low,
            },
        );
        runtime
    }

    fn notification_first_runtime() -> CodexAgentRuntime {
        let mut runtime = pending_turn_runtime();
        runtime.turns.clear();
        runtime
    }

    fn completable_runtime() -> CodexAgentRuntime {
        let mut runtime = active_reasoning_runtime();
        let turn = runtime
            .turns
            .get_mut(&(
                "thr-notification".to_owned(),
                "turn-notification".to_owned(),
            ))
            .expect("active turn fixture");
        turn.items.clear();
        runtime
    }

    fn token_usage_notification(input_tokens: i64, output_tokens: i64) -> Value {
        token_usage_notification_with_total(input_tokens, output_tokens, 123456781, 123456782)
    }

    fn token_usage_notification_with_total(
        input_tokens: i64,
        output_tokens: i64,
        total_input_tokens: i64,
        total_output_tokens: i64,
    ) -> Value {
        json!({
            "method": "thread/tokenUsage/updated",
            "params": {
                "threadId": "thr-notification",
                "turnId": "turn-notification",
                "tokenUsage": {
                    "last": {
                        "cachedInputTokens": 3,
                        "inputTokens": input_tokens,
                        "outputTokens": output_tokens,
                        "reasoningOutputTokens": 5,
                        "totalTokens": input_tokens + output_tokens
                    },
                    "total": {
                        "cachedInputTokens": 123456780,
                        "inputTokens": total_input_tokens,
                        "outputTokens": total_output_tokens,
                        "reasoningOutputTokens": 123456783,
                        "totalTokens": 987654321
                    },
                    "modelContextWindow": 128000
                }
            }
        })
    }

    fn cumulative_usage(input_tokens: u64, output_tokens: u64) -> CumulativeTurnUsageCounts {
        CumulativeTurnUsageCounts {
            cached_input_tokens: 100,
            cache_write_input_tokens: Some(0),
            input_tokens,
            output_tokens,
            reasoning_output_tokens: 5,
            total_tokens: input_tokens + output_tokens,
        }
    }

    fn realistic_thread_started(thread_id: &str) -> Value {
        json!({
            "method": "thread/started",
            "params": {
                "thread": {
                    "id": thread_id,
                    "preview": "private-preview-canary",
                    "ephemeral": true,
                    "modelProvider": "openai",
                    "createdAt": 1,
                    "updatedAt": 1,
                    "status": {"type": "idle"},
                    "cwd": "/private-cwd-canary",
                    "cliVersion": CURRENT_STANDALONE_CLI_VERSION,
                    "source": "appServer",
                    "sessionId": "private-session-canary",
                    "turns": [],
                    "agentNickname": null,
                    "agentRole": null,
                    "canAcceptDirectInput": true,
                    "extra": null,
                    "forkedFromId": null,
                    "gitInfo": null,
                    "historyMode": "legacy",
                    "name": null,
                    "parentThreadId": null,
                    "path": null,
                    "recencyAt": 1,
                    "section": null,
                    "sectionEnteredAt": null,
                    "threadSource": null
                }
            }
        })
    }

    fn turn_started(thread_id: &str, turn_id: &str) -> Value {
        json!({
            "method": "turn/started",
            "params": {
                "threadId": thread_id,
                "turn": {
                    "id": turn_id,
                    "status": "inProgress",
                    "items": [],
                    "error": null
                }
            }
        })
    }

    fn turn_start_response(request_id: u64, turn_id: &str) -> Value {
        json!({
            "id": request_id,
            "result": {
                "turn": {
                    "id": turn_id,
                    "status": "inProgress",
                    "items": [],
                    "error": null
                }
            }
        })
    }

    #[test]
    fn correlation_mismatch_provenance_is_fixed_class_and_content_free() {
        assert_eq!(
            correlation_mismatch_response_marker(),
            "steadymux-owned-child: correlation-mismatch source=response class=correlation"
        );

        let method = "thread/tokenUsage/updated";
        let marker = correlation_mismatch_notification_marker(method);
        let expected_sha256 = Sha256::digest(method.as_bytes())
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect::<String>();
        assert_eq!(
            marker,
            format!(
                "steadymux-owned-child: correlation-mismatch source=notification handler=usage mismatch=correlation method-sha256={expected_sha256}"
            )
        );
        assert!(!marker.contains(method));
        assert!(!marker.contains("thr-notification"));
        assert!(!marker.contains("turn-notification"));

        assert!(correlation_mismatch_notification_marker("item/started").contains("handler=item"));
        assert!(
            correlation_mismatch_notification_marker("mcpServer/startupStatus/updated")
                .contains("handler=mcp")
        );
        assert!(correlation_mismatch_notification_marker("hook/started").contains("handler=hook"));
        assert!(
            correlation_mismatch_notification_marker("remoteControl/status/changed")
                .contains("handler=control")
        );
    }

    #[test]
    fn crossed_notification_correlation_refuses_without_mutating_turn_state() {
        let mut runtime = notification_first_runtime();
        let before = format!("{runtime:?}");

        assert_eq!(
            ingest(&mut runtime, turn_started("thr-other", "turn-other")),
            Err(RuntimeError::CorrelationMismatch)
        );
        assert_eq!(format!("{runtime:?}"), before);
    }

    #[test]
    fn notification_first_turn_start_reconciles_exact_response_once() {
        let mut runtime = notification_first_runtime();
        assert_eq!(
            ingest(
                &mut runtime,
                turn_started("thr-notification", "turn-notification")
            ),
            Ok(RuntimeEvent::TurnStarted {
                thread_id: "thr-notification".to_owned(),
                turn_id: "turn-notification".to_owned()
            })
        );
        assert_eq!(runtime.pending_turn_started.len(), 1);
        assert!(
            runtime
                .turns
                .get(&(
                    "thr-notification".to_owned(),
                    "turn-notification".to_owned()
                ))
                .is_some_and(|turn| {
                    turn.model == "gpt-5.6-luna"
                        && turn.effort == ReasoningEffort::Low
                        && turn.lifecycle_started
                })
        );

        assert_eq!(
            ingest(&mut runtime, turn_start_response(1, "turn-notification")),
            Ok(RuntimeEvent::TurnAccepted {
                thread_id: "thr-notification".to_owned(),
                turn_id: "turn-notification".to_owned()
            })
        );
        assert!(runtime.pending_turn_started.is_empty());
        assert!(
            runtime
                .turns
                .get(&(
                    "thr-notification".to_owned(),
                    "turn-notification".to_owned()
                ))
                .is_some_and(|turn| turn.lifecycle_started)
        );
        assert_eq!(
            ingest(
                &mut runtime,
                turn_started("thr-notification", "turn-notification")
            ),
            Err(RuntimeError::CorrelationMismatch)
        );
    }

    #[test]
    fn response_first_turn_start_remains_valid() {
        let mut runtime = notification_first_runtime();
        assert!(matches!(
            ingest(&mut runtime, turn_start_response(1, "turn-notification")),
            Ok(RuntimeEvent::TurnAccepted { .. })
        ));
        assert!(runtime.pending_turn_started.is_empty());
        assert_eq!(
            ingest(
                &mut runtime,
                turn_started("thr-notification", "turn-notification")
            ),
            Ok(RuntimeEvent::TurnStarted {
                thread_id: "thr-notification".to_owned(),
                turn_id: "turn-notification".to_owned()
            })
        );
    }

    #[test]
    fn response_first_usage_before_turn_started_completes_exactly_once() {
        let mut runtime = notification_first_runtime();
        assert!(matches!(
            ingest(&mut runtime, turn_start_response(1, "turn-notification")),
            Ok(RuntimeEvent::TurnAccepted { .. })
        ));
        assert_eq!(
            ingest(&mut runtime, token_usage_notification(12, 34)),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored),
            "the accepted response creates the only record usage may update"
        );
        assert_eq!(
            ingest(
                &mut runtime,
                turn_started("thr-notification", "turn-notification")
            ),
            Ok(RuntimeEvent::TurnStarted {
                thread_id: "thr-notification".to_owned(),
                turn_id: "turn-notification".to_owned(),
            })
        );
        let terminal = ingest(
            &mut runtime,
            json!({
                "method": "turn/completed",
                "params": {
                    "threadId": "thr-notification",
                    "turn": {
                        "id": "turn-notification",
                        "status": "completed",
                        "items": [],
                        "error": null
                    }
                }
            }),
        )
        .expect("response-first usage becomes terminal only after turn/started");
        assert!(matches!(
            terminal,
            RuntimeEvent::TurnCompleted(TurnCompletionReceipt {
                input_tokens: Some(12),
                output_tokens: Some(34),
                ..
            })
        ));
    }

    #[test]
    fn response_first_usage_accepts_monotonic_totals_but_refuses_mismatched_turns() {
        let mut before_accepted = notification_first_runtime();
        assert_eq!(
            ingest(&mut before_accepted, token_usage_notification(12, 34)),
            Err(RuntimeError::ProtocolFailure),
            "a notification cannot create a turn"
        );

        let mut runtime = notification_first_runtime();
        ingest(&mut runtime, turn_start_response(1, "turn-notification"))
            .expect("accepted turn response");
        assert_eq!(
            ingest(&mut runtime, token_usage_notification(12, 34)),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored)
        );
        assert_eq!(
            ingest(&mut runtime, token_usage_notification(12, 34)),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored),
            "an identical authoritative duplicate is idempotent"
        );
        assert_eq!(
            ingest(
                &mut runtime,
                token_usage_notification_with_total(13, 34, 123456782, 123456783),
            ),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored),
            "a monotonic cumulative projection accepts the latest per-turn usage even before lifecycle start"
        );

        let mut mismatched = notification_first_runtime();
        ingest(&mut mismatched, turn_start_response(1, "turn-notification"))
            .expect("accepted turn response");
        let mut wrong_ids = token_usage_notification(12, 34);
        wrong_ids["params"]["turnId"] = json!("turn-other");
        assert_eq!(
            ingest(&mut mismatched, wrong_ids),
            Err(RuntimeError::ProtocolFailure),
            "usage must bind to the accepted response turn"
        );
        assert_eq!(
            ingest(&mut mismatched, token_usage_notification(12, 34)),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored),
            "a mismatched update cannot consume the accepted turn's usage slot"
        );
    }

    #[test]
    fn duplicate_notification_first_turn_start_is_rejected() {
        let mut runtime = notification_first_runtime();
        ingest(
            &mut runtime,
            turn_started("thr-notification", "turn-notification"),
        )
        .expect("first exactly correlated notification");
        assert_eq!(
            ingest(
                &mut runtime,
                turn_started("thr-notification", "turn-notification")
            ),
            Err(RuntimeError::CorrelationMismatch)
        );
        assert_eq!(runtime.pending_turn_started.len(), 1);
    }

    #[test]
    fn notification_first_requires_exactly_one_pending_turn_start() {
        let mut runtime = notification_first_runtime();
        runtime.pending.insert(
            2,
            PendingRequest::TurnStart {
                thread_id: "thr-notification".to_owned(),
                model: "gpt-5.6-luna".to_owned(),
                effort: ReasoningEffort::Low,
            },
        );
        assert_eq!(
            ingest(
                &mut runtime,
                turn_started("thr-notification", "turn-notification")
            ),
            Err(RuntimeError::CorrelationMismatch)
        );
        assert!(runtime.pending_turn_started.is_empty());
    }

    #[test]
    fn notification_first_rejects_unmatched_thread() {
        let mut runtime = notification_first_runtime();
        assert_eq!(
            ingest(
                &mut runtime,
                turn_started("thr-unmatched", "turn-notification")
            ),
            Err(RuntimeError::CorrelationMismatch)
        );
        assert!(runtime.pending_turn_started.is_empty());
    }

    #[test]
    fn notification_first_mismatched_response_consumes_marker_and_fails_closed() {
        let mut runtime = notification_first_runtime();
        ingest(
            &mut runtime,
            turn_started("thr-notification", "turn-notification"),
        )
        .expect("first exactly correlated notification");
        assert_eq!(
            ingest(&mut runtime, turn_start_response(1, "turn-mismatch")),
            Err(RuntimeError::CorrelationMismatch)
        );
        assert!(runtime.pending_turn_started.is_empty());
        assert!(runtime.turns.is_empty());
    }

    #[test]
    fn notification_first_turn_start_capacity_is_bounded() {
        let mut runtime = pending_turn_runtime();
        let turn = runtime
            .turns
            .values()
            .next()
            .expect("turn state fixture")
            .clone();
        runtime.turns.clear();
        for index in 0..MAX_TURNS {
            runtime.turns.insert(
                (
                    format!("thr-capacity-{index}"),
                    format!("turn-capacity-{index}"),
                ),
                turn.clone(),
            );
        }
        assert_eq!(
            ingest(
                &mut runtime,
                turn_started("thr-notification", "turn-notification")
            ),
            Err(RuntimeError::PendingCapacityExceeded)
        );
        assert!(runtime.pending_turn_started.is_empty());
    }

    #[test]
    fn exact_disabled_remote_control_status_is_content_free() {
        let mut runtime = CodexAgentRuntime::new();
        let event = ingest(
            &mut runtime,
            json!({
                "method": "remoteControl/status/changed",
                "params": {
                    "status": "disabled",
                    "serverName": "private-server-canary",
                    "installationId": "private-installation-canary",
                    "environmentId": null
                },
                "emittedAtMs": 1
            }),
        )
        .expect("exact reviewed control notification");

        assert_eq!(event, RuntimeEvent::NonSemanticNotificationIgnored);
        let debug = format!("{event:?}");
        assert!(!debug.contains("private-server-canary"));
        assert!(!debug.contains("private-installation-canary"));
    }

    #[test]
    fn json_rpc_rejection_discards_message_and_data_content() {
        let mut runtime = pending_turn_runtime();
        let error = ingest(
            &mut runtime,
            json!({
                "jsonrpc": "2.0",
                "id": 1,
                "error": {
                    "code": -32603,
                    "message": "private-json-rpc-message-canary",
                    "data": {"content": "private-json-rpc-data-canary"}
                }
            }),
        )
        .expect_err("JSON-RPC error response must reject the request");

        assert_eq!(error, RuntimeError::JsonRpcRejected);
        let debug = format!("{error:?} {error} {runtime:?}");
        assert!(!debug.contains("private-json-rpc-message-canary"));
        assert!(!debug.contains("private-json-rpc-data-canary"));
    }

    #[test]
    fn malformed_json_rpc_error_is_a_protocol_failure() {
        let mut runtime = pending_turn_runtime();

        assert_eq!(
            ingest(
                &mut runtime,
                json!({"id": 1, "error": {"message": "private-malformed-canary"}}),
            ),
            Err(RuntimeError::ProtocolFailure)
        );
        assert!(!format!("{runtime:?}").contains("private-malformed-canary"));
    }

    #[test]
    fn completed_turn_exposes_only_authoritative_last_turn_usage() {
        let mut runtime = completable_runtime();
        assert_eq!(
            ingest(&mut runtime, token_usage_notification(12, 34)),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored)
        );
        let terminal = ingest(
            &mut runtime,
            json!({
                "method": "turn/completed",
                "params": {
                    "threadId": "thr-notification",
                    "turn": {
                        "id": "turn-notification",
                        "status": "completed",
                        "items": [],
                        "error": null
                    }
                }
            }),
        )
        .expect("completed turn with one usage update");
        let RuntimeEvent::TurnCompleted(receipt) = terminal else {
            panic!("expected terminal receipt");
        };
        assert_eq!(receipt.input_tokens, Some(12));
        assert_eq!(receipt.output_tokens, Some(34));
        let debug = format!("{receipt:?} {runtime:?}");
        assert!(!debug.contains("987654321"));
        assert!(!debug.contains("123456781"));
    }

    #[test]
    fn progressive_cumulative_usage_projects_the_latest_per_turn_snapshot_at_completion() {
        let mut runtime = completable_runtime();
        ingest(
            &mut runtime,
            token_usage_notification_with_total(12, 34, 1_000, 2_000),
        )
        .expect("initial complete cumulative projection");
        ingest(
            &mut runtime,
            token_usage_notification_with_total(7, 41, 1_010, 2_010),
        )
        .expect("later cumulative projection may replace a varying last snapshot");
        ingest(
            &mut runtime,
            token_usage_notification_with_total(7, 41, 1_010, 2_010),
        )
        .expect("an identical cumulative projection is idempotent");

        let terminal = ingest(
            &mut runtime,
            json!({
                "method": "turn/completed",
                "params": {
                    "threadId": "thr-notification",
                    "turn": {
                        "id": "turn-notification",
                        "status": "completed",
                        "items": [],
                        "error": null
                    }
                }
            }),
        )
        .expect("a turn with lifecycle and cumulative usage completes");
        assert!(matches!(
            terminal,
            RuntimeEvent::TurnCompleted(TurnCompletionReceipt {
                input_tokens: Some(7),
                output_tokens: Some(41),
                ..
            })
        ));
        let debug = format!("{runtime:?}");
        for forbidden in ["1000", "1010", "2000", "2010"] {
            assert!(
                !debug.contains(forbidden),
                "cumulative values must remain private"
            );
        }
    }

    #[test]
    fn nonmonotonic_cumulative_usage_refuses_without_mutating_the_latest_snapshot() {
        let mut runtime = completable_runtime();
        ingest(
            &mut runtime,
            token_usage_notification_with_total(12, 34, 1_000, 2_000),
        )
        .expect("initial complete cumulative projection");
        let before = runtime.turns.values().next().expect("turn fixture").clone();
        assert_eq!(
            ingest(
                &mut runtime,
                token_usage_notification_with_total(7, 41, 999, 2_010),
            ),
            Err(RuntimeError::ProtocolFailure),
            "a decrease in a validated cumulative projection remains fail closed"
        );
        let after = runtime.turns.values().next().expect("turn fixture");
        assert_eq!(after.usage, before.usage);
        assert_eq!(after.total_usage, before.total_usage);
    }

    #[test]
    fn completed_turn_requires_authoritative_usage_but_ignores_identical_duplicate() {
        let mut missing = completable_runtime();
        assert_eq!(
            ingest(
                &mut missing,
                json!({"method":"turn/completed","params":{"threadId":"thr-notification","turn":{"id":"turn-notification","status":"completed","items":[],"error":null}}}),
            ),
            Err(RuntimeError::ProtocolFailure)
        );

        let mut duplicate = completable_runtime();
        ingest(&mut duplicate, token_usage_notification(12, 34)).expect("first usage update");
        assert_eq!(
            ingest(&mut duplicate, token_usage_notification(12, 34)),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored),
            "the app-server may repeat an unchanged authoritative update"
        );
    }

    #[test]
    fn interrupted_or_failed_turn_usage_is_optional_but_never_invented() {
        let mut failed = completable_runtime();
        let failed = ingest(
            &mut failed,
            json!({"method":"turn/completed","params":{"threadId":"thr-notification","turn":{"id":"turn-notification","status":"failed","items":[],"error":{"codexErrorInfo":"other"}}}}),
        )
        .expect("failed turn without usage remains representable");
        assert!(matches!(
            failed,
            RuntimeEvent::TurnCompleted(TurnCompletionReceipt {
                input_tokens: None,
                output_tokens: None,
                ..
            })
        ));

        let mut interrupted = completable_runtime();
        interrupted
            .turns
            .get_mut(&(
                "thr-notification".to_owned(),
                "turn-notification".to_owned(),
            ))
            .expect("interrupt turn fixture")
            .interrupt_requested = true;
        ingest(&mut interrupted, token_usage_notification(7, 9))
            .expect("single interrupted usage update");
        let interrupted = ingest(
            &mut interrupted,
            json!({"method":"turn/completed","params":{"threadId":"thr-notification","turn":{"id":"turn-notification","status":"interrupted","items":[],"error":null}}}),
        )
        .expect("interrupted turn with usage remains representable");
        assert!(matches!(
            interrupted,
            RuntimeEvent::TurnCompleted(TurnCompletionReceipt {
                input_tokens: Some(7),
                output_tokens: Some(9),
                ..
            })
        ));
    }

    #[test]
    #[cfg(unix)]
    fn exact_ephemeral_empty_thread_started_is_content_free() {
        let mut runtime = active_reasoning_runtime();
        let event = ingest(&mut runtime, realistic_thread_started("thr-notification"))
            .expect("reviewed ephemeral thread notification");

        assert_eq!(event, RuntimeEvent::NonSemanticNotificationIgnored);
        let debug = format!("{event:?} {runtime:?}");
        for canary in [
            "private-preview-canary",
            "private-cwd-canary",
            "private-session-canary",
            "thr-notification",
        ] {
            assert!(!debug.contains(canary));
        }
    }

    #[test]
    #[cfg(unix)]
    fn thread_started_requires_registered_correlation() {
        let mut runtime = active_reasoning_runtime();
        assert_eq!(
            ingest(&mut runtime, realistic_thread_started("thr-unregistered")),
            Err(RuntimeError::CorrelationMismatch)
        );
    }

    #[test]
    fn malformed_extended_or_side_effecting_thread_started_fails_closed() {
        let mut malformed = Vec::new();

        let mut missing = realistic_thread_started("thr-notification");
        missing["params"]["thread"]
            .as_object_mut()
            .expect("thread fixture object")
            .remove("ephemeral");
        malformed.push(missing);

        let mut extended = realistic_thread_started("thr-notification");
        extended["params"]["thread"]["future"] = json!(true);
        malformed.push(extended);

        let mut persistent = realistic_thread_started("thr-notification");
        persistent["params"]["thread"]["ephemeral"] = json!(false);
        malformed.push(persistent);

        let mut materialized = realistic_thread_started("thr-notification");
        materialized["params"]["thread"]["path"] = json!("/private/thread.jsonl");
        malformed.push(materialized);

        let mut with_turn = realistic_thread_started("thr-notification");
        with_turn["params"]["thread"]["turns"] = json!([{"id":"turn-existing"}]);
        malformed.push(with_turn);

        let mut subagent = realistic_thread_started("thr-notification");
        subagent["params"]["thread"]["source"] = json!({"subAgent":"review"});
        malformed.push(subagent);

        let mut git = realistic_thread_started("thr-notification");
        git["params"]["thread"]["gitInfo"] = json!({"branch":"private-branch"});
        malformed.push(git);

        let mut extra = realistic_thread_started("thr-notification");
        extra["params"]["thread"]["extra"] = json!({"private":"payload"});
        malformed.push(extra);

        let mut section = realistic_thread_started("thr-notification");
        section["params"]["thread"]["section"] = json!({"id":"private-section"});
        malformed.push(section);

        let mut section_entered = realistic_thread_started("thr-notification");
        section_entered["params"]["thread"]["sectionEnteredAt"] = json!(1);
        malformed.push(section_entered);

        let mut direct_input = realistic_thread_started("thr-notification");
        direct_input["params"]["thread"]["canAcceptDirectInput"] = json!(false);
        malformed.push(direct_input);

        let mut history = realistic_thread_started("thr-notification");
        history["params"]["thread"]["historyMode"] = json!("future");
        malformed.push(history);

        let mut oversized = realistic_thread_started("thr-notification");
        oversized["params"]["thread"]["preview"] = json!("x".repeat(MAX_TEXT_BYTES + 1));
        malformed.push(oversized);

        let mut extended_params = realistic_thread_started("thr-notification");
        extended_params["params"]["future"] = json!(true);
        malformed.push(extended_params);

        for (index, fixture) in malformed.into_iter().enumerate() {
            let mut runtime = active_reasoning_runtime();
            assert_eq!(
                ingest(&mut runtime, fixture),
                Err(RuntimeError::ProtocolFailure),
                "malformed notification fixture {index}"
            );
        }
    }

    #[test]
    fn exact_no_tools_non_semantic_notifications_are_content_free() {
        let fixtures = [
            json!({"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"active","activeFlags":[]}}}),
            json!({"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"active","activeFlags":["waitingOnUserInput"]}}}),
            json!({"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"idle"}}}),
            json!({"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification","tokenUsage":{"last":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10},"total":{"cacheWriteInputTokens":0,"cachedInputTokens":11,"inputTokens":12,"outputTokens":13,"reasoningOutputTokens":14,"totalTokens":50},"modelContextWindow":128000}}}),
            json!({"method":"item/reasoning/summaryPartAdded","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","summaryIndex":0}}),
            json!({"method":"item/reasoning/summaryTextDelta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","summaryIndex":0,"delta":"private-summary-canary"}}),
            json!({"method":"item/reasoning/textDelta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","contentIndex":0,"delta":"private-reasoning-canary"}}),
        ];
        let mut runtime = active_reasoning_runtime();
        for fixture in fixtures {
            let event = ingest(&mut runtime, fixture).expect("reviewed non-semantic notification");
            assert_eq!(event, RuntimeEvent::NonSemanticNotificationIgnored);
            let debug = format!("{event:?}");
            assert!(!debug.contains("private-summary-canary"));
            assert!(!debug.contains("private-reasoning-canary"));
            assert!(!debug.contains("thr-notification"));
            assert!(!debug.contains("turn-notification"));
            assert!(!debug.contains("reasoning-notification"));
        }
    }

    #[test]
    fn thread_status_active_flags_match_the_installed_protocol_enum() {
        let mut runtime = active_reasoning_runtime();
        for invalid in [
            json!({"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"active","activeFlags":["unknown"]}}}),
            json!({"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"active","activeFlags":["waitingOnApproval","waitingOnUserInput","waitingOnApproval"]}}}),
            json!({"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"active","activeFlags":[1]}}}),
        ] {
            assert!(matches!(
                ingest(&mut runtime, invalid),
                Err(RuntimeError::ProtocolFailure)
            ));
        }
    }

    #[test]
    fn installed_mcp_startup_notifications_are_content_free() {
        let fixtures = [
            json!({
                "method": "mcpServer/startupStatus/updated",
                "params": {
                    "threadId": "thr-notification",
                    "name": "openai-api-key-local-confirmation",
                    "status": "starting",
                    "error": null,
                    "failureReason": null
                },
                "emittedAtMs": 1_786_054_396_424_u64
            }),
            json!({
                "method": "mcpServer/startupStatus/updated",
                "params": {
                    "threadId": "thr-notification",
                    "name": "context7",
                    "status": "ready",
                    "error": null,
                    "failureReason": null
                },
                "emittedAtMs": 1_786_054_397_661_u64
            }),
            json!({
                "method": "mcpServer/startupStatus/updated",
                "params": {
                    "name": "context7",
                    "status": "starting"
                },
                "emittedAtMs": 1_786_054_397_662_u64
            }),
            json!({
                "method": "mcpServer/startupStatus/updated",
                "params": {
                    "threadId": null,
                    "name": "context7",
                    "status": "cancelled",
                    "error": null,
                    "failureReason": null
                },
                "emittedAtMs": 1_786_054_397_663_u64
            }),
            json!({
                "method": "mcpServer/startupStatus/updated",
                "params": {
                    "threadId": "thr-notification",
                    "name": "context7",
                    "status": "failed",
                    "error": null,
                    "failureReason": "reauthenticationRequired"
                },
                "emittedAtMs": 1_786_054_397_664_u64
            }),
            json!({
                "method": "mcpServer/startupStatus/updated",
                "params": {
                    "threadId": "thr-notification",
                    "name": "context7",
                    "status": "failed",
                    "error": "MCP startup failed",
                    "failureReason": null
                },
                "emittedAtMs": 1_786_054_397_665_u64
            }),
        ];
        let mut runtime = active_reasoning_runtime();

        for fixture in fixtures {
            let event = ingest(&mut runtime, fixture).expect("live-shaped MCP startup status");
            assert_eq!(event, RuntimeEvent::NonSemanticNotificationIgnored);
            let debug = format!("{event:?} {runtime:?}");
            assert!(!debug.contains("openai-api-key-local-confirmation"));
            assert!(!debug.contains("context7"));
            assert!(!debug.contains("thr-notification"));
            assert!(!debug.contains("MCP startup failed"));
        }
    }

    #[test]
    fn malformed_extended_or_diagnostic_mcp_startup_notifications_fail_closed() {
        let malformed = [
            json!({"threadId":"thr-notification","name":"context7","status":"ready","error":"","failureReason":null}),
            json!({"threadId":"thr-notification","name":"context7","status":"ready","error":"startup\nfailed","failureReason":null}),
            json!({"threadId":"thr-notification","name":"context7","status":"ready","error":1,"failureReason":null}),
            json!({"threadId":"thr-notification","name":"context7","status":"failed","error":null,"failureReason":"future"}),
            json!({"threadId":"thr-notification","name":"","status":"ready","error":null,"failureReason":null}),
            json!({"threadId":"thr-notification","name":"context7"}),
            json!({"name":"context7","status":"future"}),
            json!({"threadId":1,"name":"context7","status":"ready"}),
            json!({"threadId":"thr-notification","name":"context7","status":"ready","error":null,"failureReason":null,"future":true}),
        ];

        for params in malformed {
            let mut runtime = active_reasoning_runtime();
            assert!(
                ingest(
                    &mut runtime,
                    json!({"method":"mcpServer/startupStatus/updated","params":params,"emittedAtMs":1}),
                )
                .is_err()
            );
        }

        let mut unregistered = active_reasoning_runtime();
        assert_eq!(
            ingest(
                &mut unregistered,
                json!({"method":"mcpServer/startupStatus/updated","params":{"threadId":"thr-unregistered","name":"context7","status":"ready","error":null,"failureReason":null},"emittedAtMs":1}),
            ),
            Err(RuntimeError::CorrelationMismatch)
        );

        let secret = "sk-private-mcp-startup-canary";
        let mut sensitive = active_reasoning_runtime();
        let error = ingest(
            &mut sensitive,
            json!({"method":"mcpServer/startupStatus/updated","params":{"threadId":"thr-notification","name":"context7","status":"ready","error":secret,"failureReason":null},"emittedAtMs":1}),
        )
        .expect_err("secret-shaped diagnostic must fail closed");
        assert_eq!(error, RuntimeError::SensitiveInputRejected);
        assert!(!format!("{error:?} {error} {sensitive:?}").contains(secret));
    }

    #[test]
    fn live_user_message_lifecycle_is_correlated_and_not_retained() {
        let item = json!({
            "type": "userMessage",
            "id": "msg-user-live",
            "clientId": "steadymux-live-canary",
            "content": [{
                "type": "text",
                "text": "Return the single token STEADYMUX_CANARY.",
                "text_elements": []
            }]
        });
        let mut runtime = active_reasoning_runtime();
        let item_count = runtime
            .turns
            .values()
            .next()
            .expect("active turn")
            .items
            .len();

        for (method, timestamp_key) in [
            ("item/started", "startedAtMs"),
            ("item/completed", "completedAtMs"),
        ] {
            let event = ingest(
                &mut runtime,
                json!({
                    "method": method,
                    "params": {
                        "item": item.clone(),
                        "threadId": "thr-notification",
                        "turnId": "turn-notification",
                        timestamp_key: 1_786_054_552_017_u64
                    },
                    "emittedAtMs": 1_786_054_552_018_u64
                }),
            )
            .expect("live-shaped user message lifecycle");
            assert_eq!(event, RuntimeEvent::NonSemanticNotificationIgnored);
        }
        assert_eq!(
            runtime
                .turns
                .values()
                .next()
                .expect("active turn")
                .items
                .len(),
            item_count
        );
        let debug = format!("{runtime:?}");
        assert!(!debug.contains("STEADYMUX_CANARY"));
        assert!(!debug.contains("steadymux-live-canary"));
        assert!(!debug.contains("msg-user-live"));
    }

    #[test]
    #[cfg(unix)]
    fn live_warning_and_rate_notifications_are_content_free_while_hooks_refuse() {
        let fixtures = [
            json!({"method":"warning","params":{"threadId":"thr-notification","message":"Skill descriptions were shortened to fit the bounded skills context budget."},"emittedAtMs":1}),
            json!({"method":"account/rateLimits/updated","params":{"rateLimits":{"limitId":"codex","limitName":null,"primary":{"usedPercent":95,"windowDurationMins":10080,"resetsAt":1_786_323_626_u64},"secondary":null,"credits":{"hasCredits":false,"unlimited":false,"balance":"0"},"individualLimit":null,"spendControlReached":null,"planType":"pro","rateLimitReachedType":null}},"emittedAtMs":1}),
        ];
        let mut runtime = active_reasoning_runtime();

        for fixture in fixtures {
            assert_eq!(
                ingest(&mut runtime, fixture),
                Ok(RuntimeEvent::NonSemanticNotificationIgnored)
            );
        }
        for method in ["hook/started", "hook/completed"] {
            assert_eq!(
                ingest(
                    &mut runtime,
                    json!({"method":method,"params":{"private":"hook payload"},"emittedAtMs":1}),
                ),
                Err(RuntimeError::ProtocolFailure)
            );
        }
        let debug = format!("{runtime:?}");
        assert!(!debug.contains("hook payload"));
        assert!(!debug.contains("usedPercent"));
    }

    #[test]
    #[cfg(unix)]
    fn exhausted_account_rate_notification_is_not_a_protocol_failure() {
        // An exhausted subscription emits rateLimitReachedType and
        // spendControlReached populated. This must be accepted as a
        // content-free non-semantic notification so the cascade can fall
        // through to the next provider instead of misclassifying it as an
        // ambiguous post-write protocol failure.
        let exhausted = json!({"method":"account/rateLimits/updated","params":{"rateLimits":{"limitId":"codex","limitName":null,"primary":{"usedPercent":100,"windowDurationMins":10080,"resetsAt":1_786_323_626_u64},"secondary":null,"credits":{"hasCredits":false,"unlimited":false,"balance":"0"},"individualLimit":null,"spendControlReached":true,"planType":"pro","rateLimitReachedType":"spend_control"}},"emittedAtMs":1});
        let mut runtime = active_reasoning_runtime();
        assert_eq!(
            ingest(&mut runtime, exhausted),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored)
        );
        let debug = format!("{runtime:?}");
        assert!(!debug.contains("spend_control"));
        assert!(!debug.contains("usedPercent"));
    }

    #[test]
    fn malformed_or_extended_no_tools_notifications_fail_closed() {
        let malformed = [
            json!({"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"active","activeFlags":["unknown"]}}}),
            json!({"method":"thread/status/changed","params":{"threadId":"thr-notification","status":{"type":"idle"},"future":true}}),
            json!({"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification","tokenUsage":{"last":{},"total":{}}}}),
            json!({"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification","tokenUsage":{"last":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10},"total":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10}},"future":true}}),
            json!({"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification","tokenUsage":{"last":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10},"total":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10},"future":true}}}),
            json!({"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification","tokenUsage":{"last":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10,"future":0},"total":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10}}}}),
            json!({"method":"item/reasoning/summaryPartAdded","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","summaryIndex":-1}}),
            json!({"method":"item/reasoning/summaryPartAdded","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","summaryIndex":0,"future":true}}),
            json!({"method":"item/reasoning/summaryTextDelta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","summaryIndex":0}}),
            json!({"method":"item/reasoning/summaryTextDelta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","summaryIndex":0,"delta":"value","future":true}}),
            json!({"method":"item/reasoning/textDelta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","contentIndex":-1,"delta":"value"}}),
            json!({"method":"item/reasoning/textDelta","params":{"threadId":"thr-notification","turnId":"turn-notification","itemId":"reasoning-notification","contentIndex":0,"delta":"value","future":true}}),
        ];
        for (index, fixture) in malformed.into_iter().enumerate() {
            let mut runtime = active_reasoning_runtime();
            assert_eq!(
                ingest(&mut runtime, fixture),
                Err(RuntimeError::ProtocolFailure),
                "malformed no-tools notification fixture {index}"
            );
        }
    }

    #[test]
    fn token_usage_breakdown_validation_accepts_exact_counts_and_refuses_shape_keys_and_values() {
        let exact = json!({
            "cachedInputTokens": 1,
            "inputTokens": 2,
            "outputTokens": 3,
            "reasoningOutputTokens": 4,
            "totalTokens": 10,
        });
        assert_eq!(
            valid_token_breakdown(Some(&exact)),
            TokenBreakdownValidation::Valid
        );
        assert_eq!(
            valid_token_breakdown(Some(&json!(null))),
            TokenBreakdownValidation::Shape
        );
        assert_eq!(
            valid_token_breakdown(Some(&json!({"inputTokens": 2, "future": 0}))),
            TokenBreakdownValidation::Keys
        );
        assert_eq!(
            valid_token_breakdown(Some(&json!({
                "cachedInputTokens": 1,
                "inputTokens": -2,
                "outputTokens": 3,
                "reasoningOutputTokens": 4,
                "totalTokens": 10,
            }))),
            TokenBreakdownValidation::Counts
        );
    }

    #[test]
    fn token_usage_refusals_preserve_turn_usage_state() {
        let invalid_notifications = [
            json!({"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification"}}),
            json!({"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification","tokenUsage":{"last":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10},"total":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10},"modelContextWindow":-1}}}),
            json!({"method":"thread/tokenUsage/updated","params":{"threadId":"thr-notification","turnId":"turn-notification","tokenUsage":{"last":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10,"future":0},"total":{"cachedInputTokens":1,"inputTokens":2,"outputTokens":3,"reasoningOutputTokens":4,"totalTokens":10}}}}),
        ];
        for (index, notification) in invalid_notifications.into_iter().enumerate() {
            let mut runtime = completable_runtime();
            assert_eq!(
                ingest(&mut runtime, notification),
                Err(RuntimeError::ProtocolFailure),
                "token usage refusal {index}"
            );
            assert!(
                runtime
                    .turns
                    .get(&(
                        String::from("thr-notification"),
                        String::from("turn-notification")
                    ))
                    .is_some_and(|turn| turn.usage.is_none())
            );
        }
    }

    #[test]
    fn token_usage_correlation_markers_commit_only_ids_and_preserve_refusal_state() {
        let runtime = completable_runtime();
        let incoming_thread_id = "thread-private-canary";
        let incoming_turn_id = "turn-private-canary";
        let marker = token_usage_correlation_marker(
            TokenUsageCorrelationRefusal::MissingPair,
            incoming_thread_id,
            incoming_turn_id,
            &runtime.turns,
        );

        assert!(marker.starts_with(concat!(
            "steadymux-owned-child: token-usage-correlation reason=missing-pair incoming-thread-sha256=",
            "39b23b6f58c848f74d5936de5a9bd933f4b53a58cef823f75c3eef10a27073ce"
        )));
        assert!(marker.contains(concat!(
            " incoming-turn-sha256=",
            "f048ecdb852620d8ac2aed4f4b73b6b087ad0bbf515943205eba3a2bc94a9a13"
        )));
        assert!(marker.contains(" known-active-threads-sha256="));
        assert!(marker.contains(" known-active-turns-sha256="));
        assert!(marker.contains(" known-active-pairs-sha256="));
        assert!(marker.ends_with(" known-active-pair-count=1"));
        for forbidden in [
            incoming_thread_id,
            incoming_turn_id,
            "thr-notification",
            "turn-notification",
            "private",
            "canary",
        ] {
            assert!(!marker.contains(forbidden), "marker leaked {forbidden}");
        }

        let mut runtime = completable_runtime();
        let mut mismatched = token_usage_notification(12, 34);
        mismatched["params"]["threadId"] = json!(incoming_thread_id);
        mismatched["params"]["turnId"] = json!(incoming_turn_id);
        assert_eq!(
            ingest(&mut runtime, mismatched),
            Err(RuntimeError::ProtocolFailure),
            "the diagnostic must not weaken correlation admission"
        );
        assert!(
            runtime
                .turns
                .get(&(
                    String::from("thr-notification"),
                    String::from("turn-notification")
                ))
                .is_some_and(|turn| turn.usage.is_none()),
            "a mismatched update must not consume the active turn usage slot"
        );

        let mut rerouted = completable_runtime();
        rerouted
            .turns
            .get_mut(&(
                String::from("thr-notification"),
                String::from("turn-notification"),
            ))
            .expect("active turn fixture")
            .model_rerouted = true;
        assert_eq!(
            ingest(&mut rerouted, token_usage_notification(12, 34)),
            Err(RuntimeError::ProtocolFailure),
            "rerouted turns remain ineligible for usage updates"
        );
        assert!(rerouted.turns.values().all(|turn| turn.usage.is_none()));

        let mut duplicate = completable_runtime();
        ingest(&mut duplicate, token_usage_notification(12, 34)).expect("first usage update");
        let usage_before = duplicate
            .turns
            .values()
            .next()
            .expect("active turn fixture")
            .usage;
        assert_eq!(
            ingest(&mut duplicate, token_usage_notification(13, 34)),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored),
            "a monotonic cumulative projection may replace the per-turn snapshot"
        );
        assert_eq!(
            duplicate
                .turns
                .values()
                .next()
                .expect("active turn fixture")
                .usage,
            Some(TurnUsageCounts {
                input_tokens: 13,
                output_tokens: 34,
            }),
            "the latest per-turn snapshot is retained after cumulative admission"
        );
        assert_ne!(
            duplicate
                .turns
                .values()
                .next()
                .expect("active turn fixture")
                .usage,
            usage_before,
        );

        for reason in [
            TokenUsageCorrelationRefusal::MissingPair,
            TokenUsageCorrelationRefusal::Rerouted,
            TokenUsageCorrelationRefusal::PriorUsage,
        ] {
            let marker = token_usage_correlation_marker(
                reason,
                incoming_thread_id,
                incoming_turn_id,
                &runtime.turns,
            );
            assert!(marker.contains(&format!(" reason={}", reason.as_str())));
        }
    }

    #[test]
    fn cumulative_usage_admission_is_monotonic_and_never_exposes_token_values() {
        let baseline = TurnUsageCounts {
            input_tokens: 12,
            output_tokens: 34,
        };
        assert_eq!(
            prior_usage_direction(
                baseline,
                TurnUsageCounts {
                    input_tokens: 13,
                    output_tokens: 34,
                },
            ),
            PriorUsageDirection::MonotonicNondecreasing
        );
        assert_eq!(
            prior_usage_direction(
                baseline,
                TurnUsageCounts {
                    input_tokens: 11,
                    output_tokens: 34,
                },
            ),
            PriorUsageDirection::DecreaseOrMixed
        );
        assert_eq!(
            prior_usage_direction(
                baseline,
                TurnUsageCounts {
                    input_tokens: 12,
                    output_tokens: 34,
                },
            ),
            PriorUsageDirection::Same
        );
        assert_eq!(
            prior_usage_direction(
                baseline,
                TurnUsageCounts {
                    input_tokens: 13,
                    output_tokens: 33,
                },
            ),
            PriorUsageDirection::DecreaseOrMixed
        );

        assert_eq!(
            total_usage_relative_to_last(baseline, cumulative_usage(12, 34),),
            TotalUsageRelativeToLast::Equal
        );
        assert_eq!(
            total_usage_relative_to_last(baseline, cumulative_usage(13, 35),),
            TotalUsageRelativeToLast::TotalNondecreasing
        );
        assert_eq!(
            total_usage_relative_to_last(baseline, cumulative_usage(11, 35),),
            TotalUsageRelativeToLast::TotalDecreaseOrMixed
        );
        assert_eq!(
            cumulative_usage_direction(cumulative_usage(12, 34), cumulative_usage(13, 35)),
            PriorUsageDirection::MonotonicNondecreasing
        );
        assert_eq!(
            cumulative_usage_direction(cumulative_usage(12, 34), cumulative_usage(11, 35)),
            PriorUsageDirection::DecreaseOrMixed
        );
        let mut cache_write_decrease = cumulative_usage(13, 35);
        cache_write_decrease.cache_write_input_tokens = Some(1);
        let mut cache_write_regression = cache_write_decrease;
        cache_write_regression.cache_write_input_tokens = Some(0);
        assert_eq!(
            cumulative_usage_direction(cache_write_decrease, cache_write_regression),
            PriorUsageDirection::DecreaseOrMixed,
            "all validated cumulative counters participate in the admission guard"
        );
        // An omitted optional cache-write count is not a regression against a
        // previously reported positive value.
        let mut cache_write_omitted = cache_write_decrease;
        cache_write_omitted.cache_write_input_tokens = None;
        assert_eq!(
            cumulative_usage_direction(cache_write_decrease, cache_write_omitted),
            PriorUsageDirection::MonotonicNondecreasing,
            "omitted optional cache-write count must not reset a positive prior"
        );

        let marker = prior_usage_direction_marker(PriorUsageDirection::DecreaseOrMixed);
        assert_eq!(
            marker,
            "steadymux-owned-child: token-usage-prior-direction=decrease-or-mixed"
        );
        for forbidden in ["12", "13", "33", "34", "input", "output"] {
            assert!(!marker.contains(forbidden), "marker leaked {forbidden}");
        }
        let projection_marker = prior_usage_projection_marker(
            TotalUsageRelativeToLast::TotalNondecreasing,
            PriorUsageDirection::MonotonicNondecreasing,
        );
        assert_eq!(
            projection_marker,
            "steadymux-owned-child: token-usage-prior-projections total-vs-last=total-nondecreasing total-vs-prior=monotonic-nondecreasing"
        );
        for forbidden in ["12", "13", "33", "34", "input", "output", "cached"] {
            assert!(
                !projection_marker.contains(forbidden),
                "projection marker leaked {forbidden}"
            );
        }

        let mut runtime = completable_runtime();
        ingest(
            &mut runtime,
            token_usage_notification_with_total(12, 34, 1_000, 2_000),
        )
        .expect("baseline usage");
        let before = runtime.turns.values().next().expect("turn state").usage;
        assert_eq!(
            ingest(
                &mut runtime,
                token_usage_notification_with_total(13, 34, 1_010, 2_010),
            ),
            Ok(RuntimeEvent::NonSemanticNotificationIgnored),
            "a complete monotonic total admits the latest per-turn projection"
        );
        assert_eq!(
            runtime.turns.values().next().expect("turn state").usage,
            Some(TurnUsageCounts {
                input_tokens: 13,
                output_tokens: 34,
            })
        );
        assert_ne!(
            runtime.turns.values().next().expect("turn state").usage,
            before
        );
        assert_eq!(
            ingest(
                &mut runtime,
                token_usage_notification_with_total(11, 34, 1_009, 2_011),
            ),
            Err(RuntimeError::ProtocolFailure),
            "a decrease in any persisted cumulative counter remains fail closed"
        );
        assert_eq!(
            runtime.turns.values().next().expect("turn state").usage,
            Some(TurnUsageCounts {
                input_tokens: 13,
                output_tokens: 34,
            }),
            "a refused cumulative projection preserves the last accepted receipt delta"
        );
        let debug = format!("{runtime:?}");
        for forbidden in ["1000", "1010", "2000", "2010"] {
            assert!(
                !debug.contains(forbidden),
                "runtime debug leaked {forbidden}"
            );
        }

        let mut rerouted = completable_runtime();
        ingest(&mut rerouted, token_usage_notification(12, 34)).expect("baseline usage");
        rerouted
            .turns
            .values_mut()
            .next()
            .expect("turn state")
            .model_rerouted = true;
        let before = rerouted.turns.values().next().expect("turn state").usage;
        assert_eq!(
            ingest(&mut rerouted, token_usage_notification(12, 34)),
            Err(RuntimeError::ProtocolFailure),
            "rerouted turns refuse even an otherwise-idempotent usage update"
        );
        assert_eq!(
            rerouted.turns.values().next().expect("turn state").usage,
            before,
            "rerouted refusal preserves previously admitted usage"
        );
    }

    #[test]
    fn malformed_remote_control_status_notifications_fail_closed() {
        let malformed = [
            json!({"status":"enabled","serverName":"server","installationId":"install","environmentId":null}),
            json!({"status":"disabled","serverName":"","installationId":"install","environmentId":null}),
            json!({"status":"disabled","serverName":"server","installationId":"","environmentId":null}),
            json!({"status":"disabled","serverName":"server","installationId":"install","environmentId":"environment"}),
            json!({"status":"disabled","serverName":"server","installationId":"install","environmentId":null,"future":true}),
        ];

        for params in malformed {
            let mut runtime = CodexAgentRuntime::new();
            assert_eq!(
                ingest(
                    &mut runtime,
                    json!({"method":"remoteControl/status/changed","params":params,"emittedAtMs":1}),
                ),
                Err(RuntimeError::ProtocolFailure)
            );
        }
    }

    #[test]
    fn unknown_notifications_still_fail_closed() {
        let mut runtime = CodexAgentRuntime::new();
        let canary = "private-payload-canary";
        let error = ingest(
            &mut runtime,
            json!({"method":"thread/future","params":{"content":canary},"emittedAtMs":1}),
        )
        .expect_err("unknown notification must remain fail closed");
        let RuntimeError::UnsupportedNotificationMethod(method) = error else {
            panic!("unknown notification must return a method-only diagnostic");
        };
        assert_eq!(method.as_str(), "thread/future");
        assert!(!format!("{method:?} {method}").contains(canary));
    }

    #[test]
    fn side_effecting_item_refusal_markers_are_content_free_and_policy_specific() {
        let item_type = "futureToolWithPrivatePayload";
        let type_marker = side_effecting_item_type_marker(item_type);
        assert_eq!(
            type_marker,
            concat!(
                "steadymux-owned-child: side-effecting-item-type-sha256=",
                "b13049c568e4b6f1440b283578562022dc3e6a51c357527f133126acd750ee75"
            )
        );
        let type_marker = side_effecting_item_type_marker(item_type);
        assert!(!type_marker.contains(item_type));
        assert!(!type_marker.contains("PrivatePayload"));
        let mcp_item = json!({
            "status": "inProgress",
            "server": "mcp-server-private-canary",
            "tool": "mcp-tool-private-canary",
            "arguments": {"content": "NO_RETENTION_MCP_ARGUMENT_CANARY"},
        });
        assert_eq!(
            mcp_tool_policy_refusal_marker(&mcp_item, McpToolLifecycle::Started),
            concat!(
                "steadymux-owned-child: side-effecting-item-refused=mcp-tool-policy ",
                "lifecycle=started status=in-progress ",
                "server-sha256=424957af3b15be998a0143b68e0adc27ec7d06e8521b4045f03525fcbae5765d ",
                "tool-sha256=0cec75c422c7b03194148b7e50165b45a60a0868d5960d449b9b2e5376d923a6"
            )
        );
        let marker = mcp_tool_policy_refusal_marker(&mcp_item, McpToolLifecycle::Started);
        assert!(!marker.contains("mcp-server-private-canary"));
        assert!(!marker.contains("mcp-tool-private-canary"));
        assert!(!marker.contains("NO_RETENTION_MCP_ARGUMENT_CANARY"));
        assert_eq!(
            mcp_tool_policy_refusal_marker(&json!({}), McpToolLifecycle::Completed),
            concat!(
                "steadymux-owned-child: side-effecting-item-refused=mcp-tool-policy ",
                "lifecycle=completed status=unknown ",
                "server-sha256=missing-or-invalid tool-sha256=missing-or-invalid"
            )
        );
    }

    #[test]
    fn rejected_item_types_remain_unretained_and_fail_closed() {
        let canary = "NO_RETENTION_ITEM_CANARY";
        for (method, item_type) in [
            ("item/started", "futureToolWithPrivatePayload"),
            ("item/completed", "futureCompletedToolWithPrivatePayload"),
        ] {
            let mut runtime = active_reasoning_runtime();
            assert_eq!(
                ingest(
                    &mut runtime,
                    json!({
                        "method": method,
                        "params": {
                            "threadId": "thr-notification",
                            "turnId": "turn-notification",
                            "item": {
                                "type": item_type,
                                "id": "side-effecting-item",
                                "private": canary,
                            },
                        },
                    }),
                ),
                Err(RuntimeError::SideEffectingItemRejected),
                "{method} must remain fail closed"
            );
            let turn = runtime
                .turns
                .values()
                .next()
                .expect("active turn remains available for terminal receipt");
            assert!(turn.tool_or_side_effect);
            assert_eq!(turn.items.len(), 1);
            let debug = format!("{runtime:?}");
            assert!(!debug.contains(item_type));
            assert!(!debug.contains(canary));
        }
    }

    #[test]
    fn web_search_items_remain_fail_closed_even_if_a_child_flag_is_omitted() {
        let canary = "NO_RETENTION_WEB_SEARCH_CANARY";
        for (method, item_type) in [
            ("item/started", "webSearch"),
            ("item/completed", "webSearch"),
        ] {
            let mut runtime = active_reasoning_runtime();
            assert_eq!(
                ingest(
                    &mut runtime,
                    json!({
                        "method": method,
                        "params": {
                            "threadId": "thr-notification",
                            "turnId": "turn-notification",
                            "item": {
                                "type": item_type,
                                "id": "web-search-item",
                                "private": canary,
                            },
                        },
                    }),
                ),
                Err(RuntimeError::SideEffectingItemRejected),
                "{method} web-search item must remain fail closed"
            );
            let turn = runtime
                .turns
                .values()
                .next()
                .expect("active turn remains available for terminal receipt");
            assert!(turn.tool_or_side_effect);
            assert!(!format!("{runtime:?}").contains(canary));
        }
    }

    #[test]
    fn no_tools_mcp_refusal_remains_fail_closed_with_digest_only_identity_markers() {
        let canary = "NO_RETENTION_MCP_CANARY";
        for (method, status, lifecycle) in [
            ("item/started", "inProgress", McpToolLifecycle::Started),
            ("item/completed", "completed", McpToolLifecycle::Completed),
        ] {
            let mut runtime = active_reasoning_runtime();
            let item = json!({
                "type": "mcpToolCall",
                "id": "mcp-tool-item",
                "status": status,
                "server": "mcp-server-private-canary",
                "tool": "mcp-tool-private-canary",
                "arguments": {"canary": canary},
                "result": {"content": "NO_RETENTION_MCP_RESULT_CANARY"},
                "error": null,
            });
            assert_eq!(
                ingest(
                    &mut runtime,
                    json!({
                        "method": method,
                        "params": {
                            "threadId": "thr-notification",
                            "turnId": "turn-notification",
                            "item": item,
                        },
                    }),
                ),
                Err(RuntimeError::SideEffectingItemRejected),
                "{method} must reject all MCP lifecycle events"
            );
            let turn = runtime
                .turns
                .values()
                .next()
                .expect("active turn remains available");
            assert!(turn.tool_or_side_effect);
            assert!(!format!("{runtime:?}").contains(canary));
            let marker = mcp_tool_policy_refusal_marker(
                &json!({
                    "status": status,
                    "server": "mcp-server-private-canary",
                    "tool": "mcp-tool-private-canary",
                    "arguments": {"canary": canary},
                }),
                lifecycle,
            );
            assert!(!marker.contains(canary));
            assert!(!marker.contains("mcp-server-private-canary"));
            assert!(!marker.contains("mcp-tool-private-canary"));
        }
    }
}