1use std::path::{Path, PathBuf};
50use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
51use std::time::Duration;
52
53use anyhow::{Context, Result, bail};
54use jiff::Timestamp;
55use serde::{Deserialize, Serialize};
56
57use crate::agent::{self, Invocation, SeatState};
58use crate::config::Config;
59use crate::plan;
60use crate::queue::{Queue, Source, Task};
61
62pub const SCHEMA: u32 = 1;
64
65const TURN_TIMEOUT: Duration = Duration::from_secs(900);
74
75const SEAT: &str = "talk";
79
80const MAGI_NOTE: &str = "magi: ";
83
84#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
86#[serde(rename_all = "lowercase")]
87pub enum Who {
88 Operator,
90 Agent,
93}
94
95#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
97#[serde(deny_unknown_fields)]
98pub struct Turn {
99 pub who: Who,
101 pub body: String,
103 pub at: Timestamp,
105}
106
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
111#[serde(rename_all = "lowercase")]
112pub enum TalkStatus {
113 Open,
116 Closed,
118}
119
120impl TalkStatus {
121 pub fn open(self) -> bool {
123 matches!(self, Self::Open)
124 }
125
126 pub fn as_str(self) -> &'static str {
128 match self {
129 Self::Open => "open",
130 Self::Closed => "closed",
131 }
132 }
133}
134
135#[derive(Debug, Clone, Serialize, Deserialize)]
137#[serde(deny_unknown_fields)]
138pub struct Talk {
139 pub schema: u32,
141 pub id: String,
143 pub repo: PathBuf,
145 pub agent: String,
147 pub status: TalkStatus,
149 pub turns: Vec<Turn>,
151 pub created_at: Timestamp,
153 pub updated_at: Timestamp,
155 seat: SeatState,
161}
162
163impl Talk {
164 pub fn short(&self) -> &str {
166 short(&self.id)
167 }
168}
169
170#[derive(Debug, Clone)]
172pub struct Talks {
173 root: PathBuf,
174 lock: Arc<Mutex<()>>,
183}
184
185impl Talks {
186 pub fn open() -> Self {
188 Self::at(crate::run::home().join("talks"))
189 }
190
191 pub fn at(root: PathBuf) -> Self {
194 Self {
195 root,
196 lock: Arc::new(Mutex::new(())),
197 }
198 }
199
200 fn guard(&self) -> MutexGuard<'_, ()> {
209 self.lock.lock().unwrap_or_else(PoisonError::into_inner)
210 }
211
212 pub fn root(&self) -> &Path {
214 &self.root
215 }
216
217 pub fn path_of(&self, id: &str) -> PathBuf {
219 self.root.join(format!("{id}.json"))
220 }
221
222 pub fn artifacts_of(&self, id: &str) -> PathBuf {
225 self.root.join(format!("{id}.artifacts"))
226 }
227
228 pub fn put(&self, t: &mut Talk) -> Result<()> {
231 std::fs::create_dir_all(&self.root)
232 .with_context(|| format!("create {}", self.root.display()))?;
233 t.updated_at = Timestamp::now();
234 let body = serde_json::to_string_pretty(t).context("serialize talk")?;
235 let path = self.path_of(&t.id);
236 let tmp = path.with_extension("json.tmp");
237 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
238 std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
239 Ok(())
240 }
241
242 pub fn get(&self, id: &str) -> Result<Talk> {
244 let resolved = self.resolve_id(id)?;
245 read_path(&self.path_of(&resolved))
246 }
247
248 pub fn list(&self) -> Vec<Talk> {
252 let mut all: Vec<Talk> = std::fs::read_dir(&self.root)
253 .into_iter()
254 .flatten()
255 .flatten()
256 .map(|e| e.path())
257 .filter(|p| p.extension().is_some_and(|x| x == "json"))
258 .filter_map(|p| read_path(&p).ok())
259 .collect();
260 all.sort_unstable_by(|a, b| {
261 let rank = |t: &Talk| u8::from(!t.status.open());
262 rank(a).cmp(&rank(b)).then_with(|| b.id.cmp(&a.id))
263 });
264 all
265 }
266
267 pub fn resolve_id(&self, prefix: &str) -> Result<String> {
269 if self.path_of(prefix).is_file() {
270 return Ok(prefix.to_owned());
271 }
272 let hits: Vec<String> = self
273 .list()
274 .into_iter()
275 .map(|t| t.id)
276 .filter(|id| id.starts_with(prefix) || id.ends_with(prefix))
277 .collect();
278 match hits.len() {
279 1 => Ok(hits.into_iter().next().expect("exactly one hit")),
280 0 => bail!("no talk matches `{prefix}`"),
281 _ => bail!(
282 "`{prefix}` matches {} talks: {}",
283 hits.len(),
284 hits.join(", ")
285 ),
286 }
287 }
288
289 pub fn revision(&self) -> u64 {
293 std::fs::read_dir(&self.root)
294 .into_iter()
295 .flatten()
296 .flatten()
297 .filter_map(|e| e.metadata().ok())
298 .filter_map(|m| m.modified().ok())
299 .filter_map(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
300 .map(|d| d.as_millis() as u64)
301 .max()
302 .unwrap_or(0)
303 }
304
305 pub fn count_open(&self) -> usize {
307 self.list().iter().filter(|t| t.status.open()).count()
308 }
309
310 pub fn remove(&self, id: &str) -> Result<()> {
323 let _guard = self.guard();
324 let resolved = self.resolve_id(id)?;
325 let path = self.path_of(&resolved);
326 std::fs::remove_file(&path).with_context(|| format!("remove {}", path.display()))?;
327 let artifacts = self.artifacts_of(&resolved);
328 if artifacts.is_dir() {
329 std::fs::remove_dir_all(&artifacts)
330 .with_context(|| format!("remove {}", artifacts.display()))?;
331 }
332 Ok(())
333 }
334}
335
336pub fn begin(store: &Talks, cfg: &Config, repo: PathBuf, agent: Option<&str>) -> Result<Talk> {
353 let repo = repo.canonicalize().unwrap_or(repo);
357 let want = agent
358 .or(cfg.roles.chatter.as_deref())
359 .or(cfg.roles.planner.as_deref());
360 let spec = plan::pick(&cfg.agents, want, &plan::installed)?;
361
362 let now = Timestamp::now();
363 let mut talk = Talk {
364 schema: SCHEMA,
365 id: new_id(),
366 repo,
367 agent: spec.id.clone(),
368 status: TalkStatus::Open,
369 turns: Vec::new(),
370 created_at: now,
371 updated_at: now,
372 seat: SeatState::new(SEAT, &spec.id, crate::rng::entropy()),
373 };
374 store.put(&mut talk)?;
375 Ok(talk)
376}
377
378pub fn record(talk: &mut Talk, store: &Talks, text: &str) -> Result<String> {
386 let _guard = store.guard();
394 let Ok(fresh) = store.get(&talk.id) else {
399 bail!("talk {} was deleted", talk.short());
400 };
401 talk.status = fresh.status;
402 if !talk.status.open() {
403 bail!(
404 "talk {} is {} and takes no more turns",
405 talk.short(),
406 talk.status.as_str()
407 );
408 }
409 let text = text.trim();
410 if text.is_empty() {
411 bail!("nothing to say");
412 }
413 talk.turns.push(Turn {
414 who: Who::Operator,
415 body: text.to_owned(),
416 at: Timestamp::now(),
417 });
418 store.put(talk)?;
419 Ok(text.to_owned())
420}
421
422pub async fn say(talk: &mut Talk, store: &Talks, cfg: &Config, text: &str) -> Result<()> {
425 let text = record(talk, store, text)?;
426 turn(talk, store, cfg, &text).await
427}
428
429pub async fn respond(talk: &mut Talk, store: &Talks, cfg: &Config, text: &str) -> Result<()> {
432 turn(talk, store, cfg, text).await
433}
434
435pub fn close(talk: &mut Talk, store: &Talks) -> Result<()> {
453 let _guard = store.guard();
454 let mut fresh = store
455 .get(&talk.id)
456 .with_context(|| format!("talk {} was deleted", talk.short()))?;
457 fresh.status = TalkStatus::Closed;
458 store.put(&mut fresh)?;
459 *talk = fresh;
460 Ok(())
461}
462
463pub fn reopen(talk: &mut Talk, store: &Talks) -> Result<()> {
474 let _guard = store.guard();
475 let mut fresh = store
476 .get(&talk.id)
477 .with_context(|| format!("talk {} was deleted", talk.short()))?;
478 fresh.status = TalkStatus::Open;
479 store.put(&mut fresh)?;
480 *talk = fresh;
481 Ok(())
482}
483
484async fn turn(talk: &mut Talk, store: &Talks, cfg: &Config, text: &str) -> Result<()> {
492 let spec = cfg
493 .agents
494 .iter()
495 .find(|a| a.id == talk.agent)
496 .with_context(|| {
497 format!(
498 "talk {} was opened with agent `{}`, which is no longer in \
499 the roster; restore it in magi.toml or start a new \
500 conversation",
501 talk.short(),
502 talk.agent
503 )
504 })?;
505
506 let resuming = agent::has_session(spec.kind, &talk.seat, cfg.graph.sessions);
507 let body = if talk.seat.turns == 0 {
508 format!(
509 "{}\n\n# Operator\n\n{text}",
510 briefing(&talk.repo, &cfg.graph.language, cfg.talk.allow_write)
511 )
512 } else if resuming {
513 text.to_owned()
514 } else {
515 format!("{}\n\n{text}", transcript(talk))
516 };
517
518 let artifacts = store.artifacts_of(&talk.id);
519 let stem = format!("turn-{}", talk.seat.turns + 1);
520 let cache_dir = cfg.cache_dir();
523 let inv = Invocation {
524 cwd: &talk.repo,
525 prompt: &body,
526 timeout: TURN_TIMEOUT,
527 allow_write: cfg.talk.allow_write,
532 sessions: cfg.graph.sessions,
533 artifacts: &artifacts,
534 stem: &stem,
535 run: &talk.id,
538 node: "chat",
539 cache_dir: cache_dir.as_deref(),
540 };
541
542 let outcome = agent::invoke(spec, &mut talk.seat, &inv).await;
543 let note = |why: String| Turn {
544 who: Who::Agent,
545 body: format!("{MAGI_NOTE}{why}"),
546 at: Timestamp::now(),
547 };
548 let (reply, failure) = match outcome {
549 Err(e) => (
550 note(format!("could not run agent `{}`: {e}", talk.agent)),
551 Some(format!("could not run agent `{}`: {e}", talk.agent)),
552 ),
553 Ok(out) if out.quota_exhausted() => {
554 let reset = out
555 .quota
556 .as_ref()
557 .and_then(|q| q.reset.clone())
558 .map_or_else(String::new, |r| format!(" (resets {r})"));
559 let why = format!(
560 "agent `{}` is out of quota{reset}; your message is saved, so \
561 say it again when the window reopens",
562 talk.agent
563 );
564 (note(why.clone()), Some(why))
565 }
566 Ok(out) if out.timed_out => {
567 let why = format!(
568 "agent `{}` did not answer within {}s; your message is saved",
569 talk.agent,
570 TURN_TIMEOUT.as_secs()
571 );
572 (note(why.clone()), Some(why))
573 }
574 Ok(out) if !out.usable() => {
575 let why = format!(
576 "agent `{}` produced no answer (exit {}); your message is saved",
577 talk.agent,
578 out.exit_code
579 .map_or_else(|| "unknown".to_owned(), |c| c.to_string())
580 );
581 (note(why.clone()), Some(why))
582 }
583 Ok(out) => (
584 Turn {
585 who: Who::Agent,
586 body: out.text.trim().to_owned(),
587 at: Timestamp::now(),
588 },
589 None,
590 ),
591 };
592
593 let _guard = store.guard();
605 let Ok(fresh) = store.get(&talk.id) else {
611 return Ok(());
612 };
613 talk.status = fresh.status;
614 talk.turns.push(reply);
615 store.put(talk)?;
616
617 match failure {
618 Some(why) => bail!("{why}"),
619 None => Ok(()),
620 }
621}
622
623fn transcript(talk: &Talk) -> String {
626 let mut out = String::from(
627 "This conversation cannot resume on the CLI's side, so here is \
628 everything said so far; answer only the last message.\n",
629 );
630 for t in &talk.turns {
631 let who = match t.who {
632 Who::Operator => "operator",
633 Who::Agent => "you",
634 };
635 out.push_str(&format!("\n## {who}\n\n{}\n", t.body.trim()));
636 }
637 out
638}
639
640pub fn briefing(repo: &Path, language: &str, allow_write: bool) -> String {
652 let write_policy = if allow_write {
653 "This repository has set `[talk] allow_write = true`, so you may \
654 write files here - but only a small, already-decided edit the \
655 operator names outright in this conversation, not an \
656 implementation. Once you have made it, say plainly what you \
657 edited. Anything bigger, or anything still open-ended, still goes \
658 through the queue below rather than being done here."
659 } else {
660 "Do not write files. Implementing a change is not this \
661 conversation's job; a separate, blind competition of agents does \
662 that, and a repository this conversation has already edited would \
663 make their diffs unjudgeable."
664 };
665 let mut out = format!(
666 "You are magi's standing conversation partner for its operator, who \
667 usually has this open on a phone. Keep replies short: no preamble, \
668 no restating what they just said.\n\n\
669 # Repository\n\n{repo}\n\n\
670 You may look around: read files, run shell commands, search history, \
671 run tests - whatever answers the question. {write_policy}\n\n\
672 # When the operator wants something done\n\n\
673 Run:\n\n\
674 magi task add --solo --repo {repo} <instruction>\n\n\
675 and tell the operator the task id it prints, so they can follow it \
676 from the Queue. Write <instruction> so that an implementer who has \
677 never seen this conversation can act on it alone - it is everything \
678 they get. Use --solo: it runs the task through one implementer \
679 straight into review instead of the usual multi-agent competition, \
680 which is the right shape for a change this conversation has already \
681 settled, rather than one still worth several independent takes.\n",
682 repo = repo.display(),
683 );
684 out.push_str(&language_note(language));
685 out
686}
687
688fn language_note(language: &str) -> String {
692 if language.trim().is_empty() || language.eq_ignore_ascii_case("en") {
693 String::new()
694 } else {
695 format!("\nHold this conversation in {language}.\n")
696 }
697}
698
699pub fn tasks_of(queue: &Queue, talk_id: &str) -> Vec<Task> {
706 let mut tasks: Vec<Task> = queue
707 .list()
708 .into_iter()
709 .filter(|t| matches!(&t.source, Source::Agent { run, .. } if run == talk_id))
710 .collect();
711 tasks.sort_unstable_by(|a, b| a.id.cmp(&b.id));
712 tasks
713}
714
715fn read_path(path: &Path) -> Result<Talk> {
716 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
717 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))
718}
719
720fn short(id: &str) -> &str {
721 id.split('-').next_back().unwrap_or(id)
722}
723
724fn new_id() -> String {
725 let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
726 let seed = crate::rng::entropy();
727 format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
728}
729
730#[cfg(test)]
731mod tests {
732 use std::collections::BTreeMap;
733
734 use crate::config::{AgentKind, AgentSpec, Graph};
735 use crate::queue::{Queue, Source, Task};
736
737 use super::*;
738
739 fn store() -> (tempfile::TempDir, Talks) {
741 let tmp = tempfile::tempdir().expect("tempdir");
742 let talks = Talks::at(tmp.path().join("talks"));
743 (tmp, talks)
744 }
745
746 fn mock_agent(dir: &Path, script: &str, env: BTreeMap<String, String>) -> AgentSpec {
750 let path = dir.join("mock-talk-agent.sh");
751 std::fs::write(&path, script).expect("write mock");
752 AgentSpec {
753 id: "mock".to_owned(),
754 kind: AgentKind::Command,
755 model: None,
756 command: vec!["sh".to_owned(), path.to_string_lossy().into_owned()],
757 extra_args: Vec::new(),
758 env,
759 prompt_delivery: None,
760 }
761 }
762
763 fn config(spec: AgentSpec) -> Config {
764 Config {
765 agents: vec![spec],
766 graph: Graph {
767 language: "en".to_owned(),
768 ..Graph::default()
769 },
770 ..Config::default()
771 }
772 }
773
774 const REPLY: &str = "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' \"$MOCK_REPLY\"\n";
776
777 const BROKEN: &str = "#!/bin/sh\ncat >/dev/null\nexit 3\n";
779
780 const ECHO: &str = "#!/bin/sh\ncat\n";
783
784 fn env(reply: &str) -> BTreeMap<String, String> {
785 BTreeMap::from([("MOCK_REPLY".to_owned(), reply.to_owned())])
786 }
787
788 #[test]
789 fn the_frozen_json_field_names_round_trip_through_disk() {
790 let (tmp, talks) = store();
791 let mut talk = Talk {
792 schema: SCHEMA,
793 id: "20260904-014455-ab12".to_owned(),
794 repo: tmp.path().to_owned(),
795 agent: "sonnet".to_owned(),
796 status: TalkStatus::Open,
797 turns: Vec::new(),
798 created_at: Timestamp::now(),
799 updated_at: Timestamp::now(),
800 seat: SeatState::new(SEAT, "sonnet", 7),
801 };
802 talks.put(&mut talk).expect("put");
803
804 let raw = std::fs::read_to_string(talks.path_of(&talk.id)).expect("read back");
805 let v: serde_json::Value = serde_json::from_str(&raw).expect("parse");
806 for field in [
807 "schema",
808 "id",
809 "repo",
810 "agent",
811 "status",
812 "turns",
813 "created_at",
814 "updated_at",
815 ] {
816 assert!(v.get(field).is_some(), "missing field `{field}`");
817 }
818 assert_eq!(v["schema"], 1);
819 assert_eq!(v["status"], "open");
820
821 let back = talks.get(&talk.id).expect("get");
822 assert_eq!(back.id, talk.id);
823 assert_eq!(back.status, TalkStatus::Open);
824 }
825
826 #[test]
827 fn opening_a_talk_takes_no_agent_turn() {
828 let (tmp, talks) = store();
829 let spec = mock_agent(tmp.path(), BROKEN, BTreeMap::new());
833 let cfg = config(spec);
834
835 let talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
836 assert_eq!(talk.status, TalkStatus::Open);
837 assert!(talk.turns.is_empty(), "nothing has been said yet");
838
839 let on_disk = talks.get(&talk.id).expect("get");
840 assert_eq!(on_disk.turns.len(), 0);
841 }
842
843 #[test]
852 fn a_talk_prefers_the_chatter_role_over_the_planner_role() {
853 let (tmp, talks) = store();
854 let planner_spec = mock_agent(tmp.path(), BROKEN, BTreeMap::new());
855 let mut chatter_spec = mock_agent(tmp.path(), BROKEN, BTreeMap::new());
856 chatter_spec.id = "chatter-mock".to_owned();
857
858 let mut cfg = Config {
859 agents: vec![planner_spec.clone(), chatter_spec.clone()],
860 graph: Graph {
861 language: "en".to_owned(),
862 ..Graph::default()
863 },
864 ..Config::default()
865 };
866 cfg.roles.planner = Some(planner_spec.id.clone());
867 cfg.roles.chatter = Some(chatter_spec.id.clone());
868
869 let talk =
870 begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin with chatter set");
871 assert_eq!(talk.agent, chatter_spec.id, "chatter must win over planner");
872
873 cfg.roles.chatter = None;
874 let fallback =
875 begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin with chatter unset");
876 assert_eq!(
877 fallback.agent, planner_spec.id,
878 "unset chatter must fall back to planner, unchanged from before this role existed"
879 );
880 }
881
882 #[tokio::test]
883 async fn the_first_turn_carries_the_briefing_and_later_turns_do_not() {
884 let (tmp, talks) = store();
885 let spec = mock_agent(tmp.path(), ECHO, BTreeMap::new());
886 let cfg = config(spec);
887 let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
888
889 say(&mut talk, &talks, &cfg, "what does the queue module do?")
890 .await
891 .expect("first turn");
892 let first_prompt = &talk.turns[1].body;
893 assert!(first_prompt.contains("magi task add --solo"));
894 assert!(first_prompt.contains("what does the queue module do?"));
895
896 say(&mut talk, &talks, &cfg, "and how is it locked?")
897 .await
898 .expect("second turn");
899 let second_prompt = &talk.turns[3].body;
900 assert!(
901 !second_prompt.contains("magi task add --solo"),
902 "the briefing is sent once, not on every turn: {second_prompt}"
903 );
904 assert!(second_prompt.contains("and how is it locked?"));
905 }
906
907 #[tokio::test]
908 async fn say_appends_the_operator_turn_then_the_agent_turn() {
909 let (tmp, talks) = store();
910 let spec = mock_agent(tmp.path(), REPLY, env("go ahead"));
911 let cfg = config(spec);
912 let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
913
914 say(&mut talk, &talks, &cfg, "can I rename this function?")
915 .await
916 .expect("say");
917
918 assert_eq!(talk.turns.len(), 2);
919 assert_eq!(talk.turns[0].who, Who::Operator);
920 assert_eq!(talk.turns[0].body, "can I rename this function?");
921 assert_eq!(talk.turns[1].who, Who::Agent);
922 assert_eq!(talk.turns[1].body, "go ahead");
923 assert_eq!(talks.get(&talk.id).expect("get").turns, talk.turns);
924 }
925
926 #[tokio::test]
927 async fn a_failed_turn_keeps_the_operator_message_and_says_what_happened() {
928 let (tmp, talks) = store();
929 let spec = mock_agent(tmp.path(), BROKEN, BTreeMap::new());
930 let cfg = config(spec);
931 let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
932
933 let err = say(&mut talk, &talks, &cfg, "check the tests")
934 .await
935 .expect_err("a turn with no answer is an error");
936 assert!(err.to_string().contains("no answer"), "{err}");
937
938 let on_disk = talks.get(&talk.id).expect("get");
939 assert_eq!(on_disk.turns.len(), 2);
940 assert_eq!(on_disk.turns[0].body, "check the tests");
941 let note = &on_disk.turns[1];
942 assert_eq!(note.who, Who::Agent);
943 assert!(note.body.starts_with(MAGI_NOTE), "{}", note.body);
944 assert!(note.body.contains("your message is saved"));
945 }
946
947 #[test]
948 fn closing_is_idempotent_and_a_closed_talk_takes_no_more_turns() {
949 let (tmp, talks) = store();
950 let spec = mock_agent(tmp.path(), REPLY, env("hi"));
951 let cfg = config(spec);
952 let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
953
954 close(&mut talk, &talks).expect("close");
955 assert_eq!(talk.status, TalkStatus::Closed);
956 close(&mut talk, &talks).expect("closing twice is not an error");
957
958 let err = record(&mut talk, &talks, "still there?").expect_err("closed talks refuse");
959 assert!(err.to_string().contains("closed"));
960 let _ = &cfg; }
962
963 #[tokio::test]
964 async fn a_close_that_lands_while_a_turn_is_in_flight_is_not_undone_by_the_reply() {
965 let (tmp, talks) = store();
966 let spec = mock_agent(tmp.path(), REPLY, env("here you go"));
967 let cfg = config(spec);
968 let mut in_flight = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
971
972 let mut closed_elsewhere = talks.get(&in_flight.id).expect("reread");
976 close(&mut closed_elsewhere, &talks).expect("close");
977 assert_eq!(
978 talks.get(&in_flight.id).expect("reread").status,
979 TalkStatus::Closed,
980 "the close landed on disk before the turn finished"
981 );
982
983 assert_eq!(in_flight.status, TalkStatus::Open);
987 respond(&mut in_flight, &talks, &cfg, "one more question")
988 .await
989 .expect("the turn itself still completes");
990
991 let on_disk = talks.get(&in_flight.id).expect("reread");
992 assert_eq!(
993 on_disk.status,
994 TalkStatus::Closed,
995 "a close must stick even when a turn that started before it finishes after it"
996 );
997 assert!(
1000 on_disk.turns.iter().any(|t| t.body == "here you go"),
1001 "the in-flight turn's own reply is still recorded: {:?}",
1002 on_disk.turns
1003 );
1004 }
1005
1006 #[test]
1007 fn a_close_that_lands_before_record_is_called_is_not_undone_by_it() {
1008 let (tmp, talks) = store();
1009 let spec = mock_agent(tmp.path(), REPLY, env("hi"));
1010 let cfg = config(spec);
1011 let mut stale = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
1014
1015 let mut closed_elsewhere = talks.get(&stale.id).expect("reread");
1018 close(&mut closed_elsewhere, &talks).expect("close");
1019 assert_eq!(
1020 talks.get(&stale.id).expect("reread").status,
1021 TalkStatus::Closed,
1022 "the close landed on disk before record was called"
1023 );
1024
1025 assert_eq!(stale.status, TalkStatus::Open);
1029 let err = record(&mut stale, &talks, "still there?")
1030 .expect_err("a close that landed first must be honored, not overwritten");
1031 assert!(err.to_string().contains("closed"));
1032
1033 let on_disk = talks.get(&stale.id).expect("reread");
1034 assert_eq!(
1035 on_disk.status,
1036 TalkStatus::Closed,
1037 "record must not resurrect a conversation closed while its snapshot was stale"
1038 );
1039 assert!(
1040 on_disk.turns.is_empty(),
1041 "the rejected turn must not have been appended: {:?}",
1042 on_disk.turns
1043 );
1044 let _ = &cfg; }
1046
1047 #[test]
1048 fn close_blocks_on_records_guard_rather_than_interleaving_with_it() {
1049 let (tmp, talks) = store();
1050 let spec = mock_agent(tmp.path(), REPLY, env("hi"));
1051 let cfg = config(spec);
1052 let talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
1053
1054 let held = talks.guard();
1058
1059 let talks2 = talks.clone();
1060 let id = talk.id.clone();
1061 let closing = std::thread::spawn(move || {
1062 let mut talk = talks2.get(&id).expect("get");
1063 close(&mut talk, &talks2).expect("close");
1064 });
1065
1066 std::thread::sleep(Duration::from_millis(50));
1067 assert!(
1068 !closing.is_finished(),
1069 "close must wait for the guard, not read and write while it is held - \
1070 a re-read alone narrows this window without closing it"
1071 );
1072
1073 drop(held);
1074 closing.join().expect("close thread panicked");
1075
1076 assert_eq!(
1077 talks.get(&talk.id).expect("reread").status,
1078 TalkStatus::Closed,
1079 "once the guard is free, close still lands"
1080 );
1081 let _ = &cfg; }
1083
1084 #[test]
1085 fn reopening_a_closed_talk_lets_it_take_turns_again_and_reopening_twice_is_not_an_error() {
1086 let (tmp, talks) = store();
1087 let spec = mock_agent(tmp.path(), REPLY, env("hi"));
1088 let cfg = config(spec);
1089 let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
1090
1091 close(&mut talk, &talks).expect("close");
1092 assert_eq!(talk.status, TalkStatus::Closed);
1093
1094 reopen(&mut talk, &talks).expect("reopen");
1095 assert_eq!(talk.status, TalkStatus::Open);
1096 assert_eq!(
1097 talks.get(&talk.id).expect("reread").status,
1098 TalkStatus::Open
1099 );
1100
1101 reopen(&mut talk, &talks).expect("reopening an open talk is not an error");
1103 assert_eq!(talk.status, TalkStatus::Open);
1104
1105 record(&mut talk, &talks, "one more thing").expect("a reopened talk takes turns again");
1106 let _ = &cfg; }
1108
1109 #[test]
1110 fn removing_a_talk_deletes_its_record_and_artifacts_and_refuses_an_unknown_id() {
1111 let (tmp, talks) = store();
1112 let spec = mock_agent(tmp.path(), REPLY, env("hi"));
1113 let cfg = config(spec);
1114 let talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
1115
1116 let artifacts = talks.artifacts_of(&talk.id);
1117 std::fs::create_dir_all(&artifacts).expect("create artifacts dir");
1118 std::fs::write(artifacts.join("turn-1.txt"), "hello").expect("write artifact");
1119
1120 talks.remove(&talk.id).expect("remove");
1121 assert!(!talks.path_of(&talk.id).is_file(), "the record is gone");
1122 assert!(!artifacts.is_dir(), "the artifacts directory is gone");
1123 assert!(
1124 talks.get(&talk.id).is_err(),
1125 "a removed talk cannot be read back"
1126 );
1127
1128 let err = talks
1129 .remove("nonexistent-id")
1130 .expect_err("unknown id refused");
1131 assert!(err.to_string().contains("no talk matches"), "{err}");
1132 let _ = &cfg; }
1134
1135 #[tokio::test]
1136 async fn a_delete_that_lands_while_a_turn_is_in_flight_is_not_undone_by_the_reply() {
1137 let (tmp, talks) = store();
1138 let spec = mock_agent(tmp.path(), REPLY, env("here you go"));
1139 let cfg = config(spec);
1140 let mut in_flight = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
1143
1144 talks.remove(&in_flight.id).expect("remove");
1145 assert!(
1146 talks.get(&in_flight.id).is_err(),
1147 "the delete landed on disk before the turn finished"
1148 );
1149
1150 respond(&mut in_flight, &talks, &cfg, "one more question")
1153 .await
1154 .expect("the turn itself still completes rather than erroring");
1155
1156 assert!(
1157 talks.get(&in_flight.id).is_err(),
1158 "a delete must stick even when a turn that started before it finishes after it"
1159 );
1160 }
1161
1162 #[test]
1163 fn a_delete_that_lands_before_record_is_called_is_not_undone_by_it() {
1164 let (tmp, talks) = store();
1165 let spec = mock_agent(tmp.path(), REPLY, env("hi"));
1166 let cfg = config(spec);
1167 let mut stale = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
1170
1171 talks.remove(&stale.id).expect("remove");
1172
1173 let err = record(&mut stale, &talks, "still there?")
1177 .expect_err("a delete that landed first must be honored, not overwritten");
1178 assert!(err.to_string().contains("deleted"), "{err}");
1179
1180 assert!(
1181 talks.get(&stale.id).is_err(),
1182 "record must not resurrect a conversation deleted while its snapshot was stale"
1183 );
1184 let _ = &cfg; }
1186
1187 #[test]
1188 fn a_delete_that_lands_before_close_is_called_is_not_undone_by_it() {
1189 let (tmp, talks) = store();
1190 let spec = mock_agent(tmp.path(), REPLY, env("hi"));
1191 let cfg = config(spec);
1192 let mut stale = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
1195
1196 talks.remove(&stale.id).expect("remove");
1197
1198 let err = close(&mut stale, &talks)
1202 .expect_err("a delete that landed first must be honored, not overwritten");
1203 assert!(err.to_string().contains("deleted"), "{err}");
1204
1205 assert!(
1206 talks.get(&stale.id).is_err(),
1207 "close must not resurrect a conversation deleted while its snapshot was stale"
1208 );
1209 let _ = &cfg; }
1211
1212 #[test]
1213 fn a_delete_that_lands_before_reopen_is_called_is_not_undone_by_it() {
1214 let (tmp, talks) = store();
1215 let spec = mock_agent(tmp.path(), REPLY, env("hi"));
1216 let cfg = config(spec);
1217 let mut stale = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
1220 close(&mut stale, &talks).expect("close");
1221
1222 talks.remove(&stale.id).expect("remove");
1223
1224 let err = reopen(&mut stale, &talks)
1228 .expect_err("a delete that landed first must be honored, not overwritten");
1229 assert!(err.to_string().contains("deleted"), "{err}");
1230
1231 assert!(
1232 talks.get(&stale.id).is_err(),
1233 "reopen must not resurrect a conversation deleted while its snapshot was stale"
1234 );
1235 let _ = &cfg; }
1237
1238 #[test]
1239 fn list_puts_open_talks_before_closed_ones() {
1240 let (tmp, talks) = store();
1241 let make = |id: &str, status: TalkStatus| {
1242 let mut t = Talk {
1243 schema: SCHEMA,
1244 id: id.to_owned(),
1245 repo: tmp.path().to_owned(),
1246 agent: "mock".to_owned(),
1247 status,
1248 turns: Vec::new(),
1249 created_at: Timestamp::now(),
1250 updated_at: Timestamp::now(),
1251 seat: SeatState::new(SEAT, "mock", 7),
1252 };
1253 talks.put(&mut t).expect("put");
1254 };
1255 make("20260901-000000-0001", TalkStatus::Open);
1256 make("20260902-000000-0002", TalkStatus::Open);
1257 make("20260903-000000-0003", TalkStatus::Closed);
1258
1259 let ids: Vec<String> = talks.list().into_iter().map(|t| t.id).collect();
1260 assert_eq!(
1261 ids,
1262 [
1263 "20260902-000000-0002",
1264 "20260901-000000-0001",
1265 "20260903-000000-0003"
1266 ]
1267 );
1268 assert_eq!(talks.count_open(), 2);
1269 }
1270
1271 #[test]
1272 fn tasks_of_finds_only_this_talks_own_tasks() {
1273 let dir = tempfile::tempdir().expect("tempdir");
1274 let queue = Queue::at(dir.path().join("queue"));
1275
1276 let mut mine = Task::new(
1277 "rework the loader".to_owned(),
1278 "rework the loader".to_owned(),
1279 PathBuf::from("/repo"),
1280 Source::Agent {
1281 run: "20260904-014455-ab12".to_owned(),
1282 node: "chat".to_owned(),
1283 },
1284 );
1285 queue.put(&mut mine).expect("put mine");
1286
1287 let mut theirs = Task::new(
1288 "unrelated".to_owned(),
1289 "unrelated".to_owned(),
1290 PathBuf::from("/repo"),
1291 Source::Agent {
1292 run: "20260904-090000-zz99".to_owned(),
1293 node: "implement".to_owned(),
1294 },
1295 );
1296 queue.put(&mut theirs).expect("put theirs");
1297
1298 let mut human = Task::new(
1299 "typed by hand".to_owned(),
1300 "typed by hand".to_owned(),
1301 PathBuf::from("/repo"),
1302 Source::Human,
1303 );
1304 queue.put(&mut human).expect("put human");
1305
1306 let found = tasks_of(&queue, "20260904-014455-ab12");
1307 assert_eq!(found.len(), 1);
1308 assert_eq!(found[0].id, mine.id);
1309 }
1310
1311 #[test]
1312 fn the_briefing_names_solo_task_add_and_not_the_task_file_spec() {
1313 let brief = briefing(Path::new("/repo"), "en", false);
1314 assert!(brief.contains("magi task add --solo"));
1315 assert!(!brief.contains(plan::TASK_FILE_SPEC));
1316 assert!(brief.contains("/repo"));
1317 assert!(!brief.contains("Hold this conversation in"));
1318 }
1319
1320 #[test]
1321 fn the_briefing_names_the_language_when_it_is_not_english() {
1322 let brief = briefing(Path::new("/repo"), "Japanese", false);
1323 assert!(brief.contains("Hold this conversation in Japanese"));
1324 }
1325
1326 #[test]
1327 fn the_briefing_forbids_writes_unless_the_repository_opted_in() {
1328 let read_only = briefing(Path::new("/repo"), "en", false);
1329 assert!(read_only.contains("Do not write files"));
1330 assert!(!read_only.contains("allow_write"));
1331
1332 let writable = briefing(Path::new("/repo"), "en", true);
1333 assert!(!writable.contains("Do not write files"));
1334 assert!(writable.contains("allow_write = true"));
1335 assert!(writable.contains("magi task add --solo"));
1338 assert!(writable.contains("say plainly what you"));
1339 }
1340}