From aff977c55a115e0fcf45287ea2643be5a808b5d9 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Sun, 6 Sep 2026 18:30:43 +0200 Subject: [PATCH] feat: require exact Codex residency resume agent-identity: mbp2025.direct.omp.bn5hrz39 agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.1.7 agent-runtime: OMP 18.1.7 tooling-profile: dotfiles@4816a7a --- src/codex_app_server.rs | 367 ++++++++++++++++++++++++++++++++++++++-- src/main.rs | 37 +++- 2 files changed, 389 insertions(+), 15 deletions(-) diff --git a/src/codex_app_server.rs b/src/codex_app_server.rs index 5e1f217f..7f7ddf92 100644 --- a/src/codex_app_server.rs +++ b/src/codex_app_server.rs @@ -93,6 +93,8 @@ const CLASSIFIED_CODEX_THREAD_ITEMS: &[&str] = &[ const RUNTIME_SCHEMA: &str = "st2.codex-runtime.v1"; const BINDING_SCHEMA: &str = "st2.codex-thread-binding.v1"; const CONTROL_STATE_SCHEMA: &str = "st2.codex-control-state.v1"; +const RESIDENCY_CHECKPOINT_SCHEMA: &str = "st2.codex-residency-checkpoint.v1"; +const RESIDENCY_CHECKPOINT_FILE: &str = "residency-checkpoint.json"; const WRAPPER_DIAGNOSTIC_SCHEMA: &str = "st2.codex-wrapper-diagnostic.v1"; const CONTROL_TUI_LOADED_REQUEST_ID: u64 = 0; const CONTROL_SUBSCRIBE_REQUEST_ID: u64 = 1; @@ -210,6 +212,21 @@ impl CodexThreadBinding { } } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct CodexResidencyCheckpoint { + schema: String, + source_generation: crate::residency::Generation, + resume_generation: crate::residency::Generation, + binding: CodexThreadBinding, +} + +impl CodexResidencyCheckpoint { + pub fn thread_id(&self) -> &str { + self.binding.thread_id() + } +} + /// The latest delivery-relevant state observed on the bound app-server control stream. /// /// `Active` permits `turn/steer`: its turn ID came from the latest unmatched `turn/started` event. @@ -1720,11 +1737,56 @@ pub fn run_controlled( identity: String, runtime_id: String, codex_argv: Vec, +) -> Result<()> { + run_controlled_with_required_resume(catalog_root, identity, runtime_id, codex_argv, None) +} + +/// Run a cold residency generation with an exact checkpointed native thread. +pub fn run_controlled_residency( + catalog_root: &Path, + identity: String, + runtime_id: String, + codex_argv: Vec, + resume_generation: crate::residency::Generation, +) -> Result<()> { + run_controlled_with_required_resume( + catalog_root, + identity, + runtime_id, + codex_argv, + Some(resume_generation), + ) +} + +fn run_controlled_with_required_resume( + catalog_root: &Path, + identity: String, + runtime_id: String, + codex_argv: Vec, + required_resume_generation: Option, ) -> Result<()> { anyhow::ensure!( !codex_argv.is_empty(), "Codex controlled launch argv is empty" ); + // Install before any protocol preflight child exists. A SIGTERM in the preflight window must + // set the stop flag instead of leaving a detached app server and stale socket behind. + crate::provider_session::install_signal_handler(); + let state_dir = state_dir(catalog_root, &identity); + secure_dir(&state_dir)?; + let _owner_lock = acquire_owner_lock(&state_dir)?; + let binding_path = state_dir.join("binding.json"); + let resume_thread = match required_resume_generation { + Some(generation) => Some(required_residency_resume( + &state_dir, + &identity, + &runtime_id, + generation, + &codex_argv[1..], + )?), + None => load_resume_thread(&binding_path, &identity, &runtime_id)?, + }; + let mut delivery = CodexDeliveryConfig::resolve(catalog_root, &identity)?; match ensure_supported_protocol(&codex_argv[0]) { // The admitted version is the one fact the gate learns that outlives it: the diagnostic @@ -1736,9 +1798,6 @@ pub fn run_controlled( } } - let state_dir = state_dir(catalog_root, &identity); - secure_dir(&state_dir)?; - let _owner_lock = acquire_owner_lock(&state_dir)?; let mut diagnostics = WrapperDiagnostics::open(&state_dir, &identity, &runtime_id)?; diagnostics.record("ownerAcquired", json!({}))?; @@ -1749,6 +1808,7 @@ pub fn run_controlled( runtime_id, codex_argv, delivery, + resume_thread, &mut diagnostics, ); match result { @@ -1777,16 +1837,9 @@ fn run_controlled_owned( runtime_id: String, codex_argv: Vec, delivery: CodexDeliveryConfig, + resume_thread: Option, diagnostics: &mut WrapperDiagnostics, ) -> Result<()> { - // Installed before ANY child exists — the hook-trust preflight spawns a detached app-server - // first, and a SIGTERM landing in that window must set the stop flag its connect loop polls - // rather than killing this wrapper around a leaked server and a stale socket. (Installing - // resets the flag, so this must also run exactly once per launch.) - crate::provider_session::install_signal_handler(); - let binding_path = state_dir.join("binding.json"); - let resume_thread = load_resume_thread(&binding_path, &identity, &runtime_id)?; - let socket_path = socket_path(catalog_root, &identity)?; let socket_dir = socket_path .parent() @@ -4003,7 +4056,85 @@ pub fn load_current_control_state( Ok(Some(state)) } -fn load_resume_thread(path: &Path, agent: &str, runtime_id: &str) -> Result> { +pub fn checkpoint_residency( + state_dir: &Path, + agent: &str, + runtime_id: &str, + source_generation: crate::residency::Generation, + resume_generation: crate::residency::Generation, +) -> Result { + anyhow::ensure!( + source_generation.0.checked_add(1) == Some(resume_generation.0), + "Codex residency checkpoint generation is not monotonic" + ); + let binding = load_thread_binding(&state_dir.join("binding.json"), agent, runtime_id)? + .with_context(|| format!("Codex runtime {runtime_id:?} has no native thread binding"))?; + let checkpoint = CodexResidencyCheckpoint { + schema: RESIDENCY_CHECKPOINT_SCHEMA.to_owned(), + source_generation, + resume_generation, + binding, + }; + atomic_json(&state_dir.join(RESIDENCY_CHECKPOINT_FILE), &checkpoint)?; + Ok(checkpoint) +} + +pub fn required_residency_resume( + state_dir: &Path, + agent: &str, + runtime_id: &str, + resume_generation: crate::residency::Generation, + authored_args: &[String], +) -> Result { + anyhow::ensure!( + resume_insertion_index(authored_args)?.is_some(), + "authored Codex session selection conflicts with mandatory residency resume" + ); + let checkpoint = load_residency_checkpoint(state_dir, agent, runtime_id, resume_generation)?; + let current = load_thread_binding(&state_dir.join("binding.json"), agent, runtime_id)? + .with_context(|| format!("Codex runtime {runtime_id:?} has no native thread binding"))?; + anyhow::ensure!( + current == checkpoint.binding, + "Codex native thread binding changed after residency checkpoint" + ); + Ok(current.thread_id) +} + +fn load_residency_checkpoint( + state_dir: &Path, + agent: &str, + runtime_id: &str, + resume_generation: crate::residency::Generation, +) -> Result { + let path = state_dir.join(RESIDENCY_CHECKPOINT_FILE); + let bytes = fs::read(&path) + .with_context(|| format!("reading Codex residency checkpoint {}", path.display()))?; + let checkpoint: CodexResidencyCheckpoint = serde_json::from_slice(&bytes)?; + anyhow::ensure!( + checkpoint.schema == RESIDENCY_CHECKPOINT_SCHEMA, + "unsupported Codex residency checkpoint schema" + ); + anyhow::ensure!( + checkpoint.resume_generation == resume_generation + && checkpoint.source_generation.0.checked_add(1) == Some(resume_generation.0), + "Codex residency checkpoint belongs to a different generation" + ); + anyhow::ensure!( + checkpoint.binding.agent == agent && checkpoint.binding.runtime_id == runtime_id, + "Codex residency checkpoint belongs to a different agent runtime" + ); + anyhow::ensure!( + !checkpoint.binding.thread_id.is_empty(), + "Codex residency checkpoint has an empty native thread id" + ); + Ok(checkpoint) +} + +fn load_thread_binding( + path: &Path, + agent: &str, + runtime_id: &str, +) -> Result> { let bytes = match fs::read(path) { Ok(bytes) => bytes, Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), @@ -4022,7 +4153,11 @@ fn load_resume_thread(path: &Path, agent: &str, runtime_id: &str) -> Result Result> { + Ok(load_thread_binding(path, agent, runtime_id)?.map(|binding| binding.thread_id)) } fn random_token() -> Result { @@ -6958,6 +7093,212 @@ mod tests { assert!(error.to_string().contains("different runtime incarnation")); } + #[test] + fn residency_checkpoint_requires_and_preserves_the_exact_bound_thread() { + let tmp = tempfile::tempdir().unwrap(); + let state = tmp.path().join("state"); + let runtime = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + atomic_json( + &state.join("binding.json"), + &CodexThreadBinding::new(&runtime, "thread-prior".into()), + ) + .unwrap(); + + let checkpoint = checkpoint_residency( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(1), + crate::residency::Generation(2), + ) + .unwrap(); + assert_eq!(checkpoint.thread_id(), "thread-prior"); + assert_eq!( + required_residency_resume( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + &["--model".into(), "gpt-test".into(), "boot".into()], + ) + .unwrap(), + "thread-prior" + ); + + atomic_json( + &state.join("binding.json"), + &CodexThreadBinding::new(&runtime, "thread-replaced".into()), + ) + .unwrap(); + let error = required_residency_resume( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + &["boot".into()], + ) + .unwrap_err(); + assert!( + error + .to_string() + .contains("changed after residency checkpoint") + ); + } + + #[test] + fn mandatory_residency_resume_refuses_missing_or_authored_session_selection() { + let tmp = tempfile::tempdir().unwrap(); + let state = tmp.path().join("state"); + let error = checkpoint_residency( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(1), + crate::residency::Generation(2), + ) + .unwrap_err(); + assert!(error.to_string().contains("has no native thread binding")); + + let runtime = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + atomic_json( + &state.join("binding.json"), + &CodexThreadBinding::new(&runtime, "thread-prior".into()), + ) + .unwrap(); + checkpoint_residency( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(1), + crate::residency::Generation(2), + ) + .unwrap(); + for authored in [ + vec!["resume".into(), "thread-other".into()], + vec!["fork".into(), "thread-other".into()], + ] { + let error = required_residency_resume( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + &authored, + ) + .unwrap_err(); + assert!( + error + .to_string() + .contains("authored Codex session selection") + ); + } + } + + #[test] + fn residency_resume_rejects_missing_corrupt_foreign_and_stale_checkpoints() { + let tmp = tempfile::tempdir().unwrap(); + let state = tmp.path().join("state"); + let runtime = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + atomic_json( + &state.join("binding.json"), + &CodexThreadBinding::new(&runtime, "thread-prior".into()), + ) + .unwrap(); + + let missing = required_residency_resume( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + &["boot".into()], + ) + .unwrap_err(); + assert!(missing.to_string().contains("residency checkpoint")); + + checkpoint_residency( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(1), + crate::residency::Generation(2), + ) + .unwrap(); + let stale = required_residency_resume( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(3), + &["boot".into()], + ) + .unwrap_err(); + assert!(stale.to_string().contains("different generation")); + + let foreign_runtime = CodexRuntime::fresh("h.other".into(), "h.other".into()).unwrap(); + atomic_json( + &state.join(RESIDENCY_CHECKPOINT_FILE), + &CodexResidencyCheckpoint { + schema: RESIDENCY_CHECKPOINT_SCHEMA.into(), + source_generation: crate::residency::Generation(1), + resume_generation: crate::residency::Generation(2), + binding: CodexThreadBinding::new(&foreign_runtime, "thread-prior".into()), + }, + ) + .unwrap(); + let foreign = required_residency_resume( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + &["boot".into()], + ) + .unwrap_err(); + assert!(foreign.to_string().contains("different agent runtime")); + + fs::write(state.join(RESIDENCY_CHECKPOINT_FILE), b"{").unwrap(); + assert!( + required_residency_resume( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + &["boot".into()], + ) + .is_err(), + "a corrupt residency checkpoint was accepted" + ); + } + + #[test] + fn mandatory_residency_failure_does_not_spawn_a_provider_child() { + let _stop_exclusive = stop_flag_tests(); + let tmp = tempfile::tempdir().unwrap(); + let marker = tmp.path().join("provider-started"); + let codex = tmp.path().join("codex"); + fs::write( + &codex, + format!("#!/bin/sh\ntouch '{}'\nexit 1\n", marker.display()), + ) + .unwrap(); + fs::set_permissions(&codex, fs::Permissions::from_mode(0o755)).unwrap(); + + let error = run_controlled_residency( + tmp.path(), + "h.worker".into(), + "h.worker".into(), + vec![codex.display().to_string(), "boot".into()], + crate::residency::Generation(2), + ) + .unwrap_err(); + + assert!( + error.to_string().contains("residency checkpoint"), + "{error:#}" + ); + assert!( + !marker.exists(), + "a provider process started before mandatory resume validation" + ); + } + #[test] fn watcher_holds_without_an_exact_turn_and_tracks_one_unmatched_lifecycle() { let runtime = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); diff --git a/src/main.rs b/src/main.rs index a879ae65..a8b46c40 100644 --- a/src/main.rs +++ b/src/main.rs @@ -137,6 +137,9 @@ enum Command { /// Exact reconciled PTY task identity for this runtime. #[arg(long)] runtime_id: String, + /// Internal cold-residency generation that must resume its exact checkpointed thread. + #[arg(long, hide = true)] + required_resume_generation: Option, /// Original structured Codex invocation, including its provider executable. #[arg(required = true, trailing_var_arg = true, allow_hyphen_values = true)] codex_argv: Vec, @@ -343,6 +346,9 @@ enum DriverCmd { identity: String, #[arg(long)] runtime_id: String, + /// Internal cold-residency generation that must resume its exact checkpointed thread. + #[arg(long, hide = true)] + required_resume_generation: Option, #[arg(required = true, trailing_var_arg = true, allow_hyphen_values = true)] argv: Vec, }, @@ -1325,11 +1331,26 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< Command::CodexAppServer { identity, runtime_id, + required_resume_generation, codex_argv, } => { let catalog = catalog_arg(None)?; let catalog = catalog.canonicalize().unwrap_or(catalog); - st2::codex_app_server::run_controlled(&catalog, identity, runtime_id, codex_argv) + match required_resume_generation { + Some(generation) => st2::codex_app_server::run_controlled_residency( + &catalog, + identity, + runtime_id, + codex_argv, + st2::residency::Generation(generation), + ), + None => st2::codex_app_server::run_controlled( + &catalog, + identity, + runtime_id, + codex_argv, + ), + } } Command::ClaudeMcp { identity } => { let catalog = catalog_arg(None)?; @@ -1339,11 +1360,23 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< Command::Driver(DriverCmd::Codex { identity, runtime_id, + required_resume_generation, argv, }) => { let catalog = catalog_arg(None)?; let catalog = catalog.canonicalize().unwrap_or(catalog); - st2::codex_app_server::run_controlled(&catalog, identity, runtime_id, argv) + match required_resume_generation { + Some(generation) => st2::codex_app_server::run_controlled_residency( + &catalog, + identity, + runtime_id, + argv, + st2::residency::Generation(generation), + ), + None => { + st2::codex_app_server::run_controlled(&catalog, identity, runtime_id, argv) + } + } } Command::Driver(DriverCmd::PiChannel { identity }) => { let catalog = catalog_arg(None)?;