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 typed ownership links for navigation,
//!   independently of consultation eligibility.
//! - **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;
    }
    let task = crate::daemon::task_of_question(tasks, q)?;
    let chat = chat_talk_of(tasks, task)?;
    talks.iter().find(|talk| talk.id == chat).cloned()
}

/// 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> {
    associated_talk_with_watches(tasks, talks, q, &[])
}

/// A typed ownership link, shared by association and the web's selective loader.
#[derive(Clone, Copy)]
pub(crate) enum QuestionTaskRef<'a> {
    Task(&'a str),
    Run(&'a str),
}

pub(crate) fn question_task_refs<'a>(
    q: &'a Question,
    watches: &'a [crate::release_watch::WatchState],
) -> Vec<QuestionTaskRef<'a>> {
    // Review-start reconciliation is filed under a task id, unlike normal
    // review questions; both the node and seat identify that producer.
    let names_task = q.run_names_task()
        || (q.node == crate::reconcile::NODE && q.seat == crate::reconcile::SEAT);
    let mut refs = vec![if names_task {
        QuestionTaskRef::Task(&q.run)
    } else {
        QuestionTaskRef::Run(&q.run)
    }];
    let mut matches = watches
        .iter()
        .filter(|w| w.question.as_deref() == Some(&q.id));
    // Duplicate records are ambiguous even if they currently name one owner.
    if let Some(watch) = matches.next()
        && matches.next().is_none()
    {
        refs.extend(watch.run.as_deref().map(QuestionTaskRef::Run));
        refs.extend(watch.held_task.as_deref().map(QuestionTaskRef::Task));
    }
    refs
}

/// Explicit links may fill a missing owner, but conflicting owners fail closed.
/// This navigation-only resolution never changes consultation eligibility.
pub(crate) fn associated_talk_with_watches(
    tasks: &[Task],
    talks: &[Talk],
    q: &Question,
    watches: &[crate::release_watch::WatchState],
) -> Option<Talk> {
    let mut owner: Option<&Task> = None;
    for reference in question_task_refs(q, watches) {
        for task in tasks.iter().filter(|task| match reference {
            QuestionTaskRef::Task(id) => task.id == id,
            QuestionTaskRef::Run(id) => task.runs.iter().any(|run| run == id),
        }) {
            if owner.is_some_and(|previous| previous.id != task.id) {
                return None;
            }
            owner = Some(task);
        }
    }
    let chat = chat_talk_of(tasks, owner?)?;
    talks.iter().find(|talk| talk.id == chat).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 and ended consultations stop counting, restoring normal
/// permissions on the next turn.
pub fn pending_consults(questions: &Questions, talk_id: &str) -> bool {
    questions.list().iter().any(|q| active_consult(q, talk_id))
}

/// Consultation activity is separate from the question's waiting state.
pub fn active_consult(q: &Question, talk_id: &str) -> bool {
    q.routed_task.is_none()
        && q.status.open()
        && q.consult_outcome.is_none()
        && q.consult.as_ref().is_some_and(|c| c.talk == talk_id)
}

/// A non-answer never wakes a waiter or executes a choice action.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ConsultOutcome {
    /// Literal owner outcome, with no execution effect.
    pub text: String,
    /// Complete latest owner send used as evidence.
    pub instruction: String,
    /// Originating conversation.
    pub talk: String,
    /// Persisted conversation turn index.
    pub turn: usize,
    /// Persisted send index within the turn.
    pub message: usize,
    /// Time consultation ended.
    pub at: Timestamp,
}

/// Closed inventory of additional consultation operations. Every execution
/// argument is bound by the owner's complete instruction, not just a task id.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub enum TaskOperation {
    /// File work in the conversation's repository with default execution options.
    Add {
        /// Complete task text.
        instruction: String,
        /// Whether to use one implementer.
        solo: bool,
    },
    /// Release a held or finished task.
    Release {
        /// Full resolved task id.
        task: String,
    },
    /// Hold a task for the owner.
    Hold {
        /// Full resolved task id.
        task: String,
        /// Complete hold reason.
        reason: String,
    },
    /// Mark work finished without running it.
    Done {
        /// Full resolved task id.
        task: String,
    },
    /// Adjust the queue priority.
    Priority {
        /// Full resolved task id.
        task: String,
        /// New priority.
        priority: i32,
    },
}

impl TaskOperation {
    /// A literal, unambiguous individual instruction; JSON avoids shell parsing.
    pub fn instruction(&self) -> String {
        format!(
            "task-operation: {}",
            serde_json::to_string(self).expect("task operation")
        )
    }
}

fn delivered_instruction<'a>(
    q: &Question,
    talk: &'a Talk,
    quote: Option<&str>,
) -> Result<(usize, usize, &'a str)> {
    if !active_consult(q, &talk.id) {
        bail!("no active consultation in this conversation");
    }
    let (turn, message, words) = current_instruction(talk)
        .context("no current persisted owner instruction; drain newer messages first")?;
    if quote != Some(words) {
        bail!("--quote must equal the entire latest persisted owner message");
    }
    let delivery = talk
        .turns
        .iter()
        .position(|t| {
            t.who == talk::Who::Operator
                && t.body.contains(crate::prompt::CHAT_CONSULT_HEADING)
                && t.body.contains(&q.id)
        })
        .context("question has not been delivered")?;
    let consult = q.consult.as_ref().expect("active consult");
    if turn < delivery
        || talk.turns[turn].at < consult.at
        || (turn == delivery
            && !talk.turns[turn].messages[..message].iter().any(|m| {
                m.body.contains(crate::prompt::CHAT_CONSULT_HEADING) && m.body.contains(&q.id)
            }))
    {
        bail!("owner instruction must follow consultation delivery");
    }
    Ok((turn, message, words))
}

/// Caller holds the conversation evidence lock until the queue mutation ends.
pub fn authorize_task_operation(
    questions: &Questions,
    talk: &Talk,
    operation: &TaskOperation,
    quote: Option<&str>,
) -> Result<()> {
    let expected = operation.instruction();
    if !questions.list().iter().any(|q| {
        delivered_instruction(q, talk, quote).is_ok_and(|(_, _, words)| {
            words
                .strip_prefix("task-operation: ")
                .and_then(|json| serde_json::from_str::<TaskOperation>(json).ok())
                .as_ref()
                == Some(operation)
        })
    }) {
        bail!(
            "task operation requires individual post-delivery owner evidence and --quote of the entire latest send; required instruction: {expected}"
        );
    }
    Ok(())
}

/// Store an evidenced alternative and end only the consultation. Caller keeps
/// the conversation evidence lock until Questions::update publishes the note.
pub fn record_other(q: &mut Question, talk: &Talk, text: &str, quote: Option<&str>) -> Result<()> {
    if text.trim().is_empty() || q.free_text() || q.choices.iter().any(|c| c == text) {
        bail!("--reply-other requires a nonempty outcome outside the offered choices");
    }
    let (turn, message, words) = delivered_instruction(q, talk, quote)?;
    // A note is the owner's literal decision, never an assistant paraphrase
    // that could smuggle an execution authorization into retained evidence.
    if words != text {
        bail!("--reply-other must equal the owner's entire latest message");
    }
    q.schema = crate::ask::SCHEMA;
    q.consult_outcome = Some(ConsultOutcome {
        text: text.to_owned(),
        instruction: words.to_owned(),
        talk: talk.id.clone(),
        turn,
        message,
        at: Timestamp::now(),
    });
    Ok(())
}

/// 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| {
        anyhow::ensure!(
            r.routed_task.is_none(),
            "task decision is delivered by its own chat turn"
        );
        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 approval answer 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 node != CHAT_NODE || QuestionKind::of(q) != QuestionKind::Merge {
        return Ok(());
    }
    let release_run = if q.run.is_empty() && q.node == crate::bump::NOTICE_NODE {
        crate::run::try_home()
            .and_then(|home| crate::release_watch::state_for_question(&home, &q.id))
            .and_then(|state| state.run)
    } else {
        None
    };
    let run = release_run.as_deref().unwrap_or(&q.run);
    if let Some(task) = tasks
        .iter()
        .find(|t| t.id == run || t.runs.iter().any(|id| id == 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 covered by the instruction.
    pub choice: String,
    /// Execution facts validated when this answer was authorized.
    #[serde(default)]
    pub question_fingerprint: String,
    /// The conversation agent interpreted this complete owner send as consent
    /// to the single persisted internal recovery recommendation.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub recovery_interpretation: Option<String>,
}

/// Immutable recommendation-time binding to question choices and targets.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RecommendationManifest {
    /// Fingerprint of the full persona reply.
    pub body: String,
    /// Full question IDs and fingerprints of their execution-relevant facts.
    pub questions: std::collections::BTreeMap<String, String>,
    /// Canonical choice and structured effect captured with the persona reply.
    /// Old transcripts cannot acquire this binding retrospectively.
    #[serde(default)]
    pub recovery: std::collections::BTreeMap<String, RecoveryRecommendation>,
}

/// Execution binding published with a single internal recovery recommendation.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RecoveryRecommendation {
    /// Canonical stored label, independent of any alias used in the reply.
    pub choice: String,
    /// Structured effect; absent only for issuer-defined legacy keep-held.
    pub action: Option<crate::ask::ChoiceAction>,
    /// Run owning the question at recommendation time.
    pub run: String,
}

/// Fingerprint of execution-relevant question facts, independent of answer state.
pub fn question_fingerprint(q: &Question) -> String {
    let facts = serde_json::json!({
        "id": q.id, "run": q.run, "node": q.node, "seat": q.seat,
        "summary": q.summary, "detail": q.detail, "choices": q.choices,
        "actions": q.actions, "merge_target": q.merge_target,
        "posting_comparison": q.posting_comparison,
        "decision_context": q.decision_context, "triage_generation": q.triage_generation,
    });
    crate::github_text::fingerprint("", &facts.to_string())
}

/// Replacement title explicitly supplied with its question/content binding.
/// The final posting gate still vets this literal text before publication.
pub fn owner_title<'a>(q: &Question, instruction: &'a str) -> Option<&'a str> {
    if instruction.starts_with("    ") || instruction.starts_with('\t') {
        return None;
    }
    let (command, title) = instruction.trim().split_once('\n')?;
    let expected = format!(
        "{}: {} {}",
        q.id,
        crate::github_text::USE_MY_TEXT,
        question_fingerprint(q)
    );
    (command == expected && !title.trim().is_empty() && !title.contains('\n'))
        .then_some(title.trim())
}

/// Capture at publication, never reconstruct new authority from old history.
pub fn recommendation_manifest(body: &str, questions: &Questions) -> RecommendationManifest {
    let snapshot = questions.list();
    RecommendationManifest {
        body: crate::github_text::fingerprint("", body),
        recovery: snapshot
            .iter()
            .filter_map(|q| {
                let choice = recommended_choice(body, q, questions)?;
                (q.status.open() && contextual_recovery(q, choice)).then(|| {
                    (
                        q.id.clone(),
                        RecoveryRecommendation {
                            choice: choice.into(),
                            action: q.actions.get(choice).cloned(),
                            run: q.run.clone(),
                        },
                    )
                })
            })
            .collect(),
        questions: snapshot
            .iter()
            .filter(|q| q.status.open() && recommended_choice(body, q, questions).is_some())
            .map(|q| (q.id.clone(), question_fingerprint(q)))
            .collect(),
    }
}

/// 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 decision.
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 {
        return Ok(None);
    }
    // Before delivery, preserve already-supported legacy consultation replies.
    if q.routed_task.is_some() && q.routed_chat.is_some() {
        validate_routed_answer(q, talks, run, reply)?;
        return Ok(None);
    }
    if reply == crate::land::HOLD && q.actions.contains_key(reply) {
        bail!("hold cannot execute a question action");
    }
    let talk = talks.get(run)?;
    if reply != crate::land::HOLD
        && current_instruction(&talk).is_some_and(|(_, _, words)| {
            words.starts_with("    ")
                || words.starts_with('\t')
                || words.trim_start().starts_with("```")
                || words.trim_start().starts_with('>')
        })
    {
        bail!("quoted or code-block owner text is not an explicit instruction for this action");
    }
    if let Some(evidence) = choice_authority(q, questions, &talk, reply, quote) {
        validate_effect(
            q,
            reply,
            &evidence.instruction,
            quote.unwrap_or_default(),
            true,
        )?;
        // A running original consultation can still be preparing a decision.
        // Holds are safe to interrupt; other choices wait for that turn to stop.
        if reply != crate::land::HOLD
            && let Some(consult) = &q.consult
            && consult.talk != run
            && talks.turn_held(&consult.talk)
        {
            bail!("original consultation is active; wait for its turn to finish before answering");
        }
        return Ok(Some(evidence));
    }
    if current_instruction(&talk).is_some_and(|(_, _, words)| batch_instruction(words)) {
        validate_effect(
            q,
            reply,
            current_instruction(&talk).unwrap().2,
            quote.unwrap_or_default(),
            true,
        )?;
        bail!(
            "batch authorization requires a current, unambiguous recommendation for this exact offered choice and unchanged question target/content"
        );
    }
    // Preserve already supported consulted-chat answers. Independent evidence
    // is required when consultation delivery cannot bind this conversation.
    if q.consult.as_ref().is_some_and(|c| c.talk == run) {
        validate_answer(q, talks, run, node, reply, quote)?;
        if QuestionKind::of(q) == QuestionKind::Ordinary && consent(q, reply) != Consent::Delegable
        {
            let (_, _, instruction) = current_instruction(&talk)
                .context("this action requires a current explicit owner instruction")?;
            validate_effect(q, reply, instruction, quote.unwrap_or_default(), false)?;
        }
        return Ok(None);
    }
    if let Some((_, _, instruction)) = current_instruction(&talk) {
        validate_effect(q, reply, instruction, quote.unwrap_or_default(), true)?;
    }
    bail!("{}", diagnose_authorization_failure(q, &talk, reply, quote))
}

/// Explain a refused binding without changing the authorization policy.
fn diagnose_authorization_failure(
    q: &Question,
    talk: &Talk,
    reply: &str,
    quote: Option<&str>,
) -> &'static str {
    if !q.free_text() && !q.choices.iter().any(|choice| choice == reply) {
        return "alias mismatch: resolve an offered label or supported action alias with `magi answer --list`";
    }
    let queued_owner = (!talk.pending.is_empty() || !talk.pending_messages.is_empty())
        && (talk.pending_messages.is_empty()
            || talk
                .pending_messages
                .iter()
                .any(|send| send.sender != Some(talk::Sender::Magi)));
    if queued_owner {
        return "stale persisted evidence: a queued owner send may retract consent; wait for it to be drained and read it again";
    }
    let Some((turn, _, instruction)) = current_instruction(talk) else {
        return "missing context: no current persisted owner send; read the conversation again";
    };
    if quote.is_some_and(|quote| quote != instruction) {
        return "stale persisted evidence: --quote must preserve the entire latest owner send; read it again";
    }
    if talk.turns[turn].at < q.asked_at {
        return "stale persisted evidence: the latest owner send predates this question; obtain a current decision";
    }
    if contextual_recovery(q, reply) {
        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";
    }
    "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"
}

/// Only structured internal recovery effects qualify. Approval and publication
/// questions never acquire contextual consent, even if actions were attached.
fn contextual_recovery(q: &Question, reply: &str) -> bool {
    QuestionKind::of(q) == QuestionKind::Ordinary
        && consent(q, reply) == Consent::Delegable
        && (matches!(q.actions.get(reply), Some(crate::ask::ChoiceAction::Resume { run }) if !run.is_empty())
            || matches!(q.actions.get(reply), Some(crate::ask::ChoiceAction::Hold))
            || crate::triage::contextual_keep_held(q, reply))
}

/// Agent-mediated interpretation, bound to a full real owner send and the
/// immutable recommendation. Call under the same locks as `authorize_answer`.
/// This does not execute recovery; the daemon retains all ownership and merge gates.
pub fn authorize_recovery(
    q: &Question,
    talks: &Talks,
    questions: &Questions,
    seat: (&str, &str),
    reply: &str,
    instruction: &str,
) -> Result<Option<AnswerAuthority>> {
    let (run, node) = seat;
    if node != CHAT_NODE || !contextual_recovery(q, reply) {
        bail!(
            "insufficient consent: contextual interpretation is limited to structured internal resume/keep-held actions; use this decision's explicit consent workflow"
        );
    }
    let talk = talks.get(run)?;
    let evidence = recovery_authority(q, questions, &talk, reply, instruction)?;
    if q.consult
        .as_ref()
        .is_some_and(|c| c.talk != run && talks.turn_held(&c.talk))
    {
        bail!("original consultation is active; wait for its turn to finish before answering");
    }
    Ok(Some(evidence))
}

fn recovery_authority(
    q: &Question,
    questions: &Questions,
    talk: &Talk,
    reply: &str,
    instruction: &str,
) -> Result<AnswerAuthority> {
    let (turn, message, latest) = current_instruction(talk)
        .context("missing context: wait for the current owner send to be persisted and drained")?;
    if instruction != latest {
        bail!(
            "stale persisted evidence: the complete latest owner send differs; read the conversation again, including any retraction"
        );
    }
    let owner_sends = talk.turns[turn]
        .messages
        .iter()
        .filter(|send| talk::sender(&talk.turns[turn], Some(send)) != talk::Sender::Magi)
        .count();
    if owner_sends != 1
        || latest.trim().is_empty()
        || latest.starts_with("    ")
        || latest.starts_with('\t')
        || latest.trim_start().starts_with(['>', '`'])
    {
        bail!("insufficient consent: quoted text or multiple owner sends require clarification");
    }
    let index = talk.turns[..turn]
        .iter()
        .rposition(|t| talk::sender(t, None) != talk::Sender::Magi)
        .context("missing context: present one explicit recovery recommendation first")?;
    let recommendation = &talk.turns[index];
    let manifest = talk.recommendations.get(&index).context(
        "missing context: no persisted recommendation binding; present the recommendation now",
    )?;
    if recommendation.who != talk::Who::Agent
        || talk::sender(recommendation, None) != talk::Sender::Persona
        || recommendation.posted_by.is_some()
        || manifest.body != crate::github_text::fingerprint("", &recommendation.body)
        || manifest.questions.get(&q.id) != Some(&question_fingerprint(q))
        || recommendation.at < q.asked_at
        || talk.turns[turn].at < recommendation.at
        || !q.status.open()
    {
        bail!(
            "stale persisted evidence: recommendation or question target changed; present the current recommendation again"
        );
    }
    if manifest.questions.len() != 1 {
        bail!(
            "ambiguous recovery context: multiple recommended decisions; clarify which question and action the owner intends"
        );
    }
    let binding = manifest.recovery.get(&q.id)
        .context("missing context: no structured recovery binding was saved; present a current recommendation")?;
    if binding.run != q.run || binding.action.as_ref() != q.actions.get(&binding.choice) {
        bail!("stale persisted evidence: the recommended recovery target changed");
    }
    if binding.choice != reply
        || recommended_choice(&recommendation.body, q, questions) != Some(reply)
    {
        bail!("alias mismatch: the selected action differs from the persisted recommendation");
    }
    Ok(AnswerAuthority {
        talk: talk.id.clone(),
        turn,
        message,
        instruction: latest.into(),
        instruction_fingerprint: crate::github_text::fingerprint("", latest),
        recommendation_turn: Some(index),
        recommendation_fingerprint: Some(manifest.body.clone()),
        question: q.id.clone(),
        choice: reply.into(),
        question_fingerprint: question_fingerprint(q),
        recovery_interpretation: Some(
            "owner approved the single current recovery recommendation".into(),
        ),
    })
}

/// Whether this turn needs the question-store write grant. Uses the same
/// persisted evidence and effect policy as the CLI; the CLI rechecks on write.
pub fn authorized_answers(questions: &Questions, talk: &Talk) -> bool {
    let quote = current_instruction(talk).map(|(_, _, body)| body);
    questions.list().iter().any(|q| {
        if q.free_text() {
            return quote
                .and_then(|s| s.split_once(':'))
                .is_some_and(|(_, reply)| {
                    choice_authority(q, questions, talk, reply.trim(), quote).is_some()
                });
        }
        q.choices.iter().any(|choice| {
            if contextual_recovery(q, choice)
                && quote.is_some_and(|words| {
                    recovery_authority(q, questions, talk, choice, words).is_ok()
                })
            {
                return true;
            }
            choice_authority(q, questions, talk, choice, quote).is_some_and(|e| {
                validate_effect(q, choice, &e.instruction, quote.unwrap_or_default(), true).is_ok()
            })
        })
    })
}

/// The complete standing-turn write grant, also used by sandbox regressions.
/// Consultations and bound owner decisions need question/evidence locks;
/// active consultations additionally need queue storage. Evidenced `done` and
/// dismissal effects need their exact run history/state directories. Never
/// grant the whole home or lift the sandbox.
pub fn answer_write_roots(
    questions: &Questions,
    talks: &Talks,
    talk: &Talk,
) -> Vec<std::path::PathBuf> {
    if !pending_consults(questions, &talk.id) && !authorized_answers(questions, talk) {
        return Vec::new();
    }
    let Some(home) = questions.root().parent() else {
        return Vec::new();
    };
    let quote = current_instruction(talk).map(|(_, _, words)| words);
    let mut roots = std::collections::BTreeSet::from([
        questions.root().to_path_buf(),
        talks.answer_lock_root(),
    ]);
    if pending_consults(questions, &talk.id) {
        roots.insert(home.join("queue"));
        if let Some(words) = quote
            && let Some(json) = words.strip_prefix("task-operation: ")
            && let Ok(op @ TaskOperation::Done { .. }) = serde_json::from_str::<TaskOperation>(json)
            && authorize_task_operation(questions, talk, &op, quote).is_ok()
            && let TaskOperation::Done { task } = op
            && let Ok(task) = crate::queue::Queue::at(home.join("queue")).get(&task)
        {
            for run in task.runs.iter().filter(|id| crate::run::is_run_id(id)) {
                let path = home.join("runs").join(run);
                if path.is_dir() {
                    roots.insert(path);
                }
            }
        }
    }
    for q in questions.list() {
        if q.status.open()
            && q.node == crate::land::APPROVAL_NODE
            && q.choices
                .iter()
                .any(|choice| choice == crate::merge_decision::DISMISS)
            && crate::run::is_run_id(&q.run)
            && authorize_answer(
                &q,
                talks,
                questions,
                &talk.id,
                CHAT_NODE,
                crate::merge_decision::DISMISS,
                quote,
            )
            .is_ok()
        {
            roots.insert(home.join("queue"));
            roots.insert(home.join("runs").join(&q.run));
        }
    }
    roots.into_iter().collect()
}

/// Compatibility name for callers of the earlier hold-only grant.
pub fn authorized_holds(questions: &Questions, talk: &Talk) -> bool {
    authorized_answers(questions, talk)
}

/// Queue state is only a diagnostic lead, never proof of lock ownership.
pub(crate) fn investigation_holder_hint(q: &Question, tasks: &[Task]) -> String {
    let task = tasks.iter().find(|task| {
        !q.run.is_empty()
            && (task.id == q.run || task.runs.contains(&q.run))
            && task.status == crate::queue::TaskStatus::Running
    });
    match task {
        Some(task) => format!(
            "Lock owner unknown. Related task {} is running and may hold the repository lock; inspect it with `magi task show {}`. ",
            task.id, task.id
        ),
        None => "Lock owner unknown. ".into(),
    }
}

/// Action consent is separate from consultation routing. This inventory uses
/// execution effects (including structured actions), not only question nodes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Consent {
    /// Explicit single or batch instruction is sufficient.
    Delegable,
    /// Exact PR/head approval is required.
    Merge,
    /// Exact offered consequence must be confirmed individually.
    Consequence,
    /// Chosen public text needs explicit confirmation.
    Posting,
}

/// Execution effects shared with question deputies. The issuer and seat are
/// part of the policy: identical labels can have different consumers.
pub fn destructive_effect(q: &Question, label: &str) -> bool {
    if label == crate::land::HOLD && !q.actions.contains_key(label) {
        return false;
    }
    // Conductor metadata describes the requested operation independently of
    // generated choice wording. Completing a task or changing settings needs
    // individual consent even when no structured ChoiceAction was attached.
    if q.node == crate::conduct::NODE
        && q.decision_context
            .as_deref()
            .and_then(|context| serde_json::from_str::<serde_json::Value>(context).ok())
            .is_some_and(|context| {
                matches!(
                    context
                        .get(3)
                        .and_then(|v| v.get(0))
                        .and_then(|v| v.as_str()),
                    Some("complete" | "configure")
                )
            })
    {
        return true;
    }
    if matches!(
        q.actions.get(label),
        Some(crate::ask::ChoiceAction::Done | crate::ask::ChoiceAction::AbandonRun { .. })
    ) {
        return true;
    }
    let at = q.choices.iter().position(|c| c == label);
    // By label, not position: the retry question orders its choices differently.
    if q.node == crate::github_text::ASK_NODE {
        return matches!(
            label,
            crate::github_text::POST_ORIGINAL
                | crate::github_text::USE_REWRITE
                | crate::github_text::USE_MY_TEXT
        );
    }
    match q.node.as_str() {
        crate::triage::NODE => at == Some(2),
        crate::triage::DEPS_NODE => matches!(at, Some(1 | 2)),
        n => n == crate::reconcile::NODE && q.seat == crate::reconcile::SEAT,
    }
}

/// Classify every offered decision, including generic structured actions.
pub fn consent(q: &Question, choice: &str) -> Consent {
    if choice == crate::land::APPROVE {
        Consent::Merge
    } else if q.node == crate::github_text::ASK_NODE
        && matches!(
            choice,
            crate::github_text::POST_ORIGINAL
                | crate::github_text::USE_REWRITE
                | crate::github_text::USE_MY_TEXT
        )
    {
        Consent::Posting
    } else if choice == crate::merge_decision::WITHDRAW
        || crate::merge_decision::selected_agent(choice).is_some()
        || destructive_effect(q, choice)
        || matches!(q.actions.get(choice), Some(crate::ask::ChoiceAction::Done))
    {
        Consent::Consequence
    } else {
        Consent::Delegable
    }
}

/// Deputies use their authenticated question thread rather than a talk.
/// A generic batch instruction cannot replace action-specific confirmation.
pub(crate) fn validate_batch_effect(
    q: &Question,
    choice: &str,
    message: &str,
    quote: &str,
) -> Result<()> {
    if batch_instruction(message) {
        validate_effect(q, choice, message, quote, false)?;
        bail!(
            "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"
        );
    }
    Ok(())
}

/// Validate action-specific evidence independently of its delivery route.
pub(crate) fn validate_effect(
    q: &Question,
    choice: &str,
    instruction: &str,
    quote: &str,
    bound: bool,
) -> Result<()> {
    let direct = format!("{}: {}", q.id, choice);
    match consent(q, choice) {
        Consent::Delegable => {}
        Consent::Merge => {
            let target = q.merge_target.as_ref().context("merge requires the specific reviewed PR and head binding; this legacy card has no recorded target")?;
            let token = format!("merge {}", target.head);
            if quote.trim() != instruction.trim()
                || !(instruction.trim() == token
                    || instruction.trim() == format!("{}: {token}", q.id))
            {
                bail!(
                    "merge requires separate explicit confirmation of this specific reviewed PR and head: `{}: {token}`; a batch recommendation is insufficient",
                    q.id
                );
            }
        }
        Consent::Consequence => {
            if quote.trim() != instruction.trim()
                || !(instruction.trim() == direct || (!bound && instruction.trim() == choice))
            {
                bail!(
                    "`{choice}` requires an individual owner confirmation of the exact offered consequence: `{direct}`; batch authorization is insufficient"
                );
            }
        }
        Consent::Posting => {
            if bound {
                let token = format!("{}: {} {}", q.id, choice, question_fingerprint(q));
                let content_matches = if choice == crate::github_text::USE_MY_TEXT {
                    owner_title(q, instruction).is_some()
                } else {
                    instruction.trim() == token
                };
                if !content_matches || quote.trim() != instruction.trim() {
                    bail!(
                        "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"
                    );
                }
            } else if !crate::github_text::posting_authorized(instruction, quote, choice) {
                bail!(
                    "posting confirmation requires an explicit, unconditional instruction for the chosen text; batch authorization is insufficient"
                );
            }
        }
    }
    Ok(())
}

/// The entire latest persisted owner send, when it is a usable instruction.
pub(crate) fn latest_owner_instruction(talk: &Talk) -> Option<&str> {
    current_instruction(talk).map(|(_, _, words)| words)
}

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 choice_authority(
    q: &Question,
    questions: &Questions,
    talk: &Talk,
    reply: &str,
    quote: Option<&str>,
) -> Option<AnswerAuthority> {
    if !q.status.open()
        || !(q.free_text() || q.choices.iter().any(|c| c == reply))
        || (reply == crate::land::HOLD && q.actions.contains_key(reply))
    {
        return None;
    }
    let (turn, message, instruction) = current_instruction(talk)?;
    if reply != crate::land::HOLD
        && (instruction.starts_with("    ") || instruction.starts_with('\t'))
    {
        return None;
    }
    if talk.turns[turn].at < q.asked_at {
        return None;
    }
    let direct = direct_choice(instruction, q, questions, reply)
        || (reply == crate::github_text::USE_MY_TEXT
            && q.node == crate::github_text::ASK_NODE
            && owner_title(q, instruction).is_some())
        || (consent(q, reply) == Consent::Posting
            && instruction.trim() == format!("{}: {} {}", q.id, reply, question_fingerprint(q)))
        || (consent(q, reply) == Consent::Merge
            && q.merge_target
                .as_ref()
                .is_some_and(|t| instruction.trim() == format!("{}: merge {}", q.id, t.head)));
    let quote_matches = if contextual_recovery(q, reply) {
        quote == Some(instruction)
    } else {
        quote.map(str::trim) == Some(instruction.trim())
    };
    let persisted_recovery = quote.is_none() && direct && contextual_recovery(q, reply);
    if !quote_matches && !persisted_recovery {
        return None;
    }
    let recommendation_turn = if direct {
        None
    } else {
        if message != 0 || q.free_text() || !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_choice(&recommendation.body, q, questions) != Some(reply)
        {
            return None;
        }
        if let Some(manifest) = talk.recommendations.get(&index) {
            if manifest.body != crate::github_text::fingerprint("", &recommendation.body)
                || manifest.questions.get(&q.id) != Some(&question_fingerprint(q))
            {
                return None;
            }
        } else if reply != crate::land::HOLD {
            // Old transcripts can prove a safe hold, but cannot establish what
            // mutable action/target was shown before manifests were persisted.
            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: reply.into(),
        question_fingerprint: question_fingerprint(q),
        recovery_interpretation: None,
    })
}

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 mut compact = compact.trim_end_matches(['!', '！', '。', '.', 'ー', '〜', '～']);
    let prefixes = [
        "おっけー",
        "オッケー〜",
        "オッケー",
        "おっけ〜",
        "おけ",
        "オケ",
        "ok",
        "OK",
        "Ok",
        "ありがとうございます",
        "ありがとう",
        "ありがと",
        "了解です",
        "了解",
        "はい",
        "じゃあ",
        "では",
    ];
    while let Some(rest) = prefixes
        .iter()
        .find_map(|prefix| compact.strip_prefix(prefix))
    {
        compact = rest.trim_start_matches(['!', '！', '、', ',', '。', '.', 'ー', '〜', '～']);
    }
    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 && questions.resolve_id(reference).is_ok_and(|id| id == q.id))
}

fn direct_choice(text: &str, q: &Question, questions: &Questions, reply: &str) -> bool {
    // One complete instruction, binding both the question and the offered
    // choice. Other actions (including closing the PR) are never inferred.
    let text = text.trim();
    let binding = text.split_once(':').or_else(|| {
        if contextual_recovery(q, reply) {
            text.split_once(char::is_whitespace)
        } else {
            None
        }
    });
    let Some((reference, choice)) = binding else {
        return false;
    };
    let reference = reference.trim().trim_matches('`');
    (reference == q.id
        || ((reply == crate::land::HOLD || contextual_recovery(q, reply))
            && refers_to(reference, q, questions)))
        && (choice.trim() == reply
            || (contextual_recovery(q, reply)
                && q.resolve_reply(choice.trim())
                    .is_ok_and(|resolved| resolved == reply)))
}

fn recommended_choice<'a>(body: &str, q: &'a Question, questions: &Questions) -> Option<&'a str> {
    // A row must match the entire offered label, including prefixes such as
    // `dismiss:` or `reimplement:`. Prefix/substring matches confer no authority.
    // 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 None;
        }
        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_map(|c| q.resolve_reply(c).ok())
                .collect();
            if other_question || references != 1 || choices.len() != 1 {
                return None;
            }
            rows.push(
                q.choices
                    .iter()
                    .find(|choice| choice.as_str() == choices[0])?
                    .as_str(),
            );
            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);
        if !refers_to(reference, q, questions) {
            return None;
        }
        let choice = choice?;
        // Preserve the legacy safe-hold wording without granting PR closure.
        let choice = if choice == "hold、その後PRを閉じる" {
            crate::land::HOLD
        } else {
            choice.trim_matches('`')
        };
        rows.push(
            q.choices
                .iter()
                .find(|offered| {
                    q.resolve_reply(choice)
                        .is_ok_and(|resolved| offered.as_str() == resolved)
                })?
                .as_str(),
        );
    }
    (rows.len() == 1).then(|| rows[0])
}

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 {
        if crate::deputy::merge_gated(q) {
            return Self::Merge;
        }
        match q.node.as_str() {
            crate::github_text::ASK_NODE => Self::Posting,
            _ => Self::Ordinary,
        }
    }
}

/// Routed decisions use the chat agent's interpretation of the owner's request,
/// while retaining conversation ownership and open-record checks. Legacy
/// consultation consent below is unchanged.
fn validate_routed_answer(q: &Question, talks: &Talks, run: &str, reply: &str) -> Result<()> {
    anyhow::ensure!(
        q.routed_chat.as_deref() == Some(run),
        "decision belongs to a different chat"
    );
    anyhow::ensure!(q.status.open(), "decision is no longer open");
    anyhow::ensure!(talks.get(run)?.status.open(), "the task chat is closed");
    let home = talks.root().parent().context("talk store has no home")?;
    let queue = crate::queue::Queue::at(home.join("queue"));
    crate::chat_control::owned_task(
        &queue,
        run,
        q.routed_task.as_deref().context("decision has no task")?,
    )?;
    anyhow::ensure!(
        q.free_text() || q.choices.iter().any(|c| c == reply),
        "decision requires an offered choice"
    );
    Ok(())
}

/// 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<()> {
    if node == CHAT_NODE && q.routed_task.is_some() && q.routed_chat.is_some() {
        return validate_routed_answer(q, talks, run, reply);
    }
    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 && reply == crate::github_text::USE_MY_TEXT {
        bail!(
            "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"
        );
    }
    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 is insufficient for merge; 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 {
        if reply == crate::land::APPROVE {
            if let Some(target) = &q.merge_target {
                bail!(
                    "merge requires the latest whole owner message and --quote to be exactly `merge {}` for the displayed PR/head",
                    target.head
                );
            }
            bail!("merge requires a verbatim quote of the latest explicit owner merge instruction");
        }
        bail!(
            "`{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"
        );
    }
    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(())
}

/// Try the short cross-process draft-publication fence for one talk.
pub fn try_publication_guard(talks: &Talks, talk_id: &str) -> Result<Option<crate::worker::Lease>> {
    crate::worker::try_lease(talks.home(), &format!("chat-publication:{talk_id}"))
}

/// Serialize queue/record -> worker acknowledgement against a drainer's
/// generation scan + promotion. Never hold this fence across an agent call.
///
/// Lock order deliberately differs: a drainer owns TurnLease then waits here;
/// ingress owns publication but only *tries* TurnLease and queues when busy.
/// Ingress must never wait for TurnLease while holding publication.
pub async fn publication_guard(talks: &Talks, talk_id: &str) -> Result<crate::worker::Lease> {
    loop {
        if let Some(guard) = try_publication_guard(talks, talk_id)? {
            return Ok(guard);
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
}

/// Reserve and launch a detached drainer for an explicitly submitted draft.
///
/// Every submission has its own durable reservation. A shared key would lose
/// arrivals between the previous drainer's last read and its job settlement.
/// The talk's existing turn lease, not the job key, serializes CLI sessions.
pub async fn spawn_turn(talks: &Talks, talk_id: &str) -> Result<crate::worker::Job> {
    spawn_turn_with_source(talks, talk_id, false, None).await
}

/// Continue an operator turn already persisted by the HTTP admission path.
/// Reserve this job before releasing the caller's turn lease.
pub async fn spawn_recorded_turn(
    talks: &Talks,
    talk_id: &str,
    turn: usize,
) -> Result<crate::worker::Job> {
    spawn_turn_with_source(talks, talk_id, false, Some(turn)).await
}

pub(crate) async fn spawn_notice_turn(talks: &Talks, talk_id: &str) -> Result<crate::worker::Job> {
    spawn_turn_with_source(talks, talk_id, true, None).await
}

async fn spawn_turn_with_source(
    talks: &Talks,
    talk_id: &str,
    automatic: bool,
    recorded_turn: Option<usize>,
) -> Result<crate::worker::Job> {
    let talk = talks.get(talk_id)?;
    let recorded = recorded_turn
        .map(|index| {
            talk.turns
                .get(index)
                .context("recorded chat admission disappeared")
        })
        .transpose()?;
    let key = format!(
        "chat:{}:{}",
        talk.id,
        crate::rng::SplitMix64::new(crate::rng::entropy()).uuid_v4()
    );
    let result = crate::worker::spawn(
        talks.home(),
        "chat",
        &key,
        serde_json::json!({
            "talk": &talk.id, "automatic": automatic, "recorded_turn": recorded_turn,
        }),
    )
    .await;
    match result {
        Err(error) if recorded.is_some() && crate::worker::not_started(&error) => {
            let index = recorded_turn.expect("recorded snapshot");
            match talk::restore_recorded_draft(
                talks, &talk.id, index, recorded.expect("recorded snapshot"),
            ) {
                Ok(true) => Err(error.context(
                    "the agent never started; your send is saved as a draft — use Resume draft",
                )),
                Ok(false) => Err(error.context(
                    "the agent never started, but the transcript changed before draft restoration; the recorded send was retained",
                )),
                Err(restore) => Err(error.context(format!(
                    "the agent never started and its recorded send was retained; draft restoration failed: {restore:#}",
                ))),
            }
        }
        other => other,
    }
}

/// Includes startup reservations, before the child acquires the turn lease.
pub fn turn_worker_active(talks: &Talks, talk_id: &str) -> bool {
    crate::worker::active(talks.home()).is_ok_and(|jobs| {
        jobs.iter().any(|job| {
            job.kind == "chat"
                && job.payload.get("talk").and_then(serde_json::Value::as_str) == Some(talk_id)
        })
    })
}

/// Task-authored transcript posts do not answer a submitted owner turn.
fn recorded_answered(talk: &Talk, index: usize) -> bool {
    index
        .checked_add(1)
        .and_then(|next| talk.turns.get(next..))
        .is_some_and(|turns| {
            turns
                .iter()
                .any(|turn| turn.who == talk::Who::Agent && turn.posted_by.is_none())
        })
}

/// Hidden worker entry point. The original talk/session and cross-process
/// lease stay authoritative even when an older worker is still responding.
pub async fn run_turn_worker(home: &std::path::Path, payload: serde_json::Value) -> Result<()> {
    let id = payload
        .get("talk")
        .and_then(serde_json::Value::as_str)
        .context("chat worker has no talk id")?;
    let automatic = payload
        .get("automatic")
        .and_then(serde_json::Value::as_bool)
        .unwrap_or(false);
    let recorded_turn = payload
        .get("recorded_turn")
        .and_then(serde_json::Value::as_u64)
        .map(usize::try_from)
        .transpose()?;
    let talks = Talks::at(home.join("talks"));
    let current = crate::worker::current_job(home)?;
    loop {
        let talk = talks.get(id)?;
        if !talk.status.open()
            || (recorded_turn.is_none()
                && talk.pending.is_empty()
                && talk.pending_attachments.is_empty())
        {
            return Ok(());
        }
        if automatic && !crate::chat_notify::pending_is_only_notices(&talk) {
            return Ok(());
        }
        if let Some(lease) = talks.claim_turn(id)? {
            let publication = publication_guard(&talks, id).await?;
            // Reload the draft after ownership is acquired: a notice must not
            // turn an unsent operator draft into an automatic submission.
            if automatic && !crate::chat_notify::pending_is_only_notices(&talks.get(id)?) {
                return Ok(());
            }
            let mut fresh = talks.get(id)?;
            let active = crate::worker::active(home)?;
            if recorded_turn.is_none() && current.as_ref().is_some_and(|current| {
                active.iter().any(|job| {
                    job.kind == "chat" && job.id != current.id
                        && job.generation != current.generation
                        && job.payload.get("talk").and_then(serde_json::Value::as_str) == Some(id)
                        && job.payload.get("recorded_turn").and_then(serde_json::Value::as_u64).is_none()
                        && (!job.payload.get("automatic").and_then(serde_json::Value::as_bool).unwrap_or(false)
                            || crate::chat_notify::pending_is_only_notices(&fresh))
                        // Sequence zero belongs to pre-ordering records. In
                        // that case relinquish rather than infer from time.
                        && job.sequence > 0
                        && (current.sequence == 0 || job.sequence > current.sequence)
                })
            }) {
                return Ok(());
            }
            if recorded_turn.is_none()
                && active.iter().any(|job| {
                    if job.kind != "chat"
                        || job.payload.get("talk").and_then(serde_json::Value::as_str) != Some(id)
                    {
                        return false;
                    }
                    let Some(index) = job
                        .payload
                        .get("recorded_turn")
                        .and_then(serde_json::Value::as_u64)
                        .and_then(|index| usize::try_from(index).ok())
                    else {
                        return false;
                    };
                    fresh
                        .turns
                        .get(index)
                        .is_some_and(|turn| turn.who == talk::Who::Operator)
                        && !recorded_answered(&fresh, index)
                })
            {
                // A submitted immediate turn is already in the transcript,
                // but its child has not acquired the lease yet. Do not let a
                // queued follow-up overtake it during process startup.
                drop(publication);
                drop(lease);
                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                continue;
            }
            let (cfg, _) = Config::discover(&fresh.repo, None)?;
            let text = if let Some(index) = recorded_turn {
                let turn = fresh
                    .turns
                    .get(index)
                    .context("recorded chat turn disappeared")?;
                anyhow::ensure!(
                    turn.who == talk::Who::Operator,
                    "recorded turn is not operator input"
                );
                if recorded_answered(&fresh, index) {
                    return Ok(());
                }
                turn.body.clone()
            } else {
                let Some(text) = talk::drain(&mut fresh, &talks)? else {
                    return Ok(());
                };
                text
            };
            drop(publication);
            // Exactly one invocation per job. A submission accepted by the
            // successor UI already owns a successor-generation reservation;
            // never pull it into this old worker after the current reply.
            return talk::respond(&lease, &mut fresh, &talks, &cfg, &text).await;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
}

#[derive(serde::Serialize, serde::Deserialize)]
struct ServiceRequest {
    service: String,
    repo: std::path::PathBuf,
    cfg: Option<Config>,
    #[serde(default)]
    explicit_config: Option<std::path::PathBuf>,
}

fn service_due(home: &std::path::Path, service: &str, now: Timestamp) -> bool {
    let interval = match service {
        "waiter" => crate::waiter::TICK,
        "deputies" => crate::deputy::TICK,
        "release-watch" => crate::release_watch::LAP,
        _ => return false,
    };
    std::fs::read_to_string(home.join("service-ticks").join(service))
        .ok()
        .and_then(|text| text.parse::<i64>().ok())
        .is_none_or(|last| now.as_second().saturating_sub(last) >= interval.as_secs() as i64)
}

fn service_has_work(
    home: &std::path::Path,
    repo: &std::path::Path,
    cfg: Option<&Config>,
    service: &str,
    now: Timestamp,
) -> bool {
    match service {
        "waiter" => crate::waiter::Waiter::new(
            Questions::at(home.join("questions")),
            home.to_path_buf(),
            cfg.cloned(),
        )
        .has_work(now),
        "deputies" => crate::deputy::Deputies::new(
            Questions::at(home.join("questions")),
            home.to_path_buf(),
            cfg.cloned(),
            repo.to_path_buf(),
            cfg.map_or(0, |cfg| cfg.daemon.max_deputies),
            std::sync::Arc::new(|| false),
        )
        .has_work(now),
        "release-watch" => cfg.is_some_and(|cfg| cfg.daemon.release_stall_minutes > 0),
        _ => false,
    }
}

/// Finite service work outlives the scheduler, but never pins a service host
/// to an obsolete executable. Its next tick is started by the new scheduler.
pub async fn start_services(
    home: &std::path::Path,
    repo: &std::path::Path,
    cfg: Option<&Config>,
    explicit_config: Option<&std::path::Path>,
) -> Result<()> {
    let active = crate::worker::active(home)?;
    for service in ["waiter", "deputies", "release-watch"] {
        let key = format!("service:{service}");
        // Adoption is by operation key, not by the new scheduler's config
        // payload/schema. An old invocation owns its started request.
        if active.iter().any(|job| job.key == key) {
            continue;
        }
        if !service_due(home, service, Timestamp::now()) {
            continue;
        }
        let service_cfg = if service == "release-watch" {
            // The original watcher reloaded settings each lap. Do this cheap
            // admission read here so disabled/idle services never fork a CLI.
            Config::discover(repo, explicit_config)
                .ok()
                .map(|(cfg, _)| cfg)
        } else {
            cfg.cloned()
        };
        if !service_has_work(home, repo, service_cfg.as_ref(), service, Timestamp::now()) {
            let stamps = home.join("service-ticks");
            std::fs::create_dir_all(&stamps)?;
            std::fs::write(
                stamps.join(service),
                Timestamp::now().as_second().to_string(),
            )?;
            continue;
        }
        let payload = serde_json::to_value(ServiceRequest {
            service: service.to_owned(),
            repo: repo.to_path_buf(),
            cfg: service_cfg,
            explicit_config: explicit_config.map(std::path::Path::to_path_buf),
        })?;
        if let Err(error) = crate::worker::spawn(home, "service", &key, payload).await {
            tracing::warn!("service {service} worker could not start: {error:#}");
        }
    }
    Ok(())
}

/// One leased service tick. Existing question leases, seat records, and release
/// receipts remain the only authorities for delivery and once-only effects.
pub async fn run_service_worker(home: &std::path::Path, payload: serde_json::Value) -> Result<()> {
    let request: ServiceRequest = serde_json::from_value(payload)?;
    anyhow::ensure!(
        matches!(
            request.service.as_str(),
            "waiter" | "deputies" | "release-watch"
        ),
        "unknown conversation service"
    );
    let stamps = home.join("service-ticks");
    std::fs::create_dir_all(&stamps)?;
    std::fs::write(
        stamps.join(&request.service),
        Timestamp::now().as_second().to_string(),
    )?;
    let questions = Questions::at(home.join("questions"));
    match request.service.as_str() {
        "waiter" => {
            crate::waiter::Waiter::new(questions, home.to_path_buf(), request.cfg)
                .tick(Timestamp::now(), &|| false)
                .await;
        }
        "deputies" => {
            let max = request.cfg.as_ref().map_or_else(
                || Config::default().daemon.max_deputies,
                |cfg| cfg.daemon.max_deputies,
            );
            let deputies = crate::deputy::Deputies::new(
                questions,
                home.to_path_buf(),
                request.cfg,
                request.repo,
                max,
                std::sync::Arc::new(|| false),
            );
            deputies.dispatch(Timestamp::now()).await?;
        }
        "release-watch" => {
            if let Ok((cfg, _)) =
                Config::discover(&request.repo, request.explicit_config.as_deref())
            {
                let (repos, minutes) =
                    crate::daemon::release_watch_settings_from_config(home, &request.repo, &cfg);
                if minutes > 0 {
                    crate::release_watch::Watcher::new(
                        Box::new(crate::release_watch::GhForge),
                        home.to_path_buf(),
                    )
                    .lap(
                        &repos,
                        (minutes as i64).saturating_mul(60),
                        Timestamp::now().as_second(),
                        &|| false,
                    )
                    .await;
                }
            }
        }
        _ => unreachable!("validated service"),
    }
    // Preserve the delay after completion even when the scheduler changes
    // executable or restarts while this tick is still running.
    std::fs::write(
        stamps.join(&request.service),
        Timestamp::now().as_second().to_string(),
    )?;
    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;

    #[test]
    fn service_cadence_is_durable_and_keeps_the_watchers_own_interval() {
        let home = tempfile::tempdir().expect("home");
        let stamps = home.path().join("service-ticks");
        std::fs::create_dir_all(&stamps).expect("stamps");
        for service in ["waiter", "deputies", "release-watch"] {
            std::fs::write(stamps.join(service), "100").expect("completion timestamp");
        }
        let at = |seconds| Timestamp::from_second(seconds).unwrap();
        assert!(!service_due(home.path(), "waiter", at(104)));
        assert!(service_due(home.path(), "waiter", at(105)));
        assert!(!service_due(home.path(), "deputies", at(104)));
        assert!(service_due(home.path(), "deputies", at(105)));
        assert!(!service_due(home.path(), "release-watch", at(399)));
        assert!(service_due(home.path(), "release-watch", at(400)));
    }

    #[test]
    fn idle_services_do_not_need_detached_processes() {
        let home = tempfile::tempdir().expect("home");
        for service in ["waiter", "deputies", "release-watch"] {
            assert!(!service_has_work(
                home.path(),
                home.path(),
                None,
                service,
                Timestamp::now(),
            ));
        }
        let mut disabled = Config::default();
        disabled.daemon.release_stall_minutes = 0;
        assert!(!service_has_work(
            home.path(),
            home.path(),
            Some(&disabled),
            "release-watch",
            Timestamp::now(),
        ));
    }

    #[test]
    fn a_task_post_does_not_answer_an_admitted_owner_send() {
        let (_home, talks, mut talk) = talks();
        talk::record(&mut talk, &talks, "owner send", Vec::new()).expect("owner turn");
        let index = talk.turns.len() - 1;
        talk::post(&talks, &talk.id, "unrelated task report", "task-seat").expect("task post");
        let mut fresh = talks.get(&talk.id).expect("persisted transcript");
        assert!(!recorded_answered(&fresh, index));
        let mut response = fresh.turns.last().expect("post").clone();
        response.posted_by = None;
        response.body = "actual persona response".to_owned();
        fresh.turns.push(response);
        assert!(recorded_answered(&fresh, index));
    }

    #[tokio::test]
    async fn explicit_release_config_overrides_the_repository_without_launching_disabled_watchers()
    {
        let home = tempfile::tempdir().expect("home");
        let repo = tempfile::tempdir().expect("repo");
        let explicit = home.path().join("serve-config.toml");
        std::fs::write(
            repo.path().join("magi.toml"),
            "[daemon]\nrelease_stall_minutes = 60\n",
        )
        .expect("repository watcher enabled");
        std::fs::write(&explicit, "[daemon]\nrelease_stall_minutes = 0\n")
            .expect("explicit watcher disabled");
        start_services(home.path(), repo.path(), None, Some(&explicit))
            .await
            .expect("idle service admission");
        let jobs = home.path().join("workers").join("jobs");
        assert!(
            std::fs::read_dir(jobs)
                .map(|mut entries| entries.next().is_none())
                .unwrap_or(true),
            "serve --config must not launch the repository-configured watcher"
        );
    }

    #[tokio::test]
    async fn publication_waiting_drainer_does_not_block_busy_queue_admission() {
        let (_home, talks, mut talk) = talks();
        let ingress = publication_guard(&talks, &talk.id)
            .await
            .expect("ingress publication");
        let turn = talks
            .claim_turn(&talk.id)
            .expect("turn claim")
            .expect("drainer turn");
        let id = talk.id.clone();
        let waiting = publication_guard(&talks, &id);
        tokio::pin!(waiting);
        std::future::poll_fn(|cx| {
            assert!(
                waiting.as_mut().poll(cx).is_pending(),
                "worker waits for publication"
            );
            std::task::Poll::Ready(())
        })
        .await;
        assert!(
            talks
                .claim_turn(&talk.id)
                .expect("nonblocking ingress claim")
                .is_none()
        );
        talk::queue(&mut talk, &talks, "accepted while busy", Vec::new())
            .expect("queue without owning turn");
        assert_eq!(
            talks.get(&talk.id).expect("durable pending").pending,
            "accepted while busy"
        );
        drop(ingress);
        let publication = tokio::time::timeout(std::time::Duration::from_secs(5), waiting)
            .await
            .expect("publication fence released")
            .expect("drainer publication");
        assert_eq!(
            talk::drain(&mut talk, &talks).expect("drain").as_deref(),
            Some("accepted while busy")
        );
        drop(publication);
        assert!(
            try_publication_guard(&talks, &talk.id)
                .expect("released fence")
                .is_some()
        );
        drop(turn);
    }

    #[tokio::test]
    async fn a_pinning_failure_restores_the_recorded_send_for_manual_resume() {
        let (_home, talks, mut talk) = talks();
        let _publication = publication_guard(&talks, &talk.id)
            .await
            .expect("publication");
        let _turn = talks
            .claim_turn(&talk.id)
            .expect("turn")
            .expect("free turn");
        talk::record(&mut talk, &talks, "keep this accepted send", Vec::new()).expect("record");
        let workers = talks.home().join("workers");
        std::fs::create_dir_all(&workers).expect("worker directory");
        std::fs::write(workers.join("generations"), "block executable publication")
            .expect("controlled pre-reservation failure");
        let error = spawn_recorded_turn(&talks, &talk.id, talk.turns.len() - 1)
            .await
            .expect_err("pinning must fail");
        assert!(crate::worker::not_started(&error));
        assert!(error.to_string().contains("saved as a draft"));
        let mut recovered = talks.get(&talk.id).expect("recovered draft");
        assert!(
            recovered.turns.is_empty(),
            "no unanswered transcript orphan"
        );
        assert_eq!(recovered.pending, "keep this accepted send");
        assert!(
            crate::chat_notify::talks_to_start(&talks).is_empty(),
            "recovery does not become an automatic retry"
        );
        assert_eq!(
            talk::drain(&mut recovered, &talks)
                .expect("manual resume")
                .as_deref(),
            Some("keep this accepted send")
        );
    }

    #[test]
    fn restoring_an_unstarted_send_retains_reply_attachments_and_later_drafts() {
        let (_home, talks, mut talk) = talks();
        talk::post(&talks, &talk.id, "original reply source", "source-task").expect("source");
        let attachment = talk::Attachment {
            id: "image".to_owned(),
            name: "photo.png".to_owned(),
            mime: "image/png".to_owned(),
            bytes: 12,
        };
        talk::record_reply(
            &mut talk,
            &talks,
            "  preserve formatting\n\nowner send",
            vec![attachment.clone()],
            Some(talk::ReplyTarget {
                turn: 0,
                message: 0,
            }),
        )
        .expect("record reply");
        let index = talk.turns.len() - 1;
        let recorded = talk.turns[index].clone();
        talk::post(&talks, &talk.id, "unrelated concurrent post", "later-task").expect("post");
        talk::queue(&mut talk, &talks, "later draft", Vec::new()).expect("later draft");
        assert!(talk::restore_recorded_draft(&talks, &talk.id, index, &recorded).expect("restore"));
        let mut recovered = talks.get(&talk.id).expect("recovered");
        assert_eq!(recovered.turns.len(), 2, "task posts are retained");
        assert_eq!(recovered.pending_messages[0], recorded.messages[0]);
        assert_eq!(recovered.pending_messages[1].body, "later draft");
        assert_eq!(recovered.pending_attachments, vec![attachment]);
        assert_eq!(
            recovered.pending,
            "  preserve formatting\n\nowner send\n\nlater draft"
        );
        assert!(
            !talk::restore_recorded_draft(&talks, &talk.id, index, &recorded).expect("idempotent")
        );
        talk::drain(&mut recovered, &talks)
            .expect("manual resume")
            .expect("draft");
        assert_eq!(
            recovered.turns.last().expect("resumed").messages[0],
            recorded.messages[0]
        );
    }

    #[test]
    fn a_started_response_is_never_rolled_back_into_a_draft() {
        let (_home, talks, mut talk) = talks();
        talk::record(&mut talk, &talks, "owner send", Vec::new()).expect("record");
        let recorded = talk.turns[0].clone();
        let mut response = recorded.clone();
        response.who = talk::Who::Agent;
        response.sender = None;
        response.body = "already answered".to_owned();
        talk.turns.push(response);
        talks.put(&mut talk).expect("response");
        assert!(
            !talk::restore_recorded_draft(&talks, &talk.id, 0, &recorded).expect("refuse rollback")
        );
        let fresh = talks.get(&talk.id).expect("unchanged");
        assert_eq!(fresh.turns.len(), 2);
        assert!(fresh.pending.is_empty());
    }

    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,
                max_turn_secs: 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(),
        })
    }

    #[test]
    fn investigation_holder_hint_is_advisory_and_requires_a_related_running_task() {
        let mut q = question("implement");
        let mut related = task(Source::Human);
        let check = |q: &Question, tasks: &[Task]| investigation_holder_hint(q, tasks);
        let hint = check(&q, std::slice::from_ref(&related));
        assert!(hint.contains("Lock owner unknown"));
        assert!(hint.contains("may hold"));
        assert!(hint.contains(&format!("magi task show {}", related.id)));
        assert_eq!(check(&q, &[]), "Lock owner unknown. ");
        q.run = "unrelated".into();
        assert_eq!(check(&q, std::slice::from_ref(&related)), check(&q, &[]));
        q.run = related.id.clone();
        assert!(check(&q, std::slice::from_ref(&related)).contains("may hold"));
        related.hold_manual(None);
        assert_eq!(check(&q, std::slice::from_ref(&related)), check(&q, &[]));
        related.status = crate::queue::TaskStatus::Running;
        q.run.clear();
        assert_eq!(check(&q, &[related]), check(&q, &[]));
    }

    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 batch_instruction_normalizes_repeated_japanese_prefixes() {
        for instruction in [
            "おっけーそれぞれ回答してー",
            "了解です、ありがとう！では、それぞれ回答してください。",
            "ありがとうございます！はい、じゃあそれぞれ回答して〜",
            "オッケー〜！おけ、それぞれ回答して",
            "OK！それぞれ回答して",
            "answer all as recommended!",
        ] {
            assert!(batch_instruction(instruction), "{instruction}");
        }
        for instruction in [
            "了解です",
            "ありがとう！それぞれ回答してもいい？",
            "もし大丈夫ならそれぞれ回答して",
            "それぞれ回答して、いや待って",
            "「それぞれ回答して」",
        ] {
            assert!(!batch_instruction(instruction), "{instruction}");
        }
    }

    #[test]
    fn recommendation_returns_only_one_exact_offered_choice() {
        let tmp = tempfile::tempdir().unwrap();
        let questions = Questions::at(tmp.path().join("questions"));
        let mut q = question("implement");
        questions.put(&mut q).unwrap();
        let table = |rows: &str| format!("| Question | Choice |\n| --- | --- |\n{rows}");
        let row = format!("| {} | SQLite |", q.id);
        assert_eq!(
            recommended_choice(&table(&row), &q, &questions),
            Some("SQLite")
        );
        assert_eq!(
            recommended_choice(&format!("{}: Redis", q.id), &q, &questions),
            Some("Redis")
        );
        for body in [
            table(&format!("{row}\n| {} | Redis |", q.id)),
            table(&format!("{row}\n{row}")),
            table(&format!("| {} | SQLite | Redis |", q.id)),
            table(&format!("| {} | SQLite maybe |", q.id)),
            format!("> {}: SQLite", q.id),
            format!("If approved:\n\n{}: SQLite", q.id),
            "No recommendation".into(),
        ] {
            assert_eq!(recommended_choice(&body, &q, &questions), None, "{body}");
        }
    }

    #[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,
                max_turn_secs: 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.")
            // Keep the historical closer; current delivery has a different policy.
            .replace("additional evidenced task operations follow the protocol above. Do not edit the repository.", "make: do not edit the repository.")
    }

    #[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());
    }
}