1use anyhow::bail;
9use leviath_core::interaction::{
10 ApprovalScope, InteractionKind, InteractionRequest, InteractionResponse,
11};
12use leviath_runtime::control_socket::{ControlClient, ControlRequest, ControlResponse};
13
14#[derive(clap::Args, Debug, Clone)]
16pub struct MsgArgs {
17 pub agent_id: String,
19 pub content: String,
21}
22
23#[derive(clap::Args, Debug, Clone)]
25pub struct CancelArgs {
26 pub run_id: String,
28 #[arg(long)]
34 pub force: bool,
35}
36
37#[derive(clap::Args, Debug, Clone)]
39pub struct PauseArgs {
40 pub run_id: String,
42}
43
44#[derive(clap::Args, Debug, Clone)]
46pub struct ResumeArgs {
47 pub run_id: String,
49}
50
51#[derive(clap::Args, Debug, Clone)]
54pub struct RespondArgs {
55 pub request_id: Option<String>,
57 pub value: Option<String>,
59 #[arg(long)]
61 pub choice: Option<usize>,
62 #[arg(long, conflicts_with = "deny")]
64 pub approve: bool,
65 #[arg(long)]
67 pub deny: bool,
68 #[arg(long)]
70 pub session: bool,
71}
72
73async fn send_bool(
77 client: &ControlClient,
78 request: ControlRequest,
79 applied_msg: &str,
80 not_found_msg: &str,
81) -> anyhow::Result<()> {
82 match client.request(&request).await {
83 Ok(ControlResponse::Ok { ok: true }) => {
84 println!("{applied_msg}");
85 Ok(())
86 }
87 Ok(ControlResponse::Ok { ok: false }) => bail!("{not_found_msg}"),
88 Ok(other) => bail!("unexpected daemon response: {other:?}"),
89 Err(e) => bail!("the leviath daemon is not reachable ({e}); start it with `lev daemon`"),
90 }
91}
92
93pub async fn send_message(client: &ControlClient, args: &MsgArgs) -> anyhow::Result<()> {
95 send_bool(
96 client,
97 ControlRequest::Message {
98 agent_id: args.agent_id.clone(),
99 content: args.content.clone(),
100 target_region: None,
101 },
102 "message delivered",
103 "no agent accepted the message",
104 )
105 .await
106}
107
108pub async fn pause_run(client: &ControlClient, args: &PauseArgs) -> anyhow::Result<()> {
111 send_bool(
112 client,
113 ControlRequest::Pause {
114 run_id: args.run_id.clone(),
115 },
116 "paused",
117 "no such run, or it is not pausable in its current state",
118 )
119 .await
120}
121
122pub async fn resume_run(client: &ControlClient, args: &ResumeArgs) -> anyhow::Result<()> {
124 send_bool(
125 client,
126 ControlRequest::Resume {
127 run_id: args.run_id.clone(),
128 },
129 "resumed",
130 "no such run, or it is not paused",
131 )
132 .await
133}
134
135pub async fn cancel_run(client: &ControlClient, args: &CancelArgs) -> anyhow::Result<()> {
143 if args.force {
144 return report_forced(
145 crate::runstate::force_cancel(&args.run_id),
146 &args.run_id,
147 None,
148 );
149 }
150 match client
151 .request(&ControlRequest::Cancel {
152 run_id: args.run_id.clone(),
153 })
154 .await
155 {
156 Ok(ControlResponse::Ok { ok: true }) => {
157 println!("cancelled");
158 Ok(())
159 }
160 Ok(ControlResponse::Ok { ok: false }) => bail!("no such run"),
161 Ok(other) => bail!("unexpected daemon response: {other:?}"),
162 Err(e) => report_forced(
165 crate::runstate::force_cancel(&args.run_id),
166 &args.run_id,
167 Some(e),
168 ),
169 }
170}
171
172fn report_forced(
176 outcome: crate::runstate::ForceCancelOutcome,
177 run_id: &str,
178 daemon_error: Option<std::io::Error>,
179) -> anyhow::Result<()> {
180 use crate::runstate::ForceCancelOutcome as O;
181 let why = match &daemon_error {
182 Some(e) => format!(" (the daemon did not answer: {e})"),
183 None => String::new(),
184 };
185 match outcome {
186 O::Terminated => {
187 println!(
188 "cancelled '{run_id}' on disk{why}; if a daemon is still running, \
189 restart it so it picks up the change"
190 );
191 Ok(())
192 }
193 O::AlreadyTerminal => {
194 println!("'{run_id}' had already finished; nothing to cancel");
195 Ok(())
196 }
197 O::NoSuchRun => match daemon_error {
198 Some(e) => bail!(
199 "the leviath daemon is not reachable ({e}), and there is no run '{run_id}' on disk"
200 ),
201 None => bail!("no such run"),
202 },
203 O::WriteFailed => bail!("could not write '{run_id}' metadata to record the cancel"),
204 }
205}
206
207fn kind_label(kind: &InteractionKind) -> &'static str {
209 match kind {
210 InteractionKind::FreeText => "free-text",
211 InteractionKind::MultipleChoice => "choice",
212 InteractionKind::Confirm => "confirm",
213 InteractionKind::ToolApproval => "tool-approval",
214 InteractionKind::EditText => "edit-text",
215 }
216}
217
218fn format_interaction(agent_id: &str, req: &InteractionRequest) -> String {
220 let mut s = format!(
221 "{} [{}] agent={} stage={}\n {}",
222 req.id,
223 kind_label(&req.kind),
224 agent_id,
225 req.stage_name,
226 req.prompt
227 );
228 for (i, opt) in req.options.iter().enumerate() {
229 s.push_str(&format!("\n {i}) {opt}"));
230 }
231 if let Some(tool) = &req.tool_name {
232 s.push_str(&format!("\n tool: {tool}"));
233 }
234 s
235}
236
237fn build_response(request_id: &str, args: &RespondArgs) -> InteractionResponse {
240 if args.approve || args.deny {
241 let scope = if args.session {
242 ApprovalScope::Session
243 } else {
244 ApprovalScope::Once
245 };
246 InteractionResponse::approval(request_id, args.approve, scope)
247 } else if let Some(index) = args.choice {
248 InteractionResponse::choice(request_id, index)
249 } else {
250 InteractionResponse::text(request_id, args.value.clone().unwrap_or_default())
251 }
252}
253
254async fn list_interactions(client: &ControlClient) -> anyhow::Result<()> {
256 match client.request(&ControlRequest::ListInteractions).await {
257 Ok(ControlResponse::Interactions { interactions }) => {
258 if interactions.is_empty() {
259 println!("no open interactions");
260 } else {
261 for (agent_id, req) in &interactions {
262 println!("{}", format_interaction(agent_id, req));
263 }
264 }
265 Ok(())
266 }
267 Ok(other) => bail!("unexpected daemon response: {other:?}"),
268 Err(e) => bail!("the leviath daemon is not reachable ({e}); start it with `lev daemon`"),
269 }
270}
271
272pub async fn respond(client: &ControlClient, args: &RespondArgs) -> anyhow::Result<()> {
275 match &args.request_id {
276 None => list_interactions(client).await,
277 Some(request_id) => {
278 send_bool(
279 client,
280 ControlRequest::AnswerInteraction {
281 response: build_response(request_id, args),
282 },
283 "answered",
284 "no such open interaction",
285 )
286 .await
287 }
288 }
289}
290
291#[cfg(test)]
292mod tests {
293 use super::*;
294 use leviath_runtime::control_socket::{ControlId, bind_control_listener, control_id};
295 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
296 use tokio::task::JoinHandle;
297
298 fn fake_daemon(dir: &std::path::Path, response_line: String) -> (ControlId, JoinHandle<()>) {
301 let id = control_id(dir);
302 let mut listener = bind_control_listener(&id).unwrap();
303 let handle = tokio::spawn(async move {
304 let stream = listener
305 .accept()
306 .await
307 .expect("accept succeeds")
308 .expect("our own connection is admitted");
309 let (read_half, mut write_half) = tokio::io::split(stream);
310 let mut lines = BufReader::new(read_half).lines();
311 let _request = lines.next_line().await.unwrap();
312 write_half
313 .write_all(response_line.as_bytes())
314 .await
315 .unwrap();
316 write_half.write_all(b"\n").await.unwrap();
317 });
318 (id, handle)
319 }
320
321 async fn with_daemon<F, Fut>(response_line: impl Into<String>, op: F) -> anyhow::Result<()>
323 where
324 F: FnOnce(ControlClient) -> Fut,
325 Fut: std::future::Future<Output = anyhow::Result<()>>,
326 {
327 let dir = tempfile::tempdir().unwrap();
328 let (id, server) = fake_daemon(dir.path(), response_line.into());
329 let result = op(ControlClient::new(id)).await;
330 server.await.unwrap();
331 result
332 }
333
334 fn msg_args() -> MsgArgs {
335 MsgArgs {
336 agent_id: "a".to_string(),
337 content: "hi".to_string(),
338 }
339 }
340
341 #[tokio::test]
342 async fn message_applied() {
343 let r = with_daemon(r#"{"result":"ok","ok":true}"#, |c| async move {
344 send_message(&c, &msg_args()).await
345 })
346 .await;
347 assert!(r.is_ok());
348 }
349
350 #[tokio::test]
351 async fn message_not_delivered() {
352 let r = with_daemon(r#"{"result":"ok","ok":false}"#, |c| async move {
353 send_message(&c, &msg_args()).await
354 })
355 .await;
356 assert!(r.unwrap_err().to_string().contains("no agent accepted"));
357 }
358
359 #[tokio::test]
360 async fn pause_applied() {
361 let r = with_daemon(r#"{"result":"ok","ok":true}"#, |c| async move {
362 pause_run(
363 &c,
364 &PauseArgs {
365 run_id: "r".to_string(),
366 },
367 )
368 .await
369 })
370 .await;
371 assert!(r.is_ok());
372 }
373
374 #[tokio::test]
375 async fn pause_refused() {
376 let r = with_daemon(r#"{"result":"ok","ok":false}"#, |c| async move {
377 pause_run(
378 &c,
379 &PauseArgs {
380 run_id: "r".to_string(),
381 },
382 )
383 .await
384 })
385 .await;
386 assert!(r.unwrap_err().to_string().contains("not pausable"));
387 }
388
389 #[tokio::test]
390 async fn resume_applied() {
391 let r = with_daemon(r#"{"result":"ok","ok":true}"#, |c| async move {
392 resume_run(
393 &c,
394 &ResumeArgs {
395 run_id: "r".to_string(),
396 },
397 )
398 .await
399 })
400 .await;
401 assert!(r.is_ok());
402 }
403
404 #[tokio::test]
405 async fn resume_refused() {
406 let r = with_daemon(r#"{"result":"ok","ok":false}"#, |c| async move {
407 resume_run(
408 &c,
409 &ResumeArgs {
410 run_id: "r".to_string(),
411 },
412 )
413 .await
414 })
415 .await;
416 assert!(r.unwrap_err().to_string().contains("not paused"));
417 }
418
419 #[tokio::test]
420 async fn cancel_applied() {
421 let r = with_daemon(r#"{"result":"ok","ok":true}"#, |c| async move {
422 cancel_run(
423 &c,
424 &CancelArgs {
425 run_id: "r".to_string(),
426 force: false,
427 },
428 )
429 .await
430 })
431 .await;
432 assert!(r.is_ok());
433 }
434
435 #[tokio::test]
436 async fn cancel_unknown_run() {
437 let r = with_daemon(r#"{"result":"ok","ok":false}"#, |c| async move {
438 cancel_run(
439 &c,
440 &CancelArgs {
441 run_id: "r".to_string(),
442 force: false,
443 },
444 )
445 .await
446 })
447 .await;
448 assert!(r.unwrap_err().to_string().contains("no such run"));
449 }
450
451 #[tokio::test]
452 async fn unexpected_response_is_an_error() {
453 let r = with_daemon(r#"{"result":"spawned","run_id":"x"}"#, |c| async move {
456 send_message(&c, &msg_args()).await
457 })
458 .await;
459 assert!(r.unwrap_err().to_string().contains("unexpected"));
460
461 let r = with_daemon(r#"{"result":"spawned","run_id":"x"}"#, |c| async move {
462 cancel_run(
463 &c,
464 &CancelArgs {
465 run_id: "r".to_string(),
466 force: false,
467 },
468 )
469 .await
470 })
471 .await;
472 assert!(r.unwrap_err().to_string().contains("unexpected"));
473 }
474
475 #[tokio::test]
478 async fn message_to_an_unreachable_daemon_is_an_error() {
479 let dir = tempfile::tempdir().unwrap();
480 let client = ControlClient::new(control_id(&dir.path().join("no-daemon")));
481 let err = send_message(&client, &msg_args()).await.unwrap_err();
482 assert!(err.to_string().contains("not reachable"));
483 }
484
485 #[tokio::test]
488 async fn forcing_a_run_whose_metadata_cannot_be_written_reports_the_failure() {
489 crate::runstate::with_isolated_runs_dir_async("ctl-force-unwritable", |_base| async {
490 let dir = crate::runstate::run_dir("blocked-1");
491 std::fs::create_dir_all(dir.join("meta.json")).unwrap();
492
493 let err = cancel_run(
494 &ControlClient::new(control_id(std::path::Path::new("/nonexistent"))),
495 &CancelArgs {
496 run_id: "blocked-1".to_string(),
497 force: true,
498 },
499 )
500 .await
501 .unwrap_err();
502 assert!(err.to_string().contains("could not write"), "got: {err}");
503 })
504 .await;
505 }
506
507 #[tokio::test]
508 async fn not_reachable_is_an_error() {
509 let dir = tempfile::tempdir().unwrap();
510 let client = ControlClient::new(control_id(&dir.path().join("no-daemon")));
511 let err = cancel_run(
512 &client,
513 &CancelArgs {
514 run_id: "r".to_string(),
515 force: false,
516 },
517 )
518 .await
519 .unwrap_err();
520 assert!(err.to_string().contains("not reachable"));
521 }
522
523 fn seed_live_run(run_id: &str) {
525 crate::runstate::create_run(&crate::runstate::RunMeta {
526 status: crate::runstate::RunStatus::Running,
527 ..crate::runstate::RunMeta::new(
528 run_id.into(),
529 "a".into(),
530 "/p".into(),
531 "t".into(),
532 None,
533 "/w".into(),
534 1,
535 )
536 })
537 .unwrap();
538 }
539
540 fn status_of(run_id: &str) -> crate::runstate::RunStatus {
541 crate::runstate::read_meta(run_id).unwrap().status
542 }
543
544 #[tokio::test]
547 async fn force_cancels_on_disk_without_a_daemon() {
548 crate::runstate::with_isolated_runs_dir_async("ctl-force-cancel", |_base| async {
549 seed_live_run("stuck-1");
550 let dir = tempfile::tempdir().unwrap();
551 let client = ControlClient::new(control_id(&dir.path().join("no-daemon")));
553
554 cancel_run(
555 &client,
556 &CancelArgs {
557 run_id: "stuck-1".to_string(),
558 force: true,
559 },
560 )
561 .await
562 .expect("forced cancel succeeds with no daemon");
563
564 assert_eq!(status_of("stuck-1"), crate::runstate::RunStatus::Cancelled);
565 })
566 .await;
567 }
568
569 #[tokio::test]
572 async fn an_unreachable_daemon_falls_back_to_cancelling_on_disk() {
573 crate::runstate::with_isolated_runs_dir_async("ctl-fallback-cancel", |_base| async {
574 seed_live_run("stuck-2");
575 let dir = tempfile::tempdir().unwrap();
576 let client = ControlClient::new(control_id(&dir.path().join("no-daemon")));
577
578 cancel_run(
579 &client,
580 &CancelArgs {
581 run_id: "stuck-2".to_string(),
582 force: false,
583 },
584 )
585 .await
586 .expect("the fallback succeeds");
587
588 assert_eq!(status_of("stuck-2"), crate::runstate::RunStatus::Cancelled);
589 })
590 .await;
591 }
592
593 #[tokio::test]
595 async fn forcing_an_already_finished_run_is_not_an_error() {
596 crate::runstate::with_isolated_runs_dir_async("ctl-force-terminal", |_base| async {
597 crate::runstate::create_run(&crate::runstate::RunMeta {
598 status: crate::runstate::RunStatus::Complete,
599 ..crate::runstate::RunMeta::new(
600 "done-1".into(),
601 "a".into(),
602 "/p".into(),
603 "t".into(),
604 None,
605 "/w".into(),
606 1,
607 )
608 })
609 .unwrap();
610
611 cancel_run(
612 &ControlClient::new(control_id(std::path::Path::new("/nonexistent"))),
613 &CancelArgs {
614 run_id: "done-1".to_string(),
615 force: true,
616 },
617 )
618 .await
619 .expect("already-finished is reported, not an error");
620
621 assert_eq!(
622 status_of("done-1"),
623 crate::runstate::RunStatus::Complete,
624 "and the recorded outcome is left intact"
625 );
626 })
627 .await;
628 }
629
630 #[tokio::test]
632 async fn forcing_an_unknown_run_reports_no_such_run() {
633 crate::runstate::with_isolated_runs_dir_async("ctl-force-missing", |_base| async {
634 let err = cancel_run(
635 &ControlClient::new(control_id(std::path::Path::new("/nonexistent"))),
636 &CancelArgs {
637 run_id: "never-existed".to_string(),
638 force: true,
639 },
640 )
641 .await
642 .unwrap_err();
643 assert!(err.to_string().contains("no such run"), "got: {err}");
644 })
645 .await;
646 }
647
648 fn respond_args() -> RespondArgs {
651 RespondArgs {
652 request_id: Some("q1".to_string()),
653 value: None,
654 choice: None,
655 approve: false,
656 deny: false,
657 session: false,
658 }
659 }
660
661 #[test]
662 fn build_response_free_text_uses_value_or_empty() {
663 let with_value = build_response(
664 "q1",
665 &RespondArgs {
666 value: Some("hello".to_string()),
667 ..respond_args()
668 },
669 );
670 assert_eq!(with_value, InteractionResponse::text("q1", "hello"));
671 assert_eq!(
673 build_response("q1", &respond_args()),
674 InteractionResponse::text("q1", "")
675 );
676 }
677
678 #[test]
679 fn build_response_choice_selects_index() {
680 let r = build_response(
681 "q1",
682 &RespondArgs {
683 choice: Some(2),
684 ..respond_args()
685 },
686 );
687 assert_eq!(r, InteractionResponse::choice("q1", 2));
688 }
689
690 #[test]
691 fn build_response_approve_and_deny_and_session_scope() {
692 let approved = build_response(
693 "q1",
694 &RespondArgs {
695 approve: true,
696 ..respond_args()
697 },
698 );
699 assert_eq!(
700 approved,
701 InteractionResponse::approval("q1", true, ApprovalScope::Once)
702 );
703 let session = build_response(
704 "q1",
705 &RespondArgs {
706 approve: true,
707 session: true,
708 ..respond_args()
709 },
710 );
711 assert_eq!(
712 session,
713 InteractionResponse::approval("q1", true, ApprovalScope::Session)
714 );
715 let denied = build_response(
716 "q1",
717 &RespondArgs {
718 deny: true,
719 ..respond_args()
720 },
721 );
722 assert_eq!(
723 denied,
724 InteractionResponse::approval("q1", false, ApprovalScope::Once)
725 );
726 }
727
728 #[test]
729 fn kind_label_covers_every_kind() {
730 for (kind, label) in [
731 (InteractionKind::FreeText, "free-text"),
732 (InteractionKind::MultipleChoice, "choice"),
733 (InteractionKind::Confirm, "confirm"),
734 (InteractionKind::ToolApproval, "tool-approval"),
735 (InteractionKind::EditText, "edit-text"),
736 ] {
737 assert_eq!(kind_label(&kind), label);
738 }
739 }
740
741 #[test]
742 fn format_interaction_renders_options_and_tool() {
743 let mut req = InteractionRequest::multiple_choice(
744 "q1",
745 "Pick",
746 vec!["a".to_string(), "b".to_string()],
747 "plan",
748 );
749 req.tool_name = Some("bash".to_string());
750 let out = format_interaction("agent-x", &req);
751 assert!(out.contains("q1 [choice] agent=agent-x stage=plan"));
752 assert!(out.contains("Pick"));
753 assert!(out.contains("0) a"));
754 assert!(out.contains("1) b"));
755 assert!(out.contains("tool: bash"));
756 }
757
758 #[tokio::test]
759 async fn respond_answers_an_interaction() {
760 let r = with_daemon(r#"{"result":"ok","ok":true}"#, |c| async move {
761 respond(&c, &respond_args()).await
762 })
763 .await;
764 assert!(r.is_ok());
765 }
766
767 #[tokio::test]
768 async fn respond_reports_no_open_interaction() {
769 let r = with_daemon(r#"{"result":"ok","ok":false}"#, |c| async move {
770 respond(&c, &respond_args()).await
771 })
772 .await;
773 assert!(r.unwrap_err().to_string().contains("no such open"));
774 }
775
776 #[tokio::test]
777 async fn respond_lists_open_interactions() {
778 let req = InteractionRequest::free_text("q1", "What now?", "plan", true);
779 let line = serde_json::to_string(&ControlResponse::Interactions {
780 interactions: vec![("agent-a".to_string(), req)],
781 })
782 .unwrap();
783 let r = with_daemon(line, |c| async move {
784 respond(
785 &c,
786 &RespondArgs {
787 request_id: None,
788 ..respond_args()
789 },
790 )
791 .await
792 })
793 .await;
794 assert!(r.is_ok());
795 }
796
797 #[tokio::test]
798 async fn respond_lists_when_none_open() {
799 let line = serde_json::to_string(&ControlResponse::Interactions {
800 interactions: vec![],
801 })
802 .unwrap();
803 let r = with_daemon(line, |c| async move {
804 respond(
805 &c,
806 &RespondArgs {
807 request_id: None,
808 ..respond_args()
809 },
810 )
811 .await
812 })
813 .await;
814 assert!(r.is_ok());
815 }
816
817 #[tokio::test]
818 async fn respond_list_rejects_unexpected_response() {
819 let r = with_daemon(r#"{"result":"ok","ok":true}"#, |c| async move {
820 respond(
821 &c,
822 &RespondArgs {
823 request_id: None,
824 ..respond_args()
825 },
826 )
827 .await
828 })
829 .await;
830 assert!(r.unwrap_err().to_string().contains("unexpected"));
831 }
832
833 #[tokio::test]
834 async fn respond_list_errors_when_daemon_absent() {
835 let dir = tempfile::tempdir().unwrap();
836 let client = ControlClient::new(control_id(&dir.path().join("no-daemon")));
837 let err = respond(
838 &client,
839 &RespondArgs {
840 request_id: None,
841 ..respond_args()
842 },
843 )
844 .await
845 .unwrap_err();
846 assert!(err.to_string().contains("not reachable"));
847 }
848}