1use std::collections::HashMap;
45use std::path::PathBuf;
46use std::time::{Duration, Instant};
47
48use anyhow::Result;
49use jiff::Timestamp;
50
51use crate::agent::{self, Invocation, SeatState};
52use crate::ask::{Lease, Question, QuestionStatus, Questions, Waiter as Note, WaiterKind, Who};
53use crate::config::{AgentSpec, Config};
54use crate::notices::{self, Notice};
55use crate::prompt;
56use crate::run::RunState;
57
58pub const TICK: Duration = Duration::from_secs(5);
60
61const DELIVERY_TIMEOUT: Duration = Duration::from_secs(15 * 60);
65
66const RETRY_AFTER: Duration = Duration::from_secs(10 * 60);
70
71#[derive(Debug, Clone, PartialEq, Eq)]
73pub enum Word {
74 Said(String),
76 Answered(String),
78}
79
80#[derive(Debug, Clone, PartialEq, Eq)]
82pub enum Action {
83 Idle,
85 Expire,
87 Deliver(Word),
89}
90
91pub fn decide(
101 q: &Question,
102 lease: Option<&Lease>,
103 seat_busy: bool,
104 default_timeout: u64,
105 now: Timestamp,
106) -> Action {
107 decide_owned(q, lease, seat_busy, default_timeout, now, false)
108}
109
110pub fn decide_owned(
120 q: &Question,
121 lease: Option<&Lease>,
122 seat_busy: bool,
123 default_timeout: u64,
124 now: Timestamp,
125 daemon_owns_action: bool,
126) -> Action {
127 if q.cwd.is_none() || lease.is_some_and(|l| l.fresh(now)) {
128 return Action::Idle;
129 }
130 match q.status {
131 QuestionStatus::Abandoned => Action::Idle,
132 QuestionStatus::Answered => match q.resolution() {
133 _ if daemon_owns_action && q.chosen_action().is_some() => Action::Idle,
134 Some(a) if !q.answer_delivered && !seat_busy => Action::Deliver(Word::Answered(a)),
135 _ => Action::Idle,
136 },
137 QuestionStatus::Open => {
138 if let Some(said) = q.unread_from_owner() {
139 if seat_busy {
140 return Action::Idle;
141 }
142 return Action::Deliver(Word::Said(said));
143 }
144 let secs = if q.answer_timeout > 0 {
145 q.answer_timeout
146 } else {
147 default_timeout
148 };
149 let deadline = q.last_activity().saturating_add(secs as i64);
150 if now.as_second() > deadline {
151 Action::Expire
152 } else {
153 Action::Idle
154 }
155 }
156 }
157}
158
159struct Target {
161 spec: AgentSpec,
162 seat: SeatState,
163 cwd: PathBuf,
164 sessions: bool,
165 allow_write: bool,
166 language: String,
167}
168
169pub struct Waiter {
172 store: Questions,
173 home: PathBuf,
174 conduct_cfg: Option<Config>,
177 default_timeout: u64,
179 memo: HashMap<String, (String, Instant)>,
181}
182
183impl Waiter {
184 pub fn new(store: Questions, home: PathBuf, conduct_cfg: Option<Config>) -> Self {
186 let default_timeout = conduct_cfg
187 .as_ref()
188 .map_or(Config::default().graph.answer_timeout, |c| {
189 c.graph.answer_timeout
190 });
191 Self {
192 store,
193 home,
194 conduct_cfg,
195 default_timeout,
196 memo: HashMap::new(),
197 }
198 }
199
200 fn seat_busy(&self, q: &Question, now: Timestamp) -> bool {
202 if q.run == crate::conduct::NODE {
203 return crate::conduct::busy(&self.home);
204 }
205 RunState::load_under(&q.run, &self.home).is_ok_and(|s| {
206 s.seats_active()
207 .any(|(k, a)| *k == q.seat && a.remaining_secs(now) > 0)
208 })
209 }
210
211 fn daemon_owns_action(&self, q: &Question) -> bool {
214 crate::daemon::task_of_question(&crate::queue::Queue::at(self.home.join("queue")).list(), q)
215 .is_some_and(|t| crate::daemon::action_standing(t, q).daemon_owns())
216 }
217
218 pub async fn tick(&mut self, now: Timestamp, halt: &(dyn Fn() -> bool + Sync)) {
223 for q in self.store.list() {
224 if halt() {
225 return;
226 }
227 let lease = self.store.read_lease(&q.id);
228 let owned = self.daemon_owns_action(&q);
229 match decide_owned(&q, lease.as_ref(), false, self.default_timeout, now, owned) {
230 Action::Idle => {}
231 Action::Expire => self.expire(&q),
232 Action::Deliver(word) => {
233 if q.deputy.is_some() {
237 if crate::deputy::exhausted_past_deadline(
241 &q,
242 crate::deputy::can_start(
243 self.conduct_cfg.as_ref(),
244 crate::deputy::agent_of(&q),
245 ),
246 self.default_timeout,
247 now,
248 ) {
249 self.expire(&q);
250 }
251 continue;
252 }
253 if self.seat_busy(&q, now) {
254 continue;
255 }
256 let key = format!("{}:{:?}", q.thread.len(), word);
257 if matches!(self.memo.get(&q.id), Some((k, until)) if *k == key && Instant::now() < *until)
258 {
259 continue;
260 }
261 if let Err(e) = self.deliver(&q, &word, halt).await {
262 tracing::warn!("question {}: {e:#}", q.short());
263 }
264 }
265 }
266 }
267 }
268
269 fn expire(&self, q: &Question) {
272 let secs = if q.answer_timeout > 0 {
273 q.answer_timeout
274 } else {
275 self.default_timeout
276 };
277 let startable =
278 crate::deputy::can_start(self.conduct_cfg.as_ref(), crate::deputy::agent_of(q));
279 let why = format!("no answer within {}s of asking", secs.max(1));
280 let done = self.store.update(&q.id, |r| {
281 let now = Timestamp::now();
285 let lease = self.store.read_lease(&r.id);
286 if decide(r, lease.as_ref(), false, self.default_timeout, now) != Action::Expire
287 && !(lease.as_ref().is_none_or(|l| !l.fresh(now))
288 && crate::deputy::exhausted_past_deadline(
289 r,
290 startable,
291 self.default_timeout,
292 now,
293 ))
294 {
295 return Ok(false);
296 }
297 r.abandon(&why);
298 r.waiter = None;
299 Ok(true)
300 });
301 match done {
302 Ok((_, false)) => {}
303 Ok((_, true)) => {
304 self.store.drop_lease(&q.id);
305 tracing::warn!(
306 "question {} went unanswered for {secs}s with nobody waiting; \
307 it stays as the record of it",
308 q.short()
309 );
310 }
311 Err(e) => tracing::warn!("could not expire question {}: {e:#}", q.short()),
312 }
313 }
314
315 pub fn can_resume(&self, q: &Question) -> bool {
317 self.target(q).is_ok()
318 }
319
320 fn target(&self, q: &Question) -> std::result::Result<Target, String> {
322 let cwd = q
323 .cwd
324 .as_deref()
325 .map(PathBuf::from)
326 .filter(|p| p.is_dir())
327 .ok_or("the directory the agent was working in is gone")?;
328 let (cfg, seat, allow_write) = if q.run == crate::conduct::NODE {
329 let cfg = self
330 .conduct_cfg
331 .clone()
332 .ok_or("the conductor's configuration is not available")?;
333 let seat = crate::conduct::load_seat(&self.home)
334 .ok_or("the conductor has no recorded session")?;
335 (cfg, seat, false)
336 } else {
337 let state = RunState::load_under(&q.run, &self.home)
338 .map_err(|_| "the run that asked is gone".to_owned())?;
339 if !state.status.resumable() {
340 return Err(format!(
341 "the run that asked has already {}",
342 state.status.as_str()
343 ));
344 }
345 let seat = state
346 .seats
347 .get(&q.seat)
348 .cloned()
349 .ok_or("the run has no record of the asking seat")?;
350 let write = matches!(q.node.as_str(), "implement" | "fix");
351 (state.config, seat, write)
352 };
353 let spec = cfg
354 .agent(&seat.agent)
355 .map_err(|_| format!("agent `{}` is no longer in the roster", seat.agent))?
356 .clone();
357 if !agent::has_session(spec.kind, &seat, cfg.graph.sessions) {
358 return Err("the seat has no session to resume".to_owned());
359 }
360 Ok(Target {
361 spec,
362 seat,
363 cwd,
364 sessions: cfg.graph.sessions,
365 allow_write,
366 language: cfg.graph.language.clone(),
367 })
368 }
369
370 fn refuse(&mut self, q: &Question, key: String, why: &str) {
372 tracing::warn!("question {} cannot be delivered: {why}", q.short());
373 notices::raise_in(
374 &self.home,
375 Notice::warn(
376 &format!("question:{}", q.id),
377 format!(
378 "Question {} \"{}\": what you said cannot reach the agent \
379 ({why}). It is recorded, but nothing will read it.",
380 q.short(),
381 q.summary
382 ),
383 ),
384 );
385 let _ = self.store.update(&q.id, |r| {
386 r.waiter = None;
387 Ok(())
388 });
389 self.memo
390 .insert(q.id.clone(), (key, Instant::now() + RETRY_AFTER));
391 }
392
393 async fn deliver(
395 &mut self,
396 q: &Question,
397 word: &Word,
398 halt: &(dyn Fn() -> bool + Sync),
399 ) -> Result<()> {
400 let key = format!("{}:{:?}", q.thread.len(), word);
401 let mut target = match self.target(q) {
402 Ok(t) => t,
403 Err(why) => {
404 self.refuse(q, key, &why);
405 return Ok(());
406 }
407 };
408
409 let claim = self.store.root().join(format!("{}.claim", q.id));
410 if !take_claim(&claim, DELIVERY_TIMEOUT + Duration::from_secs(60)) {
411 return Ok(());
412 }
413 let _release = Release(claim);
414
415 let q = self.store.get(&q.id)?;
418 let now = Timestamp::now();
419 let lease = self.store.read_lease(&q.id);
420 let owned = self.daemon_owns_action(&q);
421 let Action::Deliver(word) =
422 decide_owned(&q, lease.as_ref(), false, self.default_timeout, now, owned)
423 else {
424 return Ok(());
425 };
426 let snapshot = q.thread.len();
427
428 let queue = crate::queue::Queue::at(self.home.join("queue"));
434 let task_id = crate::daemon::task_of_question(&queue.list(), &q).map(|t| t.id.clone());
435 let _task_claim = match (&task_id, q.chosen_action()) {
439 (Some(id), Some(_)) => match queue.claim(id) {
440 Ok(c) => Some(c),
441 Err(e) => {
442 tracing::debug!(
443 "question {}: task claim not available ({e:#}), deferring delivery",
444 q.short()
445 );
446 return Ok(());
447 }
448 },
449 _ => None,
450 };
451 let task_still_ours = || match &task_id {
454 Some(id) => queue
455 .get(id)
456 .map(|t| !crate::daemon::action_standing(&t, &q).daemon_owns())
457 .unwrap_or(true),
458 None => true,
459 };
460 if !task_still_ours() {
461 return Ok(());
462 }
463
464 self.store.beat(&q.id, WaiterKind::Daemon);
465 if !task_still_ours() {
468 return Ok(());
469 }
470 self.store.update(&q.id, |r| {
471 r.waiter = Some(Note {
472 kind: WaiterKind::Daemon,
473 since: now,
474 });
475 Ok(())
476 })?;
477 tracing::info!(
478 "question {}: the asker is gone, resuming seat {} with the owner's word",
479 q.short(),
480 q.seat
481 );
482
483 let thread: Vec<(&str, &str)> = q
484 .thread
485 .iter()
486 .map(|t| {
487 (
488 if t.who == Who::Operator {
489 "operator"
490 } else {
491 "agent"
492 },
493 t.body.as_str(),
494 )
495 })
496 .collect();
497 let shown = match word {
500 Word::Said(_) => &thread[..q.delivered_turns.min(thread.len())],
501 Word::Answered(_) => &thread[..],
502 };
503 let owner_word = match &word {
504 Word::Said(s) => prompt::OwnerWord::Said(s),
505 Word::Answered(a) => prompt::OwnerWord::Answered(a),
506 };
507 let body = prompt::question_resumed(
508 &q.id,
509 &q.summary,
510 &q.detail,
511 shown,
512 &owner_word,
513 &target.language,
514 );
515
516 let artifacts = self.store.root().join(format!("{}.delivery", q.id));
517 let stem = format!("deliver-{}", now.as_second());
518 let cache_dir = None;
519 let inv = Invocation {
520 cwd: &target.cwd,
521 prompt: &body,
522 timeout: DELIVERY_TIMEOUT,
523 allow_write: target.allow_write,
524 unsandboxed: false,
525 sessions: target.sessions,
526 artifacts: &artifacts,
527 stem: &stem,
528 run: &q.run,
529 node: &q.node,
530 cache_dir,
531 attachments: &[],
532 writable: &[],
533 };
534
535 let store = self.store.clone();
536 let id = q.id.clone();
537 let out = {
538 let fut = agent::invoke(&target.spec, &mut target.seat, &inv);
539 tokio::pin!(fut);
540 let mut beat = tokio::time::interval(Duration::from_secs(1));
541 let mut beats = 0u32;
542 loop {
543 tokio::select! {
544 r = &mut fut => break Some(r),
545 _ = beat.tick() => {
546 if halt() {
547 break None;
548 }
549 beats += 1;
550 if beats % 20 == 0 {
551 store.beat(&id, WaiterKind::Daemon);
552 }
553 }
554 }
555 }
556 };
557
558 let Some(out) = out else {
559 let _ = self.store.update(&q.id, |r| {
561 r.waiter = None;
562 Ok(())
563 });
564 return Ok(());
565 };
566 let out = match out {
567 Ok(o) if o.usable() => o,
568 other => {
569 let why = match other {
570 Ok(o) if o.timed_out => "the resumed turn timed out".to_owned(),
571 Ok(o) if o.quota_exhausted() => "the agent is out of quota".to_owned(),
572 Ok(o) => format!("the resumed turn failed (exit {:?})", o.exit_code),
573 Err(e) => format!("the agent could not be started: {e:#}"),
574 };
575 self.refuse(&q, key, &why);
576 return Ok(());
577 }
578 };
579
580 let text = out.text.trim().to_owned();
581 self.store.update(&q.id, |r| {
582 r.waiter = None;
583 match &word {
584 Word::Answered(_) => r.answer_delivered = true,
585 Word::Said(_) => {
586 r.delivered_turns = r.delivered_turns.max(snapshot);
587 let agent_spoke = r.thread[snapshot.min(r.thread.len())..]
592 .iter()
593 .any(|t| t.who == Who::Agent);
594 if !agent_spoke && r.thread.len() == snapshot && r.status.open() {
595 let choices = r.choices.clone();
596 r.reply(text, choices)?;
597 }
598 }
599 }
600 Ok(())
601 })?;
602 self.memo.remove(&q.id);
603 Ok(())
604 }
605}
606
607pub(crate) fn take_claim(path: &std::path::Path, stale_after: Duration) -> bool {
611 let attempt = || {
612 std::fs::OpenOptions::new()
613 .write(true)
614 .create_new(true)
615 .open(path)
616 .is_ok()
617 };
618 if attempt() {
619 return true;
620 }
621 let stale = std::fs::metadata(path)
622 .and_then(|m| m.modified())
623 .ok()
624 .and_then(|t| t.elapsed().ok())
625 .is_some_and(|age| age > stale_after);
626 if stale {
627 let _ = std::fs::remove_file(path);
628 return attempt();
629 }
630 false
631}
632
633pub(crate) struct Release(pub(crate) PathBuf);
635
636impl Drop for Release {
637 fn drop(&mut self) {
638 let _ = std::fs::remove_file(&self.0);
639 }
640}
641
642pub async fn run(mut waiter: Waiter, stop: crate::daemon::Stop) {
648 let halt = {
649 let stop = stop.clone();
650 move || stop.parking()
651 };
652 while !stop.stopped() {
653 waiter.tick(Timestamp::now(), &halt).await;
654 tokio::time::sleep(TICK).await;
655 }
656}
657
658#[cfg(test)]
659mod tests {
660 use super::*;
661
662 fn ts(secs: i64) -> Timestamp {
663 Timestamp::from_second(secs).unwrap()
664 }
665
666 fn asked(at: i64) -> Question {
667 let mut q = Question::new(
668 "run".into(),
669 "implement".into(),
670 "impl-A".into(),
671 "which?".into(),
672 String::new(),
673 vec![],
674 );
675 q.cwd = Some("/tmp".into());
676 q.asked_at = ts(at);
677 q.answer_timeout = 1000;
678 q
679 }
680
681 fn lease(at: i64) -> Lease {
682 Lease {
683 kind: WaiterKind::Asker,
684 pid: 1,
685 beat_at: ts(at),
686 }
687 }
688
689 #[test]
690 fn a_fresh_lease_means_nobody_else_acts() {
691 let mut q = asked(0);
692 q.say("why?").unwrap();
693 assert_eq!(
694 decide(&q, Some(&lease(95)), false, 86_400, ts(100)),
695 Action::Idle
696 );
697 }
698
699 #[test]
700 fn a_stale_lease_delivers_what_the_owner_said() {
701 let mut q = asked(0);
702 q.say("why?").unwrap();
703 assert_eq!(
704 decide(&q, Some(&lease(0)), false, 86_400, ts(500)),
705 Action::Deliver(Word::Said("why?".into()))
706 );
707 assert_eq!(
708 decide(&q, None, true, 86_400, ts(500)),
709 Action::Idle,
710 "a seat still mid-turn is not resumed"
711 );
712 }
713
714 #[test]
715 fn an_answer_nobody_read_is_delivered_once() {
716 let mut q = asked(0);
717 q.choices = vec!["A".into()];
718 q.answer(crate::ask::Answer::Choice("A".into())).unwrap();
719 assert_eq!(
720 decide(&q, None, false, 86_400, ts(10)),
721 Action::Deliver(Word::Answered("A".into()))
722 );
723 q.answer_delivered = true;
724 assert_eq!(decide(&q, None, false, 86_400, ts(10)), Action::Idle);
725 }
726
727 #[test]
728 fn only_asked_questions_and_only_before_the_deadline() {
729 let mut q = asked(0);
730 assert_eq!(decide(&q, None, false, 86_400, ts(999)), Action::Idle);
731 assert_eq!(decide(&q, None, false, 86_400, ts(1001)), Action::Expire);
732 q.cwd = None;
733 assert_eq!(decide(&q, None, false, 86_400, ts(1001)), Action::Idle);
734 }
735
736 #[test]
737 fn the_deadline_runs_from_the_last_turn_not_from_asking() {
738 let mut q = asked(0);
739 q.thread.push(crate::ask::Turn {
740 who: Who::Agent,
741 body: "context".into(),
742 at: ts(4000),
743 note: None,
744 });
745 q.delivered_turns = 1;
746 assert_eq!(decide(&q, None, false, 86_400, ts(4500)), Action::Idle);
747 assert_eq!(decide(&q, None, false, 86_400, ts(5001)), Action::Expire);
748 }
749
750 #[test]
751 fn a_word_given_in_time_is_delivered_after_the_deadline() {
752 let mut q = asked(0);
753 q.say("wait").unwrap();
754 assert!(matches!(
755 decide(&q, None, false, 86_400, ts(5000)),
756 Action::Deliver(_)
757 ));
758 }
759
760 #[test]
761 fn an_action_answer_is_left_to_the_daemon() {
762 let mut q = asked(0);
763 q.choices = vec!["A".into()];
764 q.actions
765 .insert("A".into(), crate::ask::ChoiceAction::Requeue);
766 q.answer(crate::ask::Answer::Choice("A".into())).unwrap();
767 assert_eq!(
768 decide_owned(&q, None, false, 86_400, ts(10), true),
769 Action::Idle,
770 "the daemon applies the action; the waiter must not also resume the seat"
771 );
772 q.answer_delivered = true;
774 assert_eq!(
775 decide_owned(&q, None, false, 86_400, ts(10), true),
776 Action::Idle
777 );
778 q.answer_delivered = false;
780 assert_eq!(
781 decide_owned(&q, None, false, 86_400, ts(10), false),
782 Action::Deliver(Word::Answered("A".into()))
783 );
784 let mut plain = asked(0);
786 plain.choices = vec!["A".into()];
787 plain
788 .answer(crate::ask::Answer::Choice("A".into()))
789 .unwrap();
790 assert_eq!(
791 decide_owned(&plain, None, false, 86_400, ts(10), true),
792 Action::Deliver(Word::Answered("A".into()))
793 );
794 }
795}