mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 11:01:16 +01:00
Tell the agent about the launcher's memory limit (#18801)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The native Runner supervises the harness and tool processes for each run > - A launcher can put a memory limit on those processes > - When a process goes above the limit, the kernel kills it, and the agent sees only exit code 137 > - Any SIGKILL gives that code, so the agent cannot know the cause > - The agent can then run the same command again and fail again, and the people who read the run see no reason > - Only the launcher knows if a limit applies, so Runner must not assume one > - The kernel gives the limit and the kill count in cgroup files > - This pull request adds an optional `--task-memory-cgroup` flag that names the task's cgroup > - The benefit is that the agent knows the limit before it starts, the agent can confirm a kill, and the run shows each kill ## Linked Issues or Issue Description Related: #18730. That pull request lets the kernel kill large tool processes under Runner. This pull request tells the agent when that occurs because of a memory limit. **What happened?** A tool process on a VM with a 6 GiB task limit went above the limit. The kernel killed it, and Runner and the agent continued. The agent received exit code 137 and no other data. It ran the same type check in different forms, and the kernel killed it five times. **Expected behavior** The agent knows about the limit. It can confirm that the limit killed its process. It can then reduce the work or report a blocker. The run shows the kills. **Steps to reproduce** Start Runner under a memory limit. Start a tool process that goes above the limit. The agent and the run show only exit code 137. **Deployment mode** Native Runner on Linux, when the launcher gives the flag. ## What Changed - Add the optional `--task-memory-cgroup <path>` flag to `paperclip-runnerd`. Runner has no default value. - Runner reads the limit from `memory.max` on the parent of that cgroup. Runner reads the kill count from `oom_kill` in `memory.events` on that cgroup. - When a limit exists, add a note to the Codex and ACPX instructions at launch. The note gives the limit, the meaning of exit code 137, the paths of the files to read, and what to do after a kill. - Record a `provider.notice.recorded` warning when `oom_kill` increases. Runner watches `memory.events` for the kernel's change signal, so an idle Runner does not read it. Runner takes the starting count before it executes commands. It does not report kills from before it started. - Without the flag, or when the parent has no limit, the instructions do not change. Without the flag, Runner does not read the files and records no notice. - Add unit and integration tests. ## Verification Tests ran in `rust:1-bookworm` on Linux arm64: - `cargo fmt --all` - `cargo clippy -p paperclip-runner-core --bin paperclip-runnerd --lib` has no warnings in the new module. - `cargo test -p paperclip-runner-core --lib -- task_memory restores_settled_session_for_explicit_run_attachment_only`: The task memory tests pass. The second filter covers the note in the ACPX session configuration. - `cargo test -p paperclip-runner-core --test task_memory`: this test uses a separate process. It checks that no note and no monitor exist before the flag value is set. It then checks the note in the Codex `thread/start` and `thread/resume` requests. The tests use files in a temporary directory that have the format of the cgroup files. In a separate test on a real kernel, a kill above a limit on a parent cgroup increased `oom_kill` in the cgroup of the killed process. It did not increase `oom_kill` in the other child. On a real kernel, an unprivileged process that watched `memory.events` through a read-only mount received the change signal at the kill. I did not run the complete suite locally. CI runs it. I did not run this with a real launcher or on a real VM. The notice payload has the same fields as other Runner notices, but no test sends it through the control plane. ## Risks - The note adds text to each provider's instructions when the launcher gives the flag. If the text makes the instructions go above the provider limit, Runner does not add the note. - The task must be able to read the cgroup files. If it cannot, the note is absent or the agent cannot read the paths, and the run continues as it does now. - A kill while the run is not ready gives a notice later. A kill after Runner records 64 notices gives no notice. The agent can still read the file. - This pull request does not send a message into an active turn, because most ACPX profiles do not support this. The agent must read the file to confirm a kill. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass: targeted tests only, as listed above - [x] I have added or updated tests where applicable - [ ] I have updated relevant documentation to reflect my changes: not applicable, because no document lists the `paperclip-runnerd` flags - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Nicky LeachandClaude Opus 5.5 authored and GitHub committed
1 parent
6ef37398c8
commit
2e37c73c05
8 files changed
+590
-3
No files matched your search
@@ -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();
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -476,6 +476,7 @@ pub fn run_durable_runner<E: CommandExecutor>(
|
||||
#[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<E: CommandExecutor>(
|
||||
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.
|
||||
|
||||
@@ -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};
|
||||
|
||||
@@ -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<PathBuf> = 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<Option<PathBuf>> = 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<u64> {
|
||||
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<u64> {
|
||||
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<String> {
|
||||
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<AtomicBool>,
|
||||
_watcher: RecommendedWatcher,
|
||||
}
|
||||
|
||||
impl TaskMemoryMonitor {
|
||||
pub fn new() -> Option<Self> {
|
||||
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<Self> {
|
||||
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<serde_json::Value> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -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::<serde_json::Value>(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();
|
||||
}
|
||||
Reference in new issue
Block a user