magi/

consult.rs

//! Handing an open question to the chat conversation its task came from.
//!
//! A task filed by the standing chat (`Source::Agent` with
//! [`crate::queue::CHAT_NODE`], whose `run` is the conversation's id) can raise
//! a question the owner would rather talk through where the task started. The
//! owner passes it on; the chat agent either answers it with `magi answer` or
//! puts the decision points to the owner in the conversation.
//!
//! Three rules, each with a reason:
//!
//! - **One place decides.** [`origin_talk`] is the only judge of "does this
//!   question have a chat to ask"; the web view and both entry points use it.
//!   [`associated_talk`] resolves the same association for navigation even
//!   when the question can no longer be consulted.
//! - **Not a choice.** The hand-over never goes into `Question::choices` and
//!   never touches `Question::answer`: the question stays `Open`, its
//!   validation and the choice actions stay as they were.
//! - **No new seat, no new waiter.** The text is queued as a draft of the
//!   existing talk, so the chat's own turn machinery (its session, its turn
//!   gate) runs it exactly as if the owner had typed it.

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

use crate::ask::{ChatConsult, Question, Questions};
use crate::config::Config;
use crate::queue::{CHAT_NODE, Task};
use crate::talk::{self, Talk, Talks};

/// The conversation `q` may be handed to, if any.
///
/// Requires an open question whose task was filed from a chat. The chat may
/// be closed: [`begin`] reopens it. Release notices are left out. Merge
/// approvals may be discussed, but answering from chat requires the owner's latest words.
pub fn origin_talk(tasks: &[Task], talks: &[Talk], q: &Question) -> Option<Talk> {
    if !q.status.open() || q.node == crate::bump::NOTICE_NODE {
        return None;
    }
    associated_talk(tasks, talks, q)
}

/// The existing conversation associated with a question, regardless of whether
/// the question is still eligible for consultation.
pub fn associated_talk(tasks: &[Task], talks: &[Talk], q: &Question) -> Option<Talk> {
    let task = crate::daemon::task_of_question(tasks, q)?;
    let run = chat_talk_of(tasks, task)?;
    talks.iter().find(|t| t.id == run).cloned()
}

/// The talk `start` descends from: the id recorded on a task
/// ([`Task::chat_talk`]) wins, so deleted ancestors do not matter; tasks
/// written before it existed are resolved by walking the ancestry.
///
/// Walk the provenance of `start` to the task a chat filed.
///
/// A follow-up goes to the task its merged run served: `FollowUp::origin_task`
/// when recorded (a missing one ends the walk - guessing another task could
/// hand the question to an unrelated chat), else the task whose `runs` hold
/// `FollowUp::run`. At most `MAX_FOLLOWUP_GENERATION + 1` tasks are looked at,
/// each once, so a cycle ends in `None`.
pub fn chat_talk_of(tasks: &[Task], start: &Task) -> Option<String> {
    let mut seen = std::collections::HashSet::new();
    let mut cur = start;
    for _ in 0..=crate::followup::MAX_FOLLOWUP_GENERATION {
        if !seen.insert(cur.id.as_str()) {
            return None;
        }
        if let Some(id) = cur.chat_talk() {
            return Some(id.to_owned());
        }
        let f = cur.followup.as_ref()?;
        cur = match &f.origin_task {
            Some(id) => tasks.iter().find(|t| &t.id == id)?,
            None => tasks.iter().find(|t| t.runs.contains(&f.run))?,
        };
    }
    None
}

/// Is any question handed to talk `talk_id` still open?
///
/// Decided from the question store, not from the newest turn's wording: the
/// owner's decision usually arrives in a later turn that carries no hand-over
/// heading, and `magi answer` still has to be able to write then. Answered or
/// abandoned questions stop counting, so the turn goes back to read-only.
pub fn pending_consults(questions: &Questions, talk_id: &str) -> bool {
    questions
        .list()
        .iter()
        .any(|q| q.status.open() && q.consult.as_ref().is_some_and(|c| c.talk == talk_id))
}

/// Record the hand-over and queue the question as the chat's next message.
///
/// Returns `false` (and changes nothing) when the question was already handed
/// over. The question is not otherwise touched: still `Open`, no thread turn.
/// When queueing fails the record is withdrawn, so the owner can try again.
pub fn begin(questions: &Questions, talks: &Talks, q: &Question, talk: &Talk) -> Result<bool> {
    let (q, fresh) = questions.update(&q.id, |r| {
        if !r.status.open() {
            bail!("question {} is already {}", r.short(), r.status.as_str());
        }
        if r.consult.is_some() {
            return Ok(false);
        }
        r.consult = Some(ChatConsult {
            talk: talk.id.clone(),
            at: Timestamp::now(),
        });
        Ok(true)
    })?;
    if !fresh {
        return Ok(false);
    }
    // Serialize dispatch with answers, but never wait for conversation locks
    // under the expiring question lock. Each retry reloads the latest question.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
    let dispatch = loop {
        let mut busy = false;
        let updated = questions.update(&q.id, |r| {
            if !r.status.open() || r.consult != q.consult {
                return Ok(false);
            }
            if talk::try_queue_consult(
                talks,
                &talk.id,
                &crate::prompt::chat_consult(r),
                Some(crate::prompt::chat_consult_display(r)),
            )?
            .is_none()
            {
                busy = true;
                bail!("the conversation store is locked by another process");
            }
            Ok(true)
        });
        if busy && std::time::Instant::now() < deadline {
            std::thread::sleep(std::time::Duration::from_millis(10));
            continue;
        }
        break updated;
    };
    if let Err(e) = dispatch {
        let _ = questions.update(&q.id, |r| {
            if r.status.open() && r.consult == q.consult {
                r.consult = None;
            }
            Ok(())
        });
        return Err(e).context("queue the question into the chat");
    }
    Ok(dispatch?.1)
}

/// Refuse a chat's answer to a merge approval whose task the operator held, as
/// `ask_settle_cmd` does for a deputy: such a hold is lifted by the operator's
/// button. The task is the one `q.run` names, directly or as one of its runs;
/// none found (an unreadable task is not listed) does not refuse. The caller
/// reads the queue before taking the question lock, so a hold placed in between
/// is missed for that one call.
pub fn refuse_operator_held(q: &Question, node: &str, tasks: &[crate::queue::Task]) -> Result<()> {
    if q.node != crate::land::APPROVAL_NODE || node != CHAT_NODE {
        return Ok(());
    }
    if let Some(task) = tasks
        .iter()
        .find(|t| t.id == q.run || t.runs.contains(&q.run))
        && task.operator_held()
    {
        bail!(
            "task {} is held by the operator; an answer cannot settle it",
            task.short()
        );
    }
    Ok(())
}

/// Authorization policy is selected by the question's node, never its reply.
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum QuestionKind {
    /// A question without a guarded public decision.
    Ordinary,
    /// Approval to merge a change.
    Merge,
    /// Confirmation of the text to publish.
    Posting,
}

/// Owner evidence retained with an answer, independent of its consultation destination.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AnswerAuthority {
    /// Conversation containing the instruction.
    pub talk: String,
    /// Zero-based persisted turn position.
    pub turn: usize,
    /// Zero-based send position within that turn.
    pub message: usize,
    /// Full instruction, not a substring selected by the assistant.
    pub instruction: String,
    /// Stable fingerprint of the complete instruction.
    pub instruction_fingerprint: String,
    /// Immediately preceding persona reply, for a batch instruction.
    pub recommendation_turn: Option<usize>,
    /// Fingerprint of the complete persona reply used as the batch manifest.
    pub recommendation_fingerprint: Option<String>,
    /// Exact decision covered by this evidence.
    pub question: String,
    /// Exact offered choice; this evidence path only authorizes hold.
    pub choice: String,
}

/// Validate an answer under the question lock. The caller must also retain
/// [`Talks::answer_guard`] from before this read until the answer is persisted.
/// Consultation is a delivery route, not the source of authority for an
/// explicit, independently bound owner hold.
pub fn authorize_answer(
    q: &Question,
    talks: &Talks,
    questions: &Questions,
    run: &str,
    node: &str,
    reply: &str,
    quote: Option<&str>,
) -> Result<Option<AnswerAuthority>> {
    if node == CHAT_NODE && QuestionKind::of(q) == QuestionKind::Merge && reply == crate::land::HOLD
    {
        if q.actions.contains_key(crate::land::HOLD) {
            bail!("hold cannot execute a question action");
        }
        let talk = talks.get(run)?;
        if let Some(evidence) = hold_authority(q, questions, &talk, quote) {
            return Ok(Some(evidence));
        }
        if q.consult.as_ref().is_none_or(|c| c.talk != run) {
            bail!(
                "hold requires --quote of the entire latest owner send, either `<question-id>: hold` \
                 or an explicit instruction to answer each immediately preceding question-by-question \
                 recommendation; the persisted send and exact question/hold binding must be current"
            );
        }
        // Keep legacy same-chat single-word holds, as well as its merge and
        // repair policies. A batch instruction never expands those policies.
    }
    validate_answer(q, talks, run, node, reply, quote)?;
    Ok(None)
}

/// Whether this turn needs the limited question-store write grant. This uses
/// the same evidence check as the CLI; the CLI rechecks against fresh records.
pub fn authorized_holds(questions: &Questions, talk: &Talk) -> bool {
    let quote = current_instruction(talk).map(|(_, _, body)| body);
    questions.list().iter().any(|q| {
        QuestionKind::of(q) == QuestionKind::Merge
            && hold_authority(q, questions, talk, quote).is_some()
    })
}

fn current_instruction(talk: &Talk) -> Option<(usize, usize, &str)> {
    // A newer queued send can retract the instruction. Wait until it is
    // drained rather than letting the running seat use older words. Explicitly
    // generated sends neither authorize nor retract an owner instruction.
    let queued_owner = (!talk.pending.is_empty() || !talk.pending_messages.is_empty())
        && (talk.pending_messages.is_empty()
            || talk
                .pending_messages
                .iter()
                .any(|m| m.sender != Some(talk::Sender::Magi)));
    if !talk.status.open() || queued_owner {
        return None;
    }
    for (index, turn) in talk.turns.iter().enumerate().rev() {
        if talk::sender(turn, None) == talk::Sender::Magi {
            continue;
        }
        // A persona reply consumes the conversational turn. Never reuse an
        // older instruction just because a generated post followed that reply.
        if turn.who != talk::Who::Operator || turn.posted_by.is_some() {
            return None;
        }
        let latest = turn
            .messages
            .iter()
            .enumerate()
            .rfind(|(_, send)| talk::sender(turn, Some(send)) == talk::Sender::Operator);
        let Some((message, send)) = latest else {
            if !turn.messages.is_empty()
                && turn
                    .messages
                    .iter()
                    .all(|m| m.sender == Some(talk::Sender::Magi))
            {
                continue;
            }
            return None;
        };
        // Older real sends can be reconstructed from recorded boundaries,
        // never from a guessed last paragraph. The whole-message grammar also
        // excludes generated legacy wrappers.
        if send.sender.is_none()
            && (turn.breaks.is_none()
                || turn
                    .messages
                    .iter()
                    .map(|m| m.body.as_str())
                    .collect::<Vec<_>>()
                    .join("\n\n")
                    != turn.body)
        {
            return None;
        }
        return Some((index, message, &send.body));
    }
    None
}

fn hold_authority(
    q: &Question,
    questions: &Questions,
    talk: &Talk,
    quote: Option<&str>,
) -> Option<AnswerAuthority> {
    if !q.status.open()
        || !q.choices.iter().any(|c| c == crate::land::HOLD)
        || q.actions.contains_key(crate::land::HOLD)
    {
        return None;
    }
    let (turn, message, instruction) = current_instruction(talk)?;
    if quote.map(str::trim) != Some(instruction.trim()) || talk.turns[turn].at < q.asked_at {
        return None;
    }
    let direct = direct_hold(instruction, q, questions);
    let recommendation_turn = if direct {
        None
    } else {
        if !batch_instruction(instruction) {
            return None;
        }
        let index = turn.checked_sub(1)?;
        let recommendation = &talk.turns[index];
        if recommendation.who != talk::Who::Agent
            || talk::sender(recommendation, None) != talk::Sender::Persona
            || recommendation.posted_by.is_some()
            || recommendation.at < q.asked_at
            || !recommended_hold(&recommendation.body, q, questions)
        {
            return None;
        }
        Some(index)
    };
    Some(AnswerAuthority {
        talk: talk.id.clone(),
        turn,
        message,
        instruction: instruction.to_owned(),
        instruction_fingerprint: crate::github_text::fingerprint("", instruction),
        recommendation_turn,
        recommendation_fingerprint: recommendation_turn
            .map(|i| crate::github_text::fingerprint("", &talk.turns[i].body)),
        question: q.id.clone(),
        choice: crate::land::HOLD.into(),
    })
}

fn batch_instruction(text: &str) -> bool {
    // Whole-message grammar: substrings in conditional, reported, quoted or
    // retracted instructions cannot match. Keep this deliberately bounded.
    let compact: String = text.chars().filter(|c| !c.is_whitespace()).collect();
    let compact = compact.trim_end_matches(['!', '！', '。', '.', 'ー']);
    let compact = compact
        .strip_prefix("ありがと！")
        .or_else(|| compact.strip_prefix("ありがとう！"))
        .unwrap_or(compact);
    matches!(
        compact,
        "じゃあそれぞれ回答して" | "それぞれ回答して" | "それぞれ回答してください"
    ) || matches!(
        text.trim()
            .trim_end_matches(['.', '!'])
            .to_ascii_lowercase()
            .as_str(),
        "answer each as recommended" | "answer all as recommended"
    )
}

fn refers_to(reference: &str, q: &Question, questions: &Questions) -> bool {
    reference == q.id
        || (reference.len() >= 4
            && reference == q.short()
            && questions.resolve_id(reference).is_ok_and(|id| id == q.id))
}

fn direct_hold(text: &str, q: &Question, questions: &Questions) -> bool {
    // One complete instruction, binding both the question and the offered
    // choice. Other actions (including closing the PR) are never inferred.
    let Some((reference, choice)) = text.trim().split_once(':') else {
        return false;
    };
    refers_to(reference.trim().trim_matches('`'), q, questions) && choice.trim() == "hold"
}

fn recommended_hold(body: &str, q: &Question, questions: &Questions) -> bool {
    // Supported list/table rows: `<id>: hold`, `<id> → hold`, or `| <id> | hold |`.
    // A row must select exactly one offered choice, rather than list options.
    // Repeated or conflicting rows are ambiguous even if one recommends hold.
    let mut rows = Vec::new();
    // Use the same Markdown parser as the transcript renderer. Quoted lazy
    // continuations, nested fences and indented examples are not recommendations.
    use comrak::nodes::NodeValue;
    let arena = comrak::Arena::new();
    let mut options = comrak::Options::default();
    options.extension.table = true;
    let document = comrak::parse_document(&arena, body, &options);
    let lines: Vec<_> = body.lines().collect();
    let mut eligible = std::collections::BTreeSet::new();
    for node in document.descendants() {
        let data = node.data.borrow();
        if node
            .ancestors()
            .any(|n| matches!(n.data.borrow().value, NodeValue::BlockQuote))
        {
            continue;
        }
        // A heading, introduction or trailing paragraph may qualify several
        // rows without naming any question. Do not guess its scope: batch
        // authority requires the whole unquoted recommendation to be clear.
        if matches!(
            data.value,
            NodeValue::Heading(_) | NodeValue::Paragraph | NodeValue::TableRow(_)
        ) && lines
            .get(data.sourcepos.start.line - 1..data.sourcepos.end.line)
            .is_some_and(|block| conditional_recommendation(&block.join(" ")))
        {
            return false;
        }
        if matches!(
            data.value,
            NodeValue::Paragraph | NodeValue::TableRow(false)
        ) {
            eligible.extend(data.sourcepos.start.line..=data.sourcepos.end.line);
        }
    }
    for (index, line) in body.lines().enumerate() {
        if !eligible.contains(&(index + 1)) {
            continue;
        }
        let line = line.trim();
        let tokens: Vec<&str> = line
            .split(|c: char| !c.is_ascii_alphanumeric() && c != '-')
            .collect();
        if !tokens.iter().any(|r| refers_to(r, q, questions)) {
            continue;
        }
        if line.starts_with('|') {
            let cells: Vec<_> = line
                .trim_matches('|')
                .split('|')
                .map(|c| c.trim().trim_matches('`'))
                .collect();
            let references = cells.iter().filter(|c| refers_to(c, q, questions)).count();
            let other_question = cells
                .iter()
                .any(|c| c.len() >= 4 && questions.resolve_id(c).is_ok_and(|id| id != q.id));
            let choices: Vec<_> = cells
                .iter()
                .filter(|c| q.choices.iter().any(|choice| choice == **c))
                .collect();
            rows.push(!other_question && references == 1 && choices == [&crate::land::HOLD]);
            continue;
        }
        let row = line.trim_start_matches(['-', '*', ' ']);
        let row = row.trim_start_matches('`');
        let end = row
            .find(|c: char| !c.is_ascii_alphanumeric() && c != '-')
            .unwrap_or(row.len());
        let (reference, rest) = row.split_at(end);
        let rest = rest.trim_start_matches('`').trim_start();
        let choice = rest
            .strip_prefix(':')
            .or_else(|| rest.strip_prefix('→'))
            .or_else(|| rest.strip_prefix('|'))
            .map(str::trim);
        rows.push(
            refers_to(reference, q, questions)
                && choice.is_some_and(|choice| {
                    // Consume the entire choice. A pipe is an alternative,
                    // not an excuse to discard it. The one supported trailing
                    // action is separate from the hold and grants no closure.
                    matches!(choice.trim_matches('`'), "hold" | "hold、その後PRを閉じる")
                }),
        );
    }
    rows == [true]
}

fn conditional_recommendation(line: &str) -> bool {
    let lower = line.to_lowercase();
    lower.split(|c: char| !c.is_ascii_alphabetic()).any(|word| {
        matches!(
            word,
            "if" | "unless"
                | "when"
                | "maybe"
                | "conditional"
                | "otherwise"
                | "only"
                | "except"
                | "or"
                | "either"
                | "provided"
                | "assuming"
                | "depending"
        )
    }) || [
        "ただし",
        "場合",
        "なら",
        "たら",
        "限り",
        "のみ",
        "かも",
        "次第",
        "条件",
        "または",
        "もしくは",
        "あるいは",
        "れば",
        "とき",
        "際",
        "時だけ",
        "必要に応じて",
    ]
    .iter()
    .any(|word| line.contains(word))
}

impl QuestionKind {
    /// Select the policy from the stored question node.
    pub fn of(q: &Question) -> Self {
        match q.node.as_str() {
            crate::land::APPROVAL_NODE => Self::Merge,
            crate::github_text::ASK_NODE => Self::Posting,
            _ => Self::Ordinary,
        }
    }
}

/// Validate guarded decisions answered by a running chat. Terminal answers
/// retain their existing behavior. Merge and posting policies both require
/// evidence from the correct conversation; posting also checks the selected text.
pub fn validate_answer(
    q: &Question,
    talks: &Talks,
    run: &str,
    node: &str,
    reply: &str,
    quote: Option<&str>,
) -> Result<()> {
    let kind = QuestionKind::of(q);
    if node != CHAT_NODE || kind == QuestionKind::Ordinary {
        return Ok(());
    }
    let posting = kind == QuestionKind::Posting;
    if posting && !q.choices.iter().any(|c| c == reply) {
        bail!("posting confirmation requires one of the offered choices");
    }
    if posting
        && !matches!(
            reply,
            crate::github_text::POST_ORIGINAL | crate::github_text::USE_REWRITE
        )
    {
        return Ok(());
    }
    let subject = if posting {
        "posting confirmation"
    } else {
        "merge approval"
    };
    let consult = q
        .consult
        .as_ref()
        .with_context(|| format!("{subject} was not handed to a chat"))?;
    if consult.talk != run {
        bail!("{subject} belongs to a different chat");
    }
    if !q.status.open() {
        bail!("{subject} is no longer open");
    }
    let talk = talks.get(run)?;
    if !talk.status.open() {
        bail!("the consulted chat is closed");
    }
    // Only words the owner wrote after this question was handed over count.
    // The generated hand-over text is stored as an operator turn (possibly
    // coalesced with owner replies), so it is cut out by its own markers.
    let at = talk
        .turns
        .iter()
        .rposition(|t| {
            t.who == talk::Who::Operator
                && t.body.contains(crate::prompt::CHAT_CONSULT_HEADING)
                && t.body.contains(&q.id)
        })
        .context("the question was not delivered to the chat yet")?;
    // A reply still waiting in `pending` is newer than every stored turn.
    let queued = latest_owner_message(
        &talk.pending,
        talk.pending_breaks.as_deref(),
        &talk.pending_messages,
        None,
        posting,
    );
    let stored = talk.turns[at..]
        .iter()
        .enumerate()
        .rev()
        .filter(|(_, t)| t.who == talk::Who::Operator)
        .find_map(|(i, t)| {
            latest_owner_message(
                &t.body,
                t.breaks.as_deref(),
                &t.messages,
                (i == 0).then_some(q.id.as_str()),
                posting,
            )
        });
    let latest = queued
        .or(stored)
        .context("no owner message after the question was handed to the chat")?;
    let quote = quote
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .with_context(|| {
            format!("chat {subject} requires --quote from the owner's latest message")
        })?;
    if posting {
        if !latest.contains(quote) {
            bail!("posting confirmation requires a verbatim quote of the latest owner message");
        }
        if !crate::github_text::posting_authorized(&latest, quote, reply) {
            bail!(
                "posting confirmation requires an explicit, unconditional instruction for the chosen text; ask the owner to clarify"
            );
        }
        return Ok(());
    }
    if reply == crate::land::APPROVE && batch_instruction(&latest) {
        bail!(
            "a batch answer instruction authorizes hold only; merging requires a separate explicit \
             confirmation of this specific reviewed PR and head through the existing approval driver"
        );
    }
    let valid = if reply == crate::land::APPROVE {
        match &q.merge_target {
            Some(target) => {
                latest.trim() == format!("merge {}", target.head) && quote == latest.trim()
            }
            None => latest.contains(quote),
        }
    } else {
        q.choices.iter().any(|c| c == reply)
            && crate::merge_decision::exact_owner_choice(reply, &latest, quote)
    };
    if !valid {
        bail!(
            "merge requires a verbatim quote of the latest owner message; hold requires the whole message to be hold"
        );
    }
    Ok(())
}

/// The owner's newest message in a turn body (or the pending draft).
///
/// Replies queued while a turn runs are joined with a blank line, which an
/// owner's own blank lines cannot be told apart from, so `talk::queue` records
/// where each message starts (`breaks`) and the newest message is taken whole.
/// With no usable record (a turn stored before it existed) the only safe cut is
/// the last paragraph: that can reject a multi-paragraph approval, and can
/// accept a `hold` that was only the last paragraph of a longer message, which
/// is harmless because silence is a hold.
#[cfg(test)]
fn latest_message(
    body: &str,
    breaks: Option<&[usize]>,
    after_block_of: Option<&str>,
) -> Option<String> {
    latest_message_for_policy(body, breaks, after_block_of, false)
}

fn latest_owner_message(
    body: &str,
    breaks: Option<&[usize]>,
    messages: &[talk::Message],
    after: Option<&str>,
    posting: bool,
) -> Option<String> {
    if messages.is_empty() {
        return latest_message_for_policy(body, breaks, after, posting);
    }
    let first = after.map_or(0, |id| {
        messages
            .iter()
            .rposition(|m| {
                m.body.contains(crate::prompt::CHAT_CONSULT_HEADING) && m.body.contains(id)
            })
            .unwrap_or(0)
    });
    // A legacy message may span several sends. Only map the recorded cuts
    // when the message bodies still match the enclosing body exactly.
    let aligned = messages
        .iter()
        .map(|m| m.body.as_str())
        .collect::<Vec<_>>()
        .join("\n\n")
        == body;
    let valid_breaks = breaks.filter(|cuts| {
        cuts.windows(2).all(|w| w[0] < w[1])
            && cuts.iter().all(|&o| {
                o >= 2 && o <= body.len() && body.is_char_boundary(o) && body[..o].ends_with("\n\n")
            })
    });
    let mut offset = 0;
    let mut latest = None;
    for (i, message) in messages.iter().enumerate() {
        let end = offset + message.body.len();
        if i >= first && message.sender != Some(talk::Sender::Magi) {
            let local_breaks = aligned.then_some(valid_breaks).flatten().map(|cuts| {
                cuts.iter()
                    .copied()
                    .filter(|&cut| cut > offset && cut <= end)
                    .map(|cut| cut - offset)
                    .collect::<Vec<_>>()
            });
            let message_breaks = if message.sender.is_some() {
                Some(&[][..])
            } else {
                local_breaks.as_deref()
            };
            if let Some(words) = latest_message_for_policy(
                &message.body,
                message_breaks,
                after.filter(|_| i == first),
                posting,
            ) {
                latest = Some(words);
            }
        }
        offset = end + 2;
    }
    latest
}

fn latest_message_for_policy(
    body: &str,
    breaks: Option<&[usize]>,
    after_block_of: Option<&str>,
    posting: bool,
) -> Option<String> {
    let cuts = breaks.filter(|b| {
        b.windows(2).all(|w| w[0] < w[1])
            && b.iter().all(|&o| {
                o >= 2 && o <= body.len() && body.is_char_boundary(o) && body[..o].ends_with("\n\n")
            })
    });
    let Some(cuts) = cuts else {
        if is_completion_notice(body) {
            return None;
        }
        let words = owner_words_for_policy(body, after_block_of, posting);
        // Legacy/invalid boundary records cannot distinguish paragraphs from
        // separate owner replies. Posting must not promote a reported final
        // paragraph into an instruction. Merge keeps its existing policy.
        if posting && words.contains("\n\n") {
            // Keep this as the newest evidence (with no usable quote), so an
            // ambiguous queued reply cannot fall back to an older instruction.
            return Some(String::new());
        }
        if posting {
            return (!words.trim().is_empty()).then_some(words);
        }
        return words
            .rsplit("\n\n")
            .map(str::trim)
            .find(|p| !p.is_empty())
            .map(str::to_owned);
    };
    let mut starts = vec![0];
    starts.extend_from_slice(cuts);
    let messages: Vec<&str> = starts
        .iter()
        .enumerate()
        .map(|(i, &s)| {
            let e = starts.get(i + 1).map_or(body.len(), |&n| n - 2);
            &body[s..e]
        })
        .collect();
    // The message carrying the hand-over: everything before it is older.
    let first = after_block_of.map_or(0, |id| {
        messages
            .iter()
            .rposition(|m| m.contains(crate::prompt::CHAT_CONSULT_HEADING) && m.contains(id))
            .unwrap_or(0)
    });
    messages
        .iter()
        .enumerate()
        .skip(first)
        .rev()
        .filter(|(_, m)| !is_completion_notice(m))
        .map(|(i, m)| owner_words_for_policy(m, after_block_of.filter(|_| i == first), posting))
        .find(|w| !w.trim().is_empty())
}

/// Is this message magi's own task or watch completion notice? It is never
/// the owner's word.
fn is_completion_notice(message: &str) -> bool {
    crate::prompt::is_completion_notice(message)
}

/// Length of a consult stored before quoted markers were defused, which may
/// quote the end phrase or a whole earlier consult in its detail. The closer is
/// the generated tail's full final sentence (an owner reply does not repeat
/// it), and a real heading met before it opens a nested block that needs its
/// own closer. A block that never closes runs to the end of the text, which can
/// only exclude more.
fn legacy_len(rest: &str) -> usize {
    use crate::prompt::CHAT_CONSULT_HEADING;
    const CLOSERS: [&str; 2] = [
        "cannot be revived. Do not edit the repository.",
        "make: do not edit the repository.",
    ];
    let mut events: Vec<(usize, usize)> = Vec::new(); // (at, 0 = open, else closer length)
    let first = rest.find(CHAT_CONSULT_HEADING).unwrap_or(0);
    for (i, _) in rest.match_indices(CHAT_CONSULT_HEADING) {
        if i > first
            && rest[..i].ends_with("# ")
            && rest[..i - 2].ends_with('\n')
            && rest[i + CHAT_CONSULT_HEADING.len()..].starts_with("\n\nThe operator passed you")
        {
            events.push((i, 0));
        }
    }
    for c in CLOSERS {
        events.extend(rest.match_indices(c).map(|(i, _)| (i, c.len())));
    }
    events.sort_unstable();
    let mut depth = 1usize;
    for (at, len) in events {
        if len == 0 {
            depth += 1;
        } else {
            depth -= 1;
            if depth == 0 {
                return at + len;
            }
        }
    }
    rest.len()
}

/// The owner's own words in an operator turn, with magi's generated hand-over
/// text cut out. `drain` stores a queued consult as an operator turn, and
/// `talk::queue` joins an owner reply sent meanwhile onto the same draft, so
/// the turn's origin cannot be told from the turn as a whole. Each generated
/// block runs from its [`CHAT_CONSULT_HEADING`](crate::prompt::CHAT_CONSULT_HEADING)
/// to the first [`CHAT_CONSULT_END`](crate::prompt::CHAT_CONSULT_END) after it
/// (quoted markers are defused by `prompt::chat_consult`; a consult stored
/// before that, recognised by its opening sentence, ends at the last one before
/// the next heading); with no end the
/// block runs to the next heading. Only the blocks are cut: owner replies
/// between them are kept, in order, even when they contain the phrase. With `after_block_of` (the turn
/// carrying the hand-over itself) everything up to the end of the last block
/// naming that question id predates the hand-over and is dropped.
#[cfg(test)]
fn owner_words(body: &str, after_block_of: Option<&str>) -> String {
    owner_words_for_policy(body, after_block_of, false)
}

fn owner_words_for_policy(body: &str, after_block_of: Option<&str>, posting: bool) -> String {
    // Remove generated blocks without removing Markdown code indentation from
    // the owner's evidence. Merge retains its existing whitespace handling.
    let trim = |text: &str| -> String {
        if posting {
            text.trim_matches(['\r', '\n']).to_owned()
        } else {
            text.trim().to_owned()
        }
    };
    use crate::prompt::{CHAT_CONSULT_END, CHAT_CONSULT_HEADING};
    // A real heading is the generated one: `# ` at a line start, followed by
    // the fixed opening sentence. A heading phrase quoted inside a question's
    // detail is not a boundary; it stays inside the block it is part of.
    let heads: Vec<usize> = body
        .match_indices(CHAT_CONSULT_HEADING)
        .map(|(i, _)| i)
        .filter(|&i| {
            let line_start = body[..i]
                .strip_suffix("# ")
                .is_some_and(|b| b.is_empty() || b.ends_with('\n'));
            line_start
                && body[i + CHAT_CONSULT_HEADING.len()..].starts_with("\n\nThe operator passed you")
        })
        .collect();
    if heads.is_empty() {
        return trim(body);
    }
    // (start, end) of each generated block, the heading's own "# " included.
    // `chat_consult` defuses every marker quoted in a question's text, so the
    // first end phrase after a heading is the block's own. A block with no end
    // before the next heading runs up to that heading (or the text's end).
    // Turns stored before the markers were defused may still quote an end
    // phrase; their tail is not recognised as part of the block.
    let mut blocks: Vec<(usize, usize)> = Vec::new();
    for &h in &heads {
        let start = if body[..h].ends_with("# ") { h - 2 } else { h };
        if blocks.last().is_some_and(|&(_, e)| start < e) {
            continue; // quoted inside the block above
        }
        let rest = &body[h..];
        let head = rest.split("\n\n## ").next().unwrap_or(rest);
        let end = if head.contains(crate::prompt::CHAT_CONSULT_DEFUSED) {
            // Current format: quoted markers are defused, so the first end
            // phrase is the block's own.
            rest.find(CHAT_CONSULT_END)
                .map_or(body.len(), |p| h + p + CHAT_CONSULT_END.len())
        } else {
            h + legacy_len(rest)
        };
        blocks.push((start, end));
    }
    let mut from = 0;
    if let Some(id) = after_block_of {
        if let Some(&(_, end)) = blocks.iter().rfind(|&&(s, e)| body[s..e].contains(id)) {
            from = end;
        } else if let Some(&(_, end)) = blocks.last() {
            from = end;
        }
    }
    let mut parts = Vec::new();
    let mut at = from;
    for &(s, e) in &blocks {
        if s >= at {
            parts.push(trim(&body[at..s]));
        }
        at = at.max(e);
    }
    parts.push(trim(&body[at..]));
    parts
        .into_iter()
        .filter(|p| !p.trim().is_empty())
        .collect::<Vec<_>>()
        .join("\n\n")
}

/// What [`start_turn`] did.
#[derive(Debug, PartialEq, Eq)]
pub enum Started {
    /// The question was handed over and the chat's turn ran to its end.
    Answered,
    /// Another process holds the talk's turn lease; nothing was changed.
    Busy,
    /// The question had already been handed to the chat; nothing was changed.
    Nothing,
}

/// [`begin`], then run the chat's turn here, under the talk's own lease.
///
/// The lease is taken *before* anything is written, so a talk somebody else is
/// running leaves no consult record and no draft behind (`Started::Busy`) and
/// the caller can simply try again later. A turn that fails after `begin`
/// leaves the record and the draft, so the chat can resume it.
pub async fn start_turn(
    questions: &Questions,
    talks: &Talks,
    q: &Question,
    talk: &Talk,
    cfg: &Config,
) -> Result<Started> {
    if q.consult.is_some() {
        return Ok(Started::Nothing);
    }
    let Some(lease) = talks.claim_turn(&talk.id)? else {
        return Ok(Started::Busy);
    };
    if !begin(questions, talks, q, talk)? {
        return Ok(Started::Nothing);
    }
    drain_turns(talks, &talk.id, lease, cfg).await?;
    Ok(Started::Answered)
}

/// Run the talk's turns under `lease` until no draft is owed.
///
/// Other starters that found the lease held queued drafts and left them to
/// us, so drain until nothing is left (as the web's drain loop does). A failed
/// turn is kept and reported at the end, not returned at once: drafts accepted
/// meanwhile are still owed an answer.
pub async fn drain_turns(
    talks: &Talks,
    talk_id: &str,
    mut lease: talk::TurnLease,
    cfg: &Config,
) -> Result<()> {
    let mut talk = talks.get(talk_id)?;
    let mut failed = None;
    loop {
        while let Some(text) = talk::drain(&mut talk, talks)? {
            if let Err(e) = talk::respond(&lease, &mut talk, talks, cfg, &text).await {
                failed.get_or_insert(e);
            }
            if !lease.beat()? {
                bail!("the turn lease for chat {} was lost", talk.short());
            }
        }
        // A draft queued after the last drain, before the lease is gone, was
        // left to us by a starter that found the lease held. Release first,
        // then look again: whoever queues later either sees no lease (and
        // starts its own turn) or is seen by this re-check.
        drop(lease);
        talk = talks.get(&talk.id)?;
        let owed = talk.status.open()
            && (!talk.pending.is_empty() || !talk.pending_attachments.is_empty());
        if !owed {
            break;
        }
        // Somebody else took the lease meanwhile: they drain it.
        let Some(again) = talks.claim_turn(&talk.id)? else {
            break;
        };
        lease = again;
    }
    if let Some(e) = failed {
        return Err(e);
    }
    Ok(())
}

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

    use super::*;
    use crate::config::{AgentKind, AgentSpec, Config};
    use crate::queue::Source;
    use crate::talk::TalkStatus;

    const RUN: &str = "20260902-000000-beef";

    fn talks() -> (tempfile::TempDir, Talks, Talk) {
        let tmp = tempfile::tempdir().expect("tempdir");
        let store = Talks::at(tmp.path().join("talks"));
        let cfg = Config {
            agents: vec![AgentSpec {
                id: "mock".to_owned(),
                kind: AgentKind::Command,
                model: None,
                command: vec!["true".to_owned()],
                extra_args: Vec::new(),
                env: BTreeMap::new(),
                prompt_delivery: None,
            }],
            ..Config::default()
        };
        let talk = talk::begin(&store, &cfg, tmp.path().to_path_buf(), Some("mock")).expect("talk");
        (tmp, store, talk)
    }

    fn task(source: Source) -> Task {
        let mut t = Task::new(
            "t".to_owned(),
            "Do it".to_owned(),
            PathBuf::from("/repo"),
            source,
        );
        t.start(RUN.to_owned());
        t
    }

    fn from_chat(talk: &Talk) -> Task {
        task(Source::Agent {
            run: talk.id.clone(),
            node: CHAT_NODE.to_owned(),
        })
    }

    fn question(node: &str) -> Question {
        Question::new(
            RUN.to_owned(),
            node.to_owned(),
            "impl-A".to_owned(),
            "Which backend?".to_owned(),
            "SQLite is simpler.".to_owned(),
            vec!["SQLite".to_owned(), "Redis".to_owned()],
        )
    }

    #[test]
    fn chat_answer_is_refused_for_an_operator_held_task() {
        let mut task = crate::queue::Task::new(
            "t".into(),
            "i".into(),
            std::path::PathBuf::from("."),
            crate::queue::Source::Human,
        );
        task.runs.push("run-1".into());
        let mut q = question(crate::land::APPROVAL_NODE);
        q.run = "run-1".into();
        let check = |t: &crate::queue::Task, node: &str| {
            refuse_operator_held(&q, node, std::slice::from_ref(t))
        };
        assert!(check(&task, CHAT_NODE).is_ok());
        task.hold_machine(None);
        assert!(check(&task, CHAT_NODE).is_ok());
        task.hold_manual(None);
        assert!(check(&task, CHAT_NODE).is_err());
        // Not a chat's answer, or another question: untouched.
        assert!(check(&task, "implement").is_ok());
        assert!(refuse_operator_held(&q, CHAT_NODE, &[]).is_ok());
    }

    #[tokio::test]
    async fn start_turn_is_busy_and_writes_nothing_while_the_lease_is_held() {
        let (tmp, store, talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question("implement");
        questions.put(&mut q).expect("put");
        let held = store.claim_turn(&talk.id).expect("claim").expect("first");

        let got = start_turn(&questions, &store, &q, &talk, &Config::default())
            .await
            .expect("start");
        assert_eq!(got, Started::Busy);
        assert!(questions.get(&q.id).unwrap().consult.is_none());
        assert!(store.get(&talk.id).unwrap().pending.is_empty());
        drop(held);
    }

    /// A `kind = "command"` agent running `script`; `LEASE` names the turn
    /// lease file so the script can look at it while the turn is in flight.
    fn scripted(tmp: &std::path::Path, script: &str, lease: &std::path::Path) -> Config {
        let path = tmp.join("mock-consult-agent.sh");
        std::fs::write(&path, script).expect("write mock");
        Config {
            agents: vec![AgentSpec {
                id: "mock".to_owned(),
                kind: AgentKind::Command,
                model: None,
                command: vec!["sh".to_owned(), path.to_string_lossy().into_owned()],
                extra_args: Vec::new(),
                env: BTreeMap::from([("LEASE".to_owned(), lease.to_string_lossy().into_owned())]),
                prompt_delivery: None,
            }],
            ..Config::default()
        }
    }

    /// Run `start_turn` with `script` as the agent and hand back its result,
    /// the store and the talk, for the caller to claim the lease afterwards.
    async fn run_turn(
        script: &str,
    ) -> (
        tempfile::TempDir,
        Result<Started>,
        Questions,
        Question,
        Talk,
    ) {
        crate::run::set_home(crate::run::test_home());
        let (tmp, store, talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question("implement");
        questions.put(&mut q).expect("put");
        let cfg = scripted(tmp.path(), script, &store.turn_path(&talk.id));
        let got = start_turn(&questions, &store, &q, &talk, &cfg).await;
        (tmp, got, questions, q, talk)
    }

    #[tokio::test]
    async fn the_lease_is_held_during_the_turn_and_free_once_it_ends() {
        // exit 7 if the lease file is missing while the agent runs.
        let (tmp, got, _questions, _q, talk) =
            run_turn("#!/bin/sh\ncat >/dev/null\n[ -f \"$LEASE\" ] || exit 7\nprintf 'ok\\n'\n")
                .await;
        assert_eq!(got.expect("turn"), Started::Answered);
        let other = Talks::at(tmp.path().join("talks"));
        assert!(
            other.claim_turn(&talk.id).expect("claim").is_some(),
            "free once the turn ended"
        );
    }

    #[tokio::test]
    async fn the_lease_is_free_again_after_a_failed_turn() {
        let (tmp, got, questions, q, talk) = run_turn("#!/bin/sh\ncat >/dev/null\nexit 3\n").await;
        assert!(got.is_err(), "the agent failed");
        assert!(
            questions.get(&q.id).unwrap().consult.is_some(),
            "the turn got as far as running"
        );
        let other = Talks::at(tmp.path().join("talks"));
        assert!(
            other.claim_turn(&talk.id).expect("claim").is_some(),
            "free after the failed turn"
        );
    }

    #[test]
    fn origin_talk_is_decided_in_one_table() {
        let (_tmp, _store, talk) = talks();
        let mut closed = talk.clone();
        closed.status = TalkStatus::Closed;
        let chat = from_chat(&talk);
        let q = question("implement");

        let hit = |tasks: &[Task], talks: &[Talk], q: &Question| origin_talk(tasks, talks, q);
        assert_eq!(
            hit(std::slice::from_ref(&chat), std::slice::from_ref(&talk), &q).map(|t| t.id),
            Some(talk.id.clone())
        );
        // Not filed from a chat.
        assert!(hit(&[task(Source::Human)], std::slice::from_ref(&talk), &q).is_none());
        let other = task(Source::Agent {
            run: talk.id.clone(),
            node: "implement".to_owned(),
        });
        assert!(hit(&[other], std::slice::from_ref(&talk), &q).is_none());
        // The conversation is gone, or no longer takes turns.
        assert!(hit(std::slice::from_ref(&chat), &[], &q).is_none());
        // A closed conversation is reopened by `begin`, so it still qualifies.
        assert!(hit(std::slice::from_ref(&chat), &[closed], &q).is_some());
        // No task owns the question.
        assert!(hit(&[], std::slice::from_ref(&talk), &q).is_none());
        // Merge approvals can be consulted; release notices cannot.
        assert!(
            hit(
                std::slice::from_ref(&chat),
                std::slice::from_ref(&talk),
                &question(crate::land::APPROVAL_NODE)
            )
            .is_some()
        );
        assert!(
            hit(
                std::slice::from_ref(&chat),
                std::slice::from_ref(&talk),
                &question(crate::bump::NOTICE_NODE)
            )
            .is_none()
        );
        // A settled question has nothing left to ask.
        let mut answered = question("implement");
        answered
            .answer(crate::ask::Answer::Choice("Redis".to_owned()))
            .unwrap();
        assert!(hit(&[chat], &[talk], &answered).is_none());
    }

    #[test]
    fn associated_chat_remains_navigable_after_consultation_and_settlement() {
        let (tmp, store, mut talk) = talks();
        let chat = from_chat(&talk);
        let tasks = std::slice::from_ref(&chat);
        let mut q = question("implement");
        let questions = Questions::at(tmp.path().join("questions"));
        questions.put(&mut q).unwrap();
        assert!(begin(&questions, &store, &q, &talk).unwrap());
        q = questions.get(&q.id).unwrap();
        let associated = |q: &Question, talk: &Talk| {
            associated_talk(tasks, std::slice::from_ref(talk), q).map(|t| t.id)
        };
        assert_eq!(associated(&q, &talk), Some(talk.id.clone()));
        assert!(q.status.open());
        assert!(q.answer.is_none());
        assert!(origin_talk(tasks, std::slice::from_ref(&talk), &q).is_some());
        q.answer(crate::ask::Answer::Choice("Redis".into()))
            .unwrap();
        assert_eq!(associated(&q, &talk), Some(talk.id.clone()));
        assert!(origin_talk(tasks, std::slice::from_ref(&talk), &q).is_none());
        talk.status = TalkStatus::Closed;
        assert_eq!(associated(&q, &talk), Some(talk.id.clone()));
        let mut q = question("implement");
        q.abandon("expired");
        assert_eq!(associated(&q, &talk), Some(talk.id.clone()));
        assert!(origin_talk(tasks, std::slice::from_ref(&talk), &q).is_none());
        assert!(associated_talk(tasks, &[], &q).is_none());
        assert!(associated_talk(&[], std::slice::from_ref(&talk), &q).is_none());
        let notice = question(crate::bump::NOTICE_NODE);
        assert_eq!(associated(&notice, &talk), Some(talk.id.clone()));
        assert!(origin_talk(tasks, &[talk], &notice).is_none());
    }

    #[test]
    fn approval_origin_requires_an_open_chat_task_and_question() {
        let (_tmp, _store, talk) = talks();
        let chat = from_chat(&talk);
        let mut q = question(crate::land::APPROVAL_NODE);
        assert!(origin_talk(&[task(Source::Human)], std::slice::from_ref(&talk), &q).is_none());
        assert!(
            origin_talk(
                &[task(Source::Agent {
                    run: talk.id.clone(),
                    node: "implement".into(),
                })],
                std::slice::from_ref(&talk),
                &q
            )
            .is_none()
        );
        assert!(origin_talk(std::slice::from_ref(&chat), &[], &q).is_none());
        let mut closed = talk.clone();
        closed.status = TalkStatus::Closed;
        assert!(origin_talk(std::slice::from_ref(&chat), &[closed], &q).is_some());
        q.abandon("expired");
        assert!(
            origin_talk(std::slice::from_ref(&chat), std::slice::from_ref(&talk), &q).is_none()
        );
        let mut q = question(crate::land::APPROVAL_NODE);
        q.answer(crate::ask::Answer::Choice("Redis".into()))
            .unwrap();
        assert!(origin_talk(&[chat], &[talk], &q).is_none());
    }

    #[test]
    fn a_conductor_question_is_found_through_the_task_id() {
        let (_tmp, _store, talk) = talks();
        let chat = from_chat(&talk);
        let mut q = question(crate::conduct::NODE);
        q.run = chat.id.clone();
        assert!(origin_talk(&[chat], &[talk], &q).is_some());
    }

    fn followup_of(parent: &Task, n: u32) -> Task {
        let mut t = Task::new(
            "f".to_owned(),
            "Fix".to_owned(),
            PathBuf::from("/repo"),
            Source::Agent {
                run: format!("merged-{n}"),
                node: "followup".to_owned(),
            },
        );
        t.start(format!("run-f{n}"));
        t.followup = Some(crate::queue::FollowUp {
            run: format!("merged-{n}"),
            origin_task: Some(parent.id.clone()),
            pr: "https://example.invalid/pr/1".to_owned(),
            findings: Vec::new(),
            generation: n,
        });
        t
    }

    fn q_for(t: &Task) -> Question {
        let mut q = question("implement");
        q.run = t.runs[0].clone();
        q
    }

    #[test]
    fn a_followup_traces_back_to_the_chat() {
        let (_tmp, _store, talk) = talks();
        let mut chat = from_chat(&talk);
        chat.runs = vec!["chat-run".to_owned()];
        let f1 = followup_of(&chat, 1);
        let f2 = followup_of(&f1, 2);
        let ts = [chat, f1.clone(), f2.clone()];
        let id = Some(talk.id.clone());
        let tk = std::slice::from_ref(&talk);
        assert_eq!(origin_talk(&ts, tk, &q_for(&f1)).map(|t| t.id), id);
        assert_eq!(origin_talk(&ts, tk, &q_for(&f2)).map(|t| t.id), id);
    }

    #[test]
    fn a_followup_without_origin_task_is_found_through_runs() {
        let (_tmp, _store, talk) = talks();
        let mut chat = from_chat(&talk);
        chat.runs = vec!["merged-1".to_owned()];
        let mut f1 = followup_of(&chat, 1);
        f1.followup.as_mut().unwrap().origin_task = None;
        let q = q_for(&f1);
        assert!(origin_talk(&[chat, f1], &[talk], &q).is_some());
    }

    #[test]
    fn a_dangling_origin_task_has_no_chat() {
        let (_tmp, _store, talk) = talks();
        let mut chat = from_chat(&talk);
        chat.runs = vec!["chat-run".to_owned()];
        let mut f1 = followup_of(&chat, 1);
        f1.followup.as_mut().unwrap().origin_task = Some("gone".to_owned());
        let q = q_for(&f1);
        assert!(origin_talk(&[chat, f1], &[talk], &q).is_none());
    }

    #[test]
    fn a_recorded_chat_survives_deleted_ancestors() {
        let (_tmp, _store, talk) = talks();
        let chat = from_chat(&talk);
        assert_eq!(chat.origin_chat.as_deref(), Some(talk.id.as_str()));
        let mut f1 = followup_of(&chat, 1);
        f1.origin_chat = chat.origin_chat.clone();
        let mut f2 = followup_of(&f1, 2);
        f2.origin_chat = f1.origin_chat.clone();
        // Neither the chat task nor f1 is in the queue any more.
        let q = q_for(&f2);
        let got = origin_talk(std::slice::from_ref(&f2), std::slice::from_ref(&talk), &q);
        assert_eq!(got.map(|t| t.id), Some(talk.id.clone()));
        // The recorded id still has to name an open chat.
        assert!(origin_talk(&[f2], &[], &q).is_none());
    }

    #[test]
    fn a_task_without_the_field_still_walks_the_ancestry() {
        let (_tmp, _store, talk) = talks();
        let mut chat = from_chat(&talk);
        chat.origin_chat = None;
        let mut f1 = followup_of(&chat, 1);
        f1.origin_chat = None;
        let q = q_for(&f1);
        assert!(origin_talk(&[chat, f1.clone()], std::slice::from_ref(&talk), &q).is_some());
        // And the old JSON shape reads with the field absent.
        let mut v = serde_json::to_value(&f1).unwrap();
        v.as_object_mut().unwrap().remove("origin_chat");
        let back: Task = serde_json::from_value(v).unwrap();
        assert!(back.origin_chat.is_none());
    }

    #[test]
    fn a_followup_cycle_ends_without_a_chat() {
        let (_tmp, _store, talk) = talks();
        let mut a = followup_of(&from_chat(&talk), 1);
        let mut b = followup_of(&a, 2);
        a.followup.as_mut().unwrap().origin_task = Some(b.id.clone());
        b.followup.as_mut().unwrap().origin_task = Some(a.id.clone());
        let q = q_for(&a);
        assert!(origin_talk(&[a, b], &[talk], &q).is_none());
    }

    #[test]
    fn pending_consults_follow_the_store_not_the_turn_text() {
        let (tmp, store, talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        assert!(!pending_consults(&questions, &talk.id), "empty store");

        let mut plain = question("implement");
        questions.put(&mut plain).unwrap();
        assert!(!pending_consults(&questions, &talk.id), "no consult record");

        let mut q = question("implement");
        questions.put(&mut q).unwrap();
        begin(&questions, &store, &q, &talk).unwrap();
        assert!(pending_consults(&questions, &talk.id));
        assert!(!pending_consults(&questions, "other-talk"));

        questions
            .update(&q.id, |q| {
                q.abandon("test");
                Ok(())
            })
            .unwrap();
        assert!(!pending_consults(&questions, &talk.id), "closed question");
    }

    #[test]
    fn begin_reopens_a_closed_chat_before_queueing() {
        let (tmp, store, mut talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question("implement");
        questions.put(&mut q).unwrap();
        talk::close(&mut talk, &store).unwrap();
        assert!(!store.get(&talk.id).unwrap().status.open());

        assert!(begin(&questions, &store, &q, &talk).unwrap());

        let after = store.get(&talk.id).unwrap();
        assert!(after.status.open(), "the chat was reopened");
        assert!(after.pending.contains(&q.id), "{}", after.pending);
        assert!(questions.get(&q.id).unwrap().consult.is_some());
    }

    #[test]
    fn begin_queues_once_and_leaves_the_question_open() {
        let (tmp, store, talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question("implement");
        questions.put(&mut q).unwrap();

        assert!(begin(&questions, &store, &q, &talk).unwrap());
        assert!(!begin(&questions, &store, &q, &talk).unwrap(), "idempotent");

        let after = questions.get(&q.id).unwrap();
        assert!(after.status.open());
        assert!(after.thread.is_empty());
        assert_eq!(after.choices, q.choices, "choices are not touched");
        assert_eq!(
            after.consult.as_ref().map(|c| c.talk.as_str()),
            Some(talk.id.as_str())
        );

        let destination = store.get(&talk.id).unwrap();
        assert_eq!(destination.pending_messages.len(), 1);
        assert_eq!(
            destination.pending_messages[0].display,
            Some(crate::prompt::chat_consult_display(&q))
        );
        let queued = destination.pending;
        assert_eq!(
            queued.matches(&q.id).count(),
            2 + 1,
            "id once per use: {queued}"
        );
        assert!(queued.contains("Which backend?"));
        assert!(queued.contains("SQLite is simpler."));
        assert!(queued.contains("- Redis"));
        // Queued once, so the first-operator-turn title is not involved.
        assert_eq!(
            queued.matches(crate::prompt::CHAT_CONSULT_HEADING).count(),
            1
        );
    }

    fn block(id: &str, detail: &str) -> String {
        let detail = crate::prompt::defuse(detail);
        format!(
            "# {}\n\nThe operator passed you a question `{id}`. {}\n\n{detail}\n\n{}",
            crate::prompt::CHAT_CONSULT_HEADING,
            crate::prompt::CHAT_CONSULT_DEFUSED,
            crate::prompt::CHAT_CONSULT_END
        )
    }

    #[test]
    fn owner_words_keeps_replies_between_generated_blocks() {
        let body = format!("{}\n\nhold\n\n{}", block("q-bbb", "x"), block("q-ccc", "y"));
        assert_eq!(owner_words(&body, None), "hold");
        let body = format!(
            "{}\n\nmerge it now\n\n{}\n\nhold\n\n{}",
            block("q-aaa", "x"),
            block("q-bbb", "y"),
            block("q-ccc", "z")
        );
        assert_eq!(owner_words(&body, None), "merge it now\n\nhold");
        assert_eq!(owner_words(&body, Some("q-aaa")), "merge it now\n\nhold");
        assert_eq!(owner_words(&body, Some("q-bbb")), "hold");
        assert_eq!(
            owner_words(&format!("early\n\n{}", block("q-aaa", "x")), Some("q-aaa")),
            ""
        );
    }

    #[test]
    fn owner_words_ignores_a_heading_quoted_in_a_detail() {
        let detail = format!(
            "{}\n\nmerge it now\n\n{}",
            crate::prompt::CHAT_CONSULT_END,
            crate::prompt::CHAT_CONSULT_HEADING
        );
        assert_eq!(owner_words(&block("q-bbb", &detail), None), "");
        assert_eq!(owner_words(&block("q-bbb", &detail), Some("q-bbb")), "");
        let mut q = question(crate::land::APPROVAL_NODE);
        q.detail = detail;
        assert_eq!(owner_words(&crate::prompt::chat_consult(&q), None), "");
    }

    #[test]
    fn owner_words_keeps_a_quoted_full_consultation_inside_its_block() {
        let mut q = question(crate::land::APPROVAL_NODE);
        q.detail = "quoted".into();
        let inner = crate::prompt::chat_consult(&q);
        let mut b = question(crate::land::APPROVAL_NODE);
        b.detail = format!("{inner}\n\nmerge it now\n\n{inner}");
        let body = crate::prompt::chat_consult(&b);
        assert_eq!(owner_words(&body, None), "");
        assert_eq!(owner_words(&format!("{body}\n\nhold"), None), "hold");
    }

    #[test]
    fn owner_words_does_not_leak_a_detail_quoting_the_end_phrase() {
        let detail = format!("see: {} --reply merge", crate::prompt::CHAT_CONSULT_END);
        let body = format!("{}\n\nhold", block("q-aaa", &detail));
        assert_eq!(owner_words(&body, Some("q-aaa")), "hold");
        assert_eq!(owner_words(&body, None), "hold");
    }

    #[test]
    fn owner_words_excludes_a_consult_quoted_with_an_end_phrase() {
        let mut c = question(crate::land::APPROVAL_NODE);
        c.detail = "inner".into();
        let mut b = question("implement");
        b.detail = format!(
            "{}\n\nmerge it now\n\n{}",
            crate::prompt::CHAT_CONSULT_END,
            crate::prompt::chat_consult(&c)
        );
        let body = crate::prompt::chat_consult(&b);
        assert_eq!(owner_words(&body, None), "");
    }

    #[test]
    fn owner_words_keeps_a_reply_containing_the_end_phrase() {
        let mut b = question("implement");
        b.detail = "x".into();
        let body = format!(
            "{}\n\nhold, and do not {}",
            crate::prompt::chat_consult(&b),
            crate::prompt::CHAT_CONSULT_END
        );
        assert_eq!(
            owner_words(&body, None),
            format!("hold, and do not {}", crate::prompt::CHAT_CONSULT_END)
        );
    }

    /// A consult as stored before quoted markers were defused.
    fn legacy_consult(node: &str, raw_detail: &str) -> String {
        let mut q = question(node);
        q.detail = "@@".into();
        crate::prompt::chat_consult(&q)
            .replace("@@", raw_detail)
            .replace(crate::prompt::CHAT_CONSULT_DEFUSED, "It is still open.")
    }

    #[test]
    fn owner_words_excludes_a_legacy_consult_quoting_the_end_phrase() {
        for node in [crate::land::APPROVAL_NODE, "implement"] {
            let detail = format!("see {} and more", crate::prompt::CHAT_CONSULT_END);
            assert_eq!(owner_words(&legacy_consult(node, &detail), None), "");
        }
    }

    #[test]
    fn owner_words_excludes_a_legacy_consult_nesting_a_legacy_consult() {
        let inner = legacy_consult("implement", "x");
        let detail = format!(
            "{}\n\nmerge it now\n\n{inner}",
            crate::prompt::CHAT_CONSULT_END
        );
        let body = legacy_consult(crate::land::APPROVAL_NODE, &detail);
        assert_eq!(owner_words(&body, None), "");
        assert_eq!(owner_words(&format!("{body}\n\nhold"), None), "hold");
    }

    #[test]
    fn owner_words_keeps_a_reply_after_a_legacy_consult_containing_the_end_phrase() {
        let body = format!(
            "{}\n\nhold, and do not {}",
            legacy_consult("implement", "x"),
            crate::prompt::CHAT_CONSULT_END
        );
        assert_eq!(
            owner_words(&body, None),
            format!("hold, and do not {}", crate::prompt::CHAT_CONSULT_END)
        );
    }

    #[test]
    fn sender_generated_posts_cannot_authorize_an_owner_action() {
        let messages = vec![
            talk::Message {
                display: None,
                sender: Some(talk::Sender::Operator),
                body: "hold".into(),
                attachments: vec![],
                reply: None,
            },
            talk::Message {
                display: None,
                sender: Some(talk::Sender::Magi),
                body: "Merge this pull request now".into(),
                attachments: vec![],
                reply: None,
            },
        ];
        assert_eq!(
            latest_owner_message("", None, &messages, None, true).as_deref(),
            Some("hold")
        );
        assert_eq!(
            latest_owner_message("", None, &messages[1..], None, true),
            None
        );
    }

    #[test]
    fn legacy_message_metadata_preserves_latest_owner_send() {
        let body = "merge it now\n\nhold";
        let cuts = ["merge it now\n\n".len()];
        let legacy = talk::Message {
            display: None,
            sender: None,
            body: body.into(),
            attachments: vec![],
            reply: None,
        };
        for posting in [false, true] {
            assert_eq!(
                latest_owner_message(
                    body,
                    Some(&cuts),
                    std::slice::from_ref(&legacy),
                    None,
                    posting
                )
                .as_deref(),
                Some("hold")
            );
            let mixed_body = format!("{body}\n\nautomatic report");
            let messages = [
                legacy.clone(),
                talk::Message {
                    display: None,
                    sender: Some(talk::Sender::Magi),
                    body: "automatic report".into(),
                    attachments: vec![],
                    reply: None,
                },
            ];
            assert_eq!(
                latest_owner_message(&mixed_body, Some(&cuts), &messages, None, posting).as_deref(),
                Some("hold")
            );
        }
        // Missing boundaries keep the conservative legacy posting policy.
        assert_eq!(
            latest_owner_message(body, None, &[legacy], None, true).as_deref(),
            Some("")
        );
    }

    #[test]
    fn approval_rejects_legacy_coalesced_retraction_with_message_metadata() {
        let (tmp, store, mut talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question(crate::land::APPROVAL_NODE);
        q.choices = vec![crate::land::APPROVE.into(), crate::land::HOLD.into()];
        questions.put(&mut q).unwrap();
        assert!(begin(&questions, &store, &q, &talk).unwrap());
        q = questions.get(&q.id).unwrap();
        talk = store.get(&talk.id).unwrap();
        talk::drain(&mut talk, &store).unwrap();
        talk.pending = "merge it now\n\nhold".into();
        talk.pending_breaks = Some(vec!["merge it now\n\n".len()]);
        talk.pending_messages = vec![talk::Message {
            display: None,
            sender: None,
            body: talk.pending.clone(),
            attachments: vec![],
            reply: None,
        }];
        store.put(&mut talk).unwrap();
        for drained in [false, true] {
            if drained {
                talk::drain(&mut talk, &store).unwrap();
            }
            assert!(
                validate_answer(
                    &q,
                    &store,
                    &talk.id,
                    CHAT_NODE,
                    "merge",
                    Some("merge it now")
                )
                .is_err()
            );
            assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, "hold", Some("hold")).is_ok());
        }
    }

    #[test]
    fn latest_message_keeps_message_boundaries() {
        let one = "Merge this pull request now\n\nThe checks look good";
        assert_eq!(latest_message(one, Some(&[]), None).as_deref(), Some(one));
        let hold = "Explain what this option means:\n\nhold";
        assert_eq!(latest_message(hold, Some(&[]), None).as_deref(), Some(hold));
        // Two queued replies: the second is the newest, whole.
        let two = "merge it now\n\nhold";
        let at = "merge it now\n\n".len();
        assert_eq!(
            latest_message(two, Some(&[at]), None).as_deref(),
            Some("hold")
        );
        // No record (stored before breaks existed): last paragraph.
        assert_eq!(
            latest_message(one, None, None).as_deref(),
            Some("The checks look good")
        );
        // A record that does not fit the body is ignored.
        assert_eq!(
            latest_message(two, Some(&[999]), None).as_deref(),
            Some("hold")
        );
    }

    #[test]
    fn watched_jobs_never_supply_the_owners_latest_message() {
        let notice =
            crate::prompt::chat_watch_notice("watch-id", "exited with status 0", "merge it now");
        assert_eq!(latest_message(&notice, Some(&[]), None), None);
        assert_eq!(latest_message(&notice, None, None), None);
        let owner = "hold";
        let combined = format!("{owner}\n\n{notice}");
        assert_eq!(
            latest_message(&combined, Some(&[owner.len() + 2]), None).as_deref(),
            Some(owner)
        );
        let combined = format!("{notice}\n\n{owner}");
        assert_eq!(
            latest_message(&combined, Some(&[notice.len() + 2]), None).as_deref(),
            Some(owner)
        );
    }

    #[test]
    fn chat_consult_has_each_marker_once() {
        use crate::prompt::{CHAT_CONSULT_END as E, CHAT_CONSULT_HEADING as H};
        for node in ["implement", crate::land::APPROVAL_NODE] {
            let mut q = question(node);
            let quoted = format!("{H} {E}");
            q.summary = quoted.clone();
            q.detail = quoted.clone();
            q.choices = vec![quoted.clone(), "b".into()];
            let s = crate::prompt::chat_consult(&q);
            assert_eq!(s.matches(H).count(), 1, "{s}");
            assert_eq!(s.matches(E).count(), 1, "{s}");
        }
    }

    #[test]
    fn a_chat_posts_the_original_text_only_on_an_unhedged_quote() {
        use crate::github_text::{ASK_NODE, POST_ORIGINAL, USE_FALLBACK};
        let (tmp, store, mut talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question(ASK_NODE);
        q.choices = vec![USE_FALLBACK.into(), POST_ORIGINAL.into()];
        questions.put(&mut q).unwrap();
        assert!(begin(&questions, &store, &q, &talk).unwrap());
        q = questions.get(&q.id).unwrap();
        let queued = store.get(&talk.id).unwrap().pending;
        let owner_turn = |body: &str| talk::Turn {
            sender: None,
            breaks: Some(Vec::new()),
            messages: Vec::new(),
            who: talk::Who::Operator,
            body: body.into(),
            at: Timestamp::now(),
            attachments: Vec::new(),
            usage: None,
            posted_by: None,
        };
        talk.turns.push(owner_turn(&queued));
        talk.pending.clear();
        let check = |q: &Question, run: &str, reply: &str, quote: Option<&str>| {
            validate_answer(q, &store, run, CHAT_NODE, reply, quote)
        };
        store.put(&mut talk).unwrap();
        // No owner word after the hand-over: nothing to quote.
        assert!(check(&q, &talk.id, POST_ORIGINAL, Some("post it")).is_err());
        talk.turns.push(owner_turn("Post the original, maybe?"));
        store.put(&mut talk).unwrap();
        assert!(check(&q, &talk.id, POST_ORIGINAL, Some("Post the original")).is_err());
        talk.turns
            .push(owner_turn("Post the original text as it is."));
        store.put(&mut talk).unwrap();
        assert!(check(&q, &talk.id, POST_ORIGINAL, None).is_err());
        assert!(check(&q, &talk.id, POST_ORIGINAL, Some("never said")).is_err());
        assert!(check(&q, &talk.id, POST_ORIGINAL, Some("Post the original text")).is_ok());
        // The safe default stays unguarded, as before.
        assert!(check(&q, &talk.id, USE_FALLBACK, None).is_ok());
    }

    #[test]
    fn a_chat_posts_the_rewrite_only_on_an_unhedged_quote() {
        use crate::github_text::{ASK_NODE, USE_FALLBACK, USE_REWRITE};
        let (tmp, store, mut talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question(ASK_NODE);
        q.choices = vec![USE_FALLBACK.into(), USE_REWRITE.into()];
        questions.put(&mut q).unwrap();
        assert!(begin(&questions, &store, &q, &talk).unwrap());
        q = questions.get(&q.id).unwrap();
        let queued = store.get(&talk.id).unwrap().pending;
        let owner_turn = |body: &str| talk::Turn {
            sender: None,
            breaks: Some(Vec::new()),
            messages: Vec::new(),
            who: talk::Who::Operator,
            body: body.into(),
            at: Timestamp::now(),
            attachments: Vec::new(),
            usage: None,
            posted_by: None,
        };
        talk.turns.push(owner_turn(&queued));
        talk.pending.clear();
        let check = |q: &Question, run: &str, reply: &str, quote: Option<&str>| {
            validate_answer(q, &store, run, CHAT_NODE, reply, quote)
        };
        store.put(&mut talk).unwrap();
        // No owner word after the hand-over: nothing to quote.
        assert!(check(&q, &talk.id, USE_REWRITE, Some("post it")).is_err());
        talk.turns.push(owner_turn("Use the rewrite, maybe?"));
        store.put(&mut talk).unwrap();
        assert!(check(&q, &talk.id, USE_REWRITE, Some("Use the rewrite")).is_err());
        talk.turns
            .push(owner_turn("Use the rewrite text as it is."));
        store.put(&mut talk).unwrap();
        assert!(check(&q, &talk.id, USE_REWRITE, None).is_err());
        assert!(check(&q, &talk.id, USE_REWRITE, Some("never said")).is_err());
        assert!(check(&q, &talk.id, USE_REWRITE, Some("Use the rewrite text")).is_ok());
        // The safe default stays unguarded, as before.
        assert!(check(&q, &talk.id, USE_FALLBACK, None).is_ok());
    }

    #[test]
    fn merge_repair_consult_requires_offered_choice_and_exact_latest_owner_request() {
        use crate::land::{APPROVAL_NODE, APPROVE, HOLD, REPAIR};
        let (tmp, store, mut talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question(APPROVAL_NODE);
        q.choices = vec![APPROVE.into(), HOLD.into(), REPAIR.into()];
        questions.put(&mut q).unwrap();
        begin(&questions, &store, &q, &talk).unwrap();
        q = questions.get(&q.id).unwrap();
        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, REPAIR, Some(REPAIR)).is_err());
        talk = store.get(&talk.id).unwrap();
        talk.turns.push(talk::Turn {
            sender: None,
            breaks: Some(Vec::new()),
            messages: Vec::new(),
            posted_by: None,
            who: talk::Who::Operator,
            body: talk.pending.clone(),
            at: Timestamp::now(),
            attachments: Vec::new(),
            usage: None,
        });
        talk.pending_messages.clear(); // This fixture now replaces the generated draft.
        talk.pending = "maybe fix findings and re-review".into();
        store.put(&mut talk).unwrap();
        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, REPAIR, Some(REPAIR)).is_err());
        talk.pending = REPAIR.into();
        talk.pending_breaks = Some(Vec::new());
        store.put(&mut talk).unwrap();
        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, REPAIR, Some(REPAIR)).is_ok());
        q.choices.retain(|c| c != REPAIR);
        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, REPAIR, Some(REPAIR)).is_err());
    }

    #[test]
    fn posting_consult_saves_only_authorized_offered_choices() {
        use crate::github_text::{ASK_NODE, POST_ORIGINAL, USE_FALLBACK, USE_REWRITE};
        for choice in [POST_ORIGINAL, USE_REWRITE] {
            let (tmp, store, mut talk) = talks();
            let questions = Questions::at(tmp.path().join("questions"));
            let mut q = question(ASK_NODE);
            q.choices = vec![USE_FALLBACK.into(), choice.into()];
            questions.put(&mut q).unwrap();
            begin(&questions, &store, &q, &talk).unwrap();
            q = questions.get(&q.id).unwrap();
            let queued = store.get(&talk.id).unwrap().pending;
            assert!(queued.contains("public posting confirmation"));
            assert!(queued.contains("not a merge approval"));
            assert!(
                queued.contains("`use fallback` choice can be saved once the owner has chosen it")
            );
            assert!(queued.contains("--quote <verbatim owner words>"));
            assert!(!queued.contains("answer it yourself with"));
            talk.turns.push(talk::Turn {
                sender: None,
                breaks: Some(Vec::new()),
                messages: Vec::new(),
                who: talk::Who::Operator,
                body: queued,
                at: Timestamp::now(),
                attachments: Vec::new(),
                usage: None,
                posted_by: None,
            });
            talk.pending.clear();
            store.put(&mut talk).unwrap();
            let save = |reply: &str, quote: Option<&str>, run: &str| {
                questions.update(&q.id, |r| {
                    validate_answer(r, &store, run, CHAT_NODE, reply, quote)?;
                    r.answer(crate::ask::Answer::Choice(reply.into()))
                })
            };
            // No owner reply, nor text from the generated question, is evidence.
            assert!(save(choice, Some(choice), &talk.id).is_err());
            let clear = if choice == POST_ORIGINAL {
                "元の説明のまま投稿して"
            } else {
                "Use the rewrite text as it is."
            };
            talk.pending = clear.into();
            store.put(&mut talk).unwrap();
            for quote in [None, Some("forged quote")] {
                let error = save(choice, quote, &talk.id).unwrap_err().to_string();
                assert!(error.contains("posting confirmation"), "{error}");
                assert!(
                    !error.contains("merge") && !error.contains("hold"),
                    "{error}"
                );
            }
            assert!(save(choice, Some(clear), "another-chat").is_err());
            let other = if choice == POST_ORIGINAL {
                USE_REWRITE
            } else {
                POST_ORIGINAL
            };
            assert!(save(other, Some(clear), &talk.id).is_err());
            let mut unconsulted = q.clone();
            unconsulted.consult = None;
            assert!(
                validate_answer(
                    &unconsulted,
                    &store,
                    &talk.id,
                    CHAT_NODE,
                    choice,
                    Some(clear)
                )
                .is_err()
            );
            let mut expired = q.clone();
            expired.abandon("expired");
            assert!(
                validate_answer(&expired, &store, &talk.id, CHAT_NODE, choice, Some(clear))
                    .is_err()
            );
            talk.status = TalkStatus::Closed;
            store.put(&mut talk).unwrap();
            assert!(save(choice, Some(clear), &talk.id).is_err());
            talk.status = TalkStatus::Open;
            for rejected in [
                "Post original was the option I saw.",
                "Use rewrite was the option I saw.",
                "Post the original, maybe?",
                "Use the rewrite only if checks pass.",
                "Post original. Cancel that.",
                "Do not post original.",
                "Use the rewrite when checks pass.",
                "元の説明で投稿してください。ただし検査が通ったら。",
                "元の説明で投稿しないで。",
                "まだ元の説明で投稿してください。",
                "ログには「元の説明で投稿しました」と記録されています。",
                "ログには「修正版で投稿しました」と記録されています。",
            ] {
                talk.pending = rejected.into();
                store.put(&mut talk).unwrap();
                let error = save(choice, Some(rejected), &talk.id)
                    .unwrap_err()
                    .to_string();
                assert!(
                    !error.contains("merge") && !error.contains("hold"),
                    "{error}"
                );
                assert!(save(choice, Some(clear), &talk.id).is_err(), "stale quote");
                let after = questions.get(&q.id).unwrap();
                assert!(after.status.open());
                assert!(after.answer.is_none());
            }
            let command = if choice == POST_ORIGINAL {
                "Post original"
            } else {
                "Use rewrite"
            };
            // A newer report must also block fallback to an older instruction.
            let mut earlier = talk.turns[0].clone();
            earlier.body = clear.into();
            talk.turns.push(earlier);
            for reported in [
                format!("The log contains:\n```\n{command}\n```"),
                format!("The log contains:\n\"\n{command}\n\""),
                format!("The log says: “Reported. {command}.”"),
                format!("The log says: “\n{command}\n”"),
            ]
            .into_iter()
            .chain(crate::github_text::reported_posting_messages(command))
            {
                talk.pending.clear();
                talk.pending_messages.clear();
                store.put(&mut talk).unwrap();
                talk::queue(&mut talk, &store, &reported, Vec::new()).unwrap();
                assert_eq!(talk.pending, reported);
                // Check both current message boundaries and legacy records.
                for breaks in [Some(Vec::new()), None] {
                    talk.pending_breaks = breaks;
                    store.put(&mut talk).unwrap();
                    for quote in [command, reported.as_str()] {
                        assert!(save(choice, Some(quote), &talk.id).is_err());
                        let after = questions.get(&q.id).unwrap();
                        assert!(after.status.open());
                        assert!(after.answer.is_none());
                    }
                }
                // Drained transcript evidence must retain the same context.
                talk::drain(&mut talk, &store).unwrap();
                for quote in [command, reported.as_str()] {
                    assert!(save(choice, Some(quote), &talk.id).is_err());
                    let after = questions.get(&q.id).unwrap();
                    assert!(after.status.open());
                    assert!(after.answer.is_none());
                }
            }
            let natural = if choice == POST_ORIGINAL {
                [
                    "OK, post the original",
                    "Yes, post original",
                    "I'd like to post the original",
                    "元の説明で投稿してください。",
                ]
            } else {
                [
                    "OK, use the rewrite",
                    "Yes, use rewrite",
                    "I'd like to use the rewrite",
                    "修正版で投稿してください。",
                ]
            };
            for words in natural {
                talk.pending = words.into();
                store.put(&mut talk).unwrap();
                validate_answer(&q, &store, &talk.id, CHAT_NODE, choice, Some(words)).unwrap();
            }
            talk.pending = clear.into();
            store.put(&mut talk).unwrap();
            save(choice, Some(clear), &talk.id).unwrap();
            assert_eq!(
                questions.get(&q.id).unwrap().resolution().as_deref(),
                Some(choice)
            );
        }
    }

    #[test]
    fn approval_consult_waits_for_latest_owner_confirmation() {
        let (tmp, store, mut talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question(crate::land::APPROVAL_NODE);
        q.choices = vec![crate::land::APPROVE.into(), crate::land::HOLD.into()];
        questions.put(&mut q).unwrap();
        assert!(begin(&questions, &store, &q, &talk).unwrap());
        assert!(!begin(&questions, &store, &q, &talk).unwrap());
        q = questions.get(&q.id).unwrap();
        assert!(q.status.open());
        assert!(q.answer.is_none());
        assert!(q.thread.is_empty());
        let queued = store.get(&talk.id).unwrap().pending;
        assert!(queued.contains("Never answer it yourself"));
        assert!(queued.contains("Silence holds"));
        assert!(queued.contains("--reply merge --quote"));
        assert!(!queued.contains("answer it yourself with"));

        let owner_turn = |body: &str| talk::Turn {
            sender: None,
            breaks: Some(Vec::new()),
            messages: Vec::new(),
            posted_by: None,
            who: talk::Who::Operator,
            body: body.into(),
            at: Timestamp::now(),
            attachments: Vec::new(),
            usage: None,
        };
        // An approval the owner gave before the hand-over is not reusable.
        let mut old = talk.clone();
        old.turns
            .insert(0, owner_turn("Please merge PR 12 after review"));
        store.put(&mut old).unwrap();
        talk.turns = old.turns.clone();
        // The drained consult draft is stored as an operator turn; it must not
        // count as the owner's words even though it contains "merge".
        talk.turns.push(owner_turn(&queued));
        store.put(&mut talk).unwrap();
        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, "merge", Some("merge")).is_err());
        assert!(
            validate_answer(
                &q,
                &store,
                &talk.id,
                CHAT_NODE,
                "merge",
                Some("Please merge PR 12")
            )
            .is_err()
        );
        // An owner reply coalesced onto the same draft still counts.
        let n = talk.turns.len();
        talk.turns[n - 1].body = format!("{queued}\n\nmerge it now");
        talk.turns[n - 1].breaks = Some(vec![queued.len() + 2]);
        store.put(&mut talk).unwrap();
        assert!(
            validate_answer(
                &q,
                &store,
                &talk.id,
                CHAT_NODE,
                "merge",
                Some("merge it now")
            )
            .is_ok()
        );
        // A later retraction in the same coalesced turn wins.
        talk.turns[n - 1].body = format!("{queued}\n\nmerge it now\n\nhold");
        talk.turns[n - 1].breaks = Some(vec![queued.len() + 2, queued.len() + 2 + 14]);
        store.put(&mut talk).unwrap();
        assert!(
            validate_answer(
                &q,
                &store,
                &talk.id,
                CHAT_NODE,
                "merge",
                Some("merge it now")
            )
            .is_err()
        );
        assert!(validate_answer(&q, &store, &talk.id, CHAT_NODE, "hold", Some("hold")).is_ok());
        // So does a reply still waiting in the draft.
        talk.turns[n - 1].body = format!("{queued}\n\nmerge it now");
        talk.turns[n - 1].breaks = Some(vec![queued.len() + 2]);
        talk.pending = "hold".into();
        store.put(&mut talk).unwrap();
        assert!(
            validate_answer(
                &q,
                &store,
                &talk.id,
                CHAT_NODE,
                "merge",
                Some("merge it now")
            )
            .is_err()
        );
        // Another question's hand-over queued after the retraction hides nothing.
        talk.pending = format!("hold\n\n{queued}");
        store.put(&mut talk).unwrap();
        assert!(
            validate_answer(
                &q,
                &store,
                &talk.id,
                CHAT_NODE,
                "merge",
                Some("merge it now")
            )
            .is_err()
        );
        talk.pending.clear();
        talk.turns[n - 1].body = queued.clone();
        store.put(&mut talk).unwrap();
        talk.turns
            .push(owner_turn("Merge this pull request please"));
        store.put(&mut talk).unwrap();
        let check = |q: &Question, run: &str, reply: &str, quote: Option<&str>| {
            validate_answer(q, &store, run, CHAT_NODE, reply, quote)
        };
        assert!(check(&q, "other-talk", "merge", Some("Merge this")).is_err());
        assert!(check(&q, &talk.id, "merge", None).is_err());
        assert!(check(&q, &talk.id, "merge", Some("never said")).is_err());
        assert!(
            check(
                &q,
                &talk.id,
                "merge",
                Some("Merge this pull request please")
            )
            .is_ok()
        );
        assert!(check(&q, &talk.id, "hold", Some("hold")).is_err());
        talk.turns
            .push(owner_turn("Wait, explain the checks first"));
        store.put(&mut talk).unwrap();
        assert!(
            check(
                &q,
                &talk.id,
                "merge",
                Some("Merge this pull request please")
            )
            .is_err()
        );
        talk.turns.push(owner_turn("Please hold"));
        store.put(&mut talk).unwrap();
        assert!(check(&q, &talk.id, "hold", Some("hold")).is_err());
        talk.turns.push(owner_turn("hold"));
        store.put(&mut talk).unwrap();
        assert!(check(&q, &talk.id, "hold", Some("hold")).is_ok());
        talk.turns
            .push(owner_turn("Merge this pull request please"));
        store.put(&mut talk).unwrap();
        q.abandon("approval expired or head changed");
        assert!(
            check(
                &q,
                &talk.id,
                "merge",
                Some("Merge this pull request please")
            )
            .is_err()
        );
        assert!(
            q.answer(crate::ask::Answer::Choice("merge".into()))
                .is_err()
        );
    }

    #[test]
    fn begin_withdraws_its_record_when_the_talk_cannot_take_it() {
        let (tmp, store, talk) = talks();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question("implement");
        questions.put(&mut q).unwrap();
        std::fs::remove_dir_all(tmp.path().join("talks")).unwrap();
        assert!(begin(&questions, &store, &q, &talk).is_err());
        assert!(questions.get(&q.id).unwrap().consult.is_none());
    }

    #[test]
    fn an_older_question_file_reads_without_a_consult() {
        let mut q = question("implement");
        q.schema = 5;
        let mut v = serde_json::to_value(&q).unwrap();
        v.as_object_mut().unwrap().remove("consult");
        let back: Question = serde_json::from_value(v).unwrap();
        assert!(back.consult.is_none());
    }
}