Skip to main content

magi/
consult.rs

1//! Handing an open question to the chat conversation its task came from.
2//!
3//! A task filed by the standing chat (`Source::Agent` with
4//! [`crate::queue::CHAT_NODE`], whose `run` is the conversation's id) can raise
5//! a question the owner would rather talk through where the task started. The
6//! owner passes it on; the chat agent either answers it with `magi answer` or
7//! puts the decision points to the owner in the conversation.
8//!
9//! Three rules, each with a reason:
10//!
11//! - **One place decides.** [`origin_talk`] is the only judge of "does this
12//!   question have a chat to ask"; the web view and both entry points use it.
13//!   [`associated_talk`] resolves the same association for navigation even
14//!   when the question can no longer be consulted.
15//! - **Not a choice.** The hand-over never goes into `Question::choices` and
16//!   never touches `Question::answer`: the question stays `Open`, its
17//!   validation and the choice actions stay as they were.
18//! - **No new seat, no new waiter.** The text is queued as a draft of the
19//!   existing talk, so the chat's own turn machinery (its session, its turn
20//!   gate) runs it exactly as if the owner had typed it.
21
22use anyhow::{Context, Result, bail};
23use jiff::Timestamp;
24
25use crate::ask::{ChatConsult, Question, Questions};
26use crate::config::Config;
27use crate::queue::{CHAT_NODE, Task};
28use crate::talk::{self, Talk, Talks};
29
30/// The conversation `q` may be handed to, if any.
31///
32/// Requires an open question whose task was filed from a chat. The chat may
33/// be closed: [`begin`] reopens it. Release notices are left out. Merge
34/// approvals may be discussed, but answering from chat requires the owner's latest words.
35pub fn origin_talk(tasks: &[Task], talks: &[Talk], q: &Question) -> Option<Talk> {
36    if !q.status.open() || q.node == crate::bump::NOTICE_NODE {
37        return None;
38    }
39    associated_talk(tasks, talks, q)
40}
41
42/// The existing conversation associated with a question, regardless of whether
43/// the question is still eligible for consultation.
44pub fn associated_talk(tasks: &[Task], talks: &[Talk], q: &Question) -> Option<Talk> {
45    let task = crate::daemon::task_of_question(tasks, q)?;
46    let run = chat_talk_of(tasks, task)?;
47    talks.iter().find(|t| t.id == run).cloned()
48}
49
50/// The talk `start` descends from: the id recorded on a task
51/// ([`Task::chat_talk`]) wins, so deleted ancestors do not matter; tasks
52/// written before it existed are resolved by walking the ancestry.
53///
54/// Walk the provenance of `start` to the task a chat filed.
55///
56/// A follow-up goes to the task its merged run served: `FollowUp::origin_task`
57/// when recorded (a missing one ends the walk - guessing another task could
58/// hand the question to an unrelated chat), else the task whose `runs` hold
59/// `FollowUp::run`. At most `MAX_FOLLOWUP_GENERATION + 1` tasks are looked at,
60/// each once, so a cycle ends in `None`.
61pub fn chat_talk_of(tasks: &[Task], start: &Task) -> Option<String> {
62    let mut seen = std::collections::HashSet::new();
63    let mut cur = start;
64    for _ in 0..=crate::followup::MAX_FOLLOWUP_GENERATION {
65        if !seen.insert(cur.id.as_str()) {
66            return None;
67        }
68        if let Some(id) = cur.chat_talk() {
69            return Some(id.to_owned());
70        }
71        let f = cur.followup.as_ref()?;
72        cur = match &f.origin_task {
73            Some(id) => tasks.iter().find(|t| &t.id == id)?,
74            None => tasks.iter().find(|t| t.runs.contains(&f.run))?,
75        };
76    }
77    None
78}
79
80/// Is any question handed to talk `talk_id` still open?
81///
82/// Decided from the question store, not from the newest turn's wording: the
83/// owner's decision usually arrives in a later turn that carries no hand-over
84/// heading, and `magi answer` still has to be able to write then. Answered or
85/// abandoned questions stop counting, so the turn goes back to read-only.
86pub fn pending_consults(questions: &Questions, talk_id: &str) -> bool {
87    questions
88        .list()
89        .iter()
90        .any(|q| q.status.open() && q.consult.as_ref().is_some_and(|c| c.talk == talk_id))
91}
92
93/// Record the hand-over and queue the question as the chat's next message.
94///
95/// Returns `false` (and changes nothing) when the question was already handed
96/// over. The question is not otherwise touched: still `Open`, no thread turn.
97/// When queueing fails the record is withdrawn, so the owner can try again.
98pub fn begin(questions: &Questions, talks: &Talks, q: &Question, talk: &Talk) -> Result<bool> {
99    let (q, fresh) = questions.update(&q.id, |r| {
100        if !r.status.open() {
101            bail!("question {} is already {}", r.short(), r.status.as_str());
102        }
103        if r.consult.is_some() {
104            return Ok(false);
105        }
106        r.consult = Some(ChatConsult {
107            talk: talk.id.clone(),
108            at: Timestamp::now(),
109        });
110        Ok(true)
111    })?;
112    if !fresh {
113        return Ok(false);
114    }
115    // Serialize dispatch with answers, but never wait for conversation locks
116    // under the expiring question lock. Each retry reloads the latest question.
117    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
118    let dispatch = loop {
119        let mut busy = false;
120        let updated = questions.update(&q.id, |r| {
121            if !r.status.open() || r.consult != q.consult {
122                return Ok(false);
123            }
124            if talk::try_queue_consult(
125                talks,
126                &talk.id,
127                &crate::prompt::chat_consult(r),
128                Some(crate::prompt::chat_consult_display(r)),
129            )?
130            .is_none()
131            {
132                busy = true;
133                bail!("the conversation store is locked by another process");
134            }
135            Ok(true)
136        });
137        if busy && std::time::Instant::now() < deadline {
138            std::thread::sleep(std::time::Duration::from_millis(10));
139            continue;
140        }
141        break updated;
142    };
143    if let Err(e) = dispatch {
144        let _ = questions.update(&q.id, |r| {
145            if r.status.open() && r.consult == q.consult {
146                r.consult = None;
147            }
148            Ok(())
149        });
150        return Err(e).context("queue the question into the chat");
151    }
152    Ok(dispatch?.1)
153}
154
155/// Refuse a chat approval answer whose task the operator held, as
156/// `ask_settle_cmd` does for a deputy: such a hold is lifted by the operator's
157/// button. The task is the one `q.run` names, directly or as one of its runs;
158/// none found (an unreadable task is not listed) does not refuse. The caller
159/// reads the queue before taking the question lock, so a hold placed in between
160/// is missed for that one call.
161pub fn refuse_operator_held(q: &Question, node: &str, tasks: &[crate::queue::Task]) -> Result<()> {
162    if node != CHAT_NODE || QuestionKind::of(q) != QuestionKind::Merge {
163        return Ok(());
164    }
165    let release_run = if q.run.is_empty() && q.node == crate::bump::NOTICE_NODE {
166        crate::run::try_home()
167            .and_then(|home| crate::release_watch::state_for_question(&home, &q.id))
168            .and_then(|state| state.run)
169    } else {
170        None
171    };
172    let run = release_run.as_deref().unwrap_or(&q.run);
173    if let Some(task) = tasks
174        .iter()
175        .find(|t| t.id == run || t.runs.iter().any(|id| id == run))
176        && task.operator_held()
177    {
178        bail!(
179            "task {} is held by the operator; an answer cannot settle it",
180            task.short()
181        );
182    }
183    Ok(())
184}
185
186/// Authorization policy is selected by the question's node, never its reply.
187#[derive(Clone, Copy, PartialEq, Eq)]
188pub enum QuestionKind {
189    /// A question without a guarded public decision.
190    Ordinary,
191    /// Approval to merge a change.
192    Merge,
193    /// Confirmation of the text to publish.
194    Posting,
195}
196
197/// Owner evidence retained with an answer, independent of its consultation destination.
198#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
199pub struct AnswerAuthority {
200    /// Conversation containing the instruction.
201    pub talk: String,
202    /// Zero-based persisted turn position.
203    pub turn: usize,
204    /// Zero-based send position within that turn.
205    pub message: usize,
206    /// Full instruction, not a substring selected by the assistant.
207    pub instruction: String,
208    /// Stable fingerprint of the complete instruction.
209    pub instruction_fingerprint: String,
210    /// Immediately preceding persona reply, for a batch instruction.
211    pub recommendation_turn: Option<usize>,
212    /// Fingerprint of the complete persona reply used as the batch manifest.
213    pub recommendation_fingerprint: Option<String>,
214    /// Exact decision covered by this evidence.
215    pub question: String,
216    /// Exact offered choice covered by the instruction.
217    pub choice: String,
218    /// Execution facts validated when this answer was authorized.
219    #[serde(default)]
220    pub question_fingerprint: String,
221    /// The conversation agent interpreted this complete owner send as consent
222    /// to the single persisted internal recovery recommendation.
223    #[serde(default, skip_serializing_if = "Option::is_none")]
224    pub recovery_interpretation: Option<String>,
225}
226
227/// Immutable recommendation-time binding to question choices and targets.
228#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
229pub struct RecommendationManifest {
230    /// Fingerprint of the full persona reply.
231    pub body: String,
232    /// Full question IDs and fingerprints of their execution-relevant facts.
233    pub questions: std::collections::BTreeMap<String, String>,
234    /// Canonical choice and structured effect captured with the persona reply.
235    /// Old transcripts cannot acquire this binding retrospectively.
236    #[serde(default)]
237    pub recovery: std::collections::BTreeMap<String, RecoveryRecommendation>,
238}
239
240/// Execution binding published with a single internal recovery recommendation.
241#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
242pub struct RecoveryRecommendation {
243    /// Canonical stored label, independent of any alias used in the reply.
244    pub choice: String,
245    /// Structured effect; absent only for issuer-defined legacy keep-held.
246    pub action: Option<crate::ask::ChoiceAction>,
247    /// Run owning the question at recommendation time.
248    pub run: String,
249}
250
251/// Fingerprint of execution-relevant question facts, independent of answer state.
252pub fn question_fingerprint(q: &Question) -> String {
253    let facts = serde_json::json!({
254        "id": q.id, "run": q.run, "node": q.node, "seat": q.seat,
255        "summary": q.summary, "detail": q.detail, "choices": q.choices,
256        "actions": q.actions, "merge_target": q.merge_target,
257        "posting_comparison": q.posting_comparison,
258        "decision_context": q.decision_context, "triage_generation": q.triage_generation,
259    });
260    crate::github_text::fingerprint("", &facts.to_string())
261}
262
263/// Replacement title explicitly supplied with its question/content binding.
264/// The final posting gate still vets this literal text before publication.
265pub fn owner_title<'a>(q: &Question, instruction: &'a str) -> Option<&'a str> {
266    if instruction.starts_with("    ") || instruction.starts_with('\t') {
267        return None;
268    }
269    let (command, title) = instruction.trim().split_once('\n')?;
270    let expected = format!(
271        "{}: {} {}",
272        q.id,
273        crate::github_text::USE_MY_TEXT,
274        question_fingerprint(q)
275    );
276    (command == expected && !title.trim().is_empty() && !title.contains('\n'))
277        .then_some(title.trim())
278}
279
280/// Capture at publication, never reconstruct new authority from old history.
281pub fn recommendation_manifest(body: &str, questions: &Questions) -> RecommendationManifest {
282    let snapshot = questions.list();
283    RecommendationManifest {
284        body: crate::github_text::fingerprint("", body),
285        recovery: snapshot
286            .iter()
287            .filter_map(|q| {
288                let choice = recommended_choice(body, q, questions)?;
289                (q.status.open() && contextual_recovery(q, choice)).then(|| {
290                    (
291                        q.id.clone(),
292                        RecoveryRecommendation {
293                            choice: choice.into(),
294                            action: q.actions.get(choice).cloned(),
295                            run: q.run.clone(),
296                        },
297                    )
298                })
299            })
300            .collect(),
301        questions: snapshot
302            .iter()
303            .filter(|q| q.status.open() && recommended_choice(body, q, questions).is_some())
304            .map(|q| (q.id.clone(), question_fingerprint(q)))
305            .collect(),
306    }
307}
308
309/// Validate an answer under the question lock. The caller must also retain
310/// [`Talks::answer_guard`] from before this read until the answer is persisted.
311/// Consultation is a delivery route, not the source of authority for an
312/// explicit, independently bound owner decision.
313pub fn authorize_answer(
314    q: &Question,
315    talks: &Talks,
316    questions: &Questions,
317    run: &str,
318    node: &str,
319    reply: &str,
320    quote: Option<&str>,
321) -> Result<Option<AnswerAuthority>> {
322    if node != CHAT_NODE {
323        return Ok(None);
324    }
325    if reply == crate::land::HOLD && q.actions.contains_key(reply) {
326        bail!("hold cannot execute a question action");
327    }
328    let talk = talks.get(run)?;
329    if reply != crate::land::HOLD
330        && current_instruction(&talk).is_some_and(|(_, _, words)| {
331            words.starts_with("    ")
332                || words.starts_with('\t')
333                || words.trim_start().starts_with("```")
334                || words.trim_start().starts_with('>')
335        })
336    {
337        bail!("quoted or code-block owner text is not an explicit instruction for this action");
338    }
339    if let Some(evidence) = choice_authority(q, questions, &talk, reply, quote) {
340        validate_effect(
341            q,
342            reply,
343            &evidence.instruction,
344            quote.unwrap_or_default(),
345            true,
346        )?;
347        // A running original consultation can still be preparing a decision.
348        // Holds are safe to interrupt; other choices wait for that turn to stop.
349        if reply != crate::land::HOLD
350            && let Some(consult) = &q.consult
351            && consult.talk != run
352            && talks.turn_held(&consult.talk)
353        {
354            bail!("original consultation is active; wait for its turn to finish before answering");
355        }
356        return Ok(Some(evidence));
357    }
358    if current_instruction(&talk).is_some_and(|(_, _, words)| batch_instruction(words)) {
359        validate_effect(
360            q,
361            reply,
362            current_instruction(&talk).unwrap().2,
363            quote.unwrap_or_default(),
364            true,
365        )?;
366        bail!(
367            "batch authorization requires a current, unambiguous recommendation for this exact offered choice and unchanged question target/content"
368        );
369    }
370    // Preserve already supported consulted-chat answers. Independent evidence
371    // is required when consultation delivery cannot bind this conversation.
372    if q.consult.as_ref().is_some_and(|c| c.talk == run) {
373        validate_answer(q, talks, run, node, reply, quote)?;
374        if QuestionKind::of(q) == QuestionKind::Ordinary && consent(q, reply) != Consent::Delegable
375        {
376            let (_, _, instruction) = current_instruction(&talk)
377                .context("this action requires a current explicit owner instruction")?;
378            validate_effect(q, reply, instruction, quote.unwrap_or_default(), false)?;
379        }
380        return Ok(None);
381    }
382    if let Some((_, _, instruction)) = current_instruction(&talk) {
383        validate_effect(q, reply, instruction, quote.unwrap_or_default(), true)?;
384    }
385    bail!("{}", diagnose_authorization_failure(q, &talk, reply, quote))
386}
387
388/// Explain a refused binding without changing the authorization policy.
389fn diagnose_authorization_failure(
390    q: &Question,
391    talk: &Talk,
392    reply: &str,
393    quote: Option<&str>,
394) -> &'static str {
395    if !q.free_text() && !q.choices.iter().any(|choice| choice == reply) {
396        return "alias mismatch: resolve an offered label or supported action alias with `magi answer --list`";
397    }
398    let queued_owner = (!talk.pending.is_empty() || !talk.pending_messages.is_empty())
399        && (talk.pending_messages.is_empty()
400            || talk
401                .pending_messages
402                .iter()
403                .any(|send| send.sender != Some(talk::Sender::Magi)));
404    if queued_owner {
405        return "stale persisted evidence: a queued owner send may retract consent; wait for it to be drained and read it again";
406    }
407    let Some((turn, _, instruction)) = current_instruction(talk) else {
408        return "missing context: no current persisted owner send; read the conversation again";
409    };
410    if quote.is_some_and(|quote| quote != instruction) {
411        return "stale persisted evidence: --quote must preserve the entire latest owner send; read it again";
412    }
413    if talk.turns[turn].at < q.asked_at {
414        return "stale persisted evidence: the latest owner send predates this question; obtain a current decision";
415    }
416    if contextual_recovery(q, reply) {
417        return "insufficient consent: bind an explicit question/action reply, or interpret approval of a single saved recommendation with --recovery-consent <entire latest owner send>; clarify only if the owner's intent is ambiguous";
418    }
419    "insufficient consent: answer requires --quote of the entire latest owner send bound to this question ID and exact offered choice, or explicit authorization of the immediately preceding question-by-question recommendations; persisted evidence must be current"
420}
421
422/// Only structured internal recovery effects qualify. Approval and publication
423/// questions never acquire contextual consent, even if actions were attached.
424fn contextual_recovery(q: &Question, reply: &str) -> bool {
425    QuestionKind::of(q) == QuestionKind::Ordinary
426        && consent(q, reply) == Consent::Delegable
427        && (matches!(q.actions.get(reply), Some(crate::ask::ChoiceAction::Resume { run }) if !run.is_empty())
428            || matches!(q.actions.get(reply), Some(crate::ask::ChoiceAction::Hold))
429            || crate::triage::contextual_keep_held(q, reply))
430}
431
432/// Agent-mediated interpretation, bound to a full real owner send and the
433/// immutable recommendation. Call under the same locks as `authorize_answer`.
434/// This does not execute recovery; the daemon retains all ownership and merge gates.
435pub fn authorize_recovery(
436    q: &Question,
437    talks: &Talks,
438    questions: &Questions,
439    seat: (&str, &str),
440    reply: &str,
441    instruction: &str,
442) -> Result<Option<AnswerAuthority>> {
443    let (run, node) = seat;
444    if node != CHAT_NODE || !contextual_recovery(q, reply) {
445        bail!(
446            "insufficient consent: contextual interpretation is limited to structured internal resume/keep-held actions; use this decision's explicit consent workflow"
447        );
448    }
449    let talk = talks.get(run)?;
450    let evidence = recovery_authority(q, questions, &talk, reply, instruction)?;
451    if q.consult
452        .as_ref()
453        .is_some_and(|c| c.talk != run && talks.turn_held(&c.talk))
454    {
455        bail!("original consultation is active; wait for its turn to finish before answering");
456    }
457    Ok(Some(evidence))
458}
459
460fn recovery_authority(
461    q: &Question,
462    questions: &Questions,
463    talk: &Talk,
464    reply: &str,
465    instruction: &str,
466) -> Result<AnswerAuthority> {
467    let (turn, message, latest) = current_instruction(talk)
468        .context("missing context: wait for the current owner send to be persisted and drained")?;
469    if instruction != latest {
470        bail!(
471            "stale persisted evidence: the complete latest owner send differs; read the conversation again, including any retraction"
472        );
473    }
474    let owner_sends = talk.turns[turn]
475        .messages
476        .iter()
477        .filter(|send| talk::sender(&talk.turns[turn], Some(send)) != talk::Sender::Magi)
478        .count();
479    if owner_sends != 1
480        || latest.trim().is_empty()
481        || latest.starts_with("    ")
482        || latest.starts_with('\t')
483        || latest.trim_start().starts_with(['>', '`'])
484    {
485        bail!("insufficient consent: quoted text or multiple owner sends require clarification");
486    }
487    let index = talk.turns[..turn]
488        .iter()
489        .rposition(|t| talk::sender(t, None) != talk::Sender::Magi)
490        .context("missing context: present one explicit recovery recommendation first")?;
491    let recommendation = &talk.turns[index];
492    let manifest = talk.recommendations.get(&index).context(
493        "missing context: no persisted recommendation binding; present the recommendation now",
494    )?;
495    if recommendation.who != talk::Who::Agent
496        || talk::sender(recommendation, None) != talk::Sender::Persona
497        || recommendation.posted_by.is_some()
498        || manifest.body != crate::github_text::fingerprint("", &recommendation.body)
499        || manifest.questions.get(&q.id) != Some(&question_fingerprint(q))
500        || recommendation.at < q.asked_at
501        || talk.turns[turn].at < recommendation.at
502        || !q.status.open()
503    {
504        bail!(
505            "stale persisted evidence: recommendation or question target changed; present the current recommendation again"
506        );
507    }
508    if manifest.questions.len() != 1 {
509        bail!(
510            "ambiguous recovery context: multiple recommended decisions; clarify which question and action the owner intends"
511        );
512    }
513    let binding = manifest.recovery.get(&q.id)
514        .context("missing context: no structured recovery binding was saved; present a current recommendation")?;
515    if binding.run != q.run || binding.action.as_ref() != q.actions.get(&binding.choice) {
516        bail!("stale persisted evidence: the recommended recovery target changed");
517    }
518    if binding.choice != reply
519        || recommended_choice(&recommendation.body, q, questions) != Some(reply)
520    {
521        bail!("alias mismatch: the selected action differs from the persisted recommendation");
522    }
523    Ok(AnswerAuthority {
524        talk: talk.id.clone(),
525        turn,
526        message,
527        instruction: latest.into(),
528        instruction_fingerprint: crate::github_text::fingerprint("", latest),
529        recommendation_turn: Some(index),
530        recommendation_fingerprint: Some(manifest.body.clone()),
531        question: q.id.clone(),
532        choice: reply.into(),
533        question_fingerprint: question_fingerprint(q),
534        recovery_interpretation: Some(
535            "owner approved the single current recovery recommendation".into(),
536        ),
537    })
538}
539
540/// Whether this turn needs the question-store write grant. Uses the same
541/// persisted evidence and effect policy as the CLI; the CLI rechecks on write.
542pub fn authorized_answers(questions: &Questions, talk: &Talk) -> bool {
543    let quote = current_instruction(talk).map(|(_, _, body)| body);
544    questions.list().iter().any(|q| {
545        if q.free_text() {
546            return quote
547                .and_then(|s| s.split_once(':'))
548                .is_some_and(|(_, reply)| {
549                    choice_authority(q, questions, talk, reply.trim(), quote).is_some()
550                });
551        }
552        q.choices.iter().any(|choice| {
553            if contextual_recovery(q, choice)
554                && quote.is_some_and(|words| {
555                    recovery_authority(q, questions, talk, choice, words).is_ok()
556                })
557            {
558                return true;
559            }
560            choice_authority(q, questions, talk, choice, quote).is_some_and(|e| {
561                validate_effect(q, choice, &e.instruction, quote.unwrap_or_default(), true).is_ok()
562            })
563        })
564    })
565}
566
567/// The complete standing-turn write grant, also used by sandbox regressions.
568/// Consultations and bound owner decisions need question/evidence locks;
569/// authorized dismissals additionally need queue claims and the exact run's
570/// ownership/state writes. Never grant the whole home or lift the sandbox.
571pub fn answer_write_roots(
572    questions: &Questions,
573    talks: &Talks,
574    talk: &Talk,
575) -> Vec<std::path::PathBuf> {
576    if !pending_consults(questions, &talk.id) && !authorized_answers(questions, talk) {
577        return Vec::new();
578    }
579    let Some(home) = questions.root().parent() else {
580        return Vec::new();
581    };
582    let quote = current_instruction(talk).map(|(_, _, words)| words);
583    let mut roots = std::collections::BTreeSet::from([
584        questions.root().to_path_buf(),
585        talks.answer_lock_root(),
586    ]);
587    for q in questions.list() {
588        if q.status.open()
589            && q.node == crate::land::APPROVAL_NODE
590            && q.choices
591                .iter()
592                .any(|choice| choice == crate::merge_decision::DISMISS)
593            && crate::run::is_run_id(&q.run)
594            && authorize_answer(
595                &q,
596                talks,
597                questions,
598                &talk.id,
599                CHAT_NODE,
600                crate::merge_decision::DISMISS,
601                quote,
602            )
603            .is_ok()
604        {
605            roots.insert(home.join("queue"));
606            roots.insert(home.join("runs").join(&q.run));
607        }
608    }
609    roots.into_iter().collect()
610}
611
612/// Compatibility name for callers of the earlier hold-only grant.
613pub fn authorized_holds(questions: &Questions, talk: &Talk) -> bool {
614    authorized_answers(questions, talk)
615}
616
617/// Queue state is only a diagnostic lead, never proof of lock ownership.
618pub(crate) fn investigation_holder_hint(q: &Question, tasks: &[Task]) -> String {
619    let task = tasks.iter().find(|task| {
620        !q.run.is_empty()
621            && (task.id == q.run || task.runs.contains(&q.run))
622            && task.status == crate::queue::TaskStatus::Running
623    });
624    match task {
625        Some(task) => format!(
626            "Lock owner unknown. Related task {} is running and may hold the repository lock; inspect it with `magi task show {}`. ",
627            task.id, task.id
628        ),
629        None => "Lock owner unknown. ".into(),
630    }
631}
632
633/// Action consent is separate from consultation routing. This inventory uses
634/// execution effects (including structured actions), not only question nodes.
635#[derive(Debug, Clone, Copy, PartialEq, Eq)]
636pub enum Consent {
637    /// Explicit single or batch instruction is sufficient.
638    Delegable,
639    /// Exact PR/head approval is required.
640    Merge,
641    /// Exact offered consequence must be confirmed individually.
642    Consequence,
643    /// Chosen public text needs explicit confirmation.
644    Posting,
645}
646
647/// Execution effects shared with question deputies. The issuer and seat are
648/// part of the policy: identical labels can have different consumers.
649pub fn destructive_effect(q: &Question, label: &str) -> bool {
650    if label == crate::land::HOLD && !q.actions.contains_key(label) {
651        return false;
652    }
653    // Conductor metadata describes the requested operation independently of
654    // generated choice wording. Completing a task or changing settings needs
655    // individual consent even when no structured ChoiceAction was attached.
656    if q.node == crate::conduct::NODE
657        && q.decision_context
658            .as_deref()
659            .and_then(|context| serde_json::from_str::<serde_json::Value>(context).ok())
660            .is_some_and(|context| {
661                matches!(
662                    context
663                        .get(3)
664                        .and_then(|v| v.get(0))
665                        .and_then(|v| v.as_str()),
666                    Some("complete" | "configure")
667                )
668            })
669    {
670        return true;
671    }
672    if matches!(q.actions.get(label), Some(crate::ask::ChoiceAction::Done)) {
673        return true;
674    }
675    let at = q.choices.iter().position(|c| c == label);
676    // By label, not position: the retry question orders its choices differently.
677    if q.node == crate::github_text::ASK_NODE {
678        return matches!(
679            label,
680            crate::github_text::POST_ORIGINAL
681                | crate::github_text::USE_REWRITE
682                | crate::github_text::USE_MY_TEXT
683        );
684    }
685    match q.node.as_str() {
686        crate::triage::NODE => at == Some(2),
687        crate::triage::DEPS_NODE => matches!(at, Some(1 | 2)),
688        n => n == crate::reconcile::NODE && q.seat == crate::reconcile::SEAT,
689    }
690}
691
692/// Classify every offered decision, including generic structured actions.
693pub fn consent(q: &Question, choice: &str) -> Consent {
694    if choice == crate::land::APPROVE {
695        Consent::Merge
696    } else if q.node == crate::github_text::ASK_NODE
697        && matches!(
698            choice,
699            crate::github_text::POST_ORIGINAL
700                | crate::github_text::USE_REWRITE
701                | crate::github_text::USE_MY_TEXT
702        )
703    {
704        Consent::Posting
705    } else if choice == crate::merge_decision::WITHDRAW
706        || crate::merge_decision::selected_agent(choice).is_some()
707        || destructive_effect(q, choice)
708        || matches!(q.actions.get(choice), Some(crate::ask::ChoiceAction::Done))
709    {
710        Consent::Consequence
711    } else {
712        Consent::Delegable
713    }
714}
715
716/// Deputies use their authenticated question thread rather than a talk.
717/// A generic batch instruction cannot replace action-specific confirmation.
718pub(crate) fn validate_batch_effect(
719    q: &Question,
720    choice: &str,
721    message: &str,
722    quote: &str,
723) -> Result<()> {
724    if batch_instruction(message) {
725        validate_effect(q, choice, message, quote, false)?;
726        bail!(
727            "a deputy's one-question thread has no persisted question-by-question recommendation manifest; use the active standing conversation's bound owner evidence rather than inferring a choice from generic batch words"
728        );
729    }
730    Ok(())
731}
732
733/// Validate action-specific evidence independently of its delivery route.
734pub(crate) fn validate_effect(
735    q: &Question,
736    choice: &str,
737    instruction: &str,
738    quote: &str,
739    bound: bool,
740) -> Result<()> {
741    let direct = format!("{}: {}", q.id, choice);
742    match consent(q, choice) {
743        Consent::Delegable => {}
744        Consent::Merge => {
745            let target = q.merge_target.as_ref().context("merge requires the specific reviewed PR and head binding; this legacy card has no recorded target")?;
746            let token = format!("merge {}", target.head);
747            if quote.trim() != instruction.trim()
748                || !(instruction.trim() == token
749                    || instruction.trim() == format!("{}: {token}", q.id))
750            {
751                bail!(
752                    "merge requires separate explicit confirmation of this specific reviewed PR and head: `{}: {token}`; a batch recommendation is insufficient",
753                    q.id
754                );
755            }
756        }
757        Consent::Consequence => {
758            if quote.trim() != instruction.trim()
759                || !(instruction.trim() == direct || (!bound && instruction.trim() == choice))
760            {
761                bail!(
762                    "`{choice}` requires an individual owner confirmation of the exact offered consequence: `{direct}`; batch authorization is insufficient"
763                );
764            }
765        }
766        Consent::Posting => {
767            if bound {
768                let token = format!("{}: {} {}", q.id, choice, question_fingerprint(q));
769                let content_matches = if choice == crate::github_text::USE_MY_TEXT {
770                    owner_title(q, instruction).is_some()
771                } else {
772                    instruction.trim() == token
773                };
774                if !content_matches || quote.trim() != instruction.trim() {
775                    bail!(
776                        "posting confirmation requires owner confirmation bound to the exact question and saved content: `{token}`; `use my text` additionally requires a newline and the literal replacement title; batch authorization is insufficient"
777                    );
778                }
779            } else if !crate::github_text::posting_authorized(instruction, quote, choice) {
780                bail!(
781                    "posting confirmation requires an explicit, unconditional instruction for the chosen text; batch authorization is insufficient"
782                );
783            }
784        }
785    }
786    Ok(())
787}
788
789fn current_instruction(talk: &Talk) -> Option<(usize, usize, &str)> {
790    // A newer queued send can retract the instruction. Wait until it is
791    // drained rather than letting the running seat use older words. Explicitly
792    // generated sends neither authorize nor retract an owner instruction.
793    let queued_owner = (!talk.pending.is_empty() || !talk.pending_messages.is_empty())
794        && (talk.pending_messages.is_empty()
795            || talk
796                .pending_messages
797                .iter()
798                .any(|m| m.sender != Some(talk::Sender::Magi)));
799    if !talk.status.open() || queued_owner {
800        return None;
801    }
802    for (index, turn) in talk.turns.iter().enumerate().rev() {
803        if talk::sender(turn, None) == talk::Sender::Magi {
804            continue;
805        }
806        // A persona reply consumes the conversational turn. Never reuse an
807        // older instruction just because a generated post followed that reply.
808        if turn.who != talk::Who::Operator || turn.posted_by.is_some() {
809            return None;
810        }
811        let latest = turn
812            .messages
813            .iter()
814            .enumerate()
815            .rfind(|(_, send)| talk::sender(turn, Some(send)) == talk::Sender::Operator);
816        let Some((message, send)) = latest else {
817            if !turn.messages.is_empty()
818                && turn
819                    .messages
820                    .iter()
821                    .all(|m| m.sender == Some(talk::Sender::Magi))
822            {
823                continue;
824            }
825            return None;
826        };
827        // Older real sends can be reconstructed from recorded boundaries,
828        // never from a guessed last paragraph. The whole-message grammar also
829        // excludes generated legacy wrappers.
830        if send.sender.is_none()
831            && (turn.breaks.is_none()
832                || turn
833                    .messages
834                    .iter()
835                    .map(|m| m.body.as_str())
836                    .collect::<Vec<_>>()
837                    .join("\n\n")
838                    != turn.body)
839        {
840            return None;
841        }
842        return Some((index, message, &send.body));
843    }
844    None
845}
846
847fn choice_authority(
848    q: &Question,
849    questions: &Questions,
850    talk: &Talk,
851    reply: &str,
852    quote: Option<&str>,
853) -> Option<AnswerAuthority> {
854    if !q.status.open()
855        || !(q.free_text() || q.choices.iter().any(|c| c == reply))
856        || (reply == crate::land::HOLD && q.actions.contains_key(reply))
857    {
858        return None;
859    }
860    let (turn, message, instruction) = current_instruction(talk)?;
861    if reply != crate::land::HOLD
862        && (instruction.starts_with("    ") || instruction.starts_with('\t'))
863    {
864        return None;
865    }
866    if talk.turns[turn].at < q.asked_at {
867        return None;
868    }
869    let direct = direct_choice(instruction, q, questions, reply)
870        || (reply == crate::github_text::USE_MY_TEXT
871            && q.node == crate::github_text::ASK_NODE
872            && owner_title(q, instruction).is_some())
873        || (consent(q, reply) == Consent::Posting
874            && instruction.trim() == format!("{}: {} {}", q.id, reply, question_fingerprint(q)))
875        || (consent(q, reply) == Consent::Merge
876            && q.merge_target
877                .as_ref()
878                .is_some_and(|t| instruction.trim() == format!("{}: merge {}", q.id, t.head)));
879    let quote_matches = if contextual_recovery(q, reply) {
880        quote == Some(instruction)
881    } else {
882        quote.map(str::trim) == Some(instruction.trim())
883    };
884    let persisted_recovery = quote.is_none() && direct && contextual_recovery(q, reply);
885    if !quote_matches && !persisted_recovery {
886        return None;
887    }
888    let recommendation_turn = if direct {
889        None
890    } else {
891        if message != 0 || q.free_text() || !batch_instruction(instruction) {
892            return None;
893        }
894        let index = turn.checked_sub(1)?;
895        let recommendation = &talk.turns[index];
896        if recommendation.who != talk::Who::Agent
897            || talk::sender(recommendation, None) != talk::Sender::Persona
898            || recommendation.posted_by.is_some()
899            || recommendation.at < q.asked_at
900            || recommended_choice(&recommendation.body, q, questions) != Some(reply)
901        {
902            return None;
903        }
904        if let Some(manifest) = talk.recommendations.get(&index) {
905            if manifest.body != crate::github_text::fingerprint("", &recommendation.body)
906                || manifest.questions.get(&q.id) != Some(&question_fingerprint(q))
907            {
908                return None;
909            }
910        } else if reply != crate::land::HOLD {
911            // Old transcripts can prove a safe hold, but cannot establish what
912            // mutable action/target was shown before manifests were persisted.
913            return None;
914        }
915        Some(index)
916    };
917    Some(AnswerAuthority {
918        talk: talk.id.clone(),
919        turn,
920        message,
921        instruction: instruction.to_owned(),
922        instruction_fingerprint: crate::github_text::fingerprint("", instruction),
923        recommendation_turn,
924        recommendation_fingerprint: recommendation_turn
925            .map(|i| crate::github_text::fingerprint("", &talk.turns[i].body)),
926        question: q.id.clone(),
927        choice: reply.into(),
928        question_fingerprint: question_fingerprint(q),
929        recovery_interpretation: None,
930    })
931}
932
933fn batch_instruction(text: &str) -> bool {
934    // Whole-message grammar: substrings in conditional, reported, quoted or
935    // retracted instructions cannot match. Keep this deliberately bounded.
936    let compact: String = text.chars().filter(|c| !c.is_whitespace()).collect();
937    let mut compact = compact.trim_end_matches(['!', '!', '。', '.', 'ー', '〜', '~']);
938    let prefixes = [
939        "おっけー",
940        "オッケー〜",
941        "オッケー",
942        "おっけ〜",
943        "おけ",
944        "オケ",
945        "ok",
946        "OK",
947        "Ok",
948        "ありがとうございます",
949        "ありがとう",
950        "ありがと",
951        "了解です",
952        "了解",
953        "はい",
954        "じゃあ",
955        "では",
956    ];
957    while let Some(rest) = prefixes
958        .iter()
959        .find_map(|prefix| compact.strip_prefix(prefix))
960    {
961        compact = rest.trim_start_matches(['!', '!', '、', ',', '。', '.', 'ー', '〜', '~']);
962    }
963    matches!(compact, "それぞれ回答して" | "それぞれ回答してください")
964        || matches!(
965            text.trim()
966                .trim_end_matches(['.', '!'])
967                .to_ascii_lowercase()
968                .as_str(),
969            "answer each as recommended" | "answer all as recommended"
970        )
971}
972
973fn refers_to(reference: &str, q: &Question, questions: &Questions) -> bool {
974    reference == q.id
975        || (reference.len() >= 4 && questions.resolve_id(reference).is_ok_and(|id| id == q.id))
976}
977
978fn direct_choice(text: &str, q: &Question, questions: &Questions, reply: &str) -> bool {
979    // One complete instruction, binding both the question and the offered
980    // choice. Other actions (including closing the PR) are never inferred.
981    let text = text.trim();
982    let binding = text.split_once(':').or_else(|| {
983        if contextual_recovery(q, reply) {
984            text.split_once(char::is_whitespace)
985        } else {
986            None
987        }
988    });
989    let Some((reference, choice)) = binding else {
990        return false;
991    };
992    let reference = reference.trim().trim_matches('`');
993    (reference == q.id
994        || ((reply == crate::land::HOLD || contextual_recovery(q, reply))
995            && refers_to(reference, q, questions)))
996        && (choice.trim() == reply
997            || (contextual_recovery(q, reply)
998                && q.resolve_reply(choice.trim())
999                    .is_ok_and(|resolved| resolved == reply)))
1000}
1001
1002fn recommended_choice<'a>(body: &str, q: &'a Question, questions: &Questions) -> Option<&'a str> {
1003    // A row must match the entire offered label, including prefixes such as
1004    // `dismiss:` or `reimplement:`. Prefix/substring matches confer no authority.
1005    // A row must select exactly one offered choice, rather than list options.
1006    // Repeated or conflicting rows are ambiguous even if one recommends hold.
1007    let mut rows = Vec::new();
1008    // Use the same Markdown parser as the transcript renderer. Quoted lazy
1009    // continuations, nested fences and indented examples are not recommendations.
1010    use comrak::nodes::NodeValue;
1011    let arena = comrak::Arena::new();
1012    let mut options = comrak::Options::default();
1013    options.extension.table = true;
1014    let document = comrak::parse_document(&arena, body, &options);
1015    let lines: Vec<_> = body.lines().collect();
1016    let mut eligible = std::collections::BTreeSet::new();
1017    for node in document.descendants() {
1018        let data = node.data.borrow();
1019        if node
1020            .ancestors()
1021            .any(|n| matches!(n.data.borrow().value, NodeValue::BlockQuote))
1022        {
1023            continue;
1024        }
1025        // A heading, introduction or trailing paragraph may qualify several
1026        // rows without naming any question. Do not guess its scope: batch
1027        // authority requires the whole unquoted recommendation to be clear.
1028        if matches!(
1029            data.value,
1030            NodeValue::Heading(_) | NodeValue::Paragraph | NodeValue::TableRow(_)
1031        ) && lines
1032            .get(data.sourcepos.start.line - 1..data.sourcepos.end.line)
1033            .is_some_and(|block| conditional_recommendation(&block.join(" ")))
1034        {
1035            return None;
1036        }
1037        if matches!(
1038            data.value,
1039            NodeValue::Paragraph | NodeValue::TableRow(false)
1040        ) {
1041            eligible.extend(data.sourcepos.start.line..=data.sourcepos.end.line);
1042        }
1043    }
1044    for (index, line) in body.lines().enumerate() {
1045        if !eligible.contains(&(index + 1)) {
1046            continue;
1047        }
1048        let line = line.trim();
1049        let tokens: Vec<&str> = line
1050            .split(|c: char| !c.is_ascii_alphanumeric() && c != '-')
1051            .collect();
1052        if !tokens.iter().any(|r| refers_to(r, q, questions)) {
1053            continue;
1054        }
1055        if line.starts_with('|') {
1056            let cells: Vec<_> = line
1057                .trim_matches('|')
1058                .split('|')
1059                .map(|c| c.trim().trim_matches('`'))
1060                .collect();
1061            let references = cells.iter().filter(|c| refers_to(c, q, questions)).count();
1062            let other_question = cells
1063                .iter()
1064                .any(|c| c.len() >= 4 && questions.resolve_id(c).is_ok_and(|id| id != q.id));
1065            let choices: Vec<_> = cells
1066                .iter()
1067                .filter_map(|c| q.resolve_reply(c).ok())
1068                .collect();
1069            if other_question || references != 1 || choices.len() != 1 {
1070                return None;
1071            }
1072            rows.push(
1073                q.choices
1074                    .iter()
1075                    .find(|choice| choice.as_str() == choices[0])?
1076                    .as_str(),
1077            );
1078            continue;
1079        }
1080        let row = line.trim_start_matches(['-', '*', ' ']);
1081        let row = row.trim_start_matches('`');
1082        let end = row
1083            .find(|c: char| !c.is_ascii_alphanumeric() && c != '-')
1084            .unwrap_or(row.len());
1085        let (reference, rest) = row.split_at(end);
1086        let rest = rest.trim_start_matches('`').trim_start();
1087        let choice = rest
1088            .strip_prefix(':')
1089            .or_else(|| rest.strip_prefix('→'))
1090            .or_else(|| rest.strip_prefix('|'))
1091            .map(str::trim);
1092        if !refers_to(reference, q, questions) {
1093            return None;
1094        }
1095        let choice = choice?;
1096        // Preserve the legacy safe-hold wording without granting PR closure.
1097        let choice = if choice == "hold、その後PRを閉じる" {
1098            crate::land::HOLD
1099        } else {
1100            choice.trim_matches('`')
1101        };
1102        rows.push(
1103            q.choices
1104                .iter()
1105                .find(|offered| {
1106                    q.resolve_reply(choice)
1107                        .is_ok_and(|resolved| offered.as_str() == resolved)
1108                })?
1109                .as_str(),
1110        );
1111    }
1112    (rows.len() == 1).then(|| rows[0])
1113}
1114
1115fn conditional_recommendation(line: &str) -> bool {
1116    let lower = line.to_lowercase();
1117    lower.split(|c: char| !c.is_ascii_alphabetic()).any(|word| {
1118        matches!(
1119            word,
1120            "if" | "unless"
1121                | "when"
1122                | "maybe"
1123                | "conditional"
1124                | "otherwise"
1125                | "only"
1126                | "except"
1127                | "or"
1128                | "either"
1129                | "provided"
1130                | "assuming"
1131                | "depending"
1132        )
1133    }) || [
1134        "ただし",
1135        "場合",
1136        "なら",
1137        "たら",
1138        "限り",
1139        "のみ",
1140        "かも",
1141        "次第",
1142        "条件",
1143        "または",
1144        "もしくは",
1145        "あるいは",
1146        "れば",
1147        "とき",
1148        "際",
1149        "時だけ",
1150        "必要に応じて",
1151    ]
1152    .iter()
1153    .any(|word| line.contains(word))
1154}
1155
1156impl QuestionKind {
1157    /// Select the policy from the stored question node.
1158    pub fn of(q: &Question) -> Self {
1159        if crate::deputy::merge_gated(q) {
1160            return Self::Merge;
1161        }
1162        match q.node.as_str() {
1163            crate::github_text::ASK_NODE => Self::Posting,
1164            _ => Self::Ordinary,
1165        }
1166    }
1167}
1168
1169/// Validate guarded decisions answered by a running chat. Terminal answers
1170/// retain their existing behavior. Merge and posting policies both require
1171/// evidence from the correct conversation; posting also checks the selected text.
1172pub fn validate_answer(
1173    q: &Question,
1174    talks: &Talks,
1175    run: &str,
1176    node: &str,
1177    reply: &str,
1178    quote: Option<&str>,
1179) -> Result<()> {
1180    let kind = QuestionKind::of(q);
1181    if node != CHAT_NODE || kind == QuestionKind::Ordinary {
1182        return Ok(());
1183    }
1184    let posting = kind == QuestionKind::Posting;
1185    if posting && !q.choices.iter().any(|c| c == reply) {
1186        bail!("posting confirmation requires one of the offered choices");
1187    }
1188    if posting && reply == crate::github_text::USE_MY_TEXT {
1189        bail!(
1190            "use my text requires current question-bound owner evidence with the content fingerprint, a newline and the literal replacement title; an old question-thread title or hold is insufficient"
1191        );
1192    }
1193    if posting
1194        && !matches!(
1195            reply,
1196            crate::github_text::POST_ORIGINAL | crate::github_text::USE_REWRITE
1197        )
1198    {
1199        return Ok(());
1200    }
1201    let subject = if posting {
1202        "posting confirmation"
1203    } else {
1204        "merge approval"
1205    };
1206    let consult = q
1207        .consult
1208        .as_ref()
1209        .with_context(|| format!("{subject} was not handed to a chat"))?;
1210    if consult.talk != run {
1211        bail!("{subject} belongs to a different chat");
1212    }
1213    if !q.status.open() {
1214        bail!("{subject} is no longer open");
1215    }
1216    let talk = talks.get(run)?;
1217    if !talk.status.open() {
1218        bail!("the consulted chat is closed");
1219    }
1220    // Only words the owner wrote after this question was handed over count.
1221    // The generated hand-over text is stored as an operator turn (possibly
1222    // coalesced with owner replies), so it is cut out by its own markers.
1223    let at = talk
1224        .turns
1225        .iter()
1226        .rposition(|t| {
1227            t.who == talk::Who::Operator
1228                && t.body.contains(crate::prompt::CHAT_CONSULT_HEADING)
1229                && t.body.contains(&q.id)
1230        })
1231        .context("the question was not delivered to the chat yet")?;
1232    // A reply still waiting in `pending` is newer than every stored turn.
1233    let queued = latest_owner_message(
1234        &talk.pending,
1235        talk.pending_breaks.as_deref(),
1236        &talk.pending_messages,
1237        None,
1238        posting,
1239    );
1240    let stored = talk.turns[at..]
1241        .iter()
1242        .enumerate()
1243        .rev()
1244        .filter(|(_, t)| t.who == talk::Who::Operator)
1245        .find_map(|(i, t)| {
1246            latest_owner_message(
1247                &t.body,
1248                t.breaks.as_deref(),
1249                &t.messages,
1250                (i == 0).then_some(q.id.as_str()),
1251                posting,
1252            )
1253        });
1254    let latest = queued
1255        .or(stored)
1256        .context("no owner message after the question was handed to the chat")?;
1257    let quote = quote
1258        .map(str::trim)
1259        .filter(|s| !s.is_empty())
1260        .with_context(|| {
1261            format!("chat {subject} requires --quote from the owner's latest message")
1262        })?;
1263    if posting {
1264        if !latest.contains(quote) {
1265            bail!("posting confirmation requires a verbatim quote of the latest owner message");
1266        }
1267        if !crate::github_text::posting_authorized(&latest, quote, reply) {
1268            bail!(
1269                "posting confirmation requires an explicit, unconditional instruction for the chosen text; ask the owner to clarify"
1270            );
1271        }
1272        return Ok(());
1273    }
1274    if reply == crate::land::APPROVE && batch_instruction(&latest) {
1275        bail!(
1276            "a batch answer instruction is insufficient for merge; merging requires a separate explicit \
1277             confirmation of this specific reviewed PR and head through the existing approval driver"
1278        );
1279    }
1280    let valid = if reply == crate::land::APPROVE {
1281        match &q.merge_target {
1282            Some(target) => {
1283                latest.trim() == format!("merge {}", target.head) && quote == latest.trim()
1284            }
1285            None => latest.contains(quote),
1286        }
1287    } else {
1288        q.choices.iter().any(|c| c == reply)
1289            && crate::merge_decision::exact_owner_choice(reply, &latest, quote)
1290    };
1291    if !valid {
1292        if reply == crate::land::APPROVE {
1293            if let Some(target) = &q.merge_target {
1294                bail!(
1295                    "merge requires the latest whole owner message and --quote to be exactly `merge {}` for the displayed PR/head",
1296                    target.head
1297                );
1298            }
1299            bail!("merge requires a verbatim quote of the latest explicit owner merge instruction");
1300        }
1301        bail!(
1302            "`{reply}` requires the latest whole owner message and --quote to equal this exact offered choice, or current question-bound owner evidence under the action-specific consent policy"
1303        );
1304    }
1305    Ok(())
1306}
1307
1308/// The owner's newest message in a turn body (or the pending draft).
1309///
1310/// Replies queued while a turn runs are joined with a blank line, which an
1311/// owner's own blank lines cannot be told apart from, so `talk::queue` records
1312/// where each message starts (`breaks`) and the newest message is taken whole.
1313/// With no usable record (a turn stored before it existed) the only safe cut is
1314/// the last paragraph: that can reject a multi-paragraph approval, and can
1315/// accept a `hold` that was only the last paragraph of a longer message, which
1316/// is harmless because silence is a hold.
1317#[cfg(test)]
1318fn latest_message(
1319    body: &str,
1320    breaks: Option<&[usize]>,
1321    after_block_of: Option<&str>,
1322) -> Option<String> {
1323    latest_message_for_policy(body, breaks, after_block_of, false)
1324}
1325
1326fn latest_owner_message(
1327    body: &str,
1328    breaks: Option<&[usize]>,
1329    messages: &[talk::Message],
1330    after: Option<&str>,
1331    posting: bool,
1332) -> Option<String> {
1333    if messages.is_empty() {
1334        return latest_message_for_policy(body, breaks, after, posting);
1335    }
1336    let first = after.map_or(0, |id| {
1337        messages
1338            .iter()
1339            .rposition(|m| {
1340                m.body.contains(crate::prompt::CHAT_CONSULT_HEADING) && m.body.contains(id)
1341            })
1342            .unwrap_or(0)
1343    });
1344    // A legacy message may span several sends. Only map the recorded cuts
1345    // when the message bodies still match the enclosing body exactly.
1346    let aligned = messages
1347        .iter()
1348        .map(|m| m.body.as_str())
1349        .collect::<Vec<_>>()
1350        .join("\n\n")
1351        == body;
1352    let valid_breaks = breaks.filter(|cuts| {
1353        cuts.windows(2).all(|w| w[0] < w[1])
1354            && cuts.iter().all(|&o| {
1355                o >= 2 && o <= body.len() && body.is_char_boundary(o) && body[..o].ends_with("\n\n")
1356            })
1357    });
1358    let mut offset = 0;
1359    let mut latest = None;
1360    for (i, message) in messages.iter().enumerate() {
1361        let end = offset + message.body.len();
1362        if i >= first && message.sender != Some(talk::Sender::Magi) {
1363            let local_breaks = aligned.then_some(valid_breaks).flatten().map(|cuts| {
1364                cuts.iter()
1365                    .copied()
1366                    .filter(|&cut| cut > offset && cut <= end)
1367                    .map(|cut| cut - offset)
1368                    .collect::<Vec<_>>()
1369            });
1370            let message_breaks = if message.sender.is_some() {
1371                Some(&[][..])
1372            } else {
1373                local_breaks.as_deref()
1374            };
1375            if let Some(words) = latest_message_for_policy(
1376                &message.body,
1377                message_breaks,
1378                after.filter(|_| i == first),
1379                posting,
1380            ) {
1381                latest = Some(words);
1382            }
1383        }
1384        offset = end + 2;
1385    }
1386    latest
1387}
1388
1389fn latest_message_for_policy(
1390    body: &str,
1391    breaks: Option<&[usize]>,
1392    after_block_of: Option<&str>,
1393    posting: bool,
1394) -> Option<String> {
1395    let cuts = breaks.filter(|b| {
1396        b.windows(2).all(|w| w[0] < w[1])
1397            && b.iter().all(|&o| {
1398                o >= 2 && o <= body.len() && body.is_char_boundary(o) && body[..o].ends_with("\n\n")
1399            })
1400    });
1401    let Some(cuts) = cuts else {
1402        if is_completion_notice(body) {
1403            return None;
1404        }
1405        let words = owner_words_for_policy(body, after_block_of, posting);
1406        // Legacy/invalid boundary records cannot distinguish paragraphs from
1407        // separate owner replies. Posting must not promote a reported final
1408        // paragraph into an instruction. Merge keeps its existing policy.
1409        if posting && words.contains("\n\n") {
1410            // Keep this as the newest evidence (with no usable quote), so an
1411            // ambiguous queued reply cannot fall back to an older instruction.
1412            return Some(String::new());
1413        }
1414        if posting {
1415            return (!words.trim().is_empty()).then_some(words);
1416        }
1417        return words
1418            .rsplit("\n\n")
1419            .map(str::trim)
1420            .find(|p| !p.is_empty())
1421            .map(str::to_owned);
1422    };
1423    let mut starts = vec![0];
1424    starts.extend_from_slice(cuts);
1425    let messages: Vec<&str> = starts
1426        .iter()
1427        .enumerate()
1428        .map(|(i, &s)| {
1429            let e = starts.get(i + 1).map_or(body.len(), |&n| n - 2);
1430            &body[s..e]
1431        })
1432        .collect();
1433    // The message carrying the hand-over: everything before it is older.
1434    let first = after_block_of.map_or(0, |id| {
1435        messages
1436            .iter()
1437            .rposition(|m| m.contains(crate::prompt::CHAT_CONSULT_HEADING) && m.contains(id))
1438            .unwrap_or(0)
1439    });
1440    messages
1441        .iter()
1442        .enumerate()
1443        .skip(first)
1444        .rev()
1445        .filter(|(_, m)| !is_completion_notice(m))
1446        .map(|(i, m)| owner_words_for_policy(m, after_block_of.filter(|_| i == first), posting))
1447        .find(|w| !w.trim().is_empty())
1448}
1449
1450/// Is this message magi's own task or watch completion notice? It is never
1451/// the owner's word.
1452fn is_completion_notice(message: &str) -> bool {
1453    crate::prompt::is_completion_notice(message)
1454}
1455
1456/// Length of a consult stored before quoted markers were defused, which may
1457/// quote the end phrase or a whole earlier consult in its detail. The closer is
1458/// the generated tail's full final sentence (an owner reply does not repeat
1459/// it), and a real heading met before it opens a nested block that needs its
1460/// own closer. A block that never closes runs to the end of the text, which can
1461/// only exclude more.
1462fn legacy_len(rest: &str) -> usize {
1463    use crate::prompt::CHAT_CONSULT_HEADING;
1464    const CLOSERS: [&str; 2] = [
1465        "cannot be revived. Do not edit the repository.",
1466        "make: do not edit the repository.",
1467    ];
1468    let mut events: Vec<(usize, usize)> = Vec::new(); // (at, 0 = open, else closer length)
1469    let first = rest.find(CHAT_CONSULT_HEADING).unwrap_or(0);
1470    for (i, _) in rest.match_indices(CHAT_CONSULT_HEADING) {
1471        if i > first
1472            && rest[..i].ends_with("# ")
1473            && rest[..i - 2].ends_with('\n')
1474            && rest[i + CHAT_CONSULT_HEADING.len()..].starts_with("\n\nThe operator passed you")
1475        {
1476            events.push((i, 0));
1477        }
1478    }
1479    for c in CLOSERS {
1480        events.extend(rest.match_indices(c).map(|(i, _)| (i, c.len())));
1481    }
1482    events.sort_unstable();
1483    let mut depth = 1usize;
1484    for (at, len) in events {
1485        if len == 0 {
1486            depth += 1;
1487        } else {
1488            depth -= 1;
1489            if depth == 0 {
1490                return at + len;
1491            }
1492        }
1493    }
1494    rest.len()
1495}
1496
1497/// The owner's own words in an operator turn, with magi's generated hand-over
1498/// text cut out. `drain` stores a queued consult as an operator turn, and
1499/// `talk::queue` joins an owner reply sent meanwhile onto the same draft, so
1500/// the turn's origin cannot be told from the turn as a whole. Each generated
1501/// block runs from its [`CHAT_CONSULT_HEADING`](crate::prompt::CHAT_CONSULT_HEADING)
1502/// to the first [`CHAT_CONSULT_END`](crate::prompt::CHAT_CONSULT_END) after it
1503/// (quoted markers are defused by `prompt::chat_consult`; a consult stored
1504/// before that, recognised by its opening sentence, ends at the last one before
1505/// the next heading); with no end the
1506/// block runs to the next heading. Only the blocks are cut: owner replies
1507/// between them are kept, in order, even when they contain the phrase. With `after_block_of` (the turn
1508/// carrying the hand-over itself) everything up to the end of the last block
1509/// naming that question id predates the hand-over and is dropped.
1510#[cfg(test)]
1511fn owner_words(body: &str, after_block_of: Option<&str>) -> String {
1512    owner_words_for_policy(body, after_block_of, false)
1513}
1514
1515fn owner_words_for_policy(body: &str, after_block_of: Option<&str>, posting: bool) -> String {
1516    // Remove generated blocks without removing Markdown code indentation from
1517    // the owner's evidence. Merge retains its existing whitespace handling.
1518    let trim = |text: &str| -> String {
1519        if posting {
1520            text.trim_matches(['\r', '\n']).to_owned()
1521        } else {
1522            text.trim().to_owned()
1523        }
1524    };
1525    use crate::prompt::{CHAT_CONSULT_END, CHAT_CONSULT_HEADING};
1526    // A real heading is the generated one: `# ` at a line start, followed by
1527    // the fixed opening sentence. A heading phrase quoted inside a question's
1528    // detail is not a boundary; it stays inside the block it is part of.
1529    let heads: Vec<usize> = body
1530        .match_indices(CHAT_CONSULT_HEADING)
1531        .map(|(i, _)| i)
1532        .filter(|&i| {
1533            let line_start = body[..i]
1534                .strip_suffix("# ")
1535                .is_some_and(|b| b.is_empty() || b.ends_with('\n'));
1536            line_start
1537                && body[i + CHAT_CONSULT_HEADING.len()..].starts_with("\n\nThe operator passed you")
1538        })
1539        .collect();
1540    if heads.is_empty() {
1541        return trim(body);
1542    }
1543    // (start, end) of each generated block, the heading's own "# " included.
1544    // `chat_consult` defuses every marker quoted in a question's text, so the
1545    // first end phrase after a heading is the block's own. A block with no end
1546    // before the next heading runs up to that heading (or the text's end).
1547    // Turns stored before the markers were defused may still quote an end
1548    // phrase; their tail is not recognised as part of the block.
1549    let mut blocks: Vec<(usize, usize)> = Vec::new();
1550    for &h in &heads {
1551        let start = if body[..h].ends_with("# ") { h - 2 } else { h };
1552        if blocks.last().is_some_and(|&(_, e)| start < e) {
1553            continue; // quoted inside the block above
1554        }
1555        let rest = &body[h..];
1556        let head = rest.split("\n\n## ").next().unwrap_or(rest);
1557        let end = if head.contains(crate::prompt::CHAT_CONSULT_DEFUSED) {
1558            // Current format: quoted markers are defused, so the first end
1559            // phrase is the block's own.
1560            rest.find(CHAT_CONSULT_END)
1561                .map_or(body.len(), |p| h + p + CHAT_CONSULT_END.len())
1562        } else {
1563            h + legacy_len(rest)
1564        };
1565        blocks.push((start, end));
1566    }
1567    let mut from = 0;
1568    if let Some(id) = after_block_of {
1569        if let Some(&(_, end)) = blocks.iter().rfind(|&&(s, e)| body[s..e].contains(id)) {
1570            from = end;
1571        } else if let Some(&(_, end)) = blocks.last() {
1572            from = end;
1573        }
1574    }
1575    let mut parts = Vec::new();
1576    let mut at = from;
1577    for &(s, e) in &blocks {
1578        if s >= at {
1579            parts.push(trim(&body[at..s]));
1580        }
1581        at = at.max(e);
1582    }
1583    parts.push(trim(&body[at..]));
1584    parts
1585        .into_iter()
1586        .filter(|p| !p.trim().is_empty())
1587        .collect::<Vec<_>>()
1588        .join("\n\n")
1589}
1590
1591/// What [`start_turn`] did.
1592#[derive(Debug, PartialEq, Eq)]
1593pub enum Started {
1594    /// The question was handed over and the chat's turn ran to its end.
1595    Answered,
1596    /// Another process holds the talk's turn lease; nothing was changed.
1597    Busy,
1598    /// The question had already been handed to the chat; nothing was changed.
1599    Nothing,
1600}
1601
1602/// [`begin`], then run the chat's turn here, under the talk's own lease.
1603///
1604/// The lease is taken *before* anything is written, so a talk somebody else is
1605/// running leaves no consult record and no draft behind (`Started::Busy`) and
1606/// the caller can simply try again later. A turn that fails after `begin`
1607/// leaves the record and the draft, so the chat can resume it.
1608pub async fn start_turn(
1609    questions: &Questions,
1610    talks: &Talks,
1611    q: &Question,
1612    talk: &Talk,
1613    cfg: &Config,
1614) -> Result<Started> {
1615    if q.consult.is_some() {
1616        return Ok(Started::Nothing);
1617    }
1618    let Some(lease) = talks.claim_turn(&talk.id)? else {
1619        return Ok(Started::Busy);
1620    };
1621    if !begin(questions, talks, q, talk)? {
1622        return Ok(Started::Nothing);
1623    }
1624    drain_turns(talks, &talk.id, lease, cfg).await?;
1625    Ok(Started::Answered)
1626}
1627
1628/// Run the talk's turns under `lease` until no draft is owed.
1629///
1630/// Other starters that found the lease held queued drafts and left them to
1631/// us, so drain until nothing is left (as the web's drain loop does). A failed
1632/// turn is kept and reported at the end, not returned at once: drafts accepted
1633/// meanwhile are still owed an answer.
1634pub async fn drain_turns(
1635    talks: &Talks,
1636    talk_id: &str,
1637    mut lease: talk::TurnLease,
1638    cfg: &Config,
1639) -> Result<()> {
1640    let mut talk = talks.get(talk_id)?;
1641    let mut failed = None;
1642    loop {
1643        while let Some(text) = talk::drain(&mut talk, talks)? {
1644            if let Err(e) = talk::respond(&lease, &mut talk, talks, cfg, &text).await {
1645                failed.get_or_insert(e);
1646            }
1647            if !lease.beat()? {
1648                bail!("the turn lease for chat {} was lost", talk.short());
1649            }
1650        }
1651        // A draft queued after the last drain, before the lease is gone, was
1652        // left to us by a starter that found the lease held. Release first,
1653        // then look again: whoever queues later either sees no lease (and
1654        // starts its own turn) or is seen by this re-check.
1655        drop(lease);
1656        talk = talks.get(&talk.id)?;
1657        let owed = talk.status.open()
1658            && (!talk.pending.is_empty() || !talk.pending_attachments.is_empty());
1659        if !owed {
1660            break;
1661        }
1662        // Somebody else took the lease meanwhile: they drain it.
1663        let Some(again) = talks.claim_turn(&talk.id)? else {
1664            break;
1665        };
1666        lease = again;
1667    }
1668    if let Some(e) = failed {
1669        return Err(e);
1670    }
1671    Ok(())
1672}
1673
1674#[cfg(test)]
1675mod tests {
1676    use std::collections::BTreeMap;
1677    use std::path::PathBuf;
1678
1679    use super::*;
1680    use crate::config::{AgentKind, AgentSpec, Config};
1681    use crate::queue::Source;
1682    use crate::talk::TalkStatus;
1683
1684    const RUN: &str = "20260902-000000-beef";
1685
1686    fn talks() -> (tempfile::TempDir, Talks, Talk) {
1687        let tmp = tempfile::tempdir().expect("tempdir");
1688        let store = Talks::at(tmp.path().join("talks"));
1689        let cfg = Config {
1690            agents: vec![AgentSpec {
1691                id: "mock".to_owned(),
1692                kind: AgentKind::Command,
1693                model: None,
1694                command: vec!["true".to_owned()],
1695                extra_args: Vec::new(),
1696                env: BTreeMap::new(),
1697                prompt_delivery: None,
1698            }],
1699            ..Config::default()
1700        };
1701        let talk = talk::begin(&store, &cfg, tmp.path().to_path_buf(), Some("mock")).expect("talk");
1702        (tmp, store, talk)
1703    }
1704
1705    fn task(source: Source) -> Task {
1706        let mut t = Task::new(
1707            "t".to_owned(),
1708            "Do it".to_owned(),
1709            PathBuf::from("/repo"),
1710            source,
1711        );
1712        t.start(RUN.to_owned());
1713        t
1714    }
1715
1716    fn from_chat(talk: &Talk) -> Task {
1717        task(Source::Agent {
1718            run: talk.id.clone(),
1719            node: CHAT_NODE.to_owned(),
1720        })
1721    }
1722
1723    #[test]
1724    fn investigation_holder_hint_is_advisory_and_requires_a_related_running_task() {
1725        let mut q = question("implement");
1726        let mut related = task(Source::Human);
1727        let check = |q: &Question, tasks: &[Task]| investigation_holder_hint(q, tasks);
1728        let hint = check(&q, std::slice::from_ref(&related));
1729        assert!(hint.contains("Lock owner unknown"));
1730        assert!(hint.contains("may hold"));
1731        assert!(hint.contains(&format!("magi task show {}", related.id)));
1732        assert_eq!(check(&q, &[]), "Lock owner unknown. ");
1733        q.run = "unrelated".into();
1734        assert_eq!(check(&q, std::slice::from_ref(&related)), check(&q, &[]));
1735        q.run = related.id.clone();
1736        assert!(check(&q, std::slice::from_ref(&related)).contains("may hold"));
1737        related.hold_manual(None);
1738        assert_eq!(check(&q, std::slice::from_ref(&related)), check(&q, &[]));
1739        related.status = crate::queue::TaskStatus::Running;
1740        q.run.clear();
1741        assert_eq!(check(&q, &[related]), check(&q, &[]));
1742    }
1743
1744    fn question(node: &str) -> Question {
1745        Question::new(
1746            RUN.to_owned(),
1747            node.to_owned(),
1748            "impl-A".to_owned(),
1749            "Which backend?".to_owned(),
1750            "SQLite is simpler.".to_owned(),
1751            vec!["SQLite".to_owned(), "Redis".to_owned()],
1752        )
1753    }
1754
1755    #[test]
1756    fn batch_instruction_normalizes_repeated_japanese_prefixes() {
1757        for instruction in [
1758            "おっけーそれぞれ回答してー",
1759            "了解です、ありがとう!では、それぞれ回答してください。",
1760            "ありがとうございます!はい、じゃあそれぞれ回答して〜",
1761            "オッケー〜!おけ、それぞれ回答して",
1762            "OK!それぞれ回答して",
1763            "answer all as recommended!",
1764        ] {
1765            assert!(batch_instruction(instruction), "{instruction}");
1766        }
1767        for instruction in [
1768            "了解です",
1769            "ありがとう!それぞれ回答してもいい?",
1770            "もし大丈夫ならそれぞれ回答して",
1771            "それぞれ回答して、いや待って",
1772            "「それぞれ回答して」",
1773        ] {
1774            assert!(!batch_instruction(instruction), "{instruction}");
1775        }
1776    }
1777
1778    #[test]
1779    fn recommendation_returns_only_one_exact_offered_choice() {
1780        let tmp = tempfile::tempdir().unwrap();
1781        let questions = Questions::at(tmp.path().join("questions"));
1782        let mut q = question("implement");
1783        questions.put(&mut q).unwrap();
1784        let table = |rows: &str| format!("| Question | Choice |\n| --- | --- |\n{rows}");
1785        let row = format!("| {} | SQLite |", q.id);
1786        assert_eq!(
1787            recommended_choice(&table(&row), &q, &questions),
1788            Some("SQLite")
1789        );
1790        assert_eq!(
1791            recommended_choice(&format!("{}: Redis", q.id), &q, &questions),
1792            Some("Redis")
1793        );
1794        for body in [
1795            table(&format!("{row}\n| {} | Redis |", q.id)),
1796            table(&format!("{row}\n{row}")),
1797            table(&format!("| {} | SQLite | Redis |", q.id)),
1798            table(&format!("| {} | SQLite maybe |", q.id)),
1799            format!("> {}: SQLite", q.id),
1800            format!("If approved:\n\n{}: SQLite", q.id),
1801            "No recommendation".into(),
1802        ] {
1803            assert_eq!(recommended_choice(&body, &q, &questions), None, "{body}");
1804        }
1805    }
1806
1807    #[test]
1808    fn chat_answer_is_refused_for_an_operator_held_task() {
1809        let mut task = crate::queue::Task::new(
1810            "t".into(),
1811            "i".into(),
1812            std::path::PathBuf::from("."),
1813            crate::queue::Source::Human,
1814        );
1815        task.runs.push("run-1".into());
1816        let mut q = question(crate::land::APPROVAL_NODE);
1817        q.run = "run-1".into();
1818        let check = |t: &crate::queue::Task, node: &str| {
1819            refuse_operator_held(&q, node, std::slice::from_ref(t))
1820        };
1821        assert!(check(&task, CHAT_NODE).is_ok());
1822        task.hold_machine(None);
1823        assert!(check(&task, CHAT_NODE).is_ok());
1824        task.hold_manual(None);
1825        assert!(check(&task, CHAT_NODE).is_err());
1826        // Not a chat's answer, or another question: untouched.
1827        assert!(check(&task, "implement").is_ok());
1828        assert!(refuse_operator_held(&q, CHAT_NODE, &[]).is_ok());
1829    }
1830
1831    #[tokio::test]
1832    async fn start_turn_is_busy_and_writes_nothing_while_the_lease_is_held() {
1833        let (tmp, store, talk) = talks();
1834        let questions = Questions::at(tmp.path().join("questions"));
1835        let mut q = question("implement");
1836        questions.put(&mut q).expect("put");
1837        let held = store.claim_turn(&talk.id).expect("claim").expect("first");
1838
1839        let got = start_turn(&questions, &store, &q, &talk, &Config::default())
1840            .await
1841            .expect("start");
1842        assert_eq!(got, Started::Busy);
1843        assert!(questions.get(&q.id).unwrap().consult.is_none());
1844        assert!(store.get(&talk.id).unwrap().pending.is_empty());
1845        drop(held);
1846    }
1847
1848    /// A `kind = "command"` agent running `script`; `LEASE` names the turn
1849    /// lease file so the script can look at it while the turn is in flight.
1850    fn scripted(tmp: &std::path::Path, script: &str, lease: &std::path::Path) -> Config {
1851        let path = tmp.join("mock-consult-agent.sh");
1852        std::fs::write(&path, script).expect("write mock");
1853        Config {
1854            agents: vec![AgentSpec {
1855                id: "mock".to_owned(),
1856                kind: AgentKind::Command,
1857                model: None,
1858                command: vec!["sh".to_owned(), path.to_string_lossy().into_owned()],
1859                extra_args: Vec::new(),
1860                env: BTreeMap::from([("LEASE".to_owned(), lease.to_string_lossy().into_owned())]),
1861                prompt_delivery: None,
1862            }],
1863            ..Config::default()
1864        }
1865    }
1866
1867    /// Run `start_turn` with `script` as the agent and hand back its result,
1868    /// the store and the talk, for the caller to claim the lease afterwards.
1869    async fn run_turn(
1870        script: &str,
1871    ) -> (
1872        tempfile::TempDir,
1873        Result<Started>,
1874        Questions,
1875        Question,
1876        Talk,
1877    ) {
1878        crate::run::set_home(crate::run::test_home());
1879        let (tmp, store, talk) = talks();
1880        let questions = Questions::at(tmp.path().join("questions"));
1881        let mut q = question("implement");
1882        questions.put(&mut q).expect("put");
1883        let cfg = scripted(tmp.path(), script, &store.turn_path(&talk.id));
1884        let got = start_turn(&questions, &store, &q, &talk, &cfg).await;
1885        (tmp, got, questions, q, talk)
1886    }
1887
1888    #[tokio::test]
1889    async fn the_lease_is_held_during_the_turn_and_free_once_it_ends() {
1890        // exit 7 if the lease file is missing while the agent runs.
1891        let (tmp, got, _questions, _q, talk) =
1892            run_turn("#!/bin/sh\ncat >/dev/null\n[ -f \"$LEASE\" ] || exit 7\nprintf 'ok\\n'\n")
1893                .await;
1894        assert_eq!(got.expect("turn"), Started::Answered);
1895        let other = Talks::at(tmp.path().join("talks"));
1896        assert!(
1897            other.claim_turn(&talk.id).expect("claim").is_some(),
1898            "free once the turn ended"
1899        );
1900    }
1901
1902    #[tokio::test]
1903    async fn the_lease_is_free_again_after_a_failed_turn() {
1904        let (tmp, got, questions, q, talk) = run_turn("#!/bin/sh\ncat >/dev/null\nexit 3\n").await;
1905        assert!(got.is_err(), "the agent failed");
1906        assert!(
1907            questions.get(&q.id).unwrap().consult.is_some(),
1908            "the turn got as far as running"
1909        );
1910        let other = Talks::at(tmp.path().join("talks"));
1911        assert!(
1912            other.claim_turn(&talk.id).expect("claim").is_some(),
1913            "free after the failed turn"
1914        );
1915    }
1916
1917    #[test]
1918    fn origin_talk_is_decided_in_one_table() {
1919        let (_tmp, _store, talk) = talks();
1920        let mut closed = talk.clone();
1921        closed.status = TalkStatus::Closed;
1922        let chat = from_chat(&talk);
1923        let q = question("implement");
1924
1925        let hit = |tasks: &[Task], talks: &[Talk], q: &Question| origin_talk(tasks, talks, q);
1926        assert_eq!(
1927            hit(std::slice::from_ref(&chat), std::slice::from_ref(&talk), &q).map(|t| t.id),
1928            Some(talk.id.clone())
1929        );
1930        // Not filed from a chat.
1931        assert!(hit(&[task(Source::Human)], std::slice::from_ref(&talk), &q).is_none());
1932        let other = task(Source::Agent {
1933            run: talk.id.clone(),
1934            node: "implement".to_owned(),
1935        });
1936        assert!(hit(&[other], std::slice::from_ref(&talk), &q).is_none());
1937        // The conversation is gone, or no longer takes turns.
1938        assert!(hit(std::slice::from_ref(&chat), &[], &q).is_none());
1939        // A closed conversation is reopened by `begin`, so it still qualifies.
1940        assert!(hit(std::slice::from_ref(&chat), &[closed], &q).is_some());
1941        // No task owns the question.
1942        assert!(hit(&[], std::slice::from_ref(&talk), &q).is_none());
1943        // Merge approvals can be consulted; release notices cannot.
1944        assert!(
1945            hit(
1946                std::slice::from_ref(&chat),
1947                std::slice::from_ref(&talk),
1948                &question(crate::land::APPROVAL_NODE)
1949            )
1950            .is_some()
1951        );
1952        assert!(
1953            hit(
1954                std::slice::from_ref(&chat),
1955                std::slice::from_ref(&talk),
1956                &question(crate::bump::NOTICE_NODE)
1957            )
1958            .is_none()
1959        );
1960        // A settled question has nothing left to ask.
1961        let mut answered = question("implement");
1962        answered
1963            .answer(crate::ask::Answer::Choice("Redis".to_owned()))
1964            .unwrap();
1965        assert!(hit(&[chat], &[talk], &answered).is_none());
1966    }
1967
1968    #[test]
1969    fn associated_chat_remains_navigable_after_consultation_and_settlement() {
1970        let (tmp, store, mut talk) = talks();
1971        let chat = from_chat(&talk);
1972        let tasks = std::slice::from_ref(&chat);
1973        let mut q = question("implement");
1974        let questions = Questions::at(tmp.path().join("questions"));
1975        questions.put(&mut q).unwrap();
1976        assert!(begin(&questions, &store, &q, &talk).unwrap());
1977        q = questions.get(&q.id).unwrap();
1978        let associated = |q: &Question, talk: &Talk| {
1979            associated_talk(tasks, std::slice::from_ref(talk), q).map(|t| t.id)
1980        };
1981        assert_eq!(associated(&q, &talk), Some(talk.id.clone()));
1982        assert!(q.status.open());
1983        assert!(q.answer.is_none());
1984        assert!(origin_talk(tasks, std::slice::from_ref(&talk), &q).is_some());
1985        q.answer(crate::ask::Answer::Choice("Redis".into()))
1986            .unwrap();
1987        assert_eq!(associated(&q, &talk), Some(talk.id.clone()));
1988        assert!(origin_talk(tasks, std::slice::from_ref(&talk), &q).is_none());
1989        talk.status = TalkStatus::Closed;
1990        assert_eq!(associated(&q, &talk), Some(talk.id.clone()));
1991        let mut q = question("implement");
1992        q.abandon("expired");
1993        assert_eq!(associated(&q, &talk), Some(talk.id.clone()));
1994        assert!(origin_talk(tasks, std::slice::from_ref(&talk), &q).is_none());
1995        assert!(associated_talk(tasks, &[], &q).is_none());
1996        assert!(associated_talk(&[], std::slice::from_ref(&talk), &q).is_none());
1997        let notice = question(crate::bump::NOTICE_NODE);
1998        assert_eq!(associated(&notice, &talk), Some(talk.id.clone()));
1999        assert!(origin_talk(tasks, &[talk], &notice).is_none());
2000    }
2001
2002    #[test]
2003    fn approval_origin_requires_an_open_chat_task_and_question() {
2004        let (_tmp, _store, talk) = talks();
2005        let chat = from_chat(&talk);
2006        let mut q = question(crate::land::APPROVAL_NODE);
2007        assert!(origin_talk(&[task(Source::Human)], std::slice::from_ref(&talk), &q).is_none());
2008        assert!(
2009            origin_talk(
2010                &[task(Source::Agent {
2011                    run: talk.id.clone(),
2012                    node: "implement".into(),
2013                })],
2014                std::slice::from_ref(&talk),
2015                &q
2016            )
2017            .is_none()
2018        );
2019        assert!(origin_talk(std::slice::from_ref(&chat), &[], &q).is_none());
2020        let mut closed = talk.clone();
2021        closed.status = TalkStatus::Closed;
2022        assert!(origin_talk(std::slice::from_ref(&chat), &[closed], &q).is_some());
2023        q.abandon("expired");
2024        assert!(
2025            origin_talk(std::slice::from_ref(&chat), std::slice::from_ref(&talk), &q).is_none()
2026        );
2027        let mut q = question(crate::land::APPROVAL_NODE);
2028        q.answer(crate::ask::Answer::Choice("Redis".into()))
2029            .unwrap();
2030        assert!(origin_talk(&[chat], &[talk], &q).is_none());
2031    }
2032
2033    #[test]
2034    fn a_conductor_question_is_found_through_the_task_id() {
2035        let (_tmp, _store, talk) = talks();
2036        let chat = from_chat(&talk);
2037        let mut q = question(crate::conduct::NODE);
2038        q.run = chat.id.clone();
2039        assert!(origin_talk(&[chat], &[talk], &q).is_some());
2040    }
2041
2042    fn followup_of(parent: &Task, n: u32) -> Task {
2043        let mut t = Task::new(
2044            "f".to_owned(),
2045            "Fix".to_owned(),
2046            PathBuf::from("/repo"),
2047            Source::Agent {
2048                run: format!("merged-{n}"),
2049                node: "followup".to_owned(),
2050            },
2051        );
2052        t.start(format!("run-f{n}"));
2053        t.followup = Some(crate::queue::FollowUp {
2054            run: format!("merged-{n}"),
2055            origin_task: Some(parent.id.clone()),
2056            pr: "https://example.invalid/pr/1".to_owned(),
2057            findings: Vec::new(),
2058            generation: n,
2059        });
2060        t
2061    }
2062
2063    fn q_for(t: &Task) -> Question {
2064        let mut q = question("implement");
2065        q.run = t.runs[0].clone();
2066        q
2067    }
2068
2069    #[test]
2070    fn a_followup_traces_back_to_the_chat() {
2071        let (_tmp, _store, talk) = talks();
2072        let mut chat = from_chat(&talk);
2073        chat.runs = vec!["chat-run".to_owned()];
2074        let f1 = followup_of(&chat, 1);
2075        let f2 = followup_of(&f1, 2);
2076        let ts = [chat, f1.clone(), f2.clone()];
2077        let id = Some(talk.id.clone());
2078        let tk = std::slice::from_ref(&talk);
2079        assert_eq!(origin_talk(&ts, tk, &q_for(&f1)).map(|t| t.id), id);
2080        assert_eq!(origin_talk(&ts, tk, &q_for(&f2)).map(|t| t.id), id);
2081    }
2082
2083    #[test]
2084    fn a_followup_without_origin_task_is_found_through_runs() {
2085        let (_tmp, _store, talk) = talks();
2086        let mut chat = from_chat(&talk);
2087        chat.runs = vec!["merged-1".to_owned()];
2088        let mut f1 = followup_of(&chat, 1);
2089        f1.followup.as_mut().unwrap().origin_task = None;
2090        let q = q_for(&f1);
2091        assert!(origin_talk(&[chat, f1], &[talk], &q).is_some());
2092    }
2093
2094    #[test]
2095    fn a_dangling_origin_task_has_no_chat() {
2096        let (_tmp, _store, talk) = talks();
2097        let mut chat = from_chat(&talk);
2098        chat.runs = vec!["chat-run".to_owned()];
2099        let mut f1 = followup_of(&chat, 1);
2100        f1.followup.as_mut().unwrap().origin_task = Some("gone".to_owned());
2101        let q = q_for(&f1);
2102        assert!(origin_talk(&[chat, f1], &[talk], &q).is_none());
2103    }
2104
2105    #[test]
2106    fn a_recorded_chat_survives_deleted_ancestors() {
2107        let (_tmp, _store, talk) = talks();
2108        let chat = from_chat(&talk);
2109        assert_eq!(chat.origin_chat.as_deref(), Some(talk.id.as_str()));
2110        let mut f1 = followup_of(&chat, 1);
2111        f1.origin_chat = chat.origin_chat.clone();
2112        let mut f2 = followup_of(&f1, 2);
2113        f2.origin_chat = f1.origin_chat.clone();
2114        // Neither the chat task nor f1 is in the queue any more.
2115        let q = q_for(&f2);
2116        let got = origin_talk(std::slice::from_ref(&f2), std::slice::from_ref(&talk), &q);
2117        assert_eq!(got.map(|t| t.id), Some(talk.id.clone()));
2118        // The recorded id still has to name an open chat.
2119        assert!(origin_talk(&[f2], &[], &q).is_none());
2120    }
2121
2122    #[test]
2123    fn a_task_without_the_field_still_walks_the_ancestry() {
2124        let (_tmp, _store, talk) = talks();
2125        let mut chat = from_chat(&talk);
2126        chat.origin_chat = None;
2127        let mut f1 = followup_of(&chat, 1);
2128        f1.origin_chat = None;
2129        let q = q_for(&f1);
2130        assert!(origin_talk(&[chat, f1.clone()], std::slice::from_ref(&talk), &q).is_some());
2131        // And the old JSON shape reads with the field absent.
2132        let mut v = serde_json::to_value(&f1).unwrap();
2133        v.as_object_mut().unwrap().remove("origin_chat");
2134        let back: Task = serde_json::from_value(v).unwrap();
2135        assert!(back.origin_chat.is_none());
2136    }
2137
2138    #[test]
2139    fn a_followup_cycle_ends_without_a_chat() {
2140        let (_tmp, _store, talk) = talks();
2141        let mut a = followup_of(&from_chat(&talk), 1);
2142        let mut b = followup_of(&a, 2);
2143        a.followup.as_mut().unwrap().origin_task = Some(b.id.clone());
2144        b.followup.as_mut().unwrap().origin_task = Some(a.id.clone());
2145        let q = q_for(&a);
2146        assert!(origin_talk(&[a, b], &[talk], &q).is_none());
2147    }
2148
2149    #[test]
2150    fn pending_consults_follow_the_store_not_the_turn_text() {
2151        let (tmp, store, talk) = talks();
2152        let questions = Questions::at(tmp.path().join("questions"));
2153        assert!(!pending_consults(&questions, &talk.id), "empty store");
2154
2155        let mut plain = question("implement");
2156        questions.put(&mut plain).unwrap();
2157        assert!(!pending_consults(&questions, &talk.id), "no consult record");
2158
2159        let mut q = question("implement");
2160        questions.put(&mut q).unwrap();
2161        begin(&questions, &store, &q, &talk).unwrap();
2162        assert!(pending_consults(&questions, &talk.id));
2163        assert!(!pending_consults(&questions, "other-talk"));
2164
2165        questions
2166            .update(&q.id, |q| {
2167                q.abandon("test");
2168                Ok(())
2169            })
2170            .unwrap();
2171        assert!(!pending_consults(&questions, &talk.id), "closed question");
2172    }
2173
2174    #[test]
2175    fn begin_reopens_a_closed_chat_before_queueing() {
2176        let (tmp, store, mut talk) = talks();
2177        let questions = Questions::at(tmp.path().join("questions"));
2178        let mut q = question("implement");
2179        questions.put(&mut q).unwrap();
2180        talk::close(&mut talk, &store).unwrap();
2181        assert!(!store.get(&talk.id).unwrap().status.open());
2182
2183        assert!(begin(&questions, &store, &q, &talk).unwrap());
2184
2185        let after = store.get(&talk.id).unwrap();
2186        assert!(after.status.open(), "the chat was reopened");
2187        assert!(after.pending.contains(&q.id), "{}", after.pending);
2188        assert!(questions.get(&q.id).unwrap().consult.is_some());
2189    }
2190
2191    #[test]
2192    fn begin_queues_once_and_leaves_the_question_open() {
2193        let (tmp, store, talk) = talks();
2194        let questions = Questions::at(tmp.path().join("questions"));
2195        let mut q = question("implement");
2196        questions.put(&mut q).unwrap();
2197
2198        assert!(begin(&questions, &store, &q, &talk).unwrap());
2199        assert!(!begin(&questions, &store, &q, &talk).unwrap(), "idempotent");
2200
2201        let after = questions.get(&q.id).unwrap();
2202        assert!(after.status.open());
2203        assert!(after.thread.is_empty());
2204        assert_eq!(after.choices, q.choices, "choices are not touched");
2205        assert_eq!(
2206            after.consult.as_ref().map(|c| c.talk.as_str()),
2207            Some(talk.id.as_str())
2208        );
2209
2210        let destination = store.get(&talk.id).unwrap();
2211        assert_eq!(destination.pending_messages.len(), 1);
2212        assert_eq!(
2213            destination.pending_messages[0].display,
2214            Some(crate::prompt::chat_consult_display(&q))
2215        );
2216        let queued = destination.pending;
2217        assert_eq!(
2218            queued.matches(&q.id).count(),
2219            2 + 1,
2220            "id once per use: {queued}"
2221        );
2222        assert!(queued.contains("Which backend?"));
2223        assert!(queued.contains("SQLite is simpler."));
2224        assert!(queued.contains("- Redis"));
2225        // Queued once, so the first-operator-turn title is not involved.
2226        assert_eq!(
2227            queued.matches(crate::prompt::CHAT_CONSULT_HEADING).count(),
2228            1
2229        );
2230    }
2231
2232    fn block(id: &str, detail: &str) -> String {
2233        let detail = crate::prompt::defuse(detail);
2234        format!(
2235            "# {}\n\nThe operator passed you a question `{id}`. {}\n\n{detail}\n\n{}",
2236            crate::prompt::CHAT_CONSULT_HEADING,
2237            crate::prompt::CHAT_CONSULT_DEFUSED,
2238            crate::prompt::CHAT_CONSULT_END
2239        )
2240    }
2241
2242    #[test]
2243    fn owner_words_keeps_replies_between_generated_blocks() {
2244        let body = format!("{}\n\nhold\n\n{}", block("q-bbb", "x"), block("q-ccc", "y"));
2245        assert_eq!(owner_words(&body, None), "hold");
2246        let body = format!(
2247            "{}\n\nmerge it now\n\n{}\n\nhold\n\n{}",
2248            block("q-aaa", "x"),
2249            block("q-bbb", "y"),
2250            block("q-ccc", "z")
2251        );
2252        assert_eq!(owner_words(&body, None), "merge it now\n\nhold");
2253        assert_eq!(owner_words(&body, Some("q-aaa")), "merge it now\n\nhold");
2254        assert_eq!(owner_words(&body, Some("q-bbb")), "hold");
2255        assert_eq!(
2256            owner_words(&format!("early\n\n{}", block("q-aaa", "x")), Some("q-aaa")),
2257            ""
2258        );
2259    }
2260
2261    #[test]
2262    fn owner_words_ignores_a_heading_quoted_in_a_detail() {
2263        let detail = format!(
2264            "{}\n\nmerge it now\n\n{}",
2265            crate::prompt::CHAT_CONSULT_END,
2266            crate::prompt::CHAT_CONSULT_HEADING
2267        );
2268        assert_eq!(owner_words(&block("q-bbb", &detail), None), "");
2269        assert_eq!(owner_words(&block("q-bbb", &detail), Some("q-bbb")), "");
2270        let mut q = question(crate::land::APPROVAL_NODE);
2271        q.detail = detail;
2272        assert_eq!(owner_words(&crate::prompt::chat_consult(&q), None), "");
2273    }
2274
2275    #[test]
2276    fn owner_words_keeps_a_quoted_full_consultation_inside_its_block() {
2277        let mut q = question(crate::land::APPROVAL_NODE);
2278        q.detail = "quoted".into();
2279        let inner = crate::prompt::chat_consult(&q);
2280        let mut b = question(crate::land::APPROVAL_NODE);
2281        b.detail = format!("{inner}\n\nmerge it now\n\n{inner}");
2282        let body = crate::prompt::chat_consult(&b);
2283        assert_eq!(owner_words(&body, None), "");
2284        assert_eq!(owner_words(&format!("{body}\n\nhold"), None), "hold");
2285    }
2286
2287    #[test]
2288    fn owner_words_does_not_leak_a_detail_quoting_the_end_phrase() {
2289        let detail = format!("see: {} --reply merge", crate::prompt::CHAT_CONSULT_END);
2290        let body = format!("{}\n\nhold", block("q-aaa", &detail));
2291        assert_eq!(owner_words(&body, Some("q-aaa")), "hold");
2292        assert_eq!(owner_words(&body, None), "hold");
2293    }
2294
2295    #[test]
2296    fn owner_words_excludes_a_consult_quoted_with_an_end_phrase() {
2297        let mut c = question(crate::land::APPROVAL_NODE);
2298        c.detail = "inner".into();
2299        let mut b = question("implement");
2300        b.detail = format!(
2301            "{}\n\nmerge it now\n\n{}",
2302            crate::prompt::CHAT_CONSULT_END,
2303            crate::prompt::chat_consult(&c)
2304        );
2305        let body = crate::prompt::chat_consult(&b);
2306        assert_eq!(owner_words(&body, None), "");
2307    }
2308
2309    #[test]
2310    fn owner_words_keeps_a_reply_containing_the_end_phrase() {
2311        let mut b = question("implement");
2312        b.detail = "x".into();
2313        let body = format!(
2314            "{}\n\nhold, and do not {}",
2315            crate::prompt::chat_consult(&b),
2316            crate::prompt::CHAT_CONSULT_END
2317        );
2318        assert_eq!(
2319            owner_words(&body, None),
2320            format!("hold, and do not {}", crate::prompt::CHAT_CONSULT_END)
2321        );
2322    }
2323
2324    /// A consult as stored before quoted markers were defused.
2325    fn legacy_consult(node: &str, raw_detail: &str) -> String {
2326        let mut q = question(node);
2327        q.detail = "@@".into();
2328        crate::prompt::chat_consult(&q)
2329            .replace("@@", raw_detail)
2330            .replace(crate::prompt::CHAT_CONSULT_DEFUSED, "It is still open.")
2331    }
2332
2333    #[test]
2334    fn owner_words_excludes_a_legacy_consult_quoting_the_end_phrase() {
2335        for node in [crate::land::APPROVAL_NODE, "implement"] {
2336            let detail = format!("see {} and more", crate::prompt::CHAT_CONSULT_END);
2337            assert_eq!(owner_words(&legacy_consult(node, &detail), None), "");
2338        }
2339    }
2340
2341    #[test]
2342    fn owner_words_excludes_a_legacy_consult_nesting_a_legacy_consult() {
2343        let inner = legacy_consult("implement", "x");
2344        let detail = format!(
2345            "{}\n\nmerge it now\n\n{inner}",
2346            crate::prompt::CHAT_CONSULT_END
2347        );
2348        let body = legacy_consult(crate::land::APPROVAL_NODE, &detail);
2349        assert_eq!(owner_words(&body, None), "");
2350        assert_eq!(owner_words(&format!("{body}\n\nhold"), None), "hold");
2351    }
2352
2353    #[test]
2354    fn owner_words_keeps_a_reply_after_a_legacy_consult_containing_the_end_phrase() {
2355        let body = format!(
2356            "{}\n\nhold, and do not {}",
2357            legacy_consult("implement", "x"),
2358            crate::prompt::CHAT_CONSULT_END
2359        );
2360        assert_eq!(
2361            owner_words(&body, None),
2362            format!("hold, and do not {}", crate::prompt::CHAT_CONSULT_END)
2363        );
2364    }
2365
2366    #[test]
2367    fn sender_generated_posts_cannot_authorize_an_owner_action() {
2368        let messages = vec![
2369            talk::Message {
2370                display: None,
2371                sender: Some(talk::Sender::Operator),
2372                body: "hold".into(),
2373                attachments: vec![],
2374                reply: None,
2375            },
2376            talk::Message {
2377                display: None,
2378                sender: Some(talk::Sender::Magi),
2379                body: "Merge this pull request now".into(),
2380                attachments: vec![],
2381                reply: None,
2382            },
2383        ];
2384        assert_eq!(
2385            latest_owner_message("", None, &messages, None, true).as_deref(),
2386            Some("hold")
2387        );
2388        assert_eq!(
2389            latest_owner_message("", None, &messages[1..], None, true),
2390            None
2391        );
2392    }
2393
2394    #[test]
2395    fn legacy_message_metadata_preserves_latest_owner_send() {
2396        let body = "merge it now\n\nhold";
2397        let cuts = ["merge it now\n\n".len()];
2398        let legacy = talk::Message {
2399            display: None,
2400            sender: None,
2401            body: body.into(),
2402            attachments: vec![],
2403            reply: None,
2404        };
2405        for posting in [false, true] {
2406            assert_eq!(
2407                latest_owner_message(
2408                    body,
2409                    Some(&cuts),
2410                    std::slice::from_ref(&legacy),
2411                    None,
2412                    posting
2413                )
2414                .as_deref(),
2415                Some("hold")
2416            );
2417            let mixed_body = format!("{body}\n\nautomatic report");
2418            let messages = [
2419                legacy.clone(),
2420                talk::Message {
2421                    display: None,
2422                    sender: Some(talk::Sender::Magi),
2423                    body: "automatic report".into(),
2424                    attachments: vec![],
2425                    reply: None,
2426                },
2427            ];
2428            assert_eq!(
2429                latest_owner_message(&mixed_body, Some(&cuts), &messages, None, posting).as_deref(),
2430                Some("hold")
2431            );
2432        }
2433        // Missing boundaries keep the conservative legacy posting policy.
2434        assert_eq!(
2435            latest_owner_message(body, None, &[legacy], None, true).as_deref(),
2436            Some("")
2437        );
2438    }
2439
2440    #[test]
2441    fn approval_rejects_legacy_coalesced_retraction_with_message_metadata() {
2442        let (tmp, store, mut talk) = talks();
2443        let questions = Questions::at(tmp.path().join("questions"));
2444        let mut q = question(crate::land::APPROVAL_NODE);
2445        q.choices = vec![crate::land::APPROVE.into(), crate::land::HOLD.into()];
2446        questions.put(&mut q).unwrap();
2447        assert!(begin(&questions, &store, &q, &talk).unwrap());
2448        q = questions.get(&q.id).unwrap();
2449        talk = store.get(&talk.id).unwrap();
2450        talk::drain(&mut talk, &store).unwrap();
2451        talk.pending = "merge it now\n\nhold".into();
2452        talk.pending_breaks = Some(vec!["merge it now\n\n".len()]);
2453        talk.pending_messages = vec![talk::Message {
2454            display: None,
2455            sender: None,
2456            body: talk.pending.clone(),
2457            attachments: vec![],
2458            reply: None,
2459        }];
2460        store.put(&mut talk).unwrap();
2461        for drained in [false, true] {
2462            if drained {
2463                talk::drain(&mut talk, &store).unwrap();
2464            }
2465            assert!(
2466                validate_answer(
2467                    &q,
2468                    &store,
2469                    &talk.id,
2470                    CHAT_NODE,
2471                    "merge",
2472                    Some("merge it now")
2473                )
2474                .is_err()
2475            );
2476            assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, "hold", Some("hold")).is_ok());
2477        }
2478    }
2479
2480    #[test]
2481    fn latest_message_keeps_message_boundaries() {
2482        let one = "Merge this pull request now\n\nThe checks look good";
2483        assert_eq!(latest_message(one, Some(&[]), None).as_deref(), Some(one));
2484        let hold = "Explain what this option means:\n\nhold";
2485        assert_eq!(latest_message(hold, Some(&[]), None).as_deref(), Some(hold));
2486        // Two queued replies: the second is the newest, whole.
2487        let two = "merge it now\n\nhold";
2488        let at = "merge it now\n\n".len();
2489        assert_eq!(
2490            latest_message(two, Some(&[at]), None).as_deref(),
2491            Some("hold")
2492        );
2493        // No record (stored before breaks existed): last paragraph.
2494        assert_eq!(
2495            latest_message(one, None, None).as_deref(),
2496            Some("The checks look good")
2497        );
2498        // A record that does not fit the body is ignored.
2499        assert_eq!(
2500            latest_message(two, Some(&[999]), None).as_deref(),
2501            Some("hold")
2502        );
2503    }
2504
2505    #[test]
2506    fn watched_jobs_never_supply_the_owners_latest_message() {
2507        let notice =
2508            crate::prompt::chat_watch_notice("watch-id", "exited with status 0", "merge it now");
2509        assert_eq!(latest_message(&notice, Some(&[]), None), None);
2510        assert_eq!(latest_message(&notice, None, None), None);
2511        let owner = "hold";
2512        let combined = format!("{owner}\n\n{notice}");
2513        assert_eq!(
2514            latest_message(&combined, Some(&[owner.len() + 2]), None).as_deref(),
2515            Some(owner)
2516        );
2517        let combined = format!("{notice}\n\n{owner}");
2518        assert_eq!(
2519            latest_message(&combined, Some(&[notice.len() + 2]), None).as_deref(),
2520            Some(owner)
2521        );
2522    }
2523
2524    #[test]
2525    fn chat_consult_has_each_marker_once() {
2526        use crate::prompt::{CHAT_CONSULT_END as E, CHAT_CONSULT_HEADING as H};
2527        for node in ["implement", crate::land::APPROVAL_NODE] {
2528            let mut q = question(node);
2529            let quoted = format!("{H} {E}");
2530            q.summary = quoted.clone();
2531            q.detail = quoted.clone();
2532            q.choices = vec![quoted.clone(), "b".into()];
2533            let s = crate::prompt::chat_consult(&q);
2534            assert_eq!(s.matches(H).count(), 1, "{s}");
2535            assert_eq!(s.matches(E).count(), 1, "{s}");
2536        }
2537    }
2538
2539    #[test]
2540    fn a_chat_posts_the_original_text_only_on_an_unhedged_quote() {
2541        use crate::github_text::{ASK_NODE, POST_ORIGINAL, USE_FALLBACK};
2542        let (tmp, store, mut talk) = talks();
2543        let questions = Questions::at(tmp.path().join("questions"));
2544        let mut q = question(ASK_NODE);
2545        q.choices = vec![USE_FALLBACK.into(), POST_ORIGINAL.into()];
2546        questions.put(&mut q).unwrap();
2547        assert!(begin(&questions, &store, &q, &talk).unwrap());
2548        q = questions.get(&q.id).unwrap();
2549        let queued = store.get(&talk.id).unwrap().pending;
2550        let owner_turn = |body: &str| talk::Turn {
2551            sender: None,
2552            breaks: Some(Vec::new()),
2553            messages: Vec::new(),
2554            who: talk::Who::Operator,
2555            body: body.into(),
2556            at: Timestamp::now(),
2557            attachments: Vec::new(),
2558            usage: None,
2559            posted_by: None,
2560        };
2561        talk.turns.push(owner_turn(&queued));
2562        talk.pending.clear();
2563        let check = |q: &Question, run: &str, reply: &str, quote: Option<&str>| {
2564            validate_answer(q, &store, run, CHAT_NODE, reply, quote)
2565        };
2566        store.put(&mut talk).unwrap();
2567        // No owner word after the hand-over: nothing to quote.
2568        assert!(check(&q, &talk.id, POST_ORIGINAL, Some("post it")).is_err());
2569        talk.turns.push(owner_turn("Post the original, maybe?"));
2570        store.put(&mut talk).unwrap();
2571        assert!(check(&q, &talk.id, POST_ORIGINAL, Some("Post the original")).is_err());
2572        talk.turns
2573            .push(owner_turn("Post the original text as it is."));
2574        store.put(&mut talk).unwrap();
2575        assert!(check(&q, &talk.id, POST_ORIGINAL, None).is_err());
2576        assert!(check(&q, &talk.id, POST_ORIGINAL, Some("never said")).is_err());
2577        assert!(check(&q, &talk.id, POST_ORIGINAL, Some("Post the original text")).is_ok());
2578        // The safe default stays unguarded, as before.
2579        assert!(check(&q, &talk.id, USE_FALLBACK, None).is_ok());
2580    }
2581
2582    #[test]
2583    fn a_chat_posts_the_rewrite_only_on_an_unhedged_quote() {
2584        use crate::github_text::{ASK_NODE, USE_FALLBACK, USE_REWRITE};
2585        let (tmp, store, mut talk) = talks();
2586        let questions = Questions::at(tmp.path().join("questions"));
2587        let mut q = question(ASK_NODE);
2588        q.choices = vec![USE_FALLBACK.into(), USE_REWRITE.into()];
2589        questions.put(&mut q).unwrap();
2590        assert!(begin(&questions, &store, &q, &talk).unwrap());
2591        q = questions.get(&q.id).unwrap();
2592        let queued = store.get(&talk.id).unwrap().pending;
2593        let owner_turn = |body: &str| talk::Turn {
2594            sender: None,
2595            breaks: Some(Vec::new()),
2596            messages: Vec::new(),
2597            who: talk::Who::Operator,
2598            body: body.into(),
2599            at: Timestamp::now(),
2600            attachments: Vec::new(),
2601            usage: None,
2602            posted_by: None,
2603        };
2604        talk.turns.push(owner_turn(&queued));
2605        talk.pending.clear();
2606        let check = |q: &Question, run: &str, reply: &str, quote: Option<&str>| {
2607            validate_answer(q, &store, run, CHAT_NODE, reply, quote)
2608        };
2609        store.put(&mut talk).unwrap();
2610        // No owner word after the hand-over: nothing to quote.
2611        assert!(check(&q, &talk.id, USE_REWRITE, Some("post it")).is_err());
2612        talk.turns.push(owner_turn("Use the rewrite, maybe?"));
2613        store.put(&mut talk).unwrap();
2614        assert!(check(&q, &talk.id, USE_REWRITE, Some("Use the rewrite")).is_err());
2615        talk.turns
2616            .push(owner_turn("Use the rewrite text as it is."));
2617        store.put(&mut talk).unwrap();
2618        assert!(check(&q, &talk.id, USE_REWRITE, None).is_err());
2619        assert!(check(&q, &talk.id, USE_REWRITE, Some("never said")).is_err());
2620        assert!(check(&q, &talk.id, USE_REWRITE, Some("Use the rewrite text")).is_ok());
2621        // The safe default stays unguarded, as before.
2622        assert!(check(&q, &talk.id, USE_FALLBACK, None).is_ok());
2623    }
2624
2625    #[test]
2626    fn merge_repair_consult_requires_offered_choice_and_exact_latest_owner_request() {
2627        use crate::land::{APPROVAL_NODE, APPROVE, HOLD, REPAIR};
2628        let (tmp, store, mut talk) = talks();
2629        let questions = Questions::at(tmp.path().join("questions"));
2630        let mut q = question(APPROVAL_NODE);
2631        q.choices = vec![APPROVE.into(), HOLD.into(), REPAIR.into()];
2632        questions.put(&mut q).unwrap();
2633        begin(&questions, &store, &q, &talk).unwrap();
2634        q = questions.get(&q.id).unwrap();
2635        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, REPAIR, Some(REPAIR)).is_err());
2636        talk = store.get(&talk.id).unwrap();
2637        talk.turns.push(talk::Turn {
2638            sender: None,
2639            breaks: Some(Vec::new()),
2640            messages: Vec::new(),
2641            posted_by: None,
2642            who: talk::Who::Operator,
2643            body: talk.pending.clone(),
2644            at: Timestamp::now(),
2645            attachments: Vec::new(),
2646            usage: None,
2647        });
2648        talk.pending_messages.clear(); // This fixture now replaces the generated draft.
2649        talk.pending = "maybe fix findings and re-review".into();
2650        store.put(&mut talk).unwrap();
2651        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, REPAIR, Some(REPAIR)).is_err());
2652        talk.pending = REPAIR.into();
2653        talk.pending_breaks = Some(Vec::new());
2654        store.put(&mut talk).unwrap();
2655        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, REPAIR, Some(REPAIR)).is_ok());
2656        q.choices.retain(|c| c != REPAIR);
2657        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, REPAIR, Some(REPAIR)).is_err());
2658    }
2659
2660    #[test]
2661    fn posting_consult_saves_only_authorized_offered_choices() {
2662        use crate::github_text::{ASK_NODE, POST_ORIGINAL, USE_FALLBACK, USE_REWRITE};
2663        for choice in [POST_ORIGINAL, USE_REWRITE] {
2664            let (tmp, store, mut talk) = talks();
2665            let questions = Questions::at(tmp.path().join("questions"));
2666            let mut q = question(ASK_NODE);
2667            q.choices = vec![USE_FALLBACK.into(), choice.into()];
2668            questions.put(&mut q).unwrap();
2669            begin(&questions, &store, &q, &talk).unwrap();
2670            q = questions.get(&q.id).unwrap();
2671            let queued = store.get(&talk.id).unwrap().pending;
2672            assert!(queued.contains("public posting confirmation"));
2673            assert!(queued.contains("not a merge approval"));
2674            assert!(
2675                queued.contains("`use fallback` choice can be saved once the owner has chosen it")
2676            );
2677            assert!(queued.contains("--quote <verbatim owner words>"));
2678            assert!(!queued.contains("answer it yourself with"));
2679            talk.turns.push(talk::Turn {
2680                sender: None,
2681                breaks: Some(Vec::new()),
2682                messages: Vec::new(),
2683                who: talk::Who::Operator,
2684                body: queued,
2685                at: Timestamp::now(),
2686                attachments: Vec::new(),
2687                usage: None,
2688                posted_by: None,
2689            });
2690            talk.pending.clear();
2691            store.put(&mut talk).unwrap();
2692            let save = |reply: &str, quote: Option<&str>, run: &str| {
2693                questions.update(&q.id, |r| {
2694                    validate_answer(r, &store, run, CHAT_NODE, reply, quote)?;
2695                    r.answer(crate::ask::Answer::Choice(reply.into()))
2696                })
2697            };
2698            // No owner reply, nor text from the generated question, is evidence.
2699            assert!(save(choice, Some(choice), &talk.id).is_err());
2700            let clear = if choice == POST_ORIGINAL {
2701                "元の説明のまま投稿して"
2702            } else {
2703                "Use the rewrite text as it is."
2704            };
2705            talk.pending = clear.into();
2706            store.put(&mut talk).unwrap();
2707            for quote in [None, Some("forged quote")] {
2708                let error = save(choice, quote, &talk.id).unwrap_err().to_string();
2709                assert!(error.contains("posting confirmation"), "{error}");
2710                assert!(
2711                    !error.contains("merge") && !error.contains("hold"),
2712                    "{error}"
2713                );
2714            }
2715            assert!(save(choice, Some(clear), "another-chat").is_err());
2716            let other = if choice == POST_ORIGINAL {
2717                USE_REWRITE
2718            } else {
2719                POST_ORIGINAL
2720            };
2721            assert!(save(other, Some(clear), &talk.id).is_err());
2722            let mut unconsulted = q.clone();
2723            unconsulted.consult = None;
2724            assert!(
2725                validate_answer(
2726                    &unconsulted,
2727                    &store,
2728                    &talk.id,
2729                    CHAT_NODE,
2730                    choice,
2731                    Some(clear)
2732                )
2733                .is_err()
2734            );
2735            let mut expired = q.clone();
2736            expired.abandon("expired");
2737            assert!(
2738                validate_answer(&expired, &store, &talk.id, CHAT_NODE, choice, Some(clear))
2739                    .is_err()
2740            );
2741            talk.status = TalkStatus::Closed;
2742            store.put(&mut talk).unwrap();
2743            assert!(save(choice, Some(clear), &talk.id).is_err());
2744            talk.status = TalkStatus::Open;
2745            for rejected in [
2746                "Post original was the option I saw.",
2747                "Use rewrite was the option I saw.",
2748                "Post the original, maybe?",
2749                "Use the rewrite only if checks pass.",
2750                "Post original. Cancel that.",
2751                "Do not post original.",
2752                "Use the rewrite when checks pass.",
2753                "元の説明で投稿してください。ただし検査が通ったら。",
2754                "元の説明で投稿しないで。",
2755                "まだ元の説明で投稿してください。",
2756                "ログには「元の説明で投稿しました」と記録されています。",
2757                "ログには「修正版で投稿しました」と記録されています。",
2758            ] {
2759                talk.pending = rejected.into();
2760                store.put(&mut talk).unwrap();
2761                let error = save(choice, Some(rejected), &talk.id)
2762                    .unwrap_err()
2763                    .to_string();
2764                assert!(
2765                    !error.contains("merge") && !error.contains("hold"),
2766                    "{error}"
2767                );
2768                assert!(save(choice, Some(clear), &talk.id).is_err(), "stale quote");
2769                let after = questions.get(&q.id).unwrap();
2770                assert!(after.status.open());
2771                assert!(after.answer.is_none());
2772            }
2773            let command = if choice == POST_ORIGINAL {
2774                "Post original"
2775            } else {
2776                "Use rewrite"
2777            };
2778            // A newer report must also block fallback to an older instruction.
2779            let mut earlier = talk.turns[0].clone();
2780            earlier.body = clear.into();
2781            talk.turns.push(earlier);
2782            for reported in [
2783                format!("The log contains:\n```\n{command}\n```"),
2784                format!("The log contains:\n\"\n{command}\n\""),
2785                format!("The log says: “Reported. {command}.”"),
2786                format!("The log says: “\n{command}\n”"),
2787            ]
2788            .into_iter()
2789            .chain(crate::github_text::reported_posting_messages(command))
2790            {
2791                talk.pending.clear();
2792                talk.pending_messages.clear();
2793                store.put(&mut talk).unwrap();
2794                talk::queue(&mut talk, &store, &reported, Vec::new()).unwrap();
2795                assert_eq!(talk.pending, reported);
2796                // Check both current message boundaries and legacy records.
2797                for breaks in [Some(Vec::new()), None] {
2798                    talk.pending_breaks = breaks;
2799                    store.put(&mut talk).unwrap();
2800                    for quote in [command, reported.as_str()] {
2801                        assert!(save(choice, Some(quote), &talk.id).is_err());
2802                        let after = questions.get(&q.id).unwrap();
2803                        assert!(after.status.open());
2804                        assert!(after.answer.is_none());
2805                    }
2806                }
2807                // Drained transcript evidence must retain the same context.
2808                talk::drain(&mut talk, &store).unwrap();
2809                for quote in [command, reported.as_str()] {
2810                    assert!(save(choice, Some(quote), &talk.id).is_err());
2811                    let after = questions.get(&q.id).unwrap();
2812                    assert!(after.status.open());
2813                    assert!(after.answer.is_none());
2814                }
2815            }
2816            let natural = if choice == POST_ORIGINAL {
2817                [
2818                    "OK, post the original",
2819                    "Yes, post original",
2820                    "I'd like to post the original",
2821                    "元の説明で投稿してください。",
2822                ]
2823            } else {
2824                [
2825                    "OK, use the rewrite",
2826                    "Yes, use rewrite",
2827                    "I'd like to use the rewrite",
2828                    "修正版で投稿してください。",
2829                ]
2830            };
2831            for words in natural {
2832                talk.pending = words.into();
2833                store.put(&mut talk).unwrap();
2834                validate_answer(&q, &store, &talk.id, CHAT_NODE, choice, Some(words)).unwrap();
2835            }
2836            talk.pending = clear.into();
2837            store.put(&mut talk).unwrap();
2838            save(choice, Some(clear), &talk.id).unwrap();
2839            assert_eq!(
2840                questions.get(&q.id).unwrap().resolution().as_deref(),
2841                Some(choice)
2842            );
2843        }
2844    }
2845
2846    #[test]
2847    fn approval_consult_waits_for_latest_owner_confirmation() {
2848        let (tmp, store, mut talk) = talks();
2849        let questions = Questions::at(tmp.path().join("questions"));
2850        let mut q = question(crate::land::APPROVAL_NODE);
2851        q.choices = vec![crate::land::APPROVE.into(), crate::land::HOLD.into()];
2852        questions.put(&mut q).unwrap();
2853        assert!(begin(&questions, &store, &q, &talk).unwrap());
2854        assert!(!begin(&questions, &store, &q, &talk).unwrap());
2855        q = questions.get(&q.id).unwrap();
2856        assert!(q.status.open());
2857        assert!(q.answer.is_none());
2858        assert!(q.thread.is_empty());
2859        let queued = store.get(&talk.id).unwrap().pending;
2860        assert!(queued.contains("Never answer it yourself"));
2861        assert!(queued.contains("Silence holds"));
2862        assert!(queued.contains("--reply merge --quote"));
2863        assert!(!queued.contains("answer it yourself with"));
2864
2865        let owner_turn = |body: &str| talk::Turn {
2866            sender: None,
2867            breaks: Some(Vec::new()),
2868            messages: Vec::new(),
2869            posted_by: None,
2870            who: talk::Who::Operator,
2871            body: body.into(),
2872            at: Timestamp::now(),
2873            attachments: Vec::new(),
2874            usage: None,
2875        };
2876        // An approval the owner gave before the hand-over is not reusable.
2877        let mut old = talk.clone();
2878        old.turns
2879            .insert(0, owner_turn("Please merge PR 12 after review"));
2880        store.put(&mut old).unwrap();
2881        talk.turns = old.turns.clone();
2882        // The drained consult draft is stored as an operator turn; it must not
2883        // count as the owner's words even though it contains "merge".
2884        talk.turns.push(owner_turn(&queued));
2885        store.put(&mut talk).unwrap();
2886        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, "merge", Some("merge")).is_err());
2887        assert!(
2888            validate_answer(
2889                &q,
2890                &store,
2891                &talk.id,
2892                CHAT_NODE,
2893                "merge",
2894                Some("Please merge PR 12")
2895            )
2896            .is_err()
2897        );
2898        // An owner reply coalesced onto the same draft still counts.
2899        let n = talk.turns.len();
2900        talk.turns[n - 1].body = format!("{queued}\n\nmerge it now");
2901        talk.turns[n - 1].breaks = Some(vec![queued.len() + 2]);
2902        store.put(&mut talk).unwrap();
2903        assert!(
2904            validate_answer(
2905                &q,
2906                &store,
2907                &talk.id,
2908                CHAT_NODE,
2909                "merge",
2910                Some("merge it now")
2911            )
2912            .is_ok()
2913        );
2914        // A later retraction in the same coalesced turn wins.
2915        talk.turns[n - 1].body = format!("{queued}\n\nmerge it now\n\nhold");
2916        talk.turns[n - 1].breaks = Some(vec![queued.len() + 2, queued.len() + 2 + 14]);
2917        store.put(&mut talk).unwrap();
2918        assert!(
2919            validate_answer(
2920                &q,
2921                &store,
2922                &talk.id,
2923                CHAT_NODE,
2924                "merge",
2925                Some("merge it now")
2926            )
2927            .is_err()
2928        );
2929        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, "hold", Some("hold")).is_ok());
2930        // So does a reply still waiting in the draft.
2931        talk.turns[n - 1].body = format!("{queued}\n\nmerge it now");
2932        talk.turns[n - 1].breaks = Some(vec![queued.len() + 2]);
2933        talk.pending = "hold".into();
2934        store.put(&mut talk).unwrap();
2935        assert!(
2936            validate_answer(
2937                &q,
2938                &store,
2939                &talk.id,
2940                CHAT_NODE,
2941                "merge",
2942                Some("merge it now")
2943            )
2944            .is_err()
2945        );
2946        // Another question's hand-over queued after the retraction hides nothing.
2947        talk.pending = format!("hold\n\n{queued}");
2948        store.put(&mut talk).unwrap();
2949        assert!(
2950            validate_answer(
2951                &q,
2952                &store,
2953                &talk.id,
2954                CHAT_NODE,
2955                "merge",
2956                Some("merge it now")
2957            )
2958            .is_err()
2959        );
2960        talk.pending.clear();
2961        talk.turns[n - 1].body = queued.clone();
2962        store.put(&mut talk).unwrap();
2963        talk.turns
2964            .push(owner_turn("Merge this pull request please"));
2965        store.put(&mut talk).unwrap();
2966        let check = |q: &Question, run: &str, reply: &str, quote: Option<&str>| {
2967            validate_answer(q, &store, run, CHAT_NODE, reply, quote)
2968        };
2969        assert!(check(&q, "other-talk", "merge", Some("Merge this")).is_err());
2970        assert!(check(&q, &talk.id, "merge", None).is_err());
2971        assert!(check(&q, &talk.id, "merge", Some("never said")).is_err());
2972        assert!(
2973            check(
2974                &q,
2975                &talk.id,
2976                "merge",
2977                Some("Merge this pull request please")
2978            )
2979            .is_ok()
2980        );
2981        assert!(check(&q, &talk.id, "hold", Some("hold")).is_err());
2982        talk.turns
2983            .push(owner_turn("Wait, explain the checks first"));
2984        store.put(&mut talk).unwrap();
2985        assert!(
2986            check(
2987                &q,
2988                &talk.id,
2989                "merge",
2990                Some("Merge this pull request please")
2991            )
2992            .is_err()
2993        );
2994        talk.turns.push(owner_turn("Please hold"));
2995        store.put(&mut talk).unwrap();
2996        assert!(check(&q, &talk.id, "hold", Some("hold")).is_err());
2997        talk.turns.push(owner_turn("hold"));
2998        store.put(&mut talk).unwrap();
2999        assert!(check(&q, &talk.id, "hold", Some("hold")).is_ok());
3000        talk.turns
3001            .push(owner_turn("Merge this pull request please"));
3002        store.put(&mut talk).unwrap();
3003        q.abandon("approval expired or head changed");
3004        assert!(
3005            check(
3006                &q,
3007                &talk.id,
3008                "merge",
3009                Some("Merge this pull request please")
3010            )
3011            .is_err()
3012        );
3013        assert!(
3014            q.answer(crate::ask::Answer::Choice("merge".into()))
3015                .is_err()
3016        );
3017    }
3018
3019    #[test]
3020    fn begin_withdraws_its_record_when_the_talk_cannot_take_it() {
3021        let (tmp, store, talk) = talks();
3022        let questions = Questions::at(tmp.path().join("questions"));
3023        let mut q = question("implement");
3024        questions.put(&mut q).unwrap();
3025        std::fs::remove_dir_all(tmp.path().join("talks")).unwrap();
3026        assert!(begin(&questions, &store, &q, &talk).is_err());
3027        assert!(questions.get(&q.id).unwrap().consult.is_none());
3028    }
3029
3030    #[test]
3031    fn an_older_question_file_reads_without_a_consult() {
3032        let mut q = question("implement");
3033        q.schema = 5;
3034        let mut v = serde_json::to_value(&q).unwrap();
3035        v.as_object_mut().unwrap().remove("consult");
3036        let back: Question = serde_json::from_value(v).unwrap();
3037        assert!(back.consult.is_none());
3038    }
3039}