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