1pub(in crate::delegate) mod lane;
26mod owed;
27
28use std::{
29 collections::HashSet,
30 path::PathBuf,
31 sync::{Arc, Mutex, Weak},
32 time::{Duration, Instant},
33};
34
35use async_trait::async_trait;
36use scv_core::{
37 ApprovalGate, ProgressSink, Tool, ToolApprovals, ToolContext, ToolError, ToolOutput, ToolRisk,
38 ToolSpec,
39};
40use scv_protocol::{JobChange, JobOutcome, JobReport, JobStatus, describe_reports};
41use serde::Deserialize;
42use serde_json::{Map, Value, json};
43use tokio::sync::{mpsc, watch};
44use tokio_util::sync::CancellationToken;
45
46use self::lane::{Lanes, Place};
47use crate::{
48 BusyBehavior,
49 args::{Timeouts, bounded, parse_args, timeout_schema},
50 delegate::{
51 agent::{AGENT_TOOL, AgentTool, Offered},
52 conversation::{self, ConversationStore},
53 output::AgentReply,
54 request::AgentArgs,
55 review::{self, Record, ReviewedTool},
56 },
57 sync::lock,
58};
59
60const MAX_FINISHED: usize = 16;
62const REPORT_REPLY_CHARS: usize = 6000;
64const REPORT_MAX_JOBS: usize = 4;
66const REPORT_RETRY_DELAYS: [Duration; 2] = [Duration::from_secs(30), Duration::from_secs(120)];
70const CANCEL_SETTLE: Duration = Duration::from_secs(10);
72const MAX_PENDING_CHANGES: usize = 64;
75const TASK_CHARS: usize = 80;
77
78pub struct BackgroundJobs {
81 limit: usize,
82 state: Mutex<JobsState>,
83 cancellation: CancellationToken,
84 finished: Option<mpsc::UnboundedSender<()>>,
86 approvals: Option<Arc<dyn ApprovalGate>>,
89 lanes: Arc<Lanes>,
91}
92
93impl std::fmt::Debug for BackgroundJobs {
94 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
95 formatter
96 .debug_struct("BackgroundJobs")
97 .field("limit", &self.limit)
98 .finish_non_exhaustive()
99 }
100}
101
102#[derive(Default)]
103struct JobsState {
104 next: u64,
105 jobs: Vec<Job>,
106 changes: Vec<(String, JobChange)>,
108 owed: owed::Owed,
110}
111
112impl JobsState {
113 fn refresh(&mut self, call_id: &str, outcome: JobOutcome) {
116 if let Some((_, change)) = self
117 .changes
118 .iter_mut()
119 .rev()
120 .find(|(call, change)| call == call_id && change.job == outcome.job)
121 {
122 change.outcome = Some(outcome);
123 }
124 }
125
126 fn record(&mut self, call_id: &str, change: JobChange) {
129 if call_id.is_empty() {
130 return;
131 }
132 if self.changes.len() >= MAX_PENDING_CHANGES {
133 let (call, dropped) = self.changes.remove(0);
134 self.owed.evicted(&call, &dropped.job);
135 }
136 self.changes.push((call_id.to_owned(), change));
137 }
138}
139
140struct Job {
141 id: String,
142 tool: String,
144 agent: String,
146 task: String,
148 started: Instant,
149 progress: ProgressSink,
150 last_progress: Option<String>,
151 cancel: CancellationToken,
153 cancelled: bool,
155 queued: bool,
158 outcome: Option<Outcome>,
159 reported: bool,
164 reporting: bool,
166 report_attempts: u32,
168 report_due: Option<tokio::time::Instant>,
170 done: watch::Receiver<bool>,
171 review: Option<Arc<Record>>,
173 cancelling: bool,
176}
177
178struct Waiting<'a> {
182 jobs: &'a BackgroundJobs,
183 job: String,
184 call: String,
185 owed: bool,
187 settled: bool,
188}
189
190impl Waiting<'_> {
191 fn settle(mut self) {
196 self.settled = true;
197 let mut state = self.jobs.state();
198 let Some(entry) = state.jobs.iter_mut().find(|entry| entry.id == self.job) else {
199 return;
200 };
201 entry.cancelling = false;
202 if !self.owed {
203 return;
204 }
205 let pending = entry.outcome.is_none().then(|| entry.outcome()).flatten();
206 let pending = pending.map(|mut outcome| {
207 outcome.review.journal_pending = true;
208 outcome
209 });
210 if let Some(outcome) = state.owed.waited(&self.job, &self.call, pending) {
211 state.refresh(&self.call, outcome);
212 }
213 }
214}
215
216impl Drop for Waiting<'_> {
217 fn drop(&mut self) {
218 if self.settled {
219 return;
220 }
221 let mut state = self.jobs.state();
222 if let Some(entry) = state.jobs.iter_mut().find(|entry| entry.id == self.job) {
223 entry.cancelling = false;
224 }
225 if !self.owed {
226 return;
227 }
228 let (call, job) = (&self.call, &self.job);
230 state
231 .changes
232 .retain(|(recorded, change)| !(recorded == call && &change.job == job));
233 let due = state.owed.abandoned(job, call);
234 drop(state);
235 if due {
236 self.jobs.wake();
237 }
238 }
239}
240
241struct Outcome {
242 output: ToolOutput,
243 elapsed: Duration,
244}
245
246#[derive(Debug, PartialEq, Eq)]
248pub enum ReportFailure {
249 Retry(Duration),
251 GiveUp {
254 attempts: u32,
255 reports: Vec<JobReport>,
256 },
257 Settled,
259}
260
261impl Drop for BackgroundJobs {
262 fn drop(&mut self) {
263 self.cancellation.cancel();
264 }
265}
266
267impl BackgroundJobs {
268 pub fn new(limit: usize, finished: Option<mpsc::UnboundedSender<()>>) -> Self {
270 Self {
271 limit,
272 state: Mutex::default(),
273 cancellation: CancellationToken::new(),
274 finished,
275 approvals: None,
276 lanes: Arc::default(),
277 }
278 }
279
280 #[must_use]
283 pub fn with_approvals(mut self, gate: Arc<dyn ApprovalGate>) -> Self {
284 self.approvals = Some(gate);
285 self
286 }
287
288 pub(crate) fn limit(&self) -> usize {
289 self.limit
290 }
291
292 fn state(&self) -> std::sync::MutexGuard<'_, JobsState> {
293 lock(&self.state)
294 }
295
296 #[allow(
306 clippy::too_many_arguments,
307 reason = "a job's call, its context, its turn, and its review, each its own"
308 )]
309 pub(in crate::delegate) fn start(
310 self: &Arc<Self>,
311 tool: Arc<dyn Tool>,
312 name: &str,
313 agent: &str,
314 arguments: Value,
315 context: &ToolContext,
316 turn: Turn,
317 review: Option<&Arc<ReviewedTool>>,
318 ) -> Result<Value, ToolError> {
319 let task = task_line(
320 arguments
321 .get("prompt")
322 .and_then(Value::as_str)
323 .unwrap_or_default(),
324 );
325 let queued = matches!(turn, Turn::Queued(_));
326 let reserved = match review {
327 Some(review) => {
328 let id = {
329 let mut state = self.state();
330 self.room(&state, queued)?;
331 state.next += 1;
332 format!("job-{}", state.next)
333 };
334 review.begin(&id)?;
335 Some(id)
336 }
337 None => None,
338 };
339 let (id, progress, done_tx, cancellation) = {
340 let mut state = self.state();
341 self.room(&state, queued)?;
343 let id = reserved.unwrap_or_else(|| {
344 state.next += 1;
345 format!("job-{}", state.next)
346 });
347 let progress = ProgressSink::buffered();
348 let (done_tx, done) = watch::channel(false);
349 let cancel = self.cancellation.child_token();
350 state.record(
351 &context.call_id,
352 JobChange {
353 job: id.clone(),
354 tool: name.to_owned(),
355 agent: agent.to_owned(),
356 status: JobStatus::Running,
357 task: task.clone(),
358 journal: review.map(|review| review.journal_id()),
359 outcome: None,
360 },
361 );
362 state.jobs.push(Job {
363 id: id.clone(),
364 tool: name.to_owned(),
365 agent: agent.to_owned(),
366 task,
367 started: Instant::now(),
368 progress: progress.clone(),
369 last_progress: None,
370 cancel: cancel.clone(),
371 cancelled: false,
372 queued,
373 outcome: None,
374 reported: false,
375 reporting: false,
376 report_attempts: 0,
377 report_due: None,
378 done,
379 review: review.map(|review| Arc::clone(review.record())),
380 cancelling: false,
381 });
382 (id, progress, done_tx, cancel)
383 };
384 let jobs = Arc::downgrade(self);
385 let job = id.clone();
386 let approvals = self.approvals.clone();
387 let workspace = context.workspace.clone();
388 let reviewed = review.map(Arc::clone);
389 tokio::spawn(async move {
390 let started = Instant::now();
391 let mut context = ToolContext::new(workspace, cancellation);
392 if let Some(gate) = &approvals {
395 context.approvals = ToolApprovals::new(Arc::clone(gate), job.clone());
396 }
397 context.progress = progress;
398 let place = turn.place();
399 let output = match &place {
400 Some(place) if queued => {
401 context.progress.report(&format!(
402 "queued: waits for conversation {}'s running turn",
403 place.handle()
404 ));
405 match place.wait_first(&context.cancellation).await {
406 Ok(()) => {
407 context.progress.take();
408 context.progress.report(&format!(
409 "started its turn in conversation {}",
410 place.handle()
411 ));
412 tool.execute(arguments, context).await
413 }
414 Err(error) => match &reviewed {
417 Some(review) => Ok(review.unstarted(&error)),
418 None => Err(error),
419 },
420 }
421 }
422 _ => tool.execute(arguments, context).await,
423 }
424 .unwrap_or_else(ToolOutput::from);
425 drop(place);
428 finish(&jobs, &job, output, started.elapsed());
429 let _ = done_tx.send(true);
430 });
431 let mut started = json!({
432 "job": id,
433 "agent": agent,
434 "status": "running",
435 "background": true,
436 "note": "The agent is working in the background. SCV reports the result in a new \
437 turn when it finishes. agent_status shows its progress and agent_cancel \
438 stops it."
439 });
440 if queued {
441 started["queued"] = true.into();
442 started["note"] = "The conversation is busy, so this prompt is queued: it runs in \
443 the background after the turns ahead of it, and SCV reports the result in a new \
444 turn when it finishes. agent_status shows where it stands and agent_cancel \
445 withdraws it."
446 .into();
447 }
448 Ok(started)
449 }
450
451 fn room(&self, state: &JobsState, queued: bool) -> Result<(), ToolError> {
454 let running = state
455 .jobs
456 .iter()
457 .filter(|job| job.outcome.is_none() && !job.queued)
458 .count();
459 if !queued && running >= self.limit {
460 return Err(ToolError::limit(format!(
461 "{running} background jobs are already running, the limit \
462 (agent.max_background). Start this one after a job finishes, or \
463 stop one with agent_cancel if the user no longer needs it."
464 )));
465 }
466 Ok(())
467 }
468
469 async fn cancel(&self, job: &str, call_id: &str) -> Result<Value, ToolError> {
480 let (mut done, owed) = {
481 let mut state = self.state();
482 let index = state
483 .jobs
484 .iter()
485 .position(|candidate| candidate.id == job)
486 .ok_or_else(|| unknown_job(job))?;
487 let entry = &mut state.jobs[index];
488 if entry.outcome.is_some() {
489 let (mut value, change) = entry.describe();
490 value["note"] = "The job had already finished.".into();
491 if let Some(change) = change {
492 state.record(call_id, change);
493 }
494 return Ok(value);
495 }
496 entry.cancelled = true;
497 entry.cancel.cancel();
498 entry.cancelling = true;
499 let done = entry.done.clone();
500 let mut owed = false;
501 if !entry.reported {
502 entry.reported = true;
503 let change = entry.change(JobStatus::Cancelled);
504 owed = entry.review.is_some();
505 state.record(call_id, change);
506 if owed {
507 state.owed.cancelling(job, call_id);
508 }
509 }
510 (done, owed)
511 };
512 let waiting = Waiting {
513 jobs: self,
514 job: job.to_owned(),
515 call: call_id.to_owned(),
516 owed,
517 settled: false,
518 };
519 let _ = tokio::time::timeout(CANCEL_SETTLE, done.wait_for(|finished| *finished)).await;
520 waiting.settle();
521 self.describe(Some(job), call_id)
522 }
523
524 pub fn take_updates(&self) -> Vec<JobOutcome> {
528 self.state().owed.take_updates()
529 }
530
531 pub fn updated(&self, jobs: &[String], delivered: bool) {
534 self.state().owed.updated(jobs, delivered);
535 }
536
537 pub fn published(&self, call_id: &str, delivered: bool) {
543 let due = self.state().owed.sent(call_id, delivered);
544 if due {
545 self.wake();
546 }
547 }
548
549 pub fn turn_ended(&self) {
552 let due = self.state().owed.turn_ended();
553 if due {
554 self.wake();
555 }
556 }
557
558 #[cfg(any(test, feature = "test-support"))]
565 pub fn owe_for_tests(&self, call_id: &str, change: JobChange, last: Option<JobOutcome>) {
566 let mut state = self.state();
567 state.owed.owe(call_id, &change, last);
568 state.record(call_id, change);
569 }
570
571 #[cfg(any(test, feature = "test-support"))]
573 pub fn stop_for_tests(&self, outcome: &JobOutcome) {
574 let due = self.state().owed.stopped(&outcome.job, outcome);
575 if due {
576 self.wake();
577 }
578 }
579
580 fn wake(&self) {
582 if let Some(finished) = &self.finished {
583 let _ = finished.send(());
584 }
585 }
586
587 async fn wait(
590 &self,
591 job: &str,
592 limit: Duration,
593 cancellation: &CancellationToken,
594 call_id: &str,
595 ) -> Result<Value, ToolError> {
596 let mut done = self
597 .state()
598 .jobs
599 .iter()
600 .find(|candidate| candidate.id == job)
601 .map(|candidate| candidate.done.clone())
602 .ok_or_else(|| unknown_job(job))?;
603 tokio::select! {
604 () = cancellation.cancelled() => return Err(ToolError::cancelled("wait cancelled")),
605 _ = tokio::time::timeout(limit, done.wait_for(|finished| *finished)) => {}
606 }
607 self.describe(Some(job), call_id)
608 }
609
610 fn describe(&self, job: Option<&str>, call_id: &str) -> Result<Value, ToolError> {
613 let mut state = self.state();
614 let mut changes = Vec::new();
615 let value = if let Some(job) = job {
616 let entry = state
617 .jobs
618 .iter_mut()
619 .find(|candidate| candidate.id == job)
620 .ok_or_else(|| unknown_job(job))?;
621 let (value, change) = entry.describe();
622 changes.extend(change);
623 value
624 } else {
625 let jobs: Vec<Value> = state
626 .jobs
627 .iter_mut()
628 .map(|entry| {
629 let (value, change) = entry.describe();
630 changes.extend(change);
631 value
632 })
633 .collect();
634 json!({ "jobs": jobs })
635 };
636 for change in changes {
637 state.record(call_id, change);
638 }
639 Ok(value)
640 }
641
642 pub fn take_changes(&self, call_id: &str) -> Vec<JobChange> {
648 let mut state = self.state();
649 let mut taken = Vec::new();
650 state.changes.retain(|(call, change)| {
651 if call == call_id {
652 taken.push(change.clone());
653 false
654 } else {
655 true
656 }
657 });
658 for change in &mut taken {
659 state.owed.take(call_id, change);
660 }
661 taken
662 }
663
664 pub fn take_unreported(&self) -> Vec<JobReport> {
669 let now = tokio::time::Instant::now();
670 let mut state = self.state();
671 state
672 .jobs
673 .iter_mut()
674 .filter(|job| job.awaits_report() && job.report_due.is_none_or(|due| due <= now))
675 .take(REPORT_MAX_JOBS)
676 .filter_map(|job| {
677 job.reporting = true;
678 job.report()
679 })
680 .collect()
681 }
682
683 pub fn report_settled(&self, jobs: &[String]) {
686 let mut state = self.state();
687 for job in state.jobs.iter_mut().filter(|job| jobs.contains(&job.id)) {
688 job.reporting = false;
689 job.reported = true;
690 job.report_due = None;
691 }
692 }
693
694 pub fn report_failed(&self, jobs: &[String], retry: bool) -> ReportFailure {
701 let mut state = self.state();
702 let mut failed: Vec<&mut Job> = state
703 .jobs
704 .iter_mut()
705 .filter(|job| job.reporting && jobs.contains(&job.id))
706 .collect();
707 for job in &mut failed {
708 job.reporting = false;
709 job.report_attempts += 1;
710 }
711 failed.retain(|job| !job.reported);
713 let Some(attempts) = failed.iter().map(|job| job.report_attempts).max() else {
714 return ReportFailure::Settled;
715 };
716 let delay = usize::try_from(attempts - 1)
717 .ok()
718 .and_then(|earlier| REPORT_RETRY_DELAYS.get(earlier));
719 if retry && let Some(&delay) = delay {
720 let due = tokio::time::Instant::now() + delay;
721 for job in failed {
722 job.report_due = Some(due);
723 }
724 return ReportFailure::Retry(delay);
725 }
726 let reports = failed
727 .into_iter()
728 .filter_map(|job| {
729 job.reported = true;
730 job.report_due = None;
731 job.report()
732 })
733 .collect();
734 ReportFailure::GiveUp { attempts, reports }
735 }
736
737 pub fn retry_now(&self) -> bool {
741 let mut state = self.state();
742 let mut waiting = false;
743 for job in state.jobs.iter_mut().filter(|job| job.awaits_report()) {
744 waiting |= job.report_due.take().is_some();
745 }
746 waiting
747 }
748
749 pub fn next_retry(&self) -> Option<tokio::time::Instant> {
752 self.state()
753 .jobs
754 .iter()
755 .filter(|job| job.awaits_report())
756 .filter_map(|job| job.report_due)
757 .min()
758 }
759
760 pub fn pending(&self) -> usize {
765 let state = self.state();
766 let mut pending: HashSet<&str> = state
767 .jobs
768 .iter()
769 .filter(|job| job.outcome.is_none() || !job.reported)
770 .map(|job| job.id.as_str())
771 .collect();
772 pending.extend(state.owed.jobs());
773 pending.len()
774 }
775
776 pub fn running(&self) -> usize {
778 self.state()
779 .jobs
780 .iter()
781 .filter(|job| job.outcome.is_none())
782 .count()
783 }
784}
785
786pub(in crate::delegate) enum Turn {
788 New,
790 Next(Place),
792 Queued(Place),
794}
795
796impl Turn {
797 fn place(self) -> Option<Place> {
798 match self {
799 Self::New => None,
800 Self::Next(place) | Self::Queued(place) => Some(place),
801 }
802 }
803}
804
805fn finish(jobs: &Weak<BackgroundJobs>, id: &str, output: ToolOutput, elapsed: Duration) {
806 let Some(jobs) = jobs.upgrade() else {
808 return;
809 };
810 {
811 let mut state = jobs.state();
812 let mut last = None;
813 if let Some(job) = state.jobs.iter_mut().find(|job| job.id == id) {
814 job.last_progress = job.progress.take().or(job.last_progress.take());
815 job.outcome = Some(Outcome { output, elapsed });
816 job.reported |= job.cancelled;
818 last = job.outcome();
819 }
820 if let Some(last) = last {
822 state.owed.stopped(id, &last);
823 }
824 let finished = state
826 .jobs
827 .iter()
828 .filter(|job| job.outcome.is_some())
829 .count();
830 let mut excess = finished.saturating_sub(MAX_FINISHED);
831 state.jobs.retain(|job| {
832 if excess > 0 && job.outcome.is_some() && job.reported && !job.cancelling {
833 excess -= 1;
834 false
835 } else {
836 true
837 }
838 });
839 }
840 if let Some(finished) = &jobs.finished {
841 let _ = finished.send(());
842 }
843}
844
845impl Job {
846 fn awaits_report(&self) -> bool {
848 self.outcome.is_some() && !self.reported && !self.reporting
849 }
850
851 fn report(&self) -> Option<JobReport> {
853 let outcome = self.outcome.as_ref()?;
854 let reply = AgentReply::read(&result_value(&outcome.output)).unwrap_or_default();
855 Some(JobReport {
856 job: self.id.clone(),
857 agent: self.agent.clone(),
858 task: self.task.clone(),
859 status: job_status(&outcome.output, &reply),
860 session: reply.session,
861 reply: bounded(
862 reply
863 .reply
864 .as_deref()
865 .unwrap_or(outcome.output.content.as_str()),
866 REPORT_REPLY_CHARS,
867 ),
868 outcome: self.outcome(),
869 })
870 }
871
872 fn outcome(&self) -> Option<JobOutcome> {
875 self.review.as_ref().map(|record| {
876 let (review, landing) = record
877 .live
878 .snapshot(self.cancelled || self.cancel.is_cancelled());
879 JobOutcome {
880 job: self.id.clone(),
881 review,
882 landing,
883 }
884 })
885 }
886
887 fn change(&self, status: JobStatus) -> JobChange {
890 JobChange {
891 job: self.id.clone(),
892 tool: self.tool.clone(),
893 agent: self.agent.clone(),
894 status,
895 task: self.task.clone(),
896 journal: None,
897 outcome: if status == JobStatus::Running {
898 None
899 } else {
900 self.outcome()
901 },
902 }
903 }
904
905 fn describe(&mut self) -> (Value, Option<JobChange>) {
908 if let Some(line) = self.progress.take() {
909 self.last_progress = Some(line);
910 }
911 let mut change = None;
912 let mut value = Map::new();
913 value.insert("job".into(), self.id.clone().into());
914 value.insert("agent".into(), self.agent.clone().into());
915 match &self.outcome {
916 None => {
917 value.insert("status".into(), "running".into());
918 value.insert(
919 "elapsed_seconds".into(),
920 self.started.elapsed().as_secs().into(),
921 );
922 if let Some(progress) = &self.last_progress {
923 value.insert("progress".into(), progress.clone().into());
924 }
925 if let Some(record) = &self.review {
926 value.insert("review".into(), record.live.status());
927 }
928 }
929 Some(outcome) => {
930 let result = result_value(&outcome.output);
931 let status = if self.cancelled {
932 JobStatus::Cancelled
933 } else {
934 job_status(
935 &outcome.output,
936 &AgentReply::read(&result).unwrap_or_default(),
937 )
938 };
939 value.insert("status".into(), status.as_str().into());
940 value.insert("elapsed_seconds".into(), outcome.elapsed.as_secs().into());
941 value.insert("result".into(), result);
942 if !self.reported {
943 self.reported = true;
944 change = Some(self.change(status));
945 }
946 }
947 }
948 (Value::Object(value), change)
949 }
950}
951
952fn result_value(output: &ToolOutput) -> Value {
954 match serde_json::from_str::<Value>(&output.content) {
955 Ok(value @ Value::Object(_)) => value,
956 _ => json!({ "reply": output.content, "is_error": output.is_error() }),
957 }
958}
959
960fn job_status(output: &ToolOutput, reply: &AgentReply) -> JobStatus {
963 reply.status.unwrap_or(if output.is_error() {
964 JobStatus::Failed
965 } else {
966 JobStatus::Completed
967 })
968}
969
970fn task_line(prompt: &str) -> String {
973 let line = prompt
974 .lines()
975 .map(str::trim)
976 .find(|line| !line.is_empty())
977 .unwrap_or_default();
978 let mut task: String = line
979 .chars()
980 .filter(|character| !character.is_control())
981 .take(TASK_CHARS)
982 .collect();
983 if line.chars().count() > TASK_CHARS {
984 task.push('…');
985 }
986 task
987}
988
989fn unknown_job(job: &str) -> ToolError {
990 ToolError::invalid_arguments(format!(
991 "unknown background job {:?}; agent_status lists this session's jobs",
992 bounded(job, 64)
993 ))
994}
995
996const REVIEWED_NOTE: &str = " SCV shows the user its own Review and Landing lines for \
999 reviewed jobs. If you mention the outcome, repeat them as given; never call unapproved \
1000 work approved or done, and never call landed work approved. If a reviewer declined, tell \
1001 the user what it said. If a landing is NOT confirmed, or the reviewer could NOT confirm \
1002 it, say so plainly, and don't fix, revert, or land anything on your own.";
1003
1004fn reviewed_note(reports: &[JobReport]) -> &'static str {
1005 if reports.iter().any(|report| report.outcome.is_some()) {
1006 REVIEWED_NOTE
1007 } else {
1008 ""
1009 }
1010}
1011
1012pub fn report_prompt(reports: &[JobReport]) -> String {
1014 format!(
1015 "[SCV background report] Delegated work you started in the background has \
1016 finished. The user did not send this message: tell them briefly what \
1017 happened and the key result.{}\n\n{}",
1018 reviewed_note(reports),
1019 describe_reports(reports)
1020 )
1021}
1022
1023pub fn delivered_note(reports: &[JobReport], error: &str) -> String {
1026 format!(
1027 "[SCV background report, already delivered] Delegated work you started in \
1028 the background has finished. Your report of it failed ({error}), so SCV \
1029 sent the user the results below directly. The user did not send this \
1030 message; repeat the results only if they ask.{}\n\n{}",
1031 reviewed_note(reports),
1032 describe_reports(reports)
1033 )
1034}
1035
1036pub(crate) struct BackgroundCapable {
1039 pub(crate) inner: Arc<AgentTool>,
1040 pub(crate) jobs: Arc<BackgroundJobs>,
1041 pub(crate) conversations: Arc<ConversationStore>,
1043 pub(crate) reviews: Option<PathBuf>,
1046 pub(crate) session: Option<String>,
1048}
1049
1050struct QueuedTool(Arc<AgentTool>);
1053
1054#[async_trait]
1055impl Tool for QueuedTool {
1056 fn spec(&self) -> ToolSpec {
1057 self.0.spec()
1058 }
1059 fn risk(&self, arguments: &Value) -> Result<ToolRisk, ToolError> {
1060 self.0.risk(arguments)
1061 }
1062 fn approval_summary(&self, arguments: &Value) -> Result<String, ToolError> {
1063 self.0.approval_summary(arguments)
1064 }
1065 async fn execute(
1066 &self,
1067 arguments: Value,
1068 context: ToolContext,
1069 ) -> Result<ToolOutput, ToolError> {
1070 self.0.execute_when_idle(arguments, context).await
1071 }
1072}
1073
1074struct Split {
1076 arguments: Value,
1077 background: Option<bool>,
1079 review: Option<Value>,
1081}
1082
1083impl Split {
1084 fn new(arguments: &Value) -> Self {
1085 let mut arguments = arguments.clone();
1086 let object = arguments.as_object_mut();
1087 let (background, review) = match object {
1088 Some(object) => (
1089 object
1090 .remove("background")
1091 .map(|value| value.as_bool() == Some(true)),
1092 object.remove("review").filter(|review| !review.is_null()),
1093 ),
1094 None => (None, None),
1095 };
1096 Self {
1097 arguments,
1098 background,
1099 review,
1100 }
1101 }
1102
1103 fn background(&self) -> bool {
1104 self.background == Some(true)
1105 }
1106}
1107
1108#[async_trait]
1109impl Tool for BackgroundCapable {
1110 fn spec(&self) -> ToolSpec {
1111 let mut spec = self.inner.spec();
1112 let reviewed = self.offers_review();
1113 if let Some(properties) = spec
1114 .parameters
1115 .get_mut("properties")
1116 .and_then(Value::as_object_mut)
1117 {
1118 properties.insert(
1119 "background".into(),
1120 json!({
1121 "type":"boolean",
1122 "description":"Run in the background: the call returns a job handle at once, \
1123 the user can keep talking to you while the agent works, and SCV reports \
1124 the result in a new turn when it finishes. Use it for any substantial \
1125 task; run in the foreground only for quick work whose result you need \
1126 within this turn."
1127 }),
1128 );
1129 if reviewed {
1130 properties.insert("review".into(), review::schema(&self.inner));
1131 }
1132 }
1133 spec.description.push_str(&format!(
1134 " Set background to true for anything beyond a quick task: the call returns a job \
1135 handle at once (at most {} running per session), SCV reports the result when the \
1136 job finishes, agent_status shows progress, and agent_cancel stops it. A background \
1137 job's own approval requests get only the answer this session would give without \
1138 asking a person.",
1139 self.jobs.limit()
1140 ));
1141 if reviewed {
1142 spec.description.push_str(
1143 " A call with review runs as a reviewed background job: an independent \
1144 reviewer checks the work in up to review.rounds rounds.",
1145 );
1146 }
1147 spec
1148 }
1149
1150 fn risk(&self, arguments: &Value) -> Result<ToolRisk, ToolError> {
1151 let split = Split::new(arguments);
1152 if let Some(review) = &split.review {
1153 self.plan(&split, review)?;
1154 }
1155 self.inner.risk(&split.arguments)
1156 }
1157
1158 fn approval_summary(&self, arguments: &Value) -> Result<String, ToolError> {
1159 let split = Split::new(arguments);
1160 let mut summary = self.inner.approval_summary(&split.arguments)?;
1161 if let Some(review) = &split.review {
1162 let plan = self.plan(&split, review)?;
1163 summary.push_str(
1164 " Runs in the background: the call returns at once and the result is \
1165 reported when the review ends.",
1166 );
1167 summary.push_str(&review::approval_summary(&plan, &self.inner)?);
1168 } else if split.background() {
1169 summary.push_str(
1170 " Runs in the background: the call returns at once and the result is \
1171 reported when the agent finishes.",
1172 );
1173 }
1174 Ok(summary)
1175 }
1176
1177 async fn execute(
1178 &self,
1179 arguments: Value,
1180 context: ToolContext,
1181 ) -> Result<ToolOutput, ToolError> {
1182 let split = Split::new(&arguments);
1183 if let Some(review) = &split.review {
1184 return self.start_reviewed(&split, review, &context);
1185 }
1186 let background = split.background();
1187 let arguments = split.arguments;
1188 let (agent, routed) = self.inner.route(&arguments)?;
1189 if background {
1191 agent.backend.risk(&routed)?;
1192 }
1193 let Some(handle) = parse_args::<AgentArgs>(&routed)?
1194 .session
1195 .filter(|session| conversation::is_handle(session))
1196 else {
1197 return if background {
1198 self.start(agent, arguments, &context, Turn::New)
1199 } else {
1200 self.inner.execute(arguments, context).await
1201 };
1202 };
1203 let place = self.jobs.lanes.join(&handle);
1205 if place.ahead == 0 && !agent.backend.busy(&routed)? {
1206 return if background {
1207 self.start(agent, arguments, &context, Turn::Next(place))
1208 } else {
1209 let output = self.inner.execute(arguments, context).await;
1211 drop(place);
1212 output
1213 };
1214 }
1215 let behavior = match agent.on_busy(&routed)? {
1216 BusyBehavior::Steer => {
1217 if let Some(output) = agent.backend.steer(routed, context.clone()).await? {
1218 return Ok(output);
1219 }
1220 agent.busy.fallback()
1221 }
1222 behavior => behavior,
1223 };
1224 match behavior {
1225 BusyBehavior::Wait if !background => {
1227 place.wait_first(&context.cancellation).await?;
1228 let output = self.inner.execute_when_idle(arguments, context).await;
1229 drop(place);
1230 output
1231 }
1232 BusyBehavior::Fail => Err(busy(&handle, &place)),
1233 _ => {
1234 queue_room(agent, &handle, &place)?;
1235 self.start(agent, arguments, &context, Turn::Queued(place))
1236 }
1237 }
1238 }
1239}
1240
1241fn busy(handle: &str, place: &Place) -> ToolError {
1244 let waiting = place.ahead.saturating_sub(1);
1246 ToolError::failed(if waiting == 0 {
1247 format!("session busy: conversation {handle} is still running a turn")
1248 } else {
1249 format!(
1250 "session busy: conversation {handle} is still running a turn, and {waiting} more \
1251 prompts wait for it"
1252 )
1253 })
1254}
1255
1256fn queue_room(agent: &Offered, handle: &str, place: &Place) -> Result<(), ToolError> {
1258 let waiting = place.ahead.saturating_sub(1);
1259 if waiting >= agent.busy.max_queued_turns {
1260 return Err(ToolError::limit(format!(
1261 "session busy: conversation {handle} is running a turn and {waiting} prompts \
1262 already wait for it, of the {} agent.max_queued_turns allows. Send this one \
1263 once they have run, or stop one with agent_cancel.",
1264 agent.busy.max_queued_turns
1265 )));
1266 }
1267 Ok(())
1268}
1269
1270impl BackgroundCapable {
1271 #[cfg(test)]
1274 pub(crate) fn plain(inner: Arc<AgentTool>, jobs: Arc<BackgroundJobs>) -> Self {
1275 Self {
1276 inner,
1277 jobs,
1278 conversations: Arc::new(ConversationStore::new(
1279 conversation::ConversationLimits {
1280 max: 8,
1281 idle: Duration::from_secs(86400),
1282 },
1283 None,
1284 )),
1285 reviews: None,
1286 session: None,
1287 }
1288 }
1289
1290 fn offers_review(&self) -> bool {
1294 self.reviews.is_some() && !self.inner.continuing().is_empty()
1295 }
1296
1297 fn plan(&self, split: &Split, review: &Value) -> Result<review::Plan, ToolError> {
1299 if !self.offers_review() {
1300 return Err(ToolError::invalid_arguments(
1301 "review is not available in this session",
1302 ));
1303 }
1304 if split.background == Some(false) {
1305 return Err(ToolError::invalid_arguments(
1306 "a reviewed call runs as a background job; omit background or set it to true",
1307 ));
1308 }
1309 let (builder, routed) = self.inner.route(&split.arguments)?;
1310 builder.backend.risk(&routed)?;
1311 review::plan(
1312 &self.inner,
1313 builder,
1314 &routed,
1315 review,
1316 self.conversations.limits().max,
1317 )
1318 }
1319
1320 fn start_reviewed(
1325 &self,
1326 split: &Split,
1327 review: &Value,
1328 context: &ToolContext,
1329 ) -> Result<ToolOutput, ToolError> {
1330 let plan = self.plan(split, review)?;
1331 let arguments = split.arguments.clone();
1332 let (agent, routed) = self.inner.route(&arguments)?;
1333 let turn = match parse_args::<AgentArgs>(&routed)?
1334 .session
1335 .filter(|session| conversation::is_handle(session))
1336 {
1337 None => Turn::New,
1338 Some(handle) => {
1339 let place = self.jobs.lanes.join(&handle);
1340 if place.ahead == 0 && !agent.backend.busy(&routed)? {
1341 Turn::Next(place)
1342 } else if agent.on_busy(&routed)? == BusyBehavior::Fail {
1343 return Err(busy(&handle, &place));
1344 } else {
1345 queue_room(agent, &handle, &place)?;
1346 Turn::Queued(place)
1347 }
1348 }
1349 };
1350 let dir = self.reviews.as_deref().ok_or_else(|| {
1351 ToolError::invalid_arguments("review is not available in this session")
1352 })?;
1353 let reviewers: Vec<String> = plan
1354 .reviewer_names()
1355 .into_iter()
1356 .map(str::to_owned)
1357 .collect();
1358 let (rounds, land) = (plan.rounds, plan.land);
1359 let tool = Arc::new(ReviewedTool::new(
1360 Arc::clone(&self.inner),
1361 Arc::clone(&self.jobs.lanes),
1362 Arc::clone(&self.conversations),
1363 plan,
1364 arguments.clone(),
1365 self.session.clone(),
1366 dir,
1367 )?);
1368 let queued = matches!(turn, Turn::Queued(_));
1369 let mut started = self.jobs.start(
1370 Arc::clone(&tool) as Arc<dyn Tool>,
1371 AGENT_TOOL,
1372 &agent.name,
1373 arguments,
1374 context,
1375 turn,
1376 Some(&tool),
1377 )?;
1378 let first = reviewers.first().cloned().unwrap_or_default();
1379 started["review"] = json!({
1380 "reviewers": reviewers,
1381 "rounds": rounds,
1382 "land": review::land_name(land),
1383 "journal": tool.journal_id(),
1384 });
1385 started["note"] = format!(
1386 "{}The agent works on it, then a fresh {first} reviews it (or the next reviewer \
1387 available), for up to {rounds} rounds. SCV reports the result, with its own Review \
1388 and Landing lines, in a new turn when the review ends. agent_status shows the \
1389 round and phase; agent_cancel stops the whole review.",
1390 if queued {
1391 "The conversation is busy, so this reviewed call is queued behind the turns \
1392 ahead of it. "
1393 } else {
1394 ""
1395 }
1396 )
1397 .into();
1398 Ok(ToolOutput::success(started.to_string()))
1399 }
1400
1401 fn start(
1403 &self,
1404 agent: &Offered,
1405 arguments: Value,
1406 context: &ToolContext,
1407 turn: Turn,
1408 ) -> Result<ToolOutput, ToolError> {
1409 let tool: Arc<dyn Tool> = if matches!(turn, Turn::Queued(_)) {
1410 Arc::new(QueuedTool(Arc::clone(&self.inner)))
1411 } else {
1412 Arc::clone(&self.inner) as Arc<dyn Tool>
1413 };
1414 let started = self.jobs.start(
1415 tool,
1416 AGENT_TOOL,
1417 &agent.name,
1418 arguments,
1419 context,
1420 turn,
1421 None,
1422 )?;
1423 Ok(ToolOutput::success(started.to_string()))
1424 }
1425}
1426
1427#[derive(Deserialize)]
1428#[serde(deny_unknown_fields)]
1429struct WaitArgs {
1430 job: String,
1431 timeout_seconds: Option<u64>,
1432}
1433
1434#[derive(Deserialize)]
1435#[serde(deny_unknown_fields)]
1436struct StatusArgs {
1437 job: Option<String>,
1438}
1439
1440pub(crate) struct WaitTool {
1442 pub(crate) jobs: Arc<BackgroundJobs>,
1443 pub(crate) timeouts: Timeouts,
1444}
1445
1446#[async_trait]
1447impl Tool for WaitTool {
1448 fn spec(&self) -> ToolSpec {
1449 let mut timeout = timeout_schema(self.timeouts);
1450 timeout["description"] = format!(
1451 "Seconds to wait before returning the job still running. Defaults to {}; at most {}.",
1452 self.timeouts.default.min(self.timeouts.max).as_secs(),
1453 self.timeouts.max.as_secs()
1454 )
1455 .into();
1456 ToolSpec {
1457 name: "agent_wait".into(),
1458 description: "Wait for a background agent job (the `job` handle an agent call with \
1459 background: true returned) to finish, and return its result. Returns early with \
1460 status running when the timeout passes. Waiting holds your turn open, so the \
1461 user cannot reach you meanwhile; usually let SCV report the result instead."
1462 .into(),
1463 parameters: json!({
1464 "type":"object",
1465 "properties":{"job":{"type":"string"},"timeout_seconds":timeout},
1466 "required":["job"],
1467 "additionalProperties":false
1468 }),
1469 }
1470 }
1471
1472 fn risk(&self, arguments: &Value) -> Result<ToolRisk, ToolError> {
1473 let args: WaitArgs = parse_args(arguments)?;
1474 self.timeouts.resolve(args.timeout_seconds)?;
1475 Ok(ToolRisk::ReadOnly)
1476 }
1477
1478 fn approval_summary(&self, arguments: &Value) -> Result<String, ToolError> {
1479 let args: WaitArgs = parse_args(arguments)?;
1480 Ok(format!(
1481 "Wait for background job {}",
1482 bounded(&args.job, 64)
1483 ))
1484 }
1485
1486 async fn execute(
1487 &self,
1488 arguments: Value,
1489 context: ToolContext,
1490 ) -> Result<ToolOutput, ToolError> {
1491 let args: WaitArgs = parse_args(&arguments)?;
1492 let limit = self.timeouts.resolve(args.timeout_seconds)?;
1493 let value = self
1494 .jobs
1495 .wait(&args.job, limit, &context.cancellation, &context.call_id)
1496 .await?;
1497 Ok(ToolOutput::success(value.to_string()))
1498 }
1499}
1500
1501pub(crate) struct StatusTool {
1503 pub(crate) jobs: Arc<BackgroundJobs>,
1504}
1505
1506#[async_trait]
1507impl Tool for StatusTool {
1508 fn spec(&self) -> ToolSpec {
1509 ToolSpec {
1510 name: "agent_status".into(),
1511 description: "Show this session's background agent jobs: running ones with their \
1512 latest progress, finished ones with their result. Pass job for one job."
1513 .into(),
1514 parameters: json!({
1515 "type":"object",
1516 "properties":{"job":{"type":"string"}},
1517 "additionalProperties":false
1518 }),
1519 }
1520 }
1521
1522 fn risk(&self, arguments: &Value) -> Result<ToolRisk, ToolError> {
1523 let _: StatusArgs = parse_args(arguments)?;
1524 Ok(ToolRisk::ReadOnly)
1525 }
1526
1527 fn approval_summary(&self, arguments: &Value) -> Result<String, ToolError> {
1528 let args: StatusArgs = parse_args(arguments)?;
1529 Ok(args.job.map_or_else(
1530 || "List background jobs".into(),
1531 |job| format!("Show background job {}", bounded(&job, 64)),
1532 ))
1533 }
1534
1535 async fn execute(
1536 &self,
1537 arguments: Value,
1538 context: ToolContext,
1539 ) -> Result<ToolOutput, ToolError> {
1540 let args: StatusArgs = parse_args(&arguments)?;
1541 let value = self.jobs.describe(args.job.as_deref(), &context.call_id)?;
1542 Ok(ToolOutput::success(value.to_string()))
1543 }
1544}
1545
1546#[derive(Deserialize)]
1547#[serde(deny_unknown_fields)]
1548struct CancelArgs {
1549 job: String,
1550}
1551
1552pub(crate) struct CancelTool {
1554 pub(crate) jobs: Arc<BackgroundJobs>,
1555}
1556
1557#[async_trait]
1558impl Tool for CancelTool {
1559 fn spec(&self) -> ToolSpec {
1560 ToolSpec {
1561 name: "agent_cancel".into(),
1562 description: "Stop a running background agent job (the `job` handle an agent call \
1563 with background: true returned), for example when the user no longer wants \
1564 it. The agent and every process it started are stopped; work it already wrote \
1565 stays. Returns the job with status cancelled, and no report turn follows."
1566 .into(),
1567 parameters: json!({
1568 "type":"object",
1569 "properties":{"job":{"type":"string"}},
1570 "required":["job"],
1571 "additionalProperties":false
1572 }),
1573 }
1574 }
1575
1576 fn risk(&self, arguments: &Value) -> Result<ToolRisk, ToolError> {
1577 let _: CancelArgs = parse_args(arguments)?;
1578 Ok(ToolRisk::Process)
1579 }
1580
1581 fn approval_summary(&self, arguments: &Value) -> Result<String, ToolError> {
1582 let args: CancelArgs = parse_args(arguments)?;
1583 Ok(format!("Stop background job {}", bounded(&args.job, 64)))
1584 }
1585
1586 async fn execute(
1587 &self,
1588 arguments: Value,
1589 context: ToolContext,
1590 ) -> Result<ToolOutput, ToolError> {
1591 let args: CancelArgs = parse_args(&arguments)?;
1592 let value = self.jobs.cancel(&args.job, &context.call_id).await?;
1593 Ok(ToolOutput::success(value.to_string()))
1594 }
1595}
1596
1597#[cfg(test)]
1598mod tests;