Synced from monorepo
Synced from monorepo Changes: - Workspace task snapshots only list incomplete backgrounded tasks - Quiet auth, LSP, and config warnings in the shell - Fix observability attributes for warm store errors, restore setup, remote tools, and preview denials - Fail closed when soak metrics are missing - Run plan-mode exit last in mixed tool batches - Allow /loop to store prompts that can terminate the loop - Make subagent maximum nesting depth configurable - Security: apply sandbox profile to the leader process that executes tools Source-Revision: 1adcd1f477870e4a97bacbd6be78c8a3bfbac46d
This commit is contained in:
grokkybara[bot]
parent
b41c75a578
commit
02d9359435
96 changed files with 2346 additions and 351 deletions
|
|
@ -85,6 +85,7 @@ chrono = { workspace = true }
|
|||
futures-util = { workspace = true }
|
||||
xai-grok-sandbox = { path = "../xai-grok-sandbox", default-features = false }
|
||||
xai-file-utils = { path = "../xai-file-utils" }
|
||||
dhat = { workspace = true, optional = true }
|
||||
|
||||
[target.'cfg(unix)'.dependencies]
|
||||
command-fds = { version = "0.3", features = ["tokio"] }
|
||||
|
|
@ -101,9 +102,11 @@ windows = { workspace = true, features = ["Win32_Storage_FileSystem"] }
|
|||
|
||||
[dev-dependencies]
|
||||
dirs = { workspace = true }
|
||||
pretty_assertions = { workspace = true }
|
||||
tokio = { workspace = true, features = ["test-util"] }
|
||||
tracing-subscriber = { workspace = true, features = ["fmt"] }
|
||||
wiremock = { workspace = true }
|
||||
xai-grok-test-support = { workspace = true }
|
||||
xai-test-utils = { workspace = true }
|
||||
|
||||
[build-dependencies]
|
||||
|
|
@ -118,3 +121,4 @@ workspace = true
|
|||
default-bazel = ["serde"]
|
||||
default = ["serde"]
|
||||
serde = []
|
||||
dhat-heap = ["dep:dhat"]
|
||||
|
|
|
|||
|
|
@ -835,6 +835,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: owner.map(|s| s.to_string()),
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -444,6 +444,7 @@ impl ProcessState {
|
|||
kind: self.kind,
|
||||
owner_session_id: self.owner_session_id.clone(),
|
||||
description: self.description.clone(),
|
||||
is_backgrounded: self.bg_status.is_backgrounded(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1617,6 +1618,7 @@ impl LocalTerminalActor {
|
|||
explicitly_killed: p.explicitly_killed,
|
||||
owner_session_id: p.owner_session_id.clone(),
|
||||
description: p.description.clone(),
|
||||
is_backgrounded: true,
|
||||
};
|
||||
self.completed_task_snapshots.insert(id.clone(), snapshot);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -219,6 +219,9 @@ pub struct TaskSnapshot {
|
|||
/// Model-supplied label for task UI / snapshots.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub description: Option<String>,
|
||||
/// True after explicit/user/auto backgrounding; false for pure foreground runs.
|
||||
#[serde(default)]
|
||||
pub is_backgrounded: bool,
|
||||
}
|
||||
|
||||
impl TaskSnapshot {
|
||||
|
|
@ -238,6 +241,11 @@ impl TaskSnapshot {
|
|||
pub fn is_outstanding(&self) -> bool {
|
||||
!self.completed
|
||||
}
|
||||
|
||||
/// Incomplete and backgrounded — tray/`tasks_snapshot` predicate (not FG in-flight).
|
||||
pub fn is_outstanding_background(&self) -> bool {
|
||||
!self.completed && self.is_backgrounded
|
||||
}
|
||||
}
|
||||
|
||||
/// Result of killing a terminal task.
|
||||
|
|
|
|||
|
|
@ -50,7 +50,8 @@ pub use lsp::LspTool;
|
|||
pub use monitor::tool::MonitorTool;
|
||||
pub use read_file::ReadFileTool;
|
||||
pub use scheduler::create::{
|
||||
SCHEDULER_CREATE_TOOL_NAME, SchedulerCreateTool, loop_schedule_instruction, loop_usage_message,
|
||||
LoopFireMode, SCHEDULER_CREATE_TOOL_NAME, SchedulerCreateTool, loop_schedule_instruction,
|
||||
loop_usage_message,
|
||||
};
|
||||
pub use scheduler::delete::{SCHEDULER_DELETE_TOOL_NAME, SchedulerDeleteTool};
|
||||
pub use scheduler::list::SchedulerListTool;
|
||||
|
|
|
|||
|
|
@ -110,6 +110,53 @@ impl MonitorInput {
|
|||
}
|
||||
}
|
||||
|
||||
// Mid-turn monitor event buffer
|
||||
|
||||
/// A monitor event notification to be surfaced as a `<system-reminder>` mid-turn.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct MonitorEventNotification {
|
||||
pub task_id: String,
|
||||
pub event_text: String,
|
||||
/// Session that owns the monitor which produced this event.
|
||||
///
|
||||
/// In leader mode every session shares one [`MonitorEventBuffer`], so the
|
||||
/// drain sites filter on this to avoid surfacing one session's monitor
|
||||
/// events inside another session's turn. `None` for legacy / non-grok-build
|
||||
/// backends, which any session drains for backwards compatibility.
|
||||
pub owner_session_id: Option<String>,
|
||||
}
|
||||
|
||||
impl MonitorEventNotification {
|
||||
/// Whether this buffered event should surface in the session whose owner id
|
||||
/// is `my_owner`. Mirrors `task_owned_by_session`: an event surfaces only
|
||||
/// when it has no recorded owner (legacy) or its owner matches the draining
|
||||
/// session. Foreign events stay buffered for their own session to drain.
|
||||
pub fn owned_by_session(&self, my_owner: Option<&str>) -> bool {
|
||||
match (my_owner, self.owner_session_id.as_deref()) {
|
||||
(Some(me), Some(owner)) => me == owner,
|
||||
_ => true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared buffer for mid-turn monitor event notifications: an [`EventQueue`]
|
||||
/// of [`MonitorEventNotification`]. Producers `push_capped`; the turn loop
|
||||
/// drains its session's events via [`drain_owned`].
|
||||
///
|
||||
/// [`EventQueue`]: xai_interjection_core::EventQueue
|
||||
pub type MonitorEventBuffer = xai_interjection_core::EventQueue<MonitorEventNotification>;
|
||||
|
||||
crate::register_resource!("grok_build", "MonitorEventBuffer", MonitorEventBuffer);
|
||||
|
||||
/// Drain only `my_owner`'s events (the buffer is shared across sessions in
|
||||
/// leader mode); owner-less legacy events drain anywhere.
|
||||
pub fn drain_owned(
|
||||
buffer: &MonitorEventBuffer,
|
||||
my_owner: Option<&str>,
|
||||
) -> Vec<MonitorEventNotification> {
|
||||
buffer.drain_matching(|e| e.owned_by_session(my_owner))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ use super::types::{ScheduledTask, SchedulerCommand, SchedulerHandle, scheduler_t
|
|||
// Canonical /loop wording lives in the light API crate so other consumers can
|
||||
// link it without the tools implementation crate; re-exported to keep paths stable.
|
||||
pub use xai_grok_tools_api::slash_commands::{
|
||||
SCHEDULER_CREATE_TOOL_NAME, loop_schedule_instruction, loop_usage_message,
|
||||
LoopFireMode, SCHEDULER_CREATE_TOOL_NAME, loop_schedule_instruction, loop_usage_message,
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
|
||||
|
|
@ -475,7 +475,7 @@ mod tests {
|
|||
#[test]
|
||||
fn loop_schedule_instruction_holds_invariants() {
|
||||
let args = "every 30 minutes do x";
|
||||
let instr = loop_schedule_instruction(args);
|
||||
let instr = loop_schedule_instruction(args, LoopFireMode::Detached);
|
||||
assert!(
|
||||
!instr.contains("10m"),
|
||||
"instruction must not default: {instr}"
|
||||
|
|
|
|||
|
|
@ -14,7 +14,6 @@ use super::types::{
|
|||
};
|
||||
|
||||
/// Cap on retained completed-subagent entries before the oldest are evicted.
|
||||
/// Public so the subagent soak test can assert the coordinator stays bounded.
|
||||
pub const MAX_COMPLETED_ENTRIES: usize = 1024;
|
||||
pub(super) const OUTPUT_UNAVAILABLE_PLACEHOLDER: &str = "[subagent output no longer available]";
|
||||
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@
|
|||
//!
|
||||
//! - `SubagentBackendResource` — backend for spawn/query/cancel (required)
|
||||
//! - `SubagentDepthCounter` — current nesting depth (optional, defaults to 0)
|
||||
//! - `MaxSubagentDepth` — max nesting (optional, defaults to [`MAX_SUBAGENT_DEPTH`])
|
||||
//! - `SessionIdResource` — current session ID for parent scoping (optional)
|
||||
//! - `SubagentForegroundWait` — host wait-window guard factory (optional)
|
||||
//! - `TaskModelValidator` — validates explicit model slugs before spawn
|
||||
|
|
@ -30,10 +31,16 @@ use crate::types::resources::SharedResources;
|
|||
use crate::types::tool::{ToolKind, ToolNamespace};
|
||||
use xai_tool_types::{SubagentCompletedOutput, SubagentIsolationMode, TaskToolInput};
|
||||
|
||||
/// Maximum nesting depth for subagents. A top-level session is depth 0;
|
||||
/// the first subagent is depth 1. Subagents cannot spawn further subagents.
|
||||
/// Default max nesting depth when [`MaxSubagentDepth`] is not injected.
|
||||
pub const MAX_SUBAGENT_DEPTH: u32 = 1;
|
||||
|
||||
pub fn effective_max_subagent_depth(resources: &crate::types::resources::Resources) -> u32 {
|
||||
resources
|
||||
.get::<MaxSubagentDepth>()
|
||||
.map(|d| d.0)
|
||||
.unwrap_or(MAX_SUBAGENT_DEPTH)
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────────────
|
||||
// Tool implementation
|
||||
// ───────────────────────────────────────────────────────────────────────────
|
||||
|
|
@ -125,10 +132,19 @@ impl xai_tool_runtime::Tool for TaskTool {
|
|||
.map(|cancellation| cancellation.0.clone());
|
||||
|
||||
// 1. Depth check
|
||||
let (depth, backend, model_validator, parent_session_id, parent_prompt_id, foreground_wait) = {
|
||||
let (
|
||||
depth,
|
||||
max_depth,
|
||||
backend,
|
||||
model_validator,
|
||||
parent_session_id,
|
||||
parent_prompt_id,
|
||||
foreground_wait,
|
||||
) = {
|
||||
let res = resources.lock().await;
|
||||
|
||||
let depth = res.get::<SubagentDepthCounter>().map(|d| d.0).unwrap_or(0);
|
||||
let max_depth = effective_max_subagent_depth(&res);
|
||||
|
||||
let backend = res
|
||||
.get::<SubagentBackendResource>()
|
||||
|
|
@ -155,6 +171,7 @@ impl xai_tool_runtime::Tool for TaskTool {
|
|||
|
||||
(
|
||||
depth,
|
||||
max_depth,
|
||||
backend,
|
||||
model_validator,
|
||||
parent_session_id,
|
||||
|
|
@ -163,9 +180,9 @@ impl xai_tool_runtime::Tool for TaskTool {
|
|||
)
|
||||
};
|
||||
|
||||
if depth >= MAX_SUBAGENT_DEPTH {
|
||||
if depth >= max_depth {
|
||||
return Err(xai_tool_runtime::ToolError::invalid_arguments(format!(
|
||||
"Subagent depth limit exceeded (current depth: {depth}, max: {MAX_SUBAGENT_DEPTH}). \
|
||||
"Subagent depth limit exceeded (current depth: {depth}, max: {max_depth}). \
|
||||
Cannot spawn further nested subagents."
|
||||
)));
|
||||
}
|
||||
|
|
@ -558,6 +575,41 @@ mod tests {
|
|||
assert!(err.contains("depth limit exceeded"), "error: {err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn raised_max_depth_allows_nested_spawn() {
|
||||
let (backend, mut rx) = make_backend();
|
||||
let mut resources = Resources::new();
|
||||
resources.insert(backend);
|
||||
resources.insert(SubagentDepthCounter(1));
|
||||
resources.insert(MaxSubagentDepth(2));
|
||||
resources.insert(SessionIdResource("child-session".to_string()));
|
||||
resources.insert(CurrentPromptIdResource("prompt-nested".to_string()));
|
||||
|
||||
let result = xai_tool_runtime::Tool::run(
|
||||
&TaskTool,
|
||||
test_ctx(resources.into_shared()),
|
||||
TaskToolInput {
|
||||
description: "nested ok".into(),
|
||||
prompt: "should be allowed at max_depth=2".into(),
|
||||
subagent_type: "explore".into(),
|
||||
run_in_background: true,
|
||||
capability_mode: None,
|
||||
isolation: None,
|
||||
resume_from: None,
|
||||
cwd: None,
|
||||
model: None,
|
||||
task_id: None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
result.is_ok(),
|
||||
"expected Ok at depth 1 with max 2: {result:?}"
|
||||
);
|
||||
let _ = rx.try_recv();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn subagent_cannot_spawn_nested_subagent() {
|
||||
let (backend, _rx) = make_backend();
|
||||
|
|
|
|||
|
|
@ -10,7 +10,8 @@
|
|||
//!
|
||||
//! - `SubagentBackendResource` — wraps an `Arc<dyn SubagentBackend>` that
|
||||
//! abstracts spawn/query/cancel (see [`super::backend`])
|
||||
//! - `SubagentDepthCounter` — tracks nesting depth (max 1, no recursive spawning)
|
||||
//! - `SubagentDepthCounter` — current nesting depth
|
||||
//! - `MaxSubagentDepth` — configured max nesting depth
|
||||
//! - `SessionIdResource` — carries the current session ID for parent scoping
|
||||
//! - `TaskModelValidator` — validates explicit model slugs before background spawn
|
||||
//!
|
||||
|
|
@ -877,51 +878,6 @@ pub struct SubagentEventSender(#[educe(Debug(ignore))] pub mpsc::UnboundedSender
|
|||
|
||||
register_resource!("grok_build", "SubagentEventSender", SubagentEventSender);
|
||||
|
||||
// Mid-turn monitor event buffer
|
||||
|
||||
/// A monitor event notification to be surfaced as a `<system-reminder>` mid-turn.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct MonitorEventNotification {
|
||||
pub task_id: String,
|
||||
pub event_text: String,
|
||||
/// Session that owns the monitor which produced this event.
|
||||
///
|
||||
/// In leader mode every session shares one [`MonitorEventBuffer`], so the
|
||||
/// drain sites filter on this to avoid surfacing one session's monitor
|
||||
/// events inside another session's turn. `None` for legacy / non-grok-build
|
||||
/// backends, which any session drains for backwards compatibility.
|
||||
pub owner_session_id: Option<String>,
|
||||
}
|
||||
|
||||
impl MonitorEventNotification {
|
||||
/// Whether this buffered event should surface in the session whose owner id
|
||||
/// is `my_owner`. Mirrors `task_owned_by_session`: an event surfaces only
|
||||
/// when it has no recorded owner (legacy) or its owner matches the draining
|
||||
/// session. Foreign events stay buffered for their own session to drain.
|
||||
pub fn owned_by_session(&self, my_owner: Option<&str>) -> bool {
|
||||
match (my_owner, self.owner_session_id.as_deref()) {
|
||||
(Some(me), Some(owner)) => me == owner,
|
||||
_ => true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared buffer for mid-turn monitor event notifications: an [`EventQueue`]
|
||||
/// of [`MonitorEventNotification`]. Producers `push_capped`; the turn loop
|
||||
/// drains its session's events via [`drain_owned`].
|
||||
pub type MonitorEventBuffer = xai_interjection_core::EventQueue<MonitorEventNotification>;
|
||||
|
||||
register_resource!("grok_build", "MonitorEventBuffer", MonitorEventBuffer);
|
||||
|
||||
/// Drain only `my_owner`'s events (the buffer is shared across sessions in
|
||||
/// leader mode); owner-less legacy events drain anywhere.
|
||||
pub fn drain_owned(
|
||||
buffer: &MonitorEventBuffer,
|
||||
my_owner: Option<&str>,
|
||||
) -> Vec<MonitorEventNotification> {
|
||||
buffer.drain_matching(|e| e.owned_by_session(my_owner))
|
||||
}
|
||||
|
||||
// Active subagent listing (compaction)
|
||||
|
||||
/// Lightweight summary of a running subagent.
|
||||
|
|
@ -952,15 +908,18 @@ pub struct SubagentListActiveRequest {
|
|||
pub respond_to: oneshot::Sender<Vec<ActiveSubagentSummary>>,
|
||||
}
|
||||
|
||||
/// Tracks nesting depth. Injected into child's Resources with depth+1.
|
||||
///
|
||||
/// Top-level sessions start at depth 0. Each child increments by 1.
|
||||
/// `TaskTool` rejects spawns when `depth >= MAX_SUBAGENT_DEPTH`.
|
||||
/// Current nesting depth (top-level = 0; child = parent + 1).
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SubagentDepthCounter(pub u32);
|
||||
|
||||
register_resource!("grok_build", "SubagentDepthCounter", SubagentDepthCounter);
|
||||
|
||||
/// Host-injected max nesting depth; absent → [`super::MAX_SUBAGENT_DEPTH`].
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct MaxSubagentDepth(pub u32);
|
||||
|
||||
register_resource!("grok_build", "MaxSubagentDepth", MaxSubagentDepth);
|
||||
|
||||
/// Session-scoped validator for model-facing `Task.model` arguments.
|
||||
///
|
||||
/// Returns an error message for an invalid slug and `None` for a valid slug.
|
||||
|
|
|
|||
|
|
@ -917,6 +917,7 @@ pub(crate) mod test_helpers {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
use crate::types::requirements::{Expr, ToolRequirement};
|
||||
use crate::types::tool::{ToolKind, ToolNamespace};
|
||||
|
||||
use super::task::MAX_SUBAGENT_DEPTH;
|
||||
use super::task::types::SubagentDepthCounter;
|
||||
|
||||
pub use xai_grok_tools_api::slash_commands::WORKFLOW_TOOL_NAME;
|
||||
|
|
@ -237,7 +236,8 @@ impl xai_tool_runtime::Tool for WorkflowTool {
|
|||
(depth, sender)
|
||||
};
|
||||
|
||||
if depth >= MAX_SUBAGENT_DEPTH {
|
||||
// Workflows stay top-level-only regardless of configurable subagent depth.
|
||||
if depth > 0 {
|
||||
return Err(xai_tool_runtime::ToolError::custom(
|
||||
"workflow_depth_exceeded",
|
||||
"Workflows can only be launched from a top-level session (subagents and \
|
||||
|
|
|
|||
|
|
@ -108,6 +108,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -255,7 +255,7 @@ fn split_wrapped_monitor_event(event_text: &str) -> Option<(&str, &str)> {
|
|||
/// Buffered `event_text` arrives pre-wrapped (`wrap_monitor_event`); it is
|
||||
/// unwrapped via [`split_wrapped_monitor_event`] with verbatim fallback.
|
||||
pub fn format_monitor_events(
|
||||
events: &[crate::implementations::grok_build::task::types::MonitorEventNotification],
|
||||
events: &[crate::implementations::grok_build::monitor::types::MonitorEventNotification],
|
||||
task_output_name: Option<&str>,
|
||||
) -> Option<String> {
|
||||
use std::fmt::Write as _;
|
||||
|
|
@ -278,7 +278,8 @@ pub fn format_monitor_events(
|
|||
))
|
||||
}
|
||||
_ => {
|
||||
type Event = crate::implementations::grok_build::task::types::MonitorEventNotification;
|
||||
type Event =
|
||||
crate::implementations::grok_build::monitor::types::MonitorEventNotification;
|
||||
let mut groups: Vec<(&str, Vec<&Event>)> = Vec::new();
|
||||
for event in events {
|
||||
match groups.iter_mut().find(|(id, _)| *id == event.task_id) {
|
||||
|
|
@ -802,6 +803,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
};
|
||||
let msg = format_bash_completion(&task, Some("get_command_or_subagent_output"), None);
|
||||
assert!(msg.contains("abc-123"));
|
||||
|
|
@ -829,6 +831,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
};
|
||||
let msg = format_monitor_completion(&task, Some("get_command_or_subagent_output"));
|
||||
assert!(
|
||||
|
|
@ -862,6 +865,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
};
|
||||
let msg = format_monitor_completion(&task, None);
|
||||
assert!(
|
||||
|
|
@ -890,6 +894,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
};
|
||||
let msg = format_bash_completion(&task, Some("get_command_or_subagent_output"), None);
|
||||
assert!(msg.contains("cargo test"));
|
||||
|
|
@ -915,6 +920,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
};
|
||||
let msg = format_bash_completion(&task, Some("get_command_or_subagent_output"), None);
|
||||
assert!(msg.contains("exit code: unknown"));
|
||||
|
|
@ -943,6 +949,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
};
|
||||
let msg = format_bash_completion(&task, Some("get_command_or_subagent_output"), None);
|
||||
assert!(
|
||||
|
|
@ -982,6 +989,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
};
|
||||
let msg = format_bash_completion(&task, Some("get_command_or_subagent_output"), None);
|
||||
assert!(
|
||||
|
|
@ -1020,6 +1028,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
};
|
||||
let msg = format_bash_completion(&task, Some("get_command_or_subagent_output"), None);
|
||||
assert!(msg.contains("exit code: 0"));
|
||||
|
|
@ -1181,6 +1190,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
}
|
||||
}
|
||||
fn make_running(id: &str) -> TaskSnapshot {
|
||||
|
|
@ -1202,6 +1212,7 @@ mod tests {
|
|||
explicitly_killed: false,
|
||||
owner_session_id: None,
|
||||
description: None,
|
||||
is_backgrounded: false,
|
||||
}
|
||||
}
|
||||
fn make_bg_started(id: &str) -> crate::types::output::BackgroundTaskStarted {
|
||||
|
|
@ -1811,7 +1822,7 @@ mod tests {
|
|||
/// reintroduced.
|
||||
#[tokio::test]
|
||||
async fn reminder_pipeline_ignores_monitor_event_buffer() {
|
||||
use crate::implementations::grok_build::task::types::{
|
||||
use crate::implementations::grok_build::monitor::types::{
|
||||
MonitorEventBuffer, MonitorEventNotification,
|
||||
};
|
||||
use crate::types::resources::Resources;
|
||||
|
|
@ -1845,7 +1856,7 @@ mod tests {
|
|||
/// own + owner-less legacy events; foreign events stay buffered.
|
||||
#[test]
|
||||
fn drain_owned_partitions_by_session_owner() {
|
||||
use crate::implementations::grok_build::task::types::{
|
||||
use crate::implementations::grok_build::monitor::types::{
|
||||
MonitorEventBuffer, MonitorEventNotification, drain_owned,
|
||||
};
|
||||
let shared_buffer = MonitorEventBuffer::default();
|
||||
|
|
@ -1882,7 +1893,7 @@ mod tests {
|
|||
/// empty => `None`.
|
||||
#[test]
|
||||
fn format_monitor_events_single_vs_batched() {
|
||||
use crate::implementations::grok_build::task::types::MonitorEventNotification;
|
||||
use crate::implementations::grok_build::monitor::types::MonitorEventNotification;
|
||||
let event = |task: &str, desc: &str, text: &str| MonitorEventNotification {
|
||||
task_id: task.to_string(),
|
||||
event_text: format!(
|
||||
|
|
@ -1900,7 +1911,7 @@ mod tests {
|
|||
single, "<monitor-event task_id=\"task-0\">\n[alpha] line 0\n</monitor-event>",
|
||||
"single event must use the lean monitor-event form"
|
||||
);
|
||||
let bare = crate::implementations::grok_build::task::types::MonitorEventNotification {
|
||||
let bare = crate::implementations::grok_build::monitor::types::MonitorEventNotification {
|
||||
task_id: "task-9".into(),
|
||||
event_text: "bare text, no wrapper".into(),
|
||||
owner_session_id: None,
|
||||
|
|
@ -1988,7 +1999,7 @@ mod tests {
|
|||
/// End-to-end multibyte safety through the formatter (single + batch).
|
||||
#[test]
|
||||
fn format_monitor_events_handles_multibyte_content() {
|
||||
use crate::implementations::grok_build::task::types::MonitorEventNotification;
|
||||
use crate::implementations::grok_build::monitor::types::MonitorEventNotification;
|
||||
let event = |task: &str, desc: &str, text: &str| MonitorEventNotification {
|
||||
task_id: task.to_string(),
|
||||
event_text: format!(
|
||||
|
|
|
|||
928
crates/codegen/xai-grok-tools/tests/test_subagent_soak.rs
Normal file
928
crates/codegen/xai-grok-tools/tests/test_subagent_soak.rs
Normal file
|
|
@ -0,0 +1,928 @@
|
|||
//! Subagent lifecycle soak: churn spawn/run/completion/eviction and assert
|
||||
//! threads, fds, and heap/RSS reach steady state. A stub `ChildRunner` drives
|
||||
//! the real coordinator/transport.
|
||||
//!
|
||||
//! SUBAGENT_SOAK_CYCLES=20000 cargo test -p xai-grok-tools \
|
||||
//! [--features dhat-heap] --test test_subagent_soak -- --ignored --nocapture
|
||||
|
||||
#![cfg(unix)]
|
||||
|
||||
#[cfg(feature = "dhat-heap")]
|
||||
#[global_allocator]
|
||||
static DHAT_ALLOC: dhat::Alloc = dhat::Alloc;
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use serde::ser::SerializeMap;
|
||||
use serde::{Serialize, Serializer};
|
||||
use strum::{EnumCount, IntoEnumIterator};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use xai_grok_test_support::env::env_parse;
|
||||
use xai_grok_test_support::resources::{ResourceGrowth, ResourceSnapshot};
|
||||
use xai_grok_tools::implementations::grok_build::task::backend::{ChannelBackend, SubagentBackend};
|
||||
use xai_grok_tools::implementations::grok_build::task::coordinator::{
|
||||
ChildCompletion, ChildControl, ChildRunOutput, ChildRunRequest, ChildRunner, CoordinatorConfig,
|
||||
LocalBoxFuture, MAX_COMPLETED_ENTRIES, StartedChild, SubagentCoordinator, SubagentProgress,
|
||||
};
|
||||
use xai_grok_tools::implementations::grok_build::task::types::{
|
||||
SubagentDescribeOutcome, SubagentOwner, SubagentRegistryCounts, SubagentRequest,
|
||||
SubagentResult, SubagentValidateTypeOutcome,
|
||||
};
|
||||
|
||||
const PARENT_SESSION_ID: &str = "subagent-soak-parent";
|
||||
|
||||
#[derive(Clone, Copy, strum::EnumCount, strum::EnumIter)]
|
||||
enum Metric {
|
||||
Rss,
|
||||
Threads,
|
||||
Fds,
|
||||
}
|
||||
|
||||
impl Metric {
|
||||
fn label(self) -> &'static str {
|
||||
match self {
|
||||
Metric::Rss => "rss",
|
||||
Metric::Threads => "threads",
|
||||
Metric::Fds => "fds",
|
||||
}
|
||||
}
|
||||
|
||||
/// RSS reports raw bytes, so its key names the unit.
|
||||
fn summary_key(self) -> &'static str {
|
||||
match self {
|
||||
Metric::Rss => "rss_bytes",
|
||||
Metric::Threads => "threads",
|
||||
Metric::Fds => "fds",
|
||||
}
|
||||
}
|
||||
|
||||
fn unit(self) -> Option<&'static str> {
|
||||
match self {
|
||||
Metric::Rss => Some("MiB"),
|
||||
Metric::Threads | Metric::Fds => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn budget(self, bounds: &Bounds) -> f64 {
|
||||
match self {
|
||||
Metric::Rss => bounds.max_rss_growth_mib as f64,
|
||||
Metric::Threads => bounds.max_thread_growth as f64,
|
||||
Metric::Fds => bounds.max_fd_growth as f64,
|
||||
}
|
||||
}
|
||||
|
||||
/// RSS growth samples are bytes; convert to MiB for the budget comparison.
|
||||
fn growth_in_budget_unit(self, raw: usize) -> f64 {
|
||||
match self {
|
||||
Metric::Rss => bytes_to_mib(raw),
|
||||
Metric::Threads | Metric::Fds => raw as f64,
|
||||
}
|
||||
}
|
||||
|
||||
/// RSS is sampled on every unix; thread and fd counts are Linux-only.
|
||||
fn expected_on_this_platform(self) -> bool {
|
||||
match self {
|
||||
Metric::Rss => true,
|
||||
Metric::Threads | Metric::Fds => cfg!(target_os = "linux"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Reads a metric's field from a snapshot or a growth delta so serialization and
|
||||
/// the gates share one projection instead of repeating it.
|
||||
trait MetricValue {
|
||||
fn value_of(&self, metric: Metric) -> Option<usize>;
|
||||
}
|
||||
|
||||
impl MetricValue for ResourceSnapshot {
|
||||
fn value_of(&self, metric: Metric) -> Option<usize> {
|
||||
// Destructure so a new resource field is a compile error here, not a
|
||||
// silently dropped metric.
|
||||
let ResourceSnapshot { rss, threads, fds } = *self;
|
||||
match metric {
|
||||
Metric::Rss => rss,
|
||||
Metric::Threads => threads,
|
||||
Metric::Fds => fds,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl MetricValue for ResourceGrowth {
|
||||
fn value_of(&self, metric: Metric) -> Option<usize> {
|
||||
let ResourceGrowth { rss, threads, fds } = *self;
|
||||
match metric {
|
||||
Metric::Rss => rss,
|
||||
Metric::Threads => threads,
|
||||
Metric::Fds => fds,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn serialize_metrics<T: MetricValue, S: Serializer>(
|
||||
value: &T,
|
||||
serializer: S,
|
||||
) -> Result<S::Ok, S::Error> {
|
||||
let mut map = serializer.serialize_map(Some(Metric::COUNT))?;
|
||||
for metric in Metric::iter() {
|
||||
map.serialize_entry(metric.summary_key(), &value.value_of(metric))?;
|
||||
}
|
||||
map.end()
|
||||
}
|
||||
|
||||
fn bytes_to_mib(bytes: usize) -> f64 {
|
||||
bytes as f64 / (1024.0 * 1024.0)
|
||||
}
|
||||
|
||||
#[cfg_attr(not(feature = "dhat-heap"), allow(dead_code))]
|
||||
#[derive(Clone, Copy, Serialize)]
|
||||
struct HeapSample {
|
||||
blocks: i64,
|
||||
bytes: i64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Serialize)]
|
||||
struct HeapMetrics {
|
||||
before: HeapSample,
|
||||
after: HeapSample,
|
||||
blocks_per_cycle: f64,
|
||||
bytes_per_cycle: f64,
|
||||
}
|
||||
|
||||
impl HeapMetrics {
|
||||
fn new(before: HeapSample, after: HeapSample, cycles: u64) -> Self {
|
||||
// `SUBAGENT_SOAK_CYCLES=0` would otherwise divide by zero and feed
|
||||
// NaN/inf into the leak gates.
|
||||
let cycles = cycles.max(1) as f64;
|
||||
Self {
|
||||
before,
|
||||
after,
|
||||
blocks_per_cycle: (after.blocks - before.blocks) as f64 / cycles,
|
||||
bytes_per_cycle: (after.bytes - before.bytes) as f64 / cycles,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct Bounds {
|
||||
#[serde(rename = "warmup_cycles")]
|
||||
warmup: u64,
|
||||
#[serde(rename = "measured_cycles")]
|
||||
measure: u64,
|
||||
concurrency: u64,
|
||||
max_thread_growth: u64,
|
||||
max_fd_growth: u64,
|
||||
max_rss_growth_mib: u64,
|
||||
max_blocks_per_cycle: f64,
|
||||
max_bytes_per_cycle: f64,
|
||||
}
|
||||
|
||||
impl Bounds {
|
||||
fn from_env() -> Self {
|
||||
Self {
|
||||
// Default warmup to the completed-entry cap so the ring is saturated
|
||||
// and the measured window observes steady-state eviction rather than
|
||||
// one-time cache fill.
|
||||
warmup: env_parse("SUBAGENT_SOAK_WARMUP", MAX_COMPLETED_ENTRIES as u64),
|
||||
measure: env_parse("SUBAGENT_SOAK_CYCLES", 512u64),
|
||||
concurrency: env_parse("SUBAGENT_SOAK_CONCURRENCY", 16u64),
|
||||
// RSS is looser than threads and fds to absorb allocator noise.
|
||||
max_thread_growth: env_parse("SUBAGENT_SOAK_MAX_THREAD_GROWTH", 8u64),
|
||||
max_fd_growth: env_parse("SUBAGENT_SOAK_MAX_FD_GROWTH", 16u64),
|
||||
max_rss_growth_mib: env_parse("SUBAGENT_SOAK_MAX_RSS_GROWTH_MIB", 256u64),
|
||||
max_blocks_per_cycle: env_parse("SUBAGENT_SOAK_MAX_BLOCKS_PER_CYCLE", 2.0f64),
|
||||
max_bytes_per_cycle: env_parse("SUBAGENT_SOAK_MAX_BYTES_PER_CYCLE", 4096.0f64),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct Measurement {
|
||||
#[serde(serialize_with = "serialize_metrics")]
|
||||
before: ResourceSnapshot,
|
||||
#[serde(serialize_with = "serialize_metrics")]
|
||||
after: ResourceSnapshot,
|
||||
#[serde(serialize_with = "serialize_metrics")]
|
||||
growth: ResourceGrowth,
|
||||
#[serde(serialize_with = "serialize_counts")]
|
||||
counts: SubagentRegistryCounts,
|
||||
heap: Option<HeapMetrics>,
|
||||
quiesced: bool,
|
||||
}
|
||||
|
||||
fn serialize_counts<S: Serializer>(
|
||||
counts: &SubagentRegistryCounts,
|
||||
serializer: S,
|
||||
) -> Result<S::Ok, S::Error> {
|
||||
// Exhaustive destructure so a new count field is a compile error here, not a
|
||||
// silently dropped summary key.
|
||||
let SubagentRegistryCounts {
|
||||
pending,
|
||||
active,
|
||||
completed,
|
||||
} = counts;
|
||||
let entries = [
|
||||
("pending", pending),
|
||||
("active", active),
|
||||
("completed", completed),
|
||||
];
|
||||
let mut map = serializer.serialize_map(Some(entries.len()))?;
|
||||
for (key, value) in entries {
|
||||
map.serialize_entry(key, value)?;
|
||||
}
|
||||
map.end()
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct Summary<'a> {
|
||||
#[serde(flatten)]
|
||||
bounds: &'a Bounds,
|
||||
#[serde(flatten)]
|
||||
measurement: &'a Measurement,
|
||||
}
|
||||
|
||||
fn heap_capture() -> Option<HeapSample> {
|
||||
#[cfg(feature = "dhat-heap")]
|
||||
{
|
||||
let stats = dhat::HeapStats::get();
|
||||
Some(HeapSample {
|
||||
blocks: stats.curr_blocks as i64,
|
||||
bytes: stats.curr_bytes as i64,
|
||||
})
|
||||
}
|
||||
#[cfg(not(feature = "dhat-heap"))]
|
||||
{
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
async fn quiesce(backend: &ChannelBackend) -> bool {
|
||||
const MAX_POLLS: usize = 200;
|
||||
const SLEEP: Duration = Duration::from_millis(5);
|
||||
for _ in 0..MAX_POLLS {
|
||||
let counts = backend.registry_counts().await;
|
||||
if counts.pending == 0 && counts.active == 0 {
|
||||
return true;
|
||||
}
|
||||
tokio::time::sleep(SLEEP).await;
|
||||
}
|
||||
let counts = backend.registry_counts().await;
|
||||
eprintln!(
|
||||
"[soak] quiesce budget expired with pending={} active={}; snapshot may be noisy",
|
||||
counts.pending, counts.active
|
||||
);
|
||||
false
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct SoakControl {
|
||||
cancellation: CancellationToken,
|
||||
}
|
||||
|
||||
impl ChildControl for SoakControl {
|
||||
type ProgressFuture = std::future::Ready<SubagentProgress>;
|
||||
|
||||
fn progress(&self) -> Self::ProgressFuture {
|
||||
std::future::ready(SubagentProgress::default())
|
||||
}
|
||||
|
||||
fn cancel(&self) {
|
||||
self.cancellation.cancel();
|
||||
}
|
||||
}
|
||||
|
||||
struct SoakRunner {
|
||||
gate: Arc<tokio::sync::Semaphore>,
|
||||
}
|
||||
|
||||
impl ChildRunner for SoakRunner {
|
||||
type Control = SoakControl;
|
||||
type CompletionData = ();
|
||||
type RunFuture = LocalBoxFuture<ChildRunOutput<()>>;
|
||||
type ValidateFuture = LocalBoxFuture<SubagentValidateTypeOutcome>;
|
||||
type DescribeFuture = LocalBoxFuture<SubagentDescribeOutcome>;
|
||||
|
||||
fn run(&self, run: ChildRunRequest<Self::Control>) -> Self::RunFuture {
|
||||
let gate = self.gate.clone();
|
||||
Box::pin(async move {
|
||||
let ChildRunRequest {
|
||||
request,
|
||||
cancellation,
|
||||
reporter,
|
||||
} = run;
|
||||
let promoted = reporter
|
||||
.started(StartedChild {
|
||||
child_session_id: request.id.clone(),
|
||||
persona: None,
|
||||
resumed_from: request.resume_from.clone(),
|
||||
child_cwd: request.cwd.clone().unwrap_or_default(),
|
||||
worktree_path: None,
|
||||
effective_model_id: "soak-model".to_owned(),
|
||||
definition_background: false,
|
||||
control: SoakControl {
|
||||
cancellation: cancellation.clone(),
|
||||
},
|
||||
})
|
||||
.await;
|
||||
if !promoted || cancellation.is_cancelled() {
|
||||
return ChildRunOutput {
|
||||
result: SubagentResult {
|
||||
success: false,
|
||||
cancelled: true,
|
||||
error: Some("cancelled before start".to_owned()),
|
||||
subagent_id: request.id.clone(),
|
||||
child_session_id: request.id,
|
||||
..Default::default()
|
||||
},
|
||||
completion_data: (),
|
||||
snapshot_ref: None,
|
||||
};
|
||||
}
|
||||
if request.id.starts_with("conc-") {
|
||||
// Hold in `active` until the concurrent phase releases the gate.
|
||||
let _ = gate.acquire().await;
|
||||
}
|
||||
ChildRunOutput {
|
||||
result: SubagentResult {
|
||||
success: true,
|
||||
output: Arc::from("soak child output"),
|
||||
subagent_id: request.id.clone(),
|
||||
child_session_id: request.id,
|
||||
tool_calls: 1,
|
||||
turns: 1,
|
||||
..Default::default()
|
||||
},
|
||||
completion_data: (),
|
||||
snapshot_ref: None,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_type(&self, _subagent_type: String, _parent: String) -> Self::ValidateFuture {
|
||||
Box::pin(std::future::ready(SubagentValidateTypeOutcome::Ok))
|
||||
}
|
||||
|
||||
fn describe_type(
|
||||
&self,
|
||||
_subagent_type: String,
|
||||
_harness_agent_type: Option<String>,
|
||||
_parent: String,
|
||||
) -> Self::DescribeFuture {
|
||||
Box::pin(std::future::ready(SubagentDescribeOutcome::Unavailable))
|
||||
}
|
||||
|
||||
fn on_completed(&self, _completion: ChildCompletion<Self::CompletionData>) {}
|
||||
}
|
||||
|
||||
fn soak_request(id: String, background: bool) -> SubagentRequest {
|
||||
SubagentRequest {
|
||||
id,
|
||||
prompt: "soak work".to_owned(),
|
||||
description: "soak child".to_owned(),
|
||||
subagent_type: "explore".to_owned(),
|
||||
parent_session_id: PARENT_SESSION_ID.to_owned(),
|
||||
parent_prompt_id: Some("soak-prompt".to_owned()),
|
||||
resume_from: None,
|
||||
cwd: None,
|
||||
runtime_overrides: Default::default(),
|
||||
run_in_background: background,
|
||||
surface_completion: true,
|
||||
await_to_completion: false,
|
||||
fork_context: false,
|
||||
owner: SubagentOwner::Task,
|
||||
cancel_token: CancellationToken::new(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_cycle(backend: &ChannelBackend, i: u64) {
|
||||
let fg = backend
|
||||
.spawn(soak_request(format!("fg-{i}"), false))
|
||||
.await
|
||||
.expect("foreground spawn round-trips through the coordinator");
|
||||
assert!(fg.success, "cycle {i}: foreground child must complete");
|
||||
|
||||
let bg_id = format!("bg-{i}");
|
||||
let bg = backend
|
||||
.spawn(soak_request(bg_id.clone(), true))
|
||||
.await
|
||||
.expect("background spawn round-trips through the coordinator");
|
||||
assert!(bg.success, "cycle {i}: background child must complete");
|
||||
|
||||
let blocking = true;
|
||||
let timeout_ms = Some(5_000);
|
||||
let snapshot = backend.query(&bg_id, blocking, timeout_ms).await;
|
||||
assert!(
|
||||
snapshot.is_some(),
|
||||
"cycle {i}: completed subagent must be queryable"
|
||||
);
|
||||
}
|
||||
|
||||
async fn await_concurrency(backend: &ChannelBackend, n: u64) -> bool {
|
||||
const MAX_POLLS: usize = 400;
|
||||
const SLEEP: Duration = Duration::from_millis(5);
|
||||
for _ in 0..MAX_POLLS {
|
||||
if backend.registry_counts().await.active as u64 >= n {
|
||||
return true;
|
||||
}
|
||||
tokio::time::sleep(SLEEP).await;
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
async fn concurrent_phase(backend: &ChannelBackend, gate: &tokio::sync::Semaphore, n: u64) {
|
||||
let handles: Vec<_> = (0..n)
|
||||
.map(|k| {
|
||||
let backend = backend.clone();
|
||||
tokio::task::spawn_local(async move {
|
||||
backend.spawn(soak_request(format!("conc-{k}"), true)).await
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
let reached = await_concurrency(backend, n).await;
|
||||
|
||||
// Release then join before asserting, so no child is left blocked on failure.
|
||||
gate.add_permits(n as usize);
|
||||
for h in handles {
|
||||
let result = h.await.expect("concurrent spawn task");
|
||||
assert!(
|
||||
result.expect("concurrent spawn round-trips").success,
|
||||
"concurrent child must complete"
|
||||
);
|
||||
}
|
||||
assert!(reached, "expected {n} concurrently active children");
|
||||
}
|
||||
|
||||
async fn warmup(backend: &ChannelBackend, cycles: u64) -> bool {
|
||||
for i in 0..cycles {
|
||||
run_cycle(backend, i).await;
|
||||
}
|
||||
quiesce(backend).await
|
||||
}
|
||||
|
||||
async fn measure(
|
||||
backend: &ChannelBackend,
|
||||
bounds: &Bounds,
|
||||
baseline_quiesced: bool,
|
||||
) -> Measurement {
|
||||
let heap_before = heap_capture();
|
||||
let before = ResourceSnapshot::capture();
|
||||
|
||||
// Continue ids past the warmup window so measured cycles use fresh entries
|
||||
// and keep exercising eviction instead of colliding with warmup ids.
|
||||
for i in bounds.warmup..(bounds.warmup + bounds.measure) {
|
||||
run_cycle(backend, i).await;
|
||||
}
|
||||
// A baseline that never drained already poisons `before`, so skip the
|
||||
// measured-window drain and report the window as not quiesced.
|
||||
let quiesced = baseline_quiesced && quiesce(backend).await;
|
||||
|
||||
let heap_after = heap_capture();
|
||||
let after = ResourceSnapshot::capture();
|
||||
let counts = backend.registry_counts().await;
|
||||
|
||||
Measurement {
|
||||
before,
|
||||
after,
|
||||
growth: after.growth_from(&before),
|
||||
counts,
|
||||
heap: heap_before
|
||||
.zip(heap_after)
|
||||
.map(|(before, after)| HeapMetrics::new(before, after, bounds.measure)),
|
||||
quiesced,
|
||||
}
|
||||
}
|
||||
|
||||
/// Takes `expected` as a parameter so the skip arm is testable on any platform.
|
||||
fn metric_failure(
|
||||
metric: Metric,
|
||||
value: Option<usize>,
|
||||
expected: bool,
|
||||
bounds: &Bounds,
|
||||
) -> Option<String> {
|
||||
let Some(raw) = value else {
|
||||
return expected.then(|| {
|
||||
format!(
|
||||
"{}: growth sample unavailable; the soak cannot bound it",
|
||||
metric.label()
|
||||
)
|
||||
});
|
||||
};
|
||||
let growth = metric.growth_in_budget_unit(raw);
|
||||
let budget = metric.budget(bounds);
|
||||
(growth > budget).then(|| {
|
||||
let unit = metric.unit().map(|u| format!(" {u}")).unwrap_or_default();
|
||||
format!(
|
||||
"{}: grew {growth:.1}{unit} over the soak (bound {budget:.1}{unit})",
|
||||
metric.label()
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn check_bounds(bounds: &Bounds, m: &Measurement) -> Vec<String> {
|
||||
// Drain first: a non-quiesced window has nonzero counts and noisy growth, so
|
||||
// report the quiesce failure alone; the gates below only mean anything once
|
||||
// drained.
|
||||
if !m.quiesced {
|
||||
return vec![
|
||||
"quiesce budget expired before the measured window drained; soak result is unreliable"
|
||||
.to_owned(),
|
||||
];
|
||||
}
|
||||
|
||||
let mut failures = Vec::new();
|
||||
if m.counts.pending != 0 {
|
||||
failures.push(format!(
|
||||
"no subagent may remain pending, saw {}",
|
||||
m.counts.pending
|
||||
));
|
||||
}
|
||||
if m.counts.active != 0 {
|
||||
failures.push(format!(
|
||||
"no subagent may remain active, saw {}",
|
||||
m.counts.active
|
||||
));
|
||||
}
|
||||
if m.counts.completed > MAX_COMPLETED_ENTRIES {
|
||||
failures.push(format!(
|
||||
"completed retention must stay bounded by its cap, saw {}",
|
||||
m.counts.completed
|
||||
));
|
||||
}
|
||||
|
||||
for metric in Metric::iter() {
|
||||
let expected = metric.expected_on_this_platform();
|
||||
if let Some(f) = metric_failure(metric, m.growth.value_of(metric), expected, bounds) {
|
||||
failures.push(f);
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(h) = m.heap {
|
||||
let measure = bounds.measure;
|
||||
if h.blocks_per_cycle > bounds.max_blocks_per_cycle {
|
||||
failures.push(format!(
|
||||
"block-count leak: {:.3} blocks/cycle retained ({} over {measure} cycles) \
|
||||
exceeds the {} gate",
|
||||
h.blocks_per_cycle,
|
||||
h.after.blocks - h.before.blocks,
|
||||
bounds.max_blocks_per_cycle
|
||||
));
|
||||
}
|
||||
if h.bytes_per_cycle > bounds.max_bytes_per_cycle {
|
||||
failures.push(format!(
|
||||
"byte leak: {:.1} bytes/cycle retained ({} over {measure} cycles) \
|
||||
exceeds the {} gate",
|
||||
h.bytes_per_cycle,
|
||||
h.after.bytes - h.before.bytes,
|
||||
bounds.max_bytes_per_cycle
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
failures
|
||||
}
|
||||
|
||||
fn assert_bounds(bounds: &Bounds, m: &Measurement) {
|
||||
let failures = check_bounds(bounds, m);
|
||||
assert!(
|
||||
failures.is_empty(),
|
||||
"subagent soak bounds violated:\n - {}",
|
||||
failures.join("\n - ")
|
||||
);
|
||||
}
|
||||
|
||||
/// Keep this the only test in the binary that creates a `dhat::Profiler`.
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
#[ignore = "subagent soak; run with --ignored (SUBAGENT_SOAK_CYCLES bounds the measured window)"]
|
||||
async fn subagent_lifecycle_soak_bounds_threads_fds_and_heap() {
|
||||
#[cfg(feature = "dhat-heap")]
|
||||
let _profiler = dhat::Profiler::builder().testing().build();
|
||||
|
||||
let bounds = Bounds::from_env();
|
||||
|
||||
let local = tokio::task::LocalSet::new();
|
||||
local
|
||||
.run_until(async move {
|
||||
let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
|
||||
let config = CoordinatorConfig {
|
||||
foreground_budget: Duration::from_secs(600),
|
||||
..CoordinatorConfig::default()
|
||||
};
|
||||
let gate = Arc::new(tokio::sync::Semaphore::new(0));
|
||||
tokio::task::spawn_local(
|
||||
SubagentCoordinator::new(command_rx, SoakRunner { gate: gate.clone() }, config)
|
||||
.run(),
|
||||
);
|
||||
let backend = ChannelBackend::new(command_tx);
|
||||
|
||||
let warmup_quiesced = warmup(&backend, bounds.warmup).await;
|
||||
// Drain the concurrent phase into the baseline; a failed drain marks
|
||||
// the window unreliable.
|
||||
concurrent_phase(&backend, &gate, bounds.concurrency).await;
|
||||
let baseline_quiesced = warmup_quiesced && quiesce(&backend).await;
|
||||
let measurement = measure(&backend, &bounds, baseline_quiesced).await;
|
||||
|
||||
let summary = Summary {
|
||||
bounds: &bounds,
|
||||
measurement: &measurement,
|
||||
};
|
||||
eprintln!(
|
||||
"SUBAGENT_SOAK_SUMMARY {}",
|
||||
serde_json::to_string(&summary).expect("summary serializes")
|
||||
);
|
||||
|
||||
assert_bounds(&bounds, &measurement);
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
mod tests {
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[test]
|
||||
fn value_of_reads_the_matching_slot_of_snapshot_and_growth() {
|
||||
let snapshot = ResourceSnapshot {
|
||||
rss: Some(11),
|
||||
threads: Some(22),
|
||||
fds: Some(33),
|
||||
};
|
||||
assert_eq!(snapshot.value_of(Metric::Rss), Some(11));
|
||||
assert_eq!(snapshot.value_of(Metric::Threads), Some(22));
|
||||
assert_eq!(snapshot.value_of(Metric::Fds), Some(33));
|
||||
|
||||
let growth = ResourceGrowth {
|
||||
rss: Some(1),
|
||||
threads: None,
|
||||
fds: Some(3),
|
||||
};
|
||||
assert_eq!(growth.value_of(Metric::Rss), Some(1));
|
||||
assert_eq!(growth.value_of(Metric::Threads), None);
|
||||
assert_eq!(growth.value_of(Metric::Fds), Some(3));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serialize_metrics_keys_match_summary_keys_in_order() {
|
||||
#[derive(Serialize)]
|
||||
struct Wrap(#[serde(serialize_with = "serialize_metrics")] ResourceSnapshot);
|
||||
let snapshot = ResourceSnapshot {
|
||||
rss: Some(1),
|
||||
threads: None,
|
||||
fds: Some(3),
|
||||
};
|
||||
let json = serde_json::to_string(&Wrap(snapshot)).expect("snapshot serializes");
|
||||
assert_eq!(json, r#"{"rss_bytes":1,"threads":null,"fds":3}"#);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bytes_to_mib_divides_by_1024_squared() {
|
||||
assert_eq!(bytes_to_mib(0), 0.0);
|
||||
assert_eq!(bytes_to_mib(1024 * 1024), 1.0);
|
||||
assert_eq!(bytes_to_mib(3 * 1024 * 1024), 3.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn growth_in_budget_unit_scales_only_rss() {
|
||||
assert_eq!(Metric::Rss.growth_in_budget_unit(2 * 1024 * 1024), 2.0);
|
||||
assert_eq!(Metric::Threads.growth_in_budget_unit(7), 7.0);
|
||||
assert_eq!(Metric::Fds.growth_in_budget_unit(7), 7.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn budget_reads_per_metric_bound() {
|
||||
let bounds = Bounds {
|
||||
warmup: 0,
|
||||
measure: 0,
|
||||
concurrency: 0,
|
||||
max_thread_growth: 3,
|
||||
max_fd_growth: 5,
|
||||
max_rss_growth_mib: 7,
|
||||
max_blocks_per_cycle: 1.0,
|
||||
max_bytes_per_cycle: 2.0,
|
||||
};
|
||||
assert_eq!(Metric::Rss.budget(&bounds), 7.0);
|
||||
assert_eq!(Metric::Threads.budget(&bounds), 3.0);
|
||||
assert_eq!(Metric::Fds.budget(&bounds), 5.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn heap_metrics_clamps_zero_cycles() {
|
||||
let before = HeapSample {
|
||||
blocks: 10,
|
||||
bytes: 100,
|
||||
};
|
||||
let after = HeapSample {
|
||||
blocks: 20,
|
||||
bytes: 400,
|
||||
};
|
||||
let heap = HeapMetrics::new(before, after, 0);
|
||||
assert!(heap.blocks_per_cycle.is_finite());
|
||||
assert!(heap.bytes_per_cycle.is_finite());
|
||||
assert_eq!(heap.blocks_per_cycle, 10.0);
|
||||
assert_eq!(heap.bytes_per_cycle, 300.0);
|
||||
}
|
||||
|
||||
fn generous_bounds() -> Bounds {
|
||||
Bounds {
|
||||
warmup: 0,
|
||||
measure: 4,
|
||||
concurrency: 4,
|
||||
max_thread_growth: 100,
|
||||
max_fd_growth: 100,
|
||||
max_rss_growth_mib: 100,
|
||||
max_blocks_per_cycle: 10.0,
|
||||
max_bytes_per_cycle: 10_000.0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Zero growth that reads as measured, unlike `ResourceGrowth::default()`.
|
||||
fn zero_growth() -> ResourceGrowth {
|
||||
ResourceGrowth {
|
||||
rss: Some(0),
|
||||
threads: Some(0),
|
||||
fds: Some(0),
|
||||
}
|
||||
}
|
||||
|
||||
fn drained(growth: ResourceGrowth, heap: Option<HeapMetrics>) -> Measurement {
|
||||
Measurement {
|
||||
before: ResourceSnapshot::default(),
|
||||
after: ResourceSnapshot::default(),
|
||||
growth,
|
||||
counts: SubagentRegistryCounts {
|
||||
pending: 0,
|
||||
active: 0,
|
||||
completed: 0,
|
||||
},
|
||||
heap,
|
||||
quiesced: true,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_passes_a_clean_drained_window() {
|
||||
let m = drained(zero_growth(), None);
|
||||
assert!(check_bounds(&generous_bounds(), &m).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_fails_when_an_expected_metric_is_unavailable() {
|
||||
let growth = ResourceGrowth {
|
||||
rss: None,
|
||||
threads: Some(0),
|
||||
fds: Some(0),
|
||||
};
|
||||
let failures = check_bounds(&generous_bounds(), &drained(growth, None));
|
||||
assert!(
|
||||
failures
|
||||
.iter()
|
||||
.any(|f| f.starts_with("rss:") && f.contains("unavailable")),
|
||||
"{failures:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metric_failure_covers_expected_missing_unexpected_missing_and_budget() {
|
||||
let b = generous_bounds();
|
||||
assert!(metric_failure(Metric::Threads, None, false, &b).is_none());
|
||||
assert!(
|
||||
metric_failure(Metric::Rss, None, true, &b)
|
||||
.unwrap()
|
||||
.contains("unavailable")
|
||||
);
|
||||
assert!(metric_failure(Metric::Fds, Some(0), true, &b).is_none());
|
||||
assert!(
|
||||
metric_failure(Metric::Rss, Some(500 * 1024 * 1024), true, &b)
|
||||
.unwrap()
|
||||
.starts_with("rss:")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_reports_non_quiesce_first_and_alone() {
|
||||
let mut m = drained(zero_growth(), None);
|
||||
m.quiesced = false;
|
||||
m.counts.pending = 3;
|
||||
let failures = check_bounds(&generous_bounds(), &m);
|
||||
assert_eq!(failures.len(), 1);
|
||||
assert!(failures[0].contains("quiesce"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_flags_over_budget_growth() {
|
||||
let growth = ResourceGrowth {
|
||||
rss: Some(200 * 1024 * 1024),
|
||||
threads: Some(0),
|
||||
fds: Some(0),
|
||||
};
|
||||
let failures = check_bounds(&generous_bounds(), &drained(growth, None));
|
||||
assert!(
|
||||
failures.iter().any(|f| f.starts_with("rss:")),
|
||||
"{failures:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_treats_the_budget_as_an_inclusive_max() {
|
||||
let growth = ResourceGrowth {
|
||||
rss: Some(100 * 1024 * 1024),
|
||||
threads: Some(100),
|
||||
fds: Some(100),
|
||||
};
|
||||
assert!(check_bounds(&generous_bounds(), &drained(growth, None)).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_flags_nonzero_counts_and_heap_leak() {
|
||||
let mut m = drained(
|
||||
zero_growth(),
|
||||
Some(HeapMetrics {
|
||||
before: HeapSample {
|
||||
blocks: 0,
|
||||
bytes: 0,
|
||||
},
|
||||
after: HeapSample {
|
||||
blocks: 0,
|
||||
bytes: 0,
|
||||
},
|
||||
blocks_per_cycle: 0.0,
|
||||
bytes_per_cycle: 1_000_000.0,
|
||||
}),
|
||||
);
|
||||
m.counts.active = 2;
|
||||
let failures = check_bounds(&generous_bounds(), &m);
|
||||
assert!(
|
||||
failures.iter().any(|f| f.contains("active")),
|
||||
"{failures:?}"
|
||||
);
|
||||
assert!(
|
||||
failures.iter().any(|f| f.contains("byte leak")),
|
||||
"{failures:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_flags_pending_while_quiesced() {
|
||||
let mut m = drained(zero_growth(), None);
|
||||
m.counts.pending = 3;
|
||||
let failures = check_bounds(&generous_bounds(), &m);
|
||||
assert!(
|
||||
failures.iter().any(|f| f.contains("pending")),
|
||||
"{failures:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_flags_completed_over_cap() {
|
||||
let mut m = drained(zero_growth(), None);
|
||||
m.counts.completed = MAX_COMPLETED_ENTRIES + 1;
|
||||
let failures = check_bounds(&generous_bounds(), &m);
|
||||
assert!(
|
||||
failures.iter().any(|f| f.contains("completed retention")),
|
||||
"{failures:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_flags_thread_and_fd_over_budget() {
|
||||
let growth = ResourceGrowth {
|
||||
rss: Some(0),
|
||||
threads: Some(200),
|
||||
fds: Some(200),
|
||||
};
|
||||
let failures = check_bounds(&generous_bounds(), &drained(growth, None));
|
||||
assert!(
|
||||
failures.iter().any(|f| f.starts_with("threads:")),
|
||||
"{failures:?}"
|
||||
);
|
||||
assert!(
|
||||
failures.iter().any(|f| f.starts_with("fds:")),
|
||||
"{failures:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn check_bounds_flags_block_count_leak() {
|
||||
let m = drained(
|
||||
zero_growth(),
|
||||
Some(HeapMetrics {
|
||||
before: HeapSample {
|
||||
blocks: 0,
|
||||
bytes: 0,
|
||||
},
|
||||
after: HeapSample {
|
||||
blocks: 0,
|
||||
bytes: 0,
|
||||
},
|
||||
blocks_per_cycle: 50.0,
|
||||
bytes_per_cycle: 0.0,
|
||||
}),
|
||||
);
|
||||
let failures = check_bounds(&generous_bounds(), &m);
|
||||
assert!(
|
||||
failures.iter().any(|f| f.contains("block-count leak")),
|
||||
"{failures:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue