1use crate::child_turn;
4use crate::session_runner::AgentSession;
5use anyhow::{Context as _, Result};
6use kiss_agent::{AgentMessage, AgentTool, DynTool, ToolResult, ToolUpdateSink};
7use kiss_ai::StopReason;
8use serde::{Deserialize, Serialize};
9use serde_json::{Value, json};
10use std::collections::HashMap;
11use std::sync::{Arc, Mutex, Weak};
12use std::time::Duration;
13use tokio::sync::{Semaphore, watch};
14use tokio_util::sync::CancellationToken;
15
16const MAX_ACTIVE_TURNS: usize = 4;
17const MAX_AGENT_RECORDS: usize = 16;
18const DEFAULT_WAIT_MS: u64 = 30_000;
19const MIN_WAIT_MS: u64 = 250;
20const MAX_WAIT_MS: u64 = 600_000;
21
22pub const SUBAGENT_SYSTEM_PROMPT: &str = "Subagent coordination:\n- Subagents share this working directory. Give each child one bounded task.\n- Fresh child context is the default. Copy parent turns only when the task needs them.\n- Use wait_agent when a child result is required. Do not use repeated list calls as polling.\n- Check child findings and edits before you give the final answer.\n- Start subagents only when the user or the current task justifies delegation.";
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
25#[serde(rename_all = "snake_case")]
26pub enum AgentStatus {
27 Queued,
28 Running,
29 Completed,
30 Failed,
31 Interrupted,
32}
33
34impl AgentStatus {
35 fn is_terminal(self) -> bool {
36 matches!(self, Self::Completed | Self::Failed | Self::Interrupted)
37 }
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
41#[serde(rename_all = "camelCase")]
42pub struct AgentInfo {
43 pub id: String,
44 pub task_name: String,
45 pub canonical_path: String,
46 pub status: AgentStatus,
47 #[serde(skip_serializing_if = "Option::is_none")]
48 pub result: Option<String>,
49 #[serde(skip_serializing_if = "Option::is_none")]
50 pub error: Option<String>,
51}
52
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub enum ForkTurns {
55 None,
56 All,
57 Recent(usize),
58}
59
60impl ForkTurns {
61 fn parse(value: Option<&Value>) -> Result<Self> {
62 match value {
63 None => Ok(Self::None),
64 Some(Value::String(value)) if value == "none" => Ok(Self::None),
65 Some(Value::String(value)) if value == "all" => Ok(Self::All),
66 Some(Value::Number(value)) => {
67 let count = value
68 .as_u64()
69 .filter(|count| *count > 0)
70 .and_then(|count| usize::try_from(count).ok())
71 .context("fork_turns must be 'none', 'all', or a positive integer")?;
72 Ok(Self::Recent(count))
73 }
74 _ => anyhow::bail!("fork_turns must be 'none', 'all', or a positive integer"),
75 }
76 }
77}
78
79struct AgentState {
80 status: AgentStatus,
81 result: Option<String>,
82 error: Option<String>,
83}
84
85struct AgentRecord {
86 id: String,
87 task_name: String,
88 canonical_path: String,
89 session: Arc<AgentSession>,
90 state: Mutex<AgentState>,
91}
92
93impl AgentRecord {
94 fn snapshot(&self) -> AgentInfo {
95 let state = self.state.lock().unwrap();
96 AgentInfo {
97 id: self.id.clone(),
98 task_name: self.task_name.clone(),
99 canonical_path: self.canonical_path.clone(),
100 status: state.status,
101 result: state.result.clone(),
102 error: state.error.clone(),
103 }
104 }
105}
106
107pub struct SubagentRuntime {
108 parent: Weak<AgentSession>,
109 records: Mutex<Vec<Arc<AgentRecord>>>,
110 targets: Mutex<HashMap<String, Arc<AgentRecord>>>,
111 permits: Arc<Semaphore>,
112 activity: watch::Sender<u64>,
113 spawn_lock: Mutex<()>,
114}
115
116impl SubagentRuntime {
117 pub(crate) fn new(parent: Weak<AgentSession>) -> Arc<Self> {
118 Self::with_permits(parent, MAX_ACTIVE_TURNS)
119 }
120
121 fn with_permits(parent: Weak<AgentSession>, permits: usize) -> Arc<Self> {
122 let (activity, _) = watch::channel(0);
123 Arc::new(Self {
124 parent,
125 records: Mutex::new(Vec::new()),
126 targets: Mutex::new(HashMap::new()),
127 permits: Arc::new(Semaphore::new(permits)),
128 activity,
129 spawn_lock: Mutex::new(()),
130 })
131 }
132
133 pub(crate) fn control_tools(self: &Arc<Self>) -> Vec<DynTool> {
134 vec![
135 Arc::new(SpawnAgentTool(self.clone())),
136 Arc::new(SendMessageTool(self.clone())),
137 Arc::new(FollowupTaskTool(self.clone())),
138 Arc::new(WaitAgentTool(self.clone())),
139 Arc::new(ListAgentsTool(self.clone())),
140 Arc::new(InterruptAgentTool(self.clone())),
141 ]
142 }
143
144 fn signal_activity(&self) {
145 self.activity
146 .send_modify(|version| *version = version.wrapping_add(1));
147 }
148
149 fn resolve(&self, target: &str) -> Result<Arc<AgentRecord>> {
150 self.targets
151 .lock()
152 .unwrap()
153 .get(target)
154 .cloned()
155 .with_context(|| format!("unknown subagent target '{target}'"))
156 }
157
158 fn list(&self) -> Vec<AgentInfo> {
159 self.records
160 .lock()
161 .unwrap()
162 .iter()
163 .map(|record| record.snapshot())
164 .collect()
165 }
166
167 fn spawn(
168 self: &Arc<Self>,
169 task_name: String,
170 prompt: String,
171 fork_turns: ForkTurns,
172 model: Option<String>,
173 reasoning_effort: Option<String>,
174 ) -> Result<AgentInfo> {
175 let _spawn_guard = self.spawn_lock.lock().unwrap();
176 validate_task_name(&task_name)?;
177 if prompt.trim().is_empty() {
178 anyhow::bail!("prompt must not be empty");
179 }
180 {
181 let records = self.records.lock().unwrap();
182 if records.len() >= MAX_AGENT_RECORDS {
183 anyhow::bail!("the subagent limit of {MAX_AGENT_RECORDS} has been reached");
184 }
185 if records.iter().any(|record| record.task_name == task_name) {
186 anyhow::bail!(
187 "task name '{task_name}' already exists. Use followup_task for that child"
188 );
189 }
190 }
191
192 let canonical_path = format!("/root/{task_name}");
193 let parent = self
194 .parent
195 .upgrade()
196 .context("the parent session is no longer available")?;
197 let child = parent.create_subagent_session(
198 &task_name,
199 &canonical_path,
200 fork_turns,
201 model.as_deref(),
202 reasoning_effort.as_deref(),
203 )?;
204 let id = child.manager.lock().unwrap().session_id().to_string();
205 let record = Arc::new(AgentRecord {
206 id: id.clone(),
207 task_name: task_name.clone(),
208 canonical_path: canonical_path.clone(),
209 session: child,
210 state: Mutex::new(AgentState {
211 status: AgentStatus::Queued,
212 result: None,
213 error: None,
214 }),
215 });
216 self.records.lock().unwrap().push(record.clone());
217 let mut targets = self.targets.lock().unwrap();
218 for target in [&id, &task_name, &canonical_path] {
219 targets.insert(target.clone(), record.clone());
220 }
221 drop(targets);
222
223 let info = record.snapshot();
224 self.signal_activity();
225 self.start_turn(record, prompt);
226 Ok(info)
227 }
228
229 fn start_turn(self: &Arc<Self>, record: Arc<AgentRecord>, prompt: String) {
230 let runtime = self.clone();
231 tokio::spawn(async move {
232 let permit = match runtime.permits.clone().acquire_owned().await {
233 Ok(permit) => permit,
234 Err(_) => return,
235 };
236 {
237 let mut state = record.state.lock().unwrap();
238 if state.status == AgentStatus::Interrupted {
239 return;
240 }
241 state.status = AgentStatus::Running;
242 }
243 runtime.signal_activity();
244
245 let outcome =
246 child_turn::run_child_turn(&runtime.parent, &record.session, prompt, None, None)
247 .await;
248 drop(permit);
249
250 let interrupted = record.state.lock().unwrap().status == AgentStatus::Interrupted;
251 if !interrupted {
252 let mut state = record.state.lock().unwrap();
253 state.status = outcome.status;
254 state.result = outcome.result;
255 state.error = outcome.error;
256 }
257 runtime.signal_activity();
258 });
259 }
260
261 fn send_message(&self, target: &str, message: String) -> Result<AgentInfo> {
262 if message.trim().is_empty() {
263 anyhow::bail!("message must not be empty");
264 }
265 let record = self.resolve(target)?;
266 let status = record.state.lock().unwrap().status;
267 if !matches!(status, AgentStatus::Queued | AgentStatus::Running) {
268 anyhow::bail!(
269 "subagent '{}' is {}. Use followup_task to start another turn",
270 record.task_name,
271 status_name(status)
272 );
273 }
274 record.session.queue_steering(AgentMessage::user(message));
275 self.signal_activity();
276 Ok(record.snapshot())
277 }
278
279 fn followup(self: &Arc<Self>, target: &str, prompt: String) -> Result<AgentInfo> {
280 if prompt.trim().is_empty() {
281 anyhow::bail!("prompt must not be empty");
282 }
283 let record = self.resolve(target)?;
284 {
285 let mut state = record.state.lock().unwrap();
286 if !state.status.is_terminal() {
287 anyhow::bail!("subagent '{}' is still busy", record.task_name);
288 }
289 state.status = AgentStatus::Queued;
290 state.result = None;
291 state.error = None;
292 }
293 let _ = record.session.reclaim_queued();
294 let info = record.snapshot();
295 self.signal_activity();
296 self.start_turn(record, prompt);
297 Ok(info)
298 }
299
300 async fn wait(
301 &self,
302 targets: &[String],
303 timeout_ms: u64,
304 cancel: CancellationToken,
305 ) -> Result<WaitOutcome> {
306 if targets.is_empty() {
307 anyhow::bail!("targets must contain at least one subagent");
308 }
309 let records = targets
310 .iter()
311 .map(|target| self.resolve(target))
312 .collect::<Result<Vec<_>>>()?;
313 let timeout_ms = timeout_ms.clamp(MIN_WAIT_MS, MAX_WAIT_MS);
314 let deadline = tokio::time::Instant::now() + Duration::from_millis(timeout_ms);
315 let mut activity = self.activity.subscribe();
316
317 loop {
318 let agents = records
319 .iter()
320 .map(|record| record.snapshot())
321 .collect::<Vec<_>>();
322 if agents.iter().any(|agent| agent.status.is_terminal()) {
323 return Ok(WaitOutcome {
324 timed_out: false,
325 agents,
326 });
327 }
328 let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
329 if remaining.is_zero() {
330 return Ok(WaitOutcome {
331 timed_out: true,
332 agents,
333 });
334 }
335 tokio::select! {
336 _ = cancel.cancelled() => anyhow::bail!("wait cancelled"),
337 result = tokio::time::timeout(remaining, activity.changed()) => {
338 if result.is_err() {
339 return Ok(WaitOutcome { timed_out: true, agents });
340 }
341 }
342 }
343 }
344 }
345
346 fn interrupt(&self, target: &str) -> Result<AgentInfo> {
347 let record = self.resolve(target)?;
348 interrupt_record(&record);
349 self.signal_activity();
350 Ok(record.snapshot())
351 }
352
353 pub(crate) fn interrupt_all(&self) {
354 let records = self.records.lock().unwrap().clone();
355 for record in records {
356 interrupt_record(&record);
357 }
358 self.signal_activity();
359 }
360
361 pub(crate) fn reset(&self) {
362 self.interrupt_all();
363 self.records.lock().unwrap().clear();
364 self.targets.lock().unwrap().clear();
365 self.signal_activity();
366 }
367}
368
369fn interrupt_record(record: &AgentRecord) {
370 let mut state = record.state.lock().unwrap();
371 if matches!(state.status, AgentStatus::Queued | AgentStatus::Running) {
372 state.status = AgentStatus::Interrupted;
373 state.result = None;
374 state.error = Some("interrupted".into());
375 record.session.abort();
376 }
377}
378
379pub(crate) fn turn_outcome(
380 session: &AgentSession,
381) -> (AgentStatus, Option<String>, Option<String>) {
382 let manager = session.manager.lock().unwrap();
383 let assistant = manager
384 .build_session_context()
385 .messages
386 .into_iter()
387 .rev()
388 .find_map(|message| match message {
389 AgentMessage::Assistant(assistant) => Some(assistant),
390 _ => None,
391 });
392 let Some(assistant) = assistant else {
393 return (
394 AgentStatus::Failed,
395 None,
396 Some("the child returned no assistant message".into()),
397 );
398 };
399 match assistant.stop_reason {
400 StopReason::Error => (
401 AgentStatus::Failed,
402 None,
403 Some(
404 assistant
405 .error_message
406 .unwrap_or_else(|| "the child request failed".into()),
407 ),
408 ),
409 StopReason::Aborted => (AgentStatus::Interrupted, None, Some("interrupted".into())),
410 _ => {
411 let text = assistant.text();
412 if text.trim().is_empty() {
413 (
414 AgentStatus::Failed,
415 None,
416 Some("the child returned an empty result".into()),
417 )
418 } else {
419 (AgentStatus::Completed, Some(text), None)
420 }
421 }
422 }
423}
424
425fn validate_task_name(name: &str) -> Result<()> {
426 if name.is_empty()
427 || name.len() > 64
428 || !name
429 .bytes()
430 .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
431 {
432 anyhow::bail!(
433 "task_name must use 1 to 64 lower-case ASCII letters, digits, or underscores"
434 );
435 }
436 Ok(())
437}
438
439fn status_name(status: AgentStatus) -> &'static str {
440 match status {
441 AgentStatus::Queued => "queued",
442 AgentStatus::Running => "running",
443 AgentStatus::Completed => "completed",
444 AgentStatus::Failed => "failed",
445 AgentStatus::Interrupted => "interrupted",
446 }
447}
448
449#[derive(Serialize)]
450#[serde(rename_all = "camelCase")]
451struct WaitOutcome {
452 timed_out: bool,
453 agents: Vec<AgentInfo>,
454}
455
456fn json_result(value: &impl Serialize) -> Result<ToolResult> {
457 Ok(ToolResult::text(serde_json::to_string_pretty(value)?))
458}
459
460#[derive(Deserialize)]
461struct SpawnArgs {
462 task_name: String,
463 prompt: String,
464 #[serde(default)]
465 fork_turns: Option<Value>,
466 #[serde(default)]
467 model: Option<String>,
468 #[serde(default)]
469 reasoning_effort: Option<String>,
470}
471
472struct SpawnAgentTool(Arc<SubagentRuntime>);
473
474#[async_trait::async_trait]
475impl AgentTool for SpawnAgentTool {
476 fn name(&self) -> &str {
477 "spawn_agent"
478 }
479
480 fn description(&self) -> String {
481 "Start a named child coding agent in the background. Fresh context is the default.".into()
482 }
483
484 fn parameters(&self) -> Value {
485 json!({
486 "type": "object",
487 "properties": {
488 "task_name": {"type": "string", "pattern": "^[a-z0-9_]{1,64}$"},
489 "prompt": {"type": "string", "minLength": 1},
490 "fork_turns": {"oneOf": [
491 {"type": "string", "enum": ["none", "all"]},
492 {"type": "integer", "minimum": 1}
493 ]},
494 "model": {"type": "string"},
495 "reasoning_effort": {"type": "string"}
496 },
497 "required": ["task_name", "prompt"],
498 "additionalProperties": false
499 })
500 }
501
502 async fn execute(
503 &self,
504 _tool_call_id: &str,
505 args: Value,
506 _cancel: CancellationToken,
507 _on_update: Option<ToolUpdateSink>,
508 ) -> Result<ToolResult> {
509 let args: SpawnArgs = serde_json::from_value(args)?;
510 let fork_turns = ForkTurns::parse(args.fork_turns.as_ref())?;
511 json_result(&self.0.spawn(
512 args.task_name,
513 args.prompt,
514 fork_turns,
515 args.model,
516 args.reasoning_effort,
517 )?)
518 }
519}
520
521#[derive(Deserialize)]
522struct MessageArgs {
523 target: String,
524 message: String,
525}
526
527struct SendMessageTool(Arc<SubagentRuntime>);
528
529#[async_trait::async_trait]
530impl AgentTool for SendMessageTool {
531 fn name(&self) -> &str {
532 "send_message"
533 }
534
535 fn description(&self) -> String {
536 "Send steering guidance to a queued or running child agent.".into()
537 }
538
539 fn parameters(&self) -> Value {
540 json!({
541 "type": "object",
542 "properties": {
543 "target": {"type": "string"},
544 "message": {"type": "string", "minLength": 1}
545 },
546 "required": ["target", "message"],
547 "additionalProperties": false
548 })
549 }
550
551 async fn execute(
552 &self,
553 _tool_call_id: &str,
554 args: Value,
555 _cancel: CancellationToken,
556 _on_update: Option<ToolUpdateSink>,
557 ) -> Result<ToolResult> {
558 let args: MessageArgs = serde_json::from_value(args)?;
559 json_result(&self.0.send_message(&args.target, args.message)?)
560 }
561}
562
563#[derive(Deserialize)]
564struct FollowupArgs {
565 target: String,
566 prompt: String,
567}
568
569struct FollowupTaskTool(Arc<SubagentRuntime>);
570
571#[async_trait::async_trait]
572impl AgentTool for FollowupTaskTool {
573 fn name(&self) -> &str {
574 "followup_task"
575 }
576
577 fn description(&self) -> String {
578 "Start a new turn in an idle child agent and keep its existing context.".into()
579 }
580
581 fn parameters(&self) -> Value {
582 json!({
583 "type": "object",
584 "properties": {
585 "target": {"type": "string"},
586 "prompt": {"type": "string", "minLength": 1}
587 },
588 "required": ["target", "prompt"],
589 "additionalProperties": false
590 })
591 }
592
593 async fn execute(
594 &self,
595 _tool_call_id: &str,
596 args: Value,
597 _cancel: CancellationToken,
598 _on_update: Option<ToolUpdateSink>,
599 ) -> Result<ToolResult> {
600 let args: FollowupArgs = serde_json::from_value(args)?;
601 json_result(&self.0.followup(&args.target, args.prompt)?)
602 }
603}
604
605#[derive(Deserialize)]
606struct WaitArgs {
607 targets: Vec<String>,
608 #[serde(default = "default_wait_ms")]
609 timeout_ms: u64,
610}
611
612fn default_wait_ms() -> u64 {
613 DEFAULT_WAIT_MS
614}
615
616struct WaitAgentTool(Arc<SubagentRuntime>);
617
618#[async_trait::async_trait]
619impl AgentTool for WaitAgentTool {
620 fn name(&self) -> &str {
621 "wait_agent"
622 }
623
624 fn description(&self) -> String {
625 "Wait for one requested child agent to finish or for a bounded timeout.".into()
626 }
627
628 fn parameters(&self) -> Value {
629 json!({
630 "type": "object",
631 "properties": {
632 "targets": {"type": "array", "items": {"type": "string"}, "minItems": 1, "uniqueItems": true},
633 "timeout_ms": {"type": "integer", "minimum": MIN_WAIT_MS, "maximum": MAX_WAIT_MS}
634 },
635 "required": ["targets"],
636 "additionalProperties": false
637 })
638 }
639
640 async fn execute(
641 &self,
642 _tool_call_id: &str,
643 args: Value,
644 cancel: CancellationToken,
645 _on_update: Option<ToolUpdateSink>,
646 ) -> Result<ToolResult> {
647 let args: WaitArgs = serde_json::from_value(args)?;
648 json_result(&self.0.wait(&args.targets, args.timeout_ms, cancel).await?)
649 }
650}
651
652struct ListAgentsTool(Arc<SubagentRuntime>);
653
654#[async_trait::async_trait]
655impl AgentTool for ListAgentsTool {
656 fn name(&self) -> &str {
657 "list_agents"
658 }
659
660 fn description(&self) -> String {
661 "List child agents in creation order with their current status.".into()
662 }
663
664 fn parameters(&self) -> Value {
665 json!({"type": "object", "properties": {}, "additionalProperties": false})
666 }
667
668 async fn execute(
669 &self,
670 _tool_call_id: &str,
671 _args: Value,
672 _cancel: CancellationToken,
673 _on_update: Option<ToolUpdateSink>,
674 ) -> Result<ToolResult> {
675 json_result(&self.0.list())
676 }
677}
678
679#[derive(Deserialize)]
680struct TargetArgs {
681 target: String,
682}
683
684struct InterruptAgentTool(Arc<SubagentRuntime>);
685
686#[async_trait::async_trait]
687impl AgentTool for InterruptAgentTool {
688 fn name(&self) -> &str {
689 "interrupt_agent"
690 }
691
692 fn description(&self) -> String {
693 "Interrupt a queued or running child agent.".into()
694 }
695
696 fn parameters(&self) -> Value {
697 json!({
698 "type": "object",
699 "properties": {"target": {"type": "string"}},
700 "required": ["target"],
701 "additionalProperties": false
702 })
703 }
704
705 async fn execute(
706 &self,
707 _tool_call_id: &str,
708 args: Value,
709 _cancel: CancellationToken,
710 _on_update: Option<ToolUpdateSink>,
711 ) -> Result<ToolResult> {
712 let args: TargetArgs = serde_json::from_value(args)?;
713 json_result(&self.0.interrupt(&args.target)?)
714 }
715}
716
717pub(crate) fn fork_messages(messages: &[AgentMessage], fork: ForkTurns) -> Vec<AgentMessage> {
718 let safe = messages
719 .iter()
720 .filter(|message| match message {
721 AgentMessage::User(_)
722 | AgentMessage::BranchSummary(_)
723 | AgentMessage::CompactionSummary(_) => true,
724 AgentMessage::Assistant(assistant) => assistant.tool_calls().next().is_none(),
725 AgentMessage::ToolResult(_)
726 | AgentMessage::BashExecution(_)
727 | AgentMessage::Custom(_) => false,
728 })
729 .cloned()
730 .collect::<Vec<_>>();
731
732 match fork {
733 ForkTurns::None => Vec::new(),
734 ForkTurns::All => safe,
735 ForkTurns::Recent(count) => {
736 let start = safe
737 .iter()
738 .enumerate()
739 .rev()
740 .filter(|(_, message)| matches!(message, AgentMessage::User(_)))
741 .nth(count.saturating_sub(1))
742 .map(|(index, _)| index)
743 .unwrap_or(0);
744 safe[start..].to_vec()
745 }
746 }
747}
748
749#[cfg(test)]
750mod tests {
751 use super::*;
752 use kiss_ai::{AssistantMessage, ContentBlock, ToolCall};
753
754 fn assistant(text: &str) -> AgentMessage {
755 let mut message = AssistantMessage::empty("fake", "fake", "fake");
756 message.content.push(ContentBlock::text(text));
757 AgentMessage::Assistant(message)
758 }
759
760 fn parent_session() -> Arc<AgentSession> {
761 let registry = kiss_ai::Registry::from_builtin();
762 let model = registry.all().first().expect("built-in model").clone();
763 AgentSession::new_with_subagents_allowed(
764 crate::SessionManager::in_memory(std::path::Path::new("/test")),
765 Vec::new(),
766 registry,
767 crate::Settings::default(),
768 "test".into(),
769 model,
770 kiss_ai::ThinkingLevel::Off,
771 None,
772 Arc::new(|_| {}),
773 false,
774 )
775 }
776
777 #[test]
778 fn task_names_are_bounded_and_portable() {
779 assert!(validate_task_name("review_2").is_ok());
780 assert!(validate_task_name("").is_err());
781 assert!(validate_task_name("Review").is_err());
782 assert!(validate_task_name("has-dash").is_err());
783 assert!(validate_task_name(&"a".repeat(65)).is_err());
784 }
785
786 #[test]
787 fn fork_turns_accepts_only_documented_values() {
788 assert_eq!(ForkTurns::parse(None).unwrap(), ForkTurns::None);
789 assert_eq!(
790 ForkTurns::parse(Some(&json!("all"))).unwrap(),
791 ForkTurns::All
792 );
793 assert_eq!(
794 ForkTurns::parse(Some(&json!(2))).unwrap(),
795 ForkTurns::Recent(2)
796 );
797 assert!(ForkTurns::parse(Some(&json!(0))).is_err());
798 assert!(ForkTurns::parse(Some(&json!("recent"))).is_err());
799 }
800
801 #[test]
802 fn context_fork_removes_tool_pairs_and_keeps_recent_turns() {
803 let mut tool_message = AssistantMessage::empty("fake", "fake", "fake");
804 tool_message.content.push(ContentBlock::ToolCall(ToolCall {
805 id: "call".into(),
806 name: "read".into(),
807 arguments: json!({}),
808 thought_signature: None,
809 }));
810 let history = vec![
811 AgentMessage::user("first"),
812 assistant("one"),
813 AgentMessage::Assistant(tool_message),
814 AgentMessage::user("second"),
815 assistant("two"),
816 ];
817
818 assert!(fork_messages(&history, ForkTurns::None).is_empty());
819 let all = fork_messages(&history, ForkTurns::All);
820 assert_eq!(all.len(), 4);
821 assert!(
822 all.iter()
823 .all(|message| !matches!(message, AgentMessage::ToolResult(_)))
824 );
825 let recent = fork_messages(&history, ForkTurns::Recent(1));
826 assert_eq!(recent, history[3..]);
827 }
828
829 #[test]
830 fn control_tool_catalog_uses_codex_names() {
831 let parent = Weak::new();
832 let runtime = SubagentRuntime::new(parent);
833 let names = runtime
834 .control_tools()
835 .into_iter()
836 .map(|tool| tool.name().to_string())
837 .collect::<Vec<_>>();
838 assert_eq!(
839 names,
840 [
841 "spawn_agent",
842 "send_message",
843 "followup_task",
844 "wait_agent",
845 "list_agents",
846 "interrupt_agent"
847 ]
848 );
849 for tool in runtime.control_tools() {
850 let schema = tool.parameters();
851 assert_eq!(schema["type"], "object", "{} schema", tool.name());
852 assert_eq!(
853 schema["additionalProperties"],
854 false,
855 "{} schema",
856 tool.name()
857 );
858 }
859 let spawn = runtime.control_tools().remove(0).parameters();
860 assert_eq!(spawn["required"], json!(["task_name", "prompt"]));
861 assert_eq!(
862 spawn["properties"]["fork_turns"]["oneOf"]
863 .as_array()
864 .unwrap()
865 .len(),
866 2
867 );
868 }
869
870 #[tokio::test]
871 async fn queued_agent_can_be_listed_interrupted_and_waited_for() {
872 let parent = parent_session();
873 let runtime = SubagentRuntime::with_permits(Arc::downgrade(&parent), 0);
874 let spawned = runtime
875 .spawn(
876 "review_tests".into(),
877 "Review the tests".into(),
878 ForkTurns::None,
879 None,
880 None,
881 )
882 .unwrap();
883 assert_eq!(spawned.status, AgentStatus::Queued);
884 assert_eq!(runtime.list(), vec![spawned.clone()]);
885 assert!(
886 runtime
887 .spawn(
888 "review_tests".into(),
889 "Duplicate".into(),
890 ForkTurns::None,
891 None,
892 None,
893 )
894 .is_err()
895 );
896
897 let waiting_runtime = runtime.clone();
898 let target = spawned.id.clone();
899 let waiting = tokio::spawn(async move {
900 waiting_runtime
901 .wait(&[target], 5_000, CancellationToken::new())
902 .await
903 .unwrap()
904 });
905 tokio::task::yield_now().await;
906 let interrupted = runtime.interrupt("review_tests").unwrap();
907 assert_eq!(interrupted.status, AgentStatus::Interrupted);
908 runtime.permits.add_permits(1);
909
910 let outcome = tokio::time::timeout(Duration::from_secs(1), waiting)
911 .await
912 .expect("mailbox wait")
913 .unwrap();
914 assert!(!outcome.timed_out);
915 assert_eq!(outcome.agents[0].status, AgentStatus::Interrupted);
916 }
917}