Skip to main content

magi/
deputy.rs

1//! Deputies: the seat that waits on a conductor's question, so the conductor
2//! never has to - and on land's merge approval, whose free-text reply used to
3//! reach nobody either.
4//!
5//! [`crate::conduct`] files a question for the operator and moves on; it is
6//! non-blocking by construction, because one task's question must not park the
7//! whole polling loop. The price used to be that nothing at all was waiting on
8//! it: the owner's free-text reply landed in a thread no agent read, while the
9//! phone said "waiting for the agent". Question f1dc on task 684d sat open
10//! forever on a "setup done".
11//!
12//! A deputy is the answer. For each open conductor question, `magi serve`
13//! runs one short-lived agent seat that is handed what the conductor knew
14//! ([`brief`]: the task, why it asked, what each option leads to), blocks on
15//! the question with `magi ask --wait`, answers back in the thread with
16//! `magi ask --thread`, and - when the owner's own words clearly pick an
17//! option - records that with `magi ask --settle`. Applying the outcome is not
18//! the deputy's job: an answered conductor question goes through the daemon's
19//! existing `resolve_blockers` / action path, exactly as one tapped on the
20//! phone does.
21//!
22//! # Invariants
23//!
24//! - **One seat per question, keyed by seat name.** `deputy-<question id>`,
25//!   never the conductor's shared seat (`conduct/seat.json`, overwritten by
26//!   every cycle) and never an agent id.
27//! - **Persisted on the question** ([`crate::ask::Deputy`]), always through
28//!   [`Questions::update`]: the owner's say and answer write the same file. A
29//!   restarted daemon resumes the seat only when [`agent::has_session`] says
30//!   it can; otherwise it starts a fresh seat and re-sends the whole context,
31//!   never assuming memory.
32//! - **Bounded.** At most `daemon.max_deputies` at once; at most
33//!   [`MAX_STARTS`] starts per question, and a restart does not reset that. A
34//!   deputy never outlives the question's `answer_timeout`: the waiter retires
35//!   the question at its deadline ([`crate::waiter`]) and the task goes to a
36//!   machine hold (`daemon::resolve_blockers`).
37//! - **Never a second agent on one question.** A fresh lease or the claim file
38//!   means somebody is already on it.
39//! - **A merge approval is served too, but stays land's.** Its deputy has no
40//!   `cwd` (so the waiter never touches it), its deadline is `asked_at +
41//!   answer_timeout` and never moves on a reply ([`deadline`]), and the only
42//!   thing that retires it is `daemon::land_resume_state`. A say alone never
43//!   merges: `--settle` accepts `merge` only for the owner's own word `merge`.
44//! - **A release-watch question is served too, and stays the watcher's.** The
45//!   escalation, local-mode approval and failed-release questions
46//!   ([`crate::release_watch`], node `release-bump`, seat `release-watch`) have
47//!   no asker, so a say reached nobody. Its deputy has no `cwd` either, a
48//!   fixed `asked_at + answer_timeout` deadline ([`deadline`]), and applies
49//!   nothing: `--settle` records a choice and the watcher applies it on its next
50//!   lap. A local approval's `merge` / `hold` go through the merge-approval
51//!   rules ([`merge_gated`]). A deputy never closes or merges the pull request.
52//! - **Every open question the owner can say something to has a listener.**
53//!   [`Kind::Triage`] serves the triage questions; [`Kind::Generic`] is the
54//!   fallback for any other question with choices and no `cwd` (the divergence
55//!   question today, a node nobody has written yet tomorrow). It is decided by
56//!   the missing `cwd`, never by node name. Only a conductor question is ever
57//!   given a `cwd` ([`Deputies::attach`]); the other kinds would otherwise be
58//!   resumed and expired by the waiter beside the deputy. A test enumerates
59//!   every node filed without an asker, so a new kind of question cannot ship
60//!   without one.
61//! - **No new authority.** A deputy does not edit, merge or touch the queue,
62//!   and `--settle` accepts only an offered label backed by a verbatim quote
63//!   of the owner, never on a task the operator holds.
64
65use std::collections::{BTreeMap, HashMap, HashSet};
66use std::path::{Path, PathBuf};
67use std::sync::Arc;
68use std::time::{Duration, Instant};
69
70use anyhow::{Context as _, Result};
71use jiff::Timestamp;
72use tokio::task::JoinSet;
73
74use crate::agent::{self, Invocation, SeatState};
75use crate::ask::{ChoiceAction, Deputy, Question, Questions, Waiter as Note, WaiterKind, Who};
76use crate::config::Config;
77use crate::notices::{self, Notice};
78use crate::prompt;
79
80/// Node name a deputy runs under (`MAGI_NODE`); `magi ask --settle` accepts
81/// it and nothing else.
82pub const NODE: &str = "deputy";
83
84/// How often the runner looks at the store.
85pub const TICK: Duration = Duration::from_secs(5);
86
87/// Most turns ever started for one question. A deputy that keeps ending
88/// without an answer is a problem for the owner, not something to retry until
89/// the quota is gone.
90pub const MAX_STARTS: u32 = 3;
91
92/// Least gap between two starts for one question.
93const RESTART_AFTER: Duration = Duration::from_secs(30);
94
95/// Added to the question's remaining time for the invocation's wall clock, so
96/// the deputy's own `magi ask --wait` reaches the deadline first.
97const SLACK_SECS: u64 = 120;
98
99/// A claim file older than this was left by a process that died.
100const CLAIM_STALE: Duration = Duration::from_secs(60);
101
102/// How long the context-handover turn may take.
103const HANDOVER_TIMEOUT: Duration = Duration::from_secs(10 * 60);
104
105/// Stops the deputy when true: the daemon is parking.
106pub type Halt = Arc<dyn Fn() -> bool + Send + Sync>;
107
108/// What the conductor knew about one question, for its deputy.
109///
110/// `reason` is the conductor's own one-line reasoning (may be empty). Each
111/// choice is listed with what picking it does: an attached action, or - for
112/// the usual conductor question - that the answer is recorded on the task and
113/// the task is unblocked with it in its instructions.
114pub fn brief(
115    task_id: &str,
116    reason: &str,
117    choices: &[String],
118    actions: &BTreeMap<String, ChoiceAction>,
119) -> String {
120    let reason = reason.trim();
121    let mut s = format!(
122        "Task {task_id} (`magi task show {task_id}`) is blocked on this question. \
123         The conductor asked because: {}\n\n\
124         When the owner answers, the daemon applies the exact attached action \
125         once, after rechecking its preconditions. You only record the choice; \
126         you never execute maintenance or edit repository or run records.",
127        if reason.is_empty() {
128            "(it recorded no reasoning)"
129        } else {
130            reason
131        }
132    );
133    if !choices.is_empty() {
134        s.push_str("\n\nWhat each option does:");
135        for c in choices {
136            match actions.get(c) {
137                Some(a) => s.push_str(&format!("\n- `{c}`: also {}", a.describe())),
138                None => s.push_str(&format!(
139                    "\n- `{c}`: no executable operation; keep held and ask for a supported choice"
140                )),
141            }
142        }
143    }
144    s
145}
146
147/// What a deputy serves: the question kinds it is attached to.
148#[derive(Debug, Clone, Copy, PartialEq, Eq)]
149pub enum Kind {
150    /// A conductor's question ([`crate::conduct::NODE`]).
151    Conduct,
152    /// The merge approval ([`crate::land::APPROVAL_NODE`]).
153    Land,
154    /// A question the release watcher filed ([`crate::release_watch`]).
155    Release,
156    /// A triage question about a held task or a stuck dependency root
157    /// ([`crate::triage::NODE`], [`crate::triage::DEPS_NODE`]).
158    Triage,
159    /// Any other question nobody asked from inside a run (no `cwd`) that offers
160    /// choices: the fallback, so a node nobody has met yet still has a listener.
161    Generic,
162}
163
164/// The kind of question a deputy serves for `q`, `None` for every other.
165///
166/// The fallback is decided by the absence of a `cwd`, never by the node name: a
167/// question filed with `magi ask` inside a run records its `cwd` and has an
168/// asker (and the waiter), and the node a seat asks from is the seat's own name
169/// (a reviewer's `review` is also the divergence question's node). A question
170/// with no choices cannot be settled and is a notice, not something to answer.
171pub fn kind_of(q: &Question) -> Option<Kind> {
172    match q.node.as_str() {
173        crate::conduct::NODE => Some(Kind::Conduct),
174        crate::land::APPROVAL_NODE => Some(Kind::Land),
175        // The seat matters: `bump` files choice-less notices on the same node,
176        // which nobody can answer and which must not cost a deputy.
177        crate::bump::NOTICE_NODE if q.seat == "release-watch" => Some(Kind::Release),
178        crate::bump::NOTICE_NODE => None,
179        crate::triage::NODE | crate::triage::DEPS_NODE => Some(Kind::Triage),
180        _ if q.cwd.is_none() && !q.choices.is_empty() => Some(Kind::Generic),
181        _ => None,
182    }
183}
184
185/// What a deputy is told about a question of no known kind: only what the
186/// question itself stored.
187pub fn generic_brief(
188    q: &Question,
189    actions: &BTreeMap<String, ChoiceAction>,
190    home: &Path,
191) -> String {
192    // `Question::run` may also hold a task id; only a run id can be shown.
193    let run = crate::run::is_run_id(&q.run).then_some(q.run.as_str());
194    let knows = if run.is_some() {
195        "You know what the question itself says and what you can learn from \
196         its run by the read-only investigation described below."
197    } else {
198        "You know only what the question itself says."
199    };
200    let mut s = format!(
201        "This question was filed by magi (node `{}`, seat `{}`) with no agent \
202         waiting on it. Magi records the owner's answer and the component that \
203         asked applies it; you apply nothing. {knows} A choice whose effect you \
204         cannot read is not yours to guess: do not settle it, ask the owner \
205         what they mean with `--thread` instead.",
206        q.node, q.seat
207    );
208    if let Some(run) = run {
209        let artifacts = home.join("runs").join(run).join("artifacts");
210        s.push_str(&format!(
211            "\n\nThis question concerns run `{run}`. You may investigate it on \
212             your own, read-only: run `magi show {run}` and read files under \
213             `{}`. When the owner asks about details (for example which part \
214             was wrong), look it up yourself and answer; do not reply that you \
215             cannot see the details, and do not tell the owner to run `magi show` \
216             themselves. Investigating never changes anything: edit, commit, push \
217             and apply nothing. If the run cannot be read, say exactly that and do \
218             not guess at the cause or at what an option does.",
219            artifacts.display()
220        ));
221    }
222    if q.node == crate::github_text::ASK_NODE {
223        s.push_str("\n\n");
224        s.push_str(&crate::github_text::deputy_brief());
225    }
226    if !q.choices.is_empty() {
227        s.push_str("\n\nWhat each option does:");
228        for c in &q.choices {
229            match actions.get(c) {
230                Some(a) => s.push_str(&format!("\n- `{c}`: also {}", a.describe())),
231                None => s.push_str(&format!(
232                    "\n- `{c}`: {}",
233                    crate::choice_effect::describe(q, c)
234                        .unwrap_or_else(|| "unsupported legacy choice; do not settle".into())
235                )),
236            }
237        }
238    }
239    s
240}
241
242/// Is picking `label` on `q` something that cannot be taken back, so that
243/// `--settle` must hold the owner's words to the same mechanical standard as a
244/// merge (`land::unhedged`)? Discarding a task deletes it; a divergence answer
245/// drops commits.
246pub fn destructive(q: &Question, label: &str) -> bool {
247    crate::consult::destructive_effect(q, label)
248}
249
250/// Does `--settle` hold `q` to the merge-approval rules (the verbatim quote of
251/// the latest message for `merge`, the whole message for `hold`)? A merge
252/// approval, and any other served question that offers `merge` (the release
253/// watcher's local-mode approval merges just as irreversibly).
254pub fn merge_gated(q: &Question) -> bool {
255    q.node == crate::land::APPROVAL_NODE
256        || (kind_of(q).is_some_and(|k| k != Kind::Conduct)
257            && q.choices.iter().any(|c| c == crate::land::APPROVE))
258}
259
260/// Does `q`'s clock run from `asked_at` and never move on a reply? True for the
261/// questions that something other than the waiter retires: a merge approval
262/// (land) and a release-watch question (the watcher, by silence being a hold).
263pub fn fixed_clock(q: &Question) -> bool {
264    matches!(kind_of(q), Some(Kind::Land | Kind::Release)) || q.node == crate::github_text::ASK_NODE
265}
266
267/// Second after which nobody is to be started or kept on `q`.
268///
269/// A conductor question runs from its last activity (`magi ask --thread`
270/// re-arms it, and the waiter retires it). A merge approval never moves: it
271/// runs from `asked_at`, exactly where `daemon::land_resume_state` abandons it,
272/// so a conversation cannot stretch the hold and land stays the only place that
273/// retires one.
274pub fn deadline(q: &Question, default_timeout: u64) -> i64 {
275    let secs = if q.answer_timeout > 0 {
276        q.answer_timeout
277    } else {
278        default_timeout
279    };
280    let from = if fixed_clock(q) {
281        q.asked_at.as_second()
282    } else {
283        q.last_activity()
284    };
285    from.saturating_add(secs as i64)
286}
287
288/// Dispatch and text admission share the fixed-clock and unread-word rules.
289pub fn restart_in_time(q: &Question, default_timeout: u64, now: Timestamp) -> bool {
290    now.as_second() <= deadline(q, default_timeout) || q.unread_from_owner().is_some()
291}
292
293/// Can `magi serve` start a deputy at all under `cfg`? Not when deputies are
294/// switched off (`daemon.max_deputies = 0`) or the config could not be read.
295///
296/// Also false when the agent the deputy would run as cannot be resolved
297/// (`agent` is the deputy's recorded agent, empty when it has none): `turn`
298/// fails on that before it counts a start, so it would never reach
299/// [`MAX_STARTS`].
300pub fn can_start(cfg: Option<&Config>, agent: &str) -> bool {
301    cfg.is_some_and(|c| {
302        c.daemon.max_deputies > 0
303            && ((!agent.is_empty() && c.agent(agent).is_ok()) || c.resolve_roles().is_ok())
304    })
305}
306
307/// The agent a question's deputy records, empty when it has none yet.
308pub fn agent_of(q: &Question) -> &str {
309    q.deputy.as_ref().map_or("", |d| d.agent.as_str())
310}
311
312/// Has this question's deputy run out of starts, or can none ever start, with
313/// the deadline gone?
314///
315/// Then nothing will ever read an unread say, and the waiter must retire the
316/// question anyway instead of deferring to a deputy that no longer starts.
317/// `startable` is [`can_start`]; a fresh lease is the caller's to check.
318pub fn exhausted_past_deadline(
319    q: &Question,
320    startable: bool,
321    default_timeout: u64,
322    now: Timestamp,
323) -> bool {
324    q.status.open()
325        && q.deputy
326            .as_ref()
327            .is_some_and(|d| d.starts >= MAX_STARTS || !startable)
328        && now.as_second() > deadline(q, default_timeout)
329}
330
331/// The deputy runner: its own task inside `magi serve`, beside the waiter.
332pub struct Deputies {
333    store: Questions,
334    home: PathBuf,
335    cfg: Option<Config>,
336    /// Working directory for a question that recorded none.
337    fallback_repo: PathBuf,
338    max: usize,
339    halt: Halt,
340    tasks: JoinSet<String>,
341    inflight: HashSet<String>,
342    /// Earliest next start per question.
343    memo: HashMap<String, Instant>,
344}
345
346impl Deputies {
347    /// A runner over `store`, with at most `max` deputies at once.
348    pub fn new(
349        store: Questions,
350        home: PathBuf,
351        cfg: Option<Config>,
352        fallback_repo: PathBuf,
353        max: usize,
354        halt: Halt,
355    ) -> Self {
356        Self {
357            store,
358            home,
359            cfg,
360            fallback_repo,
361            max,
362            halt,
363            tasks: JoinSet::new(),
364            inflight: HashSet::new(),
365            memo: HashMap::new(),
366        }
367    }
368
369    fn default_timeout(&self) -> u64 {
370        self.cfg
371            .as_ref()
372            .map_or(Config::default().graph.answer_timeout, |c| {
373                c.graph.answer_timeout
374            })
375    }
376
377    fn reap(&mut self) {
378        while let Some(done) = self.tasks.try_join_next() {
379            if let Ok(id) = done {
380                self.inflight.remove(&id);
381            }
382        }
383        if self.tasks.is_empty() {
384            self.inflight.clear();
385        }
386    }
387
388    /// Give a question the record a deputy needs: the deputy itself (with a
389    /// brief rebuilt from what the question stored, when it was filed before
390    /// deputies existed - a lost reason is not invented) and the deadline.
391    ///
392    /// A conductor question also gets the working directory the waiter and
393    /// `magi ask --wait` use. Every other kind never does: `cwd` is what makes a
394    /// question the waiter's (it would resume and expire it beside the deputy),
395    /// and a release-watch question, a merge approval, a triage question and a
396    /// fallback one have no asker to resume.
397    fn attach(&self, q: &Question, kind: Kind) -> Option<Question> {
398        let default_timeout = self.default_timeout();
399        let repo = self.fallback_repo.to_string_lossy().into_owned();
400        let state = match kind {
401            Kind::Land => crate::run::RunState::load(&q.run).ok(),
402            Kind::Conduct | Kind::Release | Kind::Triage | Kind::Generic => None,
403        };
404        // The deadline land itself enforces for this run.
405        let timeout = state
406            .as_ref()
407            .map_or(default_timeout, |s| s.config.graph.answer_timeout);
408        self.store
409            .update(&q.id, |r| {
410                if r.deputy.is_none() {
411                    r.deputy = Some(Deputy::new(match kind {
412                        Kind::Conduct => brief(&r.run, &r.detail, &r.choices, &r.actions),
413                        Kind::Land => crate::land::deputy_brief(r, state.as_ref()),
414                        Kind::Release => crate::release_watch::deputy_brief(r, &self.home),
415                        Kind::Triage => crate::triage::deputy_brief(
416                            r,
417                            &crate::queue::Queue::at(self.home.join("queue")),
418                        ),
419                        Kind::Generic => generic_brief(r, &r.actions, &self.home),
420                    }));
421                }
422                if kind == Kind::Conduct && r.cwd.is_none() {
423                    r.cwd = Some(repo.clone());
424                }
425                if r.answer_timeout == 0 {
426                    r.answer_timeout = match kind {
427                        Kind::Conduct => default_timeout,
428                        Kind::Land => timeout,
429                        Kind::Release | Kind::Triage | Kind::Generic => default_timeout,
430                    };
431                }
432                Ok(())
433            })
434            .map(|(r, ())| r)
435            .map_err(|e| tracing::warn!("question {}: cannot attach a deputy: {e:#}", q.short()))
436            .ok()
437    }
438
439    /// Look at every open conductor question once and start a deputy on each
440    /// that has none, within the limits. Returns without waiting for them.
441    pub fn tick(&mut self, now: Timestamp) {
442        self.reap();
443        for q in self.store.list() {
444            if (self.halt)() {
445                return;
446            }
447            let Some(kind) = kind_of(&q) else {
448                continue;
449            };
450            if !q.status.open() {
451                continue;
452            }
453            let needs_cwd = kind == Kind::Conduct && q.cwd.is_none();
454            let q = if q.deputy.is_none() || needs_cwd || q.answer_timeout == 0 {
455                match self.attach(&q, kind) {
456                    Some(q) => q,
457                    None => continue,
458                }
459            } else {
460                q
461            };
462            let Some(dep) = q.deputy.as_ref() else {
463                continue;
464            };
465            if self.inflight.contains(&q.id)
466                || self.store.read_lease(&q.id).is_some_and(|l| l.fresh(now))
467            {
468                continue;
469            }
470            if dep.starts >= MAX_STARTS {
471                continue;
472            }
473            // Past the deadline the waiter retires the question; a say that
474            // arrived in time is still read first.
475            if !restart_in_time(&q, self.default_timeout(), now) {
476                continue;
477            }
478            if self.inflight.len() >= self.max || !can_start(self.cfg.as_ref(), dep.agent.as_str())
479            {
480                continue;
481            }
482            if matches!(self.memo.get(&q.id), Some(until) if Instant::now() < *until) {
483                continue;
484            }
485            self.memo
486                .insert(q.id.clone(), Instant::now() + RESTART_AFTER);
487            self.inflight.insert(q.id.clone());
488            let job = Job {
489                store: self.store.clone(),
490                home: self.home.clone(),
491                cfg: self.cfg.clone(),
492                fallback_repo: self.fallback_repo.clone(),
493                halt: Arc::clone(&self.halt),
494            };
495            let id = q.id.clone();
496            self.tasks.spawn(async move {
497                if let Err(e) = job.turn(&id).await {
498                    tracing::warn!("deputy for question {}: {e:#}", crate::ask::short_id(&id));
499                }
500                id
501            });
502        }
503    }
504
505    /// Wait for every deputy turn in flight. The daemon never calls this; the
506    /// tests do, to look at the record once a turn is over.
507    pub async fn drain(&mut self) {
508        while let Some(done) = self.tasks.join_next().await {
509            if let Ok(id) = done {
510                self.inflight.remove(&id);
511            }
512        }
513        self.inflight.clear();
514    }
515}
516
517/// Everything one deputy turn needs, owned so it can run detached.
518struct Job {
519    store: Questions,
520    home: PathBuf,
521    cfg: Option<Config>,
522    fallback_repo: PathBuf,
523    halt: Halt,
524}
525
526impl Job {
527    /// Run one invocation, beating the lease; `None` when the daemon is
528    /// parking and the turn was dropped.
529    async fn drive(
530        &self,
531        spec: &crate::config::AgentSpec,
532        seat: &mut SeatState,
533        inv: &Invocation<'_>,
534        id: &str,
535    ) -> Option<Result<agent::AgentOutput>> {
536        let fut = agent::invoke(spec, seat, inv);
537        tokio::pin!(fut);
538        let mut beat = tokio::time::interval(Duration::from_secs(1));
539        let mut beats = 0u32;
540        loop {
541            tokio::select! {
542                r = &mut fut => break Some(r),
543                _ = beat.tick() => {
544                    if (self.halt)() {
545                        break None;
546                    }
547                    beats += 1;
548                    if beats % 20 == 0 {
549                        self.store.beat(id, WaiterKind::Deputy);
550                    }
551                }
552            }
553        }
554    }
555
556    /// The question settled or expired during the handover: release it without
557    /// starting the long turn. The start stays counted - the handover ran.
558    fn park_quietly(&self, id: &str) {
559        let _ = self.store.update(id, |r| {
560            r.waiter = None;
561            Ok(())
562        });
563        self.store.drop_lease(id);
564    }
565
566    /// Parking: the turn is dropped and the start refunded. The seat stays as
567    /// last persisted - after the handover turn, resumable.
568    fn park(&self, id: &str) {
569        let _ = self.store.update(id, |r| {
570            if let Some(d) = r.deputy.as_mut() {
571                d.starts = d.starts.saturating_sub(1);
572            }
573            r.waiter = None;
574            Ok(())
575        });
576        self.store.drop_lease(id);
577    }
578
579    async fn turn(&self, id: &str) -> Result<()> {
580        let claim = self.store.root().join(format!("{id}.deputy-claim"));
581        // The claim only covers the decision to start and the write that records
582        // it, so a daemon that dies holding it blocks a restart for a minute,
583        // not for a turn's length. The lease guards the turn itself.
584        if !crate::waiter::take_claim(&claim, CLAIM_STALE) {
585            return Ok(());
586        }
587        let release = crate::waiter::Release(claim);
588
589        // Decided again under the claim, on the record as it is now.
590        let q = self.store.get(id)?;
591        let now = Timestamp::now();
592        let Some(dep) = q.deputy.clone() else {
593            return Ok(());
594        };
595        if !q.status.open()
596            || dep.starts >= MAX_STARTS
597            || self.store.read_lease(&q.id).is_some_and(|l| l.fresh(now))
598        {
599            return Ok(());
600        }
601        let cfg = self
602            .cfg
603            .as_ref()
604            .context("the deputy's configuration is not available")?;
605        let spec = match cfg.agent(&dep.agent) {
606            Ok(s) if !dep.agent.is_empty() => s.clone(),
607            _ => {
608                cfg.resolve_roles()
609                    .context("resolving the deputy's agent")?
610                    .conductor
611            }
612        };
613        // The same CLI conversation when it can be resumed; otherwise a fresh
614        // seat - and either way the prompt carries the whole context.
615        let key = crate::ask::deputy_seat_key(&q.id);
616        let (mut seat, resumed) = match dep.seat.clone() {
617            Some(s)
618                if s.agent == spec.id && agent::has_session(spec.kind, &s, cfg.graph.sessions) =>
619            {
620                (s, true)
621            }
622            _ => (SeatState::new(&key, &spec.id, crate::rng::entropy()), false),
623        };
624        // A release-watch question has no `cwd`; its watch record names the
625        // checkout the pull request belongs to.
626        let recorded = q.cwd.clone().or_else(|| match kind_of(&q) {
627            Some(Kind::Release) => {
628                crate::release_watch::state_for_question(&self.home, &q.id).map(|st| st.repo)
629            }
630            // These record a task id in `run` (or a run id for a divergence's
631            // sibling kinds): the task's repository, when it can be found.
632            Some(Kind::Triage | Kind::Generic) => crate::queue::Queue::at(self.home.join("queue"))
633                .get(&q.run)
634                .ok()
635                .map(|t| t.repo.to_string_lossy().into_owned())
636                .filter(|r| !r.is_empty()),
637            _ => None,
638        });
639        let cwd = recorded
640            .as_deref()
641            .map(PathBuf::from)
642            .filter(|p| p.is_dir())
643            .unwrap_or_else(|| self.fallback_repo.clone());
644
645        // A settle's note is part of what the seat said, so a resumed seat
646        // reads its own report back.
647        let bodies: Vec<String> = q
648            .thread
649            .iter()
650            .map(|t| match &t.note {
651                Some(n) => format!("{}\n(note: {n})", t.body),
652                None => t.body.clone(),
653            })
654            .collect();
655        let thread: Vec<(&str, &str)> = q
656            .thread
657            .iter()
658            .zip(&bodies)
659            .map(|(t, body)| {
660                (
661                    if t.who == Who::Operator {
662                        "operator"
663                    } else {
664                        "agent"
665                    },
666                    body.as_str(),
667                )
668            })
669            .collect();
670        let read = &thread[..q.delivered_turns.min(thread.len())];
671        let unread = q.unread_from_owner();
672        let snapshot = q.thread.len();
673        let current_brief = format!(
674            "{}\n\n{}\n\n{}",
675            dep.brief,
676            crate::choice_effect::brief(&q),
677            crate::queue::dependency_facts(
678                &crate::queue::Queue::at(self.home.join("queue")),
679                &self.store
680            )
681        );
682        let body = prompt::deputy(&prompt::DeputyPrompt {
683            id: &q.id,
684            summary: &q.summary,
685            detail: &q.detail,
686            brief: &current_brief,
687            choices: &q.choices,
688            thread: read,
689            unread: unread.as_deref(),
690            resumed,
691            handover: false,
692            kind: kind_of(&q).unwrap_or(Kind::Conduct),
693            language: &cfg.graph.language,
694        });
695
696        // Recorded before the turn starts, so a daemon that dies mid-turn
697        // leaves a start counted and the seat on the record.
698        self.store.beat(&q.id, WaiterKind::Deputy);
699        let starts = dep.starts + 1;
700        let first = seat.clone();
701        self.store.update(&q.id, |r| {
702            if let Some(d) = r.deputy.as_mut() {
703                d.agent = spec.id.clone();
704                d.seat = Some(first);
705                d.starts = starts;
706            }
707            r.waiter = Some(Note {
708                kind: WaiterKind::Deputy,
709                since: now,
710            });
711            Ok(())
712        })?;
713        drop(release);
714        tracing::info!(
715            "question {}: deputy seat {} {} (start {starts}/{MAX_STARTS})",
716            q.short(),
717            seat.key,
718            if resumed { "resuming" } else { "starting" }
719        );
720
721        let left = (deadline(&q, cfg.graph.answer_timeout) - now.as_second()).max(0) as u64;
722        let artifacts = self.store.root().join(format!("{}.deputy", q.id));
723        let cache_dir = cfg.cache_dir();
724        // `magi ask` writes the question record and its lock, which a read-only
725        // sandbox refuses, so the seat cannot be read-only. It is told never to
726        // edit anything; the daemon, not the deputy, applies outcomes.
727        let allow_write = true;
728        // The question store is outside the repository, and `magi ask` writes it.
729        // A merge approval's deputy also files follow-up tasks with `magi task
730        // add`, which writes the queue; that is the one other place it writes.
731        let writable = [self.store.root().to_path_buf(), self.home.join("queue")];
732        macro_rules! invocation {
733            ($prompt:expr, $stem:expr, $timeout:expr) => {
734                Invocation {
735                    scope: crate::agent::InvocationScope::Repository,
736                    coordination: cfg.coordination.as_ref(),
737                    fetched_base: None,
738                    cwd: &cwd,
739                    prompt: $prompt,
740                    timeout: $timeout,
741                    allow_write,
742                    unsandboxed: false,
743                    sessions: cfg.graph.sessions,
744                    artifacts: &artifacts,
745                    stem: $stem,
746                    // The question's own run key (the task id) is what lets this
747                    // seat's `magi ask` pass the ownership check on a conductor
748                    // question.
749                    run: &q.run,
750                    node: NODE,
751                    cache_dir: cache_dir.as_deref(),
752                    attachments: &[],
753                    writable: &writable,
754                }
755            };
756        }
757
758        // A fresh seat first takes a short turn that ends: `agent::invoke` only
759        // learns the CLI's session id when it returns, and the real turn blocks
760        // in `magi ask` for hours, so a daemon stopped mid-wait would otherwise
761        // leave nothing to resume. This turn persists the seat before the wait.
762        let mut early = None;
763        if !resumed && cfg.graph.sessions {
764            let hbody = prompt::deputy(&prompt::DeputyPrompt {
765                id: &q.id,
766                summary: &q.summary,
767                detail: &q.detail,
768                brief: &current_brief,
769                choices: &q.choices,
770                thread: read,
771                unread: None,
772                resumed: false,
773                handover: true,
774                kind: kind_of(&q).unwrap_or(Kind::Conduct),
775                language: &cfg.graph.language,
776            });
777            let hstem = format!("handover-{starts}");
778            // Bounded by the question's own deadline, not only by its own cap: the
779            // lease it beats would otherwise keep an expired question alive.
780            let hlimit = HANDOVER_TIMEOUT.min(Duration::from_secs(left.max(1)));
781            let hinv = invocation!(&hbody, &hstem, hlimit);
782            let Some(done) = self.drive(&spec, &mut seat, &hinv, &q.id).await else {
783                self.park(&q.id);
784                return Ok(());
785            };
786            let kept = seat.clone();
787            self.store.update(&q.id, |r| {
788                if let Some(d) = r.deputy.as_mut() {
789                    d.seat = Some(kept);
790                }
791                Ok(())
792            })?;
793            if !matches!(&done, Ok(o) if o.usable()) {
794                early = Some(done);
795            }
796        }
797
798        // The handover may have been slow: look at the question again, and
799        // measure the long turn from what is left now.
800        let left = if early.is_none() && !resumed && cfg.graph.sessions {
801            let now = Timestamp::now();
802            let again = self.store.get(&q.id)?;
803            let left = (deadline(&again, cfg.graph.answer_timeout) - now.as_second()).max(0) as u64;
804            if !again.status.open() || (left == 0 && again.unread_from_owner().is_none()) {
805                self.park_quietly(&q.id);
806                return Ok(());
807            }
808            left
809        } else {
810            left
811        };
812        let out = match early {
813            Some(done) => done,
814            None => {
815                let stem = format!("turn-{starts}");
816                let timeout = Duration::from_secs(left.max(60) + SLACK_SECS);
817                let inv = invocation!(&body, &stem, timeout);
818                match self.drive(&spec, &mut seat, &inv, &q.id).await {
819                    Some(out) => out,
820                    None => {
821                        self.park(&q.id);
822                        return Ok(());
823                    }
824                }
825            }
826        };
827
828        let (text, why) = match out {
829            Ok(o) if o.usable() => (Some(o.text.trim().to_owned()), None),
830            Ok(o) if o.timed_out => (None, Some("its turn timed out".to_owned())),
831            Ok(o) if o.quota_exhausted() => (None, Some("the agent is out of quota".to_owned())),
832            Ok(o) => (
833                None,
834                Some(format!(
835                    "its turn failed ({})",
836                    notices::exit_status(o.exit_code)
837                )),
838            ),
839            Err(e) => (None, Some(format!("the agent could not be started: {e:#}"))),
840        };
841        let kept = seat.clone();
842        self.store.update(&q.id, |r| {
843            if let Some(d) = r.deputy.as_mut() {
844                d.seat = Some(kept);
845            }
846            r.waiter = None;
847            // Retire ownership under the same lock used by text admission.
848            self.store.drop_lease(&q.id);
849            // The owner's words were in the prompt. An agent that read them but
850            // answered in prose instead of `magi ask --thread` still spoke;
851            // keep it on the record so the owner reads it.
852            if let Some(text) = text
853                && unread.is_some()
854                && !text.is_empty()
855                && r.status.open()
856                && r.thread.len() == snapshot
857            {
858                r.delivered_turns = r.delivered_turns.max(snapshot);
859                let choices = r.choices.clone();
860                r.reply(text, choices)?;
861            }
862            Ok(())
863        })?;
864        if let Some(why) = why {
865            tracing::warn!("question {}: the deputy ended: {why}", q.short());
866            notices::raise_in(
867                &self.home,
868                Notice::warn(
869                    &format!("deputy-turn:{}", q.id),
870                    format!(
871                        "Question {} \"{}\": its follow-up agent stopped ({why}).",
872                        q.short(),
873                        q.summary
874                    ),
875                ),
876            );
877        }
878        Ok(())
879    }
880}
881
882/// The runner's loop, until `stop` is asked for.
883pub async fn run(mut deputies: Deputies, stop: crate::daemon::Stop) {
884    while !stop.stopped() {
885        deputies.tick(Timestamp::now());
886        tokio::time::sleep(TICK).await;
887    }
888}