magi/
chat_notify.rs

1//! Telling a chat how a task it filed ended.
2//!
3//! A chat is single-turn: it speaks only when the operator writes, so "I will
4//! tell you when it finishes" cannot be kept by the agent. This module keeps
5//! it for the agent. When a task filed by a chat ([`Task::filed_by_chat`])
6//! reaches a final state, a fixed English notice is queued as a draft of the
7//! originating talk, and the talk's own turn machinery (its session, its
8//! lease) runs it. No new seat and no new waiter: the same shape as
9//! [`crate::consult`].
10//!
11//! - **A reconciler, not a hook.** [`sweep`] looks at the queue on every
12//!   daemon tick, so whoever put the task in its final state (the loop, `magi
13//!   task done`, the web UI, a restart) is covered the same way.
14//! - **Once.** The task records what was reported ([`Task::chat_notice`]); the
15//!   talk records the accepted key in the same write as the draft
16//!   ([`talk::queue_once_matching`]). A crash between the two writes is repaired by the
17//!   next sweep, which finds the key in the talk and only fixes the task.
18//!   Lock order is task, then talk.
19//! - **Closed or gone is skipped,** never reopened (unlike a consult).
20//! - **A parking upgrade starts nothing.** The draft stays durable.
21
22use std::path::Path;
23
24use anyhow::{Result, bail};
25use jiff::Timestamp;
26
27use crate::config::Config;
28use crate::prompt::{CHAT_TASK_NOTICE_HEADING, ChatTaskNotice, chat_task_notice};
29use crate::queue::{ChatNotice, ChatNoticeOutcome, Queue, Task, TaskStatus};
30use crate::talk::{self, Talks};
31
32/// Facts about a task's latest run, read tolerantly: an unreadable record
33/// only leaves them out of the notice.
34#[derive(Debug, Default, PartialEq, Eq)]
35pub struct RunFacts {
36    /// The pull request url, if the run opened one.
37    pub pr: Option<String>,
38    /// The winning candidate's summary, if there is one.
39    pub summary: Option<String>,
40}
41
42/// Read [`RunFacts`] from `<home>/runs/<id>/run.json`.
43pub fn run_facts(home: &Path, run: &str) -> RunFacts {
44    let Ok(body) = std::fs::read_to_string(home.join("runs").join(run).join("run.json")) else {
45        return RunFacts::default();
46    };
47    let Ok(v) = serde_json::from_str::<serde_json::Value>(&body) else {
48        return RunFacts::default();
49    };
50    let text = |x: Option<&serde_json::Value>| {
51        x.and_then(|x| x.as_str())
52            .map(str::trim)
53            .filter(|s| !s.is_empty())
54            .map(str::to_owned)
55    };
56    let winner = text(v.pointer("/tally/winner"));
57    let summary = v
58        .get("candidates")
59        .and_then(|c| c.as_array())
60        .and_then(|cs| {
61            cs.iter()
62                .find(|c| winner.is_some() && text(c.get("label")).as_deref() == winner.as_deref())
63        })
64        .and_then(|c| text(c.get("summary")));
65    RunFacts {
66        pr: text(v.pointer("/pr/url")),
67        summary,
68    }
69}
70
71/// The notice text for `task`, built from what magi recorded.
72pub fn notice_text(task: &Task, facts: &RunFacts) -> String {
73    let (status, reason) = match task.status {
74        TaskStatus::Done => ("done", None),
75        TaskStatus::Held => (
76            "held (needs a decision before it runs again)",
77            task.hold_reason.as_deref().or(task.last_error.as_deref()),
78        ),
79        TaskStatus::Blocked => ("blocked", task.block_reason.as_deref()),
80        _ => ("not final", None),
81    };
82    chat_task_notice(&ChatTaskNotice {
83        id: &task.id,
84        title: &task.title,
85        status,
86        reason,
87        run: task.runs.last().map(String::as_str),
88        pr: facts.pr.as_deref(),
89        summary: facts.summary.as_deref(),
90    })
91}
92
93/// Queue the notice for `task` into its talk, once. `Skipped` when the talk is
94/// gone or closed; an error only for a transient failure worth retrying.
95/// The boolean records whether a new draft was actually added.
96fn deliver(
97    talks: &Talks,
98    home: &Path,
99    task: &Task,
100    talk_id: &str,
101    key: &str,
102) -> Result<(ChatNoticeOutcome, bool)> {
103    let gone = || !talks.list().iter().any(|t| t.id == talk_id);
104    let mut talk = match talks.get(talk_id) {
105        Ok(t) => t,
106        Err(_) if gone() => return Ok((ChatNoticeOutcome::Skipped, false)),
107        Err(e) => return Err(e),
108    };
109    if !talk.status.open() {
110        return Ok((ChatNoticeOutcome::Skipped, false));
111    }
112    let facts = task
113        .runs
114        .last()
115        .map(|r| run_facts(home, r))
116        .unwrap_or_default();
117    let text = notice_text(task, &facts);
118    let prefix = format!("{}:", task.id);
119    let previous = |stored: &str| {
120        key == "done"
121            && stored
122                .strip_prefix(&prefix)
123                .is_some_and(crate::queue::is_done_notice_key)
124    };
125    match talk::queue_once_matching(
126        &mut talk,
127        talks,
128        &text,
129        &format!("{prefix}{key}"),
130        &previous,
131    ) {
132        Ok(added) => Ok((ChatNoticeOutcome::Posted, added)),
133        Err(e) => match talks.get(talk_id) {
134            Ok(t) if t.status.open() => Err(e),
135            Ok(_) => Ok((ChatNoticeOutcome::Skipped, false)),
136            Err(_) if gone() => Ok((ChatNoticeOutcome::Skipped, false)),
137            Err(_) => Err(e),
138        },
139    }
140}
141
142/// Report every chat-filed task whose final state has not been reported.
143/// Returns the talk ids that got a new draft.
144pub fn sweep(queue: &Queue, talks: &Talks, home: &Path) -> Vec<String> {
145    let mut posted = Vec::new();
146    for task in queue.list() {
147        if task.chat_notice_due().is_none() {
148            continue;
149        }
150        let result = queue.modify(&task.id, |t| {
151            // Re-judged under the task's lock, on the stored record.
152            let (Some(key), Some(talk_id)) =
153                (t.chat_notice_due(), t.filed_by_chat().map(str::to_owned))
154            else {
155                return false;
156            };
157            match deliver(talks, home, t, &talk_id, &key) {
158                Ok((outcome, added)) => {
159                    if added {
160                        posted.push(talk_id);
161                    }
162                    t.chat_notice = Some(ChatNotice {
163                        key,
164                        at: Timestamp::now(),
165                        outcome,
166                    });
167                    true
168                }
169                Err(e) => {
170                    tracing::warn!("chat notice for task {} not queued: {e:#}", t.short());
171                    false
172                }
173            }
174        });
175        if let Err(e) = result {
176            tracing::warn!("chat notice for task {}: {e:#}", task.short());
177        }
178    }
179    posted.sort();
180    posted.dedup();
181    posted
182}
183
184/// Open talks holding a completion notice in their draft. Only these are
185/// started on the daemon's own initiative: an operator's unsent draft is not.
186pub fn talks_to_start(talks: &Talks) -> Vec<String> {
187    talks
188        .list()
189        .into_iter()
190        .filter(|t| t.status.open() && pending_is_only_notices(t))
191        .map(|t| t.id)
192        .collect()
193}
194
195/// Is the draft made of completion notices alone? A notice appended to an
196/// operator's unsent draft must not start that draft on the daemon's initiative.
197fn pending_is_only_notices(t: &talk::Talk) -> bool {
198    let head = format!("# {CHAT_TASK_NOTICE_HEADING}\n");
199    let body = &t.pending;
200    if body.is_empty() || !t.pending_attachments.is_empty() {
201        return false;
202    }
203    let Some(cuts) = t.pending_breaks.as_deref() else {
204        return false;
205    };
206    let mut starts = vec![0];
207    starts.extend_from_slice(cuts);
208    starts
209        .iter()
210        .all(|&s| body.is_char_boundary(s) && body.get(s..).is_some_and(|m| m.starts_with(&head)))
211}
212
213/// Start the chat's turn for every talk owed a notice, under the talk's own
214/// lease, each spawned into `turns` (the daemon's in-flight set, which its exit
215/// waits for). `parking` (an upgrade is parking) starts
216/// nothing. Returns how many turns were started; a talk whose lease is held
217/// elsewhere is left to its holder, who drains the draft before letting go.
218pub fn start_turns(talks: &Talks, parking: bool, turns: &mut tokio::task::JoinSet<()>) -> usize {
219    if parking {
220        return 0;
221    }
222    let mut started = 0;
223    for id in talks_to_start(talks) {
224        let lease = match talks.claim_turn(&id) {
225            Ok(Some(lease)) => lease,
226            Ok(None) => continue,
227            Err(e) => {
228                tracing::warn!("chat {id}: turn lease: {e:#}");
229                continue;
230            }
231        };
232        started += 1;
233        let talks = talks.clone();
234        turns.spawn(async move {
235            let run = async {
236                let talk = talks.get(&id)?;
237                let (cfg, _) = Config::discover(&talk.repo, None)?;
238                crate::consult::drain_turns(&talks, &id, lease, &cfg).await
239            };
240            if let Err(e) = run.await {
241                tracing::warn!("chat {id}: notice turn failed: {e:#}");
242            }
243        });
244    }
245    started
246}
247
248/// The task a `magi talk post` comes from: `run` (the posting seat's run id)
249/// must belong to a task filed by exactly this talk.
250pub fn authorize_post(tasks: &[Task], run: &str, node: &str, talk_id: &str) -> Result<()> {
251    if run.trim().is_empty() {
252        bail!("MAGI_RUN is not set: `magi talk post` is for a task seat of the chat's own task");
253    }
254    if node.trim().is_empty() {
255        bail!("MAGI_NODE is not set: `magi talk post` is for a task seat of the chat's own task");
256    }
257    if node == crate::queue::CHAT_NODE {
258        bail!("a chat seat cannot post to its own transcript");
259    }
260    let task = tasks
261        .iter()
262        .find(|t| t.runs.iter().any(|r| r == run))
263        .ok_or_else(|| anyhow::anyhow!("run {run} belongs to no task"))?;
264    match task.filed_by_chat() {
265        Some(id) if id == talk_id => Ok(()),
266        _ => bail!("run {run} does not belong to a task filed from chat {talk_id}"),
267    }
268}
269
270#[cfg(test)]
271mod tests {
272    use super::*;
273    use crate::queue::Source;
274
275    #[test]
276    fn run_facts_reads_the_winners_summary_and_the_pr() {
277        let tmp = tempfile::tempdir().expect("tempdir");
278        let dir = tmp.path().join("runs").join("r1");
279        std::fs::create_dir_all(&dir).expect("dir");
280        std::fs::write(
281            dir.join("run.json"),
282            r#"{"candidates":[{"label":"A","summary":"first"},{"label":"B","summary":"second"}],
283                "tally":{"winner":"B"},"pr":{"url":"https://example.test/pull/7"}}"#,
284        )
285        .expect("write");
286        let f = run_facts(tmp.path(), "r1");
287        assert_eq!(f.summary.as_deref(), Some("second"));
288        assert_eq!(f.pr.as_deref(), Some("https://example.test/pull/7"));
289        assert_eq!(run_facts(tmp.path(), "missing"), RunFacts::default());
290    }
291
292    #[test]
293    fn the_notice_names_the_task_state_reason_and_links() {
294        let mut t = Task::new(
295            "fix it".to_owned(),
296            "fix it".to_owned(),
297            ".".into(),
298            Source::Agent {
299                run: "talk".to_owned(),
300                node: crate::queue::CHAT_NODE.to_owned(),
301            },
302        );
303        t.start("run-1".to_owned());
304        t.fail("the gate failed", 1);
305        let text = notice_text(
306            &t,
307            &RunFacts {
308                pr: Some("https://example.test/pull/7".to_owned()),
309                summary: None,
310            },
311        );
312        assert!(text.starts_with("# Update on a task you filed"));
313        for needle in [t.id.as_str(), "held", "the gate failed", "run-1", "pull/7"] {
314            assert!(text.contains(needle), "{needle}: {text}");
315        }
316    }
317
318    #[test]
319    fn a_post_needs_a_run_of_a_task_this_chat_filed() {
320        let mut t = Task::new(
321            "x".to_owned(),
322            "x".to_owned(),
323            ".".into(),
324            Source::Agent {
325                run: "talk-a".to_owned(),
326                node: crate::queue::CHAT_NODE.to_owned(),
327            },
328        );
329        t.start("run-1".to_owned());
330        let tasks = [t];
331        assert!(authorize_post(&tasks, "run-1", "implement", "talk-a").is_ok());
332        assert!(authorize_post(&tasks, "run-1", "implement", "talk-b").is_err());
333        assert!(authorize_post(&tasks, "", "implement", "talk-a").is_err());
334        assert!(authorize_post(&tasks, "run-1", "", "talk-a").is_err());
335        assert!(authorize_post(&tasks, "run-1", "chat", "talk-a").is_err());
336        assert!(authorize_post(&tasks, "run-2", "implement", "talk-a").is_err());
337    }
338}
magi/chat_notify.rs

magi/
chat_notify.rs
