1use std::path::{Path, PathBuf};
63use std::time::{Duration, Instant};
64
65use anyhow::{Context as _, Result, bail};
66use serde::{Deserialize, Serialize};
67
68use crate::agent::{self, Invocation, SeatState};
69use crate::chat;
70use crate::config::{AgentSpec, Config};
71use crate::git;
72use crate::plan;
73use crate::prompt;
74use crate::verdict::{self, Proposal};
75
76#[derive(Debug, Clone, Serialize, Deserialize)]
80pub struct AdvisorRecord {
81 pub seat: String,
83 pub agent: String,
85 #[serde(default, skip_serializing_if = "Option::is_none")]
87 pub proposal: Option<Proposal>,
88 #[serde(default, skip_serializing_if = "Option::is_none")]
90 pub error: Option<String>,
91 pub duration_ms: u64,
93}
94
95#[derive(Debug, Clone, Serialize, Deserialize, Default)]
100pub struct Advice {
101 pub records: Vec<AdvisorRecord>,
103}
104
105impl Advice {
106 pub fn proposals(&self) -> Vec<(&str, &Proposal)> {
108 self.records
109 .iter()
110 .filter_map(|r| r.proposal.as_ref().map(|p| (r.seat.as_str(), p)))
111 .collect()
112 }
113}
114
115pub async fn run(
124 config: &Config,
125 repo: &Path,
126 draft: &Path,
127 dir: &Path,
128 id: &str,
129) -> Result<Advice> {
130 let requirements = std::fs::read_to_string(draft).with_context(|| {
131 format!(
132 "no task file at {} - the leader was asked to write one there",
133 draft.display()
134 )
135 })?;
136
137 let seats = config.advisors().context("resolving advisor seats")?;
138 if seats.is_empty() {
139 bail!(
140 "`[graph] advisors` is 0, so there is nobody to deliberate with; \
141 the interview draft is unchanged at {0} - file it as-is with \
142 `magi task add --file {0}`, or set `[graph] advisors` above 0 \
143 and re-run `magi plan`.",
144 draft.display()
145 );
146 }
147
148 let worktrees = checkout_worktrees(repo, dir, id, seats.len())
149 .await
150 .with_context(|| {
151 format!(
152 "could not prepare a disposable checkout for the advisor \
153 seats; the interview draft is unchanged at {d} - file it \
154 as-is with `magi task add --file {d}`, or retry `magi plan`.",
155 d = draft.display(),
156 )
157 })?;
158
159 let outcome = deliberate(
162 &requirements,
163 &seats,
164 &worktrees,
165 &DeliberationCtx {
166 config,
167 draft,
168 dir,
169 id,
170 language: &config.graph.language,
171 timeout: Duration::from_secs(config.graph.timeout_judge.max(1)),
176 seed: crate::rng::entropy(),
177 },
178 )
179 .await;
180
181 remove_worktrees(repo, &worktrees).await;
182
183 outcome
184}
185
186async fn checkout_worktrees(repo: &Path, dir: &Path, id: &str, n: usize) -> Result<Vec<PathBuf>> {
196 let root = dir.join(format!("{id}.repo"));
197 let mut paths = Vec::with_capacity(n);
198 for i in 0..n {
199 let wt = root.join(format!("advisor-{}", i + 1));
200 if let Err(e) = git::worktree_add_detached(repo, &wt, "HEAD").await {
201 remove_worktrees(repo, &paths).await;
204 return Err(e);
205 }
206 paths.push(wt);
207 }
208 Ok(paths)
209}
210
211async fn remove_worktrees(repo: &Path, worktrees: &[PathBuf]) {
217 for wt in worktrees {
218 if let Err(e) = git::worktree_remove(repo, wt).await {
219 tracing::warn!(
220 "could not remove disposable advisor worktree {}: {e:#}",
221 wt.display()
222 );
223 }
224 }
225 if let Some(root) = worktrees.first().and_then(|w| w.parent()) {
227 let _ = std::fs::remove_dir(root);
228 }
229}
230
231struct DeliberationCtx<'a> {
235 config: &'a Config,
236 draft: &'a Path,
237 dir: &'a Path,
238 id: &'a str,
239 language: &'a str,
240 timeout: Duration,
241 seed: u64,
242}
243
244async fn deliberate(
248 requirements: &str,
249 seats: &[AgentSpec],
250 worktrees: &[PathBuf],
251 ctx: &DeliberationCtx<'_>,
252) -> Result<Advice> {
253 let draft = ctx.draft;
254 let artifacts = ctx.dir.join(format!("{}.advisors", ctx.id));
255
256 let advice = gather(
257 seats,
258 requirements,
259 worktrees,
260 &GatherCtx {
261 artifacts: &artifacts,
262 run: ctx.id,
263 language: ctx.language,
264 timeout: ctx.timeout,
265 seed: ctx.seed,
266 },
267 )
268 .await;
269
270 let advice_path = ctx.dir.join(format!("{}.advisors.json", ctx.id));
274 std::fs::write(
275 &advice_path,
276 serde_json::to_string_pretty(&advice).context("serialize the advisor records")?,
277 )
278 .with_context(|| format!("write {}", advice_path.display()))?;
279
280 let proposals = advice.proposals();
281 if proposals.is_empty() {
282 bail!(
283 "none of {n} advisor seat(s) produced a usable design proposal \
284 (see {record}); the interview draft is unchanged at {d} - file \
285 it as-is with `magi task add --file {d}`, or retry `magi plan`.",
286 n = seats.len(),
287 record = advice_path.display(),
288 d = draft.display(),
289 );
290 }
291
292 let planner = plan::pick(
293 &ctx.config.agents,
294 ctx.config.roles.planner.as_deref(),
295 &plan::installed,
296 )
297 .context("resolving the planner seat for design synthesis")?;
298 let mut seat = SeatState::new("plan-synthesis", &planner.id, ctx.seed);
299 let synth_prompt = prompt::synthesize(requirements, &proposals, ctx.language);
300 let out = agent::invoke(
301 &planner,
302 &mut seat,
303 &Invocation {
304 cwd: &worktrees[0],
311 prompt: &synth_prompt,
312 timeout: ctx.timeout,
313 allow_write: false,
314 sessions: false,
315 artifacts: &artifacts,
316 stem: "synthesis",
317 run: ctx.id,
318 node: "plan-advise",
319 cache_dir: None,
320 },
321 )
322 .await
323 .with_context(|| {
324 format!(
325 "the planner seat could not synthesize the design proposals; the \
326 interview draft is unchanged at {0} - file it as-is with `magi \
327 task add --file {0}`, or retry `magi plan`.",
328 draft.display()
329 )
330 })?;
331
332 if !out.usable() {
333 bail!(
334 "the planner seat produced nothing usable while synthesizing the \
335 design proposals; the interview draft is unchanged at {0} - file \
336 it as-is with `magi task add --file {0}`, or retry `magi plan`.",
337 draft.display()
338 );
339 }
340
341 let synthesized = chat::extract_draft(&out.text).with_context(|| {
342 format!(
343 "the planner seat's reply had no fenced ```task block; the \
344 interview draft is unchanged at {0} - file it as-is with `magi \
345 task add --file {0}`, or retry `magi plan`.",
346 draft.display()
347 )
348 })?;
349
350 if let Err(problems) = plan::review_draft(&synthesized) {
358 let hard: Vec<&String> = problems
359 .iter()
360 .filter(|p| p.as_str() != plan::SHORT_DRAFT)
361 .collect();
362 if !hard.is_empty() {
363 let list = hard
364 .iter()
365 .map(|p| format!(" - {p}"))
366 .collect::<Vec<_>>()
367 .join("\n");
368 bail!(
369 "the planner seat's synthesis is not a usable task file:\n{list}\n\n\
370 the interview draft is unchanged at {d} - file it as-is with \
371 `magi task add --file {d}`, or retry `magi plan`.",
372 d = draft.display(),
373 );
374 }
375 }
376
377 std::fs::write(draft, &synthesized).with_context(|| format!("write {}", draft.display()))?;
378
379 Ok(advice)
380}
381
382struct GatherCtx<'a> {
386 artifacts: &'a Path,
387 run: &'a str,
388 language: &'a str,
389 timeout: Duration,
390 seed: u64,
391}
392
393async fn gather(
400 seats: &[AgentSpec],
401 requirements: &str,
402 worktrees: &[PathBuf],
403 ctx: &GatherCtx<'_>,
404) -> Advice {
405 let n = seats.len();
406 let mut set = tokio::task::JoinSet::new();
407 for (i, spec) in seats.iter().cloned().enumerate() {
408 let cwd = worktrees[i].clone();
409 let requirements = requirements.to_owned();
410 let artifacts = ctx.artifacts.to_owned();
411 let run = ctx.run.to_owned();
412 let language = ctx.language.to_owned();
413 let timeout = ctx.timeout;
414 let seed = ctx.seed;
415 let key = format!("advisor-{}", i + 1);
416 set.spawn(async move {
417 let mut seat = SeatState::new(&key, &spec.id, seed ^ (i as u64 + 1));
418 let prompt = prompt::advisor(&requirements, i + 1, n, &language);
419 let started = Instant::now();
420 let outcome = agent::invoke(
421 &spec,
422 &mut seat,
423 &Invocation {
424 cwd: &cwd,
425 prompt: &prompt,
426 timeout,
427 allow_write: false,
428 sessions: false,
429 artifacts: &artifacts,
430 stem: &key,
431 run: &run,
432 node: "plan-advise",
433 cache_dir: None,
434 },
435 )
436 .await;
437 to_record(key, spec.id, started.elapsed(), outcome)
438 });
439 }
440 let mut records = Vec::with_capacity(n);
441 while let Some(res) = set.join_next().await {
442 records.push(match res {
443 Ok(rec) => rec,
444 Err(e) => AdvisorRecord {
445 seat: "?".to_owned(),
446 agent: "?".to_owned(),
447 proposal: None,
448 error: Some(format!("advisor task panicked: {e}")),
449 duration_ms: 0,
450 },
451 });
452 }
453 records.sort_by(|a, b| a.seat.cmp(&b.seat));
456 Advice { records }
457}
458
459fn to_record(
460 seat: String,
461 agent_id: String,
462 elapsed: Duration,
463 outcome: Result<agent::AgentOutput>,
464) -> AdvisorRecord {
465 match outcome {
466 Ok(out) if out.usable() => {
467 match verdict::extract_json::<Proposal>(&out.text)
468 .and_then(|p| p.validate().map(|()| p))
469 {
470 Ok(proposal) => AdvisorRecord {
471 seat,
472 agent: agent_id,
473 proposal: Some(proposal),
474 error: None,
475 duration_ms: out.duration_ms,
476 },
477 Err(e) => AdvisorRecord {
478 seat,
479 agent: agent_id,
480 proposal: None,
481 error: Some(e.to_string()),
482 duration_ms: out.duration_ms,
483 },
484 }
485 }
486 Ok(out) => AdvisorRecord {
487 seat,
488 agent: agent_id,
489 proposal: None,
490 error: Some(if out.timed_out {
491 "timed out".to_owned()
492 } else {
493 format!("exit {:?}: {}", out.exit_code, out.text.trim())
494 }),
495 duration_ms: out.duration_ms,
496 },
497 Err(e) => AdvisorRecord {
498 seat,
499 agent: agent_id,
500 proposal: None,
501 error: Some(e.to_string()),
502 duration_ms: elapsed.as_millis() as u64,
503 },
504 }
505}
506
507#[cfg(test)]
508mod tests {
509 use super::*;
510 use crate::config::{AgentKind, Graph, Roles};
511
512 fn command(id: &str, output: &str) -> AgentSpec {
518 AgentSpec {
519 id: id.to_owned(),
520 kind: AgentKind::Command,
521 model: None,
522 command: vec![
523 "sh".to_owned(),
524 "-c".to_owned(),
525 format!("cat >/dev/null && cat <<'EOF'\n{output}\nEOF"),
526 ],
527 extra_args: Vec::new(),
528 env: Default::default(),
529 prompt_delivery: None,
530 }
531 }
532
533 fn proposal_json(approach: &str) -> String {
534 format!(
535 "```json\n{{\"approach\":\"{approach}\",\"key_tradeoff\":\"t\",\
536 \"risks\":[\"r\"],\"touches\":[\"src/a.rs\"],\
537 \"why_not_naive\":\"w\"}}\n```"
538 )
539 }
540
541 fn good_draft() -> String {
542 "# Rework the config loader\n\
543 \n\
544 ## Context\n\
545 \n\
546 placeholder context.\n\
547 \n\
548 ## Change\n\
549 \n\
550 placeholder change.\n\
551 \n\
552 ## Constraints\n\
553 \n\
554 No new dependencies.\n\
555 \n\
556 ## Completion criteria\n\
557 \n\
558 - [ ] it works\n\
559 \n\
560 ## Out of scope\n\
561 \n\
562 nothing\n"
563 .to_owned()
564 }
565
566 fn synthesized_task_block() -> String {
567 format!(
568 "```task\n{}```",
569 good_draft().replace("placeholder", "synthesized")
570 )
571 }
572
573 fn init_repo(dir: &Path) {
578 let run = |args: &[&str]| {
579 let out = std::process::Command::new("git")
580 .args(args)
581 .current_dir(dir)
582 .output()
583 .expect("spawn git");
584 assert!(
585 out.status.success(),
586 "git {args:?} failed: {}",
587 String::from_utf8_lossy(&out.stderr)
588 );
589 };
590 std::fs::create_dir_all(dir).unwrap();
591 run(&["init", "-b", "main"]);
592 run(&["config", "user.name", "magi test"]);
593 run(&["config", "user.email", "magi@example.com"]);
594 std::fs::write(dir.join("README.md"), "# fixture\n").unwrap();
595 run(&["add", "-A"]);
596 run(&["commit", "-m", "init"]);
597 }
598
599 #[tokio::test]
600 async fn gather_records_every_seat_including_one_that_fails() {
601 let seats = vec![
602 command("sage-a", &proposal_json("do X")),
603 command("sage-b", "not json at all"),
604 ];
605 let dir = tempfile::tempdir().unwrap();
606 let worktrees = vec![dir.path().join("wt-1"), dir.path().join("wt-2")];
607 for wt in &worktrees {
608 std::fs::create_dir_all(wt).unwrap();
609 }
610 let advice = gather(
611 &seats,
612 "the requirements",
613 &worktrees,
614 &GatherCtx {
615 artifacts: &dir.path().join("artifacts"),
616 run: "test-run",
617 language: "en",
618 timeout: Duration::from_secs(30),
619 seed: 7,
620 },
621 )
622 .await;
623
624 assert_eq!(advice.records.len(), 2);
625 assert_eq!(advice.records[0].seat, "advisor-1");
626 assert_eq!(advice.records[1].seat, "advisor-2");
627 let ok = advice.records[0]
628 .proposal
629 .as_ref()
630 .expect("advisor-1 parses");
631 assert_eq!(ok.approach, "do X");
632 assert!(advice.records[1].proposal.is_none());
633 assert!(advice.records[1].error.is_some());
634 }
635
636 fn config(agents: Vec<AgentSpec>, advisors: usize) -> Config {
637 Config {
638 agents,
639 roles: Roles {
640 advisors: vec!["sage-a".to_owned(), "sage-b".to_owned()],
641 planner: Some("planner".to_owned()),
642 ..Roles::default()
643 },
644 graph: Graph {
645 advisors,
646 ..Graph::default()
647 },
648 ..Config::default()
649 }
650 }
651
652 #[tokio::test]
660 async fn a_relative_path_write_from_an_advisor_lands_in_its_worktree_not_the_operators_repository()
661 {
662 let tmp = tempfile::tempdir().unwrap();
663 let repo = tmp.path().join("repo");
664 init_repo(&repo);
665 let dir = tmp.path().join("drafts");
666 std::fs::create_dir_all(&dir).unwrap();
667 let draft = dir.join("20260906-000000-kl12.md");
668 std::fs::write(&draft, good_draft()).unwrap();
669
670 let writer = AgentSpec {
674 id: "sage-a".to_owned(),
675 kind: AgentKind::Command,
676 model: None,
677 command: vec![
678 "sh".to_owned(),
679 "-c".to_owned(),
680 format!(
681 "cat >/dev/null && touch leaked-by-advisor.txt && cat <<'EOF'\n{}\nEOF",
682 proposal_json("do X")
683 ),
684 ],
685 extra_args: Vec::new(),
686 env: Default::default(),
687 prompt_delivery: None,
688 };
689
690 let cfg = config(
691 vec![
692 writer,
693 command("sage-b", &proposal_json("do Y")),
694 command("planner", &synthesized_task_block()),
695 ],
696 2,
697 );
698
699 run(&cfg, &repo, &draft, &dir, "20260906-000000-kl12")
700 .await
701 .expect("deliberation still succeeds even though a seat wrote something");
702
703 assert!(
704 !repo.join("leaked-by-advisor.txt").exists(),
705 "an advisor's write must land in its disposable worktree, never in the operator's repository"
706 );
707 }
708
709 #[tokio::test]
710 async fn run_writes_the_raw_records_and_overwrites_the_draft_with_the_synthesis() {
711 let tmp = tempfile::tempdir().unwrap();
712 let repo = tmp.path().join("repo");
713 init_repo(&repo);
714 let dir = tmp.path().join("drafts");
715 std::fs::create_dir_all(&dir).unwrap();
716 let draft = dir.join("20260906-000000-ab12.md");
717 std::fs::write(&draft, good_draft()).unwrap();
718
719 let cfg = config(
720 vec![
721 command("sage-a", &proposal_json("do X")),
722 command("sage-b", &proposal_json("do Y")),
723 command("planner", &synthesized_task_block()),
724 ],
725 2,
726 );
727
728 let advice = run(&cfg, &repo, &draft, &dir, "20260906-000000-ab12")
729 .await
730 .expect("deliberation succeeds");
731 assert_eq!(advice.proposals().len(), 2);
732
733 let advice_path = dir.join("20260906-000000-ab12.advisors.json");
734 let raw = std::fs::read_to_string(&advice_path).expect("raw record on disk");
735 let reread: Advice = serde_json::from_str(&raw).expect("parses back");
736 assert_eq!(reread.records.len(), 2);
737
738 let final_draft = std::fs::read_to_string(&draft).unwrap();
739 assert!(
740 final_draft.contains("synthesized context"),
741 "the draft must be overwritten with the synthesis: {final_draft}"
742 );
743 assert!(final_draft.contains("## Completion criteria"));
744
745 assert!(
749 !dir.join("20260906-000000-ab12.repo").exists(),
750 "advisor worktrees must be cleaned up after the run"
751 );
752 }
753
754 #[tokio::test]
755 async fn run_leaves_the_draft_untouched_when_no_advisor_produces_a_proposal() {
756 let tmp = tempfile::tempdir().unwrap();
757 let repo = tmp.path().join("repo");
758 init_repo(&repo);
759 let dir = tmp.path().join("drafts");
760 std::fs::create_dir_all(&dir).unwrap();
761 let draft = dir.join("20260906-000000-cd34.md");
762 let original = good_draft();
763 std::fs::write(&draft, &original).unwrap();
764
765 let cfg = config(
766 vec![
767 command("sage-a", "garbage"),
768 command("sage-b", "also garbage"),
769 command("planner", &synthesized_task_block()),
770 ],
771 2,
772 );
773
774 let err = run(&cfg, &repo, &draft, &dir, "20260906-000000-cd34")
775 .await
776 .expect_err("no proposal must fail the stage");
777 let msg = err.to_string();
778 assert!(msg.contains(&draft.display().to_string()), "{msg}");
779 assert!(msg.contains("magi task add --file"), "{msg}");
780 assert_eq!(
781 std::fs::read_to_string(&draft).unwrap(),
782 original,
783 "the interview draft must survive a total advisor failure"
784 );
785 assert!(dir.join("20260906-000000-cd34.advisors.json").is_file());
788 }
789
790 #[tokio::test]
791 async fn run_leaves_the_draft_untouched_when_the_planner_replies_with_no_task_block() {
792 let tmp = tempfile::tempdir().unwrap();
793 let repo = tmp.path().join("repo");
794 init_repo(&repo);
795 let dir = tmp.path().join("drafts");
796 std::fs::create_dir_all(&dir).unwrap();
797 let draft = dir.join("20260906-000000-ef56.md");
798 let original = good_draft();
799 std::fs::write(&draft, &original).unwrap();
800
801 let cfg = config(
802 vec![
803 command("sage-a", &proposal_json("do X")),
804 command("sage-b", &proposal_json("do Y")),
805 command("planner", "sure, here is my answer with no fence"),
806 ],
807 2,
808 );
809
810 let err = run(&cfg, &repo, &draft, &dir, "20260906-000000-ef56")
811 .await
812 .expect_err("a synthesis with no task block must fail the stage");
813 let msg = err.to_string();
814 assert!(msg.contains(&draft.display().to_string()), "{msg}");
815 assert_eq!(std::fs::read_to_string(&draft).unwrap(), original);
816 }
817
818 #[tokio::test]
824 async fn run_leaves_the_draft_untouched_when_the_synthesis_fence_never_closes() {
825 let tmp = tempfile::tempdir().unwrap();
826 let repo = tmp.path().join("repo");
827 init_repo(&repo);
828 let dir = tmp.path().join("drafts");
829 std::fs::create_dir_all(&dir).unwrap();
830 let draft = dir.join("20260906-000000-ij90.md");
831 let original = good_draft();
832 std::fs::write(&draft, &original).unwrap();
833
834 let cfg = config(
835 vec![
836 command("sage-a", &proposal_json("do X")),
837 command("sage-b", &proposal_json("do Y")),
838 command("planner", "```task\n# incomplete"),
839 ],
840 2,
841 );
842
843 let err = run(&cfg, &repo, &draft, &dir, "20260906-000000-ij90")
844 .await
845 .expect_err("an incomplete synthesis must not become the task file");
846 let msg = err.to_string();
847 assert!(msg.contains(&draft.display().to_string()), "{msg}");
848 assert!(msg.contains("not a usable task file"), "{msg}");
849 assert_eq!(
850 std::fs::read_to_string(&draft).unwrap(),
851 original,
852 "the interview draft must survive an incomplete synthesis"
853 );
854 }
855
856 #[tokio::test]
857 async fn run_reports_a_missing_draft_against_the_path_the_leader_was_given() {
858 let tmp = tempfile::tempdir().unwrap();
859 let dir = tmp.path().join("drafts");
860 std::fs::create_dir_all(&dir).unwrap();
861 let draft = dir.join("never-written.md");
862
863 let cfg = config(vec![command("sage-a", &proposal_json("x"))], 1);
864 let msg = run(&cfg, tmp.path(), &draft, &dir, "never-written")
865 .await
866 .expect_err("nothing to deliberate over")
867 .to_string();
868 assert!(msg.contains(&draft.display().to_string()), "{msg}");
869 }
870
871 #[tokio::test]
872 async fn zero_advisors_is_an_error_that_still_names_the_draft() {
873 let tmp = tempfile::tempdir().unwrap();
874 let dir = tmp.path().join("drafts");
875 std::fs::create_dir_all(&dir).unwrap();
876 let draft = dir.join("20260906-000000-gh78.md");
877 std::fs::write(&draft, good_draft()).unwrap();
878
879 let cfg = config(vec![command("sage-a", &proposal_json("x"))], 0);
880 let msg = run(&cfg, tmp.path(), &draft, &dir, "20260906-000000-gh78")
881 .await
882 .expect_err("nobody to deliberate with")
883 .to_string();
884 assert!(msg.contains("advisors` is 0"), "{msg}");
885 assert!(msg.contains(&draft.display().to_string()), "{msg}");
886 }
887}