bamboo-engine 2026.10.22

//! Authoritative permission refresh for approved-tool replay.
//!
//! An approval only authorizes the posture that is still active when the
//! suspended tool is about to re-enter the executor. All replay surfaces use
//! this helper so a stale in-memory Auto/Bypass snapshot cannot outrun a newer
//! durable Default/Plan transition.

use bamboo_agent_core::storage::Storage;
use bamboo_agent_core::tools::{
    plan_mode_allows_tool, ExecutingSupervisorObservation, PendingToolOperation,
    RetainedToolExecution, ToolCall, ToolExecutionSessionFlags, ToolExecutor, ToolResult,
    PENDING_TOOL_OPERATION_METADATA_KEY,
};
use bamboo_agent_core::{AgentError, PendingQuestionSource, Session};
use bamboo_domain::{permission_request_generation, AgentRuntimeState, PermissionMode};
use bamboo_tools::permission::{
    PermissionConfig, PermissionDecision, PermissionDecisionKind, PermissionDecisionReceipt,
    PermissionRequest,
};
use serde::{Deserialize, Serialize};

const PERMISSION_REPLAY_APPROVALS_METADATA_KEY: &str = "permission.replay_approvals.v1";

/// Original host dispatch, retained through result application without reading
/// a possibly replaced Session or trusting the tool's display payload.
pub(crate) struct PermissionReplayOrigin {
    caller: Option<bamboo_domain::SessionCallerReference>,
    observation: Option<ExecutingSupervisorObservation>,
    session_id: String,
    tool_call: ToolCall,
    execution_name: String,
    execution_contract: Option<String>,
    contract_known: bool,
}

impl PermissionReplayOrigin {
    pub(crate) fn new(
        caller: Option<bamboo_domain::SessionCallerReference>,
        observation: Option<ExecutingSupervisorObservation>,
        session_id: &str,
        tool_call: &ToolCall,
        execution_name: &str,
        execution_contract: Option<&str>,
        contract_known: bool,
    ) -> Self {
        Self {
            caller,
            observation,
            session_id: session_id.into(),
            tool_call: tool_call.clone(),
            execution_name: execution_name.into(),
            execution_contract: execution_contract.map(str::to_owned),
            contract_known,
        }
    }

    pub(crate) fn has_supervisor_observation(&self) -> bool {
        self.observation.is_some()
    }

    pub(crate) fn bind_waiting_message(&self, message: &mut bamboo_agent_core::Message) {
        let Ok(payload) = serde_json::from_str::<serde_json::Value>(&message.content) else {
            return;
        };
        if payload.get("status").and_then(serde_json::Value::as_str)
            != Some("awaiting_permission_approval")
        {
            return;
        }
        let Some(request) = payload
            .get("permission_request")
            .cloned()
            .and_then(|value| serde_json::from_value::<PermissionRequest>(value).ok())
        else {
            return;
        };
        if request.session_id != self.session_id
            || request.request_id != self.tool_call.id
            || request.tool_name != self.execution_name
            || request.request_generation.trim().is_empty()
            || message.tool_call_id.as_deref() != Some(self.tool_call.id.as_str())
        {
            return;
        }
        let mut metadata = serde_json::json!({"permission_request": request});
        if let Some(caller) = self.caller.as_ref() {
            metadata[PENDING_TOOL_OPERATION_METADATA_KEY] =
                PendingToolOperation::capture_with_contract(
                    caller.clone(),
                    &message.id,
                    &self.tool_call,
                    &self.execution_name,
                    &request.request_generation,
                    self.execution_contract.as_deref(),
                    self.contract_known,
                )
                .metadata_value();
        }
        if let Some(observation) = self.observation {
            metadata[ExecutingSupervisorObservation::PERMISSION_REPLAY_METADATA_KEY] = observation
                .permission_replay_record(
                    &self.session_id,
                    &message.id,
                    &self.tool_call,
                    &self.execution_name,
                    &request.request_generation,
                );
        }
        message.metadata = Some(metadata);
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
struct PermissionReplayApproval {
    request: PermissionRequest,
    decision: PermissionDecision,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ReparkedPermissionApproval {
    pub question: String,
    pub options: Vec<String>,
    pub allow_custom: bool,
}

/// Result of refreshing permission posture immediately before approval replay.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ApprovalReplayDecision {
    /// The approved call may enter the executor with these exact live flags.
    Execute(ToolExecutionSessionFlags),
    /// The approval became stale because the latest posture is Plan and the
    /// original tool mutates state. Callers consume the marker and record a
    /// failed result, but must not emit `ToolStart` or enter the executor.
    BlockedByPlan(ToolExecutionSessionFlags),
    /// The durable Root tool catalog no longer admits the resolved execution
    /// owner. Callers consume the marker as a failed result without restoring
    /// a remembered grant, emitting ToolStart, or entering the executor.
    BlockedByRootToolAuthority,
    /// The original execution owner is no longer registered. A legacy replay
    /// consumes its stale marker as a failed result without entering a tool.
    BlockedByUnavailableTool,
}

/// Exact history target of one approved tool replay.
///
/// The result-message position is captured together with the closest preceding
/// tool call so a provider-reused tool-call id cannot redirect replay to an
/// older operation or overwrite an older result.
pub struct PermissionReplayTarget {
    tool_call: ToolCall,
    result_message_index: usize,
    result_message_id: String,
    request_generation: Option<String>,
}

impl PermissionReplayTarget {
    pub fn tool_call(&self) -> &ToolCall {
        &self.tool_call
    }

    pub fn request_generation(&self) -> Option<&str> {
        self.request_generation.as_deref()
    }
}

fn retained_operation(
    session: &Session,
    target: &PermissionReplayTarget,
) -> Result<Option<PendingToolOperation>, AgentError> {
    PendingToolOperation::restore(
        session,
        target.result_message_index,
        &target.tool_call,
        target.request_generation(),
    )
    .map_err(|error| AgentError::Tool(error.into()))
}

/// Restore the original generic caller only from the validated host record.
/// Legacy readable history has no generic observation; do not mint one from
/// the Session that happened to be reloaded for this replay.
pub fn restore_permission_replay_caller(
    session: &Session,
    target: &PermissionReplayTarget,
) -> Result<Option<bamboo_domain::SessionCallerReference>, AgentError> {
    retained_operation(session, target)
        .map(|operation| operation.map(|operation| operation.caller().clone()))
}

/// Restore semantic selection only from retained Host evidence. Missing v1
/// provenance cannot identify a canonical v2 spawn owner after a name collision.
pub fn restore_permission_replay_execution(
    session: &Session,
    target: &PermissionReplayTarget,
    execution_name: &str,
) -> Result<Option<RetainedToolExecution>, AgentError> {
    let operation = retained_operation(session, target)?;
    let (known, contract) = if let Some(operation) = operation.as_ref() {
        if operation.execution_name() != execution_name {
            return Err(AgentError::Tool("retained execution owner changed".into()));
        }
        if operation.has_execution_contract_record() {
            (
                operation.execution_contract_known(),
                operation.execution_contract(),
            )
        } else {
            if execution_name == "spawn_session" && session.independent_root_spawn_contract() {
                return Err(unknown_spawn_contract());
            }
            (execution_name == "spawn_session", None)
        }
    } else {
        let legacy_binding = session.messages[target.result_message_index]
            .metadata
            .as_ref()
            .is_some_and(|metadata| {
                metadata
                    .get(ExecutingSupervisorObservation::PERMISSION_REPLAY_METADATA_KEY)
                    .is_some()
            });
        if !legacy_binding {
            return Ok(None);
        }
        ExecutingSupervisorObservation::restore_permission_replay_record(
            session,
            target.result_message_index,
            &target.tool_call,
            execution_name,
            target.request_generation(),
        )
        .map_err(|error| AgentError::Tool(error.into()))?;
        if execution_name == "spawn_session" && session.independent_root_spawn_contract() {
            return Err(unknown_spawn_contract());
        }
        (execution_name == "spawn_session", None)
    };
    if execution_name == "spawn_session" && !known {
        return Err(unknown_spawn_contract());
    }
    RetainedToolExecution::new(&target.tool_call, execution_name, known, contract)
        .map(Some)
        .map_err(|error| AgentError::Tool(error.to_string()))
}

fn unknown_spawn_contract() -> AgentError {
    AgentError::Tool("retained spawn operation lacks concrete execution-contract provenance; history and pending markers are preserved. Resend explicitly; no Root or Child was started".into())
}

fn retained_request(
    session: &Session,
    target: &PermissionReplayTarget,
) -> Result<Option<PermissionRequest>, AgentError> {
    let message = session
        .messages
        .get(target.result_message_index)
        .filter(|message| message.id == target.result_message_id)
        .ok_or_else(|| AgentError::Tool("permission replay result occurrence changed".into()))?;
    let payload = serde_json::from_str::<serde_json::Value>(&message.content).ok();
    let decode = |value: &serde_json::Value| {
        serde_json::from_value::<PermissionRequest>(value.clone())
            .map_err(|_| AgentError::Tool("permission replay request is malformed".into()))
    };
    let durable = message
        .metadata
        .as_ref()
        .and_then(|metadata| metadata.get("permission_request"))
        .map(decode)
        .transpose()?;
    let displayed = payload
        .as_ref()
        .and_then(|payload| payload.get("permission_request"))
        .map(decode)
        .transpose()?;
    if matches!((&durable, &displayed), (Some(a), Some(b)) if a != b) {
        return Err(AgentError::Tool(
            "permission replay request payload and metadata disagree".into(),
        ));
    }
    let request = durable.or(displayed);
    if request.as_ref().is_some_and(|request| {
        message.role != bamboo_domain::Role::Tool
            || request.session_id != session.id
            || request.request_id != target.tool_call.id
            || request.request_generation.trim().is_empty()
            || Some(request.request_generation.as_str()) != target.request_generation()
    }) {
        return Err(AgentError::Tool(
            "permission replay request identity changed".into(),
        ));
    }
    Ok(request)
}

/// Retained replay selects its original exact owner, never the fresh-call alias.
/// Old non-spawn calls keep their established resolver only while it agrees
/// with the retained typed owner. Ambiguous historical spawn calls need a new
/// explicit instruction, with their history and pending markers left intact.
pub fn resolve_permission_replay_owner(
    session: &Session,
    target: &PermissionReplayTarget,
    executor: &dyn ToolExecutor,
) -> Result<Option<String>, AgentError> {
    session
        .check_execution_compatibility()
        .map_err(|error| AgentError::Tool(error.to_string()))?;
    let operation = retained_operation(session, target)?;
    let request = retained_request(session, target)?;
    if let Some(operation) = operation {
        let request = request
            .ok_or_else(|| AgentError::Tool("pending operation lacks its typed request".into()))?;
        if operation.execution_name() != request.tool_name {
            return Err(AgentError::Tool(
                "permission replay owner differs from its frozen operation".into(),
            ));
        }
        let execution =
            restore_permission_replay_execution(session, target, operation.execution_name())?
                .expect("retained operation provides execution selection");
        return Ok((execution.owner_is_available(executor)
            && replay_framework_instance_available(
                session,
                operation.execution_name(),
                Some(&execution),
                executor,
            ))
        .then(|| operation.execution_name().to_owned()));
    }
    let legacy_supervisor_binding = session.messages[target.result_message_index]
        .metadata
        .as_ref()
        .is_some_and(|metadata| {
            metadata
                .get(ExecutingSupervisorObservation::PERMISSION_REPLAY_METADATA_KEY)
                .is_some()
        });
    if legacy_supervisor_binding {
        let request = request.ok_or_else(|| {
            AgentError::Tool("Supervisor pending operation lacks its typed request".into())
        })?;
        ExecutingSupervisorObservation::restore_permission_replay_record(
            session,
            target.result_message_index,
            &target.tool_call,
            &request.tool_name,
            target.request_generation(),
        )
        .map_err(|error| AgentError::Tool(error.into()))?;
        let execution = restore_permission_replay_execution(session, target, &request.tool_name)?
            .expect("validated legacy binding provides execution selection");
        return Ok((execution.owner_is_available(executor)
            && replay_framework_instance_available(
                session,
                &request.tool_name,
                Some(&execution),
                executor,
            ))
        .then_some(request.tool_name));
    }
    if bamboo_domain::independent_root_spawn_reference(&target.tool_call.function.name) {
        return Err(AgentError::Tool("retained spawn operation has no frozen owner/arguments; history and pending state are preserved. Resend the intended operation explicitly; no Root or Child was started".into()));
    }
    let owner = executor.resolve_reference_for_session(session, &target.tool_call.function.name);
    if let (Some(request), Some(owner)) = (&request, &owner) {
        if request.tool_name != *owner {
            return Err(AgentError::Tool(
                "permission replay owner differs from the retained request".into(),
            ));
        }
    }
    Ok(owner.filter(|name| replay_framework_instance_available(session, name, None, executor)))
}

fn replay_framework_instance_available(
    session: &Session,
    name: &str,
    retained: Option<&RetainedToolExecution>,
    executor: &dyn ToolExecutor,
) -> bool {
    let kind = bamboo_agent_core::tools::FrameworkSessionToolKind::from_execution_name(name);
    let root_semantics = retained.and_then(RetainedToolExecution::execution_contract)
        == Some(bamboo_agent_core::tools::FRAMEWORK_ROOT_SPAWN_CONTRACT_V1);
    // A known retained custom/legacy owner remains that exact operation. A
    // normal Child's own frozen ceiling still checks whether it is callable;
    // do not reinterpret it as the newly installed framework Root instance.
    let retained_ordinary_legacy_spawn = name == "spawn_session"
        && !session.model_orchestration_only_enabled()
        && retained.is_some_and(|retained| retained.contract_known() && !root_semantics);
    if kind.is_none()
        || retained_ordinary_legacy_spawn
        || !(root_semantics
            || session.model_orchestration_only_enabled()
            || session.kind == bamboo_domain::SessionKind::Child)
    {
        return true;
    }
    match retained {
        Some(retained) => retained.framework_instance_is_available(executor),
        None => executor
            .framework_session_tool_for_exact(name)
            .is_some_and(|proof| Some(proof.kind()) == kind && proof.is_current()),
    }
}

/// Locate the concrete approved invocation, newest-first and optionally bound
/// to the server-issued permission generation.
pub fn find_permission_replay_target(
    session: &Session,
    tool_call_id: &str,
    request_generation: Option<&str>,
) -> Option<PermissionReplayTarget> {
    let (result_message_index, result_message) =
        session
            .messages
            .iter()
            .enumerate()
            .rev()
            .find(|(_, message)| {
                message.tool_call_id.as_deref() == Some(tool_call_id)
                    && match request_generation {
                        Some(generation) => {
                            permission_request_generation(message).as_deref() == Some(generation)
                        }
                        None => true,
                    }
            })?;
    let tool_call = session.messages[..result_message_index]
        .iter()
        .rev()
        .find_map(|message| {
            message.tool_calls.as_ref().and_then(|calls| {
                calls
                    .iter()
                    .rev()
                    .find(|call| call.id == tool_call_id)
                    .cloned()
            })
        })?;
    let result_generation = permission_request_generation(result_message);
    Some(PermissionReplayTarget {
        tool_call,
        result_message_index,
        result_message_id: result_message.id.clone(),
        request_generation: result_generation,
    })
}

/// Replace only the tool-result message captured by
/// [`find_permission_replay_target`].
pub fn apply_permission_replay_result(
    session: &mut Session,
    target: &PermissionReplayTarget,
    content: String,
    success: bool,
) -> bool {
    let Some(message) = session.messages.get_mut(target.result_message_index) else {
        return false;
    };
    if message.id != target.result_message_id
        || message.tool_call_id.as_deref() != Some(target.tool_call.id.as_str())
        || target
            .request_generation
            .as_deref()
            .is_some_and(|generation| {
                permission_request_generation(message).as_deref() != Some(generation)
            })
    {
        return false;
    }
    message.content = content;
    message.tool_success = Some(success);
    true
}

fn message_permission_contract(
    session: &Session,
    target: &PermissionReplayTarget,
) -> Result<(PermissionRequest, PermissionDecisionReceipt), AgentError> {
    let message = session
        .messages
        .get(target.result_message_index)
        .filter(|message| message.id == target.result_message_id)
        .ok_or_else(|| {
            AgentError::Tool("permission replay result occurrence changed".to_string())
        })?;
    let metadata = message.metadata.as_ref().ok_or_else(|| {
        AgentError::Tool("permission replay is missing durable typed metadata".to_string())
    })?;
    let request = metadata
        .get("permission_request")
        .cloned()
        .and_then(|value| serde_json::from_value::<PermissionRequest>(value).ok())
        .ok_or_else(|| {
            AgentError::Tool("permission replay is missing its durable request".to_string())
        })?;
    let receipt = metadata
        .get("permission_decision_receipt")
        .cloned()
        .and_then(|value| serde_json::from_value::<PermissionDecisionReceipt>(value).ok())
        .ok_or_else(|| {
            AgentError::Tool(
                "permission replay is missing its durable decision receipt".to_string(),
            )
        })?;
    Ok((request, receipt))
}

fn validate_replay_approval(
    session: &Session,
    tool_call_id: &str,
    approval: &PermissionReplayApproval,
) -> Result<(), AgentError> {
    let request = &approval.request;
    let decision = &approval.decision;
    // expected_policy_revision was checked as an ingress CAS. A refresh can
    // retain this request generation while the durable request has its old revision.
    if request.session_id != session.id
        || request.request_id != tool_call_id
        || request.request_generation.trim().is_empty()
        || decision.request_id != request.request_id
        || decision.request_generation != request.request_generation
        || !request.allowed_decisions.contains(&decision.decision)
    {
        return Err(AgentError::Tool(
            "permission replay approval identity is invalid".to_string(),
        ));
    }
    if decision.decision == PermissionDecisionKind::AllowGlobal && !decision.confirm_global {
        return Err(AgentError::Tool(
            "permission replay global approval lacks explicit confirmation".to_string(),
        ));
    }
    Ok(())
}

fn selected_replay_matcher(
    approval: &PermissionReplayApproval,
) -> Result<bamboo_tools::permission::PermissionMatcher, AgentError> {
    let matcher_id = approval.decision.matcher_id.as_deref().ok_or_else(|| {
        AgentError::Tool("remembered permission replay lacks matcher identity".to_string())
    })?;
    approval
        .request
        .suggested_matchers
        .iter()
        .find(|matcher| matcher.id == matcher_id)
        .cloned()
        .ok_or_else(|| {
            AgentError::Tool("remembered permission replay matcher is unavailable".to_string())
        })
}

/// Pure admission, including present-invalid authority on legacy and Plan
/// paths. Run this before changing replay markers or installing any grants.
pub fn validate_permission_replay_authority(
    session: &Session,
    target: &PermissionReplayTarget,
    execution_name: &str,
) -> Result<Option<ExecutingSupervisorObservation>, AgentError> {
    session
        .check_execution_compatibility()
        .map_err(|error| AgentError::Tool(error.to_string()))?;
    if let Some(operation) = retained_operation(session, target)? {
        if operation.execution_name() != execution_name {
            return Err(AgentError::Tool(
                "permission replay owner differs from its frozen operation".into(),
            ));
        }
    }
    let observation = ExecutingSupervisorObservation::restore_permission_replay_record(
        session,
        target.result_message_index,
        &target.tool_call,
        execution_name,
        target.request_generation(),
    )
    .map_err(|error| AgentError::Tool(error.into()))?;
    if target.request_generation().is_some() {
        let approvals = validated_replay_approvals(session, target)?;
        if approvals.is_empty()
            || approvals
                .iter()
                .any(|approval| approval.request.tool_name != execution_name)
        {
            return Err(AgentError::Tool(
                "permission replay approval owner does not match dispatch".into(),
            ));
        }
    }
    Ok(observation)
}

/// An unanswered native approval can be displayed again without a receipt,
/// but its host identity must already match the waiting operation. This pure
/// check never returns an executable observation or installs a permission.
pub fn validate_pending_permission_replay_authority(
    session: &Session,
    target: &PermissionReplayTarget,
    execution_name: &str,
) -> Result<(), AgentError> {
    if let Some(operation) = retained_operation(session, target)? {
        if operation.execution_name() != execution_name {
            return Err(AgentError::Tool(
                "pending approval owner differs from its frozen operation".into(),
            ));
        }
    }
    let observation = ExecutingSupervisorObservation::restore_permission_replay_record(
        session,
        target.result_message_index,
        &target.tool_call,
        execution_name,
        target.request_generation(),
    )
    .map_err(|error| AgentError::Tool(error.into()))?;
    if observation.is_some() {
        let metadata = session.messages[target.result_message_index]
            .metadata
            .as_ref()
            .expect("validated host metadata");
        let request = metadata
            .get("permission_request")
            .cloned()
            .and_then(|value| serde_json::from_value::<PermissionRequest>(value).ok())
            .ok_or_else(|| {
                AgentError::Tool(
                    "Supervisor waiting approval is missing its durable request".into(),
                )
            })?;
        if request.session_id != session.id
            || request.request_id != target.tool_call.id
            || request.tool_name != execution_name
            || target.request_generation() != Some(request.request_generation.as_str())
            || metadata.get("permission_decision_receipt").is_some()
        {
            return Err(AgentError::Tool(
                "Supervisor waiting approval does not match its request".into(),
            ));
        }
    }
    Ok(())
}

fn validated_replay_approvals(
    session: &Session,
    target: &PermissionReplayTarget,
) -> Result<Vec<PermissionReplayApproval>, AgentError> {
    let Some(replay_generation) = target.request_generation() else {
        return Ok(Vec::new());
    };
    let message = session
        .messages
        .get(target.result_message_index)
        .filter(|message| message.id == target.result_message_id)
        .ok_or_else(|| AgentError::Tool("permission replay result occurrence changed".into()))?;
    let mut approvals = message
        .metadata
        .as_ref()
        .and_then(|metadata| metadata.get(PERMISSION_REPLAY_APPROVALS_METADATA_KEY))
        .cloned()
        .map(serde_json::from_value::<Vec<PermissionReplayApproval>>)
        .transpose()
        .map_err(|error| {
            AgentError::Tool(format!(
                "permission replay approval ledger is invalid: {error}"
            ))
        })?
        .unwrap_or_default();
    let (request, receipt) = message_permission_contract(session, target)?;
    if receipt.session_id != session.id || request.request_generation != replay_generation {
        return Err(AgentError::Tool(
            "permission replay receipt does not match the active generation".into(),
        ));
    }
    approvals.push(PermissionReplayApproval {
        request,
        decision: receipt.decision,
    });
    if approvals.len() > 64 {
        return Err(AgentError::Tool(
            "permission replay approval ledger exceeded its safety bound".into(),
        ));
    }
    for approval in &approvals {
        validate_replay_approval(session, &target.tool_call.id, approval)?;
        match approval.decision.decision {
            PermissionDecisionKind::AllowOnce => {}
            PermissionDecisionKind::AllowSession | PermissionDecisionKind::AllowGlobal => {
                selected_replay_matcher(approval)?;
            }
            PermissionDecisionKind::AllowWorkspace => {
                let workspace = session
                    .workspace
                    .as_deref()
                    .map(str::trim)
                    .filter(|workspace| !workspace.is_empty());
                if workspace != approval.request.workspace_path.as_deref() {
                    return Err(AgentError::Tool(
                        "workspace permission replay does not match the durable session workspace"
                            .into(),
                    ));
                }
                selected_replay_matcher(approval)?;
            }
            PermissionDecisionKind::DenyOnce | PermissionDecisionKind::DenySession => {
                return Err(AgentError::Tool(
                    "denied permission unexpectedly carried an execution replay marker".into(),
                ));
            }
        }
    }
    Ok(approvals)
}

/// Rebuild the exact runtime capabilities needed by a typed approved replay.
///
/// The current receipt and any earlier contexts approved for the same concrete
/// invocation live on the generation-bound result message. This lets a daemon
/// restart recover an AllowOnce/AllowSession without broadening it, and lets a
/// tool with multiple permission contexts restart its checks while retaining
/// only the contexts the operator already approved.
pub fn restore_permission_replay_authorization(
    config: &PermissionConfig,
    session: &Session,
    target: &PermissionReplayTarget,
    execution_name: &str,
) -> Result<(), AgentError> {
    validate_permission_replay_authority(session, target, execution_name)?;
    let approvals = validated_replay_approvals(session, target)?;
    let Some(replay_generation) = target.request_generation() else {
        return Ok(());
    };
    config.set_session_workspace(
        session.id.clone(),
        session
            .workspace
            .as_deref()
            .map(str::trim)
            .filter(|workspace| !workspace.is_empty())
            .map(ToOwned::to_owned),
    );
    // Every record and matcher was validated before the first installation.
    for approval in &approvals {
        match approval.decision.decision {
            PermissionDecisionKind::AllowOnce => config
                .grant_once_for_generation(
                    &session.id,
                    &target.tool_call.id,
                    replay_generation,
                    approval.request.permission_type,
                    approval.request.resource.clone(),
                )
                .map_err(AgentError::Tool)?,
            PermissionDecisionKind::AllowSession => config
                .grant_typed_scoped_session_permission(
                    &session.id,
                    approval.request.permission_type,
                    selected_replay_matcher(approval)?,
                )
                .map_err(AgentError::Tool)?,
            PermissionDecisionKind::AllowWorkspace | PermissionDecisionKind::AllowGlobal => {}
            PermissionDecisionKind::DenyOnce | PermissionDecisionKind::DenySession => {
                unreachable!("validated above")
            }
        }
    }
    Ok(())
}

/// Convert a second typed permission gate encountered during replay into a new
/// durable pending question. The prior approval is appended to a bounded
/// per-invocation ledger; no model successor may start until every context has
/// been separately approved and the tool returns a real result.
pub fn repark_permission_replay(
    session: &mut Session,
    target: &PermissionReplayTarget,
    result: &ToolResult,
    execution_name: &str,
) -> Result<Option<ReparkedPermissionApproval>, AgentError> {
    let payload = match serde_json::from_str::<serde_json::Value>(&result.result) {
        Ok(payload)
            if result.display_preference.as_deref() == Some("request_permissions")
                && payload.get("status").and_then(serde_json::Value::as_str)
                    == Some("awaiting_permission_approval") =>
        {
            payload
        }
        _ => return Ok(None),
    };
    // Validate the old durable identity/receipt without installing grants again.
    let original_observation =
        validate_permission_replay_authority(session, target, execution_name)?;
    let original_operation = retained_operation(session, target)?;
    let prior_approvals = validated_replay_approvals(session, target)?;
    if payload
        .get(ExecutingSupervisorObservation::PERMISSION_REPLAY_METADATA_KEY)
        .is_some()
        || payload.get(PENDING_TOOL_OPERATION_METADATA_KEY).is_some()
    {
        return Err(AgentError::Tool(
            "Supervisor authority cannot originate in a replay result payload".into(),
        ));
    }
    let new_request = payload
        .get("permission_request")
        .cloned()
        .and_then(|value| serde_json::from_value::<PermissionRequest>(value).ok())
        .ok_or_else(|| {
            AgentError::Tool("replayed permission gate omitted typed request".to_string())
        })?;
    if new_request.session_id != session.id
        || new_request.request_id != target.tool_call.id
        || new_request.request_generation.trim().is_empty()
        || new_request.tool_name != execution_name
        || target.request_generation() == Some(new_request.request_generation.as_str())
        || prior_approvals
            .iter()
            .any(|approval| approval.request.request_generation == new_request.request_generation)
    {
        return Err(AgentError::Tool(
            "replayed permission gate returned an invalid next-generation identity".to_string(),
        ));
    }
    let question = payload
        .get("question")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("Permission required")
        .to_string();
    let options = payload
        .get("options")
        .and_then(serde_json::Value::as_array)
        .map(|options| {
            options
                .iter()
                .filter_map(serde_json::Value::as_str)
                .map(ToOwned::to_owned)
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();
    let allow_custom = payload
        .get("allow_custom")
        .and_then(serde_json::Value::as_bool)
        .unwrap_or(false);

    let (old_request, old_receipt) = message_permission_contract(session, target)?;
    if old_receipt.session_id != session.id {
        return Err(AgentError::Tool(
            "permission replay receipt session identity changed".to_string(),
        ));
    }
    let old_approval = PermissionReplayApproval {
        request: old_request,
        decision: old_receipt.decision,
    };
    validate_replay_approval(session, target.tool_call.id.as_str(), &old_approval)?;

    let message = session
        .messages
        .get_mut(target.result_message_index)
        .filter(|message| {
            message.id == target.result_message_id
                && message.tool_call_id.as_deref() == Some(target.tool_call.id.as_str())
        })
        .ok_or_else(|| {
            AgentError::Tool("permission replay result occurrence changed".to_string())
        })?;
    let metadata = message
        .metadata
        .get_or_insert_with(|| serde_json::Value::Object(Default::default()));
    let object = metadata.as_object_mut().ok_or_else(|| {
        AgentError::Tool("permission replay result metadata is malformed".to_string())
    })?;
    let mut approvals = object
        .get(PERMISSION_REPLAY_APPROVALS_METADATA_KEY)
        .cloned()
        .map(serde_json::from_value::<Vec<PermissionReplayApproval>>)
        .transpose()
        .map_err(|error| {
            AgentError::Tool(format!(
                "permission replay approval ledger is invalid: {error}"
            ))
        })?
        .unwrap_or_default();
    if !approvals.iter().any(|approval| approval == &old_approval) {
        approvals.push(old_approval);
    }
    if approvals.len() > 64 {
        return Err(AgentError::Tool(
            "permission replay approval ledger exceeded its safety bound".to_string(),
        ));
    }
    object.insert(
        PERMISSION_REPLAY_APPROVALS_METADATA_KEY.to_string(),
        serde_json::to_value(approvals).expect("permission replay approvals serialize"),
    );
    if let Some(original) = original_operation {
        object.insert(
            PENDING_TOOL_OPERATION_METADATA_KEY.to_string(),
            original
                .for_generation(&new_request.request_generation)
                .metadata_value(),
        );
    }
    if let Some(original) = original_observation {
        object.insert(
            ExecutingSupervisorObservation::PERMISSION_REPLAY_METADATA_KEY.to_string(),
            original.permission_replay_record(
                &session.id,
                &target.result_message_id,
                &target.tool_call,
                execution_name,
                &new_request.request_generation,
            ),
        );
    }
    object.insert(
        "permission_request".to_string(),
        serde_json::to_value(&new_request).expect("permission request serializes"),
    );
    object.remove("permission_decision_receipt");
    message.content.clone_from(&result.result);
    message.tool_success = Some(result.success);

    session.set_pending_question_with_source(
        target.tool_call.id.clone(),
        target.tool_call.function.name.clone(),
        question.clone(),
        options.clone(),
        allow_custom,
        PendingQuestionSource::PauseTool,
    );
    session
        .metadata
        .remove(crate::session_app::respond::PERMISSION_REEXECUTE_METADATA_KEY);
    session
        .metadata
        .remove(crate::session_app::respond::PERMISSION_REEXECUTE_GENERATION_METADATA_KEY);
    for stale_resume_marker in [
        "clarification_resume_pending",
        "conclusion_with_options_resume_pending",
        "execute.startup_handoff_at",
    ] {
        session.metadata.remove(stale_resume_marker);
    }
    session.metadata.insert(
        "runtime.suspend_reason".to_string(),
        "awaiting_clarification".to_string(),
    );
    Ok(Some(ReparkedPermissionApproval {
        question,
        options,
        allow_custom,
    }))
}

/// Strictly reload and adopt the authoritative permission posture for an
/// approved tool call. `execution_name` is the resolved exact execution owner,
/// or `None` when a legacy call's owner is no longer registered.
///
/// Storage errors, missing sessions, and malformed typed posture fail closed.
/// This function does not touch the replay marker, so callers can retain it
/// when returning the error. The coherent typed-mode/audit pair and Plan state
/// are committed to `session` only after every fallible validation succeeds.
pub async fn refresh_approval_replay_posture(
    storage: &dyn Storage,
    session: &mut Session,
    configured_mode: PermissionMode,
    execution_name: Option<&str>,
) -> Result<ApprovalReplayDecision, AgentError> {
    session
        .check_execution_compatibility()
        .map_err(|error| AgentError::Tool(error.to_string()))?;
    let latest = storage
        .load_execution_admission_snapshot(&session.id)
        .await
        .map_err(|error| {
            AgentError::Tool(format!(
                "authoritative approval replay posture refresh failed closed: {error}"
            ))
        })?
        .ok_or_else(|| {
            AgentError::Tool(
                "authoritative approval replay posture refresh failed closed: session missing"
                    .to_string(),
            )
        })?;
    latest
        .check_execution_compatibility()
        .map_err(|error| AgentError::Tool(error.to_string()))?;
    let latest_runtime = latest.agent_runtime_state.as_ref().ok_or_else(|| {
        AgentError::Tool(
            "authoritative approval replay posture refresh failed closed: typed mode missing"
                .to_string(),
        )
    })?;
    let disk_mode = latest_runtime.effective_permission_mode();
    let current_mode = session
        .agent_runtime_state
        .as_ref()
        .map(AgentRuntimeState::effective_permission_mode)
        .unwrap_or_default();

    let should_adopt_audit = bamboo_domain::disk_permission_posture_is_fresher(
        current_mode,
        &session.metadata,
        disk_mode,
        &latest.metadata,
    );
    let fresher_audit = bamboo_domain::fresher_disk_permission_audit(
        current_mode,
        &session.metadata,
        disk_mode,
        &latest.metadata,
    );
    if should_adopt_audit && fresher_audit.is_none() {
        return Err(AgentError::Tool(
            "authoritative approval replay posture refresh failed closed: complete audit unavailable"
                .to_string(),
        ));
    }

    // A replay may arrive after this Root ID was deleted and recreated, or
    // after a host narrowed its tool catalog. Validate the same durable birth
    // and revision proof used by normal tool dispatch before any grant or
    // replay-marker mutation.
    session
        .adopt_root_tool_authority_from(&latest)
        .map_err(|error| {
            AgentError::Tool(format!(
                "authoritative approval replay Root tool authority failed closed: {error}"
            ))
        })?;
    if let Some(audit) = fresher_audit {
        audit.write_to(&mut session.metadata);
    }
    let runtime = session
        .agent_runtime_state
        .get_or_insert_with(|| AgentRuntimeState::new(latest_runtime.run_id.clone()));
    runtime.set_permission_mode(disk_mode);
    runtime.plan_mode = latest_runtime.plan_mode.clone();

    let flags =
        ToolExecutionSessionFlags::from_session_and_configured_mode(session, configured_mode);
    let Some(execution_name) = execution_name else {
        return Ok(ApprovalReplayDecision::BlockedByUnavailableTool);
    };
    if !session.allows_model_tool_execution(execution_name) {
        Ok(ApprovalReplayDecision::BlockedByRootToolAuthority)
    } else if flags.plan_read_only && !plan_mode_allows_tool(execution_name) {
        Ok(ApprovalReplayDecision::BlockedByPlan(flags))
    } else {
        Ok(ApprovalReplayDecision::Execute(flags))
    }
}

/// Refresh the live policy, then preserve the historical custom-owner mode
/// boundary. The common spawn name is for the proven framework Root tool;
/// joining that allowlist cannot promote a retained custom implementation.
pub async fn refresh_retained_approval_replay_posture(
    storage: &dyn Storage,
    session: &mut Session,
    configured_mode: PermissionMode,
    execution_name: Option<&str>,
    execution: Option<&RetainedToolExecution>,
) -> Result<ApprovalReplayDecision, AgentError> {
    let decision =
        refresh_approval_replay_posture(storage, session, configured_mode, execution_name).await?;
    if matches!(decision, ApprovalReplayDecision::Execute(_))
        && execution_name == Some("spawn_session")
        && session.model_orchestration_only_enabled()
        && execution.and_then(RetainedToolExecution::execution_contract)
            != Some(bamboo_agent_core::tools::FRAMEWORK_ROOT_SPAWN_CONTRACT_V1)
    {
        return Ok(ApprovalReplayDecision::BlockedByRootToolAuthority);
    }
    Ok(decision)
}

#[cfg(test)]
pub(crate) mod framework_fixture;

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use bamboo_agent_core::storage::Storage;
    use bamboo_agent_core::tools::{FunctionCall, ToolCall, ToolResult};
    use bamboo_agent_core::{Message, Session};
    use bamboo_domain::{
        record_permission_audit, resolve_permission_mode, AgentRuntimeState, PermissionAuditSeed,
        PermissionMode, PlanModeState, PlanModeStatus, SessionPermissionMode,
    };
    use bamboo_tools::permission::{
        PermissionConfig, PermissionDecision, PermissionDecisionKind, PermissionDecisionReceipt,
        PermissionMatcher, PermissionMatcherKind, PermissionReasonCode, PermissionRequest,
        PermissionType, RiskLevel,
    };
    use tokio::sync::RwLock;

    use super::*;

    struct ReplayStorage {
        session: RwLock<Option<Session>>,
        fail_load: bool,
    }

    #[async_trait::async_trait]
    impl Storage for ReplayStorage {
        async fn save_session(&self, _session: &Session) -> std::io::Result<()> {
            Ok(())
        }

        async fn load_session(&self, _session_id: &str) -> std::io::Result<Option<Session>> {
            unreachable!("approval replay must use an explicit canonical/control-plane read")
        }

        async fn load_execution_admission_snapshot(
            &self,
            session_id: &str,
        ) -> std::io::Result<Option<Session>> {
            self.load_runtime_control_plane(session_id).await
        }

        async fn load_runtime_control_plane(
            &self,
            _session_id: &str,
        ) -> std::io::Result<Option<Session>> {
            if self.fail_load {
                return Err(std::io::Error::other("injected posture read failure"));
            }
            Ok(self.session.read().await.clone())
        }

        async fn delete_session(&self, _session_id: &str) -> std::io::Result<bool> {
            Ok(false)
        }
    }

    fn session_with_mode(id: &str, mode: SessionPermissionMode) -> Session {
        let mut session = Session::new(id, "model");
        let mut runtime = AgentRuntimeState::new("run");
        runtime.set_permission_mode(mode);
        session.agent_runtime_state = Some(runtime);
        let resolution = resolve_permission_mode(mode, PermissionMode::Default);
        record_permission_audit(
            &mut session.metadata,
            &PermissionAuditSeed::bamboo_runtime(1, resolution),
            Some("2026-07-31T00:00:00Z"),
        )
        .unwrap();
        session
    }

    fn permission_request(
        session_id: &str,
        generation: &str,
        resource: &str,
        decisions: Vec<PermissionDecisionKind>,
    ) -> PermissionRequest {
        PermissionRequest {
            request_id: "call-1".to_string(),
            request_generation: generation.to_string(),
            session_id: session_id.to_string(),
            workspace_path: Some("/workspace/a".to_string()),
            tool_name: "multi_context".to_string(),
            permission_type: PermissionType::ExecuteCommand,
            resource: resource.to_string(),
            operation_summary: format!("execute {resource}"),
            risk_level: RiskLevel::High,
            reason_code: PermissionReasonCode::RiskThreshold,
            effective_mode: PermissionMode::Default,
            bypass_requested: false,
            auto_approve_requested: false,
            policy_revision: 0,
            matched_rule: None,
            allowed_decisions: decisions,
            suggested_matchers: vec![PermissionMatcher {
                id: "exact_resource".to_string(),
                kind: PermissionMatcherKind::ExactResource,
                value: resource.to_string(),
            }],
        }
    }

    fn decision_receipt(
        session_id: &str,
        generation: &str,
        decision: PermissionDecisionKind,
    ) -> PermissionDecisionReceipt {
        PermissionDecisionReceipt {
            session_id: session_id.to_string(),
            decision: PermissionDecision {
                request_id: "call-1".to_string(),
                request_generation: generation.to_string(),
                decision,
                matcher_id: matches!(
                    decision,
                    PermissionDecisionKind::AllowSession
                        | PermissionDecisionKind::AllowWorkspace
                        | PermissionDecisionKind::AllowGlobal
                        | PermissionDecisionKind::DenySession
                )
                .then(|| "exact_resource".to_string()),
                expected_policy_revision: None,
                confirm_global: decision == PermissionDecisionKind::AllowGlobal,
            },
            decided_at: chrono::Utc::now(),
        }
    }

    fn approved_replay_session(
        generation: &str,
        resource: &str,
        decision: PermissionDecisionKind,
    ) -> Session {
        approved_replay_session_for("replay", "multi_context", generation, resource, decision)
    }

    fn approved_replay_session_for(
        session_id: &str,
        tool_name: &str,
        generation: &str,
        resource: &str,
        decision: PermissionDecisionKind,
    ) -> Session {
        let mut session = session_with_mode(session_id, SessionPermissionMode::Default);
        session.workspace = Some("/workspace/a".to_string());
        session.add_message(Message::assistant(
            "",
            Some(vec![ToolCall {
                id: "call-1".to_string(),
                tool_type: "function".to_string(),
                function: FunctionCall {
                    name: tool_name.to_string(),
                    arguments: "{}".to_string(),
                },
            }]),
        ));
        let mut request = permission_request(
            &session.id,
            generation,
            resource,
            vec![
                PermissionDecisionKind::AllowOnce,
                PermissionDecisionKind::AllowSession,
                PermissionDecisionKind::AllowWorkspace,
                PermissionDecisionKind::AllowGlobal,
                PermissionDecisionKind::DenyOnce,
                PermissionDecisionKind::DenySession,
            ],
        );
        request.tool_name = tool_name.to_string();
        let receipt = decision_receipt(&session.id, generation, decision);
        let mut result = Message::tool_result("call-1", "Selected response: Approve");
        result.id = "result-1".to_string();
        result.metadata = Some(serde_json::json!({
            "permission_request": request,
            "permission_decision_receipt": receipt,
        }));
        session.add_message(result);
        session.metadata.insert(
            crate::session_app::respond::PERMISSION_REEXECUTE_METADATA_KEY.to_string(),
            "call-1".to_string(),
        );
        session.metadata.insert(
            crate::session_app::respond::PERMISSION_REEXECUTE_GENERATION_METADATA_KEY.to_string(),
            generation.to_string(),
        );
        session.metadata.insert(
            "clarification_resume_pending".to_string(),
            "true".to_string(),
        );
        session.metadata.insert(
            "conclusion_with_options_resume_pending".to_string(),
            "true".to_string(),
        );
        session.metadata.insert(
            "execute.startup_handoff_at".to_string(),
            "2026-08-14T00:00:00Z".to_string(),
        );
        session
    }

    #[tokio::test]
    async fn common_spawn_name_does_not_promote_retained_custom_after_live_ultra_selection() {
        for version in [None, Some(1), Some(2)] {
            for contract in [
                None,
                Some("custom.spawn.v1"),
                Some(bamboo_agent_core::tools::FRAMEWORK_ROOT_SPAWN_CONTRACT_V1),
            ] {
                let mut owned =
                    session_with_mode("replay-original-owner", SessionPermissionMode::Default);
                owned.session_version = version;
                owned.metadata.insert(
                    "permission_reexecute_tool_call_id".into(),
                    "retained-custom".into(),
                );
                owned.add_message(Message::user("keep the original work"));
                let markers = owned.metadata.clone();
                let messages = serde_json::to_value(&owned.messages).unwrap();
                let mut latest = owned.clone();
                latest.set_root_orchestration_only(true).unwrap();
                let storage = ReplayStorage {
                    session: RwLock::new(Some(latest)),
                    fail_load: false,
                };
                let call = ToolCall {
                    id: "retained-custom".into(),
                    tool_type: "function".into(),
                    function: FunctionCall {
                        name: "spawn_session".into(),
                        arguments: "{}".into(),
                    },
                };
                let execution =
                    RetainedToolExecution::new(&call, "spawn_session", true, contract).unwrap();
                let decision = refresh_retained_approval_replay_posture(
                    &storage,
                    &mut owned,
                    PermissionMode::Default,
                    Some("spawn_session"),
                    Some(&execution),
                )
                .await
                .unwrap();
                if contract == Some(bamboo_agent_core::tools::FRAMEWORK_ROOT_SPAWN_CONTRACT_V1) {
                    assert!(matches!(decision, ApprovalReplayDecision::Execute(_)));
                } else {
                    assert_eq!(decision, ApprovalReplayDecision::BlockedByRootToolAuthority);
                }
                assert!(owned.root_orchestration_only_enabled());
                assert_eq!(owned.metadata, markers);
                assert_eq!(serde_json::to_value(&owned.messages).unwrap(), messages);
            }
        }
    }

    #[tokio::test]
    async fn latest_plan_blocks_mutating_replay_before_executor_entry() {
        let mut latest = session_with_mode("plan", SessionPermissionMode::Auto);
        latest.agent_runtime_state.as_mut().unwrap().plan_mode = Some(PlanModeState {
            entered_at: chrono::Utc::now(),
            pre_permission_mode: "auto".to_string(),
            plan_file_path: None,
            status: PlanModeStatus::Exploring,
        });
        let mut owned = session_with_mode("plan", SessionPermissionMode::Bypass);
        owned.created_at = latest.created_at;
        let storage: Arc<dyn Storage> = Arc::new(ReplayStorage {
            session: RwLock::new(Some(latest)),
            fail_load: false,
        });

        let decision = refresh_approval_replay_posture(
            storage.as_ref(),
            &mut owned,
            PermissionMode::Default,
            Some("Write"),
        )
        .await
        .unwrap();

        assert_eq!(
            decision,
            ApprovalReplayDecision::BlockedByPlan(
                bamboo_agent_core::tools::ToolExecutionSessionFlags {
                    bypass_permissions: false,
                    auto_approve_permissions: true,
                    plan_read_only: true,
                }
            )
        );
    }

    #[tokio::test]
    async fn latest_auto_and_bypass_map_to_distinct_executor_flags() {
        for (mode, expected_bypass, expected_auto) in [
            (SessionPermissionMode::Auto, false, true),
            (SessionPermissionMode::Bypass, true, false),
        ] {
            let latest = session_with_mode("flags", mode);
            let mut owned = session_with_mode("flags", SessionPermissionMode::Default);
            owned.created_at = latest.created_at;
            let storage: Arc<dyn Storage> = Arc::new(ReplayStorage {
                session: RwLock::new(Some(latest)),
                fail_load: false,
            });

            let decision = refresh_approval_replay_posture(
                storage.as_ref(),
                &mut owned,
                PermissionMode::Default,
                Some("Write"),
            )
            .await
            .unwrap();
            let ApprovalReplayDecision::Execute(flags) = decision else {
                panic!("non-Plan replay must remain executable");
            };
            assert_eq!(flags.bypass_permissions, expected_bypass);
            assert_eq!(flags.auto_approve_permissions, expected_auto);
            assert!(!flags.plan_read_only);
        }
    }

    #[tokio::test]
    async fn supervisor_delegation_first_blocks_implementation_approval_replay_in_both_modes() {
        use bamboo_domain::{SessionAuthorityIdentity, DEFAULT_SUPERVISOR_SESSION_ID};
        for ultra in [false, true] {
            let mut owned =
                session_with_mode(DEFAULT_SUPERVISOR_SESSION_ID, SessionPermissionMode::Auto);
            owned.authority_identity = SessionAuthorityIdentity::Supervisor {
                incarnation_id: uuid::Uuid::new_v4(),
            };
            owned.set_root_orchestration_only(ultra).unwrap();
            let storage = ReplayStorage {
                session: RwLock::new(Some(owned.clone())),
                fail_load: false,
            };
            for name in [
                "Read",
                "Grep",
                "Glob",
                "GetFileInfo",
                "ViewImage",
                "Write",
                "Edit",
                "exec_command",
                "write_stdin",
                "load_skill",
                "workflow_run",
            ] {
                assert_eq!(
                    refresh_approval_replay_posture(
                        &storage,
                        &mut owned,
                        PermissionMode::Default,
                        Some(name)
                    )
                    .await
                    .unwrap(),
                    ApprovalReplayDecision::BlockedByRootToolAuthority,
                    "{ultra}: {name}"
                );
            }
            for name in ["Plan", "SubAgent", "session_control"] {
                assert!(
                    matches!(
                        refresh_approval_replay_posture(
                            &storage,
                            &mut owned,
                            PermissionMode::Default,
                            Some(name)
                        )
                        .await
                        .unwrap(),
                        ApprovalReplayDecision::Execute(_)
                    ),
                    "{ultra}: {name}"
                );
            }
        }
    }

    #[tokio::test]
    async fn supervisor_retained_once_cannot_override_role_or_consume_replay_markers() {
        use bamboo_domain::{SessionAuthorityIdentity, DEFAULT_SUPERVISOR_SESSION_ID};
        for ultra in [false, true] {
            let mut owned = approved_replay_session_for(
                DEFAULT_SUPERVISOR_SESSION_ID,
                "exec_command",
                "retained-generation",
                "echo isolated-probe",
                PermissionDecisionKind::AllowOnce,
            );
            owned.authority_identity = SessionAuthorityIdentity::Supervisor {
                incarnation_id: uuid::Uuid::new_v4(),
            };
            owned.set_root_orchestration_only(ultra).unwrap();
            let target =
                find_permission_replay_target(&owned, "call-1", Some("retained-generation"))
                    .unwrap();
            let original =
                ExecutingSupervisorObservation::capture_from_executing_session(&owned).unwrap();
            let record = original.permission_replay_record(
                &owned.id,
                &target.result_message_id,
                &target.tool_call,
                "exec_command",
                "retained-generation",
            );
            owned.messages[target.result_message_index]
                .metadata
                .as_mut()
                .unwrap()["permission.executing_supervisor.v1"] = record;
            assert_eq!(
                validate_permission_replay_authority(&owned, &target, "exec_command").unwrap(),
                Some(original),
                "the retained typed receipt is valid, not the refusal cause"
            );
            let permissions = PermissionConfig::new();
            // An isolated historical in-memory grant is present before refresh.
            restore_permission_replay_authorization(&permissions, &owned, &target, "exec_command")
                .unwrap();
            let storage = ReplayStorage {
                session: RwLock::new(Some(owned.clone())),
                fail_load: false,
            };
            let markers = owned.metadata.clone();
            let history = serde_json::to_value(&owned.messages).unwrap();
            let execution =
                RetainedToolExecution::new(&target.tool_call, "exec_command", true, None).unwrap();
            assert_eq!(
                refresh_retained_approval_replay_posture(
                    &storage,
                    &mut owned,
                    PermissionMode::Default,
                    Some("exec_command"),
                    Some(&execution),
                )
                .await
                .unwrap(),
                ApprovalReplayDecision::BlockedByRootToolAuthority
            );
            assert_eq!(owned.metadata, markers);
            assert_eq!(serde_json::to_value(&owned.messages).unwrap(), history);
            assert!(
                permissions.consume_once_for_generation(
                    &owned.id,
                    "call-1",
                    "retained-generation",
                    PermissionType::ExecuteCommand,
                    "echo isolated-probe",
                ),
                "role refusal must not consume the old grant"
            );
        }
    }

    #[tokio::test]
    async fn durable_root_tool_authority_blocks_denied_replay_owner() {
        let mut owned = session_with_mode("root-replay", SessionPermissionMode::Auto);
        let mut latest = owned.clone();
        latest.set_root_orchestration_only(true).unwrap();
        let storage = ReplayStorage {
            session: RwLock::new(Some(latest)),
            fail_load: false,
        };

        assert_eq!(
            refresh_approval_replay_posture(
                &storage,
                &mut owned,
                PermissionMode::Default,
                Some("Write"),
            )
            .await
            .unwrap(),
            ApprovalReplayDecision::BlockedByRootToolAuthority
        );
        assert!(owned.root_orchestration_only_enabled());
        assert!(matches!(
            refresh_approval_replay_posture(
                &storage,
                &mut owned,
                PermissionMode::Default,
                Some("Read")
            )
            .await
            .unwrap(),
            ApprovalReplayDecision::Execute(_)
        ));
        assert_eq!(
            refresh_approval_replay_posture(&storage, &mut owned, PermissionMode::Default, None)
                .await
                .unwrap(),
            ApprovalReplayDecision::BlockedByUnavailableTool
        );
    }

    #[tokio::test]
    async fn stale_or_corrupt_root_authority_keeps_replay_retryable() {
        for invalid_birth in [true, false] {
            let mut latest = session_with_mode("stale-root-replay", SessionPermissionMode::Auto);
            let mut owned = latest.clone();
            owned.metadata.insert(
                crate::session_app::respond::PERMISSION_REEXECUTE_METADATA_KEY.into(),
                "call-1".into(),
            );
            if invalid_birth {
                latest.created_at += chrono::Duration::microseconds(1);
            } else {
                latest.root_orchestration_only = true;
            }
            let storage = ReplayStorage {
                session: RwLock::new(Some(latest)),
                fail_load: false,
            };
            let before = owned.clone();
            let error = refresh_approval_replay_posture(
                &storage,
                &mut owned,
                PermissionMode::Default,
                Some("Write"),
            )
            .await
            .expect_err("stale Root authority must fail closed");
            assert!(error
                .to_string()
                .contains("Root tool authority failed closed"));
            assert_eq!(owned.created_at, before.created_at);
            assert_eq!(owned.metadata, before.metadata);
        }
    }

    #[tokio::test]
    async fn real_v2_missing_or_corrupt_root_proof_keeps_replay_marker() {
        for corrupt in [false, true] {
            let directory = tempfile::tempdir().unwrap();
            let store = bamboo_storage::SessionStoreV2::new(directory.path().to_path_buf())
                .await
                .unwrap();
            let mut owned = session_with_mode("v2-root-replay", SessionPermissionMode::Auto);
            owned.metadata.insert(
                crate::session_app::respond::PERMISSION_REEXECUTE_METADATA_KEY.into(),
                "call-1".into(),
            );
            store.save_session(&owned).await.unwrap();
            let mut latest = owned.clone();
            latest.set_root_orchestration_only(true).unwrap();
            store.save_session(&latest).await.unwrap();
            let proof = store
                .sessions_root_dir()
                .join(&owned.id)
                .join("root-tool-authority.json");
            if corrupt {
                tokio::fs::write(&proof, b"{").await.unwrap();
            } else {
                tokio::fs::remove_file(&proof).await.unwrap();
            }

            let before = owned.clone();
            let error = refresh_approval_replay_posture(
                &store,
                &mut owned,
                PermissionMode::Default,
                Some("Write"),
            )
            .await
            .expect_err("invalid durable Root proof must fail closed");
            assert!(error.to_string().contains("failed closed"));
            assert_eq!(owned.metadata, before.metadata);
            assert_eq!(
                owned.root_tool_authority_revision,
                before.root_tool_authority_revision
            );
        }
    }

    #[tokio::test]
    async fn missing_or_failed_authoritative_load_fails_closed_without_mutation() {
        for storage in [
            ReplayStorage {
                session: RwLock::new(None),
                fail_load: false,
            },
            ReplayStorage {
                session: RwLock::new(None),
                fail_load: true,
            },
        ] {
            let mut owned = session_with_mode("missing", SessionPermissionMode::Auto);
            let before = owned.clone();
            let error = refresh_approval_replay_posture(
                &storage,
                &mut owned,
                PermissionMode::Default,
                Some("Write"),
            )
            .await
            .expect_err("unavailable durable posture must fail closed");
            assert!(error.to_string().contains("failed closed"));
            assert_eq!(owned.agent_runtime_state, before.agent_runtime_state);
            assert_eq!(owned.metadata, before.metadata);
        }
    }

    #[test]
    fn restart_restores_refreshed_typed_allow_once_and_session_authorizations() {
        for expected_policy_revision in [None, Some(0), Some(1)] {
            let mut allow_once = approved_replay_session(
                "generation-1",
                "context-one",
                PermissionDecisionKind::AllowOnce,
            );
            // A same-generation refresh leaves the original durable request at 0.
            allow_once
                .messages
                .last_mut()
                .unwrap()
                .metadata
                .as_mut()
                .unwrap()["permission_decision_receipt"]["decision"]["expected_policy_revision"] =
                serde_json::json!(expected_policy_revision);
            let target =
                find_permission_replay_target(&allow_once, "call-1", Some("generation-1")).unwrap();
            assert!(
                validate_permission_replay_authority(&allow_once, &target, "multi_context")
                    .unwrap()
                    .is_none()
            );
            let restarted = PermissionConfig::new();
            restore_permission_replay_authorization(
                &restarted,
                &allow_once,
                &target,
                "multi_context",
            )
            .unwrap();
            assert!(!restarted.consume_once_for_generation(
                "replay",
                "call-1",
                "generation-1",
                PermissionType::ExecuteCommand,
                "different-context"
            ));
            assert!(restarted.consume_once_for_generation(
                "replay",
                "call-1",
                "generation-1",
                PermissionType::ExecuteCommand,
                "context-one"
            ));
            assert!(!restarted.consume_once_for_generation(
                "replay",
                "call-1",
                "generation-1",
                PermissionType::ExecuteCommand,
                "context-one"
            ));
        }

        let allow_session = approved_replay_session(
            "generation-session",
            "git status",
            PermissionDecisionKind::AllowSession,
        );
        let target =
            find_permission_replay_target(&allow_session, "call-1", Some("generation-session"))
                .unwrap();
        let restarted = PermissionConfig::new();
        restore_permission_replay_authorization(
            &restarted,
            &allow_session,
            &target,
            "multi_context",
        )
        .unwrap();
        assert!(restarted.is_scoped_session_granted(
            "replay",
            PermissionType::ExecuteCommand,
            "git status"
        ));
    }

    #[test]
    fn typed_replay_rejects_changed_execution_owner_before_restoring_grants() {
        let session = approved_replay_session(
            "generation-owner",
            "git status",
            PermissionDecisionKind::AllowOnce,
        );
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-owner")).unwrap();
        let restarted = PermissionConfig::new();
        let error = restore_permission_replay_authorization(
            &restarted,
            &session,
            &target,
            "new_exact_custom_owner",
        )
        .unwrap_err();
        assert!(error.to_string().contains("approval owner"));
        assert!(!restarted.consume_once_for_generation(
            "replay",
            "call-1",
            "generation-owner",
            PermissionType::ExecuteCommand,
            "git status"
        ));
    }

    #[test]
    fn workspace_replay_uses_only_durable_session_workspace() {
        let session = approved_replay_session(
            "generation-workspace",
            "cargo test",
            PermissionDecisionKind::AllowWorkspace,
        );
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-workspace"))
                .unwrap();
        let restarted = PermissionConfig::new();
        restore_permission_replay_authorization(&restarted, &session, &target, "multi_context")
            .unwrap();
        assert_eq!(
            restarted.session_workspace("replay").as_deref(),
            Some("/workspace/a")
        );

        let mut mismatched = session;
        mismatched.workspace = Some("/workspace/b".to_string());
        let target =
            find_permission_replay_target(&mismatched, "call-1", Some("generation-workspace"))
                .unwrap();
        assert!(restore_permission_replay_authorization(
            &PermissionConfig::new(),
            &mismatched,
            &target,
            "multi_context"
        )
        .is_err());
    }

    #[test]
    fn second_permission_context_reparks_and_accumulates_exact_approvals() {
        let mut session = approved_replay_session(
            "generation-1",
            "context-one",
            PermissionDecisionKind::AllowOnce,
        );
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-1")).unwrap();
        let next_request = permission_request(
            &session.id,
            "generation-2",
            "context-two",
            vec![
                PermissionDecisionKind::AllowOnce,
                PermissionDecisionKind::DenyOnce,
            ],
        );
        let next_result = ToolResult {
            success: true,
            result: serde_json::json!({
                "status": "awaiting_permission_approval",
                "question": "Approve context two?",
                "options": ["Approve", "Deny"],
                "allow_custom": false,
                "permission_request": next_request,
            })
            .to_string(),
            display_preference: Some("request_permissions".to_string()),
            images: Vec::new(),
        };

        let reparked =
            repark_permission_replay(&mut session, &target, &next_result, "multi_context")
                .unwrap()
                .expect("second context must park");
        assert_eq!(reparked.question, "Approve context two?");
        assert_eq!(
            session
                .pending_question
                .as_ref()
                .map(|pending| pending.tool_call_id.as_str()),
            Some("call-1")
        );
        for marker in [
            crate::session_app::respond::PERMISSION_REEXECUTE_METADATA_KEY,
            crate::session_app::respond::PERMISSION_REEXECUTE_GENERATION_METADATA_KEY,
            "clarification_resume_pending",
            "conclusion_with_options_resume_pending",
            "execute.startup_handoff_at",
        ] {
            assert!(
                !session.metadata.contains_key(marker),
                "stale marker {marker}"
            );
        }

        let message = session.messages.last_mut().unwrap();
        message.content = "Selected response: Approve".to_string();
        message
            .metadata
            .as_mut()
            .unwrap()
            .as_object_mut()
            .unwrap()
            .insert(
                "permission_decision_receipt".to_string(),
                serde_json::to_value(decision_receipt(
                    "replay",
                    "generation-2",
                    PermissionDecisionKind::AllowOnce,
                ))
                .unwrap(),
            );
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-2")).unwrap();
        let restarted = PermissionConfig::new();
        restore_permission_replay_authorization(&restarted, &session, &target, "multi_context")
            .unwrap();
        for resource in ["context-one", "context-two"] {
            assert!(restarted.consume_once_for_generation(
                "replay",
                "call-1",
                "generation-2",
                PermissionType::ExecuteCommand,
                resource,
            ));
        }
    }

    struct FrozenOwnerProbe {
        names: Vec<&'static str>,
        resolver_queries: std::sync::atomic::AtomicUsize,
    }

    #[test]
    fn retained_framework_replay_needs_actual_instance_not_same_contract_and_keeps_original_bytes()
    {
        use framework_fixture::{park, FixtureTools};
        for version in [None, Some(1), Some(2)] {
            let mut session = Session::new("proof-replay", "model");
            session.session_version = version;
            let call = ToolCall {
                id: "parked-root".into(),
                tool_type: "function".into(),
                function: FunctionCall {
                    name: "default::spawnSession".into(),
                    arguments: "{ \"message\":\"retained original bytes\" }".into(),
                },
            };
            let host = FixtureTools::new(&["spawn_session"]);
            let target = park(
                &mut session,
                &call,
                "spawn_session",
                true,
                Some(bamboo_agent_core::tools::FRAMEWORK_ROOT_SPAWN_CONTRACT_V1),
            );
            let original = serde_json::to_value(&session).unwrap();
            assert_eq!(
                resolve_permission_replay_owner(&session, &target, host.as_ref())
                    .unwrap()
                    .as_deref(),
                Some("spawn_session")
            );
            host.proofs_enabled
                .store(false, std::sync::atomic::Ordering::SeqCst);
            assert_eq!(
                host.execution_contract_for_exact("spawn_session"),
                Some(bamboo_agent_core::tools::FRAMEWORK_ROOT_SPAWN_CONTRACT_V1)
            );
            assert!(
                resolve_permission_replay_owner(&session, &target, host.as_ref())
                    .unwrap()
                    .is_none()
            );
            let shadow = FrozenOwnerProbe {
                names: vec!["spawn_session"],
                resolver_queries: Default::default(),
            };
            assert!(resolve_permission_replay_owner(&session, &target, &shadow)
                .unwrap()
                .is_none());
            assert_eq!(
                shadow
                    .resolver_queries
                    .load(std::sync::atomic::Ordering::SeqCst),
                0
            );
            assert_eq!(target.tool_call(), &call);
            assert_eq!(serde_json::to_value(&session).unwrap(), original);
            assert_eq!(host.gates(), 0);
            assert!(host.invocations().is_empty());
        }
    }

    #[async_trait::async_trait]
    impl ToolExecutor for FrozenOwnerProbe {
        async fn execute(&self, _: &ToolCall) -> Result<ToolResult, bamboo_agent_core::ToolError> {
            panic!("pure owner inspection must not enter any executor")
        }
        fn list_tools(&self) -> Vec<bamboo_agent_core::ToolSchema> {
            self.names.iter().map(|name| serde_json::from_value(serde_json::json!({
                "type":"function", "function":{"name":name, "description":"fixture", "parameters":{"type":"object"}}
            })).unwrap()).collect()
        }
        fn resolve_reference_for_session(&self, _: &Session, _: &str) -> Option<String> {
            self.resolver_queries
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            // Simulate the future fresh-call contract or a newly shadowing
            // exact custom registration. Retained replay must not consult it.
            Some("spawn_session".into())
        }
        fn exact_tool_owner(&self, name: &str) -> Option<&dyn ToolExecutor> {
            self.names
                .contains(&name)
                .then_some(self as &dyn ToolExecutor)
        }
    }

    fn frozen_spawn_session(version: Option<u32>, owner: &str) -> Session {
        let mut session = approved_replay_session(
            "generation-1",
            "first-context",
            PermissionDecisionKind::AllowOnce,
        );
        session.session_version = version;
        let call = session.messages[0]
            .tool_calls
            .as_mut()
            .unwrap()
            .first_mut()
            .unwrap();
        call.function.name = "default::spawnSession".into();
        call.function.arguments = "{ \"prompt\":\"retained bytes\" }".into();
        let call = call.clone();
        let caller = bamboo_domain::SessionCallerReference::from(&session);
        let result = session.messages.last_mut().unwrap();
        result.metadata.as_mut().unwrap()["permission_request"]["tool_name"] =
            serde_json::json!(owner);
        result.metadata.as_mut().unwrap()[PENDING_TOOL_OPERATION_METADATA_KEY] =
            PendingToolOperation::capture(caller, &result.id, &call, owner, "generation-1")
                .metadata_value();
        session
    }

    #[test]
    fn frozen_spawn_owner_is_independent_of_fresh_alias_and_durable_reopen() {
        for version in [None, Some(1), Some(2)] {
            for owner in ["SubAgent", "spawn_session", "spawnSession"] {
                let session = frozen_spawn_session(version, owner);
                let session: Session =
                    serde_json::from_value(serde_json::to_value(session).unwrap()).unwrap();
                let target =
                    find_permission_replay_target(&session, "call-1", Some("generation-1"))
                        .unwrap();
                let probe = FrozenOwnerProbe {
                    names: vec!["SubAgent", "spawn_session", "spawnSession"],
                    resolver_queries: Default::default(),
                };
                if version == Some(2) && owner == "spawn_session" {
                    let before = serde_json::to_value(&session).unwrap();
                    assert!(resolve_permission_replay_owner(&session, &target, &probe).is_err());
                    assert_eq!(serde_json::to_value(&session).unwrap(), before);
                    continue;
                }
                assert_eq!(
                    resolve_permission_replay_owner(&session, &target, &probe)
                        .unwrap()
                        .as_deref(),
                    Some(owner)
                );
                assert_eq!(
                    probe
                        .resolver_queries
                        .load(std::sync::atomic::Ordering::SeqCst),
                    0
                );
                assert_eq!(target.tool_call().function.name, "default::spawnSession");
                assert_eq!(
                    target.tool_call().function.arguments,
                    "{ \"prompt\":\"retained bytes\" }"
                );
            }
        }
    }

    #[test]
    fn replay_caller_requires_its_original_host_record_and_never_recaptures() {
        let session = frozen_spawn_session(None, "SubAgent");
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-1")).unwrap();
        let original = bamboo_domain::SessionCallerReference::from(&session);
        assert_eq!(
            restore_permission_replay_caller(&session, &target).unwrap(),
            Some(original)
        );

        let mut replacement = session.clone();
        replacement.created_at += chrono::Duration::seconds(1);
        let before = serde_json::to_value(&replacement).unwrap();
        assert!(restore_permission_replay_caller(&replacement, &target).is_err());
        assert_eq!(serde_json::to_value(&replacement).unwrap(), before);

        let mut legacy = session.clone();
        legacy
            .messages
            .last_mut()
            .unwrap()
            .metadata
            .as_mut()
            .unwrap()
            .as_object_mut()
            .unwrap()
            .remove(PENDING_TOOL_OPERATION_METADATA_KEY);
        assert!(restore_permission_replay_caller(&legacy, &target)
            .unwrap()
            .is_none());
        let mut forged = legacy;
        forged.messages.last_mut().unwrap().content = session
            .messages
            .last()
            .unwrap()
            .metadata
            .as_ref()
            .unwrap()
            .to_string();
        assert!(restore_permission_replay_caller(&forged, &target).is_err());
    }

    #[test]
    fn unknown_legacy_spawn_refuses_without_consuming_markers_or_reinterpreting() {
        for version in [None, Some(1), Some(2)] {
            for typed in [false, true] {
                let mut session = frozen_spawn_session(version, "SubAgent");
                let metadata = session
                    .messages
                    .last_mut()
                    .unwrap()
                    .metadata
                    .as_mut()
                    .unwrap()
                    .as_object_mut()
                    .unwrap();
                metadata.remove(PENDING_TOOL_OPERATION_METADATA_KEY);
                if !typed {
                    metadata.clear();
                }
                let target = find_permission_replay_target(&session, "call-1", None).unwrap();
                let before = serde_json::to_value(&session).unwrap();
                let probe = FrozenOwnerProbe {
                    names: vec!["SubAgent", "spawn_session"],
                    resolver_queries: Default::default(),
                };
                let error = resolve_permission_replay_owner(&session, &target, &probe).unwrap_err();
                assert!(error.to_string().contains("Resend"), "{error}");
                assert_eq!(serde_json::to_value(&session).unwrap(), before);
                assert_eq!(
                    probe
                        .resolver_queries
                        .load(std::sync::atomic::Ordering::SeqCst),
                    0
                );
            }
        }
    }

    #[test]
    fn frozen_owner_missing_or_changed_arguments_never_fall_through_to_new_root() {
        let mut session = frozen_spawn_session(None, "SubAgent");
        let probe = FrozenOwnerProbe {
            names: vec!["spawn_session"],
            resolver_queries: Default::default(),
        };
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-1")).unwrap();
        assert!(resolve_permission_replay_owner(&session, &target, &probe)
            .unwrap()
            .is_none());
        session.messages[0].tool_calls.as_mut().unwrap()[0]
            .function
            .arguments = "{\"project_id\":\"unapproved\"}".into();
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-1")).unwrap();
        let before = serde_json::to_value(&session).unwrap();
        assert!(resolve_permission_replay_owner(&session, &target, &probe).is_err());
        assert_eq!(serde_json::to_value(&session).unwrap(), before);
        assert_eq!(
            probe
                .resolver_queries
                .load(std::sync::atomic::Ordering::SeqCst),
            0
        );
    }

    #[test]
    fn initial_permission_park_binds_ordinary_call_and_rejects_payload_authority_before_answer() {
        let mut session = Session::new("ordinary", "model");
        let call = ToolCall {
            id: "call-1".into(),
            tool_type: "function".into(),
            function: FunctionCall {
                name: "default::spawnSession".into(),
                arguments: "{ \"prompt\": \"bounded\" }".into(),
            },
        };
        session.add_message(Message::assistant("", Some(vec![call.clone()])));
        let mut request = permission_request(
            &session.id,
            "generation-1",
            "first-context",
            vec![PermissionDecisionKind::AllowOnce],
        );
        request.tool_name = "SubAgent".into();
        let mut result = Message::tool_result(&call.id, serde_json::json!({"status":"awaiting_permission_approval", "permission_request": request}).to_string());
        PermissionReplayOrigin::new(
            Some(bamboo_domain::SessionCallerReference::from(&session)),
            None,
            &session.id,
            &call,
            "SubAgent",
            None,
            true,
        )
        .bind_waiting_message(&mut result);
        assert!(result
            .metadata
            .as_ref()
            .unwrap()
            .get(PENDING_TOOL_OPERATION_METADATA_KEY)
            .is_some());
        session.add_message(result);
        let target = find_permission_replay_target(&session, "call-1", None).unwrap();
        assert!(retained_operation(&session, &target).unwrap().is_some());
        let payload: serde_json::Value =
            serde_json::from_str(&session.messages[1].content).unwrap();
        let mut injected = payload;
        injected[PENDING_TOOL_OPERATION_METADATA_KEY] = serde_json::json!({"version":1});
        session.messages[1].content = injected.to_string();
        let before = serde_json::to_value(&session).unwrap();
        assert!(retained_operation(&session, &target).is_err());
        assert_eq!(serde_json::to_value(&session).unwrap(), before);
    }

    #[test]
    fn second_permission_gate_preserves_frozen_operation_and_rotates_only_generation() {
        let mut session = approved_replay_session(
            "generation-1",
            "context-one",
            PermissionDecisionKind::AllowOnce,
        );
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-1")).unwrap();
        let original = PendingToolOperation::capture(
            bamboo_domain::SessionCallerReference::from(&session),
            &session.messages.last().unwrap().id,
            target.tool_call(),
            "multi_context",
            "generation-1",
        );
        session
            .messages
            .last_mut()
            .unwrap()
            .metadata
            .as_mut()
            .unwrap()[PENDING_TOOL_OPERATION_METADATA_KEY] = original.metadata_value();
        let next = ToolResult {
            success: true,
            result: serde_json::json!({"status":"awaiting_permission_approval", "question":"second context", "options":["Approve","Deny"],
                "permission_request":permission_request(&session.id, "generation-2", "context-two", vec![PermissionDecisionKind::AllowOnce])}).to_string(),
            display_preference: Some("request_permissions".into()), images: Vec::new(),
        };
        assert!(
            repark_permission_replay(&mut session, &target, &next, "multi_context")
                .unwrap()
                .is_some()
        );
        let next_target =
            find_permission_replay_target(&session, "call-1", Some("generation-2")).unwrap();
        assert_eq!(
            retained_operation(&session, &next_target)
                .unwrap()
                .unwrap()
                .metadata_value(),
            original.for_generation("generation-2").metadata_value()
        );
        assert!(retained_operation(&session, &target).is_err());
    }

    #[test]
    fn legacy_supervisor_exact_binding_is_read_without_migration_or_live_alias_resolution() {
        let mut session = frozen_spawn_session(Some(1), "SubAgent");
        session.id = bamboo_domain::DEFAULT_SUPERVISOR_SESSION_ID.into();
        session.root_session_id = session.id.clone();
        session.authority_identity = bamboo_domain::SessionAuthorityIdentity::Supervisor {
            incarnation_id: uuid::Uuid::new_v4(),
        };
        let target =
            find_permission_replay_target(&session, "call-1", Some("generation-1")).unwrap();
        let observation =
            ExecutingSupervisorObservation::capture_from_executing_session(&session).unwrap();
        let record = observation.permission_replay_record(
            &session.id,
            &session.messages.last().unwrap().id,
            target.tool_call(),
            "SubAgent",
            "generation-1",
        );
        let id = session.id.clone();
        let metadata = session
            .messages
            .last_mut()
            .unwrap()
            .metadata
            .as_mut()
            .unwrap();
        metadata
            .as_object_mut()
            .unwrap()
            .remove(PENDING_TOOL_OPERATION_METADATA_KEY);
        metadata["permission_request"]["session_id"] = serde_json::json!(&id);
        metadata["permission_decision_receipt"]["session_id"] = serde_json::json!(&id);
        metadata[ExecutingSupervisorObservation::PERMISSION_REPLAY_METADATA_KEY] = record;
        let before = serde_json::to_value(&session).unwrap();
        let probe = FrozenOwnerProbe {
            names: vec!["SubAgent", "spawn_session"],
            resolver_queries: Default::default(),
        };
        assert_eq!(
            resolve_permission_replay_owner(&session, &target, &probe)
                .unwrap()
                .as_deref(),
            Some("SubAgent")
        );
        assert_eq!(
            validate_permission_replay_authority(&session, &target, "SubAgent").unwrap(),
            Some(observation)
        );
        assert_eq!(
            probe
                .resolver_queries
                .load(std::sync::atomic::Ordering::SeqCst),
            0
        );
        assert_eq!(serde_json::to_value(&session).unwrap(), before);
    }
}