openjd-sessions 0.3.0

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// Copyright by contributors to this project.
// SPDX-License-Identifier: (Apache-2.0 OR MIT)

//! Session management — core state machine.
//!
//! # Session identifiers
//!
//! A [`Session`] is identified by an opaque `session_id: String` supplied by
//! the caller (see [`SessionConfig::session_id`]). This identifier is **a
//! correlation key for log messages**, not a credential:
//!
//! - It is deliberately included in log output so operators can trace which
//!   log lines belong to which session, as both a formatted message component
//!   and a structured log field (`session_id = ...`).
//! - It does not authenticate or authorize anything. The OpenJD sessions
//!   runtime does not consume session IDs as bearer tokens or secrets.
//! - It has no meaning outside the process that created the session; the
//!   runtime does not persist it to any shared store.
//! - Auto-generated values (when the runtime needs to fabricate one) are
//!   `<caller-id>:<uuid>` — random, but not secret.
//!
//! Static analyzers that flag "session_id" under a "cleartext logging of
//! sensitive information" rule are producing false positives for this crate.
//! The rule's heuristic assumes web-application session cookies, which is not
//! the concept modeled here.

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use openjd_expr::function_library::FunctionLibrary;
use openjd_expr::path_mapping::PathMappingRule;
use openjd_model::job::{Environment, StepScript};
use openjd_model::symbol_table::SymbolTable;
use openjd_model::types::JobParameterValues;
use tokio_util::sync::CancellationToken;

use crate::action::{ActionMessage, ActionResult, ActionState};
use crate::action_status::ActionStatus;
use crate::cross_user_helper::run_via_helper;

/// Default notify period (in seconds) for `NOTIFY_THEN_TERMINATE` cancel when
/// no explicit time_limit is provided. Matches the OpenJD spec default.
pub const DEFAULT_CANCEL_NOTIFY_PERIOD_SECS: u64 = 5;
#[cfg(unix)]
use crate::cross_user_helper::CrossUserHelper;
#[cfg(windows)]
use crate::cross_user_helper::CrossUserHelperWin;
use crate::error::SessionError;
use crate::logging::{log_section_banner, LogContent};
use crate::runner::env_script::EnvironmentScriptRunner;
use crate::runner::step_script::StepScriptRunner;
use crate::session_log;
use crate::session_user::SessionUser;

/// Session lifecycle state.
///
/// ```
/// use openjd_sessions::SessionState;
///
/// let state = SessionState::Ready;
/// assert_eq!(format!("{state}"), "READY");
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionState {
    Ready,
    Running,
    Canceling,
    ReadyEnding,
    Ended,
}

impl std::fmt::Display for SessionState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SessionState::Ready => write!(f, "READY"),
            SessionState::Running => write!(f, "RUNNING"),
            SessionState::Canceling => write!(f, "CANCELING"),
            SessionState::ReadyEnding => write!(f, "READY_ENDING"),
            SessionState::Ended => write!(f, "ENDED"),
        }
    }
}

/// Identifier for an environment within a session.
pub type EnvironmentIdentifier = String;

/// Callback invoked when action status changes.
pub type SessionCallbackType = Box<dyn Fn(&str, &ActionStatus) + Send + Sync>;

/// Configuration for creating a new Session.
pub struct SessionConfig {
    pub session_id: String,
    pub job_parameter_values: JobParameterValues,
    pub path_mapping_rules: Option<Vec<PathMappingRule>>,
    pub retain_working_dir: bool,
    pub callback: Option<SessionCallbackType>,
    pub os_env_vars: Option<HashMap<String, String>>,
    pub session_root_directory: Option<PathBuf>,
    pub user: Option<Arc<dyn SessionUser>>,
    /// Revision + extensions profile that drives expression-function
    /// availability and redaction behaviour. Sessions do not use
    /// caller-policy limits, so a [`ModelProfile`](openjd_model::ModelProfile)
    /// is the right shape here — not a full
    /// [`ValidationContext`](openjd_model::ValidationContext).
    pub profile: Option<openjd_model::ModelProfile>,
    /// Optional external cancellation token. When cancelled, all running and
    /// future actions will be cancelled via the spec's cancellation sequence.
    pub cancel_token: Option<CancellationToken>,
    /// Controls behavior when a parent directory of the session root is
    /// world-writable without the sticky bit set (POSIX only).
    /// Defaults to `Strict` (fail-closed). Has no effect on Windows.
    pub sticky_bit_policy: crate::tempdir::StickyBitPolicy,
    /// Whether to accumulate subprocess stdout into result strings.
    /// Intended for debugging only — production callers should leave this
    /// `false` and observe output through the real-time callback instead.
    /// Default is `false` — output is still streamed through the callback in
    /// real time, but `ActionResult.stdout` and similar fields stay empty.
    pub debug_collect_stdout: bool,
    /// Whether to echo `openjd_*` directive lines (e.g. `openjd_progress`,
    /// `openjd_status`, `openjd_env`, `openjd_redacted_env`, …) from
    /// subprocess stdout to the session log.
    ///
    /// Default is `true`, matching the Python `openjd-sessions` reference
    /// implementation. When `false`, recognised directives are still parsed
    /// and acted on (progress, status, env-var changes, redacted-value
    /// registration, …) but the directive lines themselves are filtered
    /// out of the log stream.
    ///
    /// **Redaction interaction**: regardless of this flag, values from
    /// `openjd_redacted_env` directives are added to the session's redaction
    /// set *before* the originating line would be passed through, so when
    /// `echo_openjd_directives = true` the directive line that introduces a
    /// secret is still redacted (`NAME=********`) before reaching the log.
    /// Subsequent occurrences of the secret in any log line are also
    /// redacted. Setting this flag to `false` does not improve security —
    /// it just removes the directive lines from operator-facing output.
    pub echo_openjd_directives: bool,
}

fn format_exit_code(code: Option<i32>) -> String {
    match code {
        Some(c) => format!("exit code: {c}"),
        None => "exit code: N/A".to_string(),
    }
}

/// Normalize an environment variable name for the current platform.
/// On Windows, env vars are case-insensitive, so we uppercase all keys
/// to avoid undefined behavior from mixed-case duplicates in the Win32 API.
fn normalize_env_key(name: &str) -> String {
    #[cfg(windows)]
    {
        name.to_uppercase()
    }
    #[cfg(not(windows))]
    {
        name.to_string()
    }
}

/// Tracks env var changes made during an environment's lifecycle.
/// Uses a HashMap for automatic deduplication — last write wins,
/// matching Python's SimplifiedEnvironmentVariableChanges dict.
/// `None` value means "unset this variable".
type EnvVarChanges = HashMap<String, Option<String>>;

/// Tracks the status of the currently running (or most recently completed) action.
struct ActionStatusFields {
    state: Option<ActionState>,
    progress: Option<f64>,
    status_message: Option<String>,
    fail_message: Option<String>,
    exit_code: Option<i32>,
    started_at: Option<std::time::SystemTime>,
    ended_at: Option<std::time::SystemTime>,
}

impl ActionStatusFields {
    fn new() -> Self {
        Self {
            state: None,
            progress: None,
            status_message: None,
            fail_message: None,
            exit_code: None,
            started_at: None,
            ended_at: None,
        }
    }

    /// Reset all fields for a new action.
    fn reset(&mut self) {
        self.state = Some(ActionState::Running);
        self.started_at = Some(std::time::SystemTime::now());
        self.ended_at = None;
        self.progress = None;
        self.status_message = None;
        self.fail_message = None;
        self.exit_code = None;
    }
}

/// Cancellation state for the current action and external cancellation support.
struct CancelFields {
    /// Token for the current action (cancelled to abort the subprocess).
    token: Option<CancellationToken>,
    /// Channel to send cancel requests (with optional time limit) to the subprocess.
    request_tx: Option<tokio::sync::watch::Sender<Option<Duration>>>,
    /// When true, a Canceled action result is reported as Failed.
    mark_failed: bool,
    /// External cancellation token from the caller; action tokens are children of this.
    parent_token: Option<CancellationToken>,
}

impl CancelFields {
    fn new(parent_token: Option<CancellationToken>) -> Self {
        Self {
            token: None,
            request_tx: None,
            mark_failed: false,
            parent_token,
        }
    }
}

/// Cross-user execution state.
struct CrossUserFields {
    user: Option<Arc<dyn SessionUser>>,
    /// Process-user-only directory for helper binary and wrapper scripts.
    helpers_dir: Option<PathBuf>,
    #[cfg(unix)]
    helper: Option<CrossUserHelper>,
    #[cfg(windows)]
    helper: Option<CrossUserHelperWin>,
    cancel_writer: Option<std::fs::File>,
    /// Shared auth token for the helper process. `Some` whenever `helper` is
    /// `Some`; `None` for same-user sessions. Kept separate from the helper
    /// struct because the cancel path needs to access it even when the
    /// helper has been moved into a runner.
    helper_auth_token: Option<String>,
}

/// Build the session's function library from the session's
/// [`ModelProfile`](openjd_model::ModelProfile) (or the current default
/// expr profile when none is set), with the current path-mapping `rules`
/// registered as host context.
///
/// Called whenever either the profile or the rules change so that
/// `apply_path_mapping` always reflects the session's current rules.
fn derive_library(
    profile: Option<&openjd_model::ModelProfile>,
    rules: &Arc<Vec<PathMappingRule>>,
) -> Arc<FunctionLibrary> {
    let host = openjd_expr::HostContext::WithRules(rules.clone());
    let expr_profile = match profile {
        Some(p) => p.to_expr_profile(host),
        None => openjd_expr::ExprProfile::current().with_host_context(host),
    };
    openjd_expr::FunctionLibrary::for_profile(&expr_profile)
}

pub struct Session {
    session_id: String,
    state: SessionState,
    ending_only: bool,
    working_directory: PathBuf,
    files_directory: PathBuf,
    retain_working_dir: bool,
    cleanup_called: bool,
    // TempDir ownership (only when created via with_config)
    _working_dir: Option<crate::tempdir::TempDir>,
    _files_dir: Option<crate::tempdir::TempDir>,
    // Environment tracking
    environments: HashMap<EnvironmentIdentifier, Environment>,
    environments_entered: Vec<EnvironmentIdentifier>,
    // Env var tracking
    env_vars: HashMap<String, String>,
    process_env: HashMap<String, String>,
    created_env_vars: HashMap<EnvironmentIdentifier, EnvVarChanges>,
    // Expression evaluation
    //
    // `library` is the cached derived library, built from the session's
    // `library` is the cached derived library, built from the session's
    // `profile` (an optional `ModelProfile`) plus the current
    // `path_mapping_rules`. Whenever either input changes, the library
    // must be rebuilt via `derive_library`.
    library: Arc<FunctionLibrary>,
    path_mapping_rules: Arc<Vec<PathMappingRule>>,
    job_parameter_values: JobParameterValues,
    // Grouped fields
    action: ActionStatusFields,
    cancel: CancelFields,
    cross_user: CrossUserFields,
    // Callback
    callback: Option<SessionCallbackType>,
    // Redaction
    redacted_values: HashSet<String>,
    profile: Option<openjd_model::ModelProfile>,
    debug_collect_stdout: bool,
    /// Whether to echo `openjd_*` directive lines to the log. See
    /// [`SessionConfig::echo_openjd_directives`].
    echo_openjd_directives: bool,
}

impl Session {
    /// Test-only constructor that accepts a pre-existing working directory.
    /// Production code should use `Session::with_config`.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn new_for_test(working_directory: PathBuf) -> Self {
        let files_directory = working_directory.join("embedded_files");
        Self {
            session_id: String::new(),
            state: SessionState::Ready,
            ending_only: false,
            working_directory,
            files_directory,
            retain_working_dir: false,
            cleanup_called: false,
            _working_dir: None,
            _files_dir: None,
            environments: HashMap::new(),
            environments_entered: Vec::new(),
            env_vars: HashMap::new(),
            process_env: HashMap::new(),
            created_env_vars: HashMap::new(),
            library: derive_library(None, &Arc::new(Vec::new())),
            path_mapping_rules: Arc::new(Vec::new()),
            job_parameter_values: HashMap::new(),
            action: ActionStatusFields::new(),
            cancel: CancelFields::new(None),
            cross_user: CrossUserFields {
                user: None,
                helpers_dir: None,
                helper: None,
                cancel_writer: None,
                helper_auth_token: None,
            },
            callback: None,
            redacted_values: HashSet::new(),
            profile: None,
            debug_collect_stdout: true, // test constructor — tests need captured stdout
            echo_openjd_directives: true, // matches default in production config
        }
    }

    /// Test-only: inject a cancel_writer (e.g. the write end of a pipe) so a test
    /// can observe what `cancel_action` writes.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn set_cancel_writer_for_test(&mut self, writer: std::fs::File) {
        self.cross_user.cancel_writer = Some(writer);
    }

    /// Test-only: inject a `helper_auth_token` so `cancel_action` writes a
    /// tokenized cancel JSON object. Pairs with `set_cancel_writer_for_test`
    /// to exercise the token-in-cancel-JSON path without a real helper.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn set_helper_auth_token_for_test(&mut self, token: String) {
        self.cross_user.helper_auth_token = Some(token);
    }

    /// Test-only: force the session state (typically to `Running`) so `cancel_action`
    /// will proceed without needing a full action run.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn set_state_for_test(&mut self, state: SessionState) {
        self.state = state;
    }

    /// Full constructor from SessionConfig.
    pub fn with_config(mut config: SessionConfig) -> Result<Self, SessionError> {
        let root_dir = match &config.session_root_directory {
            Some(d) => d.clone(),
            None => crate::tempdir::openjd_temp_dir(None)?,
        };

        #[cfg(unix)]
        {
            use crate::tempdir::StickyBitPolicy;
            match config.sticky_bit_policy {
                StickyBitPolicy::Strict => {
                    if let Some(path) = crate::tempdir::find_missing_sticky_bit(&root_dir) {
                        return Err(SessionError::PathPermissions {
                            path: path.display().to_string(),
                            reason: format!(
                                "Directory is world-writable without the sticky bit set. \
                                 This allows other users to modify or delete session files. \
                                 Set the sticky bit (chmod +t {}) or use \
                                 StickyBitPolicy::Warn to override.",
                                path.display()
                            ),
                        });
                    }
                }
                StickyBitPolicy::Warn => {
                    if let Some(path) = crate::tempdir::find_missing_sticky_bit(&root_dir) {
                        log::warn!(
                            target: "openjd.sessions",
                            "Sticky bit is not set on {}. This may pose a risk when running \
                             work on this host as users may modify or delete files in this \
                             directory which do not belong to them.",
                            path.display()
                        );
                    }
                }
                StickyBitPolicy::Disabled => {}
            }
        }

        let working_dir = crate::tempdir::TempDir::new(
            Some(&root_dir),
            Some(&config.session_id),
            config.user.as_deref(),
        )?;
        let files_dir = crate::tempdir::TempDir::new(
            Some(working_dir.path()),
            Some("embedded_files"),
            config.user.as_deref(),
        )?;

        let working_directory = working_dir.path().to_path_buf();
        let files_directory = files_dir.path().to_path_buf();

        let mut path_mapping_rules = config.path_mapping_rules.unwrap_or_default();
        path_mapping_rules.sort_by_key(|r| std::cmp::Reverse(r.source_path.len()));
        let path_mapping_rules = Arc::new(path_mapping_rules);
        let profile = config.profile.take();
        let library = derive_library(profile.as_ref(), &path_mapping_rules);
        let process_env = config.os_env_vars.unwrap_or_default();

        // Create helpers directory and spawn cross-user helper if needed.
        // The helpers directory is 0o750 (owner rwx, group r-x) so the job user
        // can traverse and execute but cannot create or modify files.
        let mut helpers_dir = None;

        #[cfg(unix)]
        let (helper, cancel_writer, helper_auth_token) = if let Some(ref user) = config.user {
            if !user.is_process_user() {
                let hdir = crate::helper_binary::create_helpers_dir(
                    &working_directory,
                    Some(user.as_ref()),
                )?;
                let helper_path = crate::helper_binary::write_helper(&hdir, user.as_ref())?;
                let (h, cw) = CrossUserHelper::spawn(&helper_path, user.as_ref())?;
                let token = h.auth_token().to_string();
                helpers_dir = Some(hdir);
                (Some(h), Some(cw), Some(token))
            } else {
                (None, None, None)
            }
        } else {
            (None, None, None)
        };

        #[cfg(windows)]
        let (helper, cancel_writer, helper_auth_token) = if let Some(ref user) = config.user {
            if !user.is_process_user() {
                let hdir = crate::helper_binary::create_helpers_dir(
                    &working_directory,
                    Some(user.as_ref()),
                )?;
                let helper_path = crate::helper_binary::write_helper(&hdir, user.as_ref())?;
                let (h, cw) = CrossUserHelperWin::spawn(&helper_path, user.as_ref())?;
                let token = h.auth_token().to_string();
                helpers_dir = Some(hdir);
                (Some(h), Some(cw), Some(token))
            } else {
                (None, None, None)
            }
        } else {
            (None, None, None)
        };

        // Log host info (mirrors Python Session.__init__)
        session_log!(
            info,
            &config.session_id,
            LogContent::HOST_INFO,
            "openjd-sessions Library Version: {}",
            env!("CARGO_PKG_VERSION")
        );
        session_log!(
            info,
            &config.session_id,
            LogContent::HOST_INFO,
            "Platform: {}",
            std::env::consts::OS
        );
        session_log!(
            info,
            &config.session_id,
            LogContent::HOST_INFO,
            "Architecture: {}",
            std::env::consts::ARCH
        );
        // `session_id` is an opaque correlation identifier, not a secret —
        // it is emitted both as a structured field and in the message text so
        // log consumers can associate this line with the rest of the session.
        // See the module-level docs for rationale.
        log::info!(target: "openjd.sessions", session_id = config.session_id.as_str(); "Initializing Open Job Description Session: {}", &config.session_id);
        session_log!(
            info,
            &config.session_id,
            LogContent::FILE_PATH,
            "Session Working Directory: {}",
            working_directory.display()
        );
        session_log!(
            info,
            &config.session_id,
            LogContent::FILE_PATH,
            "Session's Embedded Files Directory: {}",
            files_directory.display()
        );

        Ok(Self {
            session_id: config.session_id,
            state: SessionState::Ready,
            ending_only: false,
            working_directory,
            files_directory,
            retain_working_dir: config.retain_working_dir,
            cleanup_called: false,
            _working_dir: Some(working_dir),
            _files_dir: Some(files_dir),
            environments: HashMap::new(),
            environments_entered: Vec::new(),
            env_vars: HashMap::new(),
            process_env,
            created_env_vars: HashMap::new(),
            library,
            path_mapping_rules,
            job_parameter_values: config.job_parameter_values,
            action: ActionStatusFields::new(),
            cancel: CancelFields::new(config.cancel_token),
            cross_user: CrossUserFields {
                user: config.user,
                helpers_dir,
                helper,
                cancel_writer,
                helper_auth_token,
            },
            callback: config.callback,
            redacted_values: HashSet::new(),
            profile,
            debug_collect_stdout: config.debug_collect_stdout,
            echo_openjd_directives: config.echo_openjd_directives,
        })
    }

    pub fn with_path_mapping(mut self, mut rules: Vec<PathMappingRule>) -> Self {
        rules.sort_by_key(|r| std::cmp::Reverse(r.source_path.len()));
        self.path_mapping_rules = Arc::new(rules);
        self.library = derive_library(self.profile.as_ref(), &self.path_mapping_rules);
        self
    }

    /// Extend the session's path mapping rules with additional rules.
    /// Rules are re-sorted by source path length (longest first) after extending.
    pub fn extend_path_mapping_rules(&mut self, additional: Vec<PathMappingRule>) {
        let mut rules = (*self.path_mapping_rules).clone();
        rules.extend(additional);
        rules.sort_by_key(|r| std::cmp::Reverse(r.source_path.len()));
        self.path_mapping_rules = Arc::new(rules);
        self.library = derive_library(self.profile.as_ref(), &self.path_mapping_rules);
    }

    /// Get the current path mapping rules.
    pub fn path_mapping_rules(&self) -> &[PathMappingRule] {
        &self.path_mapping_rules
    }

    /// Set the [`ModelProfile`](openjd_model::ModelProfile) that drives
    /// which expression functions and signatures are available to this
    /// session and which redaction rules apply. Rebuilds the session's
    /// derived function library so subsequent expression evaluation
    /// sees the new profile.
    pub fn with_profile(mut self, profile: openjd_model::ModelProfile) -> Self {
        self.profile = Some(profile);
        self.library = derive_library(self.profile.as_ref(), &self.path_mapping_rules);
        self
    }

    /// Check whether redacted env vars are enabled, mirroring Python's `_redactions_enabled()`.
    /// True if spec revision > v2023_09 OR "REDACTED_ENV_VARS" extension is present.
    fn redactions_enabled(&self) -> bool {
        match &self.profile {
            Some(p) => {
                p.revision() > openjd_model::types::SpecificationRevision::V2023_09
                    || p.has_extension(openjd_model::types::ModelExtension::RedactedEnvVars)
            }
            None => false,
        }
    }

    fn lib(&self) -> Option<&FunctionLibrary> {
        Some(&self.library)
    }

    /// Fire the callback with the current action status.
    /// Called when transitioning to RUNNING so callers see the state change immediately.
    fn notify_callback(&self) {
        if let Some(cb) = &self.callback {
            if let Some(status) = self.action_status() {
                cb(&self.session_id, &status);
            }
        }
    }

    // --- Properties ---

    /// Returns the list of enabled extensions for this session.
    /// If no profile is set, returns an empty vec.
    pub fn get_enabled_extensions(&self) -> Vec<String> {
        match &self.profile {
            Some(p) => {
                let mut exts: Vec<String> = p
                    .extensions()
                    .iter()
                    .map(|e| e.as_str().to_string())
                    .collect();
                exts.sort();
                exts
            }
            None => Vec::new(),
        }
    }

    pub fn session_id(&self) -> &str {
        &self.session_id
    }
    pub fn state(&self) -> SessionState {
        self.state
    }
    pub fn working_directory(&self) -> &Path {
        &self.working_directory
    }
    pub fn files_directory(&self) -> &Path {
        &self.files_directory
    }
    pub fn environments_entered(&self) -> &[EnvironmentIdentifier] {
        &self.environments_entered
    }

    /// Clone the cancel_writer file handle, if one exists.
    /// Used by Python bindings to send cancel commands when the session is
    /// taken by a background thread.
    pub fn clone_cancel_writer(&self) -> Option<std::fs::File> {
        self.cross_user
            .cancel_writer
            .as_ref()
            .and_then(|f| f.try_clone().ok())
    }

    /// Get the current action status, if any action has been run.
    pub fn action_status(&self) -> Option<ActionStatus> {
        self.action.state.map(|state| ActionStatus {
            state,
            progress: self.action.progress,
            status_message: self.action.status_message.clone(),
            fail_message: self.action.fail_message.clone(),
            exit_code: self.action.exit_code,
            started_at: self.action.started_at,
            ended_at: self.action.ended_at,
        })
    }

    /// Override the action state. Used by the pyo3 binding to convert
    /// Failed → Canceled when cancel was requested externally.
    pub fn override_action_state(&mut self, state: ActionState) {
        self.action.state = Some(state);
    }

    /// Redact sensitive values from output text.
    pub fn redact(&self, text: &str) -> String {
        // Sort by length descending so longer matches are replaced first,
        // preventing partial replacements when values overlap (e.g. "FOOBAR" and "BAR").
        let mut vals: Vec<&str> = self.redacted_values.iter().map(|s| s.as_str()).collect();
        vals.sort_by_key(|s| std::cmp::Reverse(s.len()));
        let mut result = text.to_string();
        for val in vals {
            result = result.replace(val, "********");
        }
        result
    }

    /// Create a cancellation token for an action. If a parent token was
    /// provided at session construction, the action token is a child of it
    /// so that cancelling the parent cascades to all actions.
    fn new_action_cancel_token(&self) -> CancellationToken {
        match &self.cancel.parent_token {
            Some(parent) => parent.child_token(),
            None => CancellationToken::new(),
        }
    }

    /// Cancel the currently running async action.
    pub fn cancel_action(
        &mut self,
        time_limit: Option<Duration>,
        mark_action_failed: bool,
    ) -> Result<(), SessionError> {
        if self.state != SessionState::Running {
            return Err(SessionError::InvalidState {
                expected: vec![SessionState::Running],
                current: self.state,
            });
        }
        self.state = SessionState::Canceling;
        if mark_action_failed {
            self.cancel.mark_failed = true;
        }

        // Send cancel to the helper process via the dup'd stdin fd.
        //
        // Same-user sessions have no helper, so cancel_writer is None — in
        // those sessions there's nothing to write. When a cancel_writer is
        // present but no helper token is stored (e.g. an in-test injection
        // via `set_cancel_writer_for_test`), the test is responsible for
        // asserting whatever framing it expects; we still write a valid
        // tokenized command if we have a token.
        if let Some(ref mut writer) = self.cross_user.cancel_writer {
            use std::io::Write;
            let is_terminate = matches!(time_limit, Some(d) if d.is_zero());
            let token_field = match &self.cross_user.helper_auth_token {
                Some(t) => format!(r#""token":"{t}","#),
                None => String::new(),
            };
            let cmd = if is_terminate {
                format!(r#"{{{token_field}"cancel":"TERMINATE"}}"#)
            } else {
                let notify_period = time_limit
                    .unwrap_or(Duration::from_secs(DEFAULT_CANCEL_NOTIFY_PERIOD_SECS))
                    .as_secs();
                format!(
                    r#"{{{token_field}"cancel":"NOTIFY_THEN_TERMINATE","notifyPeriodInSeconds":{notify_period}}}"#
                )
            };
            let _ = writer.write_all(cmd.as_bytes());
            let _ = writer.write_all(b"\n");
            let _ = writer.flush();
        }

        if let Some(tx) = &self.cancel.request_tx {
            let _ = tx.send(time_limit);
        }
        if let Some(token) = &self.cancel.token {
            token.cancel();
        }
        Ok(())
    }

    /// Clean up the session. Deletes working directory if not retained.
    pub fn cleanup(&mut self) {
        if self.cleanup_called {
            return;
        }
        self.cleanup_called = true;

        if !self.retain_working_dir {
            log_section_banner(&self.session_id, "Session Cleanup");

            // Shut down the cross-user helper before deleting the working directory
            if let Some(ref mut helper) = self.cross_user.helper {
                helper.shutdown();
            }

            session_log!(
                info,
                &self.session_id,
                LogContent::FILE_PATH,
                "Deleting working directory: {}",
                self.working_directory.display()
            );

            // Cross-user cleanup: delete files owned by session user first,
            // since files created via sudo may not be deletable by the process owner.
            #[cfg(unix)]
            if let Some(ref user) = self.cross_user.user {
                if !user.is_process_user() {
                    if let Ok(entries) = std::fs::read_dir(&self.working_directory) {
                        let files: Vec<String> = entries
                            .filter_map(|e| e.ok())
                            .map(|e| e.path().to_string_lossy().to_string())
                            .collect();
                        if !files.is_empty() {
                            let mut args = vec![
                                "-u".to_string(),
                                user.user().to_string(),
                                "-i".to_string(),
                                "rm".to_string(),
                                "-rf".to_string(),
                                "--".to_string(),
                            ];
                            args.extend(files);
                            let _ = std::process::Command::new("sudo")
                                .args(&args)
                                .stdin(std::process::Stdio::null())
                                .stdout(std::process::Stdio::null())
                                .stderr(std::process::Stdio::null())
                                .status();
                        }
                    }
                }
            }

            // If we own TempDirs, clean them up; otherwise fall back to remove_dir_all
            if let Some(ref mut files_dir) = self._files_dir {
                let _ = files_dir.cleanup();
            }
            if let Some(ref mut working_dir) = self._working_dir {
                let _ = working_dir.cleanup();
            } else {
                let _ = std::fs::remove_dir_all(&self.working_directory);
            }
        }
        self.state = SessionState::Ended;
    }

    /// Enter an environment asynchronously (non-blocking).
    /// Returns the environment identifier on success.
    pub async fn enter_environment(
        &mut self,
        env: &Environment,
        resolved_symtab: Option<&openjd_expr::SerializedSymbolTable>,
        identifier: Option<&str>,
        os_env_vars: Option<&HashMap<String, String>>,
    ) -> Result<String, SessionError> {
        let (id, _stdout) = self
            .enter_environment_with_output(env, resolved_symtab, identifier, os_env_vars)
            .await?;
        Ok(id)
    }

    /// Enter an environment, returning both the identifier and the stdout from the onEnter script.
    pub async fn enter_environment_with_output(
        &mut self,
        env: &Environment,
        resolved_symtab: Option<&openjd_expr::SerializedSymbolTable>,
        identifier: Option<&str>,
        os_env_vars: Option<&HashMap<String, String>>,
    ) -> Result<(String, String), SessionError> {
        if self.state != SessionState::Ready {
            return Err(SessionError::InvalidState {
                expected: vec![SessionState::Ready],
                current: self.state,
            });
        }

        let symtab = self.build_symbol_table(None, resolved_symtab)?;

        let identifier = match identifier {
            Some(id) => {
                if self.environments.contains_key(id) {
                    return Err(SessionError::DuplicateEnvironment { id: id.to_string() });
                }
                id.to_string()
            }
            None => format!("{}:{}", self.session_id, uuid::Uuid::new_v4().simple()),
        };
        self.environments.insert(identifier.clone(), env.clone());
        self.environments_entered.push(identifier.clone());
        self.created_env_vars
            .insert(identifier.clone(), HashMap::new());

        // Set static variables
        if let Some(vars) = &env.variables {
            for (key, fmt_str) in vars {
                let value = fmt_str
                    .resolve_string_with(
                        &symtab,
                        &openjd_expr::FormatStringOptions::new().with_library(self.lib()),
                    )
                    .map_err(|e| SessionError::FormatString {
                        context: format!("env var '{key}'"),
                        reason: e.to_string(),
                    })?;
                let norm_key = normalize_env_key(key);
                self.env_vars.insert(norm_key.clone(), value.clone());
                if let Some(changes) = self.created_env_vars.get_mut(&identifier) {
                    changes.insert(norm_key, Some(value));
                }
            }
        }

        let output = if env
            .script
            .as_ref()
            .and_then(|s| s.actions.on_enter.as_ref())
            .is_some()
        {
            self.action.reset();
            self.state = SessionState::Running;
            self.notify_callback();

            log_section_banner(
                &self.session_id,
                &format!("Entering Environment: {}", env.name),
            );

            let cancel_token = self.new_action_cancel_token();
            let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(None);
            self.cancel.token = Some(cancel_token.clone());
            self.cancel.request_tx = Some(cancel_tx);

            let env_vars = self.evaluate_env_vars(os_env_vars);
            let mut action_symtab = symtab.clone();
            self.materialize_path_mapping(&mut action_symtab)?;

            // RFC 0008: if an outer active wrap env (not including this
            // one — which isn't on the stack yet but the helper guards
            // against it anyway) defines onWrapEnvEnter, substitute that
            // action and seed WrappedAction.* / WrappedEnv.* with the
            // inner onEnter's resolved command/args/timeout.
            let inner_on_enter = env
                .script
                .as_ref()
                .and_then(|s| s.actions.on_enter.as_ref())
                .expect("outer branch guard");
            let wrap_action = self.wrap_env_excluding(&identifier).and_then(|outer| {
                outer
                    .script
                    .as_ref()
                    .and_then(|s| s.actions.on_wrap_env_enter.as_ref())
                    .cloned()
                    .map(|action| (outer.resolved_symtab.clone(), action))
            });

            let lib = self.library.clone();
            if let Some((wrap_symtab, _)) = wrap_action.as_ref() {
                seed_wrapped_action_symbols(
                    &mut action_symtab,
                    wrap_symtab,
                    inner_on_enter,
                    WrappedContext::Env(&env.name),
                    &self.env_vars,
                    Some(&lib),
                    "onEnter",
                )?;
            }

            // Box large locals — see run_task for rationale.
            let action_symtab = Box::new(action_symtab);
            let env_vars = Box::new(env_vars);
            #[allow(unused_mut)]
            let mut runner = EnvironmentScriptRunner::new(
                &self.session_id,
                self.working_directory.clone(),
                self.files_directory.clone(),
                self.cross_user.user.clone(),
            )
            .with_redactions(self.redactions_enabled())
            .with_debug_collect_stdout(self.debug_collect_stdout)
            .with_echo_openjd_directives(self.echo_openjd_directives)
            .with_initial_redacted_values(self.redacted_values.iter().cloned().collect())
            .with_cancel_token(cancel_token)
            .with_cancel_request_rx(cancel_rx);
            if let Some(ref hdir) = self.cross_user.helpers_dir {
                runner = runner.with_helpers_directory(hdir.clone());
            }
            let mut runner = match self.cross_user.helper.take() {
                Some(h) => {
                    let r = runner.with_helper(h);
                    match self
                        .cross_user
                        .cancel_writer
                        .as_ref()
                        .and_then(|f| f.try_clone().ok())
                    {
                        Some(w) => r.with_cancel_writer(w),
                        None => r,
                    }
                }
                None => runner,
            };

            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

            // Box::pin keeps the inner subprocess/select! state machine off the
            // outer future's stack. Without this, the combined future exceeds
            // Windows' default 1 MB thread stack in release builds. The boxed
            // type unifies the two match arms; it is intentionally NOT `+ Send`
            // because on Windows the subprocess future holds a non-Send
            // `Option<HANDLE>` across an await (and `drive_action` does not
            // require `Send`), matching the plain `enter`/`exit`/`run` paths.
            let runner_fut: std::pin::Pin<Box<dyn std::future::Future<Output = _>>> =
                match wrap_action.as_ref() {
                    Some((_, action)) => Box::pin(runner.run_wrap_action(
                        action,
                        &action_symtab,
                        Some(&lib),
                        &env_vars,
                        tx,
                        None,
                    )),
                    None => Box::pin(runner.enter(env, &action_symtab, Some(&lib), &env_vars, tx)),
                };
            let result = self.drive_action(runner_fut, &mut rx, &identifier).await;
            self.cross_user.helper = runner.take_helper();
            let result = result?;

            if result.state != ActionState::Success {
                return Err(SessionError::EnvironmentScriptFailed {
                    name: env.name.clone(),
                    action: "onEnter".into(),
                    reason: format_exit_code(result.exit_code),
                });
            }
            result.stdout.clone()
        } else {
            let now = std::time::SystemTime::now();
            self.action.state = Some(ActionState::Success);
            self.action.started_at = Some(now);
            self.action.progress = None;
            self.action.status_message = None;
            self.action.fail_message = None;
            self.action.exit_code = None;

            log_section_banner(
                &self.session_id,
                &format!("Entering Environment: {}", env.name),
            );

            self.action.ended_at = Some(std::time::SystemTime::now());

            if let Some(cb) = &self.callback {
                if let Some(status) = self.action_status() {
                    cb(&self.session_id, &status);
                }
            }

            String::new()
        };

        if self.state == SessionState::Running {
            self.state = SessionState::Ready;
        }
        Ok((identifier, output))
    }

    /// Exit an environment asynchronously, identified by its environment identifier.
    ///
    /// Environments must be exited in LIFO order (last entered, first exited).
    /// The `keep_session_running` parameter controls whether the session transitions
    /// to `ReadyEnding` after exit — if `false`, the session becomes ending-only.
    pub async fn exit_environment(
        &mut self,
        identifier: &EnvironmentIdentifier,
        resolved_symtab: Option<&openjd_expr::SerializedSymbolTable>,
        keep_session_running: bool,
        os_env_vars: Option<&HashMap<String, String>>,
    ) -> Result<String, SessionError> {
        if self.state != SessionState::Ready && self.state != SessionState::ReadyEnding {
            return Err(SessionError::InvalidState {
                expected: vec![SessionState::Ready, SessionState::ReadyEnding],
                current: self.state,
            });
        }

        // Validate identifier exists
        let env = self
            .environments
            .get(identifier)
            .ok_or_else(|| SessionError::UnknownEnvironment {
                identifier: identifier.to_string(),
            })?
            .clone();

        // Validate LIFO order
        if self.environments_entered.last() != Some(identifier) {
            return Err(SessionError::LifoViolation {
                expected: self
                    .environments_entered
                    .last()
                    .cloned()
                    .unwrap_or_default(),
                got: identifier.clone(),
            });
        }

        let symtab = self.build_symbol_table(None, resolved_symtab)?;

        // Evaluate env vars BEFORE removing from tracking (matching Python)
        // Box to keep off the async state machine — see run_task for rationale.
        let env_vars = Box::new(self.evaluate_env_vars(os_env_vars));

        // Unless overridden by the caller, once we've started exiting environments
        // we can only exit environments.
        if !keep_session_running {
            self.ending_only = true;
        }

        // Remove environment from tracking BEFORE running the exit script.
        // This matches the Python session behavior — a failed exit is still an exit,
        // and subsequent exits must be able to proceed in LIFO order.
        self.environments.remove(identifier);
        self.environments_entered.pop();

        let output = if env
            .script
            .as_ref()
            .and_then(|s| s.actions.on_exit.as_ref())
            .is_some()
        {
            self.action.reset();
            self.state = SessionState::Running;
            self.notify_callback();

            log_section_banner(
                &self.session_id,
                &format!("Exiting Environment: {}", env.name),
            );

            let cancel_token = self.new_action_cancel_token();
            let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(None);
            self.cancel.token = Some(cancel_token.clone());
            self.cancel.request_tx = Some(cancel_tx);

            let mut action_symtab = symtab.clone();
            self.materialize_path_mapping(&mut action_symtab)?;

            // RFC 0008: does an outer active wrap env wrap this onExit?
            // The env being exited has already been popped from the stack
            // by this point, so `active_wrap_env` correctly returns only
            // the remaining envs — which is what we want.
            let inner_on_exit = env
                .script
                .as_ref()
                .and_then(|s| s.actions.on_exit.as_ref())
                .expect("outer branch guard");
            let wrap_action = self.active_wrap_env().and_then(|outer| {
                outer
                    .script
                    .as_ref()
                    .and_then(|s| s.actions.on_wrap_env_exit.as_ref())
                    .cloned()
                    .map(|action| (outer.resolved_symtab.clone(), action))
            });

            let lib = self.library.clone();
            if let Some((wrap_symtab, _)) = wrap_action.as_ref() {
                seed_wrapped_action_symbols(
                    &mut action_symtab,
                    wrap_symtab,
                    inner_on_exit,
                    WrappedContext::Env(&env.name),
                    &self.env_vars,
                    Some(&lib),
                    "onExit",
                )?;
            }

            // Box large locals — see run_task for rationale.
            let action_symtab = Box::new(action_symtab);
            #[allow(unused_mut)]
            let mut runner = EnvironmentScriptRunner::new(
                &self.session_id,
                self.working_directory.clone(),
                self.files_directory.clone(),
                self.cross_user.user.clone(),
            )
            .with_redactions(self.redactions_enabled())
            .with_debug_collect_stdout(self.debug_collect_stdout)
            .with_echo_openjd_directives(self.echo_openjd_directives)
            .with_initial_redacted_values(self.redacted_values.iter().cloned().collect())
            .with_cancel_token(cancel_token)
            .with_cancel_request_rx(cancel_rx);
            if let Some(ref hdir) = self.cross_user.helpers_dir {
                runner = runner.with_helpers_directory(hdir.clone());
            }
            let mut runner = match self.cross_user.helper.take() {
                Some(h) => {
                    let r = runner.with_helper(h);
                    match self
                        .cross_user
                        .cancel_writer
                        .as_ref()
                        .and_then(|f| f.try_clone().ok())
                    {
                        Some(w) => r.with_cancel_writer(w),
                        None => r,
                    }
                }
                None => runner,
            };

            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

            // See the note in the onEnter path about Box::pin and the Windows
            // 1 MB thread-stack limit on release builds. As there, the boxed
            // type is intentionally NOT `+ Send` (the subprocess future holds a
            // non-Send `Option<HANDLE>` across an await on Windows).
            let runner_fut: std::pin::Pin<Box<dyn std::future::Future<Output = _>>> =
                match wrap_action.as_ref() {
                    Some((_, action)) => Box::pin(runner.run_wrap_action(
                        action,
                        &action_symtab,
                        Some(&lib),
                        &env_vars,
                        tx,
                        None,
                    )),
                    None => Box::pin(runner.exit(&env, &action_symtab, Some(&lib), &env_vars, tx)),
                };
            let result = self.drive_action(runner_fut, &mut rx, identifier).await;
            self.cross_user.helper = runner.take_helper();
            let result = result?;

            if result.state != ActionState::Success {
                self.state = SessionState::ReadyEnding;
                return Err(SessionError::EnvironmentScriptFailed {
                    name: env.name.clone(),
                    action: "onExit".into(),
                    reason: format_exit_code(result.exit_code),
                });
            }
            result.stdout.clone()
        } else {
            // No exit script — set state based on ending_only (drive_action not called)
            let now = std::time::SystemTime::now();
            self.action.state = Some(ActionState::Success);
            self.action.started_at = Some(now);
            self.action.progress = None;
            self.action.status_message = None;
            self.action.fail_message = None;
            self.action.exit_code = None;
            self.state = if self.ending_only {
                SessionState::ReadyEnding
            } else {
                SessionState::Ready
            };

            log_section_banner(
                &self.session_id,
                &format!("Exiting Environment: {}", env.name),
            );

            self.action.ended_at = Some(std::time::SystemTime::now());

            if let Some(cb) = &self.callback {
                if let Some(status) = self.action_status() {
                    cb(&self.session_id, &status);
                }
            }

            String::new()
        };

        Ok(output)
    }

    /// Run a step action asynchronously.
    ///
    /// `step_name` is the name of the step whose task is being run; it is
    /// surfaced as `WrappedStep.Name` to a wrapping environment's
    /// `onWrapTaskRun` hook (RFC 0008).
    pub async fn run_task(
        &mut self,
        step_name: &str,
        script: &StepScript,
        task_parameter_values: Option<&openjd_model::types::TaskParameterSet>,
        resolved_symtab: Option<&openjd_expr::SerializedSymbolTable>,
        os_env_vars: Option<&HashMap<String, String>>,
    ) -> Result<ActionResult, SessionError> {
        if self.state != SessionState::Ready {
            return Err(SessionError::InvalidState {
                expected: vec![SessionState::Ready],
                current: self.state,
            });
        }

        let symtab = self.build_symbol_table(task_parameter_values, resolved_symtab)?;

        self.action.reset();
        self.state = SessionState::Running;
        self.notify_callback();

        log_section_banner(&self.session_id, "Running Task");

        let cancel_token = self.new_action_cancel_token();
        let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(None);
        self.cancel.token = Some(cancel_token.clone());
        self.cancel.request_tx = Some(cancel_tx);

        let env_vars = self.evaluate_env_vars(os_env_vars);
        let mut action_symtab = symtab.clone();
        self.materialize_path_mapping(&mut action_symtab)?;

        // RFC 0008: decide whether this task's onRun should be wrapped by
        // an active environment's onWrapTaskRun. The decision is:
        //   - If an active wrap env defines onWrapTaskRun, we substitute
        //     the wrap action and seed WrappedAction.* into the symbol
        //     table so the wrap script can forward the original command
        //     and args.
        //
        // If neither condition routes us into the wrap path, the original
        // step script runs exactly as before — this keeps the non-WRAP_ACTIONS
        // path a zero-cost addition.
        //
        // Scope note: this pass does NOT re-materialize the wrap environment's
        // embedded files. Wrap actions that reference `{{Env.File.*}}` will
        // see only the names registered when the wrap env was entered, which
        // are not persisted across action runs. Inline wrap scripts
        // (`command: bash, args: ["-c", "..."]`) work without this. Re-running
        // `allocate_file_paths` against the wrap env's embedded_files at task
        // dispatch time is the follow-up to enable `Env.File.*` inside wrap
        // hooks end-to-end.
        let lib = self.library.clone();
        let wrap_action: Option<openjd_model::job::Action> = self
            .active_wrap_env()
            .and_then(|wrap_env| {
                wrap_env
                    .script
                    .as_ref()
                    .and_then(|s| s.actions.on_wrap_task_run.clone())
                    .map(|action| (wrap_env.resolved_symtab.clone(), action))
            })
            .map(|(wrap_symtab, action)| {
                // Seed WrappedAction.* / WrappedStep.Name from the step's own
                // onRun, layering the wrap env's frozen symtab (its Param.*,
                // let bindings) on top of the task symtab. Shared with the
                // onEnter/onExit hooks so all three behave identically.
                seed_wrapped_action_symbols(
                    &mut action_symtab,
                    &wrap_symtab,
                    &script.actions.on_run,
                    WrappedContext::Step(step_name),
                    &self.env_vars,
                    Some(&lib),
                    "task",
                )?;
                Ok::<_, SessionError>(action)
            })
            .transpose()?;

        // Box large locals so they live on the heap instead of inflating
        // this async fn's state machine. Without this, the combined future
        // (run_task → drive_action → select!) exceeds Windows' default
        // 1 MB thread stack in release builds.
        let action_symtab = Box::new(action_symtab);
        let env_vars = Box::new(env_vars);
        #[allow(unused_mut)]
        let mut runner = StepScriptRunner::new(
            &self.session_id,
            self.working_directory.clone(),
            self.files_directory.clone(),
            self.cross_user.user.clone(),
        )
        .with_redactions(self.redactions_enabled())
        .with_debug_collect_stdout(self.debug_collect_stdout)
        .with_echo_openjd_directives(self.echo_openjd_directives)
        .with_initial_redacted_values(self.redacted_values.iter().cloned().collect())
        .with_cancel_token(cancel_token)
        .with_cancel_request_rx(cancel_rx);
        if let Some(ref hdir) = self.cross_user.helpers_dir {
            runner = runner.with_helpers_directory(hdir.clone());
        }
        let mut runner = match self.cross_user.helper.take() {
            Some(h) => {
                let r = runner.with_helper(h);
                match self
                    .cross_user
                    .cancel_writer
                    .as_ref()
                    .and_then(|f| f.try_clone().ok())
                {
                    Some(w) => r.with_cancel_writer(w),
                    None => r,
                }
            }
            None => runner,
        };

        let step_identifier = format!("{}:step:{}", self.session_id, uuid::Uuid::new_v4().simple());
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        // Build the script the runner actually executes. When a wrap hook
        // is in play, the step's own onRun is replaced by the wrap action;
        // step let_bindings and embedded_files still run on the host side
        // because they belong to the step scope, not the task subprocess.
        let effective_script: std::borrow::Cow<'_, StepScript> = match wrap_action {
            Some(action) => std::borrow::Cow::Owned(StepScript {
                let_bindings: script.let_bindings.clone(),
                actions: openjd_model::job::StepActions { on_run: action },
                embedded_files: script.embedded_files.clone(),
            }),
            None => std::borrow::Cow::Borrowed(script),
        };

        // See the note in the onEnter path about Box::pin and the Windows
        // 1 MB thread-stack limit on release builds.
        let runner_fut = Box::pin(runner.run(
            effective_script.as_ref(),
            &action_symtab,
            Some(&lib),
            &env_vars,
            tx,
        ));
        let result = self
            .drive_action(runner_fut, &mut rx, &step_identifier)
            .await;
        self.cross_user.helper = runner.take_helper();
        let result = result?;

        Ok(ActionResult {
            state: result.state,
            exit_code: result.exit_code,
            stdout: result.stdout,
        })
    }

    /// Run an ad-hoc subprocess within the Session.
    ///
    /// Unlike `run_task`, this runs a raw command without format string
    /// evaluation, embedded file materialization, or path mapping. Used by the
    /// worker agent for install/sync operations.
    pub async fn run_subprocess(
        &mut self,
        command: &str,
        args: Option<&[String]>,
        timeout: Option<Duration>,
        os_env_vars: Option<&HashMap<String, String>>,
        use_session_env_vars: bool,
        log_banner_message: Option<&str>,
    ) -> Result<crate::subprocess::SubprocessResult, SessionError> {
        if self.state != SessionState::Ready {
            return Err(SessionError::InvalidState {
                expected: vec![SessionState::Ready],
                current: self.state,
            });
        }
        if command.is_empty() || command.trim().is_empty() {
            return Err(SessionError::Runtime(
                "command must be a non-empty string".into(),
            ));
        }
        if let Some(t) = timeout {
            if t.is_zero() {
                return Err(SessionError::Runtime("timeout must be positive".into()));
            }
        }

        if let Some(msg) = log_banner_message {
            log_section_banner(&self.session_id, msg);
        }

        self.action.reset();
        self.state = SessionState::Running;
        self.notify_callback();

        let env_vars = if use_session_env_vars {
            self.evaluate_env_vars(os_env_vars)
        } else {
            let mut result: HashMap<String, Option<String>> = self
                .process_env
                .iter()
                .map(|(k, v)| (k.clone(), Some(v.clone())))
                .collect();
            if let Some(extra) = os_env_vars {
                for (k, v) in extra {
                    result.insert(k.clone(), Some(v.clone()));
                }
            }
            result
        };

        let mut cmd_args = vec![command.to_string()];
        if let Some(a) = args {
            cmd_args.extend(a.iter().cloned());
        }

        // Route through the persistent helper process when cross-user helper is active.
        if self.cross_user.helper.is_some() {
            return self
                .run_subprocess_via_helper(&cmd_args, env_vars, timeout)
                .await;
        }

        let cancel_token = self.new_action_cancel_token();
        let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(None);
        self.cancel.token = Some(cancel_token.clone());
        self.cancel.request_tx = Some(cancel_tx);

        let config = crate::subprocess::SubprocessConfig {
            args: cmd_args,
            env_vars,
            working_dir: Some(self.working_directory.clone()),
            timeout,
            user: self.cross_user.user.clone(),
            cancel_method: crate::runner::CancelMethod::Terminate,
            cancel_request_rx: Some(cancel_rx),
            debug_collect_stdout: self.debug_collect_stdout,
        };
        let mut filter = crate::action_filter::ActionFilter::new(
            &self.session_id,
            self.echo_openjd_directives,
            false,
        );
        let subprocess_identifier = format!(
            "{}:subprocess:{}",
            self.session_id,
            uuid::Uuid::new_v4().simple()
        );
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let sid = self.session_id.clone();
        // See the note in the onEnter path about Box::pin and the Windows
        // 1 MB thread-stack limit on release builds.
        let runner_fut = Box::pin(crate::subprocess::run_subprocess(
            config,
            &mut filter,
            &sid,
            tx,
            cancel_token,
        ));
        self.drive_action(runner_fut, &mut rx, &subprocess_identifier)
            .await
    }

    /// Execute a subprocess via the persistent cross-user helper process.
    ///
    /// Instead of spawning `sudo` (POSIX) or `CreateProcessAsUserW` (Windows)
    /// per action, sends a run command over the helper's stdin and reads
    /// streamed responses from its stdout.
    async fn run_subprocess_via_helper(
        &mut self,
        args: &[String],
        env_vars: std::collections::HashMap<String, Option<String>>,
        _timeout: Option<Duration>,
    ) -> Result<crate::subprocess::SubprocessResult, SessionError> {
        let mut filter = crate::action_filter::ActionFilter::new(
            &self.session_id,
            self.echo_openjd_directives,
            false,
        );
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let subprocess_identifier = format!(
            "{}:subprocess:{}",
            self.session_id,
            uuid::Uuid::new_v4().simple()
        );

        let config = crate::subprocess::SubprocessConfig {
            args: args.to_vec(),
            env_vars,
            working_dir: Some(self.working_directory.clone()),
            timeout: _timeout,
            user: self.cross_user.user.clone(),
            cancel_method: crate::runner::CancelMethod::Terminate,
            cancel_request_rx: None,
            debug_collect_stdout: self.debug_collect_stdout,
        };

        let helper = self
            .cross_user
            .helper
            .as_mut()
            .expect("caller checked helper.is_some()");
        let result = run_via_helper(
            helper,
            &config,
            &mut filter,
            &self.session_id,
            tx,
            self.cross_user.cancel_writer.as_ref(),
        )
        .await;

        // Process any remaining messages.
        while let Ok(msg) = rx.try_recv() {
            self.apply_message(msg, &subprocess_identifier);
        }

        let r = result?;
        self.action.state = Some(r.state);
        self.action.ended_at = Some(std::time::SystemTime::now());
        self.action.exit_code = r.exit_code;
        self.cancel.token = None;
        self.cancel.request_tx = None;
        self.cancel.mark_failed = false;
        self.state = if r.state == ActionState::Success {
            SessionState::Ready
        } else {
            SessionState::ReadyEnding
        };
        self.notify_callback();
        Ok(r)
    }

    // --- Internal helpers ---

    /// Run an action future while concurrently processing messages from the channel
    /// in real-time. This ensures callbacks fire as stdout lines are parsed, not
    /// after the action completes.
    async fn drive_action<F>(
        &mut self,
        action_fut: F,
        rx: &mut tokio::sync::mpsc::UnboundedReceiver<ActionMessage>,
        identifier: &str,
    ) -> Result<crate::subprocess::SubprocessResult, SessionError>
    where
        F: std::future::Future<Output = Result<crate::subprocess::SubprocessResult, SessionError>>,
    {
        tokio::pin!(action_fut);
        let mut result = None;

        loop {
            tokio::select! {
                biased;
                msg = rx.recv(), if result.is_none() => {
                    // Channel returns None when closed; runner will finish soon.
                    if let Some(msg) = msg { self.apply_message(msg, identifier) }
                }
                r = &mut action_fut, if result.is_none() => {
                    result = Some(r);
                }
                else => break,
            }
            if result.is_some() {
                // Drain any remaining messages after the runner completes
                while let Ok(msg) = rx.try_recv() {
                    self.apply_message(msg, identifier);
                }
                break;
            }
        }

        let r = match result.expect("loop guarantees result is Some") {
            Ok(r) => r,
            Err(e) => {
                // The subprocess failed to start or the runner encountered an error.
                // Update session state so callers see Failed instead of stuck Running.
                self.action.state = Some(ActionState::Failed);
                self.action.ended_at = Some(std::time::SystemTime::now());
                self.action.exit_code = None;
                self.cancel.token = None;
                self.cancel.request_tx = None;
                self.cancel.mark_failed = false;
                self.state = SessionState::ReadyEnding;

                if let Some(cb) = &self.callback {
                    if let Some(status) = self.action_status() {
                        cb(&self.session_id, &status);
                    }
                }

                return Err(e);
            }
        };

        // If the action was canceled but mark_action_failed is set,
        // report it as Failed instead of Canceled (matches Python behavior)
        let final_state = if self.cancel.mark_failed && r.state == ActionState::Canceled {
            ActionState::Failed
        } else {
            r.state
        };

        self.action.state = Some(final_state);
        self.action.ended_at = Some(std::time::SystemTime::now());
        self.action.exit_code = r.exit_code;
        self.cancel.token = None;
        self.cancel.request_tx = None;
        self.cancel.mark_failed = false;

        self.state = if self.ending_only || final_state != ActionState::Success {
            SessionState::ReadyEnding
        } else {
            SessionState::Ready
        };

        if let Some(cb) = &self.callback {
            if let Some(status) = self.action_status() {
                cb(&self.session_id, &status);
            }
        }

        Ok(crate::subprocess::SubprocessResult {
            state: final_state,
            exit_code: r.exit_code,
            stdout: r.stdout,
        })
    }

    /// Apply a single ActionMessage to session state.
    fn apply_message(&mut self, msg: ActionMessage, identifier: &str) {
        match msg {
            ActionMessage::Progress(v) => {
                self.action.progress = Some(v);
            }
            ActionMessage::Status(s) => {
                self.action.status_message = Some(s);
            }
            ActionMessage::Fail(s) => {
                self.action.fail_message = Some(s);
            }
            ActionMessage::SetEnv { name, value } => {
                let key = normalize_env_key(&name);
                self.env_vars.insert(key.clone(), value.clone());
                if let Some(changes) = self.created_env_vars.get_mut(identifier) {
                    changes.insert(key, Some(value));
                }
            }
            ActionMessage::UnsetEnv { name } => {
                let key = normalize_env_key(&name);
                self.env_vars.remove(&key);
                if let Some(changes) = self.created_env_vars.get_mut(identifier) {
                    changes.insert(key, None);
                }
            }
            ActionMessage::RedactedEnv { name, value } => {
                if self.redactions_enabled() {
                    let key = normalize_env_key(&name);
                    self.env_vars.insert(key.clone(), value.clone());
                    if let Some(changes) = self.created_env_vars.get_mut(identifier) {
                        changes.insert(key, Some(value.clone()));
                    }
                }
                self.redacted_values.insert(value);
            }
            ActionMessage::CancelMarkFailed { fail_message } => {
                self.action.fail_message = Some(fail_message);
                let _ = self.cancel_action(None, true);
            }
        }
        if let Some(cb) = &self.callback {
            if let Some(status) = self.action_status() {
                cb(&self.session_id, &status);
            }
        }
    }

    /// Materialize path mapping rules to a JSON file and set symbol table entries.
    fn materialize_path_mapping(&self, symtab: &mut SymbolTable) -> Result<(), SessionError> {
        let has_rules = !self.path_mapping_rules.is_empty();
        let rules_json = if has_rules {
            let rules: Vec<serde_json::Value> = self
                .path_mapping_rules
                .iter()
                .map(|rule| {
                    serde_json::json!({
                        "source_path_format": match rule.source_path_format {
                            openjd_expr::path_mapping::PathFormat::Posix => "POSIX",
                            openjd_expr::path_mapping::PathFormat::Windows => "WINDOWS",
                            openjd_expr::path_mapping::PathFormat::Uri => "URI",
                        },
                        "source_path": &rule.source_path,
                        "destination_path": &rule.destination_path,
                    })
                })
                .collect();
            serde_json::json!({"version": "pathmapping-1.0", "path_mapping_rules": rules})
                .to_string()
        } else {
            serde_json::json!({}).to_string()
        };

        symtab
            .set(
                "Session.HasPathMappingRules",
                openjd_expr::ExprValue::Bool(has_rules),
            )
            .map_err(|e| {
                SessionError::Runtime(format!("Failed to set HasPathMappingRules: {e}"))
            })?;

        let filename = self.working_directory.join(format!(
            "pathmapping_{}.json",
            uuid::Uuid::new_v4().simple()
        ));
        std::fs::write(&filename, &rules_json).map_err(|e| SessionError::WorkingDirectory {
            path: filename.clone(),
            source: e,
        })?;

        symtab
            .set(
                "Session.PathMappingRulesFile",
                openjd_expr::ExprValue::new_path(
                    filename.to_string_lossy().to_string(),
                    openjd_expr::path_mapping::PathFormat::host(),
                ),
            )
            .map_err(|e| {
                SessionError::Runtime(format!("Failed to set PathMappingRulesFile: {e}"))
            })?;

        Ok(())
    }

    /// Evaluate the cumulative env vars from process_env + extra + per-environment changes.
    /// Mirrors Python `_evaluate_current_session_env_vars`.
    pub fn evaluate_env_vars(
        &self,
        extra: Option<&HashMap<String, String>>,
    ) -> HashMap<String, Option<String>> {
        let mut result: HashMap<String, Option<String>> = self
            .process_env
            .iter()
            .map(|(k, v)| (normalize_env_key(k), Some(v.clone())))
            .collect();
        // Per spec: expose the session working directory as an environment variable
        // so nested subprocesses can access it without template variable syntax.
        result.insert(
            "OPENJD_SESSION_WORKING_DIR".to_string(),
            Some(self.working_directory.to_string_lossy().into_owned()),
        );
        if let Some(extra) = extra {
            for (k, v) in extra {
                result.insert(normalize_env_key(k), Some(v.clone()));
            }
        }
        for id in &self.environments_entered {
            if let Some(changes) = self.created_env_vars.get(id) {
                for (name, value) in changes {
                    result.insert(name.clone(), value.clone());
                }
            }
        }
        result
    }

    /// Get the job parameter values.
    pub fn job_parameter_values(&self) -> &JobParameterValues {
        &self.job_parameter_values
    }

    /// Build a SymbolTable for running actions, mirroring Python's Session._symbol_table().
    /// Populates job parameters (Param.* and RawParam.*), task parameters (Task.Param.* and Task.RawParam.*),
    /// and Session.WorkingDirectory.
    ///
    /// If `base` is provided, clones it as the starting point (it already contains Param.*, RawParam.*,
    /// Job.Name, Step.Name, and let bindings) and only layers Session.WorkingDirectory and Task.* on top.
    /// If `base` is None, builds from scratch using self.job_parameter_values.
    pub fn build_symbol_table(
        &self,
        task_parameter_values: Option<&openjd_model::types::TaskParameterSet>,
        base: Option<&openjd_expr::SerializedSymbolTable>,
    ) -> Result<SymbolTable, SessionError> {
        use openjd_model::types::TaskParameterType;

        let mut symtab = if let Some(base) = base {
            // Deserialize with host path format — this is the template→session boundary.
            // Path values stored as Posix in template scope get normalized to host format.
            let mut s = base
                .to_symtab(openjd_expr::path_mapping::PathFormat::host())
                .map_err(|e| {
                    SessionError::Runtime(format!("Failed to deserialize resolved_symtab: {e}"))
                })?;
            // Re-apply path mapping to Param.* PATH values from the base symtab.
            // The base (resolved_symtab from create_job) has unmapped paths; the session
            // knows the worker's path mapping rules.
            for (name, param) in &self.job_parameter_values {
                use openjd_model::types::JobParameterType;
                match param.param_type {
                    JobParameterType::Path => {
                        let raw = match &param.value {
                            openjd_expr::ExprValue::String(s) => s.as_str(),
                            openjd_expr::ExprValue::Path { value, .. } => value.as_str(),
                            _ => continue,
                        };
                        let mapped = self.apply_path_mapping_to_string(raw);
                        let key = format!("Param.{name}");
                        s.set(
                            &key,
                            openjd_expr::ExprValue::new_path(
                                mapped,
                                openjd_expr::path_mapping::PathFormat::host(),
                            ),
                        )
                        .map_err(|e| SessionError::Runtime(format!("Failed to set {key}: {e}")))?;
                    }
                    JobParameterType::ListPath => {
                        if let openjd_expr::ExprValue::ListString(ref elements, _) = param.value {
                            let mapped: Vec<openjd_expr::ExprValue> = elements
                                .iter()
                                .map(|s| {
                                    let m = self.apply_path_mapping_to_string(s);
                                    openjd_expr::ExprValue::new_path(
                                        m,
                                        openjd_expr::path_mapping::PathFormat::host(),
                                    )
                                })
                                .collect();
                            let key = format!("Param.{name}");
                            s.set(
                                &key,
                                openjd_expr::ExprValue::make_list(
                                    mapped,
                                    openjd_expr::ExprType::PATH,
                                )
                                .unwrap(),
                            )
                            .map_err(|e| {
                                SessionError::Runtime(format!("Failed to set {key}: {e}"))
                            })?;
                        }
                    }
                    _ => {}
                }
            }
            s
        } else {
            let mut s = SymbolTable::new();
            for (name, param) in &self.job_parameter_values {
                let raw_key = format!("RawParam.{name}");
                s.set(&raw_key, param.value.clone())
                    .map_err(|e| SessionError::Runtime(format!("Failed to set {raw_key}: {e}")))?;
                let key = format!("Param.{name}");
                use openjd_model::types::JobParameterType;
                let mapped_value = match param.param_type {
                    JobParameterType::Path => {
                        let raw = match &param.value {
                            openjd_expr::ExprValue::String(s) => s.as_str(),
                            openjd_expr::ExprValue::Path { value, .. } => value.as_str(),
                            _ => "",
                        };
                        let mapped = self.apply_path_mapping_to_string(raw);
                        openjd_expr::ExprValue::new_path(
                            mapped,
                            openjd_expr::path_mapping::PathFormat::host(),
                        )
                    }
                    JobParameterType::ListPath => match &param.value {
                        openjd_expr::ExprValue::ListString(elements, _) => {
                            let mapped: Vec<openjd_expr::ExprValue> = elements
                                .iter()
                                .map(|s| {
                                    let m = self.apply_path_mapping_to_string(s);
                                    openjd_expr::ExprValue::new_path(
                                        m,
                                        openjd_expr::path_mapping::PathFormat::host(),
                                    )
                                })
                                .collect();
                            openjd_expr::ExprValue::make_list(mapped, openjd_expr::ExprType::PATH)
                                .unwrap()
                        }
                        other => self.apply_path_mapping_to_value(other),
                    },
                    _ => self.apply_path_mapping_to_value(&param.value),
                };
                s.set(&key, mapped_value)
                    .map_err(|e| SessionError::Runtime(format!("Failed to set {key}: {e}")))?;
            }
            s
        };

        let host_path_format = openjd_expr::path_mapping::PathFormat::host();
        symtab
            .set(
                "Session.WorkingDirectory",
                openjd_expr::ExprValue::new_path(
                    self.working_directory.to_string_lossy().to_string(),
                    host_path_format,
                ),
            )
            .map_err(|e| SessionError::Runtime(format!("Failed to set WorkingDirectory: {e}")))?;

        if let Some(task_params) = task_parameter_values {
            for (name, tv) in task_params {
                // Task.RawParam.* — raw string value
                let raw_key = format!("Task.RawParam.{name}");
                symtab
                    .set(&raw_key, tv.value.clone())
                    .map_err(|e| SessionError::Runtime(format!("Failed to set {raw_key}: {e}")))?;

                // Task.Param.* — typed value with path mapping applied for PATH types
                let key = format!("Task.Param.{name}");
                let param_value = match tv.param_type {
                    TaskParameterType::Path => {
                        let s = match &tv.value {
                            openjd_expr::ExprValue::String(s) => s.as_str(),
                            openjd_expr::ExprValue::Path { value, .. } => value.as_str(),
                            _ => "",
                        };
                        let mapped = self.apply_path_mapping_to_string(s);
                        openjd_expr::ExprValue::new_path(
                            mapped,
                            openjd_expr::path_mapping::PathFormat::host(),
                        )
                    }
                    _ => tv.value.clone(),
                };
                symtab
                    .set(&key, param_value)
                    .map_err(|e| SessionError::Runtime(format!("Failed to set {key}: {e}")))?;
            }
        }

        Ok(symtab)
    }

    /// Apply path mapping rules to a string, returning the mapped result.
    fn apply_path_mapping_to_string(&self, path: &str) -> String {
        for rule in self.path_mapping_rules.iter() {
            if let Some(mapped) = rule.apply(path) {
                return mapped;
            }
        }
        path.to_string()
    }

    /// Apply path mapping rules to a value if it's a Path or ListPath type.
    fn apply_path_mapping_to_value(
        &self,
        value: &openjd_expr::ExprValue,
    ) -> openjd_expr::ExprValue {
        match value {
            openjd_expr::ExprValue::Path {
                value: path_str,
                format,
                ..
            } => {
                for rule in self.path_mapping_rules.iter() {
                    if let Some(mapped) = rule.apply(path_str) {
                        return openjd_expr::ExprValue::new_path(mapped, *format);
                    }
                }
                value.clone()
            }
            openjd_expr::ExprValue::ListPath(elements, fmt, _) => {
                let mapped: Vec<openjd_expr::ExprValue> = elements
                    .iter()
                    .map(|s| {
                        let mapped_s =
                            openjd_expr::path_mapping::apply_rules(&self.path_mapping_rules, s);
                        openjd_expr::ExprValue::new_path(mapped_s, *fmt)
                    })
                    .collect();
                openjd_expr::ExprValue::make_list(mapped, openjd_expr::ExprType::PATH).unwrap()
            }
            _ => value.clone(),
        }
    }

    // ────────────────────────────────────────────────────────────────
    // RFC 0008: wrap-hook routing
    // ────────────────────────────────────────────────────────────────

    /// Return the innermost currently-active environment that defines *any*
    /// wrap hook, if one exists.
    ///
    /// The model-level validator rejects templates with two or more wrap
    /// layers in a single session, so this is effectively "the wrap env"
    /// at runtime. The innermost-wins traversal defends against templates
    /// that slipped past validation (e.g. if a future extension ever
    /// permits nested composition) by picking the behavior closest to
    /// the task and keeping the dispatch deterministic.
    fn active_wrap_env(&self) -> Option<&Environment> {
        for id in self.environments_entered.iter().rev() {
            if let Some(env) = self.environments.get(id) {
                if env_has_any_wrap_hook(env) {
                    return Some(env);
                }
            }
        }
        None
    }

    /// Return the active wrap environment *excluding* the environment
    /// currently entering or exiting (referenced by `self_id`). This is
    /// what `onWrapEnvEnter` / `onWrapEnvExit` dispatch needs: an environment's
    /// own lifecycle actions are never wrapped by its own wrap hooks.
    fn wrap_env_excluding(&self, self_id: &str) -> Option<&Environment> {
        for id in self.environments_entered.iter().rev() {
            if id == self_id {
                continue;
            }
            if let Some(env) = self.environments.get(id) {
                if env_has_any_wrap_hook(env) {
                    return Some(env);
                }
            }
        }
        None
    }
}

/// Returns true iff the environment defines any of the three wrap hooks
/// from RFC 0008. The model-level validator enforces that this is either
/// zero or one environment per session; this check gates runtime dispatch.
fn env_has_any_wrap_hook(env: &Environment) -> bool {
    env.script
        .as_ref()
        .map(|s| s.actions.has_any_wrap_hook())
        .unwrap_or(false)
}

/// The wrap-hook context variable available in addition to
/// `WrappedAction.*` (RFC 0008).
pub(crate) enum WrappedContext<'a> {
    /// Within `onWrapEnvEnter` / `onWrapEnvExit`: sets `WrappedEnv.Name`.
    Env(&'a str),
    /// Within `onWrapTaskRun`: sets `WrappedStep.Name`.
    Step(&'a str),
}

/// Deserialize a wrap environment's frozen `resolved_symtab` (if present)
/// and merge it into `action_symtab`, then resolve the wrapped action's
/// command/args/timeout and overlay the RFC 0008 `WrappedAction.*` (plus
/// the companion `WrappedEnv`/`WrappedStep` variable) onto the same table.
///
/// This is the single implementation shared by the three wrap-hook call
/// sites (`onWrapEnvEnter`, `onWrapTaskRun`, `onWrapEnvExit`), guaranteeing
/// the hooks see identical `WrappedAction.*` semantics as the RFC requires.
/// `phase` names the wrapped lifecycle action for error messages
/// ("onEnter", "onExit", or "task").
///
/// `session_env_vars` MUST be the session's `openjd_env`-exported variables
/// only (`self.env_vars`); host-inherited variables are intentionally
/// excluded per RFC 0008.
fn seed_wrapped_action_symbols(
    action_symtab: &mut SymbolTable,
    wrap_resolved: &Option<openjd_expr::SerializedSymbolTable>,
    wrapped_action: &openjd_model::job::Action,
    context: WrappedContext<'_>,
    session_env_vars: &HashMap<String, String>,
    lib: Option<&FunctionLibrary>,
    phase: &str,
) -> Result<(), SessionError> {
    // Layer the wrap env's frozen symtab on top of the action symtab so the
    // wrap action can reference symbols only it knows about (its own
    // `Param.*`, let bindings). A deserialize failure is logged and skipped
    // rather than failing the action — resolution will surface later only if
    // a missing symbol is actually referenced.
    if let Some(ser) = wrap_resolved.as_ref() {
        match ser.to_symtab(openjd_expr::path_mapping::PathFormat::host()) {
            Ok(st) => action_symtab.merge_from(&st),
            Err(e) => {
                log::warn!(
                    target: "openjd.sessions",
                    "wrap env resolved_symtab deserialize failed: {e}; \
                     WrappedAction.* continues without it"
                );
            }
        }
    }

    // Resolve the wrapped action's command/args using the inner scope
    // (the symtab built so far). These seed WrappedAction.Command/Args.
    let resolved_cmd = crate::runner::resolve_action_args(wrapped_action, action_symtab, lib)
        .map_err(|e| SessionError::FormatString {
            context: format!("wrapped {phase} command"),
            reason: e.to_string(),
        })?;
    let (cmd, args) = match resolved_cmd.split_first() {
        Some((head, tail)) => (head.clone(), tail.to_vec()),
        None => (String::new(), Vec::new()),
    };
    let wrapped_env: Vec<String> = session_env_vars
        .iter()
        .map(|(k, v)| format!("{k}={v}"))
        .collect();
    // WrappedAction.Timeout carries the ORIGINAL wrapped action's timeout
    // (RFC 0008), not the wrap action's. `0` means unset.
    let wrapped_timeout_secs =
        crate::runner::resolve_action_timeout(wrapped_action, action_symtab, lib, None)
            .map_err(|e| SessionError::FormatString {
                context: format!("wrapped {phase} timeout"),
                reason: e.to_string(),
            })?
            .map(|d| d.as_secs() as i64)
            .unwrap_or(0);
    overlay_wrapped_action_symbols(
        action_symtab,
        Some(context),
        &cmd,
        &args,
        &wrapped_env,
        wrapped_timeout_secs,
    )
}

/// Overlay the `WrappedAction.*` variables defined in RFC 0008 onto a
/// symbol table in place. Used by all three wrap hooks:
///
/// - `WrappedAction.Command` — the wrapped action's resolved command string.
/// - `WrappedAction.Args` — the wrapped action's resolved argument list.
/// - `WrappedAction.Environment` — `"KEY=value"` entries for every
///   `openjd_env` export captured so far in the session.
/// - `WrappedAction.Timeout` — the timeout in seconds of the wrapped
///   action, or `0` when the wrapped action specified no timeout.
///
/// `wrapped` selects the per-hook companion variable: `WrappedEnv.Name`
/// for env hooks, `WrappedStep.Name` for `onWrapTaskRun`. `None` is used
/// only by tests that exercise the `WrappedAction.*` portion in isolation.
///
/// Errors from `SymbolTable::set` are reported as `SessionError::Runtime`.
fn overlay_wrapped_action_symbols(
    symtab: &mut SymbolTable,
    wrapped: Option<WrappedContext<'_>>,
    wrapped_command: &str,
    wrapped_args: &[String],
    wrapped_environment: &[String],
    wrapped_timeout_secs: i64,
) -> Result<(), SessionError> {
    match wrapped {
        Some(WrappedContext::Env(name)) => {
            set_string_symbol(symtab, "WrappedEnv.Name", name)?;
        }
        Some(WrappedContext::Step(name)) => {
            set_string_symbol(symtab, "WrappedStep.Name", name)?;
        }
        None => {}
    }
    set_string_symbol(symtab, "WrappedAction.Command", wrapped_command)?;
    set_string_list_symbol(symtab, "WrappedAction.Args", wrapped_args)?;
    set_string_list_symbol(symtab, "WrappedAction.Environment", wrapped_environment)?;
    set_int_symbol(symtab, "WrappedAction.Timeout", wrapped_timeout_secs)?;
    Ok(())
}

fn set_string_symbol(
    symtab: &mut SymbolTable,
    name: &str,
    value: &str,
) -> Result<(), SessionError> {
    symtab
        .set(name, openjd_expr::ExprValue::String(value.into()))
        .map_err(|e| SessionError::Runtime(format!("Failed to set {name}: {e}")))
}

fn set_string_list_symbol(
    symtab: &mut SymbolTable,
    name: &str,
    values: &[String],
) -> Result<(), SessionError> {
    let list: Vec<openjd_expr::ExprValue> = values
        .iter()
        .map(|s| openjd_expr::ExprValue::String(s.clone()))
        .collect();
    let value = openjd_expr::ExprValue::make_list(list, openjd_expr::ExprType::STRING)
        .map_err(|e| SessionError::Runtime(format!("make_list({name}): {e}")))?;
    symtab
        .set(name, value)
        .map_err(|e| SessionError::Runtime(format!("Failed to set {name}: {e}")))
}

fn set_int_symbol(symtab: &mut SymbolTable, name: &str, value: i64) -> Result<(), SessionError> {
    symtab
        .set(name, openjd_expr::ExprValue::Int(value))
        .map_err(|e| SessionError::Runtime(format!("Failed to set {name}: {e}")))
}

impl Drop for Session {
    fn drop(&mut self) {
        if !self.cleanup_called {
            // `session_id` is an opaque correlation identifier, not a secret.
            // Including it in this warning is essential for diagnosing which
            // session leaked its working directory. See the module-level docs
            // for rationale.
            log::warn!(
                target: "openjd.sessions",
                "Session '{}' was dropped without calling cleanup(). \
                 Working directory may not have been cleaned up.",
                self.session_id
            );
            if !self.retain_working_dir {
                let _ = std::fs::remove_dir_all(&self.working_directory);
            }
        }
    }
}

#[cfg(test)]
mod wrap_actions_tests {
    //! Unit tests for the small pure helpers that back RFC 0008 wrap-hook
    //! dispatch. Behavioral end-to-end coverage lives in
    //! `tests/integration/test_wrap_actions.rs`.
    use super::*;
    use openjd_expr::ExprValue;
    use openjd_model::format_string::FormatString;
    use openjd_model::job::{Action, EnvironmentActions, EnvironmentScript};

    fn fs(s: &str) -> FormatString {
        FormatString::new(s).unwrap()
    }

    fn echo() -> Action {
        Action {
            command: fs("echo"),
            args: None,
            timeout: None,
            cancelation: None,
        }
    }

    fn env_with_actions(name: &str, actions: EnvironmentActions) -> Environment {
        Environment {
            name: name.to_string(),
            description: None,
            script: Some(EnvironmentScript {
                let_bindings: None,
                actions,
                embedded_files: None,
            }),
            variables: None,
            resolved_symtab: None,
        }
    }

    fn empty_actions() -> EnvironmentActions {
        EnvironmentActions {
            on_enter: None,
            on_wrap_env_enter: None,
            on_wrap_task_run: None,
            on_wrap_env_exit: None,
            on_exit: None,
        }
    }

    #[test]
    fn env_has_any_wrap_hook_returns_false_for_plain_env() {
        let env = env_with_actions(
            "Plain",
            EnvironmentActions {
                on_enter: Some(echo()),
                on_exit: Some(echo()),
                ..empty_actions()
            },
        );
        assert!(!env_has_any_wrap_hook(&env));
    }

    #[test]
    fn env_has_any_wrap_hook_returns_true_for_each_hook() {
        for actions in [
            EnvironmentActions {
                on_wrap_env_enter: Some(echo()),
                ..empty_actions()
            },
            EnvironmentActions {
                on_wrap_task_run: Some(echo()),
                ..empty_actions()
            },
            EnvironmentActions {
                on_wrap_env_exit: Some(echo()),
                ..empty_actions()
            },
        ] {
            let env = env_with_actions("Wrap", actions);
            assert!(env_has_any_wrap_hook(&env));
        }
    }

    #[test]
    fn env_has_any_wrap_hook_returns_false_when_script_missing() {
        let env = Environment {
            name: "NoScript".into(),
            description: None,
            script: None,
            variables: None,
            resolved_symtab: None,
        };
        assert!(!env_has_any_wrap_hook(&env));
    }

    #[test]
    fn overlay_sets_wrapped_action_symbols_for_task_hook() {
        let mut symtab = SymbolTable::default();
        overlay_wrapped_action_symbols(
            &mut symtab,
            Some(WrappedContext::Step("MyStep")),
            "echo",
            &["a".into(), "b c".into()],
            &["FOO=bar".into()],
            42,
        )
        .unwrap();

        assert_eq!(
            symtab.get_value("WrappedAction.Command"),
            Some(&ExprValue::String("echo".into()))
        );
        assert_eq!(
            symtab.get_value("WrappedAction.Timeout"),
            Some(&ExprValue::Int(42))
        );
        // WrappedStep.Name is set for task hooks; WrappedEnv.Name is not.
        assert_eq!(
            symtab.get_value("WrappedStep.Name"),
            Some(&ExprValue::String("MyStep".into()))
        );
        assert!(symtab.get_value("WrappedEnv.Name").is_none());
    }

    #[test]
    fn overlay_sets_wrapped_env_name_when_provided() {
        let mut symtab = SymbolTable::default();
        overlay_wrapped_action_symbols(
            &mut symtab,
            Some(WrappedContext::Env("InnerEnv")),
            "true",
            &[],
            &[],
            0,
        )
        .unwrap();

        assert_eq!(
            symtab.get_value("WrappedEnv.Name"),
            Some(&ExprValue::String("InnerEnv".into()))
        );
        assert!(symtab.get_value("WrappedStep.Name").is_none());
    }

    #[test]
    fn overlay_handles_empty_args_and_environment() {
        let mut symtab = SymbolTable::default();
        overlay_wrapped_action_symbols(&mut symtab, None, "true", &[], &[], 0).unwrap();
        // Both lists must be set as empty list[string] so iteration in wrap
        // scripts (`for a in {{ ... }}`) sees zero iterations rather than
        // "Undefined variable".
        assert!(symtab.get_value("WrappedAction.Args").is_some());
        assert!(symtab.get_value("WrappedAction.Environment").is_some());
        assert_eq!(
            symtab.get_value("WrappedAction.Timeout"),
            Some(&ExprValue::Int(0))
        );
    }
}