diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index c7cee722c59..1e445b261d1 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -426,7 +426,11 @@ impl AcpxProviderDescriptor { grok_disable_web_search: self.grok_disable_web_search.clone(), permission_mode_pinned: self.permission_mode_pinned, provider_policy: self.provider_policy.clone(), - system_instructions: self.instructions.clone(), + system_instructions: crate::task_memory::instructions_with_memory_note( + &self.instructions, + crate::acpx_provider_session::MAX_SYSTEM_INSTRUCTIONS_BYTES, + ) + .into_owned(), runtime_context: self.runtime_context.clone(), tool_set: provider_tool_set, expected_identity, @@ -3958,6 +3962,26 @@ mod tests { session_config.system_instructions, descriptor_value["instructions"] ); + // The note reaches the provider without changing the durable descriptor. + let task_cgroup = directory.join("tasks").join("task-a"); + fs::create_dir_all(&task_cgroup).unwrap(); + fs::write(directory.join("tasks").join("memory.max"), "6442450944\n").unwrap(); + crate::task_memory::TEST_TASK_CGROUP.with(|path| *path.borrow_mut() = Some(task_cgroup)); + let session_config = refreshed + .session_config( + attached.state.as_ref().unwrap().tool_set.clone(), + attached.state.as_ref().unwrap().identity.clone(), + Some(&launch_profile), + ) + .unwrap(); + crate::task_memory::TEST_TASK_CGROUP.with(|path| *path.borrow_mut() = None); + assert!(session_config + .system_instructions + .starts_with(descriptor_value["instructions"].as_str().unwrap())); + assert!(session_config + .system_instructions + .contains("6 GiB memory limit")); + assert_eq!(refreshed.instructions, descriptor_value["instructions"]); // Both persistence and the sidecar launch config receive the new grant. let persisted: AcpxDurableState = serde_json::from_slice(&fs::read(attached.state_path()).unwrap()).unwrap(); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs index 943f3b78aa8..d0597a74313 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs @@ -24,7 +24,7 @@ use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE const MAX_ID_CHARS: usize = 240; const MAX_MODEL_CHARS: usize = 240; -const MAX_SYSTEM_INSTRUCTIONS_BYTES: usize = 1024 * 1024; +pub(crate) const MAX_SYSTEM_INSTRUCTIONS_BYTES: usize = 1024 * 1024; const MAX_JSON_SAFE_INTEGER: u64 = 9_007_199_254_740_991; #[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index 13d237e3276..0362b8a554a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -495,6 +495,13 @@ fn apply_oom_score_adj(args: &[String]) -> Result<(), LocalRunnerError> { Ok(()) } +fn apply_task_memory_cgroup(args: &[String]) -> Result<(), LocalRunnerError> { + if let Some(path) = optional_value(args, "--task-memory-cgroup")? { + paperclip_runner_core::task_memory::set_task_cgroup(PathBuf::from(path)); + } + Ok(()) +} + fn run(args: &[String]) -> Result<(), LocalRunnerError> { #[cfg(unix)] if matches!( @@ -515,6 +522,7 @@ fn run(args: &[String]) -> Result<(), LocalRunnerError> { return Ok(()); } apply_oom_score_adj(args)?; + apply_task_memory_cgroup(args)?; if args.iter().any(|argument| argument == "--connect-url") || args.iter().any(|argument| argument == "--listen-address") || args.iter().any(|argument| argument == "--listen-port") diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs index 33449c81a66..fefedb0501a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs @@ -1254,7 +1254,13 @@ impl CodexProvider { } else { "baseInstructions" }; - params_object.insert(instruction_field.to_owned(), json!(config.instructions)); + params_object.insert( + instruction_field.to_owned(), + json!(crate::task_memory::instructions_with_memory_note( + &config.instructions, + MAX_INSTRUCTIONS_BYTES, + )), + ); if provider.permission_profile == "paperclip-runner-external-sandbox" { // The execution target (for example Daytona) is the OS sandbox. // Codex must not try to create nested user/network namespaces, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index 13a61f31271..ddfd4034c94 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -476,6 +476,7 @@ pub fn run_durable_runner( #[cfg(unix)] let mut notifications = crate::runtime_notifications::RuntimeNotifications::bind(&config.state_dir).ok(); + let mut task_memory = crate::task_memory::TaskMemoryMonitor::new(); // Bind listener mode or resolve dial mode before processing commands. Dial // reconnects retain the same validated addresses so DNS cannot redirect a // retry after the trust decision. @@ -887,6 +888,9 @@ pub fn run_durable_runner( if let Some(notifications) = notifications.as_mut() { notifications.poll(&config, &mut state, &store); } + if let Some(task_memory) = task_memory.as_mut() { + task_memory.poll(&config, &mut state, &store); + } let now = current_unix_ms()?; if welcome.lease_renewal_version == Some(1) && now >= next_lease_renewal { // A failed write may still have reached the controller. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs index df2670ec9ea..6f859e098b6 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs @@ -38,6 +38,7 @@ pub mod question_response; pub mod replay; mod stable_identity; pub mod stdio_mcp; +pub mod task_memory; use std::error::Error; use std::fmt::{self, Display, Formatter}; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/task_memory.rs b/packages/paperclip-runner/runner/crates/runner-core/src/task_memory.rs new file mode 100644 index 00000000000..5279f4f3fef --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/task_memory.rs @@ -0,0 +1,468 @@ +//! Memory limit that a launcher applies to the processes Runner supervises. +//! +//! The kernel's cgroup files are the only source of these facts, so an environment +//! that passes no flag gets no note and no notices. + +use crate::durable::{ + DurableRunnerConfig, DurableRunnerError, DurableState, DurableStateStore, EventPriority, +}; +use notify::{RecommendedWatcher, RecursiveMode, Watcher}; +use serde_json::json; +use std::borrow::Cow; +use std::fs; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, OnceLock}; + +const MAX_NOTICES: u32 = 64; +const BYTES_PER_GIB: u64 = 1 << 30; + +static TASK_CGROUP: OnceLock = OnceLock::new(); + +// Lib tests share a process, so a process-wide path would reach unrelated tests. +#[cfg(test)] +thread_local! { + pub(crate) static TEST_TASK_CGROUP: std::cell::RefCell> = const { + std::cell::RefCell::new(None) + }; +} + +/// Set once at startup from the launch flag. +pub fn set_task_cgroup(path: PathBuf) { + let _ = TASK_CGROUP.set(path); +} + +/// The limit on the parent cgroup, which tasks share. "max" means no limit. +fn read_limit(task_cgroup: &Path) -> Option { + let content = fs::read_to_string(task_cgroup.parent()?.join("memory.max")).ok()?; + content.trim().parse().ok().filter(|limit| *limit > 0) +} + +/// The kernel counts an out-of-memory kill in the victim's cgroup, so this is +/// specific to the task even though the limit is shared. +fn read_kills(task_cgroup: &Path) -> Option { + fs::read_to_string(task_cgroup.join("memory.events")) + .ok()? + .lines() + .find_map(|line| line.strip_prefix("oom_kill "))? + .trim() + .parse() + .ok() +} + +fn format_limit(bytes: u64) -> String { + if bytes.is_multiple_of(BYTES_PER_GIB) { + format!("{} GiB", bytes / BYTES_PER_GIB) + } else { + format!("{:.1} GiB", bytes as f64 / BYTES_PER_GIB as f64) + } +} + +fn memory_note(limit_bytes: u64, task_cgroup: &Path) -> Option { + let shared = task_cgroup.parent()?; + Some(format!( + "All task processes on this computer share a {limit} memory limit, and memory used by a \ + concurrent task counts toward it. Above the limit, the kernel kills a process with \ + SIGKILL, usually the largest, and the shell reports exit code 137. Other causes can also \ + produce that exit code. To check for a memory kill, read the oom_kill count in \ + {task}/memory.events. Read \ + {shared}/memory.current and {shared}/memory.peak for the shared use. After such a kill, \ + the same command will probably fail again, so reduce its memory use or split the work. \ + If the work cannot fit, report that as a blocker.", + limit = format_limit(limit_bytes), + task = task_cgroup.display(), + shared = shared.display(), + )) +} + +fn with_note_from<'a>(task_cgroup: &Path, instructions: &'a str, max_bytes: usize) -> Cow<'a, str> { + let Some(note) = read_limit(task_cgroup).and_then(|limit| memory_note(limit, task_cgroup)) + else { + return Cow::Borrowed(instructions); + }; + let separator = if instructions.is_empty() { "" } else { "\n\n" }; + // Providers reject oversized instructions, and the note is not worth a failed launch. + if instructions.len() + separator.len() + note.len() > max_bytes { + return Cow::Borrowed(instructions); + } + Cow::Owned(format!("{instructions}{separator}{note}")) +} + +/// Adds the limit to instructions for a provider whose tools run under it. +pub fn instructions_with_memory_note(instructions: &str, max_bytes: usize) -> Cow<'_, str> { + #[cfg(test)] + if let Some(path) = TEST_TASK_CGROUP.with(|path| path.borrow().clone()) { + return with_note_from(&path, instructions, max_bytes); + } + match TASK_CGROUP.get() { + Some(path) => with_note_from(path, instructions, max_bytes), + None => Cow::Borrowed(instructions), + } +} + +/// Records a run notice when the kernel reports a new kill in this task. The +/// kernel signals each change to memory.events, so an idle Runner reads nothing. +pub struct TaskMemoryMonitor { + task_cgroup: PathBuf, + reported: u64, + notices: u32, + changed: Arc, + _watcher: RecommendedWatcher, +} + +impl TaskMemoryMonitor { + pub fn new() -> Option { + TASK_CGROUP.get().cloned().and_then(Self::for_cgroup) + } + + /// Runner takes the starting count before it executes any command, so work it + /// starts is counted and a restarted Runner does not report earlier kills. + fn for_cgroup(task_cgroup: PathBuf) -> Option { + let changed = Arc::new(AtomicBool::new(false)); + let signal = Arc::clone(&changed); + let mut watcher = notify::recommended_watcher(move |_| { + signal.store(true, Ordering::Relaxed); + }) + .ok()?; + watcher + .watch( + &task_cgroup.join("memory.events"), + RecursiveMode::NonRecursive, + ) + .ok()?; + Some(Self { + reported: read_kills(&task_cgroup)?, + task_cgroup, + notices: 0, + changed, + _watcher: watcher, + }) + } + + pub fn poll( + &mut self, + config: &DurableRunnerConfig, + state: &mut DurableState, + store: &DurableStateStore, + ) { + // A change stays pending until a run can record it. + if self.notices >= MAX_NOTICES + || !self.changed.load(Ordering::Relaxed) + || state.lifecycle != "ready" + || state.warm_transition.is_some() + || state.run_id != config.run_id + { + return; + } + self.changed.store(false, Ordering::Relaxed); + let Some(kills) = read_kills(&self.task_cgroup) else { + return; + }; + if kills <= self.reported { + return; + } + if self.record(kills, config, state, store).is_ok() { + self.notices += 1; + self.reported = kills; + } else { + self.changed.store(true, Ordering::Relaxed); + } + } + + fn record( + &self, + total: u64, + config: &DurableRunnerConfig, + state: &mut DurableState, + store: &DurableStateStore, + ) -> Result<(), DurableRunnerError> { + let new_kills = total - self.reported; + let killed = if new_kills == 1 { + "a process".to_owned() + } else { + format!("{new_kills} processes") + }; + let mut next = state.clone(); + next.enqueue_event( + config, + "provider.notice.recorded", + EventPriority::P1, + json!({ + "schema": "paperclip.provider.notice.v1", + "noticeId": format!("task-memory-kill-{total}"), + "severity": "warning", + "category": "task_out_of_memory_kill", + "scope": "environment", + "recoverable": true, + "userActionable": false, + // The count also rises for a VM-wide kill, so it does not prove the limit caused it. + "summary": format!( + "The kernel's out-of-memory killer killed {killed} in this task ({total} so far)." + ), + }), + )?; + store.save(&next)?; + *state = next; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use std::time::{Duration, Instant}; + use uuid::Uuid; + + struct TestDirectory(PathBuf); + impl TestDirectory { + fn create() -> Self { + let path = std::env::temp_dir().join(format!("task-memory-{}", Uuid::new_v4())); + fs::create_dir_all(path.join("tasks").join("task-a")).unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap(); + } + Self(path) + } + fn task_cgroup(&self) -> PathBuf { + self.0.join("tasks").join("task-a") + } + fn write_limit(&self, limit: &str) { + fs::write(self.0.join("tasks").join("memory.max"), limit).unwrap(); + } + fn write_kills(&self, kills: u64) { + fs::write( + self.task_cgroup().join("memory.events"), + format!("low 0\nhigh 0\nmax 7\noom 2\noom_kill {kills}\noom_group_kill 0\n"), + ) + .unwrap(); + } + } + impl Drop for TestDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } + } + + fn config(path: PathBuf) -> DurableRunnerConfig { + DurableRunnerConfig { + connect_url: "ws://127.0.0.1/runner".into(), + ca_bundle_path: None, + state_dir: path, + runner_instance_id: "runner-1".into(), + environment_lease_id: "lease-1".into(), + run_id: "run-1".into(), + normalized_session_id: "session-1".into(), + turn_id: "turn-1".into(), + item_id: "item-1".into(), + runner_version: "1".into(), + runner_digest: "sha256:digest".into(), + acpx_launch_profile: None, + opencode_launch_profile: None, + cursor_cloud_launch_profile: None, + max_outbox_bytes: 1024 * 1024, + p0_reserve_bytes: 64 * 1024, + max_frame_bytes: 64 * 1024, + reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, + max_runtime: Duration::ZERO, + } + } + + fn setup(directory: &TestDirectory) -> (DurableRunnerConfig, DurableStateStore, DurableState) { + let config = config(directory.0.clone()); + let store = DurableStateStore::new(&config.state_dir).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + state.lifecycle = "ready".into(); + (config, store, state) + } + + fn notices(state: &DurableState) -> Vec { + state + .outbox + .iter() + .filter(|event| { + event + .envelope + .pointer("/payload/eventType") + .and_then(|v| v.as_str()) + == Some("provider.notice.recorded") + }) + .map(|event| event.envelope["payload"]["payload"].clone()) + .collect() + } + + /// Polls until a watcher event has had time to produce the wanted notices. + fn poll_for_notices( + monitor: &mut TaskMemoryMonitor, + config: &DurableRunnerConfig, + state: &mut DurableState, + store: &DurableStateStore, + wanted: usize, + ) { + let deadline = Instant::now() + Duration::from_secs(10); + while notices(state).len() < wanted && Instant::now() < deadline { + monitor.poll(config, state, store); + std::thread::sleep(Duration::from_millis(10)); + } + // Later events must not add a notice. + std::thread::sleep(Duration::from_millis(100)); + monitor.poll(config, state, store); + assert_eq!(notices(state).len(), wanted); + } + + fn monitor(directory: &TestDirectory) -> TaskMemoryMonitor { + directory.write_limit("6442450944\n"); + directory.write_kills(0); + TaskMemoryMonitor::for_cgroup(directory.task_cgroup()).unwrap() + } + + #[test] + fn note_states_the_limit_the_files_and_the_way_out() { + let directory = TestDirectory::create(); + directory.write_limit("6442450944\n"); + let task = directory.task_cgroup(); + let with_note = with_note_from(&task, "Do the work.", 1024 * 1024); + assert!(with_note.starts_with("Do the work.\n\nAll task processes")); + let events = format!("{}/memory.events", task.display()); + let current = format!("{}/memory.current", task.parent().unwrap().display()); + for fact in ["6 GiB", "exit code 137", &events, ¤t, "blocker"] { + assert!(with_note.contains(fact), "missing {fact}"); + } + assert!(with_note_from(&task, "", 1024 * 1024).starts_with("All task processes")); + } + + #[test] + fn instructions_stay_unchanged_without_a_limit() { + let directory = TestDirectory::create(); + let task = directory.task_cgroup(); + assert_eq!(with_note_from(&task, "Do the work.", 4096), "Do the work."); + for limit in ["max\n", "0\n", "not a number"] { + directory.write_limit(limit); + assert_eq!(with_note_from(&task, "Do the work.", 4096), "Do the work."); + } + } + + #[test] + fn note_never_pushes_instructions_over_the_provider_limit() { + let directory = TestDirectory::create(); + directory.write_limit("6442450944\n"); + let task = directory.task_cgroup(); + let note_bytes = with_note_from(&task, "", usize::MAX).len(); + let fits = "x".repeat(10); + assert_eq!( + with_note_from(&task, &fits, 10 + 2 + note_bytes).len(), + 10 + 2 + note_bytes + ); + assert_eq!(with_note_from(&task, &fits, 10 + 2 + note_bytes - 1), fits); + } + + #[test] + fn limit_keeps_a_fraction_when_it_is_not_whole() { + assert_eq!(format_limit(6 << 30), "6 GiB"); + assert_eq!(format_limit(3 << 29), "1.5 GiB"); + } + + #[test] + fn monitor_reports_new_kills_once_with_a_valid_notice() { + let directory = TestDirectory::create(); + let (config, store, mut state) = setup(&directory); + let mut monitor = monitor(&directory); + poll_for_notices(&mut monitor, &config, &mut state, &store, 0); + + directory.write_kills(1); + poll_for_notices(&mut monitor, &config, &mut state, &store, 1); + directory.write_kills(4); + poll_for_notices(&mut monitor, &config, &mut state, &store, 2); + + let notices = notices(&state); + assert_eq!(notices[0]["schema"], "paperclip.provider.notice.v1"); + assert_eq!(notices[0]["noticeId"], "task-memory-kill-1"); + assert_eq!(notices[0]["severity"], "warning"); + assert_eq!(notices[0]["scope"], "environment"); + assert_eq!( + notices[0]["summary"], + "The kernel's out-of-memory killer killed a process in this task (1 so far)." + ); + assert_eq!( + notices[1]["summary"], + "The kernel's out-of-memory killer killed 3 processes in this task (4 so far)." + ); + } + + #[test] + fn monitor_ignores_only_kills_from_before_it_started() { + let directory = TestDirectory::create(); + let (config, store, mut state) = setup(&directory); + directory.write_limit("6442450944\n"); + directory.write_kills(5); + let mut monitor = TaskMemoryMonitor::for_cgroup(directory.task_cgroup()).unwrap(); + // This kill comes after the start but before the first poll. + directory.write_kills(6); + poll_for_notices(&mut monitor, &config, &mut state, &store, 1); + assert_eq!( + notices(&state)[0]["summary"], + "The kernel's out-of-memory killer killed a process in this task (6 so far)." + ); + } + + #[test] + fn monitor_waits_for_an_active_run_and_keeps_the_kill() { + let directory = TestDirectory::create(); + let (config, store, mut state) = setup(&directory); + let mut monitor = monitor(&directory); + directory.write_kills(2); + state.lifecycle = "reconnecting".into(); + poll_for_notices(&mut monitor, &config, &mut state, &store, 0); + state.lifecycle = "ready".into(); + poll_for_notices(&mut monitor, &config, &mut state, &store, 1); + } + + #[test] + fn monitor_reads_nothing_until_the_file_changes() { + let directory = TestDirectory::create(); + let (config, store, mut state) = setup(&directory); + let mut monitor = monitor(&directory); + directory.write_kills(3); + std::thread::sleep(Duration::from_millis(100)); + // A read would now report 3 kills, so only the missing signal can stop it. + monitor.changed.store(false, Ordering::Relaxed); + monitor.poll(&config, &mut state, &store); + assert!(notices(&state).is_empty()); + monitor.changed.store(true, Ordering::Relaxed); + monitor.poll(&config, &mut state, &store); + assert_eq!(notices(&state).len(), 1); + } + + #[test] + fn monitor_bounds_notices() { + let directory = TestDirectory::create(); + let (config, store, mut state) = setup(&directory); + let mut monitor = monitor(&directory); + for count in 1..=u64::from(MAX_NOTICES) { + directory.write_kills(count); + let deadline = Instant::now() + Duration::from_secs(10); + while notices(&state).len() < count as usize && Instant::now() < deadline { + monitor.poll(&config, &mut state, &store); + std::thread::sleep(Duration::from_millis(1)); + } + } + assert_eq!(notices(&state).len(), MAX_NOTICES as usize); + directory.write_kills(u64::from(MAX_NOTICES) + 1); + poll_for_notices( + &mut monitor, + &config, + &mut state, + &store, + MAX_NOTICES as usize, + ); + } + + #[test] + fn monitor_needs_a_cgroup_with_memory_accounting() { + let directory = TestDirectory::create(); + assert!(TaskMemoryMonitor::for_cgroup(directory.task_cgroup()).is_none()); + } +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/task_memory.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/task_memory.rs new file mode 100644 index 00000000000..6d7797495c6 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/task_memory.rs @@ -0,0 +1,76 @@ +use std::borrow::Cow; +use std::fs; +use std::path::PathBuf; + +use paperclip_runner_core::codex_provider::{CodexProvider, CodexProviderConfig}; +use paperclip_runner_core::task_memory::{ + instructions_with_memory_note, set_task_cgroup, TaskMemoryMonitor, +}; + +// A separate test binary: the flag's process-wide path starts unset here, and this +// one test owns the order of the before and after states. +#[test] +fn only_the_launch_flag_adds_the_note_and_the_monitor() { + assert!(matches!( + instructions_with_memory_note("Do the work.", 4096), + Cow::Borrowed("Do the work.") + )); + assert!(TaskMemoryMonitor::new().is_none()); + + let directory = std::env::temp_dir().join(format!("runner-memory-{}", uuid::Uuid::new_v4())); + fs::create_dir(&directory).unwrap(); + let task_cgroup: PathBuf = directory.join("task-a"); + fs::create_dir(&task_cgroup).unwrap(); + fs::write(directory.join("memory.max"), "6442450944\n").unwrap(); + fs::write(task_cgroup.join("memory.events"), "oom_kill 0\n").unwrap(); + set_task_cgroup(task_cgroup); + + let instructions = instructions_with_memory_note("Do the work.", 4096); + assert!(instructions.starts_with("Do the work.\n\nAll task processes")); + assert!(instructions.contains("6 GiB")); + assert!(TaskMemoryMonitor::new().is_some()); + + let log = directory.join("requests.ndjson"); + let config = CodexProviderConfig { + provider: "codex".to_owned(), + driver: "codex_app_server".to_owned(), + provider_version: "fake-1".to_owned(), + command: PathBuf::from(env!("CARGO_BIN_EXE_fake-codex-app-server")), + args: vec![ + "--state-file".to_owned(), + directory.join("state.json").display().to_string(), + "--call-log".to_owned(), + directory.join("calls.log").display().to_string(), + "--request-log".to_owned(), + log.display().to_string(), + ], + cwd: std::env::current_dir().unwrap().display().to_string(), + model: Some("test-model".to_owned()), + provider_session_id: None, + instructions: "Do the work.".to_owned(), + approval_policy: "never".to_owned(), + externally_sandboxed: false, + read_only: false, + include_skill_instructions: None, + conversation_mode: None, + }; + let mut provider = CodexProvider::start(&config, None).unwrap(); + let thread_id = provider.thread_id().to_owned(); + provider.shutdown().unwrap(); + let mut resumed = CodexProvider::start(&config, Some(&thread_id)).unwrap(); + resumed.shutdown().unwrap(); + let requests = fs::read_to_string(&log).unwrap(); + for method in ["thread/start", "thread/resume"] { + let request: serde_json::Value = requests + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .find(|request| request["method"] == method) + .unwrap(); + assert_eq!( + request["params"]["developerInstructions"].as_str().unwrap(), + instructions.as_ref(), + "{method}" + ); + } + fs::remove_dir_all(directory).unwrap(); +}