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`]). 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.
95fn deliver(
96    talks: &Talks,
97    home: &Path,
98    task: &Task,
99    talk_id: &str,
100    key: &str,
101) -> Result<ChatNoticeOutcome> {
102    let gone = || !talks.list().iter().any(|t| t.id == talk_id);
103    let mut talk = match talks.get(talk_id) {
104        Ok(t) => t,
105        Err(_) if gone() => return Ok(ChatNoticeOutcome::Skipped),
106        Err(e) => return Err(e),
107    };
108    if !talk.status.open() {
109        return Ok(ChatNoticeOutcome::Skipped);
110    }
111    let facts = task
112        .runs
113        .last()
114        .map(|r| run_facts(home, r))
115        .unwrap_or_default();
116    let text = notice_text(task, &facts);
117    match talk::queue_once(&mut talk, talks, &text, &format!("{}:{key}", task.id)) {
118        Ok(_) => Ok(ChatNoticeOutcome::Posted),
119        Err(e) => match talks.get(talk_id) {
120            Ok(t) if t.status.open() => Err(e),
121            Ok(_) => Ok(ChatNoticeOutcome::Skipped),
122            Err(_) if gone() => Ok(ChatNoticeOutcome::Skipped),
123            Err(_) => Err(e),
124        },
125    }
126}
127
128/// Report every chat-filed task whose final state has not been reported.
129/// Returns the talk ids that got a new draft.
130pub fn sweep(queue: &Queue, talks: &Talks, home: &Path) -> Vec<String> {
131    let mut posted = Vec::new();
132    for task in queue.list() {
133        if task.chat_notice_due().is_none() {
134            continue;
135        }
136        let result = queue.modify(&task.id, |t| {
137            // Re-judged under the task's lock, on the stored record.
138            let (Some(key), Some(talk_id)) =
139                (t.chat_notice_due(), t.filed_by_chat().map(str::to_owned))
140            else {
141                return false;
142            };
143            match deliver(talks, home, t, &talk_id, &key) {
144                Ok(outcome) => {
145                    if outcome == ChatNoticeOutcome::Posted {
146                        posted.push(talk_id);
147                    }
148                    t.chat_notice = Some(ChatNotice {
149                        key,
150                        at: Timestamp::now(),
151                        outcome,
152                    });
153                    true
154                }
155                Err(e) => {
156                    tracing::warn!("chat notice for task {} not queued: {e:#}", t.short());
157                    false
158                }
159            }
160        });
161        if let Err(e) = result {
162            tracing::warn!("chat notice for task {}: {e:#}", task.short());
163        }
164    }
165    posted.sort();
166    posted.dedup();
167    posted
168}
169
170/// Open talks holding a completion notice in their draft. Only these are
171/// started on the daemon's own initiative: an operator's unsent draft is not.
172pub fn talks_to_start(talks: &Talks) -> Vec<String> {
173    talks
174        .list()
175        .into_iter()
176        .filter(|t| t.status.open() && pending_is_only_notices(t))
177        .map(|t| t.id)
178        .collect()
179}
180
181/// Is the draft made of completion notices alone? A notice appended to an
182/// operator's unsent draft must not start that draft on the daemon's initiative.
183fn pending_is_only_notices(t: &talk::Talk) -> bool {
184    let head = format!("# {CHAT_TASK_NOTICE_HEADING}\n");
185    let body = &t.pending;
186    if body.is_empty() || !t.pending_attachments.is_empty() {
187        return false;
188    }
189    let Some(cuts) = t.pending_breaks.as_deref() else {
190        return false;
191    };
192    let mut starts = vec![0];
193    starts.extend_from_slice(cuts);
194    starts
195        .iter()
196        .all(|&s| body.is_char_boundary(s) && body.get(s..).is_some_and(|m| m.starts_with(&head)))
197}
198
199/// Start the chat's turn for every talk owed a notice, under the talk's own
200/// lease, each spawned into `turns` (the daemon's in-flight set, which its exit
201/// waits for). `parking` (an upgrade is parking) starts
202/// nothing. Returns how many turns were started; a talk whose lease is held
203/// elsewhere is left to its holder, who drains the draft before letting go.
204pub fn start_turns(talks: &Talks, parking: bool, turns: &mut tokio::task::JoinSet<()>) -> usize {
205    if parking {
206        return 0;
207    }
208    let mut started = 0;
209    for id in talks_to_start(talks) {
210        let lease = match talks.claim_turn(&id) {
211            Ok(Some(lease)) => lease,
212            Ok(None) => continue,
213            Err(e) => {
214                tracing::warn!("chat {id}: turn lease: {e:#}");
215                continue;
216            }
217        };
218        started += 1;
219        let talks = talks.clone();
220        turns.spawn(async move {
221            let run = async {
222                let talk = talks.get(&id)?;
223                let (cfg, _) = Config::discover(&talk.repo, None)?;
224                crate::consult::drain_turns(&talks, &id, lease, &cfg).await
225            };
226            if let Err(e) = run.await {
227                tracing::warn!("chat {id}: notice turn failed: {e:#}");
228            }
229        });
230    }
231    started
232}
233
234/// The task a `magi talk post` comes from: `run` (the posting seat's run id)
235/// must belong to a task filed by exactly this talk.
236pub fn authorize_post(tasks: &[Task], run: &str, node: &str, talk_id: &str) -> Result<()> {
237    if run.trim().is_empty() {
238        bail!("MAGI_RUN is not set: `magi talk post` is for a task seat of the chat's own task");
239    }
240    if node.trim().is_empty() {
241        bail!("MAGI_NODE is not set: `magi talk post` is for a task seat of the chat's own task");
242    }
243    if node == crate::queue::CHAT_NODE {
244        bail!("a chat seat cannot post to its own transcript");
245    }
246    let task = tasks
247        .iter()
248        .find(|t| t.runs.iter().any(|r| r == run))
249        .ok_or_else(|| anyhow::anyhow!("run {run} belongs to no task"))?;
250    match task.filed_by_chat() {
251        Some(id) if id == talk_id => Ok(()),
252        _ => bail!("run {run} does not belong to a task filed from chat {talk_id}"),
253    }
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259    use crate::queue::Source;
260
261    #[test]
262    fn run_facts_reads_the_winners_summary_and_the_pr() {
263        let tmp = tempfile::tempdir().expect("tempdir");
264        let dir = tmp.path().join("runs").join("r1");
265        std::fs::create_dir_all(&dir).expect("dir");
266        std::fs::write(
267            dir.join("run.json"),
268            r#"{"candidates":[{"label":"A","summary":"first"},{"label":"B","summary":"second"}],
269                "tally":{"winner":"B"},"pr":{"url":"https://example.test/pull/7"}}"#,
270        )
271        .expect("write");
272        let f = run_facts(tmp.path(), "r1");
273        assert_eq!(f.summary.as_deref(), Some("second"));
274        assert_eq!(f.pr.as_deref(), Some("https://example.test/pull/7"));
275        assert_eq!(run_facts(tmp.path(), "missing"), RunFacts::default());
276    }
277
278    #[test]
279    fn the_notice_names_the_task_state_reason_and_links() {
280        let mut t = Task::new(
281            "fix it".to_owned(),
282            "fix it".to_owned(),
283            ".".into(),
284            Source::Agent {
285                run: "talk".to_owned(),
286                node: crate::queue::CHAT_NODE.to_owned(),
287            },
288        );
289        t.start("run-1".to_owned());
290        t.fail("the gate failed", 1);
291        let text = notice_text(
292            &t,
293            &RunFacts {
294                pr: Some("https://example.test/pull/7".to_owned()),
295                summary: None,
296            },
297        );
298        assert!(text.starts_with("# Update on a task you filed"));
299        for needle in [t.id.as_str(), "held", "the gate failed", "run-1", "pull/7"] {
300            assert!(text.contains(needle), "{needle}: {text}");
301        }
302    }
303
304    #[test]
305    fn a_post_needs_a_run_of_a_task_this_chat_filed() {
306        let mut t = Task::new(
307            "x".to_owned(),
308            "x".to_owned(),
309            ".".into(),
310            Source::Agent {
311                run: "talk-a".to_owned(),
312                node: crate::queue::CHAT_NODE.to_owned(),
313            },
314        );
315        t.start("run-1".to_owned());
316        let tasks = [t];
317        assert!(authorize_post(&tasks, "run-1", "implement", "talk-a").is_ok());
318        assert!(authorize_post(&tasks, "run-1", "implement", "talk-b").is_err());
319        assert!(authorize_post(&tasks, "", "implement", "talk-a").is_err());
320        assert!(authorize_post(&tasks, "run-1", "", "talk-a").is_err());
321        assert!(authorize_post(&tasks, "run-1", "chat", "talk-a").is_err());
322        assert!(authorize_post(&tasks, "run-2", "implement", "talk-a").is_err());
323    }
324}
magi/chat_notify.rs

magi/
chat_notify.rs
