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