magi-cli 0.10.0

Blind multi-agent implementation competition: N agents implement, M judges rank blind, deliberate, vote privately, winner survives double review + E2E gate
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
//! The standing conversation: a place to think out loud with an agent between
//! tasks, reachable from a phone.
//!
//! [`crate::chat`] is an interview with one purpose - arrive at a task file
//! and file it - and it ends the moment that happens. This module is the other
//! kind of conversation an operator wants: one that stays open. Ask a
//! question, have the agent read a file or run a command to check something,
//! talk through an idea, and when it is time to act, tell it to file the work
//! rather than do it here. The conversation does not end; it is what the
//! operator opens the next time something comes up.
//!
//! # Talking is not implementing
//!
//! Every turn here runs with `allow_write: false` by default - the same
//! restriction [`crate::chat`] puts on its own interview, for the same
//! reason: not security, but attribution. An agent that edits a checkout
//! mid-conversation leaves a diff that belongs to no run and passed no
//! review, and on a repository entered into magi's blind competition that
//! makes every candidate's diff unjudgeable. That is why the default holds
//! regardless of what a repository's own `magi.toml` says about anything
//! else. When the operator wants a change made, the agent is told to run
//! `magi task add --solo` ([`briefing`]) rather than reach for an editor: the
//! change goes through magi's own queue, on the repository's own terms, and
//! the operator can watch it happen instead of trusting that it did.
//!
//! `[talk] allow_write` ([`crate::config::Talk::allow_write`]) lets a
//! specific repository opt out of that default - a dotfiles or personal
//! config checkout that is never entered into a competition and never
//! reviewed has nothing for the restriction to protect, and filing a task for
//! a one-line edit there is pure overhead. Turning it on does not turn this
//! conversation into an implementer: [`briefing`] still sends everything
//! bigger than a small, operator-named edit to the queue, and still tells the
//! agent to say what it changed.
//!
//! `--solo` rather than a plain `magi task add` is the point of pairing this
//! module with [`crate::queue::Task::solo`]. A task that came out of a
//! conversation the operator just had is a decision already made, not a
//! design question worth three independent takes - so it runs through one
//! implementer and straight into review, the way [`crate::graph::Runner`]
//! already degrades a single-candidate run.
//!
//! # Shape
//!
//! The same split [`crate::chat`] and [`crate::queue`] use: [`Talk`] is data
//! plus pure helpers, [`Talks`] owns the I/O and is constructed with its root,
//! so every test here drives a real store in a temp directory rather than the
//! operator's own home.

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;

use anyhow::{Context, Result, bail};
use jiff::Timestamp;
use serde::{Deserialize, Serialize};

use crate::agent::{self, Invocation, SeatState};
use crate::config::Config;
use crate::plan;
use crate::queue::{Queue, Source, Task};

/// On-disk format for a conversation. Bumped when a field's meaning changes.
pub const SCHEMA: u32 = 1;

/// Wall-clock limit for one agent turn.
///
/// Fifteen minutes, three times [`crate::chat::TURN_TIMEOUT`]. A planning turn
/// answers a question about intent; a turn here is expected to run several
/// shell commands and read their output before answering one - "what does
/// this function do", "is this still true", "run the tests and tell me" - and
/// a five-minute budget cuts that off mid-investigation on exactly the
/// conversation meant to support it.
const TURN_TIMEOUT: Duration = Duration::from_secs(900);

/// Seat name for the conversation's agent, scoping its CLI-side session away
/// from every other seat magi ever opens - the same rule [`crate::chat`]
/// applies to its own interviewer.
const SEAT: &str = "talk";

/// Prefix on a turn magi wrote rather than an agent. See
/// [`crate::chat::MAGI_NOTE`], which this mirrors.
const MAGI_NOTE: &str = "magi: ";

/// Who said something.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Who {
    /// The operator.
    Operator,
    /// The conversation's agent - or magi itself, reporting that a turn
    /// failed. See [`MAGI_NOTE`].
    Agent,
}

/// One message in the conversation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Turn {
    /// Who wrote it.
    pub who: Who,
    /// What they said.
    pub body: String,
    /// When it was said.
    pub at: Timestamp,
}

/// Where a conversation is in its life. Unlike [`crate::chat::ChatStatus`]
/// there is no `filed`: this conversation can file any number of tasks
/// without ending, so it only ever moves once, from open to closed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum TalkStatus {
    /// Still open; the operator may say more, and may have already filed work
    /// out of it.
    Open,
    /// Closed by hand. Kept on disk as a record.
    Closed,
}

impl TalkStatus {
    /// Is this conversation still live?
    pub fn open(self) -> bool {
        matches!(self, Self::Open)
    }

    /// Wire form, for the phone and for logs.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Open => "open",
            Self::Closed => "closed",
        }
    }
}

/// One standing conversation.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Talk {
    /// On-disk format version.
    pub schema: u32,
    /// Conversation id, e.g. `20260904-014455-ab12`.
    pub id: String,
    /// Repository this conversation is about.
    pub repo: PathBuf,
    /// Roster agent id holding the conversation.
    pub agent: String,
    /// Current state.
    pub status: TalkStatus,
    /// Everything said, oldest first.
    pub turns: Vec<Turn>,
    /// When the conversation was opened.
    pub created_at: Timestamp,
    /// Last change to this file.
    pub updated_at: Timestamp,
    /// The CLI-side conversation, so a turn after the first costs one
    /// sentence instead of the whole transcript. Not `pub` for the same
    /// reason [`crate::chat::Chat`]'s is not: it is magi's bookkeeping, and a
    /// caller that edited it would detach the record from the conversation
    /// the model actually holds.
    seat: SeatState,
}

impl Talk {
    /// Short form used in lists and notifications, matching a run's short id.
    pub fn short(&self) -> &str {
        short(&self.id)
    }
}

/// A conversation store on disk.
#[derive(Debug, Clone)]
pub struct Talks {
    root: PathBuf,
    /// Serializes the read-modify-write cycle that reads a talk, decides
    /// something from its `status`, and writes the whole record back.
    /// [`close`], [`record`] and the tail of [`turn`] all take this before
    /// that cycle rather than after just the read: a re-read narrows the
    /// window another writer can land in, but does not close it, since
    /// nothing stopped that other writer's own put from landing between this
    /// call's re-read and its own put. Shared across every clone, since every
    /// clone is a handle onto the same files.
    lock: Arc<Mutex<()>>,
}

impl Talks {
    /// The operator's conversations, `<home>/talks`.
    pub fn open() -> Self {
        Self::at(crate::run::home().join("talks"))
    }

    /// A store at an explicit root. Tests use this, which is why none of them
    /// need the operator's real home.
    pub fn at(root: PathBuf) -> Self {
        Self {
            root,
            lock: Arc::new(Mutex::new(())),
        }
    }

    /// Claim the right to read-modify-write a talk's `status`. A plain
    /// `std::sync::Mutex`, not an async one: every caller holds it across a
    /// handful of small file operations and never across an `.await`, so
    /// blocking the thread briefly is the right tool, not a reason to reach
    /// for `tokio::sync::Mutex`. Poisoning recovers rather than propagates -
    /// one panicking caller must not wedge every talk in the store the way it
    /// would wedge the loop's own lock; see [`crate::web`]'s `lock_or_recover`,
    /// which this mirrors.
    fn guard(&self) -> MutexGuard<'_, ()> {
        self.lock.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Directory holding the conversation files.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Path for one conversation id.
    pub fn path_of(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.json"))
    }

    /// Where one conversation's prompts and CLI output are kept, beside the
    /// record rather than inside it - see [`crate::chat::Chats::artifacts_of`].
    pub fn artifacts_of(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.artifacts"))
    }

    /// Write a conversation, atomically, so a process killed mid-write leaves
    /// the previous state readable rather than a truncated file.
    pub fn put(&self, t: &mut Talk) -> Result<()> {
        std::fs::create_dir_all(&self.root)
            .with_context(|| format!("create {}", self.root.display()))?;
        t.updated_at = Timestamp::now();
        let body = serde_json::to_string_pretty(t).context("serialize talk")?;
        let path = self.path_of(&t.id);
        let tmp = path.with_extension("json.tmp");
        std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
        std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
        Ok(())
    }

    /// Load a conversation by id or unambiguous id prefix.
    pub fn get(&self, id: &str) -> Result<Talk> {
        let resolved = self.resolve_id(id)?;
        read_path(&self.path_of(&resolved))
    }

    /// Every conversation on disk: open first, then newest first - the same
    /// ordering [`crate::chat::Chats::list`] uses, for the same reason: what
    /// the operator is still using belongs above what they are done with.
    pub fn list(&self) -> Vec<Talk> {
        let mut all: Vec<Talk> = std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .map(|e| e.path())
            .filter(|p| p.extension().is_some_and(|x| x == "json"))
            .filter_map(|p| read_path(&p).ok())
            .collect();
        all.sort_unstable_by(|a, b| {
            let rank = |t: &Talk| u8::from(!t.status.open());
            rank(a).cmp(&rank(b)).then_with(|| b.id.cmp(&a.id))
        });
        all
    }

    /// Expand an id prefix to exactly one conversation id.
    pub fn resolve_id(&self, prefix: &str) -> Result<String> {
        if self.path_of(prefix).is_file() {
            return Ok(prefix.to_owned());
        }
        let hits: Vec<String> = self
            .list()
            .into_iter()
            .map(|t| t.id)
            .filter(|id| id.starts_with(prefix) || id.ends_with(prefix))
            .collect();
        match hits.len() {
            1 => Ok(hits.into_iter().next().expect("exactly one hit")),
            0 => bail!("no talk matches `{prefix}`"),
            _ => bail!(
                "`{prefix}` matches {} talks: {}",
                hits.len(),
                hits.join(", ")
            ),
        }
    }

    /// Change detection token, the same shape as
    /// [`crate::chat::Chats::revision`]: the newest modification time in the
    /// store, in milliseconds.
    pub fn revision(&self) -> u64 {
        std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .filter_map(|e| e.metadata().ok())
            .filter_map(|m| m.modified().ok())
            .filter_map(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_millis() as u64)
            .max()
            .unwrap_or(0)
    }

    /// How many conversations are still open.
    pub fn count_open(&self) -> usize {
        self.list().iter().filter(|t| t.status.open()).count()
    }

    /// Remove a conversation from disk, record and artifacts both. The
    /// operator's way of saying "not just done, gone" - [`close`] alone
    /// leaves the record as history.
    ///
    /// Takes [`Talks::guard`] for the same reason [`close`] does: a delete
    /// racing a [`record`] or the tail of [`turn`] must not land between
    /// their own read and write, or the file removed here would look, to
    /// them, like a record that simply has not been written yet. The other
    /// half of that story is on their side - both check under this same
    /// guard that the record they are about to write is still there, and
    /// give up without writing if it is not, which is what stops their `put`
    /// from resurrecting a conversation this call already removed.
    pub fn remove(&self, id: &str) -> Result<()> {
        let _guard = self.guard();
        let resolved = self.resolve_id(id)?;
        let path = self.path_of(&resolved);
        std::fs::remove_file(&path).with_context(|| format!("remove {}", path.display()))?;
        let artifacts = self.artifacts_of(&resolved);
        if artifacts.is_dir() {
            std::fs::remove_dir_all(&artifacts)
                .with_context(|| format!("remove {}", artifacts.display()))?;
        }
        Ok(())
    }
}

/// Open a conversation. Unlike [`crate::chat::start`] this takes no agent
/// turn: there is no idea to answer yet, and a conversation the operator has
/// not said anything into yet is a normal, valid thing to have sitting on the
/// phone.
///
/// `agent` beats `[roles] chatter`, which beats `[roles] planner` -
/// [`plan::pick`] run against the same preference order [`crate::chat::start`]
/// uses for its own resident conversation. This is a standing chat, not an
/// interview, so `chatter` rather than `planner` is the field this
/// conversation is actually about; `planner` remains the fallback so an
/// operator who never set `chatter` sees no change. `chatter` exists at all
/// because this conversation stays open far longer than a single planning
/// interview, and opening it against the same seat as a judge is what
/// produced the `agent ... did not answer within 300s` timeout that led to
/// splitting the two roles apart - see `[roles] chatter`'s own doc in
/// [`crate::config`].
pub fn begin(store: &Talks, cfg: &Config, repo: PathBuf, agent: Option<&str>) -> Result<Talk> {
    // Absolute, for the same reason `chat::start` canonicalizes: a relative
    // path means the wrong repository once anything other than this process
    // reads it back.
    let repo = repo.canonicalize().unwrap_or(repo);
    let want = agent
        .or(cfg.roles.chatter.as_deref())
        .or(cfg.roles.planner.as_deref());
    let spec = plan::pick(&cfg.agents, want, &plan::installed)?;

    let now = Timestamp::now();
    let mut talk = Talk {
        schema: SCHEMA,
        id: new_id(),
        repo,
        agent: spec.id.clone(),
        status: TalkStatus::Open,
        turns: Vec::new(),
        created_at: now,
        updated_at: now,
        seat: SeatState::new(SEAT, &spec.id, crate::rng::entropy()),
    };
    store.put(&mut talk)?;
    Ok(talk)
}

/// Append the operator's turn and flush it, without invoking anything.
///
/// Split out of [`say`] for the same reason [`crate::chat::record`] is split
/// out of [`crate::chat::say`]: `POST /api/talks/{id}/say` answers once the
/// message is safely on disk, and runs the agent's half in the background -
/// see that function's doc for why holding the connection for a turn that can
/// run fifteen minutes is the wrong shape for a phone.
pub fn record(talk: &mut Talk, store: &Talks, text: &str) -> Result<String> {
    // `web::talk_say` reads the talk, then awaits config discovery before
    // calling this - a gap a concurrent `POST /api/talks/{id}/close` can land
    // in. The guard held for the rest of this function is what actually closes
    // that gap: re-reading status without it only shrinks the window a
    // concurrent `close` could land in between this call's own read and its
    // `put`, it does not remove it. See [`Talks::guard`] and the matching
    // guard in `turn`, which this mirrors.
    let _guard = store.guard();
    // A concurrent `Talks::remove` can have landed in that same gap. `put`
    // writes unconditionally, so trusting the stale `talk` here would recreate
    // the file a delete just removed - the record must still be there for a
    // turn to have anywhere to append to.
    let Ok(fresh) = store.get(&talk.id) else {
        bail!("talk {} was deleted", talk.short());
    };
    talk.status = fresh.status;
    if !talk.status.open() {
        bail!(
            "talk {} is {} and takes no more turns",
            talk.short(),
            talk.status.as_str()
        );
    }
    let text = text.trim();
    if text.is_empty() {
        bail!("nothing to say");
    }
    talk.turns.push(Turn {
        who: Who::Operator,
        body: text.to_owned(),
        at: Timestamp::now(),
    });
    store.put(talk)?;
    Ok(text.to_owned())
}

/// One operator turn and one agent turn, appended - the synchronous form, used
/// by tests and by anything that is fine waiting out the turn itself.
pub async fn say(talk: &mut Talk, store: &Talks, cfg: &Config, text: &str) -> Result<()> {
    let text = record(talk, store, text)?;
    turn(talk, store, cfg, &text).await
}

/// The agent's half of a turn: invoke, append, flush. Pairs with [`record`],
/// the same way [`crate::chat::respond`] pairs with [`crate::chat::record`].
pub async fn respond(talk: &mut Talk, store: &Talks, cfg: &Config, text: &str) -> Result<()> {
    turn(talk, store, cfg, text).await
}

/// Close a conversation. Idempotent: closing an already-closed conversation is
/// not an error, since the operator's intent - "I am done with this" - is
/// already satisfied.
///
/// Re-reads the record under [`Talks::guard`] rather than trusting the
/// caller's copy of `talk`, and writes that fresh copy back rather than the
/// one passed in. `web::talk_close` loads `talk` and calls this right after
/// with no gap of its own, but without the guard that load can still land
/// between a `record` or `turn` elsewhere reading the file and writing it
/// back - and a close built on the older snapshot would put it right back,
/// silently dropping whatever turn the other call had just appended.
///
/// If the re-read fails, this errors rather than falling back to the
/// caller's stale copy: `talk::begin` always `put`s the record before handing
/// out a `Talk`, so the only way a re-read can fail is a concurrent
/// [`Talks::remove`] having deleted it, and writing the stale copy back would
/// resurrect exactly what that delete removed.
pub fn close(talk: &mut Talk, store: &Talks) -> Result<()> {
    let _guard = store.guard();
    let mut fresh = store
        .get(&talk.id)
        .with_context(|| format!("talk {} was deleted", talk.short()))?;
    fresh.status = TalkStatus::Closed;
    store.put(&mut fresh)?;
    *talk = fresh;
    Ok(())
}

/// Reopen a closed conversation. Idempotent for the same reason [`close`] is:
/// reopening an already-open conversation is not an error, since the
/// operator's intent - "I want to keep talking about this" - is already
/// satisfied.
///
/// Written symmetrically with [`close`]: re-reads the record under
/// [`Talks::guard`] rather than trusting the caller's copy of `talk`, writes
/// that fresh copy back rather than the one passed in, and errors rather than
/// falling back to the stale copy if the re-read fails, for the same reasons
/// `close`'s doc gives.
pub fn reopen(talk: &mut Talk, store: &Talks) -> Result<()> {
    let _guard = store.guard();
    let mut fresh = store
        .get(&talk.id)
        .with_context(|| format!("talk {} was deleted", talk.short()))?;
    fresh.status = TalkStatus::Open;
    store.put(&mut fresh)?;
    *talk = fresh;
    Ok(())
}

/// Invoke the conversation's agent once and append what it said.
///
/// The first turn ever taken carries the full [`briefing`], because nothing
/// else has told the agent what this conversation is or what it may do.
/// Every turn after that behaves like [`crate::chat`]'s: resend nothing when
/// the CLI can resume its own session, and fall back to [`transcript`] only
/// when it cannot.
async fn turn(talk: &mut Talk, store: &Talks, cfg: &Config, text: &str) -> Result<()> {
    let spec = cfg
        .agents
        .iter()
        .find(|a| a.id == talk.agent)
        .with_context(|| {
            format!(
                "talk {} was opened with agent `{}`, which is no longer in \
                 the roster; restore it in magi.toml or start a new \
                 conversation",
                talk.short(),
                talk.agent
            )
        })?;

    let resuming = agent::has_session(spec.kind, &talk.seat, cfg.graph.sessions);
    let body = if talk.seat.turns == 0 {
        format!(
            "{}\n\n# Operator\n\n{text}",
            briefing(&talk.repo, &cfg.graph.language, cfg.talk.allow_write)
        )
    } else if resuming {
        text.to_owned()
    } else {
        format!("{}\n\n{text}", transcript(talk))
    };

    let artifacts = store.artifacts_of(&talk.id);
    let stem = format!("turn-{}", talk.seat.turns + 1);
    // The chat's build cache is the same shared one the graph's seats get, so
    // a conversation that compiles does not mint another multi-GB target dir.
    let cache_dir = cfg.cache_dir();
    let inv = Invocation {
        cwd: &talk.repo,
        prompt: &body,
        timeout: TURN_TIMEOUT,
        // Off unless this repository's own config opts in - see
        // `crate::config::Talk::allow_write` and this module's doc for why
        // the default keeps a conversational edit from landing in a checkout
        // no run or review can claim.
        allow_write: cfg.talk.allow_write,
        sessions: cfg.graph.sessions,
        artifacts: &artifacts,
        stem: &stem,
        // The conversation's own id, so `magi task add` run from inside it is
        // attributed to this conversation - see `Source::Agent`.
        run: &talk.id,
        node: "chat",
        cache_dir: cache_dir.as_deref(),
    };

    let outcome = agent::invoke(spec, &mut talk.seat, &inv).await;
    let note = |why: String| Turn {
        who: Who::Agent,
        body: format!("{MAGI_NOTE}{why}"),
        at: Timestamp::now(),
    };
    let (reply, failure) = match outcome {
        Err(e) => (
            note(format!("could not run agent `{}`: {e}", talk.agent)),
            Some(format!("could not run agent `{}`: {e}", talk.agent)),
        ),
        Ok(out) if out.quota_exhausted() => {
            let reset = out
                .quota
                .as_ref()
                .and_then(|q| q.reset.clone())
                .map_or_else(String::new, |r| format!(" (resets {r})"));
            let why = format!(
                "agent `{}` is out of quota{reset}; your message is saved, so \
                 say it again when the window reopens",
                talk.agent
            );
            (note(why.clone()), Some(why))
        }
        Ok(out) if out.timed_out => {
            let why = format!(
                "agent `{}` did not answer within {}s; your message is saved",
                talk.agent,
                TURN_TIMEOUT.as_secs()
            );
            (note(why.clone()), Some(why))
        }
        Ok(out) if !out.usable() => {
            let why = format!(
                "agent `{}` produced no answer (exit {}); your message is saved",
                talk.agent,
                out.exit_code
                    .map_or_else(|| "unknown".to_owned(), |c| c.to_string())
            );
            (note(why.clone()), Some(why))
        }
        Ok(out) => (
            Turn {
                who: Who::Agent,
                body: out.text.trim().to_owned(),
                at: Timestamp::now(),
            },
            None,
        ),
    };

    // A close landed on disk while this turn was in flight is read back here
    // rather than trusted from the snapshot this call started with. `store`
    // holds nothing else this function does not itself own - the turn guard
    // in `web::Ui::begin_talk_turn` keeps `turns` and `seat` this call's
    // alone to mutate - but `status` is not behind that guard, and an
    // operator's close must stick: the whole point of ending a conversation
    // is that an agent's answer to the last message before the close cannot
    // silently reopen it. The guard is what makes that read-then-write
    // section atomic with `close`'s own - taken only for this tail and not
    // for the whole invocation above, so one talk's fifteen-minute turn does
    // not block another talk's close from proceeding.
    let _guard = store.guard();
    // A delete is the more final version of that same race: `put` writes
    // unconditionally, so a talk removed while this turn was in flight must
    // stay removed rather than being written back with this turn's reply
    // appended to it. The reply is simply given up on - there is no
    // conversation left for it to belong to.
    let Ok(fresh) = store.get(&talk.id) else {
        return Ok(());
    };
    talk.status = fresh.status;
    talk.turns.push(reply);
    store.put(talk)?;

    match failure {
        Some(why) => bail!("{why}"),
        None => Ok(()),
    }
}

/// Everything said so far, as prose, for a CLI that cannot resume its own
/// conversation. See [`crate::chat::transcript`], which this mirrors.
fn transcript(talk: &Talk) -> String {
    let mut out = String::from(
        "This conversation cannot resume on the CLI's side, so here is \
         everything said so far; answer only the last message.\n",
    );
    for t in &talk.turns {
        let who = match t.who {
            Who::Operator => "operator",
            Who::Agent => "you",
        };
        out.push_str(&format!("\n## {who}\n\n{}\n", t.body.trim()));
    }
    out
}

/// The briefing the agent opens with, sent once as part of its first turn.
///
/// Pure, so the properties that matter can be asserted without an interview:
/// it names `magi task add --solo` (the route this conversation always has to
/// changing anything) and it does not carry
/// [`crate::plan::TASK_FILE_SPEC`] - that spec describes a task *file*, which
/// belongs to the planning interview and would tell this agent to write one
/// here instead of filing through the queue. `allow_write` only ever adds an
/// extra permission on top of that; it never removes the queue as an option,
/// which is why both branches keep the same `# When the operator wants
/// something done` section - `write_policy` is the only part that changes.
pub fn briefing(repo: &Path, language: &str, allow_write: bool) -> String {
    let write_policy = if allow_write {
        "This repository has set `[talk] allow_write = true`, so you may \
         write files here - but only a small, already-decided edit the \
         operator names outright in this conversation, not an \
         implementation. Once you have made it, say plainly what you \
         edited. Anything bigger, or anything still open-ended, still goes \
         through the queue below rather than being done here."
    } else {
        "Do not write files. Implementing a change is not this \
         conversation's job; a separate, blind competition of agents does \
         that, and a repository this conversation has already edited would \
         make their diffs unjudgeable."
    };
    let mut out = format!(
        "You are magi's standing conversation partner for its operator, who \
         usually has this open on a phone. Keep replies short: no preamble, \
         no restating what they just said.\n\n\
         # Repository\n\n{repo}\n\n\
         You may look around: read files, run shell commands, search history, \
         run tests - whatever answers the question. {write_policy}\n\n\
         # When the operator wants something done\n\n\
         Run:\n\n\
         magi task add --solo --repo {repo} <instruction>\n\n\
         and tell the operator the task id it prints, so they can follow it \
         from the Queue. Write <instruction> so that an implementer who has \
         never seen this conversation can act on it alone - it is everything \
         they get. Use --solo: it runs the task through one implementer \
         straight into review instead of the usual multi-agent competition, \
         which is the right shape for a change this conversation has already \
         settled, rather than one still worth several independent takes.\n",
        repo = repo.display(),
    );
    out.push_str(&language_note(language));
    out
}

/// The operator is talking, so their language matters here more than in most
/// prompts magi sends - see [`crate::chat::language_note`], which this
/// mirrors.
fn language_note(language: &str) -> String {
    if language.trim().is_empty() || language.eq_ignore_ascii_case("en") {
        String::new()
    } else {
        format!("\nHold this conversation in {language}.\n")
    }
}

/// Queue tasks this conversation has filed, oldest first.
///
/// A task is this conversation's when its [`Source::Agent`] names this
/// conversation's id as `run` - which is exactly what happens when
/// `magi task add` is run from inside a turn, because [`turn`] passes the
/// conversation's own id as [`Invocation::run`].
pub fn tasks_of(queue: &Queue, talk_id: &str) -> Vec<Task> {
    let mut tasks: Vec<Task> = queue
        .list()
        .into_iter()
        .filter(|t| matches!(&t.source, Source::Agent { run, .. } if run == talk_id))
        .collect();
    tasks.sort_unstable_by(|a, b| a.id.cmp(&b.id));
    tasks
}

fn read_path(path: &Path) -> Result<Talk> {
    let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
    serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))
}

fn short(id: &str) -> &str {
    id.split('-').next_back().unwrap_or(id)
}

fn new_id() -> String {
    let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
    let seed = crate::rng::entropy();
    format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use crate::config::{AgentKind, AgentSpec, Graph};
    use crate::queue::{Queue, Source, Task};

    use super::*;

    /// A store of its own, with no process-global state.
    fn store() -> (tempfile::TempDir, Talks) {
        let tmp = tempfile::tempdir().expect("tempdir");
        let talks = Talks::at(tmp.path().join("talks"));
        (tmp, talks)
    }

    /// A `kind = "command"` agent whose whole behaviour is a POSIX shell
    /// script - see `chat`'s tests for why no test here may spawn a real
    /// agent CLI.
    fn mock_agent(dir: &Path, script: &str, env: BTreeMap<String, String>) -> AgentSpec {
        let path = dir.join("mock-talk-agent.sh");
        std::fs::write(&path, script).expect("write mock");
        AgentSpec {
            id: "mock".to_owned(),
            kind: AgentKind::Command,
            model: None,
            command: vec!["sh".to_owned(), path.to_string_lossy().into_owned()],
            extra_args: Vec::new(),
            env,
            prompt_delivery: None,
        }
    }

    fn config(spec: AgentSpec) -> Config {
        Config {
            agents: vec![spec],
            graph: Graph {
                language: "en".to_owned(),
                ..Graph::default()
            },
            ..Config::default()
        }
    }

    /// Echo a canned reply, ignoring the prompt on stdin.
    const REPLY: &str = "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' \"$MOCK_REPLY\"\n";

    /// Say nothing and fail, the way a CLI that cannot start does.
    const BROKEN: &str = "#!/bin/sh\ncat >/dev/null\nexit 3\n";

    /// Reply with the prompt it was given, so a test can inspect exactly what
    /// the agent received on stdin.
    const ECHO: &str = "#!/bin/sh\ncat\n";

    fn env(reply: &str) -> BTreeMap<String, String> {
        BTreeMap::from([("MOCK_REPLY".to_owned(), reply.to_owned())])
    }

    #[test]
    fn the_frozen_json_field_names_round_trip_through_disk() {
        let (tmp, talks) = store();
        let mut talk = Talk {
            schema: SCHEMA,
            id: "20260904-014455-ab12".to_owned(),
            repo: tmp.path().to_owned(),
            agent: "sonnet".to_owned(),
            status: TalkStatus::Open,
            turns: Vec::new(),
            created_at: Timestamp::now(),
            updated_at: Timestamp::now(),
            seat: SeatState::new(SEAT, "sonnet", 7),
        };
        talks.put(&mut talk).expect("put");

        let raw = std::fs::read_to_string(talks.path_of(&talk.id)).expect("read back");
        let v: serde_json::Value = serde_json::from_str(&raw).expect("parse");
        for field in [
            "schema",
            "id",
            "repo",
            "agent",
            "status",
            "turns",
            "created_at",
            "updated_at",
        ] {
            assert!(v.get(field).is_some(), "missing field `{field}`");
        }
        assert_eq!(v["schema"], 1);
        assert_eq!(v["status"], "open");

        let back = talks.get(&talk.id).expect("get");
        assert_eq!(back.id, talk.id);
        assert_eq!(back.status, TalkStatus::Open);
    }

    #[test]
    fn opening_a_talk_takes_no_agent_turn() {
        let (tmp, talks) = store();
        // A script that would fail loudly if it were ever run: `begin` must
        // not invoke anything, since there is nothing yet for an agent to
        // answer.
        let spec = mock_agent(tmp.path(), BROKEN, BTreeMap::new());
        let cfg = config(spec);

        let talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
        assert_eq!(talk.status, TalkStatus::Open);
        assert!(talk.turns.is_empty(), "nothing has been said yet");

        let on_disk = talks.get(&talk.id).expect("get");
        assert_eq!(on_disk.turns.len(), 0);
    }

    /// `roles.chatter`, not `roles.planner`, decides who holds this
    /// conversation - the same distinction [`crate::chat::start`] makes for
    /// its own resident chat, and for the same reason: a Talk stays open far
    /// longer than a `magi plan` interview, and opening it against the same
    /// seat as a judge is what produced the `agent ... did not answer within
    /// 300s` timeout `[roles] chatter` exists to avoid. See
    /// `a_chat_prefers_the_chatter_role_over_the_planner_role` in
    /// `src/chat.rs`, which this mirrors.
    #[test]
    fn a_talk_prefers_the_chatter_role_over_the_planner_role() {
        let (tmp, talks) = store();
        let planner_spec = mock_agent(tmp.path(), BROKEN, BTreeMap::new());
        let mut chatter_spec = mock_agent(tmp.path(), BROKEN, BTreeMap::new());
        chatter_spec.id = "chatter-mock".to_owned();

        let mut cfg = Config {
            agents: vec![planner_spec.clone(), chatter_spec.clone()],
            graph: Graph {
                language: "en".to_owned(),
                ..Graph::default()
            },
            ..Config::default()
        };
        cfg.roles.planner = Some(planner_spec.id.clone());
        cfg.roles.chatter = Some(chatter_spec.id.clone());

        let talk =
            begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin with chatter set");
        assert_eq!(talk.agent, chatter_spec.id, "chatter must win over planner");

        cfg.roles.chatter = None;
        let fallback =
            begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin with chatter unset");
        assert_eq!(
            fallback.agent, planner_spec.id,
            "unset chatter must fall back to planner, unchanged from before this role existed"
        );
    }

    #[tokio::test]
    async fn the_first_turn_carries_the_briefing_and_later_turns_do_not() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), ECHO, BTreeMap::new());
        let cfg = config(spec);
        let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        say(&mut talk, &talks, &cfg, "what does the queue module do?")
            .await
            .expect("first turn");
        let first_prompt = &talk.turns[1].body;
        assert!(first_prompt.contains("magi task add --solo"));
        assert!(first_prompt.contains("what does the queue module do?"));

        say(&mut talk, &talks, &cfg, "and how is it locked?")
            .await
            .expect("second turn");
        let second_prompt = &talk.turns[3].body;
        assert!(
            !second_prompt.contains("magi task add --solo"),
            "the briefing is sent once, not on every turn: {second_prompt}"
        );
        assert!(second_prompt.contains("and how is it locked?"));
    }

    #[tokio::test]
    async fn say_appends_the_operator_turn_then_the_agent_turn() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("go ahead"));
        let cfg = config(spec);
        let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        say(&mut talk, &talks, &cfg, "can I rename this function?")
            .await
            .expect("say");

        assert_eq!(talk.turns.len(), 2);
        assert_eq!(talk.turns[0].who, Who::Operator);
        assert_eq!(talk.turns[0].body, "can I rename this function?");
        assert_eq!(talk.turns[1].who, Who::Agent);
        assert_eq!(talk.turns[1].body, "go ahead");
        assert_eq!(talks.get(&talk.id).expect("get").turns, talk.turns);
    }

    #[tokio::test]
    async fn a_failed_turn_keeps_the_operator_message_and_says_what_happened() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), BROKEN, BTreeMap::new());
        let cfg = config(spec);
        let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        let err = say(&mut talk, &talks, &cfg, "check the tests")
            .await
            .expect_err("a turn with no answer is an error");
        assert!(err.to_string().contains("no answer"), "{err}");

        let on_disk = talks.get(&talk.id).expect("get");
        assert_eq!(on_disk.turns.len(), 2);
        assert_eq!(on_disk.turns[0].body, "check the tests");
        let note = &on_disk.turns[1];
        assert_eq!(note.who, Who::Agent);
        assert!(note.body.starts_with(MAGI_NOTE), "{}", note.body);
        assert!(note.body.contains("your message is saved"));
    }

    #[test]
    fn closing_is_idempotent_and_a_closed_talk_takes_no_more_turns() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("hi"));
        let cfg = config(spec);
        let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        close(&mut talk, &talks).expect("close");
        assert_eq!(talk.status, TalkStatus::Closed);
        close(&mut talk, &talks).expect("closing twice is not an error");

        let err = record(&mut talk, &talks, "still there?").expect_err("closed talks refuse");
        assert!(err.to_string().contains("closed"));
        let _ = &cfg; // config kept only to build the agent above
    }

    #[tokio::test]
    async fn a_close_that_lands_while_a_turn_is_in_flight_is_not_undone_by_the_reply() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("here you go"));
        let cfg = config(spec);
        // The in-flight turn's own handle: loaded once, the way a spawned
        // background task in `web::talk_say` holds one for the whole turn.
        let mut in_flight = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        // The operator closes the conversation through a *different* handle
        // while the turn above is still running - exactly what a close typed
        // on the phone while an agent is mid-answer looks like.
        let mut closed_elsewhere = talks.get(&in_flight.id).expect("reread");
        close(&mut closed_elsewhere, &talks).expect("close");
        assert_eq!(
            talks.get(&in_flight.id).expect("reread").status,
            TalkStatus::Closed,
            "the close landed on disk before the turn finished"
        );

        // The turn's own handle still says `open` - it was loaded before the
        // close - and finishing it must not resurrect the conversation the
        // operator already ended.
        assert_eq!(in_flight.status, TalkStatus::Open);
        respond(&mut in_flight, &talks, &cfg, "one more question")
            .await
            .expect("the turn itself still completes");

        let on_disk = talks.get(&in_flight.id).expect("reread");
        assert_eq!(
            on_disk.status,
            TalkStatus::Closed,
            "a close must stick even when a turn that started before it finishes after it"
        );
        // The reply is not lost either: a turn already in flight when the
        // operator closed still gets its answer recorded.
        assert!(
            on_disk.turns.iter().any(|t| t.body == "here you go"),
            "the in-flight turn's own reply is still recorded: {:?}",
            on_disk.turns
        );
    }

    #[test]
    fn a_close_that_lands_before_record_is_called_is_not_undone_by_it() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("hi"));
        let cfg = config(spec);
        // The handle `web::talk_say` would have read before awaiting config
        // discovery, then carried across that await into `record`.
        let mut stale = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        // The operator closes the conversation through a *different* handle
        // in the gap between that read and the call to `record` below.
        let mut closed_elsewhere = talks.get(&stale.id).expect("reread");
        close(&mut closed_elsewhere, &talks).expect("close");
        assert_eq!(
            talks.get(&stale.id).expect("reread").status,
            TalkStatus::Closed,
            "the close landed on disk before record was called"
        );

        // The stale handle still says `open` - it was loaded before the
        // close - so a `record` that trusted it would append a turn and
        // write the conversation back open, undoing the close.
        assert_eq!(stale.status, TalkStatus::Open);
        let err = record(&mut stale, &talks, "still there?")
            .expect_err("a close that landed first must be honored, not overwritten");
        assert!(err.to_string().contains("closed"));

        let on_disk = talks.get(&stale.id).expect("reread");
        assert_eq!(
            on_disk.status,
            TalkStatus::Closed,
            "record must not resurrect a conversation closed while its snapshot was stale"
        );
        assert!(
            on_disk.turns.is_empty(),
            "the rejected turn must not have been appended: {:?}",
            on_disk.turns
        );
        let _ = &cfg; // config kept only to build the agent above
    }

    #[test]
    fn close_blocks_on_records_guard_rather_than_interleaving_with_it() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("hi"));
        let cfg = config(spec);
        let talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        // Hold the same guard `record`'s read-modify-write section holds for
        // the whole of its own read-then-write, standing in for `record`
        // being paused between its read and its `put`.
        let held = talks.guard();

        let talks2 = talks.clone();
        let id = talk.id.clone();
        let closing = std::thread::spawn(move || {
            let mut talk = talks2.get(&id).expect("get");
            close(&mut talk, &talks2).expect("close");
        });

        std::thread::sleep(Duration::from_millis(50));
        assert!(
            !closing.is_finished(),
            "close must wait for the guard, not read and write while it is held - \
             a re-read alone narrows this window without closing it"
        );

        drop(held);
        closing.join().expect("close thread panicked");

        assert_eq!(
            talks.get(&talk.id).expect("reread").status,
            TalkStatus::Closed,
            "once the guard is free, close still lands"
        );
        let _ = &cfg; // config kept only to build the agent above
    }

    #[test]
    fn reopening_a_closed_talk_lets_it_take_turns_again_and_reopening_twice_is_not_an_error() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("hi"));
        let cfg = config(spec);
        let mut talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        close(&mut talk, &talks).expect("close");
        assert_eq!(talk.status, TalkStatus::Closed);

        reopen(&mut talk, &talks).expect("reopen");
        assert_eq!(talk.status, TalkStatus::Open);
        assert_eq!(
            talks.get(&talk.id).expect("reread").status,
            TalkStatus::Open
        );

        // Idempotent: reopening an already-open talk is not an error.
        reopen(&mut talk, &talks).expect("reopening an open talk is not an error");
        assert_eq!(talk.status, TalkStatus::Open);

        record(&mut talk, &talks, "one more thing").expect("a reopened talk takes turns again");
        let _ = &cfg; // config kept only to build the agent above
    }

    #[test]
    fn removing_a_talk_deletes_its_record_and_artifacts_and_refuses_an_unknown_id() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("hi"));
        let cfg = config(spec);
        let talk = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        let artifacts = talks.artifacts_of(&talk.id);
        std::fs::create_dir_all(&artifacts).expect("create artifacts dir");
        std::fs::write(artifacts.join("turn-1.txt"), "hello").expect("write artifact");

        talks.remove(&talk.id).expect("remove");
        assert!(!talks.path_of(&talk.id).is_file(), "the record is gone");
        assert!(!artifacts.is_dir(), "the artifacts directory is gone");
        assert!(
            talks.get(&talk.id).is_err(),
            "a removed talk cannot be read back"
        );

        let err = talks
            .remove("nonexistent-id")
            .expect_err("unknown id refused");
        assert!(err.to_string().contains("no talk matches"), "{err}");
        let _ = &cfg; // config kept only to build the agent above
    }

    #[tokio::test]
    async fn a_delete_that_lands_while_a_turn_is_in_flight_is_not_undone_by_the_reply() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("here you go"));
        let cfg = config(spec);
        // The in-flight turn's own handle, loaded before the delete lands -
        // the same shape as the matching close test above.
        let mut in_flight = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        talks.remove(&in_flight.id).expect("remove");
        assert!(
            talks.get(&in_flight.id).is_err(),
            "the delete landed on disk before the turn finished"
        );

        // The turn's own handle has no way to know the record is gone -
        // finishing it must not write the file back into existence.
        respond(&mut in_flight, &talks, &cfg, "one more question")
            .await
            .expect("the turn itself still completes rather than erroring");

        assert!(
            talks.get(&in_flight.id).is_err(),
            "a delete must stick even when a turn that started before it finishes after it"
        );
    }

    #[test]
    fn a_delete_that_lands_before_record_is_called_is_not_undone_by_it() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("hi"));
        let cfg = config(spec);
        // The handle `web::talk_say` would have read before awaiting config
        // discovery, then carried across that await into `record`.
        let mut stale = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        talks.remove(&stale.id).expect("remove");

        // The stale handle has no way to know the record is gone - a
        // `record` that trusted it would append a turn and write the
        // conversation back into existence.
        let err = record(&mut stale, &talks, "still there?")
            .expect_err("a delete that landed first must be honored, not overwritten");
        assert!(err.to_string().contains("deleted"), "{err}");

        assert!(
            talks.get(&stale.id).is_err(),
            "record must not resurrect a conversation deleted while its snapshot was stale"
        );
        let _ = &cfg; // config kept only to build the agent above
    }

    #[test]
    fn a_delete_that_lands_before_close_is_called_is_not_undone_by_it() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("hi"));
        let cfg = config(spec);
        // `web::talk_close` loads `talk` and calls `close` right after - this
        // stands in for a delete landing in that gap.
        let mut stale = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");

        talks.remove(&stale.id).expect("remove");

        // The stale handle has no way to know the record is gone - a `close`
        // that fell back to it would write the conversation back into
        // existence, closed.
        let err = close(&mut stale, &talks)
            .expect_err("a delete that landed first must be honored, not overwritten");
        assert!(err.to_string().contains("deleted"), "{err}");

        assert!(
            talks.get(&stale.id).is_err(),
            "close must not resurrect a conversation deleted while its snapshot was stale"
        );
        let _ = &cfg; // config kept only to build the agent above
    }

    #[test]
    fn a_delete_that_lands_before_reopen_is_called_is_not_undone_by_it() {
        let (tmp, talks) = store();
        let spec = mock_agent(tmp.path(), REPLY, env("hi"));
        let cfg = config(spec);
        // `web::talk_reopen` loads `talk` and calls `reopen` right after -
        // this stands in for a delete landing in that gap.
        let mut stale = begin(&talks, &cfg, tmp.path().to_owned(), None).expect("begin");
        close(&mut stale, &talks).expect("close");

        talks.remove(&stale.id).expect("remove");

        // The stale handle has no way to know the record is gone - a
        // `reopen` that fell back to it would write the conversation back
        // into existence, open.
        let err = reopen(&mut stale, &talks)
            .expect_err("a delete that landed first must be honored, not overwritten");
        assert!(err.to_string().contains("deleted"), "{err}");

        assert!(
            talks.get(&stale.id).is_err(),
            "reopen must not resurrect a conversation deleted while its snapshot was stale"
        );
        let _ = &cfg; // config kept only to build the agent above
    }

    #[test]
    fn list_puts_open_talks_before_closed_ones() {
        let (tmp, talks) = store();
        let make = |id: &str, status: TalkStatus| {
            let mut t = Talk {
                schema: SCHEMA,
                id: id.to_owned(),
                repo: tmp.path().to_owned(),
                agent: "mock".to_owned(),
                status,
                turns: Vec::new(),
                created_at: Timestamp::now(),
                updated_at: Timestamp::now(),
                seat: SeatState::new(SEAT, "mock", 7),
            };
            talks.put(&mut t).expect("put");
        };
        make("20260901-000000-0001", TalkStatus::Open);
        make("20260902-000000-0002", TalkStatus::Open);
        make("20260903-000000-0003", TalkStatus::Closed);

        let ids: Vec<String> = talks.list().into_iter().map(|t| t.id).collect();
        assert_eq!(
            ids,
            [
                "20260902-000000-0002",
                "20260901-000000-0001",
                "20260903-000000-0003"
            ]
        );
        assert_eq!(talks.count_open(), 2);
    }

    #[test]
    fn tasks_of_finds_only_this_talks_own_tasks() {
        let dir = tempfile::tempdir().expect("tempdir");
        let queue = Queue::at(dir.path().join("queue"));

        let mut mine = Task::new(
            "rework the loader".to_owned(),
            "rework the loader".to_owned(),
            PathBuf::from("/repo"),
            Source::Agent {
                run: "20260904-014455-ab12".to_owned(),
                node: "chat".to_owned(),
            },
        );
        queue.put(&mut mine).expect("put mine");

        let mut theirs = Task::new(
            "unrelated".to_owned(),
            "unrelated".to_owned(),
            PathBuf::from("/repo"),
            Source::Agent {
                run: "20260904-090000-zz99".to_owned(),
                node: "implement".to_owned(),
            },
        );
        queue.put(&mut theirs).expect("put theirs");

        let mut human = Task::new(
            "typed by hand".to_owned(),
            "typed by hand".to_owned(),
            PathBuf::from("/repo"),
            Source::Human,
        );
        queue.put(&mut human).expect("put human");

        let found = tasks_of(&queue, "20260904-014455-ab12");
        assert_eq!(found.len(), 1);
        assert_eq!(found[0].id, mine.id);
    }

    #[test]
    fn the_briefing_names_solo_task_add_and_not_the_task_file_spec() {
        let brief = briefing(Path::new("/repo"), "en", false);
        assert!(brief.contains("magi task add --solo"));
        assert!(!brief.contains(plan::TASK_FILE_SPEC));
        assert!(brief.contains("/repo"));
        assert!(!brief.contains("Hold this conversation in"));
    }

    #[test]
    fn the_briefing_names_the_language_when_it_is_not_english() {
        let brief = briefing(Path::new("/repo"), "Japanese", false);
        assert!(brief.contains("Hold this conversation in Japanese"));
    }

    #[test]
    fn the_briefing_forbids_writes_unless_the_repository_opted_in() {
        let read_only = briefing(Path::new("/repo"), "en", false);
        assert!(read_only.contains("Do not write files"));
        assert!(!read_only.contains("allow_write"));

        let writable = briefing(Path::new("/repo"), "en", true);
        assert!(!writable.contains("Do not write files"));
        assert!(writable.contains("allow_write = true"));
        // Still names the queue for anything past a small named edit, and
        // still tells the agent to report what it changed.
        assert!(writable.contains("magi task add --solo"));
        assert!(writable.contains("say plainly what you"));
    }
}