diff --git a/crates/openjd-model/src/job/create_job/instantiate.rs b/crates/openjd-model/src/job/create_job/instantiate.rs index 4430a533..9780d50f 100644 --- a/crates/openjd-model/src/job/create_job/instantiate.rs +++ b/crates/openjd-model/src/job/create_job/instantiate.rs @@ -501,9 +501,6 @@ fn filter_symtab_for_step( if let Some(d) = &f.data { d.copy_used_symtab_values(full, &mut filtered); } - if let Some(n) = &f.filename { - n.copy_used_symtab_values(full, &mut filtered); - } } } if let Some(bindings) = &s.let_bindings { @@ -525,9 +522,6 @@ fn filter_symtab_for_step( if let Some(d) = &f.data { d.copy_used_symtab_values(full, &mut filtered); } - if let Some(n) = &f.filename { - n.copy_used_symtab_values(full, &mut filtered); - } } } } @@ -617,9 +611,6 @@ fn collect_all_accessed_symbols( if let Some(d) = &f.data { collect_from_fs(d, &mut symbols); } - if let Some(n) = &f.filename { - collect_from_fs(n, &mut symbols); - } } } if let Some(bindings) = &s.let_bindings { @@ -650,9 +641,6 @@ fn collect_all_accessed_symbols( if let Some(d) = &f.data { collect_from_fs(d, &mut symbols); } - if let Some(n) = &f.filename { - collect_from_fs(n, &mut symbols); - } } } } @@ -717,9 +705,6 @@ fn filter_symtab_for_environment(env: &job::Environment, full: &SymbolTable) -> if let Some(d) = &f.data { d.copy_used_symtab_values(full, &mut filtered); } - if let Some(n) = &f.filename { - n.copy_used_symtab_values(full, &mut filtered); - } } } if let Some(bindings) = &es.let_bindings { @@ -763,9 +748,6 @@ fn collect_env_accessed_symbols(env: &job::Environment) -> std::collections::Has if let Some(d) = &f.data { symbols.extend(d.accessed_symbols()); } - if let Some(n) = &f.filename { - symbols.extend(n.accessed_symbols()); - } } } if let Some(bindings) = &es.let_bindings { diff --git a/crates/openjd-model/src/job/mod.rs b/crates/openjd-model/src/job/mod.rs index 41fe1433..914ea294 100644 --- a/crates/openjd-model/src/job/mod.rs +++ b/crates/openjd-model/src/job/mod.rs @@ -203,7 +203,7 @@ pub struct EmbeddedFile { pub name: String, #[serde(alias = "type")] pub file_type: FileType, - pub filename: Option, + pub filename: Option, pub data: Option, pub runnable: Option, pub end_of_line: Option, diff --git a/crates/openjd-model/src/template/environment.rs b/crates/openjd-model/src/template/environment.rs index c712b180..20abd675 100644 --- a/crates/openjd-model/src/template/environment.rs +++ b/crates/openjd-model/src/template/environment.rs @@ -38,7 +38,7 @@ pub struct EmbeddedFile { pub name: String, #[serde(rename = "type")] pub file_type: FileType, - pub filename: Option, + pub filename: Option, pub data: Option, pub runnable: Option, #[serde(rename = "endOfLine")] diff --git a/crates/openjd-model/src/template/step.rs b/crates/openjd-model/src/template/step.rs index 2fa14a5a..33649072 100644 --- a/crates/openjd-model/src/template/step.rs +++ b/crates/openjd-model/src/template/step.rs @@ -110,7 +110,7 @@ impl StepTemplate { embedded_files: Some(vec![EmbeddedFile { name: embedded_name, file_type: crate::types::FileType::Text, - filename: Some(FormatString::new(&filename).unwrap()), + filename: Some(filename), data: Some(FormatString::new(&sa.script).map_err(|e| { crate::ModelError::DecodeValidation(format!( "SimpleAction script format string error: {e}" diff --git a/crates/openjd-model/src/template/validate_v2023_09/format_strings.rs b/crates/openjd-model/src/template/validate_v2023_09/format_strings.rs index 94450f87..a8408b29 100644 --- a/crates/openjd-model/src/template/validate_v2023_09/format_strings.rs +++ b/crates/openjd-model/src/template/validate_v2023_09/format_strings.rs @@ -224,7 +224,11 @@ fn build_task_scope_symtab( } } - // Task.File.* from step script embedded files + // Task.File.* from step script embedded files. In scope for script-level + // `let` bindings too: `filename` is a plain string (never an expression), + // so the runtime allocates embedded file paths — defining Task.File.* — + // before evaluating `let`, mirroring the environment runner's Env.File.* + // ordering. if let Some(script) = step .resolve_syntax_sugar() .ok() @@ -662,7 +666,10 @@ pub fn validate_format_strings( if let Some(script) = &step.script { let script_path = path_field(&step_path, "script"); - // Script-level let bindings (TASK scope — host_lib) + // Script-level let bindings (TASK scope — host_lib). Task.File.* + // is in scope: file paths are allocated before `let` evaluation + // at runtime (filenames are plain strings, so allocation cannot + // depend on `let` values). if let Some(bindings) = &script.let_bindings { let let_path = path_field(&script_path, "let"); if !expr_active { @@ -763,15 +770,8 @@ pub fn validate_format_strings( errors, ); } - if let Some(filename) = &f.filename { - validate_fs( - filename, - &task_symtab, - &host_lib, - &path_field(&f_path, "filename"), - errors, - ); - } + // `filename` is a plain string per the 2023-09 schema + // (not @fmtstring) — no format-string validation. } } @@ -1186,16 +1186,8 @@ fn validate_env_format_strings( } validate_fs(data, symtab, lib, &data_path, errors); } - if let Some(filename) = &f.filename { - let filename_path = path_field(&f_path, "filename"); - if !expr_active && filename.has_complex_expressions() { - errors.add( - &filename_path, - "complex expressions require the EXPR extension.", - ); - } - validate_fs(filename, symtab, lib, &filename_path, errors); - } + // `filename` is a plain string per the 2023-09 schema + // (not @fmtstring) — no format-string validation. } } } diff --git a/crates/openjd-model/src/template/validate_v2023_09/limits.rs b/crates/openjd-model/src/template/validate_v2023_09/limits.rs index e743277d..83e05ccf 100644 --- a/crates/openjd-model/src/template/validate_v2023_09/limits.rs +++ b/crates/openjd-model/src/template/validate_v2023_09/limits.rs @@ -78,7 +78,7 @@ pub fn enforce_limits(jt: &JobTemplate, limits: &EffectiveLimits, errors: &mut V ); } if let Some(filename) = &f.filename { - if filename.raw().len() > limits.max_filename_len { + if filename.len() > limits.max_filename_len { errors.add( &path_field(&f_path, "filename"), format!("exceeds {} characters.", limits.max_filename_len), @@ -139,7 +139,7 @@ fn enforce_environment_limits( ); } if let Some(filename) = &f.filename { - if filename.raw().len() > limits.max_filename_len { + if filename.len() > limits.max_filename_len { errors.add( &path_field(&f_path, "filename"), format!("exceeds {} characters.", limits.max_filename_len), diff --git a/crates/openjd-model/src/template/validate_v2023_09/structure.rs b/crates/openjd-model/src/template/validate_v2023_09/structure.rs index 5ad6aad8..25907388 100644 --- a/crates/openjd-model/src/template/validate_v2023_09/structure.rs +++ b/crates/openjd-model/src/template/validate_v2023_09/structure.rs @@ -925,7 +925,7 @@ fn validate_embedded_files( } } if let Some(filename) = &f.filename { - let fname = filename.raw(); + let fname = filename.as_str(); if fname.is_empty() { errors.add(&path_field(&f_path, "filename"), "must not be empty."); } diff --git a/crates/openjd-model/tests/integration/test_create_job.rs b/crates/openjd-model/tests/integration/test_create_job.rs index 3c142fb0..7446df88 100644 --- a/crates/openjd-model/tests/integration/test_create_job.rs +++ b/crates/openjd-model/tests/integration/test_create_job.rs @@ -1641,9 +1641,9 @@ fn test_scope_boundary_embedded_file_filename_not_resolved() { ); let files = job.steps[0].script.embedded_files.as_ref().unwrap(); assert_eq!( - files[0].filename.as_ref().unwrap().raw(), - "{{Param.Val}}", - "EmbeddedFile filename is SESSION/TASK scope — must NOT be resolved during create_job" + files[0].filename.as_deref(), + Some("{{Param.Val}}"), + "EmbeddedFile filename is a plain string (not @fmtstring) — braces are literal text, never resolved" ); } @@ -1760,8 +1760,8 @@ fn test_scope_boundary_env_embedded_files_not_resolved() { .embedded_files .as_ref() .unwrap(); - assert_eq!(files[0].filename.as_ref().unwrap().raw(), "{{Param.Val}}", - "Environment embedded file filename is SESSION scope — must NOT be resolved during create_job"); + assert_eq!(files[0].filename.as_deref(), Some("{{Param.Val}}"), + "Environment embedded file filename is a plain string (not @fmtstring) — braces are literal text, never resolved"); assert_eq!( files[0].data.as_ref().unwrap().raw(), "echo {{Param.Val}}", diff --git a/crates/openjd-model/tests/integration/test_let_bindings.rs b/crates/openjd-model/tests/integration/test_let_bindings.rs index 0af368ea..93d3018b 100644 --- a/crates/openjd-model/tests/integration/test_let_bindings.rs +++ b/crates/openjd-model/tests/integration/test_let_bindings.rs @@ -639,7 +639,10 @@ fn test_env_file_has_path_properties() { #[test] fn test_task_file_has_path_properties() { - // Task.File.* is PATH type, so .stem should work in script-level let bindings + // Task.File.* is PATH type and IS in scope for script-level let bindings: + // `filename` is a plain string (never an expression), so the runtime + // allocates embedded file paths — defining Task.File.* — before `let` + // evaluation, mirroring Env.File.* in environment scripts. let s = r#"{ "specificationVersion": "jobtemplate-2023-09", "extensions": ["EXPR"], diff --git a/crates/openjd-model/tests/integration/test_scope_library_split.rs b/crates/openjd-model/tests/integration/test_scope_library_split.rs index 3c4a2220..59544ade 100644 --- a/crates/openjd-model/tests/integration/test_scope_library_split.rs +++ b/crates/openjd-model/tests/integration/test_scope_library_split.rs @@ -171,7 +171,11 @@ fn task_scope_embedded_file_data_accepts_apply_path_mapping() { } #[test] -fn task_scope_embedded_file_filename_accepts_apply_path_mapping() { +fn task_scope_embedded_file_filename_is_plain_string() { + // `filename` is a plain string per the 2023-09 schema (not @fmtstring): + // brace syntax is literal text, so no format-string/library validation + // applies (though it still fails the runtime basename check if it + // contains separators — none here). let t = job_with_path_param( r#""test""#, &simple_step( diff --git a/crates/openjd-sessions/src/embedded_files.rs b/crates/openjd-sessions/src/embedded_files.rs index 33d26a8d..2044e6ba 100644 --- a/crates/openjd-sessions/src/embedded_files.rs +++ b/crates/openjd-sessions/src/embedded_files.rs @@ -178,14 +178,14 @@ fn random_hex_filename() -> String { /// rejecting `/` and `\` in the filename. /// /// This function is a defense-in-depth check at the point where the -/// resolved filename is joined to the target directory. It protects against: +/// filename is joined to the target directory. It protects against: /// - Bugs or gaps in model-layer validation. /// - Call paths that reach the session layer without going through full -/// model validation. -/// - Any implementation-level format-string substitution in the filename -/// (the current model stores `filename` as a `FormatString`) that could -/// introduce separators or traversal components from symbol values at -/// session time. +/// model validation (e.g. an `EmbeddedFile` constructed directly). +/// +/// Note: `filename` is a plain string per the 2023-09 schema (not +/// `@fmtstring`), so no expression substitution can introduce separators +/// at session time — the value checked here is exactly the template value. /// /// Rejects filenames that: /// - Are empty. @@ -268,27 +268,23 @@ impl EmbeddedFiles { ); for file in files { let symbol = symtab_key(self.scope, &file.name); - let filename = if let Some(ref fname_fs) = file.filename { - let resolved = fname_fs - .resolve_string_with(symtab, &openjd_expr::FormatStringOptions::new()) - .map_err(|e| SessionError::FormatString { - context: format!("embedded file '{}' filename", file.name), - reason: e.to_string(), - })?; + let filename = if let Some(ref fname) = file.filename { + // `filename` is a plain string per the 2023-09 schema (not + // @fmtstring) — use it literally, no expression resolution. + // // Defense-in-depth: the spec requires `filename` to be a // plain basename and the model layer rejects path separators // in the raw template. Re-check the value here before it is // joined to the target directory, in case model validation - // was bypassed or the resolved value differs from the raw - // template. - validate_resolved_filename(&resolved).map_err(|reason| { + // was bypassed. + validate_resolved_filename(fname).map_err(|reason| { SessionError::EmbeddedFilePath { name: file.name.clone(), - filename: resolved.clone(), + filename: fname.clone(), reason, } })?; - self.target_directory.join(resolved) + self.target_directory.join(fname) } else { let name = random_hex_filename(); let path = self.target_directory.join(&name); diff --git a/crates/openjd-sessions/src/lib.rs b/crates/openjd-sessions/src/lib.rs index 4fec652a..6faf7bcb 100644 --- a/crates/openjd-sessions/src/lib.rs +++ b/crates/openjd-sessions/src/lib.rs @@ -41,7 +41,9 @@ pub use error::SessionError; pub use logging::LogContent; pub use openjd_expr::path_mapping::{PathFormat, PathMappingRule}; pub use runner::{CancelMethod, ScriptRunnerState}; -pub use session::{EnvironmentIdentifier, Session, SessionConfig, SessionState}; +pub use session::{ + EnvironmentIdentifier, Session, SessionCancelHandle, SessionConfig, SessionState, +}; #[cfg(windows)] pub use session_user::BadCredentialsError; #[cfg(unix)] diff --git a/crates/openjd-sessions/src/runner/step_script.rs b/crates/openjd-sessions/src/runner/step_script.rs index 6526ed4f..5ca0b838 100644 --- a/crates/openjd-sessions/src/runner/step_script.rs +++ b/crates/openjd-sessions/src/runner/step_script.rs @@ -119,18 +119,14 @@ impl StepScriptRunner { env_vars: &HashMap>, message_tx: mpsc::UnboundedSender, ) -> Result { - // Step scripts: evaluate let bindings first, then materialize files - let mut final_symtab = if let Some(bindings) = &script.let_bindings { - evaluate_let_bindings(bindings, symtab, library, openjd_expr::PathFormat::host()) - .map_err(|e| SessionError::FormatString { - context: "let bindings".into(), - reason: e.to_string(), - })? - } else { - symtab.clone() - }; - - if let Some(files) = &script.embedded_files { + // Step scripts: allocate embedded file paths first — `filename` is a + // plain string (never an expression) so allocation has no dependency + // on `let` values, and it defines the `Task.File.*` symbols that + // `let` bindings may reference. File *contents* are written after + // `let` evaluation so `data` expressions can use let-bound values. + // This matches the environment runner's ordering for `Env.File.*`. + let mut final_symtab = symtab.clone(); + let ef = if let Some(files) = &script.embedded_files { let mut ef = EmbeddedFiles::new( EmbeddedFilesScope::Step, self.base.files_directory.clone(), @@ -138,6 +134,25 @@ impl StepScriptRunner { ) .with_user(self.base.user.clone()); ef.allocate_file_paths(files, &mut final_symtab)?; + Some(ef) + } else { + None + }; + + if let Some(bindings) = &script.let_bindings { + final_symtab = evaluate_let_bindings( + bindings, + &final_symtab, + library, + openjd_expr::PathFormat::host(), + ) + .map_err(|e| SessionError::FormatString { + context: "let bindings".into(), + reason: e.to_string(), + })?; + } + + if let Some(ef) = ef { ef.write_file_contents(&final_symtab, library)?; } diff --git a/crates/openjd-sessions/src/session.rs b/crates/openjd-sessions/src/session.rs index 8bc67069..ce815d6c 100644 --- a/crates/openjd-sessions/src/session.rs +++ b/crates/openjd-sessions/src/session.rs @@ -28,6 +28,7 @@ use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::sync::Mutex as StdMutex; use std::time::Duration; use openjd_expr::function_library::FunctionLibrary; @@ -203,14 +204,22 @@ impl ActionStatusFields { } } -/// Cancellation state for the current action and external cancellation support. -struct CancelFields { +/// Cancellation state for the current action, shared with any +/// [`SessionCancelHandle`]s so cancellation can be requested from another +/// thread while the `Session` itself is owned by an action-running thread. +struct CancelShared { /// Token for the current action (cancelled to abort the subprocess). token: Option, /// Channel to send cancel requests (with optional time limit) to the subprocess. request_tx: Option>>, /// When true, a Canceled action result is reported as Failed. mark_failed: bool, +} + +/// Cancellation state for the current action and external cancellation support. +struct CancelFields { + /// Per-action cancellation state, shared with `SessionCancelHandle`s. + shared: Arc>, /// External cancellation token from the caller; action tokens are children of this. parent_token: Option, } @@ -218,12 +227,133 @@ struct CancelFields { impl CancelFields { fn new(parent_token: Option) -> Self { Self { - token: None, - request_tx: None, - mark_failed: false, + shared: Arc::new(StdMutex::new(CancelShared { + token: None, + request_tx: None, + mark_failed: false, + })), parent_token, } } + + /// Lock the shared state, recovering from a poisoned mutex. The state is + /// plain data (no invariants across fields), so recovery is safe. + fn lock(&self) -> std::sync::MutexGuard<'_, CancelShared> { + self.shared + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn set_action( + &self, + token: CancellationToken, + request_tx: tokio::sync::watch::Sender>, + ) { + let mut shared = self.lock(); + shared.token = Some(token); + shared.request_tx = Some(request_tx); + } + + fn reset(&self) { + let mut shared = self.lock(); + shared.token = None; + shared.request_tx = None; + shared.mark_failed = false; + } +} + +/// A thread-safe handle for cancelling a `Session`'s currently running +/// action from another thread — including while the `Session` value itself +/// is exclusively borrowed (or owned) by the thread driving the action. +/// +/// Obtain one via [`Session::cancel_handle`]. The handle stays valid for the +/// life of the session and can be used repeatedly: each action installs its +/// own cancellation token, and [`SessionCancelHandle::cancel`] cancels +/// whichever action is running at the time of the call. +/// +/// Cancellation delivered through this handle follows the action's own +/// cancelation method (e.g. NOTIFY_THEN_TERMINATE), exactly like +/// [`Session::cancel_action`]. One difference: the handle cannot update the +/// session's `state` field (the `Session` is owned elsewhere), so the +/// `Running` → `Canceling` transition that `cancel_action` performs is not +/// reflected; the state moves directly to the terminal value when the +/// canceled action finishes. Callers tracking a Canceling phase must do so +/// themselves. +pub struct SessionCancelHandle { + shared: Arc>, + /// Dup'd cancel-command pipe to the cross-user helper, when one exists. + cancel_writer: Option, + helper_auth_token: Option, +} + +impl SessionCancelHandle { + /// Request cancellation of the currently running action. + /// + /// * `time_limit` — same semantics as [`Session::cancel_action`]: an + /// urgent bound on the cancel; `Some(0)` turns a notify-then-terminate + /// cancel into an immediate terminate. + /// * `mark_action_failed` — report the canceled action as Failed + /// instead of Canceled. + /// + /// Returns `true` if a running action was found and cancellation was + /// delivered; `false` if no action was running. + pub fn cancel(&self, time_limit: Option, mark_action_failed: bool) -> bool { + // Hold the lock across the whole delivery so the target cannot + // change out from under us: `set_action`/`reset` (start/end of an + // action on the session thread) take the same lock, so the helper + // pipe command, the time-limit send, and the token cancel are all + // guaranteed to refer to the same action. + let mut shared = self + .shared + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let Some(token) = shared.token.clone() else { + return false; + }; + if mark_action_failed { + shared.mark_failed = true; + } + + // Cross-user sessions: also deliver the cancel command to the helper + // process over the dup'd pipe, mirroring Session::cancel_action. + if let Some(ref writer) = self.cancel_writer { + send_helper_cancel_command(writer, self.helper_auth_token.as_deref(), time_limit); + } + + if let Some(tx) = &shared.request_tx { + let _ = tx.send(time_limit); + } + token.cancel(); + true + } +} + +/// Write a tokenized cancel command to the cross-user helper's cancel pipe. +/// Shared by [`Session::cancel_action`] and [`SessionCancelHandle::cancel`]. +fn send_helper_cancel_command( + mut writer: &std::fs::File, + auth_token: Option<&str>, + time_limit: Option, +) { + use std::io::Write; + let is_terminate = matches!(time_limit, Some(d) if d.is_zero()); + let token_field = match 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(); } /// Cross-user execution state. @@ -709,7 +839,7 @@ impl Session { } self.state = SessionState::Canceling; if mark_action_failed { - self.cancel.mark_failed = true; + self.cancel.lock().mark_failed = true; } // Send cancel to the helper process via the dup'd stdin fd. @@ -720,37 +850,82 @@ impl Session { // 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(ref writer) = self.cross_user.cancel_writer { + send_helper_cancel_command( + writer, + self.cross_user.helper_auth_token.as_deref(), + time_limit, + ); } - if let Some(tx) = &self.cancel.request_tx { + let (token, request_tx) = { + let shared = self.cancel.lock(); + (shared.token.clone(), shared.request_tx.clone()) + }; + if let Some(tx) = request_tx { let _ = tx.send(time_limit); } - if let Some(token) = &self.cancel.token { + if let Some(token) = token { token.cancel(); } Ok(()) } + /// Get a thread-safe handle for cancelling this session's running action + /// from another thread, e.g. while the `Session` value is owned by the + /// thread driving the action. See [`SessionCancelHandle`]. + pub fn cancel_handle(&self) -> SessionCancelHandle { + let cancel_writer = self.clone_cancel_writer(); + if self.cross_user.cancel_writer.is_some() && cancel_writer.is_none() { + // try_clone failed: the handle will still cancel same-user + // subprocesses via the token, but helper-routed processes + // would not receive the pipe command. Surface it rather than + // failing silently at cancel time. + session_log!( + warn, + &self.session_id, + LogContent::PROCESS_CONTROL, + "Failed to duplicate the cross-user cancel pipe for a cancel \ + handle; cancels via this handle will not reach helper-run \ + processes" + ); + } + SessionCancelHandle { + shared: self.cancel.shared.clone(), + cancel_writer, + helper_auth_token: self.cross_user.helper_auth_token.clone(), + } + } + + /// Record a failure that occurred while setting up an action that has + /// already been reported as Running (e.g. wrap-symbol seeding or path + /// mapping materialization failing before the subprocess dispatch). + /// + /// Without this, an early `?` return would leave the session state + /// permanently `Running` with no terminal `ActionStatus` — callers + /// polling for completion (notably the Python bindings) would hang, and + /// nothing about the failure would appear in the session log. + /// + /// Marks the action Failed with the error message, resets cancellation + /// state, drops out of Running, logs the failure, and notifies the + /// callback. Returns the error so call sites can use it in `map_err`. + fn fail_action_setup(&mut self, e: SessionError) -> SessionError { + session_log!( + error, + &self.session_id, + LogContent::EXCEPTION_INFO, + "Action failed to run: {e}" + ); + self.action.state = Some(ActionState::Failed); + self.action.fail_message = Some(e.to_string()); + self.action.ended_at = Some(std::time::SystemTime::now()); + self.action.exit_code = None; + self.cancel.reset(); + self.state = SessionState::ReadyEnding; + self.notify_callback(); + e + } + /// Clean up the session. Deletes working directory if not retained. pub fn cleanup(&mut self) { if self.cleanup_called { @@ -901,12 +1076,12 @@ impl Session { 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); + self.cancel.set_action(cancel_token.clone(), 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)?; + self.materialize_path_mapping(&mut action_symtab) + .map_err(|e| self.fail_action_setup(e))?; // RFC 0008: if an outer active wrap env (not including this // one — which isn't on the stack yet but the helper guards @@ -937,7 +1112,8 @@ impl Session { &self.env_vars, Some(&lib), "onEnter", - )?; + ) + .map_err(|e| self.fail_action_setup(e))?; } // Box large locals — see run_task for rationale. @@ -1114,11 +1290,11 @@ impl Session { 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); + self.cancel.set_action(cancel_token.clone(), cancel_tx); let mut action_symtab = symtab.clone(); - self.materialize_path_mapping(&mut action_symtab)?; + self.materialize_path_mapping(&mut action_symtab) + .map_err(|e| self.fail_action_setup(e))?; // RFC 0008: does an outer active wrap env wrap this onExit? // The env being exited has already been popped from the stack @@ -1148,7 +1324,8 @@ impl Session { &self.env_vars, Some(&lib), "onExit", - )?; + ) + .map_err(|e| self.fail_action_setup(e))?; } // Box large locals — see run_task for rationale. @@ -1280,12 +1457,12 @@ impl Session { 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); + self.cancel.set_action(cancel_token.clone(), 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)?; + self.materialize_path_mapping(&mut action_symtab) + .map_err(|e| self.fail_action_setup(e))?; // RFC 0008: decide whether this task's onRun should be wrapped by // an active environment's onWrapTaskRun. The decision is: @@ -1332,7 +1509,8 @@ impl Session { )?; Ok::<_, SessionError>(action) }) - .transpose()?; + .transpose() + .map_err(|e| self.fail_action_setup(e))?; // Box large locals so they live on the heap instead of inflating // this async fn's state machine. Without this, the combined future @@ -1479,8 +1657,7 @@ impl Session { 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); + self.cancel.set_action(cancel_token.clone(), cancel_tx); let config = crate::subprocess::SubprocessConfig { args: cmd_args, @@ -1528,6 +1705,20 @@ impl Session { env_vars: std::collections::HashMap>, _timeout: Option, ) -> Result { + // Register per-action cancel state even though the helper protocol — + // not the token — carries the cancel to the subprocess: the token's + // presence is what tells a `SessionCancelHandle` that an action is + // in flight, so `cancel()` proceeds to write the helper pipe command + // (and returns true) instead of reporting "nothing running". + // + // The watch receiver goes into the SubprocessConfig: run_via_helper + // classifies the exit as Canceled only when the cancel-request + // channel has fired (rx.has_changed()), so without it a canceled + // helper subprocess would be misreported as Failed. + let cancel_token = self.new_action_cancel_token(); + let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(None); + self.cancel.set_action(cancel_token, cancel_tx); + let mut filter = crate::action_filter::ActionFilter::new( &self.session_id, self.echo_openjd_directives, @@ -1547,7 +1738,7 @@ impl Session { timeout: _timeout, user: self.cross_user.user.clone(), cancel_method: crate::runner::CancelMethod::Terminate, - cancel_request_rx: None, + cancel_request_rx: Some(cancel_rx), debug_collect_stdout: self.debug_collect_stdout, }; @@ -1571,20 +1762,34 @@ impl Session { self.apply_message(msg, &subprocess_identifier); } - let r = result?; - self.action.state = Some(r.state); + // A helper-run failure must not leave the session wedged in Running + // with no terminal ActionStatus — same contract as drive_action's + // error branch. + let r = result.map_err(|e| self.fail_action_setup(e))?; + + // Same mark_failed conversion as drive_action: a cancel requested + // with mark_action_failed=true (e.g. via SessionCancelHandle) must + // report the canceled action as Failed, not Canceled. + let final_state = if self.cancel.lock().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 r.state == ActionState::Success { + self.cancel.reset(); + self.state = if final_state == ActionState::Success { SessionState::Ready } else { SessionState::ReadyEnding }; self.notify_callback(); - Ok(r) + Ok(crate::subprocess::SubprocessResult { + state: final_state, + ..r + }) } // --- Internal helpers --- @@ -1629,13 +1834,20 @@ impl Session { 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. + // Update session state so callers see Failed instead of stuck Running, + // record the reason on the action status, and log it — otherwise the + // failure is invisible to log-followers and status pollers alike. + session_log!( + error, + &self.session_id, + LogContent::EXCEPTION_INFO, + "Action failed to run: {e}" + ); self.action.state = Some(ActionState::Failed); + self.action.fail_message = Some(e.to_string()); 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.cancel.reset(); self.state = SessionState::ReadyEnding; if let Some(cb) = &self.callback { @@ -1650,7 +1862,7 @@ impl Session { // 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 { + let final_state = if self.cancel.lock().mark_failed && r.state == ActionState::Canceled { ActionState::Failed } else { r.state @@ -1659,9 +1871,7 @@ impl Session { 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.cancel.reset(); self.state = if self.ending_only || final_state != ActionState::Success { SessionState::ReadyEnding diff --git a/crates/openjd-sessions/tests/integration/test_cross_user.rs b/crates/openjd-sessions/tests/integration/test_cross_user.rs index 5556d06c..90e372f6 100644 --- a/crates/openjd-sessions/tests/integration/test_cross_user.rs +++ b/crates/openjd-sessions/tests/integration/test_cross_user.rs @@ -337,6 +337,85 @@ async fn test_cross_user_session_run_subprocess() { session.cleanup(); } +// === Cross-user Session cancel-handle test === + +/// A `SessionCancelHandle` must be able to cancel a subprocess that runs via +/// the cross-user helper: the helper path registers per-action cancel state, +/// and the handle delivers the cancel command over the helper pipe. This is +/// the cross-user counterpart of the same-user handle tests in +/// `test_session.rs` (which cannot exercise helper routing). +#[tokio::test(flavor = "multi_thread")] +#[ignore] +async fn test_cross_user_session_cancel_handle_cancels_helper_subprocess() { + let user = require_target_user(); + let mut session = make_session(user); + let handle = session.cancel_handle(); + + let delivered: Arc>> = Arc::new(std::sync::Mutex::new(None)); + let delivered_clone = delivered.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(500)).await; + *delivered_clone.lock().unwrap() = Some(handle.cancel(None, false)); + }); + + let started = std::time::Instant::now(); + let result = session + .run_subprocess("sleep", Some(&["60".to_string()]), None, None, true, None) + .await; + let elapsed = started.elapsed(); + + assert_eq!( + *delivered.lock().unwrap(), + Some(true), + "handle must find the in-flight helper action and deliver the cancel" + ); + assert!( + elapsed < Duration::from_secs(30), + "cancel must interrupt the 60s sleep, took {elapsed:?}" + ); + match result { + // With the cancel-request channel wired into the helper config, the + // exit must classify as Canceled rather than Failed. + Ok(r) => assert_eq!( + r.state, + ActionState::Canceled, + "canceled helper subprocess must classify as Canceled" + ), + Err(_) => { /* helper-side cancel may surface as an error result */ } + } + session.cleanup(); +} + +/// `mark_action_failed=true` must convert a canceled helper subprocess to +/// Failed, matching drive_action's conversion and the +/// `SessionCancelHandle::cancel` contract. +#[tokio::test(flavor = "multi_thread")] +#[ignore] +async fn test_cross_user_session_cancel_handle_mark_failed_helper_subprocess() { + let user = require_target_user(); + let mut session = make_session(user); + let handle = session.cancel_handle(); + + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(500)).await; + handle.cancel(None, true); + }); + + let result = session + .run_subprocess("sleep", Some(&["60".to_string()]), None, None, true, None) + .await; + + match result { + Ok(r) => assert_eq!( + r.state, + ActionState::Failed, + "mark_action_failed must report the canceled helper subprocess as Failed" + ), + Err(_) => { /* helper-side cancel may surface as an error result */ } + } + session.cleanup(); +} + // === Cross-user TempDir tests === /// TempDir with target user has correct group ownership and mode 0o770. diff --git a/crates/openjd-sessions/tests/integration/test_cross_user_windows.rs b/crates/openjd-sessions/tests/integration/test_cross_user_windows.rs index b86eefe8..6a12fcbb 100644 --- a/crates/openjd-sessions/tests/integration/test_cross_user_windows.rs +++ b/crates/openjd-sessions/tests/integration/test_cross_user_windows.rs @@ -548,6 +548,105 @@ async fn test_cross_user_session_run_subprocess() { session.cleanup(); } +// === Session-level: SessionCancelHandle cancels a helper-routed subprocess === + +/// A `SessionCancelHandle` must be able to cancel a subprocess that runs via +/// the cross-user helper: the helper path registers per-action cancel state, +/// and the handle delivers the cancel command over the helper pipe. This is +/// the cross-user counterpart of the same-user handle tests in +/// `test_session.rs` (which cannot exercise helper routing). +#[tokio::test(flavor = "multi_thread")] +#[ignore] +async fn test_cross_user_session_cancel_handle_cancels_helper_subprocess() { + let user = require_windows_user(); + let mut session = make_session(user); + let handle = session.cancel_handle(); + + let delivered: Arc>> = Arc::new(std::sync::Mutex::new(None)); + let delivered_clone = delivered.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + *delivered_clone.lock().unwrap() = Some(handle.cancel(None, false)); + }); + + let started = std::time::Instant::now(); + let result = session + .run_subprocess( + "powershell", + Some(&[ + "-Command".to_string(), + "Start-Sleep -Seconds 60".to_string(), + ]), + None, + None, + true, + None, + ) + .await; + let elapsed = started.elapsed(); + + assert_eq!( + *delivered.lock().unwrap(), + Some(true), + "handle must find the in-flight helper action and deliver the cancel" + ); + assert!( + elapsed < std::time::Duration::from_secs(30), + "cancel must interrupt the 60s sleep, took {elapsed:?}" + ); + match result { + // With the cancel-request channel wired into the helper config, the + // exit must classify as Canceled rather than Failed. + Ok(r) => assert_eq!( + r.state, + ActionState::Canceled, + "canceled helper subprocess must classify as Canceled" + ), + Err(_) => { /* helper-side cancel may surface as an error result */ } + } + session.cleanup(); +} + +/// `mark_action_failed=true` must convert a canceled helper subprocess to +/// Failed, matching drive_action's conversion and the +/// `SessionCancelHandle::cancel` contract. +#[tokio::test(flavor = "multi_thread")] +#[ignore] +async fn test_cross_user_session_cancel_handle_mark_failed_helper_subprocess() { + let user = require_windows_user(); + let mut session = make_session(user); + let handle = session.cancel_handle(); + + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + handle.cancel(None, true); + }); + + let result = session + .run_subprocess( + "powershell", + Some(&[ + "-Command".to_string(), + "Start-Sleep -Seconds 60".to_string(), + ]), + None, + None, + true, + None, + ) + .await; + + match result { + Ok(r) => assert_eq!( + r.state, + ActionState::Failed, + "mark_action_failed must report the canceled helper subprocess as Failed" + ), + Err(_) => { /* helper-side cancel may surface as an error result */ } + } + session.cleanup(); +} + // === Session-level: cleanup with cross-user files === /// Session cleanup deletes the working directory even when files were created diff --git a/crates/openjd-sessions/tests/integration/test_embedded_files.rs b/crates/openjd-sessions/tests/integration/test_embedded_files.rs index b46c5ba1..017251db 100644 --- a/crates/openjd-sessions/tests/integration/test_embedded_files.rs +++ b/crates/openjd-sessions/tests/integration/test_embedded_files.rs @@ -253,6 +253,55 @@ fn test_materialize_resolved_data() { assert_eq!(fs::read_to_string(&bar_path).unwrap(), resolved); } +// === Filename is a plain string (2023-09 schema: not @fmtstring) === + +/// `{{ }}` sequences in an embedded file's `filename` are literal text — +/// never resolved as expressions — even when the referenced symbol exists +/// in the symbol table. Pins the sessions-layer half of the plain-string +/// filename contract (the model layer pins the decode half). +#[test] +fn literal_braces_in_filename_are_not_resolved() { + use openjd_expr::format_string::FormatString; + use openjd_expr::ExprValue; + use openjd_model::job::EmbeddedFile; + use openjd_model::symbol_table::SymbolTable; + use openjd_model::types::FileType; + + let tmp = TempDir::new().unwrap(); + let mut ef = EmbeddedFiles::new( + EmbeddedFilesScope::Step, + tmp.path().to_path_buf(), + "test-session", + ); + let file = EmbeddedFile { + name: "f".to_string(), + file_type: FileType::Text, + filename: Some("literal_{{Param.X}}.txt".to_string()), + data: Some(FormatString::new("hello").unwrap()), + runnable: None, + end_of_line: None, + }; + // Param.X IS defined — a regression to format-string semantics would + // silently resolve rather than error. + let mut st = SymbolTable::new(); + st.set("Param.X", ExprValue::String("resolved".to_string())) + .unwrap(); + + ef.allocate_file_paths(&[file], &mut st).unwrap(); + ef.write_file_contents(&st, None).unwrap(); + + let literal = tmp.path().join("literal_{{Param.X}}.txt"); + assert!( + literal.exists(), + "file must be created under the literal filename" + ); + assert_eq!(fs::read_to_string(&literal).unwrap(), "hello"); + assert!( + !tmp.path().join("literal_resolved.txt").exists(), + "filename must not be expression-resolved" + ); +} + // === Path traversal defense-in-depth (CWE-22) === // // Per spec (§6.1.1), embedded file `filename` must be a plain basename and @@ -260,39 +309,30 @@ fn test_materialize_resolved_data() { // time. These tests exercise the sessions-layer check that re-validates // the filename before it is joined to the target directory. This check // catches traversal strings that could reach the session layer via any -// bypass of model validation, including implementation-level format-string -// substitution (the current model stores `filename` as a `FormatString`). +// bypass of model validation (e.g. an `EmbeddedFile` constructed directly +// rather than decoded from a validated template). mod path_traversal { use super::*; use openjd_expr::format_string::FormatString; - use openjd_expr::ExprValue; use openjd_model::job::EmbeddedFile; use openjd_model::symbol_table::SymbolTable; use openjd_model::types::FileType; use openjd_sessions::SessionError; - /// Build an `EmbeddedFile` with a `filename` template that references - /// `Param.Evil` — the tainted value is supplied via the symbol table so - /// the traversal string bypasses any raw-template validation. - fn file_with_tainted_filename(name: &str) -> EmbeddedFile { + /// Build an `EmbeddedFile` whose `filename` carries a traversal string + /// directly, as if model validation had been bypassed. + fn file_with_tainted_filename(name: &str, filename: &str) -> EmbeddedFile { EmbeddedFile { name: name.to_string(), file_type: FileType::Text, - filename: Some(FormatString::new("{{Param.Evil}}").unwrap()), + filename: Some(filename.to_string()), data: Some(FormatString::new("echo hello").unwrap()), runnable: None, end_of_line: None, } } - fn symtab_with_evil(value: &str) -> SymbolTable { - let mut st = SymbolTable::new(); - st.set("Param.Evil", ExprValue::String(value.to_string())) - .unwrap(); - st - } - fn allocate_with_tainted(value: &str) -> Result<(), SessionError> { let tmp = TempDir::new().unwrap(); let mut ef = EmbeddedFiles::new( @@ -300,8 +340,8 @@ mod path_traversal { tmp.path().to_path_buf(), "test-session", ); - let mut st = symtab_with_evil(value); - let file = file_with_tainted_filename("evil"); + let mut st = SymbolTable::new(); + let file = file_with_tainted_filename("evil", value); ef.allocate_file_paths(&[file], &mut st) } diff --git a/crates/openjd-sessions/tests/integration/test_session.rs b/crates/openjd-sessions/tests/integration/test_session.rs index c8ef8b85..9a22b6cc 100644 --- a/crates/openjd-sessions/tests/integration/test_session.rs +++ b/crates/openjd-sessions/tests/integration/test_session.rs @@ -2837,6 +2837,93 @@ mod cancel_escalation { assert_eq!(msgs[0]["token"].as_str().unwrap(), "AbCdEfGhIjKlMnOpQrStUv",); assert_eq!(msgs[0]["cancel"].as_str().unwrap(), "TERMINATE"); } + + /// `SessionCancelHandle` must deliver the helper pipe command like + /// `cancel_action` does. Unlike `cancel_action`, the handle only fires + /// when a per-action token is registered, so this drives a real running + /// action and cancels it mid-flight from another task. + /// + /// Note: this covers handle → pipe delivery. Full cancellation of a + /// helper-ROUTED subprocess (helper spawned for a cross-user session) + /// requires a second OS user and is exercised by the cross-user + /// integration tests' environments. + #[tokio::test(flavor = "multi_thread")] + async fn handle_cancel_writes_notify_command_to_helper_pipe() { + let tmp = TempDir::new().unwrap(); + let mut s = Session::new_for_test(tmp.path().to_path_buf()); + let writer_path = tmp.path().join("cancel_writer.log"); + let writer = OpenOptions::new() + .create(true) + .append(true) + .open(&writer_path) + .unwrap(); + s.set_cancel_writer_for_test(writer); + s.set_helper_auth_token_for_test("HandleTok123".into()); + + // Handle taken after the writer is injected, so it dup's the pipe. + let handle = s.cancel_handle(); + let delivered: Arc>> = Arc::new(Mutex::new(None)); + let delivered_clone = delivered.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + *delivered_clone.lock().unwrap() = Some(handle.cancel(None, false)); + }); + + let r = s + .run_task("t", &step("sh", vec!["-c", "sleep 30"]), None, None, None) + .await + .unwrap(); + assert_eq!(r.state, ActionState::Canceled); + assert_eq!(*delivered.lock().unwrap(), Some(true)); + + let msgs = read_cancel_messages(&writer_path); + assert_eq!( + msgs.len(), + 1, + "handle must write exactly one cancel command to the helper pipe" + ); + assert_eq!(msgs[0]["cancel"].as_str().unwrap(), "NOTIFY_THEN_TERMINATE"); + assert_eq!(msgs[0]["notifyPeriodInSeconds"].as_u64().unwrap(), 5); + assert_eq!( + msgs[0]["token"].as_str().unwrap(), + "HandleTok123", + "handle pipe command must carry the helper auth token" + ); + } + + /// Urgent cancel (`time_limit = 0`) through the handle writes TERMINATE. + #[tokio::test(flavor = "multi_thread")] + async fn handle_urgent_cancel_writes_terminate_command_to_helper_pipe() { + let tmp = TempDir::new().unwrap(); + let mut s = Session::new_for_test(tmp.path().to_path_buf()); + let writer_path = tmp.path().join("cancel_writer.log"); + let writer = OpenOptions::new() + .create(true) + .append(true) + .open(&writer_path) + .unwrap(); + s.set_cancel_writer_for_test(writer); + + let handle = s.cancel_handle(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + handle.cancel(Some(Duration::from_secs(0)), false); + }); + + let r = s + .run_task("t", &step("sh", vec!["-c", "sleep 30"]), None, None, None) + .await + .unwrap(); + assert_eq!(r.state, ActionState::Canceled); + + let msgs = read_cancel_messages(&writer_path); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["cancel"].as_str().unwrap(), "TERMINATE"); + assert!( + msgs[0].get("notifyPeriodInSeconds").is_none(), + "TERMINATE should not include notifyPeriodInSeconds" + ); + } } /// Test Option 1: Session-level test for the cancel race condition. @@ -3178,3 +3265,262 @@ async fn test_echo_openjd_directives_true_redacts_redacted_env_in_log() { ); }); } + +// ══════════════════════════════════════════════════════════════ +// SessionCancelHandle — external cancellation while the Session +// is owned by the thread/task driving the action +// ══════════════════════════════════════════════════════════════ + +#[tokio::test] +async fn test_cancel_handle_idle_returns_false() { + let tmp = TempDir::new().unwrap(); + let s = Session::new_for_test(tmp.path().to_path_buf()); + let handle = s.cancel_handle(); + assert!( + !handle.cancel(None, false), + "no action running — cancel must report nothing to cancel" + ); + assert_eq!(s.state(), SessionState::Ready); +} + +#[tokio::test] +async fn test_cancel_handle_cancels_running_action() { + let tmp = TempDir::new().unwrap(); + let mut s = Session::new_for_test(tmp.path().to_path_buf()); + let handle = s.cancel_handle(); + + let delivered: Arc>> = Arc::new(Mutex::new(None)); + let delivered_clone = delivered.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + *delivered_clone.lock().unwrap() = Some(handle.cancel(None, false)); + }); + + let r = s + .run_task("t", &step("sh", vec!["-c", "sleep 30"]), None, None, None) + .await + .unwrap(); + assert_eq!(r.state, ActionState::Canceled); + assert_eq!(s.state(), SessionState::ReadyEnding); + assert_eq!( + *delivered.lock().unwrap(), + Some(true), + "handle must report the cancel as delivered" + ); +} + +#[tokio::test] +async fn test_cancel_handle_mark_action_failed() { + let tmp = TempDir::new().unwrap(); + let mut s = Session::new_for_test(tmp.path().to_path_buf()); + let handle = s.cancel_handle(); + + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + handle.cancel(None, true); + }); + + let r = s + .run_task("t", &step("sh", vec!["-c", "sleep 30"]), None, None, None) + .await + .unwrap(); + assert_eq!( + r.state, + ActionState::Failed, + "mark_action_failed must report the canceled action as Failed" + ); +} + +/// The same handle stays usable across actions, and a cancel does not +/// poison later actions (each action installs a fresh token — unlike the +/// one-shot `SessionConfig::cancel_token`, whose cancellation is permanent). +#[tokio::test] +async fn test_cancel_handle_reusable_across_actions() { + let tmp = TempDir::new().unwrap(); + let mut s = Session::new_for_test(tmp.path().to_path_buf()); + let handle = s.cancel_handle(); + + // Enter two environments (quick onEnter; E1 has a slow onExit). + let e2 = env_with_enter("E2", "sh", vec!["-c", "echo enter2"]); + let e1 = Environment { + name: "E1".into(), + description: None, + script: Some(EnvironmentScript { + let_bindings: None, + actions: EnvironmentActions { + on_enter: Some(action("sh", vec!["-c", "echo enter1"])), + on_exit: Some(action("sh", vec!["-c", "sleep 30"])), + on_wrap_env_enter: None, + on_wrap_task_run: None, + on_wrap_env_exit: None, + }, + embedded_files: None, + }), + variables: None, + resolved_symtab: None, + }; + let id1 = s.enter_environment(&e1, None, None, None).await.unwrap(); + let id2 = s.enter_environment(&e2, None, None, None).await.unwrap(); + + // Action 1: cancel a running task via the handle. + let h = s.cancel_handle(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + h.cancel(None, false); + }); + let r = s + .run_task("t", &step("sh", vec!["-c", "sleep 30"]), None, None, None) + .await + .unwrap(); + assert_eq!(r.state, ActionState::Canceled); + + // Action 2 (E2 exit, quick): must run normally after the cancel — the + // cancel must not have poisoned subsequent actions. + s.exit_environment(&id2, None, true, None).await.unwrap(); + assert_eq!( + s.action_status().unwrap().state, + ActionState::Success, + "action after a handle cancel must not be born-canceled" + ); + + // Action 3 (E1 exit, slow): cancel again via the ORIGINAL handle. + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + handle.cancel(None, false); + }); + let _ = s.exit_environment(&id1, None, true, None).await; + assert_eq!( + s.action_status().unwrap().state, + ActionState::Canceled, + "the same handle must be reusable for later actions" + ); +} + +// ══════════════════════════════════════════════════════════════ +// fail_action_setup — failures after the Running transition must +// produce a terminal Failed status, not wedge the session +// ══════════════════════════════════════════════════════════════ + +/// RFC 0008 wrap seeding resolves the wrapped (inner) action's timeout; +/// an unresolvable expression there previously returned early with the +/// session stuck in Running forever and nothing logged. +#[tokio::test] +async fn test_wrap_seed_failure_reports_failed_action() { + let tmp = TempDir::new().unwrap(); + let statuses: Arc>> = Arc::new(Mutex::new(Vec::new())); + let statuses_clone = statuses.clone(); + let config = SessionConfig { + session_id: "wrap-seed-fail".into(), + job_parameter_values: HashMap::new(), + path_mapping_rules: None, + retain_working_dir: false, + callback: Some(Box::new(move |_sid, status| { + statuses_clone.lock().unwrap().push(status.state); + })), + os_env_vars: None, + session_root_directory: Some(tmp.path().to_path_buf()), + user: None, + profile: None, + cancel_token: None, + debug_collect_stdout: true, + echo_openjd_directives: true, + sticky_bit_policy: openjd_sessions::StickyBitPolicy::Disabled, + }; + let mut s = Session::with_config(config).unwrap(); + + // Wrap environment declaring all three wrap hooks. + let wrap_env = Environment { + name: "WrapEnv".into(), + description: None, + script: Some(EnvironmentScript { + let_bindings: None, + actions: EnvironmentActions { + on_enter: Some(action("sh", vec!["-c", "echo wrap-enter"])), + on_exit: Some(action("sh", vec!["-c", "echo wrap-exit"])), + on_wrap_env_enter: Some(action("sh", vec!["-c", "echo wrapped"])), + on_wrap_task_run: Some(action("sh", vec!["-c", "echo wrapped"])), + on_wrap_env_exit: Some(action("sh", vec!["-c", "echo wrapped"])), + }, + embedded_files: None, + }), + variables: None, + resolved_symtab: None, + }; + s.enter_environment(&wrap_env, None, None, None) + .await + .unwrap(); + + // Inner environment whose onEnter timeout references an undefined + // symbol — seeding WrappedAction.Timeout fails during action setup. + let inner = Environment { + name: "Inner".into(), + description: None, + script: Some(EnvironmentScript { + let_bindings: None, + actions: EnvironmentActions { + on_enter: Some(Action { + command: fs("sh"), + args: Some(vec![fs("-c"), fs("echo inner")]), + timeout: Some(fs("{{ Param.Missing }}")), + cancelation: None, + }), + on_exit: None, + on_wrap_env_enter: None, + on_wrap_task_run: None, + on_wrap_env_exit: None, + }, + embedded_files: None, + }), + variables: None, + resolved_symtab: None, + }; + + let result = s.enter_environment(&inner, None, None, None).await; + assert!(result.is_err(), "setup failure must surface as an error"); + + // The session must NOT be wedged in Running, and the failure must be + // observable through the action status and the callback. + assert_eq!(s.state(), SessionState::ReadyEnding); + let status = s.action_status().expect("terminal action status"); + assert_eq!(status.state, ActionState::Failed); + let msg = status.fail_message.expect("fail_message must be recorded"); + assert!( + msg.contains("timeout"), + "fail_message should identify the failing field, got: {msg}" + ); + assert!( + statuses.lock().unwrap().contains(&ActionState::Failed), + "callback must observe the Failed transition" + ); +} + +/// A runner error during dispatch (here: an undefined symbol in the action's +/// args, which fails command-line resolution inside the runner) must surface +/// through drive_action's error branch with a recorded `fail_message` — not +/// as a bare Failed status with no message and nothing logged, and not as a +/// session wedged in Running. +#[tokio::test] +async fn test_runner_resolution_error_records_fail_message() { + let tmp = TempDir::new().unwrap(); + let mut s = Session::new_for_test(tmp.path().to_path_buf()); + + let script = step("sh", vec!["-c", "echo {{ No.Such.Symbol }}"]); + let result = s.run_task("t", &script, None, None, None).await; + assert!( + result.is_err(), + "resolution failure must surface as an error" + ); + + assert_eq!( + s.state(), + SessionState::ReadyEnding, + "session must not be wedged in Running" + ); + let status = s.action_status().expect("terminal action status"); + assert_eq!(status.state, ActionState::Failed); + let msg = status.fail_message.expect("fail_message must be recorded"); + assert!( + msg.contains("No.Such.Symbol"), + "fail_message should identify the unresolvable symbol, got: {msg}" + ); +} diff --git a/crates/openjd-sessions/tests/integration/test_session_env_step.rs b/crates/openjd-sessions/tests/integration/test_session_env_step.rs index f2ab75af..c8456c08 100644 --- a/crates/openjd-sessions/tests/integration/test_session_env_step.rs +++ b/crates/openjd-sessions/tests/integration/test_session_env_step.rs @@ -174,7 +174,7 @@ async fn test_env_with_embedded_files() { embedded_files: Some(vec![EmbeddedFile { name: "Script".to_string(), file_type: openjd_model::types::FileType::Text, - filename: Some(fs("script.txt")), + filename: Some("script.txt".to_string()), data: Some(fs("file content here")), runnable: None, end_of_line: None, @@ -535,7 +535,7 @@ async fn test_env_with_let_bindings_and_embedded_files() { embedded_files: Some(vec![EmbeddedFile { name: "Config".to_string(), file_type: openjd_model::types::FileType::Text, - filename: Some(fs("config.txt")), + filename: Some("config.txt".to_string()), data: Some(fs("config data")), runnable: None, end_of_line: None, @@ -576,7 +576,7 @@ async fn test_step_with_let_bindings_and_embedded_files() { embedded_files: Some(vec![EmbeddedFile { name: "Data".to_string(), file_type: openjd_model::types::FileType::Text, - filename: Some(fs("data.txt")), + filename: Some("data.txt".to_string()), data: Some(fs("{{ greeting }}")), runnable: None, end_of_line: None, diff --git a/crates/openjd-sessions/tests/scenarios/let_bindings/let_bindings_scenario.yaml b/crates/openjd-sessions/tests/scenarios/let_bindings/let_bindings_scenario.yaml index ab113339..96924f57 100644 --- a/crates/openjd-sessions/tests/scenarios/let_bindings/let_bindings_scenario.yaml +++ b/crates/openjd-sessions/tests/scenarios/let_bindings/let_bindings_scenario.yaml @@ -63,6 +63,7 @@ expect: - "step_first_file_stem=file1" - 'step_file_stems=["file1", "file2", "file3"]' # Task-level bindings for frame 1 (step_int//10 = 2, so frames 1-2) + - "task_cfg_stem=config" - "task_frame=1" - "task_combined=21" - "task_output_name=hello_world_1.exr" diff --git a/crates/openjd-sessions/tests/scenarios/let_bindings/let_bindings_template.yaml b/crates/openjd-sessions/tests/scenarios/let_bindings/let_bindings_template.yaml index 60c3cd62..0395751d 100644 --- a/crates/openjd-sessions/tests/scenarios/let_bindings/let_bindings_template.yaml +++ b/crates/openjd-sessions/tests/scenarios/let_bindings/let_bindings_template.yaml @@ -126,11 +126,14 @@ steps: - task_combined = step_int + Task.Param.Frame - task_output_name = step_string + "_" + string(Task.Param.Frame) + ".exr" - task_msg = "Frame " + string(Task.Param.Frame) + " of " + string(step_int // 10) + # Task.File.* is available to script-level lets: file paths are allocated before let evaluation + - task_cfg_stem = Task.File.config.stem # Embedded file using let binding values embeddedFiles: - name: config type: TEXT - filename: "config_{{task_frame}}.txt" + # filename is a plain string (not @fmtstring) per the 2023-09 schema + filename: "config.txt" data: | # Config for frame {{task_frame}} output_name={{task_output_name}} @@ -143,6 +146,7 @@ steps: - | echo "=== Embedded file content ===" cat {{repr_sh(Task.File.config)}} + echo task_cfg_stem={{repr_sh(task_cfg_stem)}} echo "=== Step-level bindings ===" echo step_int={{repr_sh(string(step_int))}} echo step_float={{repr_sh(string(step_float))}} diff --git a/specs/cli/run.md b/specs/cli/run.md index 81be45e6..53d5544b 100644 --- a/specs/cli/run.md +++ b/specs/cli/run.md @@ -236,8 +236,8 @@ differs from the CLI's iteration order. For each step, embedded file paths are pre-computed before task iteration: -1. Filenames are resolved against the step symbol table (format strings may reference - step-level variables) +1. Filenames are plain strings per the 2023-09 schema (not `@fmtstring`) and + are used literally 2. File paths are constructed relative to the session working directory 3. For each task, file contents are resolved against the task symbol table and written via `openjd_sessions::embedded_files::write_embedded_file()` diff --git a/specs/model/job-types.md b/specs/model/job-types.md index 87d40b4d..803a1cc6 100644 --- a/specs/model/job-types.md +++ b/specs/model/job-types.md @@ -159,7 +159,7 @@ included instead. pub struct EmbeddedFile { pub name: String, pub file_type: FileType, // Typed enum, not String - pub filename: Option, // Session/task-scope + pub filename: Option, // Plain string (not @fmtstring) pub data: Option, // Session/task-scope pub runnable: Option, pub end_of_line: Option, // Typed enum, not String diff --git a/specs/model/public-api.md b/specs/model/public-api.md index 975e8d53..0097d1eb 100644 --- a/specs/model/public-api.md +++ b/specs/model/public-api.md @@ -556,7 +556,7 @@ pub struct job::EnvironmentActions { pub struct job::EmbeddedFile { pub name: String, pub file_type: FileType, - pub filename: Option, + pub filename: Option, pub data: Option, pub runnable: Option, pub end_of_line: Option, diff --git a/specs/model/template-types.md b/specs/model/template-types.md index 814be367..8f83755a 100644 --- a/specs/model/template-types.md +++ b/specs/model/template-types.md @@ -129,7 +129,7 @@ pub struct EnvironmentScript { pub struct EmbeddedFile { pub name: String, pub file_type: String, // "type" field; must be "TEXT" - pub filename: Option, + pub filename: Option, // Plain string (not @fmtstring) pub data: Option, pub runnable: Option, pub end_of_line: Option, // FEATURE_BUNDLE_1: "LF", "CRLF", "AUTO" diff --git a/specs/model/validation.md b/specs/model/validation.md index 8a4f899f..497fa69d 100644 --- a/specs/model/validation.md +++ b/specs/model/validation.md @@ -63,7 +63,9 @@ checks have nothing to walk): **template scope** instead — no `Session.*`/`Env.File.*`. `let` bindings require EXPR; with EXPR they are validated and type-checked into the symbol table. Complex expressions (anything beyond a bare `{{Name.Path}}` reference) require EXPR in - variables, action commands/args, and embedded-file data/filename. + variables, action commands/args, and embedded-file data. Embedded-file + `filename` is a plain string (not `@fmtstring`) — brace syntax in it is + literal text and no format-string validation applies. - **Pass 10** — WRAP_ACTIONS gating (see below). ## Pass 5: Limits Enforcement diff --git a/specs/sessions/embedded-files.md b/specs/sessions/embedded-files.md index 226c6404..3fedf93f 100644 --- a/specs/sessions/embedded-files.md +++ b/specs/sessions/embedded-files.md @@ -66,7 +66,8 @@ use `Env.File.*`, step-scoped files use `Task.File.*`. For each embedded file: 1. Determine the file path: - - If `filename` is specified: resolve its format string, validate it as a + - If `filename` is specified: it is a plain string per the 2023-09 schema + (not `@fmtstring`) — use it literally, validate it as a safe single path component (see [Filename path-traversal defense-in-depth](#filename-path-traversal-defense-in-depth)), then join with `files_directory`. - Otherwise: `files_directory / {random_hex}` (hash-based name for uniqueness) @@ -153,14 +154,12 @@ that produce multi-line strings get consistent line endings. ## Integration with Runners -Environment scripts use the full two-phase flow: +Environment and step scripts both use the full two-phase flow: ``` allocate_file_paths() → evaluate_let_bindings() → write_file_contents() ``` -Step scripts use a simplified flow (let bindings evaluated first): -``` -evaluate_let_bindings() → allocate_file_paths() + write_file_contents() -``` - -See [runners.md](runners.md) for why the ordering differs. +This is what makes `Env.File.*` / `Task.File.*` available to `let` bindings +while letting file `data` reference let-bound values. It is only possible +because `filename` is a plain string (2023-09 schema, not `@fmtstring`), so +path allocation never depends on `let` values. diff --git a/specs/sessions/public-api.md b/specs/sessions/public-api.md index 586ce828..0af53cb0 100644 --- a/specs/sessions/public-api.md +++ b/specs/sessions/public-api.md @@ -199,6 +199,11 @@ impl Session { mark_action_failed: bool, ) -> Result<(), SessionError>; + /// Thread-safe handle for cancelling the running action while the + /// `Session` value is owned by the thread driving it. Reusable for + /// the life of the session; see session.md § SessionCancelHandle. + pub fn cancel_handle(&self) -> SessionCancelHandle; + /// Remove the working directory and associated temp files. /// Called automatically on drop as a safety net, but callers /// should call it explicitly at the end of a session to surface diff --git a/specs/sessions/runners.md b/specs/sessions/runners.md index d05eb5ad..79602717 100644 --- a/specs/sessions/runners.md +++ b/specs/sessions/runners.md @@ -193,21 +193,22 @@ pub async fn run( The ordering differs from environment scripts: -1. Evaluate let bindings first (they can reference `Task.Param.*` but not `Task.File.*` - since step-level let bindings are evaluated before embedded files) -2. Materialize embedded files (allocate paths + write contents, using the let-binding- - enriched symbol table) -3. Resolve action args and run subprocess - -### Why the ordering differs from EnvironmentScriptRunner - -In the spec, `StepScript.let` bindings are scoped to the step script and are evaluated -before embedded files. This means let bindings can't reference `Task.File.*` paths -(unlike environment scripts where let bindings can reference `Env.File.*`). The simpler -ordering reflects this: let bindings first, then files, then action. - -The Python library follows the same ordering distinction between environment and step -scripts. +1. Allocate embedded file paths (registers `Task.File.*` in the symbol table; + `filename` is a plain string per the 2023-09 schema, so allocation has no + dependency on let bindings) +2. Evaluate let bindings (they can reference `Task.Param.*` and `Task.File.*`) +3. Write embedded file contents (using the let-binding-enriched symbol table) +4. Resolve action args and run subprocess + +### Ordering matches EnvironmentScriptRunner + +Step scripts use the same two-phase flow as environment scripts: paths first, +then let bindings, then file contents. `Task.File.*` is therefore available to +script-level `let` bindings, mirroring `Env.File.*` in environment scripts — +and matching decode-time validation, which seeds `Task.File.*` into the +let-binding scope. (An earlier revision evaluated step let bindings before +file allocation, which made `Task.File.*` in a `let` validate but fail at +runtime.) ## resolve_action_timeout diff --git a/specs/sessions/session.md b/specs/sessions/session.md index 0c700061..21dca446 100644 --- a/specs/sessions/session.md +++ b/specs/sessions/session.md @@ -319,6 +319,29 @@ Unset takes precedence over set within the same environment, matching the spec. The `time_limit` parameter sets a deadline for the cancelation to complete. If the subprocess doesn't exit within the limit, it's forcefully killed. +### SessionCancelHandle + +`cancel_action` requires `&mut Session`, which is unavailable while an action-driving +thread owns the `Session` (the embedding pattern used by the Python bindings). For that +case, `Session::cancel_handle()` returns a `SessionCancelHandle` — a thread-safe, +reusable handle over the shared per-action cancel state (token, cancel-request watch +channel, `mark_failed` flag, all behind one `Arc>`). + +`SessionCancelHandle::cancel(time_limit, mark_action_failed) -> bool`: + +- Returns `false` if no action is running (no per-action token installed). +- Otherwise delivers exactly what `cancel_action` delivers — the helper pipe cancel + command (cross-user sessions), the `time_limit` over the watch channel, and the + token cancel — while holding the shared lock so the target action cannot change + mid-delivery. +- Unlike `cancel_action`, it cannot set the session's transient `Canceling` state + (the `Session` is owned elsewhere); the state moves directly to the terminal value + when the canceled action finishes. + +Every action installs fresh cancel state, so a cancel never poisons later actions +(contrast with `SessionConfig::cancel_token`, whose cancellation is permanent) and +the same handle works for the life of the session. + ## Cleanup See [Session Lifecycle Contract](#session-lifecycle-contract) for the full cleanup