use super::support::*;
use crate::{
agent::steering::{AgentSteering, SteeringRejected},
subagents::control::SubagentControls,
};
const STEERING: &str = "Use $review and report only the changed files";
struct SteeredChildProvider {
controls: SubagentControls,
requests: Mutex<Vec<ProviderRequest>>,
sessions_root: PathBuf,
}
impl Provider for SteeredChildProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let mut requests = self.requests.lock().unwrap();
requests.push(request);
if requests.len() == 1 {
let control = self.controls.get(&ActivityId::new("batch/g1")).unwrap();
assert!(control.is_active());
control.steering.try_enqueue(STEERING.into()).unwrap();
} else {
assert!(
requests
.last()
.unwrap()
.messages()
.iter()
.any(|message| message.content.contains(STEERING))
);
let persisted = std::fs::read_dir(self.sessions_root.join("subagents"))
.unwrap()
.filter_map(Result::ok)
.filter(|entry| {
entry
.path()
.extension()
.is_some_and(|extension| extension == "jsonl")
})
.any(|entry| {
std::fs::read_to_string(entry.path())
.unwrap()
.contains(STEERING)
});
assert!(
persisted,
"steering must be durable before the next provider request"
);
}
on_event(ProviderEvent::TextDelta(
if requests.len() == 1 {
"initial answer"
} else {
"changed files: main.rs"
}
.into(),
))?;
on_event(ProviderEvent::Done)
}
}
fn run_child(id: &str, cfg: &SubagentRunConfig) -> SubagentTaskResult {
run_one_subagent(SubagentRunInput {
id: id.into(),
task: SubagentTask {
intent: "inspect changes".into(),
agent: None,
identity: None,
context: None,
cwd: None,
},
cwd: cfg.parent_cwd.clone(),
config: cfg,
cancellation: cfg.cancellation.clone(),
batch_id: &ActivityId::new("batch"),
reporter: None,
finisher: TaskActivityFinisher::new(),
})
}
#[test]
fn child_steering_is_persisted_before_provider_continuation_and_registry_is_removed() {
let temp = tempfile::tempdir().unwrap();
let controls = SubagentControls::default();
let provider = Arc::new(SteeredChildProvider {
controls: controls.clone(),
requests: Mutex::new(Vec::new()),
sessions_root: temp.path().join("sessions"),
});
let mut cfg = config(provider.clone(), temp.path());
cfg.sessions_root = Some(provider.sessions_root.clone());
cfg.subagent_controls = Some(controls.clone());
let result = run_child("g1", &cfg);
assert_eq!(
result.status,
SubagentStatus::Completed,
"{:?}",
result.error
);
assert!(result.output.contains("changed files: main.rs"));
assert_eq!(provider.requests.lock().unwrap().len(), 2);
assert!(controls.get(&ActivityId::new("batch/g1")).is_none());
}
struct FailedSteeredChildProvider(SubagentControls);
impl Provider for FailedSteeredChildProvider {
fn stream_cancellable(
&self,
_request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let control = self.0.get(&ActivityId::new("batch/g1")).unwrap();
control.steering.try_enqueue(STEERING.into()).unwrap();
on_event(ProviderEvent::TextDelta("partial answer".into()))?;
anyhow::bail!("child failed after accepting steering")
}
}
#[test]
fn failed_child_retains_pending_steering_for_readonly_recall() {
let temp = tempfile::tempdir().unwrap();
let controls = SubagentControls::default();
let mut cfg = config(
Arc::new(FailedSteeredChildProvider(controls.clone())),
temp.path(),
);
cfg.subagent_controls = Some(controls.clone());
assert_eq!(run_child("g1", &cfg).status, SubagentStatus::Failed);
let control = controls.get(&ActivityId::new("batch/g1")).unwrap();
assert!(!control.is_active());
assert!(controls.has_pending_input());
assert_eq!(
control.steering.try_enqueue_nonblocking("restart".into()),
Err(SteeringRejected::Closed)
);
assert_eq!(
control.steering.restore_pending_messages(|text| {
assert_eq!(text, STEERING);
true
}),
Ok(true)
);
assert!(!controls.has_pending_input());
}
struct SelectiveCancelProvider {
controls: SubagentControls,
selected: Mutex<Option<crate::subagents::control::SubagentControl>>,
calls: AtomicUsize,
}
impl Provider for SelectiveCancelProvider {
fn stream_cancellable(
&self,
_request: ProviderRequest,
cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
let control = self.controls.get(&ActivityId::new("batch/g1")).unwrap();
control.cancel();
*self.selected.lock().unwrap() = Some(control);
cancellation.check()?;
panic!("selected child must be canceled");
}
cancellation.check()?;
on_event(ProviderEvent::TextDelta("sibling completed".into()))?;
on_event(ProviderEvent::Done)
}
}
#[test]
fn selected_child_cancel_preserves_parent_and_sibling_and_closes_stale_handle() {
let temp = tempfile::tempdir().unwrap();
let controls = SubagentControls::default();
let provider = Arc::new(SelectiveCancelProvider {
controls: controls.clone(),
selected: Mutex::new(None),
calls: AtomicUsize::new(0),
});
let mut cfg = config(provider.clone(), temp.path());
cfg.subagent_controls = Some(controls.clone());
let events = Arc::new(Mutex::new(Vec::new()));
let captured = events.clone();
cfg.activity_sender = Some(Arc::new(move |event| captured.lock().unwrap().push(event)));
let canceled = run_child("g1", &cfg);
assert_eq!(canceled.status, SubagentStatus::Failed);
assert!(!cfg.cancellation.is_canceled());
let sibling = run_child("g2", &cfg);
assert_eq!(sibling.status, SubagentStatus::Completed);
assert_eq!(sibling.output, "sibling completed");
let selected = provider.selected.lock().unwrap().take().unwrap();
assert!(!selected.is_active());
assert_eq!(
selected.steering.try_enqueue("too late".into()),
Err(SteeringRejected::Closed)
);
assert!(controls.get(&ActivityId::new("batch/g1")).is_none());
assert!(events.lock().unwrap().iter().any(|event| matches!(event, ActivityEvent::Finished { id, status: ActivityStatus::Canceled, .. } if id == &ActivityId::new("batch/g1"))));
}
#[test]
fn final_enqueue_is_either_reserved_for_persistence_or_rejected_as_closed() {
for _ in 0..64 {
let steering = AgentSteering::new();
let producer = steering.clone();
let barrier = Arc::new(std::sync::Barrier::new(2));
let producer_barrier = barrier.clone();
let enqueue = thread::spawn(move || {
producer_barrier.wait();
producer.try_enqueue("late input".into())
});
barrier.wait();
let reservation = steering.reserve_collapsed_or_close();
match enqueue.join().unwrap() {
Ok(_) => {
let reservation = reservation.expect("accepted input must be reserved, not lost");
assert!(reservation.text().contains("late input"));
assert!(
steering
.acknowledge_reserved_prompt(reservation.text())
.is_some()
);
drop(reservation);
assert!(steering.reserve_collapsed_or_close().is_none());
}
Err(error) => {
assert_eq!(error, SteeringRejected::Closed);
assert!(reservation.is_none());
}
}
assert_eq!(
steering.try_enqueue("after completion".into()),
Err(SteeringRejected::Closed)
);
}
}
#[test]
fn nested_children_share_registry_with_distinct_runtime_activity_ids() {
let temp = tempfile::tempdir().unwrap();
let controls = SubagentControls::default();
let mut cfg = config(Arc::new(NestedSubagentsProvider::new()), temp.path());
cfg.subagent_controls = Some(controls.clone());
let registered = Arc::new(Mutex::new(BTreeSet::new()));
let captured = registered.clone();
let registry = controls.clone();
cfg.activity_sender = Some(Arc::new(move |event| {
if let ActivityEvent::Started {
id,
kind: ActivityKind::SubagentTask,
..
} = event
&& let Some(control) = registry.get(&id)
{
assert!(control.is_active());
captured.lock().unwrap().insert(id);
}
}));
let result = run_child("g1", &cfg);
assert_eq!(
result.status,
SubagentStatus::Completed,
"{:?}",
result.error
);
let registered = registered.lock().unwrap();
assert_eq!(
registered.len(),
2,
"parent child and nested child must both be addressable"
);
for id in registered.iter() {
assert!(controls.get(id).is_none());
}
}
#[test]
fn parent_cancellation_reaches_registered_descendants_and_unwind_closes_handles() {
let controls = SubagentControls::default();
let flag = Arc::new(AtomicBool::new(false));
let parent = AgentCancellation::new(flag.clone());
let (child, handle) = parent.child_token();
let (descendant, _) = child.child_token();
let id = ActivityId::new("child");
let registration = controls.register(id.clone(), AgentSteering::new(), handle, child.clone());
let control = controls.get(&id).unwrap();
flag.store(true, Ordering::SeqCst);
assert!(child.check().is_err());
assert!(descendant.check().is_err());
assert!(!control.is_active());
assert!(
std::panic::catch_unwind(move || {
let _registration = registration;
panic!("child worker panic");
})
.is_err()
);
assert!(!control.is_active());
assert!(controls.get(&id).is_none());
assert_eq!(
control.steering.try_enqueue("after panic".into()),
Err(SteeringRejected::Closed)
);
}