Skip to main content

scv_tools/delegate/
background.rs

1//! Background delegations: an `agent` call with `background: true` returns a
2//! job handle at once while the agent keeps working; `agent_status` and
3//! `agent_wait` observe the job, `agent_cancel` stops it, and the session is
4//! told when it finishes so the server can report it in a turn of its own.
5//!
6//! Each call that starts a job, or shows the model a job's result, leaves a
7//! [`JobChange`] under its call ID, which the server hands to the session's
8//! clients with the call's `tool.completed` ([`BackgroundJobs::take_changes`]).
9//!
10//! A finished job stays unreported until a report turn about it succeeds. A
11//! failed one makes it due again after a delay, a bounded number of times;
12//! after that, or when another turn must not be tried, the server reports it
13//! to the client directly ([`BackgroundJobs::report_failed`]).
14//!
15//! A call that continues a conversation takes its place in that
16//! conversation's lane as it arrives. When a turn or another call is
17//! ahead, the call's busy policy decides: it steers the running turn, waits
18//! in the foreground, fails, or becomes a queued job that runs once its
19//! place comes up.
20//!
21//! A call with `review` runs as a reviewed job ([`review`](super::review)):
22//! the same job, whose one call runs the whole builder and reviewer loop and
23//! whose changes and report carry SCV's outcome.
24
25pub(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
60/// Finished jobs a session keeps for `agent_status` beyond the running ones.
61const MAX_FINISHED: usize = 16;
62/// Characters of a finished job's reply quoted in a server-started report turn.
63const REPORT_REPLY_CHARS: usize = 6000;
64/// Jobs one report turn covers; any more wait for the next.
65const REPORT_MAX_JOBS: usize = 4;
66/// How long after each failed report turn the next one is due: after the
67/// first failure, then after the second. Once they run out, the job is
68/// reported directly.
69const REPORT_RETRY_DELAYS: [Duration; 2] = [Duration::from_secs(30), Duration::from_secs(120)];
70/// How long `agent_cancel` waits for a stopped job to settle.
71const CANCEL_SETTLE: Duration = Duration::from_secs(10);
72/// Job changes kept for calls whose `tool.completed` has not taken them,
73/// such as a call whose turn was aborted; the oldest go first.
74const MAX_PENDING_CHANGES: usize = 64;
75/// Characters of a delegated prompt's first line kept as its job's task.
76const TASK_CHARS: usize = 80;
77
78/// One session's background jobs. Dropping the store (with the session's
79/// tools) cancels every job still running.
80pub struct BackgroundJobs {
81    limit: usize,
82    state: Mutex<JobsState>,
83    cancellation: CancellationToken,
84    /// Woken whenever a job finishes, so the session can report it.
85    finished: Option<mpsc::UnboundedSender<()>>,
86    /// Decides a running job's nested approval requests, since no turn is
87    /// left to carry them to a person. Without it they are denied.
88    approvals: Option<Arc<dyn ApprovalGate>>,
89    /// The `agent` calls continuing each conversation, in arrival order.
90    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 by the ID of the call that made them, oldest first.
107    changes: Vec<(String, JobChange)>,
108    /// The final outcomes of cancelled reviewed jobs the session is owed.
109    owed: owed::Owed,
110}
111
112impl JobsState {
113    /// Bring the outcome in the change the call `call_id` made for
114    /// `outcome`'s job up to date, if no client has taken it yet.
115    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    /// Keep `change` for the call `call_id`, which a client learns of with
127    /// that call's `tool.completed`. A call without an ID has no event.
128    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    /// The tool that started it, `agent`.
143    tool: String,
144    /// The agent that runs it, such as `codex`.
145    agent: String,
146    /// The first line of the delegated prompt, shortened.
147    task: String,
148    started: Instant,
149    progress: ProgressSink,
150    last_progress: Option<String>,
151    /// Stops this job alone.
152    cancel: CancellationToken,
153    /// `agent_cancel` stopped it.
154    cancelled: bool,
155    /// A busy conversation's queued prompt, which `agent.max_queued_turns`
156    /// bounds instead of `agent.max_background`.
157    queued: bool,
158    outcome: Option<Outcome>,
159    /// The model has seen the result (through `agent_wait`, `agent_status`,
160    /// or a report turn that completed), the user stopped its report turn,
161    /// the server reported it directly, or the model asked for the stop with
162    /// `agent_cancel`, so it needs no report turn.
163    reported: bool,
164    /// A report turn about it runs.
165    reporting: bool,
166    /// Report turns about it that failed.
167    report_attempts: u32,
168    /// After a failed report turn: when the next one is due.
169    report_due: Option<tokio::time::Instant>,
170    done: watch::Receiver<bool>,
171    /// A reviewed job's record, whose outcome its changes and report state.
172    review: Option<Arc<Record>>,
173    /// `agent_cancel` waits for it, and describes it after: never pruned
174    /// meanwhile.
175    cancelling: bool,
176}
177
178/// `agent_cancel` waiting for a job to stop. Settled after the wait, or,
179/// if the call is dropped first, as when its turn is aborted, its change
180/// withdrawn and any outcome it owed left to an update.
181struct Waiting<'a> {
182    jobs: &'a BackgroundJobs,
183    job: String,
184    call: String,
185    /// The job is reviewed: the session is owed its final outcome.
186    owed: bool,
187    settled: bool,
188}
189
190impl Waiting<'_> {
191    /// The wait is over: the change the call made says the final outcome
192    /// if the job stopped, else the outcome as decided, its journal still
193    /// open; [`BackgroundJobs::take_changes`] says the final one if the job
194    /// stops before the change is sent.
195    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        // Its call is gone, so its change is never sent.
229        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/// What a failed report turn leaves its jobs to.
247#[derive(Debug, PartialEq, Eq)]
248pub enum ReportFailure {
249    /// Another report turn about them is due after this long.
250    Retry(Duration),
251    /// No more report turns: the server reports them directly. `attempts`
252    /// counts the report turns that failed.
253    GiveUp {
254        attempts: u32,
255        reports: Vec<JobReport>,
256    },
257    /// The model saw each of them through a call during the turn.
258    Settled,
259}
260
261impl Drop for BackgroundJobs {
262    fn drop(&mut self) {
263        self.cancellation.cancel();
264    }
265}
266
267impl BackgroundJobs {
268    /// At most `limit` jobs run at once. `finished` is woken as jobs finish.
269    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    /// Decide running jobs' nested approval requests with `gate`, which must
281    /// never grant more than the session's foreground would.
282    #[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    /// Start `tool` with `arguments` in the background for the call in
297    /// `context` (whose workspace and call ID it takes; the job has its own
298    /// cancellation), on `agent`, and return its job's
299    /// `{"job","agent","status":"running","background":true}` description.
300    /// A reviewed job's `review` records its start in its journal first, and
301    /// the call fails if it cannot. That write happens outside the job
302    /// table's lock, so a slow disk never holds up the session's other job
303    /// calls: the job's number is taken first, and a failed write leaves it
304    /// unused.
305    #[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            // Checked again: another call may have started a job meanwhile.
342            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            // No turn is left to ask a person, so nested approval requests
393            // get the session's unattended answer, or are denied.
394            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                        // A reviewed job ends its review even when its loop
415                        // never ran, so its result says how it ended.
416                        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            // Free the conversation before the result is visible, so a call
426            // continuing it after `agent_wait` finds it idle.
427            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    /// Refuse a job beyond `agent.max_background`; a queued prompt counts
452    /// toward `agent.max_queued_turns` instead.
453    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    /// Stop `job` for the call `call_id` and wait briefly for it to settle,
470    /// then describe it. A stopped job needs no report turn: the model asked
471    /// for the stop.
472    ///
473    /// A reviewed job's decision is fixed at once, but its journal goes on
474    /// until it really stops, so the session is owed its final outcome
475    /// ([`owed`]): the call's change says it if the job stops before the
476    /// change is sent, and otherwise says the journal is still open, and a
477    /// `background.updated` follows ([`Self::take_updates`]). The wait
478    /// reads only the job's live state, never its journal.
479    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    /// The final outcomes of cancelled reviewed jobs the session is owed
525    /// and has not been sent, for the server's `background.updated`. Each
526    /// is being sent until [`Self::updated`] says how that went.
527    pub fn take_updates(&self) -> Vec<JobOutcome> {
528        self.state().owed.take_updates()
529    }
530
531    /// The update naming `jobs` reached the client, when `delivered`, which
532    /// ends what was owed; or it could not be sent, which leaves them due.
533    pub fn updated(&self, jobs: &[String], delivered: bool) {
534        self.state().owed.updated(jobs, delivered);
535    }
536
537    /// The `tool.completed` of the call `call_id`, carrying the changes
538    /// [`Self::take_changes`] gave, reached the client, when `delivered`, or
539    /// could not be sent, as when its turn was cancelled first. A change
540    /// that did not deliver a reviewed job's final outcome leaves it to a
541    /// `background.updated`.
542    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    /// A turn ended: no change of its calls can be sent any more, so what
550    /// such a change owed goes to a `background.updated`.
551    pub fn turn_ended(&self) {
552        let due = self.state().owed.turn_ended();
553        if due {
554            self.wake();
555        }
556    }
557
558    /// For tests of how a client of this store delivers reviewed jobs'
559    /// outcomes, such as the server's, with the `test-support` feature: owe
560    /// the session the final outcome of `change.job`, through `change`,
561    /// recorded for the call `call_id` once `agent_cancel` waited, which says
562    /// the journal is still open; `last` is the final outcome if the job has
563    /// stopped.
564    #[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    /// For the same tests: the job of `outcome` stops with it.
572    #[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    /// Wake the session, as when a job finishes, for what is now due.
581    fn wake(&self) {
582        if let Some(finished) = &self.finished {
583            let _ = finished.send(());
584        }
585    }
586
587    /// Wait up to `limit` for `job` and describe it for the call `call_id`;
588    /// a finished job is then marked seen.
589    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    /// Describe one job, or every job this session remembers, for the call
611    /// `call_id`; finished jobs described are marked seen.
612    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    /// The jobs the call `call_id` started or showed the model the result of,
643    /// for its `tool.completed`, which then reports how its sending went
644    /// ([`Self::published`]). A cancelled reviewed job's change says its
645    /// final outcome if the job has stopped by now, superseding the
646    /// snapshot taken when the cancel's wait ended.
647    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    /// Finished jobs the model has not seen yet whose report is due, at most
665    /// a few, for a report turn. Each stays unreported, and is not taken
666    /// again, until the turn ends: [`report_settled`](Self::report_settled)
667    /// or [`report_failed`](Self::report_failed).
668    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    /// The report turn about `jobs` ended without failing: it completed, so
684    /// the model has seen them, or the user cancelled it. Either settles them.
685    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    /// The report turn about `jobs` failed, so the model has not seen them
695    /// (a failed turn leaves no history). With `retry`, another turn is due
696    /// after a delay, up to three turns in all; otherwise, or once they are
697    /// used up, the jobs are settled and returned for the server to report
698    /// directly. A turn that reported several jobs decides for all of them,
699    /// by the one that failed most.
700    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        // One the turn's model looked up through a call is settled already.
712        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    /// Make every report waiting out a failed turn due now, as once a turn
738    /// succeeds and the model is evidently reachable again. Whether any was
739    /// waiting.
740    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    /// When the earliest report waiting out a failed turn is due; it may
750    /// already have passed.
751    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    /// Jobs still running or whose result is not reported yet, including
761    /// those a report turn is reporting now, and cancelled reviewed jobs
762    /// whose final outcome the session is still owed, until it was sent;
763    /// each job once.
764    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    /// Whether any job is still running.
777    pub fn running(&self) -> usize {
778        self.state()
779            .jobs
780            .iter()
781            .filter(|job| job.outcome.is_none())
782            .count()
783    }
784}
785
786/// How a job's call takes its conversation's turn.
787pub(in crate::delegate) enum Turn {
788    /// A new conversation, which nothing else can reach yet.
789    New,
790    /// The conversation's next turn, which nothing was ahead of.
791    Next(Place),
792    /// A prompt queued behind the turns ahead of it.
793    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    // The session ended first: its jobs were cancelled with it.
807    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            // Stopped on request: the model already knows.
817            job.reported |= job.cancelled;
818            last = job.outcome();
819        }
820        // What a cancel owed of it is now final.
821        if let Some(last) = last {
822            state.owed.stopped(id, &last);
823        }
824        // Keep every running job and the newest finished ones.
825        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    /// Finished, unseen by the model, and not in a report turn.
847    fn awaits_report(&self) -> bool {
848        self.outcome.is_some() && !self.reported && !self.reporting
849    }
850
851    /// The finished job's result, for a report.
852    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    /// A reviewed job's outcome as it stands. A cancelled job's is what it
873    /// was when the cancel arrived.
874    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    /// This job with `status`, as its clients learn of it: a reviewed job's
888    /// settling change carries its outcome.
889    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    /// The job as the model reads it, and its change when this is the first
906    /// time the model sees its result.
907    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
952/// The agent tool's structured result, or its text when it was not JSON.
953fn 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
960/// How a finished job ended: its agent's reported status, else whether its
961/// call failed.
962fn 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
970/// The first non-empty line of `prompt`, at most [`TASK_CHARS`] characters,
971/// which names a job to people.
972fn 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
996/// What the model is told when a report includes reviewed jobs, whose
997/// outcome SCV states itself.
998const 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
1012/// The server-started turn that reports finished jobs to the model.
1013pub 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
1023/// What the model is told of jobs the server reported directly, because
1024/// their report turns failed with `error`: the user has their results.
1025pub 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
1036/// The `agent` tool, able to run a call in the background as well, and to
1037/// run one as a reviewed job.
1038pub(crate) struct BackgroundCapable {
1039    pub(crate) inner: Arc<AgentTool>,
1040    pub(crate) jobs: Arc<BackgroundJobs>,
1041    /// The session's conversations, which a reviewed job pins and releases.
1042    pub(crate) conversations: Arc<ConversationStore>,
1043    /// Where reviewed jobs keep their journals; without it, `review` is not
1044    /// offered.
1045    pub(crate) reviews: Option<PathBuf>,
1046    /// The SCV session, which a review's journal names.
1047    pub(crate) session: Option<String>,
1048}
1049
1050/// The `agent` tool for a queued prompt, whose turn has come by the time it
1051/// runs, so no busy policy applies to it again.
1052struct 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
1074/// An agent call's arguments with `background` and `review` split off.
1075struct Split {
1076    arguments: Value,
1077    /// `background` as the call gave it.
1078    background: Option<bool>,
1079    /// The `review` object, when the call gave one.
1080    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        // Validate before returning a job handle, so a bad call fails now.
1190        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        // Taken before anything runs or waits, so calls keep their order.
1204        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                // Held until the turn ends, so a later call finds it busy.
1210                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            // A background call cannot hold its caller, so it queues instead.
1226            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
1241/// The `session busy` error for a call to conversation `handle` that found
1242/// a turn running, or calls waiting, at `place`.
1243fn busy(handle: &str, place: &Place) -> ToolError {
1244    // Calls ahead of this one other than the running turn.
1245    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
1256/// Refuse a prompt queued at `place` beyond `agent.max_queued_turns`.
1257fn 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    /// `inner` with `jobs` and a store of its own, without reviews: for
1272    /// tests of plain background calls.
1273    #[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    /// Whether calls may set `review`: background jobs exist (this tool),
1291    /// the session keeps journals, and an offered agent can continue a
1292    /// conversation.
1293    fn offers_review(&self) -> bool {
1294        self.reviews.is_some() && !self.inner.continuing().is_empty()
1295    }
1296
1297    /// Check a reviewed call, before approval and again before it starts.
1298    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    /// Start a reviewed call as a background job. A busy builder
1321    /// conversation queues it, or fails it with `on_busy: fail`; it never
1322    /// waits, since it is a background job, and never steers, since its
1323    /// prompt opens a review of its own.
1324    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    /// Start the call on `agent` as a background job of this session.
1402    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
1440/// `agent_wait`: block until a background job finishes, or the timeout.
1441pub(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
1501/// `agent_status`: describe one background job or all of them.
1502pub(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
1552/// `agent_cancel`: stop one running background job.
1553pub(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;