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