mx 0.1.229

use anyhow::{Result, bail};
use std::collections::HashMap;

use crate::engage::{MatchResult, fuzzy_match};
use crate::knowledge::KnowledgeEntry;
use crate::store::{AgentContext, KnowledgeStore, WakeCascade};
use crate::wake_chunk::{
    ChunkPlan, PhraseMatch, chunk_threshold, compare_phrase, compute_chunks, extract_auto_phrase,
    salient_candidates,
};
use crate::wake_guess::{Bucket, MatchKind, WakeGuessRow, content_hash};
use crate::wake_token::*;

/// The phrases a chunk's guess is matched against, and where they came from.
///
/// On chunk 0 an `Authored` set is every authored phrase on the bloom, not
/// just one of them: a single-chunk bloom with three phrases used to be
/// matched against phrase 0 only, which made phrases 1 and 2 unreachable
/// (#450). Every other set is one generated phrase, except a later chunk
/// with nothing to phrase from or whose every candidate repeated an earlier
/// reveal: its set is empty and its source is `Unphrased`.
struct ChunkPhrases {
    phrases: Vec<String>,
    source: PhraseSource,
}

/// Pick the phrase set for chunk `chunk_idx` of a bloom whose whole content
/// is `content`, chunked by `plan`.
///
/// Chunk 0 is the only chunk with authored phrases: it gets all of them, or
/// one `auto` phrase when the bloom has none. Each later chunk gets one phrase
/// from its own body text, labelled `derived` when the bloom has authored
/// phrases and `auto` when it has none. The reveal of a chunk shows its
/// phrases, so a later chunk matched against them would be matched against
/// its own answer key (#469). A later chunk walks its body sentences in order
/// and takes the first that matches nothing an earlier chunk's reveal showed,
/// capped or whole, which is why every earlier chunk's set is rebuilt here.
/// When the chunk has no candidate (no body text, heading or line, e.g. all
/// code) or every candidate collides, the set is empty and the source is
/// `Unphrased`: the chunk is still prompted and guessed, and any guess on it
/// is revealed.
fn phrases_for_chunk(
    entry: &KnowledgeEntry,
    plan: &ChunkPlan,
    content: &str,
    chunk_idx: u16,
) -> ChunkPhrases {
    let authored = authored_phrases(entry);
    let has_authored = !authored.is_empty();
    let first = if has_authored {
        authored
    } else {
        vec![extract_auto_phrase(
            chunk_text(plan, content, 0),
            &entry.title,
        )]
    };
    if chunk_idx == 0 {
        return ChunkPhrases {
            phrases: first,
            source: if has_authored {
                PhraseSource::Authored
            } else {
                PhraseSource::Auto
            },
        };
    }

    let mut shown_before = first;
    shown_before.push(entry.title.clone());
    let mut phrases: Vec<String> = Vec::new();
    for k in 1..=chunk_idx {
        let text = chunk_text(plan, content, k);
        phrases = salient_candidates(text)
            .into_iter()
            .find(|c| {
                best_match(&c.phrase, &shown_before).0 == MatchKind::None
                    && best_match(&c.text, &shown_before).0 == MatchKind::None
            })
            .map(|c| c.phrase)
            .into_iter()
            .collect();
        shown_before.extend(phrases.iter().cloned());
    }
    let source = if phrases.is_empty() {
        PhraseSource::Unphrased
    } else if has_authored {
        PhraseSource::Derived
    } else {
        PhraseSource::Auto
    };
    ChunkPhrases { phrases, source }
}

/// Compare a guess against every phrase in the set and keep the best result.
/// `exact` beats `close`; among equals the earliest phrase wins.
///
/// `exact` means identical after trimming and nothing else. Anything that
/// needed normalizing or fuzzing to match is `close`, whichever matcher found
/// it — the fuzzy matcher strips punctuation, so letting its exact through
/// would rank a guess wrapped in quotes above the same guess bare.
fn best_match(guess: &str, phrases: &[String]) -> (MatchKind, Option<usize>) {
    let mut best: Option<usize> = None;
    for (idx, phrase) in phrases.iter().enumerate() {
        let kind = match compare_phrase(guess, phrase) {
            PhraseMatch::Exact => MatchKind::Exact,
            PhraseMatch::Tolerant => MatchKind::Close,
            PhraseMatch::Mismatch => match fuzzy_match(guess, phrase) {
                MatchResult::Exact | MatchResult::Close => MatchKind::Close,
                MatchResult::Partial | MatchResult::Wrong => MatchKind::None,
            },
        };
        match kind {
            MatchKind::Exact => return (MatchKind::Exact, Some(idx)),
            MatchKind::Close if best.is_none() => best = Some(idx),
            _ => {}
        }
    }
    match best {
        Some(idx) => (MatchKind::Close, Some(idx)),
        None => (MatchKind::None, None),
    }
}

/// Compute the sum of chunk counts across all blooms in the session, using
/// the current in-memory content.
fn total_chunks_across_cascade(
    session: &WakeSession,
    blooms: &HashMap<String, KnowledgeEntry>,
) -> usize {
    let threshold = chunk_threshold();
    let mut total: usize = 0;
    for id in &session.bloom_ids {
        if let Some(entry) = blooms.get(id) {
            let content = bloom_content(entry);
            let plan = compute_chunks(&content, threshold);
            total += plan.total as usize;
        } else {
            total += 1; // fallback — treat missing blooms as 1-chunk
        }
    }
    total
}

/// Formatted body or summary or placeholder — the string used for chunking.
fn bloom_content(entry: &KnowledgeEntry) -> String {
    entry
        .body
        .clone()
        .or_else(|| entry.summary.clone())
        .unwrap_or_else(|| "(no content)".to_string())
}

/// The title as the responder sees it: decorated with `(Part N/M)` for a
/// chunked bloom, plain otherwise.
fn title_for_chunk(entry: &KnowledgeEntry, chunk_idx: u16, plan: &ChunkPlan) -> String {
    if plan.total > 1 {
        format!("{} (Part {}/{})", entry.title, chunk_idx + 1, plan.total)
    } else {
        entry.title.clone()
    }
}

fn chunk_ref(chunk_idx: u16, plan: &ChunkPlan) -> Option<ChunkRef> {
    if plan.total > 1 {
        Some(ChunkRef {
            index: chunk_idx + 1,
            total: plan.total,
            oversized: if plan.is_oversized(chunk_idx) {
                Some(true)
            } else {
                None
            },
        })
    } else {
        None
    }
}

/// The text of the chunk at `chunk_idx` under `plan`.
fn chunk_text<'a>(plan: &ChunkPlan, content: &'a str, chunk_idx: u16) -> &'a str {
    if plan.total > 1 {
        plan.chunk(content, chunk_idx)
    } else {
        content
    }
}

/// Build the prompt for one chunk: the title, and nothing that would answer it.
fn build_prompt_for_chunk(
    entry: &KnowledgeEntry,
    chunk_idx: u16,
    plan: &ChunkPlan,
    content: &str,
) -> BloomPrompt {
    let resolved = phrases_for_chunk(entry, plan, content, chunk_idx);
    BloomPrompt {
        id: entry.id.clone(),
        title: title_for_chunk(entry, chunk_idx, plan),
        phrase_source: resolved.source.as_str().to_string(),
        chunk: chunk_ref(chunk_idx, plan),
    }
}

fn reveal_from_row(row: &WakeGuessRow, bloom: &KnowledgeEntry) -> BloomFull {
    let content = bloom_content(bloom);
    let plan = compute_chunks(&content, chunk_threshold());
    let chunk = (row.chunk_total > 1).then(|| ChunkRef {
        index: row.chunk_index + 1,
        total: row.chunk_total,
        oversized: (plan.total == row.chunk_total && plan.is_oversized(row.chunk_index))
            .then_some(true),
    });
    BloomFull {
        id: row.bloom_id.clone(),
        title: row.title_shown.clone(),
        phrases: row.phrases.clone(),
        phrase_source: row.phrase_source.clone(),
        content: chunk_text(&plan, &content, row.chunk_index).to_string(),
        chunk,
    }
}

/// Build the full reveal for one chunk. `content` is the *chunk's* content,
/// not the whole bloom.
fn build_full_for_chunk(
    entry: &KnowledgeEntry,
    chunk_idx: u16,
    plan: &ChunkPlan,
    content: &str,
    phrases: Vec<String>,
    source: PhraseSource,
) -> BloomFull {
    BloomFull {
        id: entry.id.clone(),
        title: title_for_chunk(entry, chunk_idx, plan),
        phrases,
        phrase_source: source.as_str().to_string(),
        content: chunk_text(plan, content, chunk_idx).to_string(),
        chunk: chunk_ref(chunk_idx, plan),
    }
}

/// The longest guess the log accepts, in CHARACTERS.
///
/// A guess is model output written verbatim into the database, and nothing
/// else on the path bounds it. Characters rather than bytes so the same guess
/// is judged the same way whatever it is written in, and a refusal rather than
/// a trim so a caller is never told its guess was logged when only part of it
/// was. Nothing on this path slices by byte index.
const MAX_GUESS_CHARS: usize = 2_000;

/// Reject a guess that is not usable data, before anything is written or
/// advanced, so the caller can simply guess again.
fn validate_guess(guess: &str) -> Result<()> {
    let chars = guess.chars().count();
    if chars > MAX_GUESS_CHARS {
        bail!(
            "Guess is {} characters; the limit is {}. Send a shorter guess.",
            chars,
            MAX_GUESS_CHARS
        );
    }
    // Matching strips every non-alphanumeric character, so a guess with none
    // is indistinguishable from an empty one — and would compare equal to any
    // phrase that also strips to nothing.
    if !guess.chars().any(|c| c.is_alphanumeric()) {
        bail!("Guess is empty. Send what the title brought to mind, then continue.");
    }
    Ok(())
}

/// Identity of the ritual run, for the session and every guess row it writes.
pub struct RitualMeta {
    pub agent: String,
    /// Wake number from `--wake`. mx does not read a counter of its own.
    pub wake: Option<i64>,
    /// Model identifier from `--model`. mx has no way to discover it.
    pub model_id: Option<String>,
}

/// Check `--wake` against the guess log before a session is opened.
///
/// The number is fixed at `--begin` for every row the ritual writes and cannot
/// be repaired afterwards, so a wrong one has to be caught here. A number
/// another session already logged rows under is refused unless `force_wake`:
/// a second ritual under it would mix first guesses with re-guesses. A number
/// below the agent's highest logged wake is allowed, with a warning.
fn check_wake_number(
    db: &dyn KnowledgeStore,
    meta: &RitualMeta,
    force_wake: bool,
) -> Result<Vec<String>> {
    let Some(wake) = meta.wake else {
        return Ok(Vec::new());
    };
    if wake <= 0 {
        bail!("--wake must be a positive wake number; got {}.", wake);
    }

    let history = db.wake_history(&meta.agent, wake)?;
    let mut warnings = Vec::new();
    let earlier = history.sessions_at_wake.len();
    if earlier > 0 {
        if !force_wake {
            bail!(
                "Wake {wake} already has guesses logged by {earlier} earlier ritual session(s) \
                 of {agent}. A second ritual under the same number mixes first guesses with \
                 re-guesses. Check the wake number, or pass --force-wake to log this ritual \
                 under wake {wake} anyway.",
                agent = meta.agent
            );
        }
        warnings.push(format!(
            "Wake {wake} already has guesses from {earlier} earlier ritual session(s); \
             logging this one under it too (--force-wake)."
        ));
    }
    if let Some(highest) = history.highest_wake {
        if wake < highest {
            warnings.push(format!(
                "Wake {wake} is lower than wake {highest}, the highest this agent has logged. \
                 If that is a typo, start again with the right --wake before answering."
            ));
        } else if wake > highest.saturating_add(1) {
            // Caught now, because once a typo like 4640 is logged it becomes
            // the highest, and every correct wake after it warns as "lower".
            warnings.push(format!(
                "Wake {wake} skips ahead of wake {highest}, the highest this agent has \
                 logged. If that is a typo, start again with the right --wake before \
                 answering."
            ));
        }
    }
    Ok(warnings)
}

/// Start a new wake ritual session.
pub fn begin_ritual(
    db: &dyn KnowledgeStore,
    cascade: &WakeCascade,
    meta: RitualMeta,
    force_wake: bool,
) -> Result<String> {
    let warnings = check_wake_number(db, &meta, force_wake)?;

    if cascade.core.is_empty() && cascade.recent.is_empty() && cascade.bridges.is_empty() {
        if !cascade.excluded.is_empty() {
            // The entries are in the graph; a tag is keeping them out. Say so,
            // and name the way back in — otherwise the only signal is an empty
            // wake set that looks like data loss.
            let counts: Vec<String> = cascade
                .excluded
                .iter()
                .map(|(tag, n)| format!("{}: {}", tag, n))
                .collect();
            bail!(
                "No blooms to wake: every entry that qualified was excluded by tag ({}). \
                 Pass --include-excluded to wake them anyway, or remove the tag.",
                counts.join(", ")
            );
        }
        bail!("No blooms to wake");
    }

    let mut session = WakeSession::new(cascade, meta.agent, meta.wake, meta.model_id);

    // Build lookup map from the cascade we already have.
    let owned_blooms: HashMap<String, KnowledgeEntry> = build_bloom_map_owned(cascade);

    // Eager total-chunks count for progress.total.
    let total_steps = total_chunks_across_cascade(&session, &owned_blooms);

    // Get first bloom + its chunk plan.
    let first_id = session
        .current_bloom_id()
        .ok_or_else(|| anyhow::anyhow!("No blooms in session"))?;
    let first_bloom = owned_blooms
        .get(first_id)
        .ok_or_else(|| anyhow::anyhow!("Bloom not found: {}", first_id))?;
    let first_content = bloom_content(first_bloom);
    let first_plan = compute_chunks(&first_content, chunk_threshold());

    let prompt = build_prompt_for_chunk(first_bloom, 0, &first_plan, &first_content);
    session.prompt = Some(prompt.clone());

    // Persist session to DB.
    let session_id = db.create_wake_session(&session)?;

    // Return signed token at step 0.
    let token = create_token(&session_id, session.step);

    let response = WakeBeginResponse {
        status: "ritual_started".to_string(),
        session: token,
        prompt,
        progress: Progress {
            current: 1,
            total: total_steps.max(1),
            bloom_current: 1,
            bloom_total: session.total_blooms(),
            buckets: None,
        },
        excluded: cascade.excluded.clone(),
        wake: session.wake,
        model: session.model_id.clone(),
        warnings,
    };

    Ok(serde_json::to_string(&response)?)
}

fn rejection(message: String, expected_id: Option<String>) -> anyhow::Error {
    WakeRejection(WakeErrorResponse {
        status: "error".to_string(),
        error: "invalid_bloom_id".to_string(),
        message,
        expected_id,
    })
    .into()
}

/// Judge one guess and show the bloom.
///
/// The guess is matched against the chunk's phrases, and the row and the
/// session advance are written in one transaction. A failed write fails the
/// call and leaves the session where it was — the guess is the data the ritual
/// exists to collect, so it is not best-effort.
///
/// A step that is already logged is answered from the log, not judged again:
/// a retry after a lost response, a second submit on one token, or a row an
/// older binary wrote without advancing the session. The first guess counts.
pub fn respond_ritual(
    db: &dyn KnowledgeStore,
    ctx: &AgentContext,
    bloom_id: &str,
    guess: &str,
    token_str: &str,
) -> Result<String> {
    let (session_id, token_step) =
        verify_token(token_str).map_err(|e| anyhow::anyhow!("Token verification failed: {}", e))?;

    let mut session = db.get_wake_session(&session_id)?.ok_or_else(|| {
        anyhow::anyhow!(
            "Session not found: {}. Run `mx memory wake --begin` to start a ritual.",
            session_id
        )
    })?;

    // `--begin` always stamps the calling agent, so a session with no agent is
    // one an older binary wrote. Walking it would file every guess under an
    // empty agent, unreachable from a log that is keyed by agent.
    if session.agent.is_empty() {
        bail!(
            "This session was created by an older version of mx and cannot be continued. \
             Run `mx memory wake --begin` to start a new ritual."
        );
    }

    // The token authorises the session, not the holder. The caller must BE the
    // agent that began it — a caller naming no agent fails to establish that
    // just as surely as one naming a different agent, so both are refused.
    //
    // The blooms are fetched with the caller's context further down, and an
    // entry that context cannot see is indistinguishable from one that was
    // deleted. Requiring the owner removes the way a caller could arrive here
    // with less visibility than the ritual was built with. It does NOT remove
    // the conflation: the owner's own visibility is not fixed for the life of
    // a ritual — another agent's public entry can be made private while this
    // one is open — and the owner then steps over it as vanished while it is
    // still in the database. Telling those two apart is open work.
    if ctx.agent_id.as_deref() != Some(session.agent.as_str()) {
        bail!(
            "This ritual was begun by another agent. Run `mx memory wake --begin` \
             to start your own."
        );
    }

    // Every write this binary makes that leaves a prompt outstanding records
    // it. Without the record there is nothing to log the guess's title from,
    // and rebuilding it from the entry as it is now is the bug this guards.
    if !session.is_complete() && session.prompt.is_none() {
        bail!(
            "This session was created by an older version of mx and cannot be continued. \
             Run `mx memory wake --begin` to start a new ritual."
        );
    }

    // The last successful write spent this token, and its response never
    // reached the caller. Answer it again with the token the caller never
    // received. Keyed on the step that write STARTED from, not on `step - 1`:
    // one call can move `step` further than one, by judging a guess and then
    // stepping over deleted entries.
    if session.prev_step == Some(token_step) && token_step < session.step {
        let all_blooms = fetch_blooms_by_ids(db, ctx, &session.bloom_ids)?;
        return resume(db, session, &session_id, token_step, &all_blooms, bloom_id);
    }

    // Anti-replay: token step must match server-side state. Every respond
    // advances the step, so a spent token lands here — and since the last
    // spent token is resumed above, one that lands here really is stale.
    if session.step != token_step {
        bail!(
            "Token out of sync: token step {} but session at step {}. Retry with the \
             token from your last successful response.",
            token_step,
            session.step
        );
    }

    if session.is_complete() {
        bail!(
            "This ritual is already complete; its summary was in the final response. \
             Run `mx memory wake --begin` to start a new one."
        );
    }

    // Entries can be deleted while a ritual is open. Fetch what is still
    // there rather than failing on the first id that is gone: an entry the
    // ritual already walked past must not brick the rest of it.
    let all_blooms = fetch_blooms_by_ids(db, ctx, &session.bloom_ids)?;

    // Entries can be deleted between one call and the next, including the one
    // the caller was just handed as `next`. Step over whatever has gone before
    // deciding what is being asked about, so a deletion mid-ritual cannot
    // strand the session on an entry that no longer exists.
    //
    // Before `validate_guess`, deliberately: a caller answering about an entry
    // that no longer exists should be told the entry is gone, not that its
    // guess was unusable. The second answer does not describe the situation,
    // and a client that retries on a guess error would retry forever.
    //
    // At every step the sweep stands on, a logged row wins over any other
    // answer: an older binary wrote the row and the session in two calls, and
    // when the second failed the session was left behind its log. Finish that
    // advance with the logged judgment — even when the entry it judged has
    // been deleted since, which is why the lookup is here and not later.
    let mut vanished: Vec<String> = Vec::new();
    loop {
        if let Some(logged) = db.get_wake_guess(&session_id, session.step)? {
            return replay(
                db,
                session,
                &session_id,
                token_step,
                &all_blooms,
                logged,
                bloom_id,
            );
        }
        if session.is_complete() || !session.current_bloom_is_missing(&all_blooms) {
            break;
        }
        if let Some(id) = session.current_bloom_id() {
            vanished.push(id.to_string());
        }
        session.advance_unjudged();
    }

    // The prompt the caller was handed. Every incomplete session carries one
    // past the guard above.
    let prompted = session.prompt.clone().ok_or_else(|| {
        anyhow::anyhow!(
            "Wake session {} has no prompt awaiting a response",
            session_id
        )
    })?;

    // The entry the caller was handed has gone. Only a call naming it is
    // answered as `bloom_missing`: it is following the sequence it was given.
    // Any other id was never handed out — even one that has also gone — and
    // answering it as vanished would consume the prompted entry on the way
    // past, or complete the session and make the mistake uncorrectable.
    if vanished.contains(&prompted.id) {
        if bloom_id != prompted.id {
            return Err(rejection(
                format!(
                    "The entry awaiting a response is {}, which has since been deleted; \
                     respond about it to be handed the next one. Got {}",
                    prompted.id, bloom_id
                ),
                Some(prompted.id),
            ));
        }
        return Ok(serde_json::to_string(&unjudged_response(
            db,
            &mut session,
            &session_id,
            token_step,
            &all_blooms,
            "bloom_missing",
            None,
        )?)?);
    }

    // The prompted entry is still there, so the sweep stepped over nothing
    // after it and the cursor stands on it — unless the prompt record was left
    // behind by an older binary.
    if session.current_bloom_id() != Some(prompted.id.as_str())
        || prompted.chunk_index() != session.current_chunk_index
    {
        bail!(
            "This session was last advanced by an older version of mx and its prompt record \
             is out of date. Run `mx memory wake --begin` to start a new ritual."
        );
    }
    let expected_id = prompted.id.clone();

    if bloom_id != expected_id {
        return Err(rejection(
            format!("Expected bloom {}, got {}", expected_id, bloom_id),
            Some(expected_id),
        ));
    }

    validate_guess(guess)?;

    let bloom = all_blooms
        .get(&expected_id)
        .ok_or_else(|| anyhow::anyhow!("Bloom not found after the missing-entry sweep"))?;

    let content = bloom_content(bloom);
    let plan = compute_chunks(&content, chunk_threshold());

    // If the bloom shrank past our chunk cursor, advance to the next bloom.
    // No guess was judged, so no row is written — but the step still ticks, so
    // the token the caller just spent stops verifying.
    if session.clamp_if_chunks_shrank(plan.total) {
        let resolved = phrases_for_chunk(bloom, &plan, &content, 0);
        let shown =
            build_full_for_chunk(bloom, 0, &plan, &content, resolved.phrases, resolved.source);
        return Ok(serde_json::to_string(&unjudged_response(
            db,
            &mut session,
            &session_id,
            token_step,
            &all_blooms,
            "chunk_truncated",
            Some(shown),
        )?)?);
    }

    let chunk_idx = session.current_chunk_index;
    let chunk_content = chunk_text(&plan, &content, chunk_idx);
    let resolved = phrases_for_chunk(bloom, &plan, &content, chunk_idx);
    // The row and the session's bucket split carry the source the prompt
    // announced, as a replay of this row will.
    let prompted_source = PhraseSource::parse(&prompted.phrase_source).ok_or_else(|| {
        anyhow::anyhow!(
            "Wake session {} prompted with unknown phrase source {}",
            session_id,
            prompted.phrase_source
        )
    })?;

    // A step announced unphrased showed no phrase, so nothing can match it,
    // even if the entry's phrases were edited before the guess landed.
    let phrases = if prompted_source == PhraseSource::Unphrased {
        Vec::new()
    } else {
        resolved.phrases
    };
    let (match_kind, match_index) = best_match(guess, &phrases);
    let bucket = match match_kind {
        MatchKind::None => Bucket::Revealed,
        _ => Bucket::Unhinted,
    };

    let row = WakeGuessRow {
        agent: session.agent.clone(),
        wake: session.wake,
        session_id: session_id.clone(),
        bloom_id: expected_id.clone(),
        chunk_index: chunk_idx,
        chunk_total: prompted.chunk_total(),
        position: session.step,
        bloom_position: session.current_index + 1,
        bloom_total: session.total_blooms(),
        title_shown: prompted.title,
        guess: guess.to_string(),
        model_id: session.model_id.clone(),
        phrase_source: prompted.phrase_source,
        phrases,
        match_kind: match_kind.as_str().to_string(),
        match_index,
        bucket: bucket.as_str().to_string(),
        content_hash: content_hash(chunk_content),
    };

    // The reveal names the step the way the row does, so this response and
    // any replay of it say the same thing.
    let shown = reveal_from_row(&row, bloom);

    let before_advance = session.clone();
    session.last_status = Some("shown".to_string());
    session.advance(plan.total, bucket, prompted_source);
    skip_missing_blooms(&mut session, &all_blooms);
    issue_next_prompt(&mut session, &all_blooms)?;

    if let Err(err) = db.record_wake_guess(&row, &session, token_step) {
        // Another call on the same token got there first: its guess is logged
        // and it is the one that counts. Anything else is a real failure.
        let Some(logged) = db.get_wake_guess(&session_id, row.position)? else {
            return Err(err);
        };
        let base = match db.get_wake_session(&session_id)? {
            Some(stored) if stored.step > logged.position => stored,
            _ => before_advance,
        };
        return replay(
            db,
            base,
            &session_id,
            token_step,
            &all_blooms,
            logged,
            bloom_id,
        );
    }

    let (progress, summary) = progress_and_summary(&session, &all_blooms);

    let response = WakeRespondResponse {
        status: "shown".to_string(),
        replayed: false,
        bucket: Some(bucket.as_str().to_string()),
        guess: Some(guess.to_string()),
        match_info: Some(MatchInfo {
            kind: match_kind.as_str().to_string(),
            phrase_index: match_index,
        }),
        bloom: Some(shown),
        session: create_token(&session_id, session.step),
        next: session.prompt.clone(),
        progress: Some(progress),
        summary,
        wake: session.wake,
        model: session.model_id.clone(),
    };

    Ok(serde_json::to_string(&response)?)
}

/// Answer again the last successful call, whose response was lost.
///
/// If that call judged a guess, its row sits somewhere in `[token_step,
/// step)` — after any entries it stepped over first — and is replayed. If it
/// judged nothing (`bloom_missing`, `chunk_truncated`), its recorded status is
/// echoed with the current token and `next` or summary; there is no row and
/// no `bloom` to show.
fn resume(
    db: &dyn KnowledgeStore,
    session: WakeSession,
    session_id: &str,
    token_step: u32,
    all_blooms: &HashMap<String, KnowledgeEntry>,
    bloom_id: &str,
) -> Result<String> {
    for position in token_step..session.step {
        if let Some(logged) = db.get_wake_guess(session_id, position)? {
            return replay(
                db, session, session_id, token_step, all_blooms, logged, bloom_id,
            );
        }
    }

    let status = session.last_status.clone().ok_or_else(|| {
        anyhow::anyhow!("Wake session {} has no record of its last step", session_id)
    })?;
    let (progress, summary) = progress_and_summary(&session, all_blooms);
    let response = WakeRespondResponse {
        status,
        replayed: true,
        bucket: None,
        guess: None,
        match_info: None,
        bloom: None,
        session: create_token(session_id, session.step),
        next: if session.is_complete() {
            None
        } else {
            session.prompt.clone()
        },
        progress: Some(progress),
        summary,
        wake: session.wake,
        model: session.model_id.clone(),
    };
    Ok(serde_json::to_string(&response)?)
}

/// Answer a step from its logged row instead of judging the guess just sent.
///
/// `session` is either still AT the logged step — the row landed and the
/// session write did not — in which case it is advanced with the LOGGED
/// judgment and saved as a compare-and-swap on `expected_step`; or already
/// past it, in which case nothing is written. Either way the response carries
/// the logged guess, its judgment, and the current token and `next` — the
/// prompt the session recorded when it was handed out, not one rebuilt from
/// the entry as it is now, so the title a retry is shown is the one its guess
/// is logged under.
fn replay(
    db: &dyn KnowledgeStore,
    mut session: WakeSession,
    session_id: &str,
    expected_step: u32,
    all_blooms: &HashMap<String, KnowledgeEntry>,
    logged: WakeGuessRow,
    bloom_id: &str,
) -> Result<String> {
    if bloom_id != logged.bloom_id {
        return Err(rejection(
            format!(
                "Step {} was a guess about bloom {}, not {}",
                logged.position, logged.bloom_id, bloom_id
            ),
            Some(logged.bloom_id),
        ));
    }

    if session.step == logged.position {
        let bucket = Bucket::parse(&logged.bucket)
            .ok_or_else(|| anyhow::anyhow!("Logged guess has unknown bucket {}", logged.bucket))?;
        let source = PhraseSource::parse(&logged.phrase_source).ok_or_else(|| {
            anyhow::anyhow!(
                "Logged guess has unknown phrase source {}",
                logged.phrase_source
            )
        })?;
        session.advance(logged.chunk_total, bucket, source);
        skip_missing_blooms(&mut session, all_blooms);
        issue_next_prompt(&mut session, all_blooms)?;
        session.last_status = Some("shown".to_string());
        db.update_wake_session(&session, expected_step)?;
    }

    let (progress, summary) = progress_and_summary(&session, all_blooms);

    let shown = all_blooms
        .get(&logged.bloom_id)
        .map(|b| reveal_from_row(&logged, b));

    let response = WakeRespondResponse {
        status: "shown".to_string(),
        replayed: true,
        bucket: Some(logged.bucket),
        guess: Some(logged.guess),
        match_info: Some(MatchInfo {
            kind: logged.match_kind,
            phrase_index: logged.match_index,
        }),
        bloom: shown,
        session: create_token(session_id, session.step),
        next: if session.is_complete() {
            None
        } else {
            session.prompt.clone()
        },
        progress: Some(progress),
        summary,
        wake: session.wake,
        model: session.model_id.clone(),
    };

    Ok(serde_json::to_string(&response)?)
}

/// Finish a step where no guess was judged: persist the already-advanced
/// session, step over any blooms that have since been deleted, and build the
/// response. Shared by the truncated-chunk and deleted-entry paths, which
/// differ only in their status and in whether there is anything to show.
fn unjudged_response(
    db: &dyn KnowledgeStore,
    session: &mut WakeSession,
    session_id: &str,
    expected_step: u32,
    all_blooms: &HashMap<String, KnowledgeEntry>,
    status: &str,
    shown: Option<BloomFull>,
) -> Result<WakeRespondResponse> {
    skip_missing_blooms(session, all_blooms);

    issue_next_prompt(session, all_blooms)?;
    let (progress, summary) = progress_and_summary(session, all_blooms);
    session.last_status = Some(status.to_string());
    db.update_wake_session(session, expected_step)?;

    Ok(WakeRespondResponse {
        status: status.to_string(),
        replayed: false,
        bucket: None,
        guess: None,
        match_info: None,
        bloom: shown,
        session: create_token(session_id, session.step),
        next: session.prompt.clone(),
        progress: Some(progress),
        summary,
        wake: session.wake,
        model: session.model_id.clone(),
    })
}

/// Step the cursor over blooms that are no longer in the database, so the
/// next prompt names an entry that still exists. Each one costs a step, the
/// same as any other advance.
fn skip_missing_blooms(session: &mut WakeSession, all_blooms: &HashMap<String, KnowledgeEntry>) {
    while !session.is_complete() && session.current_bloom_is_missing(all_blooms) {
        session.advance_unjudged();
    }
}

/// Fetch the blooms of a session that still exist.
///
/// Deliberately tolerant: a bloom the ritual already walked past can be
/// deleted from another shell, and re-fetching every id on every respond used
/// to hard-fail on the first one missing — bricking a session that could
/// neither be finished nor cleared. Callers handle an absent current bloom.
fn fetch_blooms_by_ids(
    db: &dyn KnowledgeStore,
    ctx: &AgentContext,
    bloom_ids: &[String],
) -> Result<HashMap<String, KnowledgeEntry>> {
    let mut map = HashMap::new();

    for id in bloom_ids {
        if let Some(entry) = db.get(id, ctx)? {
            map.insert(id.clone(), entry);
        }
    }

    Ok(map)
}

/// Build owned lookup map of all blooms from cascade.
fn build_bloom_map_owned(cascade: &WakeCascade) -> HashMap<String, KnowledgeEntry> {
    let mut map = HashMap::new();

    for entry in &cascade.core {
        map.insert(entry.id.clone(), entry.clone());
    }
    for entry in &cascade.recent {
        map.insert(entry.id.clone(), entry.clone());
    }
    for entry in &cascade.bridges {
        map.insert(entry.id.clone(), entry.clone());
    }

    map
}

/// Build the prompt for the cursor's position and record it on the session,
/// for the write that persists the cursor to persist with it. `None` once the
/// ritual is complete. The caller must already have stepped over deleted
/// entries, so the entry under the cursor is present.
fn issue_next_prompt(
    session: &mut WakeSession,
    all_blooms: &HashMap<String, KnowledgeEntry>,
) -> Result<()> {
    session.prompt = match session.current_bloom_id() {
        None => None,
        Some(next_id) => {
            let next_bloom = all_blooms
                .get(next_id)
                .ok_or_else(|| anyhow::anyhow!("Next bloom not found: {}", next_id))?;
            let next_content = bloom_content(next_bloom);
            let next_plan = compute_chunks(&next_content, chunk_threshold());
            Some(build_prompt_for_chunk(
                next_bloom,
                session.current_chunk_index,
                &next_plan,
                &next_content,
            ))
        }
    };
    Ok(())
}

/// Current progress, and the summary once the ritual is complete.
fn progress_and_summary(
    session: &WakeSession,
    all_blooms: &HashMap<String, KnowledgeEntry>,
) -> (Progress, Option<Summary>) {
    // Re-compute total chunks for progress (cheap; keeps the total fresh for
    // mid-ritual edits).
    let total_chunks = total_chunks_across_cascade(session, all_blooms).max(1);

    // `current` is the chunk being worked on: one past the ones already
    // walked. When the ritual is complete there is no such chunk, so it is the
    // count walked — otherwise the last response of every ritual would report
    // a position past the end, next to a summary that says otherwise. Clamped
    // because a step where no guess was judged still ticks, and the recomputed
    // total need not have a chunk for it.
    let walked = session.step as usize;
    let display_current = if session.is_complete() {
        walked
    } else {
        walked + 1
    }
    .min(total_chunks);

    let bloom_current = (session.current_index + 1).min(session.total_blooms());
    let buckets = session.bucket_totals();

    let progress = Progress {
        current: display_current,
        total: total_chunks,
        bloom_current,
        bloom_total: session.total_blooms(),
        buckets: Some(buckets.totals()),
    };

    let summary = session.is_complete().then(|| Summary {
        chunks: session.step as usize,
        blooms: session.total_blooms(),
        buckets,
        unjudged: session.unjudged_count,
    });
    (progress, summary)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::wake_chunk::PhraseMatch;

    // =====================================================================
    // Fixtures. Every bloom here is invented for the test.
    // =====================================================================

    fn test_entry() -> KnowledgeEntry {
        // KnowledgeEntry has no Default; use serde_json round-trip to
        // construct a minimal valid entry (all fields have #[serde(default)]
        // except id/title/category_id).
        serde_json::from_str::<KnowledgeEntry>(
            r#"{"id":"kn-test","category_id":"bloom","title":"Test","body":"body"}"#,
        )
        .expect("test entry deserialize")
    }

    fn entry_with_phrases(phrases: Vec<&str>) -> KnowledgeEntry {
        let mut e = test_entry();
        e.wake_phrases = phrases.into_iter().map(|s| s.to_string()).collect();
        e
    }

    fn test_cascade(entries: Vec<KnowledgeEntry>) -> WakeCascade {
        WakeCascade {
            core: entries,
            ..Default::default()
        }
    }

    fn meta() -> RitualMeta {
        RitualMeta {
            agent: "test-agent".to_string(),
            wake: Some(7),
            model_id: Some("test-model".to_string()),
        }
    }

    /// A bloom big enough to split into several chunks, with H2 sections so
    /// the chunker has semantic break points.
    fn make_large_bloom(target_bytes: usize, phrases: Vec<&str>) -> KnowledgeEntry {
        let mut body = String::new();
        let mut section = 0;
        while body.len() < target_bytes {
            section += 1;
            body.push_str(&format!(
                "\n## Section {section}\n\n\
                 This is section {section} of the test bloom. It contains \
                 enough text that multiple sections will cross the chunking \
                 threshold. The wake ritual should walk each chunk in turn.\n\n\
                 - bullet one for section {section}\n\
                 - bullet two for section {section}\n\
                 - bullet three for section {section}\n\n"
            ));
        }
        let mut e = test_entry();
        e.title = "Ops".to_string();
        e.body = Some(body);
        e.wake_phrases = phrases.into_iter().map(|s| s.to_string()).collect();
        e
    }

    /// A guess that matches nothing in any fixture here.
    const MISS: &str = "zzz quibbling flugelhorn marmalade";

    fn token_from_response(json: &serde_json::Value) -> String {
        json.get("session")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string()
    }

    fn begin(store: &MockStore, cascade: &WakeCascade) -> serde_json::Value {
        serde_json::from_str(&begin_ritual(store, cascade, meta(), false).unwrap()).unwrap()
    }

    /// Respond as the agent that `meta()` began the ritual with.
    ///
    /// Deliberately NOT an anonymous context: driving respond with no agent is
    /// the one case the ownership guard lets through, and a helper that did it
    /// by default would make every test here depend on that hole staying open.
    fn respond(store: &MockStore, bloom_id: &str, guess: &str, token: &str) -> serde_json::Value {
        let ctx = AgentContext::for_agent("test-agent");
        serde_json::from_str(&respond_ritual(store, &ctx, bloom_id, guess, token).unwrap()).unwrap()
    }

    /// Respond expecting a structured refusal, and return its JSON payload.
    fn reject(store: &MockStore, bloom_id: &str, guess: &str, token: &str) -> serde_json::Value {
        let ctx = AgentContext::for_agent("test-agent");
        let err = respond_ritual(store, &ctx, bloom_id, guess, token)
            .expect_err("expected a structured refusal");
        let rejection = err
            .downcast::<WakeRejection>()
            .unwrap_or_else(|e| panic!("expected a WakeRejection, got: {e}"));
        serde_json::to_value(&rejection.0).unwrap()
    }

    // =====================================================================
    // phrases_for_chunk — which phrases a chunk is matched against
    // =====================================================================

    /// The whole content and a plan that cuts it exactly into `chunks`.
    fn planned(chunks: &[&str]) -> (String, ChunkPlan) {
        let mut content = String::new();
        let mut boundaries = Vec::new();
        for (idx, chunk) in chunks.iter().enumerate() {
            if idx > 0 {
                boundaries.push(content.len());
            }
            content.push_str(chunk);
        }
        let plan = ChunkPlan {
            total: chunks.len() as u16,
            boundaries,
            oversized: vec![false; chunks.len()],
        };
        (content, plan)
    }

    #[test]
    fn authored_chunk_is_matched_against_every_authored_phrase() {
        let e = entry_with_phrases(vec!["alpha", "beta", "gamma"]);
        let (content, plan) = planned(&[
            "Chunk zero talks about orchards.\n",
            "The cellar floods every spring.\n",
            "Lamps hang along the north wall.\n",
            "Nobody waters the fern on Sundays.\n",
            "The gate squeaks in cold weather.\n",
        ]);
        let resolved = phrases_for_chunk(&e, &plan, &content, 0);
        assert_eq!(resolved.source, PhraseSource::Authored);
        assert_eq!(resolved.phrases, vec!["alpha", "beta", "gamma"]);

        // A later chunk inside the authored count no longer reuses the list
        // chunk 0's reveal just showed (#469): it derives from its own body.
        let resolved = phrases_for_chunk(&e, &plan, &content, 2);
        assert_eq!(resolved.source, PhraseSource::Derived);
        assert_eq!(resolved.phrases, vec!["Lamps hang along the north wall."]);
        assert_ne!(resolved.phrases, vec!["alpha", "beta", "gamma"]);
    }

    #[test]
    fn chunk_beyond_authored_count_derives_one_phrase() {
        let e = entry_with_phrases(vec!["alpha"]);
        let (content, plan) = planned(&[
            "Chunk zero talks about orchards.\n",
            "The cellar floods every spring.\n",
            "Lamps hang along the north wall.\n",
            "\n## Derived heading here\n\nbody text",
            "\nThe gate squeaks in cold weather.\n",
        ]);
        let resolved = phrases_for_chunk(&e, &plan, &content, 3);
        assert_eq!(resolved.source, PhraseSource::Derived);
        assert_eq!(resolved.phrases, vec!["body text"]);
    }

    #[test]
    fn phraseless_bloom_gets_one_auto_phrase() {
        let e = entry_with_phrases(vec![]);
        let (content, plan) = planned(&[
            "## Opening\n\nFirst words of the entry.\n",
            "Middle words sit here.\n",
            "## A heading\n\nbody",
        ]);
        let resolved = phrases_for_chunk(&e, &plan, &content, 2);
        assert_eq!(resolved.source, PhraseSource::Auto);
        assert_eq!(resolved.phrases, vec!["body"]);
    }

    #[test]
    fn legacy_single_wake_phrase_is_treated_as_authored() {
        let mut e = test_entry();
        e.wake_phrase = Some("legacy phrase".to_string());
        let (content, plan) = planned(&["chunk"]);
        let resolved = phrases_for_chunk(&e, &plan, &content, 0);
        assert_eq!(resolved.source, PhraseSource::Authored);
        assert_eq!(resolved.phrases, vec!["legacy phrase"]);
    }

    /// Walk every chunk of `e` and check that no later chunk's set matches
    /// anything an earlier reveal showed, in either direction, and that a set
    /// is empty exactly when its source is `Unphrased`.
    fn assert_no_later_chunk_matches_an_earlier_reveal(
        e: &KnowledgeEntry,
        plan: &ChunkPlan,
        content: &str,
    ) {
        let mut shown_before: Vec<String> = Vec::new();
        for k in 0..plan.total {
            let resolved = phrases_for_chunk(e, plan, content, k);
            assert_eq!(
                resolved.phrases.is_empty(),
                resolved.source == PhraseSource::Unphrased,
                "chunk {k}: {:?} labelled {:?}",
                resolved.phrases,
                resolved.source
            );
            if k > 0 {
                for phrase in &resolved.phrases {
                    assert_eq!(
                        best_match(phrase, &shown_before),
                        (MatchKind::None, None),
                        "chunk {k} phrase {phrase:?} matches an earlier reveal {shown_before:?}"
                    );
                }
                for shown in &shown_before {
                    assert_eq!(
                        best_match(shown, &resolved.phrases),
                        (MatchKind::None, None),
                        "earlier reveal {shown:?} matches chunk {k}'s set {:?}",
                        resolved.phrases
                    );
                }
            }
            shown_before.extend(resolved.phrases);
        }
    }

    #[test]
    fn a_later_chunk_is_never_matched_against_an_earlier_reveal() {
        let e = make_large_bloom(95_000, vec!["alpha", "beta", "gamma"]);
        let content = bloom_content(&e);
        let plan = compute_chunks(&content, chunk_threshold());
        assert!(plan.total >= 3, "fixture must chunk; got {}", plan.total);
        assert_no_later_chunk_matches_an_earlier_reveal(&e, &plan, &content);
    }

    mod reveal_collision_props {
        use super::*;
        use proptest::prelude::*;

        const CHUNK: &str = "(## Section [0-9]\n\n)?((alpha|beta|Part [1-4]/4|This is section [0-9]) ?(of the bloom)?\\. ){0,4}";

        proptest! {
            #[test]
            fn prop_no_later_chunk_matches_an_earlier_reveal(
                authored in prop::collection::vec("(alpha|beta|This is section [0-9]\\.)", 0..3),
                chunks in prop::collection::vec(CHUNK, 2..6),
            ) {
                let e = entry_with_phrases(authored.iter().map(String::as_str).collect());
                let parts: Vec<&str> = chunks.iter().map(String::as_str).collect();
                let (content, plan) = planned(&parts);
                assert_no_later_chunk_matches_an_earlier_reveal(&e, &plan, &content);
            }
        }
    }

    #[test]
    fn a_later_chunk_guessing_an_authored_phrase_is_revealed() {
        let store = MockStore::new();
        let bloom = make_large_bloom(95_000, vec!["alpha", "beta", "gamma"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let first = respond(&store, &bloom_id, MISS, &token_from_response(&begin_json));
        assert_eq!(
            first["bloom"]["phrases"],
            serde_json::json!(["alpha", "beta", "gamma"])
        );
        assert_eq!(first["next"]["phrase_source"], "derived");

        let second = respond(&store, &bloom_id, "beta", &token_from_response(&first));
        assert_eq!(second["bucket"], "revealed");
        assert_eq!(second["match"]["kind"], "none");
        assert_eq!(second["bloom"]["phrase_source"], "derived");
        let rows = store.guesses.borrow();
        assert_eq!(rows[1].chunk_index, 1);
        assert_eq!(rows[1].phrase_source, "derived");
        assert_eq!(rows[1].bucket, "revealed");
    }

    #[test]
    fn a_derived_phrase_repeating_an_authored_phrase_takes_the_next_sentence() {
        // The author quoted chunk 2's opening sentence as a phrase, so chunk
        // 1's reveal shows it.
        let e = entry_with_phrases(vec!["The weir holds back the river.", "millstone"]);
        let chunk_two = "\n## The Weir\n\nThe weir holds back the river. It was rebuilt twice.\n";
        let (content, plan) = planned(&["The mill sits on a bend.\n", chunk_two]);

        let resolved = phrases_for_chunk(&e, &plan, &content, 1);
        assert_eq!(resolved.source, PhraseSource::Derived);
        assert_eq!(resolved.phrases, vec!["It was rebuilt twice."]);
    }

    #[test]
    fn a_chunk_whose_every_sentence_repeats_an_earlier_reveal_has_no_phrase() {
        let e = entry_with_phrases(vec![
            "The weir holds back the river.",
            "It was rebuilt twice.",
        ]);
        let chunk_two = "\n## The Weir\n\nThe weir holds back the river. It was rebuilt twice.\n";
        let (content, plan) = planned(&["The mill sits on a bend.\n", chunk_two]);

        let resolved = phrases_for_chunk(&e, &plan, &content, 1);
        assert!(resolved.phrases.is_empty(), "{:?}", resolved.phrases);
        assert_eq!(resolved.source, PhraseSource::Unphrased);
    }

    /// A bloom whose every section repeats its one authored phrase, padded
    /// with fenced filler so it chunks. Fences are never split, so every later
    /// chunk's only body sentences are that phrase.
    fn make_repetitive_bloom() -> KnowledgeEntry {
        let filler = "let pears = 4;\n".repeat(150);
        let mut body = String::new();
        let mut section = 0;
        while body.len() < 40_000 {
            section += 1;
            body.push_str(&format!(
                "\n## Section {section}\n\nNothing new here.\n\n```\n{filler}```\n\n"
            ));
        }
        let mut e = test_entry();
        e.title = "Repeats".to_string();
        e.body = Some(body);
        e.wake_phrases = vec!["Nothing new here.".to_string()];
        e
    }

    #[test]
    fn a_phraseless_chunk_reveals_any_guess() {
        let store = MockStore::new();
        let bloom = make_repetitive_bloom();
        let bloom_id = bloom.id.clone();
        let plan = compute_chunks(&bloom_content(&bloom), chunk_threshold());
        assert!(plan.total >= 2, "fixture must chunk; got {}", plan.total);
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let first = respond(&store, &bloom_id, MISS, &token_from_response(&begin_json));
        assert_eq!(first["next"]["phrase_source"], "unphrased");

        let mut resp = respond(
            &store,
            &bloom_id,
            "Nothing new here.",
            &token_from_response(&first),
        );
        assert_eq!(resp["bucket"], "revealed");
        assert_eq!(resp["match"]["kind"], "none");
        assert_eq!(resp["bloom"]["phrases"], serde_json::json!([]));
        assert_eq!(resp["bloom"]["phrase_source"], "unphrased");
        {
            let rows = store.guesses.borrow();
            assert_eq!(rows[1].chunk_index, 1);
            assert_eq!(rows[1].phrase_source, "unphrased");
            assert!(rows[1].phrases.is_empty(), "{:?}", rows[1].phrases);
        }

        for _ in 2..plan.total {
            resp = respond(&store, &bloom_id, MISS, &token_from_response(&resp));
        }
        assert_eq!(
            resp["summary"]["buckets"]["revealed"]["unphrased"],
            u64::from(plan.total - 1)
        );
    }

    /// One chunk per section: the section text, then a fenced filler block
    /// big enough that two sections never share a chunk window. Every section
    /// after the first must open with `## ` or `---`.
    fn sectioned(id: &str, title: &str, authored: &[&str], sections: &[&str]) -> KnowledgeEntry {
        let filler = "let pears = 4;\n".repeat(1300);
        let mut body = String::new();
        for s in sections {
            body.push_str(s);
            body.push_str("\n\n```\n");
            body.push_str(&filler);
            body.push_str("```\n\n");
        }
        let mut e = test_entry();
        e.id = id.to_string();
        e.title = title.to_string();
        e.body = Some(body);
        e.wake_phrases = authored.iter().map(|s| s.to_string()).collect();
        let content = bloom_content(&e);
        let plan = compute_chunks(&content, chunk_threshold());
        assert_eq!(
            plan.total as usize,
            sections.len(),
            "fixture must chunk per section"
        );
        for (k, s) in sections.iter().enumerate() {
            assert!(
                chunk_text(&plan, &content, k as u16).contains(s.trim()),
                "chunk {k} must hold its section"
            );
        }
        e
    }

    /// Walk a whole ritual, the guess for step `i` chosen by `guess_for`.
    fn run_ritual(
        blooms: &[KnowledgeEntry],
        mut guess_for: impl FnMut(usize) -> String,
    ) -> Vec<serde_json::Value> {
        let store = MockStore::new();
        for b in blooms {
            store.seed(b);
        }
        let first = begin(&store, &test_cascade(blooms.to_vec()));
        let mut prompt = first["prompt"].clone();
        let mut token = token_from_response(&first);
        let mut out = Vec::new();
        let mut step = 0;
        while prompt.is_object() {
            let id = prompt["id"].as_str().unwrap().to_string();
            let r = respond(&store, &id, &guess_for(step), &token);
            token = token_from_response(&r);
            prompt = r["next"].clone();
            out.push(r);
            step += 1;
            assert!(step < 64, "ritual never ended");
        }
        out
    }

    /// Every prompt and reveal shows the title with its `Part k/M` suffix, so
    /// a chunk with nothing to phrase from must not be phrased `Part k/M`.
    /// Chunk 0's reveal showed `Mill (Part 1/3)`; guessing its suffix on
    /// chunk 1 must not match.
    #[test]
    fn code_only_chunk_is_not_matched_by_an_earlier_title_suffix() {
        let filler = "let pears = 4;\n".repeat(1300);
        let mut e = test_entry();
        e.id = "kn-code".into();
        e.title = "Mill".into();
        e.wake_phrases = vec!["millstone".into()];
        e.body = Some(format!(
            "## Mill\n\nThe mill sits on a bend.\n\n```\n{filler}```\n\n\n```\n{filler}```\n\n## Gate\n\nThe gate squeaks.\n\n```\n{filler}```\n"
        ));
        let content = bloom_content(&e);
        let plan = compute_chunks(&content, chunk_threshold());
        assert_eq!(plan.total, 3, "fixture must chunk");
        assert!(
            !chunk_text(&plan, &content, 1).contains("##"),
            "chunk 1 is code only"
        );
        let mut failures = Vec::new();
        for g in ["Part 1/3", "Mill (Part 1/3)", "Part 2/3"] {
            let rs = run_ritual(std::slice::from_ref(&e), |i| {
                if i == 1 {
                    g.to_string()
                } else {
                    MISS.to_string()
                }
            });
            if rs[1]["bucket"] != "revealed" || rs[1]["bloom"]["phrase_source"] != "unphrased" {
                failures.push(format!(
                    "{g:?} -> {} {} vs {} ({})",
                    rs[1]["bucket"],
                    rs[1]["match"]["kind"],
                    rs[1]["bloom"]["phrases"],
                    rs[1]["bloom"]["phrase_source"]
                ));
            }
        }
        assert!(
            failures.is_empty(),
            "a code-only chunk was phrased:\n{}",
            failures.join("\n")
        );
    }

    /// Chunk 1's first sentence is the long authored phrase chunk 0's reveal
    /// just showed; capped, it would be a verbatim prefix of that reveal. The
    /// uncapped sentence is checked, so chunk 1 takes its next sentence.
    #[test]
    fn capped_phrase_is_not_a_prefix_of_an_earlier_reveal() {
        let long = "The weir holds back the river through every spring flood and every autumn storm while the millers \
                    argue about who should pay for the repairs to its crumbling stones.";
        assert!(long.chars().count() > 100, "the sentence is capped");
        let e = entry_with_phrases(vec![long]);
        let (content, plan) = planned(&[
            "The mill sits on a bend.\n",
            &format!("\n## Weir\n\n{long} The sluice was oiled in May.\n"),
        ]);
        let one = phrases_for_chunk(&e, &plan, &content, 1);
        assert_eq!(one.source, PhraseSource::Derived);
        assert_eq!(one.phrases, vec!["The sluice was oiled in May."]);
        assert_eq!(
            best_match(long, &one.phrases),
            (MatchKind::None, None),
            "copying the reveal whole is safe"
        );
        for p in &one.phrases {
            let stem = p.trim_end_matches('\u{2026}');
            assert!(
                !long.starts_with(stem),
                "chunk 1's phrase {p:?} is the first {} chars of the authored phrase chunk 0 revealed",
                stem.chars().count()
            );
        }
    }

    /// The prompt announced `unphrased`; the owner edits the phrases before
    /// the guess lands. The row keeps the announced source, so an `unphrased`
    /// row must still have no phrases and be `revealed`.
    #[test]
    fn phrase_edit_mid_ritual_never_logs_an_unhinted_unphrased_row() {
        let mut e = sectioned(
            "kn-weir",
            "Weir",
            &["The weir holds back the river.", "It was rebuilt twice."],
            &[
                "## Mill\n\nThe mill sits on a bend.",
                "## All\n\nThe weir holds back the river. It was rebuilt twice.",
            ],
        );
        let store = MockStore::new();
        store.seed(&e);
        let b = begin(&store, &test_cascade(vec![e.clone()]));
        let r0 = respond(&store, &e.id, MISS, &token_from_response(&b));
        assert_eq!(r0["next"]["phrase_source"], "unphrased");
        e.wake_phrases = vec!["millstone".into()];
        store.seed(&e);
        let r1 = respond(
            &store,
            &e.id,
            "The weir holds back the river.",
            &token_from_response(&r0),
        );
        let rows = store.guesses.borrow();
        let row = &rows[1];
        assert_eq!(row.phrase_source, "unphrased", "response {r1}");
        assert!(row.phrases.is_empty(), "phrases {:?}", row.phrases);
        assert_eq!(row.bucket, "revealed");
    }

    #[test]
    fn single_chunk_phrase_sets_are_unchanged() {
        let e = entry_with_phrases(vec!["alpha", "beta", "gamma"]);
        let (content, plan) = planned(&["One small entry about kilns.\n"]);
        let resolved = phrases_for_chunk(&e, &plan, &content, 0);
        assert_eq!(resolved.source, PhraseSource::Authored);
        assert_eq!(resolved.phrases, vec!["alpha", "beta", "gamma"]);

        // Chunk 0 of a phraseless entry keeps the auto ladder's heading tier.
        let mut bare = entry_with_phrases(vec![]);
        bare.title = "Kilns".to_string();
        let (content, plan) = planned(&["## Kiln Schedule\n\nFiring starts on Monday.\n"]);
        let resolved = phrases_for_chunk(&bare, &plan, &content, 0);
        assert_eq!(resolved.source, PhraseSource::Auto);
        assert_eq!(resolved.phrases, vec!["Kiln Schedule"]);
        assert_eq!(
            resolved.phrases,
            vec![extract_auto_phrase(&content, &bare.title)]
        );
    }

    // =====================================================================
    // best_match — exact beats close, earliest phrase wins
    // =====================================================================

    #[test]
    fn best_match_reports_the_index_of_the_matching_phrase() {
        let phrases = vec!["alpha".to_string(), "beta".to_string(), "gamma".to_string()];
        assert_eq!(best_match("gamma", &phrases), (MatchKind::Exact, Some(2)));
        assert_eq!(best_match("beta", &phrases), (MatchKind::Exact, Some(1)));
        assert_eq!(best_match(MISS, &phrases), (MatchKind::None, None));
    }

    #[test]
    fn best_match_prefers_an_exact_hit_over_an_earlier_close_one() {
        let phrases = vec!["Alpha".to_string(), "beta".to_string()];
        // "alpha" is a case-tolerant (close) hit on phrase 0; "beta" would be
        // exact. Ask for the exact one and confirm it is not shadowed.
        assert_eq!(compare_phrase("alpha", &phrases[0]), PhraseMatch::Tolerant);
        assert_eq!(best_match("beta", &phrases), (MatchKind::Exact, Some(1)));
        assert_eq!(best_match("alpha", &phrases), (MatchKind::Close, Some(0)));
    }

    // =====================================================================
    // The one-guess flow
    // =====================================================================

    #[test]
    fn a_matching_guess_is_unhinted_and_shows_the_bloom() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let token = token_from_response(&begin_json);
        let resp = respond(&store, &bloom_id, "alpha", &token);

        assert_eq!(resp["status"], "shown");
        assert_eq!(resp["bucket"], "unhinted");
        assert_eq!(resp["guess"], "alpha");
        assert_eq!(resp["match"]["kind"], "exact");
        assert_eq!(resp["match"]["phrase_index"], 0);
        assert_eq!(resp["bloom"]["content"], "body");
        assert_eq!(resp["bloom"]["phrase_source"], "authored");
    }

    #[test]
    fn a_missed_guess_is_revealed_and_still_shows_the_bloom() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let token = token_from_response(&begin_json);
        let resp = respond(&store, &bloom_id, MISS, &token);

        assert_eq!(resp["status"], "shown");
        assert_eq!(resp["bucket"], "revealed");
        assert_eq!(resp["match"]["kind"], "none");
        assert!(resp["match"]["phrase_index"].is_null());
        // The bloom is shown on a miss too — there is no second attempt.
        assert_eq!(resp["bloom"]["content"], "body");
        assert_eq!(resp["bloom"]["phrases"][0], "alpha");
    }

    #[test]
    fn a_second_respond_on_one_token_replays_the_first_guess() {
        // One guess per bloom: a retry on the same token (a double submit, or
        // a lost response) is answered with what was logged, not judged again.
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);
        let mut second = entry_with_phrases(vec!["bravo"]);
        second.id = "kn-second".to_string();
        store.seed(&second);

        let begin_json = begin(&store, &test_cascade(vec![bloom, second]));
        let token = token_from_response(&begin_json);

        let first = respond(&store, &bloom_id, MISS, &token);
        assert_eq!(first["status"], "shown");
        assert!(first.get("replayed").is_none());

        // Same token, and this time the right answer. It does not count.
        let again = respond(&store, &bloom_id, "alpha", &token);
        assert_eq!(again["replayed"], true);
        assert_eq!(
            again["guess"], MISS,
            "the logged guess, not the retyped one"
        );
        assert_eq!(again["bucket"], "revealed");
        assert_eq!(again["match"]["kind"], "none");
        assert_eq!(again["session"], first["session"]);
        assert_eq!(again["next"]["id"], "kn-second");
        assert_eq!(store.guesses.borrow().len(), 1, "no second row");
        let sessions = store.sessions.borrow();
        let session = sessions.values().next().unwrap();
        assert_eq!(session.step, 1, "the replay does not advance again");
        assert_eq!(session.revealed_count, 1);
        assert_eq!(session.unhinted_count, 0);
    }

    #[test]
    fn a_lost_final_response_is_resumed_with_the_summary() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let token = token_from_response(&begin_json);
        let first = respond(&store, &bloom_id, "alpha", &token);
        assert!(first["summary"].is_object());

        // The final response never arrived; the caller retries.
        let resumed = respond(&store, &bloom_id, "alpha", &token);
        assert_eq!(resumed["replayed"], true);
        assert_eq!(resumed["bucket"], "unhinted");
        assert_eq!(resumed["summary"], first["summary"]);
        assert_eq!(resumed["bloom"]["content"], "body");
        assert!(resumed.get("next").is_none());
    }

    #[test]
    fn a_resume_naming_the_wrong_bloom_is_refused() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);
        let mut second = entry_with_phrases(vec!["bravo"]);
        second.id = "kn-second".to_string();
        store.seed(&second);

        let begin_json = begin(&store, &test_cascade(vec![bloom, second]));
        let token = token_from_response(&begin_json);
        respond(&store, &bloom_id, MISS, &token);

        let refusal = reject(&store, "kn-second", "bravo", &token);
        assert_eq!(refusal["error"], "invalid_bloom_id");
        assert_eq!(refusal["expected_id"], bloom_id.as_str());
    }

    #[test]
    fn a_token_two_steps_stale_is_refused_and_the_error_names_the_recovery() {
        let store = MockStore::new();
        let mut blooms = Vec::new();
        for (id, phrase) in [("kn-1", "one"), ("kn-2", "two"), ("kn-3", "three")] {
            let mut b = entry_with_phrases(vec![phrase]);
            b.id = id.to_string();
            store.seed(&b);
            blooms.push(b);
        }
        let begin_json = begin(&store, &test_cascade(blooms));
        let token0 = token_from_response(&begin_json);
        let r1 = respond(&store, "kn-1", "one", &token0);
        respond(&store, "kn-2", "two", &token_from_response(&r1));

        let err = respond_ritual(
            &store,
            &AgentContext::for_agent("test-agent"),
            "kn-1",
            "one",
            &token0,
        )
        .unwrap_err()
        .to_string();
        assert!(err.contains("Token out of sync"), "{err}");
        assert!(
            err.contains("retry with the token from your last successful response")
                || err.contains("Retry with the token from your last successful response"),
            "{err}"
        );
        assert_eq!(store.guesses.borrow().len(), 2);
    }

    #[test]
    fn a_half_written_step_is_replayed_and_the_session_catches_up() {
        // The released binary wrote the row, then the session, in two calls.
        // A failure between them left the row logged and the session behind,
        // and every retry then tripped the unique index. Seed exactly that.
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);
        let mut second = entry_with_phrases(vec!["bravo"]);
        second.id = "kn-second".to_string();
        store.seed(&second);

        let begin_json = begin(&store, &test_cascade(vec![bloom, second]));
        let token = token_from_response(&begin_json);
        let session_id = store.sessions.borrow().keys().next().unwrap().clone();

        store.guesses.borrow_mut().push(WakeGuessRow {
            agent: "test-agent".to_string(),
            wake: Some(7),
            session_id: session_id.clone(),
            bloom_id: bloom_id.clone(),
            chunk_index: 0,
            chunk_total: 1,
            position: 0,
            bloom_position: 1,
            bloom_total: 2,
            title_shown: "Test".to_string(),
            guess: "alpha".to_string(),
            model_id: Some("test-model".to_string()),
            phrase_source: "authored".to_string(),
            phrases: vec!["alpha".to_string()],
            match_kind: "exact".to_string(),
            match_index: Some(0),
            bucket: "unhinted".to_string(),
            content_hash: content_hash("body"),
        });

        // The retry sends a different guess; the logged one counts.
        let resp = respond(&store, &bloom_id, MISS, &token);
        assert_eq!(resp["replayed"], true);
        assert_eq!(resp["guess"], "alpha");
        assert_eq!(resp["bucket"], "unhinted");
        assert_eq!(resp["match"]["kind"], "exact");
        assert_eq!(resp["next"]["id"], "kn-second");
        assert_eq!(store.guesses.borrow().len(), 1, "exactly one row");
        {
            let sessions = store.sessions.borrow();
            let session = &sessions[&session_id];
            assert_eq!(session.step, 1, "the session caught up");
            assert_eq!(
                session.unhinted_count, 1,
                "counted with the logged judgment"
            );
        }

        // The token the replay handed back walks on.
        let resp = respond(&store, "kn-second", "bravo", &token_from_response(&resp));
        assert_eq!(resp["status"], "shown");
        assert_eq!(resp["summary"]["buckets"]["unhinted"]["authored"], 2);
    }

    #[test]
    fn the_response_carries_no_hint_ladder_and_no_retired_vocabulary() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let raw = respond_ritual(
            &store,
            &AgentContext::for_agent("test-agent"),
            &bloom_id,
            MISS,
            &token_from_response(&begin_json),
        )
        .unwrap();

        for retired in [
            "hint",
            "attempt",
            "remembered",
            "needed_help",
            "incorrect",
            "skipped",
            "match_type",
            "derived_phrase_mismatch",
            "wake_phrase_count",
            "matched_phrase",
            "all_phrases",
            "blooms_complete",
        ] {
            let key = format!("\"{retired}\"");
            assert!(
                !raw.contains(&key),
                "respond payload still carries {key}: {raw}"
            );
        }
    }

    #[test]
    fn a_guess_matching_the_second_or_third_authored_phrase_is_unhinted() {
        // The old matcher compared chunk i against wake_phrases[i] only, so on
        // a single-chunk bloom phrases 1 and 2 could never match (#450).
        for (guess, expected_index) in [("beta", 1), ("gamma", 2)] {
            let store = MockStore::new();
            let bloom = entry_with_phrases(vec!["alpha", "beta", "gamma"]);
            let bloom_id = bloom.id.clone();
            store.seed(&bloom);

            let begin_json = begin(&store, &test_cascade(vec![bloom]));
            let resp = respond(&store, &bloom_id, guess, &token_from_response(&begin_json));

            assert_eq!(resp["bucket"], "unhinted", "guess {guess:?}");
            assert_eq!(resp["match"]["kind"], "exact", "guess {guess:?}");
            assert_eq!(resp["match"]["phrase_index"], expected_index);
            assert_eq!(store.guesses.borrow()[0].match_index, Some(expected_index));
        }
    }

    #[test]
    fn an_authored_phrase_matches_case_insensitively() {
        // Authored phrases used to need a case-sensitive exact match.
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["The Long Way Round"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let resp = respond(
            &store,
            &bloom_id,
            "the long way round.",
            &token_from_response(&begin_json),
        );
        assert_eq!(resp["bucket"], "unhinted");
        assert_eq!(resp["match"]["kind"], "close");
        assert_eq!(resp["match"]["phrase_index"], 0);
    }

    fn seed_ids(store: &MockStore, ids: &[(&str, &str)]) -> Vec<KnowledgeEntry> {
        ids.iter()
            .map(|(id, phrase)| {
                let mut b = entry_with_phrases(vec![phrase]);
                b.id = id.to_string();
                store.seed(&b);
                b
            })
            .collect()
    }

    #[test]
    fn a_judged_step_that_also_stepped_over_a_deleted_entry_resumes() {
        // One call judges kn-1 and then steps over the deleted kn-2, moving
        // step 0 -> 2. A `step - 1` resume check could not see token 0.
        let store = MockStore::new();
        let blooms = seed_ids(
            &store,
            &[("kn-1", "one"), ("kn-2", "two"), ("kn-3", "three")],
        );
        let begin_json = begin(&store, &test_cascade(blooms));
        let token0 = token_from_response(&begin_json);
        store.blooms.borrow_mut().remove("kn-2");

        let first = respond(&store, "kn-1", "one", &token0);
        assert_eq!(first["next"]["id"], "kn-3");
        assert_eq!(store.sessions.borrow().values().next().unwrap().step, 2);

        let resumed = respond(&store, "kn-1", "one", &token0);
        assert_eq!(resumed["replayed"], true);
        assert_eq!(resumed["status"], "shown");
        assert_eq!(resumed["guess"], "one");
        assert_eq!(resumed["session"], first["session"]);
        assert_eq!(resumed["next"]["id"], "kn-3");
        assert_eq!(store.guesses.borrow().len(), 1);

        // Once the caller walks on, token 0 is genuinely stale.
        respond(&store, "kn-3", "three", &token_from_response(&first));
        let err = respond_ritual(
            &store,
            &AgentContext::for_agent("test-agent"),
            "kn-1",
            "one",
            &token0,
        )
        .unwrap_err()
        .to_string();
        assert!(err.contains("Token out of sync"), "{err}");
    }

    #[test]
    fn a_lost_bloom_missing_response_resumes_with_its_status() {
        let store = MockStore::new();
        let blooms = seed_ids(&store, &[("kn-1", "one"), ("kn-2", "two")]);
        let begin_json = begin(&store, &test_cascade(blooms));
        let token0 = token_from_response(&begin_json);
        store.blooms.borrow_mut().remove("kn-1");

        let first = respond(&store, "kn-1", "one", &token0);
        assert_eq!(first["status"], "bloom_missing");

        let resumed = respond(&store, "kn-1", "one", &token0);
        assert_eq!(resumed["status"], "bloom_missing");
        assert_eq!(resumed["replayed"], true);
        assert!(resumed.get("bucket").is_none() && resumed.get("guess").is_none());
        assert_eq!(resumed["session"], first["session"]);
        assert_eq!(resumed["next"]["id"], "kn-2");
        assert!(store.guesses.borrow().is_empty(), "nothing was judged");

        // The resumed token walks on.
        let resp = respond(&store, "kn-2", "two", &token_from_response(&resumed));
        assert_eq!(resp["status"], "shown");
        assert_eq!(resp["summary"]["unjudged"], 1);
    }

    #[test]
    fn a_lost_chunk_truncated_response_resumes_with_its_status() {
        let store = MockStore::new();
        let big = make_large_bloom(95_000, vec!["alpha", "beta", "gamma"]);
        let big_id = big.id.clone();
        store.seed(&big);
        let mut small = entry_with_phrases(vec!["bravo"]);
        small.id = "kn-small".to_string();
        store.seed(&small);

        let begin_json = begin(&store, &test_cascade(vec![big, small]));
        let mut token = token_from_response(&begin_json);
        for _ in 0..2 {
            token = token_from_response(&respond(&store, &big_id, MISS, &token));
        }
        store.mutate_bloom(&big_id, |e| {
            e.body = Some("shrunk to one chunk.".to_string())
        });

        let first = respond(&store, &big_id, "ignored", &token);
        assert_eq!(first["status"], "chunk_truncated");

        let resumed = respond(&store, &big_id, "ignored", &token);
        assert_eq!(resumed["status"], "chunk_truncated");
        assert_eq!(resumed["replayed"], true);
        assert_eq!(resumed["session"], first["session"]);
        assert_eq!(resumed["next"]["id"], "kn-small");
        assert_eq!(store.guesses.borrow().len(), 2);
    }

    #[test]
    fn a_half_written_step_whose_entry_was_since_deleted_still_replays() {
        // The logged row wins over the bloom_missing answer: the guess was
        // made and logged, and it is what the step counts.
        let store = MockStore::new();
        let blooms = seed_ids(&store, &[("kn-1", "one"), ("kn-2", "two")]);
        let begin_json = begin(&store, &test_cascade(blooms));
        let token0 = token_from_response(&begin_json);
        let session_id = store.sessions.borrow().keys().next().unwrap().clone();
        store.guesses.borrow_mut().push(WakeGuessRow {
            agent: "test-agent".to_string(),
            wake: Some(7),
            session_id: session_id.clone(),
            bloom_id: "kn-1".to_string(),
            chunk_index: 0,
            chunk_total: 1,
            position: 0,
            bloom_position: 1,
            bloom_total: 2,
            title_shown: "Test".to_string(),
            guess: "one".to_string(),
            model_id: None,
            phrase_source: "authored".to_string(),
            phrases: vec!["one".to_string()],
            match_kind: "exact".to_string(),
            match_index: Some(0),
            bucket: "unhinted".to_string(),
            content_hash: content_hash("body"),
        });
        store.blooms.borrow_mut().remove("kn-1");

        let resp = respond(&store, "kn-1", MISS, &token0);
        assert_eq!(resp["status"], "shown");
        assert_eq!(resp["replayed"], true);
        assert_eq!(resp["guess"], "one");
        assert!(resp.get("bloom").is_none(), "the entry is gone");
        assert_eq!(resp["next"]["id"], "kn-2");
        let sessions = store.sessions.borrow();
        assert_eq!(sessions[&session_id].step, 1);
        assert_eq!(sessions[&session_id].unhinted_count, 1);
        assert_eq!(sessions[&session_id].unjudged_count, 0);
    }

    #[test]
    fn the_loser_of_a_double_submit_gets_the_winners_answer() {
        // Two calls on one token: the other one lands its row and advance
        // between this call's lookup and its write. The loser used to get a
        // raw index error; it now gets the logged answer.
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);
        let mut second = entry_with_phrases(vec!["bravo"]);
        second.id = "kn-second".to_string();
        store.seed(&second);

        let begin_json = begin(&store, &test_cascade(vec![bloom, second]));
        let token = token_from_response(&begin_json);
        let session_id = store.sessions.borrow().keys().next().unwrap().clone();

        let mut winner_session = store.sessions.borrow()[&session_id].clone();
        winner_session.advance(1, Bucket::Unhinted, PhraseSource::Authored);
        issue_next_prompt(&mut winner_session, &store.blooms.borrow()).unwrap();
        let winner_row = WakeGuessRow {
            agent: "test-agent".to_string(),
            wake: Some(7),
            session_id: session_id.clone(),
            bloom_id: bloom_id.clone(),
            chunk_index: 0,
            chunk_total: 1,
            position: 0,
            bloom_position: 1,
            bloom_total: 2,
            title_shown: "Test".to_string(),
            guess: "alpha".to_string(),
            model_id: Some("test-model".to_string()),
            phrase_source: "authored".to_string(),
            phrases: vec!["alpha".to_string()],
            match_kind: "exact".to_string(),
            match_index: Some(0),
            bucket: "unhinted".to_string(),
            content_hash: content_hash("body"),
        };
        *store.race_winner.borrow_mut() = Some((winner_row, winner_session));

        let resp = respond(&store, &bloom_id, MISS, &token);
        assert_eq!(resp["replayed"], true);
        assert_eq!(
            resp["guess"], "alpha",
            "the winner's guess is the one logged"
        );
        assert_eq!(resp["bucket"], "unhinted");
        assert_eq!(resp["next"]["id"], "kn-second");
        assert_eq!(store.guesses.borrow().len(), 1);
        assert_eq!(store.sessions.borrow()[&session_id].step, 1);
    }

    // =====================================================================
    // match.kind is monotonic: exact means identical after trimming
    // =====================================================================

    #[test]
    fn exact_means_identical_after_trim_and_nothing_else() {
        let phrases = vec!["The Long Way Round".to_string()];
        assert_eq!(
            best_match("The Long Way Round", &phrases),
            (MatchKind::Exact, Some(0))
        );
        assert_eq!(
            best_match("  The Long Way Round \n", &phrases),
            (MatchKind::Exact, Some(0))
        );
    }

    #[test]
    fn a_guess_differing_only_by_case_is_close() {
        let phrases = vec!["The Long Way Round".to_string()];
        assert_eq!(
            best_match("the long way round", &phrases),
            (MatchKind::Close, Some(0))
        );
    }

    #[test]
    fn a_guess_wrapped_in_quotes_is_not_exact() {
        // Review counter-example: the fuzzy matcher strips punctuation, so the
        // quoted guess used to come back EXACT while the bare lowercase one
        // was only close. A farther guess must never get the better verdict.
        let phrases = vec!["The Long Way Round".to_string()];
        let bare = best_match("the long way round", &phrases).0;
        for quoted in ["\"the long way round\"", "\"The Long Way Round\""] {
            let kind = best_match(quoted, &phrases).0;
            assert_eq!(kind, MatchKind::Close, "{quoted}");
            assert_eq!(kind, bare, "{quoted} must not outrank the bare guess");
        }
    }

    #[test]
    fn a_guess_missing_an_apostrophe_is_close() {
        let phrases = vec!["Q's den".to_string()];
        assert_eq!(best_match("q's den", &phrases), (MatchKind::Close, Some(0)));
        assert_eq!(best_match("Qs den", &phrases), (MatchKind::Close, Some(0)));
    }

    // =====================================================================
    // --wake: echoed, checked against the log, never silently misfiled
    // =====================================================================

    fn begin_at(store: &MockStore, wake: Option<i64>, force: bool) -> Result<serde_json::Value> {
        let bloom = entry_with_phrases(vec!["alpha"]);
        store.seed(&bloom);
        let out = begin_ritual(
            store,
            &test_cascade(vec![bloom]),
            RitualMeta {
                agent: "test-agent".to_string(),
                wake,
                model_id: Some("test-model".to_string()),
            },
            force,
        )?;
        Ok(serde_json::from_str(&out).unwrap())
    }

    /// Walk a one-bloom ritual under `wake` so the log holds a row for it.
    fn walk_at(store: &MockStore, wake: i64, force: bool) -> serde_json::Value {
        let begin_json = begin_at(store, Some(wake), force).unwrap();
        respond(store, "kn-test", "alpha", &token_from_response(&begin_json))
    }

    #[test]
    fn the_wake_and_model_are_echoed_on_begin_and_every_respond() {
        let store = MockStore::new();
        let begin_json = begin_at(&store, Some(7), false).unwrap();
        assert_eq!(begin_json["wake"], 7);
        assert_eq!(begin_json["model"], "test-model");
        assert!(
            begin_json.get("warnings").is_none(),
            "nothing to warn about"
        );

        let resp = respond(
            &store,
            "kn-test",
            "alpha",
            &token_from_response(&begin_json),
        );
        assert_eq!(resp["wake"], 7);
        assert_eq!(resp["model"], "test-model");
    }

    #[test]
    fn an_omitted_wake_is_echoed_as_null() {
        let store = MockStore::new();
        let begin_json = begin_at(&store, None, false).unwrap();
        assert!(begin_json["wake"].is_null());
        assert!(
            begin_json.get("wake").is_some(),
            "the key is always present"
        );
    }

    #[test]
    fn a_wake_number_that_is_not_positive_is_refused() {
        for wake in [0, -1, -463] {
            let store = MockStore::new();
            let err = begin_at(&store, Some(wake), false).unwrap_err().to_string();
            assert!(err.contains("positive"), "{wake}: {err}");
            assert!(store.sessions.borrow().is_empty(), "no session opened");
        }
    }

    #[test]
    fn a_wake_number_another_ritual_logged_under_is_refused() {
        let store = MockStore::new();
        walk_at(&store, 463, false);

        let err = begin_at(&store, Some(463), false).unwrap_err().to_string();
        assert!(err.contains("Wake 463 already has guesses"), "{err}");
        assert!(
            err.contains("--force-wake"),
            "the refusal names the override: {err}"
        );
        assert_eq!(store.sessions.borrow().len(), 1, "no second session opened");
    }

    #[test]
    fn force_wake_opens_the_ritual_anyway_and_says_so() {
        let store = MockStore::new();
        walk_at(&store, 463, false);

        let begin_json = begin_at(&store, Some(463), true).unwrap();
        assert_eq!(begin_json["status"], "ritual_started");
        assert_eq!(begin_json["wake"], 463);
        let warnings = begin_json["warnings"].as_array().unwrap();
        assert!(
            warnings
                .iter()
                .any(|w| w.as_str().unwrap().contains("--force-wake")),
            "{warnings:?}"
        );
    }

    #[test]
    fn a_wake_number_below_the_highest_logged_is_warned_about() {
        let store = MockStore::new();
        walk_at(&store, 463, false);

        let begin_json = begin_at(&store, Some(46), false).unwrap();
        assert_eq!(begin_json["status"], "ritual_started");
        let warnings = begin_json["warnings"].as_array().unwrap();
        assert_eq!(warnings.len(), 1, "{warnings:?}");
        assert!(
            warnings[0]
                .as_str()
                .unwrap()
                .contains("lower than wake 463"),
            "{warnings:?}"
        );

        // The next wake up is not warned about.
        let begin_json = begin_at(&store, Some(464), false).unwrap();
        assert!(begin_json.get("warnings").is_none());
    }

    #[test]
    fn a_wake_number_that_skips_ahead_of_the_highest_logged_is_warned_about() {
        let store = MockStore::new();
        walk_at(&store, 463, false);

        let begin_json = begin_at(&store, Some(4640), false).unwrap();
        let warnings = begin_json["warnings"].as_array().unwrap();
        assert_eq!(warnings.len(), 1, "{warnings:?}");
        assert!(
            warnings[0]
                .as_str()
                .unwrap()
                .contains("skips ahead of wake 463"),
            "{warnings:?}"
        );
    }

    #[test]
    fn another_agents_rows_do_not_block_a_wake_number() {
        let store = MockStore::new();
        walk_at(&store, 463, false);
        for row in store.guesses.borrow_mut().iter_mut() {
            row.agent = "someone-else".to_string();
        }
        let begin_json = begin_at(&store, Some(463), false).unwrap();
        assert_eq!(begin_json["status"], "ritual_started");
        assert!(begin_json.get("warnings").is_none());
    }

    // =====================================================================
    // The guess log
    // =====================================================================

    #[test]
    fn each_respond_writes_exactly_one_row_with_both_positions() {
        let store = MockStore::new();
        let mut first = entry_with_phrases(vec!["alpha"]);
        first.id = "kn-first".to_string();
        first.title = "First".to_string();
        let mut second = entry_with_phrases(vec!["bravo"]);
        second.id = "kn-second".to_string();
        second.title = "Second".to_string();
        store.seed(&first);
        store.seed(&second);

        let begin_json = begin(&store, &test_cascade(vec![first, second]));
        let mut token = token_from_response(&begin_json);
        for (bloom_id, guess) in [("kn-first", "alpha"), ("kn-second", MISS)] {
            let resp = respond(&store, bloom_id, guess, &token);
            token = token_from_response(&resp);
        }

        let rows = store.guesses.borrow();
        assert_eq!(rows.len(), 2, "one row per respond");

        assert_eq!(rows[0].bloom_id, "kn-first");
        assert_eq!(rows[0].position, 0);
        assert_eq!(rows[0].bloom_position, 1);
        assert_eq!(rows[0].bloom_total, 2);
        assert_eq!(rows[0].chunk_index, 0);
        assert_eq!(rows[0].chunk_total, 1);
        assert_eq!(rows[0].title_shown, "First");
        assert_eq!(rows[0].bucket, "unhinted");
        assert_eq!(rows[0].match_kind, "exact");
        assert_eq!(rows[0].phrases, vec!["alpha"]);
        assert_eq!(
            rows[0].content_hash,
            crate::wake_guess::content_hash("body")
        );

        assert_eq!(rows[1].bloom_id, "kn-second");
        assert_eq!(rows[1].position, 1);
        assert_eq!(rows[1].bloom_position, 2);
        assert_eq!(rows[1].bucket, "revealed");
        assert_eq!(rows[1].match_kind, "none");
        assert_eq!(rows[1].match_index, None);
        assert_eq!(rows[1].guess, MISS);
    }

    #[test]
    fn the_wake_and_model_flags_land_on_every_row() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        respond(
            &store,
            &bloom_id,
            "alpha",
            &token_from_response(&begin_json),
        );

        let rows = store.guesses.borrow();
        assert_eq!(rows[0].wake, Some(7));
        assert_eq!(rows[0].model_id.as_deref(), Some("test-model"));
        assert_eq!(rows[0].agent, "test-agent");
    }

    #[test]
    fn an_omitted_wake_number_logs_as_absent() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json: serde_json::Value = serde_json::from_str(
            &begin_ritual(
                &store,
                &test_cascade(vec![bloom]),
                RitualMeta {
                    agent: "test-agent".to_string(),
                    wake: None,
                    model_id: None,
                },
                false,
            )
            .unwrap(),
        )
        .unwrap();
        respond(
            &store,
            &bloom_id,
            "alpha",
            &token_from_response(&begin_json),
        );

        let rows = store.guesses.borrow();
        assert_eq!(rows[0].wake, None);
        assert_eq!(rows[0].model_id, None);
    }

    #[test]
    fn a_chunked_bloom_advances_position_but_not_bloom_position() {
        let store = MockStore::new();
        let bloom = make_large_bloom(95_000, vec!["alpha", "beta", "gamma"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let total_chunks = begin_json["progress"]["total"].as_u64().unwrap();
        assert!(total_chunks >= 3, "fixture must chunk; got {total_chunks}");

        let mut token = token_from_response(&begin_json);
        for _ in 0..total_chunks {
            let resp = respond(&store, &bloom_id, MISS, &token);
            token = token_from_response(&resp);
        }

        let rows = store.guesses.borrow();
        assert_eq!(rows.len(), total_chunks as usize, "one row per chunk");
        for (idx, row) in rows.iter().enumerate() {
            assert_eq!(row.position, idx as u32, "position advances per chunk");
            assert_eq!(row.bloom_position, 1, "still the same bloom");
            assert_eq!(row.chunk_index, idx as u16);
            assert!(
                row.title_shown.contains(&format!("(Part {}/", idx + 1)),
                "chunked title must carry its part suffix: {}",
                row.title_shown
            );
        }
    }

    #[test]
    fn a_chunk_truncated_response_writes_no_row() {
        let store = MockStore::new();
        let bloom = make_large_bloom(95_000, vec!["alpha", "beta", "gamma"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let mut token = token_from_response(&begin_json);

        // Walk two chunks, then shrink the bloom under the cursor.
        for _ in 0..2 {
            let resp = respond(&store, &bloom_id, MISS, &token);
            token = token_from_response(&resp);
        }
        assert_eq!(store.guesses.borrow().len(), 2);

        store.mutate_bloom(&bloom_id, |entry| {
            entry.body = Some("shrunk down to a single tiny chunk now.".to_string());
        });

        let resp = respond(&store, &bloom_id, "ignored", &token);
        assert_eq!(resp["status"], "chunk_truncated");
        assert!(
            resp.get("bucket").is_none(),
            "no bucket without a judgement"
        );
        assert!(resp.get("guess").is_none());
        assert_eq!(
            store.guesses.borrow().len(),
            2,
            "a truncated chunk judges no guess, so it logs none"
        );
    }

    #[test]
    fn a_failed_row_write_fails_the_respond_and_leaves_the_session_put() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let token = token_from_response(&begin_json);

        store.fail_guess_write.set(true);
        let err = respond_ritual(
            &store,
            &AgentContext::for_agent("test-agent"),
            &bloom_id,
            "alpha",
            &token,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("guess log unavailable"),
            "the write failure must surface, got: {err}"
        );

        {
            let sessions = store.sessions.borrow();
            let session = sessions.values().next().expect("session must survive");
            assert_eq!(
                session.step, 0,
                "session must not advance past a lost guess"
            );
            assert_eq!(session.current_index, 0);
            assert_eq!(session.unhinted_count, 0);
        }

        // With the log back, the same token still works.
        store.fail_guess_write.set(false);
        let resp = respond(&store, &bloom_id, "alpha", &token);
        assert_eq!(resp["status"], "shown");
        assert_eq!(store.guesses.borrow().len(), 1);
    }

    #[test]
    fn a_bloom_deleted_while_it_is_on_the_table_is_stepped_over() {
        // The brick case above covers an entry already walked past. This is the
        // harder one: the entry the ritual is asking about right now.
        let store = MockStore::new();
        let mut first = entry_with_phrases(vec!["alpha"]);
        first.id = "kn-first".to_string();
        let mut second = entry_with_phrases(vec!["bravo"]);
        second.id = "kn-second".to_string();
        store.seed(&first);
        store.seed(&second);

        let begin_json = begin(&store, &test_cascade(vec![first, second]));
        let token = token_from_response(&begin_json);
        store.blooms.borrow_mut().remove("kn-first");

        let resp = respond(&store, "kn-first", "alpha", &token);
        assert_eq!(resp["status"], "bloom_missing");
        assert!(resp.get("bloom").is_none(), "there is nothing to show");
        assert!(resp.get("bucket").is_none(), "nothing was judged");
        assert!(
            store.guesses.borrow().is_empty(),
            "a vanished bloom logs no guess"
        );

        // The ritual keeps going: the next entry is served and answerable.
        assert_eq!(resp["next"]["id"], "kn-second");
        let resp = respond(&store, "kn-second", "bravo", &token_from_response(&resp));
        assert_eq!(resp["status"], "shown");
        assert_eq!(resp["bucket"], "unhinted");
        // The skipped step is counted as walked but sits in no bucket.
        assert_eq!(resp["summary"]["chunks"], 2);
        assert_eq!(resp["summary"]["buckets"]["unhinted"]["authored"], 1);
        assert_eq!(resp["summary"]["buckets"]["revealed"]["authored"], 0);
    }

    #[test]
    fn a_bloom_deleted_further_down_the_sequence_is_never_prompted_for() {
        // The `next` pointer must name an entry that still exists.
        let store = MockStore::new();
        let mut first = entry_with_phrases(vec!["alpha"]);
        first.id = "kn-first".to_string();
        let mut second = entry_with_phrases(vec!["bravo"]);
        second.id = "kn-second".to_string();
        let mut third = entry_with_phrases(vec!["charlie"]);
        third.id = "kn-third".to_string();
        store.seed(&first);
        store.seed(&second);
        store.seed(&third);

        let begin_json = begin(&store, &test_cascade(vec![first, second, third]));
        let token = token_from_response(&begin_json);
        store.blooms.borrow_mut().remove("kn-second");

        let resp = respond(&store, "kn-first", "alpha", &token);
        assert_eq!(resp["status"], "shown");
        assert_eq!(
            resp["next"]["id"], "kn-third",
            "the deleted entry must be stepped over, not prompted for: {resp}"
        );
    }

    // =====================================================================
    // The prompt snapshot: a row records what the guesser was shown (#467)
    // =====================================================================

    fn titled(id: &str, title: &str, phrase: &str) -> KnowledgeEntry {
        let mut e = entry_with_phrases(vec![phrase]);
        e.id = id.to_string();
        e.title = title.to_string();
        e
    }

    fn only_session(store: &MockStore) -> WakeSession {
        store.sessions.borrow().values().next().unwrap().clone()
    }

    #[test]
    fn a_title_edited_after_its_prompt_is_logged_as_prompted() {
        let store = MockStore::new();
        let first = titled("kn-first", "First", "alpha");
        let second = titled("kn-second", "Second", "bravo");
        store.seed(&first);
        store.seed(&second);

        // Edited between --begin and the respond.
        let begin_json = begin(&store, &test_cascade(vec![first, second]));
        assert_eq!(begin_json["prompt"]["title"], "First");
        store.mutate_bloom("kn-first", |e| e.title = "First, renamed".to_string());
        let resp = respond(
            &store,
            "kn-first",
            "alpha",
            &token_from_response(&begin_json),
        );

        // Edited between the respond that handed out `next` and the next one.
        assert_eq!(resp["next"]["title"], "Second");
        store.mutate_bloom("kn-second", |e| e.title = "Second, renamed".to_string());
        respond(&store, "kn-second", "bravo", &token_from_response(&resp));

        let rows = store.guesses.borrow();
        assert_eq!(rows[0].title_shown, "First");
        assert_eq!(rows[1].title_shown, "Second");
    }

    #[test]
    fn a_bloom_rechunked_after_its_prompt_is_logged_with_the_prompted_part() {
        let store = MockStore::new();
        let bloom = make_large_bloom(95_000, vec!["alpha", "beta", "gamma"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        let prompted_total = begin_json["prompt"]["chunk"]["total"].as_u64().unwrap() as u16;
        let resp = respond(&store, &bloom_id, MISS, &token_from_response(&begin_json));
        let prompted_title = format!("Ops (Part 2/{prompted_total})");
        assert_eq!(resp["next"]["title"], prompted_title.as_str());

        store.mutate_bloom(&bloom_id, |e| {
            e.body = make_large_bloom(160_000, vec![]).body;
        });
        let current_total = compute_chunks(
            &bloom_content(&store.blooms.borrow()[&bloom_id]),
            chunk_threshold(),
        )
        .total;
        assert_ne!(current_total, prompted_total, "the fixture must re-chunk");

        let spent = token_from_response(&resp);
        let resp = respond(&store, &bloom_id, MISS, &spent);
        {
            let rows = store.guesses.borrow();
            assert_eq!(rows[1].title_shown, prompted_title);
            assert_eq!(rows[1].chunk_total, prompted_total);
            assert_eq!(rows[1].chunk_index, 1);
        }

        // Navigation is the bloom as it is now, exactly as before this change:
        // the cursor advanced under the current plan, not the prompted one.
        let next_title = format!("Ops (Part 3/{current_total})");
        assert_eq!(resp["next"]["title"], next_title.as_str());
        let after = only_session(&store);
        assert_eq!(after.current_chunk_index, 2);

        // A retry of the spent token is answered from the row and the record,
        // and moves nothing.
        let again = respond(&store, &bloom_id, "alpha", &spent);
        assert_eq!(again["replayed"], true);
        assert_eq!(again["bloom"]["title"], prompted_title.as_str());
        assert_eq!(again["bloom"], resp["bloom"]);
        assert_eq!(again["next"], resp["next"]);
        let replayed = only_session(&store);
        assert_eq!(replayed.step, after.step);
        assert_eq!(replayed.current_chunk_index, after.current_chunk_index);
        assert_eq!(store.guesses.borrow().len(), 2);
    }

    #[test]
    fn an_unchanged_entry_logs_the_row_it_always_did() {
        let store = MockStore::new();
        let mut big = make_large_bloom(60_000, vec!["alpha"]);
        big.id = "kn-big".to_string();
        let plain = titled("kn-plain", "Plain", "bravo");
        store.seed(&big);
        store.seed(&plain);

        let begin_json = begin(&store, &test_cascade(vec![big.clone(), plain.clone()]));
        let big_total = compute_chunks(&bloom_content(&big), chunk_threshold()).total;
        assert!(big_total > 1, "the fixture must chunk");

        let mut steps: Vec<(&KnowledgeEntry, u16, &str)> = Vec::new();
        for idx in 0..big_total {
            steps.push((&big, idx, if idx == 0 { "alpha" } else { MISS }));
        }
        steps.push((&plain, 0, "bravo"));

        let mut token = token_from_response(&begin_json);
        for (entry, _, guess) in &steps {
            token = token_from_response(&respond(&store, &entry.id, guess, &token));
        }

        // The row as main builds it, from the entry at respond time.
        let rows = store.guesses.borrow();
        assert_eq!(rows.len(), steps.len());
        for (position, (entry, chunk_idx, guess)) in steps.iter().enumerate() {
            let content = bloom_content(entry);
            let plan = compute_chunks(&content, chunk_threshold());
            let chunk_content = chunk_text(&plan, &content, *chunk_idx);
            let resolved = phrases_for_chunk(entry, &plan, &content, *chunk_idx);
            let (match_kind, match_index) = best_match(guess, &resolved.phrases);
            let bucket = match match_kind {
                MatchKind::None => Bucket::Revealed,
                _ => Bucket::Unhinted,
            };
            let expected = WakeGuessRow {
                agent: "test-agent".to_string(),
                wake: Some(7),
                session_id: rows[0].session_id.clone(),
                bloom_id: entry.id.clone(),
                chunk_index: *chunk_idx,
                chunk_total: plan.total,
                position: position as u32,
                bloom_position: if entry.id == "kn-big" { 1 } else { 2 },
                bloom_total: 2,
                title_shown: title_for_chunk(entry, *chunk_idx, &plan),
                guess: guess.to_string(),
                model_id: Some("test-model".to_string()),
                phrase_source: resolved.source.as_str().to_string(),
                phrases: resolved.phrases,
                match_kind: match_kind.as_str().to_string(),
                match_index,
                bucket: bucket.as_str().to_string(),
                content_hash: content_hash(chunk_content),
            };
            // Debug prints every field, so this compares all of them.
            assert_eq!(
                format!("{:?}", rows[position]),
                format!("{expected:?}"),
                "row {position}"
            );
        }
    }

    #[test]
    fn a_session_awaiting_a_response_without_a_prompt_record_is_refused() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        let bloom_id = bloom.id.clone();
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        // What an older binary leaves: a session with no prompt record.
        let session_id = only_session(&store).session_id;
        store
            .sessions
            .borrow_mut()
            .get_mut(&session_id)
            .unwrap()
            .prompt = None;

        let err = respond_ritual(
            &store,
            &AgentContext::for_agent("test-agent"),
            &bloom_id,
            "alpha",
            &token_from_response(&begin_json),
        )
        .expect_err("a session with no prompt record must be refused");
        let msg = err.to_string();
        assert!(
            msg.contains("older version") && msg.contains("--begin"),
            "{msg}"
        );
        assert!(store.guesses.borrow().is_empty(), "nothing is logged");
        let session = only_session(&store);
        assert_eq!(session.step, 0, "nothing is advanced");
        assert_eq!(session.prev_step, None, "nothing is written");
    }

    #[test]
    fn a_resumed_next_is_the_prompt_its_guess_is_logged_under() {
        let store = MockStore::new();
        let first = titled("kn-first", "First", "alpha");
        let second = titled("kn-second", "Second", "bravo");
        store.seed(&first);
        store.seed(&second);

        let begin_json = begin(&store, &test_cascade(vec![first, second]));
        let token0 = token_from_response(&begin_json);
        let lost = respond(&store, "kn-first", "alpha", &token0);
        assert_eq!(lost["next"]["title"], "Second");

        // The response was lost, and the next entry is renamed before the retry.
        store.mutate_bloom("kn-second", |e| e.title = "Second, renamed".to_string());
        let resumed = respond(&store, "kn-first", "alpha", &token0);
        assert_eq!(resumed["replayed"], true);
        assert_eq!(resumed["next"], lost["next"], "the prompt as handed out");

        respond(&store, "kn-second", "bravo", &token_from_response(&resumed));
        assert_eq!(store.guesses.borrow()[1].title_shown, "Second");
    }

    #[test]
    fn a_retry_repeats_the_reveal_of_an_entry_edited_before_its_response() {
        // A retry after a lost response is meant to be the same answer. The
        // edit lands between the prompt and the respond; nothing changes after.
        let store = MockStore::new();
        let first = titled("kn-first", "First", "alpha");
        let second = titled("kn-second", "Second", "bravo");
        store.seed(&first);
        store.seed(&second);
        let b = begin(&store, &test_cascade(vec![first, second]));
        let t0 = token_from_response(&b);
        store.mutate_bloom("kn-first", |e| {
            e.title = "First, renamed".to_string();
            e.wake_phrases.clear();
        });

        let original = respond(&store, "kn-first", MISS, &t0);
        assert_eq!(original["bloom"]["title"], "First");
        assert_eq!(original["bloom"]["phrase_source"], "authored");
        let retry = respond(&store, "kn-first", MISS, &t0);
        assert_eq!(retry["replayed"], true);
        assert_eq!(retry["bloom"], original["bloom"]);
        let rows = store.guesses.borrow();
        assert_eq!(rows[0].title_shown, "First");
        assert_eq!(rows[0].phrase_source, "authored");
    }

    #[test]
    fn the_row_and_the_summary_count_the_prompted_phrase_source() {
        let store = MockStore::new();
        let bloom = titled("kn-only", "Only", "alpha");
        store.seed(&bloom);

        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        assert_eq!(begin_json["prompt"]["phrase_source"], "authored");
        // The authored phrase is removed after the prompt said "authored".
        store.mutate_bloom("kn-only", |e| e.wake_phrases.clear());

        let resp = respond(&store, "kn-only", MISS, &token_from_response(&begin_json));
        assert_eq!(
            resp["bloom"]["phrase_source"], "authored",
            "the reveal names the step as the row does"
        );
        assert_eq!(store.guesses.borrow()[0].phrase_source, "authored");
        assert_eq!(resp["summary"]["buckets"]["revealed"]["authored"], 1);
        assert_eq!(resp["summary"]["buckets"]["revealed"]["auto"], 0);
    }

    #[test]
    fn a_guess_about_an_entry_never_prompted_for_is_refused() {
        // The caller is handed kn-gone, which is deleted. Answering about
        // kn-last instead would log a title nobody was shown.
        let store = MockStore::new();
        seed_ids(
            &store,
            &[
                ("kn-first", "alpha"),
                ("kn-gone", "bravo"),
                ("kn-last", "charlie"),
            ],
        );
        let blooms: Vec<KnowledgeEntry> = ["kn-first", "kn-gone", "kn-last"]
            .iter()
            .map(|id| store.blooms.borrow()[*id].clone())
            .collect();
        let begin_json = begin(&store, &test_cascade(blooms));
        let resp = respond(
            &store,
            "kn-first",
            "alpha",
            &token_from_response(&begin_json),
        );
        let token = token_from_response(&resp);
        store.blooms.borrow_mut().remove("kn-gone");

        let refusal = reject(&store, "kn-last", "charlie", &token);
        assert_eq!(refusal["error"], "invalid_bloom_id");
        assert_eq!(refusal["expected_id"], "kn-gone");
        assert_eq!(store.guesses.borrow().len(), 1, "nothing is logged");
        assert_eq!(only_session(&store).step, 1, "nothing is advanced");

        // Following the refusal walks on.
        let resp = respond(&store, "kn-gone", "bravo", &token);
        assert_eq!(resp["status"], "bloom_missing");
        assert_eq!(resp["next"]["id"], "kn-last");
    }

    #[test]
    fn naming_a_deleted_entry_that_was_never_prompted_for_is_refused() {
        // kn-2 is prompted; kn-2 and kn-3 are deleted. kn-3 was never handed
        // out, so naming it is a wrong id, the same as naming kn-4.
        let store = MockStore::new();
        let blooms = seed_ids(
            &store,
            &[
                ("kn-1", "alpha"),
                ("kn-2", "bravo"),
                ("kn-3", "charlie"),
                ("kn-4", "delta"),
            ],
        );
        let b = begin(&store, &test_cascade(blooms));
        let r = respond(&store, "kn-1", "alpha", &token_from_response(&b));
        let t = token_from_response(&r);
        store.blooms.borrow_mut().remove("kn-2");
        store.blooms.borrow_mut().remove("kn-3");

        let refusal = reject(&store, "kn-3", "charlie", &t);
        assert_eq!(refusal["error"], "invalid_bloom_id");
        assert_eq!(refusal["expected_id"], "kn-2");
        assert_eq!(store.guesses.borrow().len(), 1, "nothing is logged");
        assert_eq!(only_session(&store).step, 1, "nothing is advanced");

        let missing = respond(&store, "kn-2", "bravo", &t);
        assert_eq!(missing["status"], "bloom_missing");
        assert_eq!(missing["next"]["id"], "kn-4");
    }

    #[test]
    fn a_wrong_id_after_the_rest_of_the_ritual_is_deleted_is_pointed_at_the_prompt() {
        // kn-2 is prompted, then it and everything after it is deleted. The
        // sweep runs the sequence out, but the refusal still names the entry
        // the caller was handed, and following it finishes the ritual.
        let store = MockStore::new();
        let blooms = seed_ids(
            &store,
            &[("kn-1", "alpha"), ("kn-2", "bravo"), ("kn-3", "charlie")],
        );
        let b = begin(&store, &test_cascade(blooms));
        let r = respond(&store, "kn-1", "alpha", &token_from_response(&b));
        let t = token_from_response(&r);
        store.blooms.borrow_mut().remove("kn-2");
        store.blooms.borrow_mut().remove("kn-3");

        let refusal = reject(&store, "kn-elsewhere", "zulu", &t);
        assert_eq!(refusal["error"], "invalid_bloom_id");
        assert_eq!(refusal["expected_id"], "kn-2");
        assert_eq!(store.guesses.borrow().len(), 1, "nothing is logged");
        let session = only_session(&store);
        assert_eq!(session.step, 1, "nothing is advanced");
        assert!(!session.is_complete());

        let done = respond(&store, "kn-2", "bravo", &t);
        assert_eq!(done["status"], "bloom_missing");
        assert!(done["next"].is_null(), "{done}");
        assert!(done["summary"].is_object());
        assert!(only_session(&store).is_complete());
    }

    // =====================================================================
    // Progress and summary
    // =====================================================================

    #[test]
    fn the_summary_reports_bucket_counts_split_by_phrase_source() {
        let store = MockStore::new();
        let mut authored = entry_with_phrases(vec!["alpha"]);
        authored.id = "kn-authored".to_string();
        let mut auto = test_entry(); // no phrases at all
        auto.id = "kn-auto".to_string();
        auto.body = Some("## The Discovery\n\nBody text here.".to_string());
        store.seed(&authored);
        store.seed(&auto);

        let begin_json = begin(&store, &test_cascade(vec![authored, auto]));
        assert!(
            begin_json["progress"].get("buckets").is_none(),
            "nothing is guessed at begin time"
        );

        let mut token = token_from_response(&begin_json);
        let resp = respond(&store, "kn-authored", "alpha", &token);
        assert_eq!(resp["progress"]["buckets"]["unhinted"], 1);
        assert_eq!(resp["progress"]["buckets"]["revealed"], 0);
        token = token_from_response(&resp);

        let resp = respond(&store, "kn-auto", MISS, &token);
        let summary = &resp["summary"];
        assert_eq!(summary["chunks"], 2);
        assert_eq!(summary["blooms"], 2);
        assert_eq!(summary["buckets"]["unhinted"]["authored"], 1);
        assert_eq!(summary["buckets"]["unhinted"]["derived"], 0);
        assert_eq!(summary["buckets"]["unhinted"]["auto"], 0);
        assert_eq!(summary["buckets"]["revealed"]["auto"], 1);
        assert_eq!(summary["buckets"]["revealed"]["authored"], 0);
        assert!(
            resp.get("next").is_none(),
            "the last respond has no next prompt"
        );
        let sessions = store.sessions.borrow();
        let session = sessions
            .values()
            .next()
            .expect("a completed session is kept");
        assert!(session.is_complete());
        assert!(
            session.completed_at.is_some(),
            "a completed session is stamped, not deleted"
        );
    }

    #[test]
    fn the_begin_response_reports_excluded_entries_per_tag() {
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        store.seed(&bloom);
        let mut cascade = test_cascade(vec![bloom]);
        cascade.excluded.insert("archive".to_string(), 2);
        cascade.excluded.insert("wake-exclude".to_string(), 1);

        let begin_json = begin(&store, &cascade);
        assert_eq!(begin_json["excluded"]["archive"], 2);
        assert_eq!(begin_json["excluded"]["wake-exclude"], 1);
    }

    #[test]
    fn a_begin_with_nothing_excluded_reports_an_empty_object() {
        // The key is always present, so a consumer reads one shape either way
        // and never has to tell an absent key from a zero count.
        let store = MockStore::new();
        let bloom = entry_with_phrases(vec!["alpha"]);
        store.seed(&bloom);
        let begin_json = begin(&store, &test_cascade(vec![bloom]));
        assert_eq!(
            begin_json["excluded"],
            serde_json::json!({}),
            "excluded must be present and empty: {begin_json}"
        );
    }

    // =====================================================================
    // Minimal in-memory KnowledgeStore for the tests above. Implements the
    // methods the ritual actually calls; every other trait method is
    // `unreachable!()` because the ritual never touches them.
    // =====================================================================

    use mock_store::MockStore;

    mod mock_store {
        use std::cell::{Cell, RefCell};
        use std::collections::HashMap;

        use anyhow::Result;

        use crate::knowledge::KnowledgeEntry;
        use crate::store::{
            AgentContext, EditResult, KnowledgeFilter, KnowledgeStore, ReinforcementResult,
            WakeCascade,
        };
        use crate::types::{
            Agent, ApplicabilityType, Category, ContentType, EntryType, MemoryBackup, Project,
            Relationship, RelationshipType, Session, SessionType, SourceType,
        };
        use crate::wake_guess::{WakeGuessRow, WakeHistory};
        use crate::wake_token::WakeSession;

        pub struct MockStore {
            pub blooms: RefCell<HashMap<String, KnowledgeEntry>>,
            pub sessions: RefCell<HashMap<String, WakeSession>>,
            pub guesses: RefCell<Vec<WakeGuessRow>>,
            /// Simulates a guess log that refuses writes.
            pub fail_guess_write: Cell<bool>,
            /// A competing call on the same token that lands its row and
            /// session advance just before the next `record_wake_guess`.
            pub race_winner: RefCell<Option<(WakeGuessRow, WakeSession)>>,
        }

        impl MockStore {
            pub fn new() -> Self {
                Self {
                    blooms: RefCell::new(HashMap::new()),
                    sessions: RefCell::new(HashMap::new()),
                    guesses: RefCell::new(Vec::new()),
                    fail_guess_write: Cell::new(false),
                    race_winner: RefCell::new(None),
                }
            }

            pub fn seed(&self, entry: &KnowledgeEntry) {
                self.blooms
                    .borrow_mut()
                    .insert(entry.id.clone(), entry.clone());
            }

            fn check_step(&self, session: &WakeSession, expected_step: u32) -> Result<()> {
                match self.sessions.borrow().get(&session.session_id) {
                    Some(stored) if stored.step == expected_step => Ok(()),
                    _ => anyhow::bail!(
                        "wake session is no longer at step {expected_step}; another call advanced it"
                    ),
                }
            }

            fn cas_session(&self, session: &WakeSession, expected_step: u32) -> Result<()> {
                self.check_step(session, expected_step)?;
                let mut stored = session.clone();
                stored.completed_at = session
                    .is_complete()
                    .then(|| chrono::Utc::now().timestamp());
                stored.prev_step = Some(expected_step);
                self.sessions
                    .borrow_mut()
                    .insert(session.session_id.clone(), stored);
                Ok(())
            }

            /// Replace a bloom in place — simulates a mid-ritual content edit.
            pub fn mutate_bloom(&self, id: &str, mutate: impl FnOnce(&mut KnowledgeEntry)) {
                let mut blooms = self.blooms.borrow_mut();
                let entry = blooms.get_mut(id).expect("bloom to mutate must exist");
                mutate(entry);
            }
        }

        impl KnowledgeStore for MockStore {
            fn get(&self, id: &str, _ctx: &AgentContext) -> Result<Option<KnowledgeEntry>> {
                Ok(self.blooms.borrow().get(id).cloned())
            }

            fn create_wake_session(&self, session: &WakeSession) -> Result<String> {
                self.sessions
                    .borrow_mut()
                    .insert(session.session_id.clone(), session.clone());
                Ok(session.session_id.clone())
            }

            fn get_wake_session(&self, session_id: &str) -> Result<Option<WakeSession>> {
                Ok(self.sessions.borrow().get(session_id).cloned())
            }

            fn update_wake_session(&self, session: &WakeSession, expected_step: u32) -> Result<()> {
                self.cas_session(session, expected_step)
            }

            /// Atomic like the real store: every check runs before anything is
            /// written, so a refused call leaves no trace.
            fn record_wake_guess(
                &self,
                row: &WakeGuessRow,
                session: &WakeSession,
                expected_step: u32,
            ) -> Result<()> {
                if self.fail_guess_write.get() {
                    anyhow::bail!("guess log unavailable");
                }
                if let Some((winner_row, winner_session)) = self.race_winner.borrow_mut().take() {
                    self.guesses.borrow_mut().push(winner_row);
                    self.cas_session(&winner_session, expected_step)?;
                }
                if self
                    .get_wake_guess(&row.session_id, row.position)?
                    .is_some()
                {
                    anyhow::bail!(
                        "unique index wake_guess_step: ({}, {}) already logged",
                        row.session_id,
                        row.position
                    );
                }
                self.check_step(session, expected_step)?;
                self.guesses.borrow_mut().push(row.clone());
                self.cas_session(session, expected_step)
            }

            fn get_wake_guess(
                &self,
                session_id: &str,
                position: u32,
            ) -> Result<Option<WakeGuessRow>> {
                Ok(self
                    .guesses
                    .borrow()
                    .iter()
                    .find(|r| r.session_id == session_id && r.position == position)
                    .cloned())
            }

            fn wake_history(&self, agent: &str, wake: i64) -> Result<WakeHistory> {
                let guesses = self.guesses.borrow();
                let mine = guesses.iter().filter(|r| r.agent == agent);
                let mut sessions_at_wake: Vec<String> = mine
                    .clone()
                    .filter(|r| r.wake == Some(wake))
                    .map(|r| r.session_id.clone())
                    .collect();
                sessions_at_wake.sort();
                sessions_at_wake.dedup();
                Ok(WakeHistory {
                    sessions_at_wake,
                    highest_wake: mine.filter_map(|r| r.wake).max(),
                })
            }

            // ---- unreachable methods (not used by wake_ritual flow) ----

            fn upsert_knowledge(&self, _entry: &KnowledgeEntry) -> Result<()> {
                unreachable!("wake ritual does not write blooms")
            }
            fn delete(&self, _id: &str, _ctx: &AgentContext) -> Result<bool> {
                unreachable!()
            }
            fn search(
                &self,
                _q: &str,
                _ctx: &AgentContext,
                _f: &KnowledgeFilter,
            ) -> Result<Vec<KnowledgeEntry>> {
                unreachable!()
            }
            fn semantic_search(
                &self,
                _emb: &[f32],
                _ctx: &AgentContext,
                _f: &KnowledgeFilter,
                _l: usize,
            ) -> Result<Vec<KnowledgeEntry>> {
                unreachable!()
            }
            fn semantic_search_scored(
                &self,
                _emb: &[f32],
                _ctx: &AgentContext,
                _f: &KnowledgeFilter,
                _l: usize,
            ) -> Result<Vec<(KnowledgeEntry, f32)>> {
                unreachable!()
            }
            fn semantic_search_entries_scored(
                &self,
                _emb: &[f32],
                _ctx: &AgentContext,
                _l: usize,
            ) -> Result<Vec<(KnowledgeEntry, f32)>> {
                unreachable!()
            }
            fn list_by_category(
                &self,
                _c: &str,
                _ctx: &AgentContext,
                _f: &KnowledgeFilter,
            ) -> Result<Vec<KnowledgeEntry>> {
                unreachable!()
            }
            fn count_by_category(
                &self,
                _c: &str,
                _ctx: &AgentContext,
                _f: &KnowledgeFilter,
            ) -> Result<usize> {
                unreachable!()
            }
            fn owned_private_matching(
                &self,
                _agent: &str,
                _query: Option<&str>,
                _filter: &KnowledgeFilter,
            ) -> Result<Vec<KnowledgeEntry>> {
                // N2: unreachable here by design — this hint query (Issue #400) is
                // only ever invoked from the List/Search handlers, never from the
                // wake ritual this MockStore exercises. The hint-count behavior is
                // covered by the SurrealDatabase-backed tests in helpers.rs.
                unreachable!()
            }
            fn list_all(&self, _ctx: &AgentContext) -> Result<Vec<KnowledgeEntry>> {
                unreachable!()
            }
            fn count(&self) -> Result<usize> {
                unreachable!()
            }
            fn wake_cascade(
                &self,
                _ctx: &AgentContext,
                _l: usize,
                _r: Option<i32>,
                _d: i64,
                _include_excluded: bool,
            ) -> Result<WakeCascade> {
                unreachable!()
            }
            fn update_activations(&self, _ids: &[String]) -> Result<()> {
                unreachable!()
            }
            fn update_summary(&self, _id: &str, _s: &str, _ctx: &AgentContext) -> Result<bool> {
                unreachable!()
            }
            fn apply_update(
                &self,
                _id: &str,
                _spec: &crate::store_update::UpdateSpec,
                _ctx: &AgentContext,
            ) -> Result<crate::store_update::UpdateOutcome> {
                unreachable!()
            }
            fn increment_activation_count(&self, _ids: &[String]) -> Result<()> {
                unreachable!()
            }
            fn query_recent_facts(&self, _d: i32) -> Result<Vec<KnowledgeEntry>> {
                unreachable!()
            }
            fn query_recent_facts_all_types(&self, _d: i32) -> Result<Vec<KnowledgeEntry>> {
                unreachable!()
            }
            fn reinforce(
                &self,
                _id: &str,
                _a: i32,
                _c: Option<i32>,
                _ctx: &AgentContext,
            ) -> Result<Option<ReinforcementResult>> {
                unreachable!()
            }
            fn edit_content(
                &self,
                _id: &str,
                _ctx: &AgentContext,
                _o: &str,
                _n: &str,
                _r: bool,
                _nth: Option<usize>,
            ) -> Result<EditResult> {
                unreachable!()
            }
            fn append_content(&self, _id: &str, _ctx: &AgentContext, _c: &str) -> Result<()> {
                unreachable!()
            }
            fn prepend_content(&self, _id: &str, _ctx: &AgentContext, _c: &str) -> Result<()> {
                unreachable!()
            }
            fn backup_content(
                &self,
                _e: &KnowledgeEntry,
                _o: &str,
                _a: Option<&str>,
            ) -> Result<String> {
                unreachable!()
            }
            fn list_backups(&self, _id: &str) -> Result<Vec<MemoryBackup>> {
                unreachable!()
            }
            fn latest_backup(&self, _id: &str) -> Result<Option<MemoryBackup>> {
                unreachable!()
            }
            fn purge_backups(&self, _id: &str, _k: usize) -> Result<()> {
                unreachable!()
            }
            fn get_tags_for_entry(&self, _id: &str) -> Result<Vec<String>> {
                unreachable!()
            }
            fn set_tags_for_entry(&self, _id: &str, _t: &[String]) -> Result<()> {
                unreachable!()
            }
            fn list_all_tags(&self, _c: Option<&str>) -> Result<Vec<String>> {
                unreachable!()
            }
            fn get_applicability_for_entry(&self, _id: &str) -> Result<Vec<String>> {
                unreachable!()
            }
            fn set_applicability_for_entry(&self, _id: &str, _ids: &[String]) -> Result<()> {
                unreachable!()
            }
            fn list_applicability_types(&self) -> Result<Vec<ApplicabilityType>> {
                unreachable!()
            }
            fn upsert_applicability_type(&self, _a: &ApplicabilityType) -> Result<()> {
                unreachable!()
            }
            fn list_categories(&self) -> Result<Vec<Category>> {
                unreachable!()
            }
            fn get_category(&self, _id: &str) -> Result<Option<Category>> {
                unreachable!()
            }
            fn upsert_category(&self, _c: &Category) -> Result<()> {
                unreachable!()
            }
            fn delete_category(&self, _id: &str) -> Result<bool> {
                unreachable!()
            }
            fn list_projects(&self, _a: bool) -> Result<Vec<Project>> {
                unreachable!()
            }
            fn get_project(&self, _id: &str) -> Result<Option<Project>> {
                unreachable!()
            }
            fn upsert_project(&self, _p: &Project) -> Result<()> {
                unreachable!()
            }
            fn get_tags_for_project(&self, _id: &str) -> Result<Vec<String>> {
                unreachable!()
            }
            fn set_tags_for_project(&self, _id: &str, _t: &[String]) -> Result<()> {
                unreachable!()
            }
            fn get_applicability_for_project(&self, _id: &str) -> Result<Vec<String>> {
                unreachable!()
            }
            fn set_applicability_for_project(&self, _id: &str, _ids: &[String]) -> Result<()> {
                unreachable!()
            }
            fn list_agents(&self) -> Result<Vec<Agent>> {
                unreachable!()
            }
            fn get_agent(&self, _id: &str) -> Result<Option<Agent>> {
                unreachable!()
            }
            fn upsert_agent(&self, _a: &Agent) -> Result<()> {
                unreachable!()
            }
            fn list_relationships_for_entry(&self, _id: &str) -> Result<Vec<Relationship>> {
                unreachable!()
            }
            fn add_relationship(&self, _f: &str, _t: &str, _r: &str) -> Result<String> {
                unreachable!()
            }
            fn delete_relationship(&self, _id: &str) -> Result<bool> {
                unreachable!()
            }
            fn get_facts_for_session(&self, _id: &str) -> Result<Vec<String>> {
                unreachable!()
            }
            fn get_entries_for_session(
                &self,
                _id: &str,
                _owner: Option<&str>,
                _category: &str,
                _ctx: &AgentContext,
            ) -> Result<Vec<crate::store::DedupCandidate>> {
                unreachable!()
            }
            fn get_session_for_fact(&self, _id: &str) -> Result<Option<String>> {
                unreachable!()
            }
            fn list_sessions(&self, _p: Option<&str>) -> Result<Vec<Session>> {
                unreachable!()
            }
            fn get_session(&self, _id: &str) -> Result<Option<Session>> {
                unreachable!()
            }
            fn upsert_session(&self, _s: &Session) -> Result<()> {
                unreachable!()
            }
            fn list_source_types(&self) -> Result<Vec<SourceType>> {
                unreachable!()
            }
            fn list_entry_types(&self) -> Result<Vec<EntryType>> {
                unreachable!()
            }
            fn list_content_types(&self) -> Result<Vec<ContentType>> {
                unreachable!()
            }
            fn list_session_types(&self) -> Result<Vec<SessionType>> {
                unreachable!()
            }
            fn list_relationship_types(&self) -> Result<Vec<RelationshipType>> {
                unreachable!()
            }
            fn list_tables(&self) -> Result<Vec<String>> {
                unreachable!()
            }

            fn delete_embedding_chunks(&self, _id: &str) -> Result<()> {
                unreachable!()
            }
            fn insert_embedding_chunk(
                &self,
                _id: &str,
                _ci: usize,
                _ct: &str,
                _to: usize,
                _tc: usize,
                _emb: &[f32],
                _m: &str,
            ) -> Result<()> {
                unreachable!()
            }
            fn semantic_search_chunks(
                &self,
                _emb: &[f32],
                _l: usize,
            ) -> Result<Vec<(String, f32)>> {
                unreachable!()
            }

            fn sweep_ghost_anchors(
                &self,
                _dry_run: bool,
            ) -> Result<crate::store::GhostSweepResult> {
                unreachable!()
            }
        }
    }

    // =====================================================================
    // Adversarial cases. Each one asserts the behaviour the spec or the
    // invariant asks for; the ones that fail are defects, not test bugs.
    // Every fixture here is invented.
    // =====================================================================
    mod adversarial {
        use super::mock_store::MockStore;
        use super::*;

        /// A guess with no alphanumeric content is not data: it is the absence
        /// of the thing the ritual exists to collect. Such a guess is refused
        /// with an error, writes no row, and does not advance the session.
        ///
        /// Refusing it also closes a matching hole. `fuzzy_match` strips every
        /// non-alphanumeric character before comparing, so a blank guess and a
        /// punctuation-only phrase both normalize to the empty string and
        /// compare EQUAL — logged as `exact`, bucketed `unhinted`.
        ///
        /// Ruled during review: refuse, no row, no advance.
        #[test]
        fn a_guess_that_normalizes_to_nothing_is_refused() {
            // The empty string, whitespace, and punctuation that survives a
            // trim but carries no content.
            for guess in ["", "   ", "\t\n", "\u{2014}", "...", "???"] {
                let store = MockStore::new();
                // An em-dash phrase normalizes to nothing too, which is what
                // makes the blank guess an `exact` match today.
                let bloom = entry_with_phrases(vec!["\u{2014}"]);
                let bloom_id = bloom.id.clone();
                store.seed(&bloom);

                let begin_json = begin(&store, &test_cascade(vec![bloom]));
                let out = respond_ritual(
                    &store,
                    &AgentContext::for_agent("test-agent"),
                    &bloom_id,
                    guess,
                    &token_from_response(&begin_json),
                );

                assert!(
                    out.is_err(),
                    "guess {guess:?} was accepted instead of refused: {out:?}"
                );
                assert!(
                    store.guesses.borrow().is_empty(),
                    "guess {guess:?} reached the log"
                );

                let sessions = store.sessions.borrow();
                let session = sessions.values().next().expect("session must survive");
                assert_eq!(
                    session.step, 0,
                    "guess {guess:?} advanced the session it was refused from"
                );
                assert_eq!(session.current_index, 0, "guess {guess:?}");
            }
        }

        /// `fuzzy_match` counts edit distance in CHARACTERS but divides by
        /// `str::len()`, which is BYTES. For multibyte text the denominator is
        /// inflated 3-4x, so the 0.8 tolerance silently widens to accept a
        /// guess that is half wrong. Identical edit ratios must get identical
        /// verdicts regardless of how the text happens to encode.
        #[test]
        fn the_match_tolerance_does_not_widen_for_multibyte_text() {
            // Control: 5 of 10 ASCII characters differ -> not a match.
            let ascii = vec!["abcdefghij".to_string()];
            assert_eq!(
                best_match("abcdeqrstu", &ascii),
                (MatchKind::None, None),
                "precondition: half-wrong ASCII is not a match"
            );

            // Same shape in hiragana: 5 of 10 characters differ.
            let kana = vec![
                "\u{3042}\u{3044}\u{3046}\u{3048}\u{304A}\u{304B}\u{304D}\u{304F}\u{3051}\u{3053}"
                    .to_string(),
            ];
            let half_wrong =
                "\u{3042}\u{3044}\u{3046}\u{3048}\u{304A}\u{3055}\u{3057}\u{3059}\u{305B}\u{305D}";
            assert_eq!(
                best_match(half_wrong, &kana),
                (MatchKind::None, None),
                "a half-wrong multibyte guess was accepted as a match"
            );
        }

        /// Deleting an entry that the ritual has already walked past must not
        /// break the rest of the ritual. `fetch_blooms_by_ids` re-fetches every
        /// id in the session on every respond and hard-fails on the first one
        /// that is gone, so removing an already-shown bloom bricks the session
        /// with no way to finish it and no way to clear it.
        #[test]
        fn deleting_an_already_shown_bloom_does_not_brick_the_ritual() {
            let store = MockStore::new();
            let mut first = entry_with_phrases(vec!["alpha"]);
            first.id = "kn-first".to_string();
            let mut second = entry_with_phrases(vec!["bravo"]);
            second.id = "kn-second".to_string();
            store.seed(&first);
            store.seed(&second);

            let begin_json = begin(&store, &test_cascade(vec![first, second]));
            let resp = respond(
                &store,
                "kn-first",
                "alpha",
                &token_from_response(&begin_json),
            );
            let token = token_from_response(&resp);

            // The first bloom is done with. Remove it, as a `memory delete`
            // from another shell would.
            store.blooms.borrow_mut().remove("kn-first");

            let out = respond_ritual(
                &store,
                &AgentContext::for_agent("test-agent"),
                "kn-second",
                "bravo",
                &token,
            );
            assert!(
                out.is_ok(),
                "a deleted, already-shown bloom killed the rest of the ritual: {:?}",
                out.err()
            );
        }

        /// A `chunk_truncated` response must retire the token it was handed.
        /// The truncation path advances the bloom cursor but never ticks
        /// `step`, so the token it returns is byte-identical to the one just
        /// spent and that spent token still verifies against the session.
        #[test]
        fn a_chunk_truncated_response_retires_the_token_it_consumed() {
            let store = MockStore::new();
            let big = make_large_bloom(95_000, vec!["alpha", "beta", "gamma"]);
            let big_id = big.id.clone();
            let mut tail = entry_with_phrases(vec!["bravo"]);
            tail.id = "kn-tail".to_string();
            tail.title = "Tail".to_string();
            store.seed(&big);
            store.seed(&tail);

            let begin_json = begin(&store, &test_cascade(vec![big, tail]));
            let mut token = token_from_response(&begin_json);
            for _ in 0..2 {
                let resp = respond(&store, &big_id, MISS, &token);
                token = token_from_response(&resp);
            }

            store.mutate_bloom(&big_id, |entry| {
                entry.body = Some("shrunk down to a single tiny chunk now.".to_string());
            });

            let spent = token.clone();
            let resp = respond(&store, &big_id, "ignored", &spent);
            assert_eq!(resp["status"], "chunk_truncated");
            assert_ne!(
                token_from_response(&resp),
                spent,
                "a truncation handed back the very token it consumed"
            );
        }

        /// The token authorises the session, not the caller, and nothing
        /// checks that the responder is the agent that began the ritual. A
        /// foreign agent can drive someone else's ritual to completion, and
        /// every row it writes is stamped with the OWNER's agent id — so the
        /// owner's guess log silently fills with guesses they never made.
        #[test]
        fn a_respond_from_another_agent_is_refused() {
            let store = MockStore::new();
            let bloom = entry_with_phrases(vec!["alpha"]);
            let bloom_id = bloom.id.clone();
            store.seed(&bloom);

            // `meta()` begins the ritual as "test-agent".
            let begin_json = begin(&store, &test_cascade(vec![bloom]));
            let out = respond_ritual(
                &store,
                &AgentContext::for_agent("some-other-agent"),
                &bloom_id,
                "alpha",
                &token_from_response(&begin_json),
            );

            assert!(out.is_err(), "another agent walked this session: {out:?}");
            assert!(
                store.guesses.borrow().is_empty(),
                "a foreign respond wrote a row stamped with the owner's agent: {:?}",
                store.guesses.borrow().first().map(|r| r.agent.clone())
            );
        }

        /// A session whose `agent` is empty cannot be one this binary began:
        /// `--begin` always stamps `MX_CURRENT_AGENT`. It is a session written
        /// by the pre-one-guess binary, whose row has no `agent` field at all
        /// and which the loader silently defaults to "". Spec 9.1 says such a
        /// session is treated as expired and the caller is told to run
        /// `--begin`. Instead it walks, and every guess row it writes carries
        /// `agent: ""` — unattributable, and invisible to every query in the
        /// log, all of which filter on the agent.
        #[test]
        fn a_session_from_an_older_binary_is_refused_rather_than_walked() {
            let store = MockStore::new();
            let bloom = entry_with_phrases(vec!["alpha"]);
            let bloom_id = bloom.id.clone();
            store.seed(&bloom);

            let mut legacy =
                WakeSession::new(&test_cascade(vec![bloom]), String::new(), None, None);
            legacy.agent = String::new();
            let session_id = legacy.session_id.clone();
            store.create_wake_session(&legacy).unwrap();
            let token = create_token(&session_id, legacy.step);

            let out = respond_ritual(
                &store,
                &AgentContext::for_agent("test-agent"),
                &bloom_id,
                "alpha",
                &token,
            );

            let refused = match out {
                Err(ref e) => e.to_string().contains("--begin"),
                Ok(_) => false,
            };
            assert!(
                refused,
                "an agent-less session was walked instead of refused: {out:?}"
            );
            assert!(
                store
                    .guesses
                    .borrow()
                    .iter()
                    .all(|row| !row.agent.is_empty()),
                "a guess row was written with an empty agent"
            );
        }

        /// `progress.current` is `step + 1` — "the chunk we are on now" — but
        /// after the LAST advance there is no chunk we are on, so the final
        /// response of every ritual reports a position one past the end:
        /// chunk 2 of 2 becomes `current: 3, total: 2`. This is in the payload
        /// the model reads, next to the summary.
        #[test]
        fn the_final_response_does_not_report_a_position_past_the_end() {
            let store = MockStore::new();
            let mut first = entry_with_phrases(vec!["alpha"]);
            first.id = "kn-first".to_string();
            let mut second = entry_with_phrases(vec!["bravo"]);
            second.id = "kn-second".to_string();
            store.seed(&first);
            store.seed(&second);

            let begin_json = begin(&store, &test_cascade(vec![first, second]));
            let mut token = token_from_response(&begin_json);
            let mut last = serde_json::Value::Null;
            for (id, guess) in [("kn-first", "alpha"), ("kn-second", "bravo")] {
                last = respond(&store, id, guess, &token);
                token = token_from_response(&last);
            }

            let current = last["progress"]["current"].as_u64().unwrap();
            let total = last["progress"]["total"].as_u64().unwrap();
            assert!(
                current <= total,
                "final progress reads {current} of {total}: {}",
                last["progress"]
            );
        }

        /// The guess is model output written verbatim into the database with
        /// no bound anywhere on the path. Ruled during review: a guess longer
        /// than 2000 CHARACTERS is REFUSED — error, no row, no advance — not
        /// truncated. A truncated guess is not what the model guessed, and the
        /// log exists to be honest about guesses; better nothing than a row
        /// that misstates one.
        ///
        /// The limit counts characters, so both boundaries are driven in two
        /// encodings. A byte-based check would wrongly refuse the 2000-kana
        /// guess (6000 bytes) that must be accepted, and a byte-based
        /// truncation would panic on a boundary landing mid-character.
        #[test]
        fn a_guess_over_two_thousand_characters_is_refused() {
            const LIMIT: usize = 2_000;

            // A fresh session per case: an accepted guess advances one.
            fn fixture(store: &MockStore) -> (String, String) {
                let bloom = entry_with_phrases(vec!["alpha"]);
                let bloom_id = bloom.id.clone();
                store.seed(&bloom);
                let begin_json = begin(store, &test_cascade(vec![bloom]));
                (bloom_id, token_from_response(&begin_json))
            }

            // ASCII, and a 3-byte character. Both boundaries are driven in
            // both encodings, and the two directions run as separate loops so
            // a failure in one does not hide the other's cases.

            // ---- exactly at the limit: accepted, and logged WHOLE ----
            for filler in ["a", "\u{3042}"] {
                let store = MockStore::new();
                let (bloom_id, token) = fixture(&store);
                let at_limit = filler.repeat(LIMIT);
                let out = respond_ritual(
                    &store,
                    &AgentContext::for_agent("test-agent"),
                    &bloom_id,
                    &at_limit,
                    &token,
                );
                assert!(
                    out.is_ok(),
                    "a guess of exactly {LIMIT} characters must be accepted \
                     (filler {filler:?}): {out:?}"
                );
                let rows = store.guesses.borrow();
                let row = rows.first().expect("an accepted guess is logged");
                assert_eq!(
                    row.guess, at_limit,
                    "a guess at the limit must be logged whole, not trimmed \
                     (filler {filler:?})"
                );
            }

            // ---- one character over: refused, no row, no advance ----
            for filler in ["a", "\u{3042}"] {
                let store = MockStore::new();
                let (bloom_id, token) = fixture(&store);
                let over = filler.repeat(LIMIT + 1);
                let out = respond_ritual(
                    &store,
                    &AgentContext::for_agent("test-agent"),
                    &bloom_id,
                    &over,
                    &token,
                );
                assert!(
                    out.is_err(),
                    "a guess of {} characters was accepted (filler {filler:?}): {out:?}",
                    LIMIT + 1
                );
                assert!(
                    store.guesses.borrow().is_empty(),
                    "an over-long guess reached the log (filler {filler:?})"
                );

                let sessions = store.sessions.borrow();
                let session = sessions.values().next().expect("session must survive");
                assert_eq!(
                    session.step, 0,
                    "an over-long guess advanced the session it was refused from \
                     (filler {filler:?})"
                );
                assert_eq!(session.current_index, 0, "filler {filler:?}");
            }
        }

        /// Spec 9.1 reasons that "sessions live for one ritual, so at most one
        /// is affected". Nothing enforces that: `created_at` is stored and then
        /// never read, so a token minted a year ago still drives its session.
        #[test]
        #[ignore = "session lifecycle is follow-up work, not part of this change"]
        fn a_stale_session_is_expired_by_age() {
            let store = MockStore::new();
            let bloom = entry_with_phrases(vec!["alpha"]);
            let bloom_id = bloom.id.clone();
            store.seed(&bloom);

            let begin_json = begin(&store, &test_cascade(vec![bloom]));
            let token = token_from_response(&begin_json);

            // Backdate the session a year.
            {
                let mut sessions = store.sessions.borrow_mut();
                let session = sessions.values_mut().next().expect("session exists");
                session.created_at -= 365 * 24 * 60 * 60;
            }

            let out = respond_ritual(
                &store,
                &AgentContext::for_agent("test-agent"),
                &bloom_id,
                "alpha",
                &token,
            );
            assert!(
                out.is_err(),
                "a year-old session token still walked the ritual: {out:?}"
            );
        }

        /// A second `--begin` abandons the first session's row. Nothing deletes
        /// it and nothing sweeps by age, so every interrupted ritual leaves a
        /// permanent row behind — each one carrying a bloom id list.
        #[test]
        #[ignore = "session lifecycle is follow-up work, not part of this change"]
        fn beginning_again_does_not_orphan_the_previous_session() {
            let store = MockStore::new();
            let bloom = entry_with_phrases(vec!["alpha"]);
            store.seed(&bloom);

            begin(&store, &test_cascade(vec![bloom.clone()]));
            begin(&store, &test_cascade(vec![bloom]));

            assert_eq!(
                store.sessions.borrow().len(),
                1,
                "a second begin left the first ritual's session behind"
            );
        }

        // =================================================================
        // Round two: the code added to close round one.
        // =================================================================

        /// The ownership guard refuses a caller naming a DIFFERENT agent but
        /// waves through a caller naming NO agent, on the reasoning that an
        /// anonymous caller cannot be impersonating anyone and the CLI always
        /// names one.
        ///
        /// Anonymity is not innocence here. The guard's job is to establish
        /// that the caller IS the owner, and `None` fails to establish that
        /// just as surely as a mismatch does. The exemption was needed because
        /// this file's own tests drove respond anonymously; they no longer do,
        /// so nothing depends on the hole staying open.
        ///
        /// It matters because of what an anonymous context does further down:
        /// entries it cannot SEE come back from `fetch_blooms_by_ids` as
        /// absent, and absent is now silently stepped over as `bloom_missing`.
        /// An anonymous caller therefore does not merely walk another agent's
        /// ritual — it walks it while skipping every entry it lacks the
        /// visibility to read, logging no row for any of them, and the owner's
        /// session is marked complete at the end as though the ritual finished.
        #[test]
        fn a_respond_from_a_caller_with_no_agent_is_refused() {
            let store = MockStore::new();
            let bloom = entry_with_phrases(vec!["alpha"]);
            let bloom_id = bloom.id.clone();
            store.seed(&bloom);

            // `meta()` begins as "test-agent".
            let begin_json = begin(&store, &test_cascade(vec![bloom]));
            let out = respond_ritual(
                &store,
                &AgentContext::public_only(),
                &bloom_id,
                "alpha",
                &token_from_response(&begin_json),
            );

            assert!(
                out.is_err(),
                "an anonymous caller walked an owned ritual: {out:?}"
            );
            assert!(
                store.guesses.borrow().is_empty(),
                "an anonymous respond filed a row under the owner's agent: {:?}",
                store.guesses.borrow().first().map(|r| r.agent.clone())
            );
        }

        /// An entry that cannot be fetched is stepped over and never judged.
        /// That much is unavoidable — there is nothing to put to the responder
        /// — so what this pins is the other half: it must not happen SILENTLY.
        /// The summary has to account for the step, so a reader is never left
        /// to notice the shortfall by subtracting one number from another.
        ///
        /// Whether such an entry was deleted or has merely become unreadable
        /// to this caller is a distinction this path cannot draw. Telling them
        /// apart is open work, and is not what this test claims.
        #[test]
        fn an_unreadable_entry_is_accounted_for_in_the_summary() {
            let store = MockStore::new();
            let mut visible = entry_with_phrases(vec!["alpha"]);
            visible.id = "kn-visible".to_string();
            let mut hidden = entry_with_phrases(vec!["bravo"]);
            hidden.id = "kn-hidden".to_string();
            store.seed(&visible);
            store.seed(&hidden);

            let begin_json = begin(&store, &test_cascade(vec![visible, hidden]));

            // Stand in for "still in the database, but this caller may not
            // read it": the entry is gone from what the fetch can return,
            // while never having been deleted.
            store.blooms.borrow_mut().remove("kn-hidden");

            let resp = respond(
                &store,
                "kn-visible",
                "alpha",
                &token_from_response(&begin_json),
            );

            // An entry the fetch cannot return can never produce a guess row —
            // there is nothing to put to the responder. What the summary must
            // not do is let that pass unremarked, leaving a reader to notice
            // the shortfall by subtracting one number from another.
            let logged = store.guesses.borrow().len();
            assert_eq!(
                resp["summary"]["blooms"], 2,
                "precondition: both entries were in the sequence: {resp}"
            );
            assert_eq!(logged, 1, "only the readable entry could be judged");
            assert_eq!(
                resp["summary"]["unjudged"], 1,
                "an entry vanished from the log and the summary did not say so: \
                 {} of 2 rows written, summary {}",
                logged, resp["summary"]
            );
            assert_eq!(
                resp["summary"]["chunks"].as_u64().unwrap(),
                logged as u64 + resp["summary"]["unjudged"].as_u64().unwrap(),
                "the chunk count must be fully explained by the two halves"
            );
        }

        /// `summary.chunks` now counts steps where nothing was judged, while
        /// the bucket totals do not. The model reads both in one object, so a
        /// summary of 3 chunks over 2 bucketed guesses invites exactly the
        /// "2 out of 3" the vocabulary section retired — and the one number
        /// that would explain the gap is the one not reported.
        ///
        /// Either the two agree, or the summary names the difference. It must
        /// not be left as a subtraction for the reader to perform.
        #[test]
        fn the_summary_does_not_leave_an_unexplained_gap_to_subtract() {
            let store = MockStore::new();
            let mut first = entry_with_phrases(vec!["alpha"]);
            first.id = "kn-first".to_string();
            let mut gone = entry_with_phrases(vec!["bravo"]);
            gone.id = "kn-gone".to_string();
            let mut last = entry_with_phrases(vec!["charlie"]);
            last.id = "kn-last".to_string();
            store.seed(&first);
            store.seed(&gone);
            store.seed(&last);

            let begin_json = begin(&store, &test_cascade(vec![first, gone, last]));
            let mut token = token_from_response(&begin_json);

            let resp = respond(&store, "kn-first", "alpha", &token);
            token = token_from_response(&resp);

            store.blooms.borrow_mut().remove("kn-gone");

            // The caller answers about the entry it was handed, learns it is
            // gone, and is handed the last one.
            let resp = respond(&store, "kn-gone", "bravo", &token);
            assert_eq!(resp["status"], "bloom_missing");
            token = token_from_response(&resp);

            let resp = respond(&store, "kn-last", "charlie", &token);
            let summary = &resp["summary"];

            let chunks = summary["chunks"].as_u64().unwrap();
            let bucketed = ["unhinted", "revealed"]
                .iter()
                .flat_map(|b| {
                    ["authored", "derived", "auto", "unphrased"]
                        .iter()
                        .map(move |s| summary["buckets"][b][s].as_u64().unwrap_or(0))
                })
                .sum::<u64>();

            assert!(
                chunks == bucketed || summary.get("unjudged").is_some(),
                "summary reports {chunks} chunks against {bucketed} bucketed \
                 guesses and does not name the difference: {summary}"
            );
        }

        // =================================================================
        // Round three: the sweep that answers about a vanished entry.
        // =================================================================

        /// Answering a caller about the entry it was just handed is right.
        /// Answering it about an entry it never named is not.
        ///
        /// The completion arm of the sweep gate is `session.is_complete() &&
        /// !vanished.is_empty()`, which does not mention `bloom_id` at all. So
        /// once the sweep consumes the rest of the sequence, ANY id is
        /// answered with `bloom_missing` — including one that was never in the
        /// session. A caller that sends a wrong id at that moment is told the
        /// ritual finished, and the session is completed on the way out, so the
        /// mistake is unrecoverable rather than correctable.
        ///
        /// `invalid_bloom_id` exists precisely to say "that is not what you
        /// were asked about", and it must not stop applying because the
        /// sequence happened to end.
        #[test]
        fn a_bogus_id_is_still_rejected_when_the_sweep_ends_the_ritual() {
            let store = MockStore::new();
            let mut first = entry_with_phrases(vec!["alpha"]);
            first.id = "kn-first".to_string();
            let mut last = entry_with_phrases(vec!["bravo"]);
            last.id = "kn-last".to_string();
            store.seed(&first);
            store.seed(&last);

            let begin_json = begin(&store, &test_cascade(vec![first, last]));
            let resp = respond(
                &store,
                "kn-first",
                "alpha",
                &token_from_response(&begin_json),
            );
            let token = token_from_response(&resp);

            // The entry the caller was just handed goes away.
            store.blooms.borrow_mut().remove("kn-last");

            // The caller names something that was never in this session.
            let resp = reject(&store, "kn-never-in-this-session", "alpha", &token);

            assert_eq!(
                resp["status"], "error",
                "an id that was never in the session must not be answered as \
                 though it were the sequence: {resp}"
            );
            assert_eq!(resp["error"], "invalid_bloom_id");
            let sessions = store.sessions.borrow();
            let session = sessions.values().next().expect("the session survives");
            assert!(
                session.completed_at.is_none() && session.step == 1,
                "a wrong id must not complete the session out from under the caller"
            );
        }

        /// The sweep runs before `validate_guess`? No — it runs after. So a
        /// caller answering about an entry that has vanished is told its GUESS
        /// was unusable, when the truth is there was nothing to guess about.
        ///
        /// Minor on its own, but it means the refusal a caller sees does not
        /// describe the situation it is in, and a client that retries on a
        /// guess error will retry forever against an entry that is gone.
        #[test]
        fn a_vanished_entry_is_reported_before_the_guess_is_judged_unusable() {
            let store = MockStore::new();
            let mut first = entry_with_phrases(vec!["alpha"]);
            first.id = "kn-first".to_string();
            let mut last = entry_with_phrases(vec!["bravo"]);
            last.id = "kn-last".to_string();
            store.seed(&first);
            store.seed(&last);

            let begin_json = begin(&store, &test_cascade(vec![first, last]));
            let resp = respond(
                &store,
                "kn-first",
                "alpha",
                &token_from_response(&begin_json),
            );
            let token = token_from_response(&resp);

            store.blooms.borrow_mut().remove("kn-last");

            // An empty guess for an entry that no longer exists. There is
            // nothing to guess about; the guess is beside the point.
            let out = respond_ritual(
                &store,
                &AgentContext::for_agent("test-agent"),
                "kn-last",
                "",
                &token,
            );

            let reported_as_missing = match out {
                Ok(ref json) => json.contains("\"status\":\"bloom_missing\""),
                Err(_) => false,
            };
            assert!(
                reported_as_missing,
                "a vanished entry was reported as a bad guess instead: {out:?}"
            );
        }

        /// `Summary.unjudged` exists because, in its own words, "deriving it is
        /// how an 'N out of M' score gets reinvented". That reasoning is not
        /// applied to `progress`, which carries the same two numbers and is
        /// read on EVERY respond rather than once at the end.
        ///
        /// After an unjudged step, `progress.current - 1` steps have completed
        /// while `progress.buckets` accounts for fewer, and the difference is
        /// again there to be subtracted — for the whole rest of the ritual, not
        /// just at the close.
        #[test]
        #[ignore = "running progress has the same gap the summary now names; \
                    reporting it there too is follow-up work, not part of this change"]
        fn running_progress_accounts_for_every_completed_step() {
            let store = MockStore::new();
            let mut first = entry_with_phrases(vec!["alpha"]);
            first.id = "kn-first".to_string();
            let mut gone = entry_with_phrases(vec!["bravo"]);
            gone.id = "kn-gone".to_string();
            let mut last = entry_with_phrases(vec!["charlie"]);
            last.id = "kn-last".to_string();
            store.seed(&first);
            store.seed(&gone);
            store.seed(&last);

            let begin_json = begin(&store, &test_cascade(vec![first, gone, last]));
            let resp = respond(
                &store,
                "kn-first",
                "alpha",
                &token_from_response(&begin_json),
            );
            let token = token_from_response(&resp);

            store.blooms.borrow_mut().remove("kn-gone");

            // The step that walks over the deleted entry. The ritual is not
            // finished, so this is `progress` without a `summary` beside it.
            let resp = respond(&store, "kn-gone", "bravo", &token);
            assert_eq!(resp["status"], "bloom_missing", "{resp}");

            let progress = &resp["progress"];
            let completed = progress["current"].as_u64().unwrap() - 1;
            let bucketed = progress["buckets"]["unhinted"].as_u64().unwrap()
                + progress["buckets"]["revealed"].as_u64().unwrap();

            assert!(
                completed == bucketed || progress.get("unjudged").is_some(),
                "progress reports {completed} completed steps against \
                 {bucketed} bucketed guesses and does not name the \
                 difference: {progress}"
            );
        }
    }

    /// Poppy's round-8 probes (mx#469), invented fixtures.
    mod poppy_r8 {
        use super::mock_store::MockStore;
        use super::*;

        const CHUNK0: &str = "## Mill\n\nThe mill sits on a bend.\n";

        fn one_guess(e: &KnowledgeEntry, chunk1: &str, guess: &str) -> serde_json::Value {
            let filler = "let pears = 4;\n".repeat(1300);
            let mut e = e.clone();
            e.body = Some(format!(
                "{CHUNK0}\n```\n{filler}```\n\n{chunk1}\n```\n{filler}```\n"
            ));
            let content = bloom_content(&e);
            let plan = compute_chunks(&content, chunk_threshold());
            assert_eq!(plan.total, 2, "fixture must chunk in two for {chunk1:?}");
            let store = MockStore::new();
            store.seed(&e);
            let b = begin(&store, &test_cascade(vec![e.clone()]));
            let r0 = respond(&store, &e.id, MISS, &token_from_response(&b));
            let r1 = respond(&store, &e.id, guess, &token_from_response(&r0));
            serde_json::json!({"prompt": r0["next"], "r": r1})
        }

        fn mill() -> KnowledgeEntry {
            let mut e = entry_with_phrases(vec!["millstone"]);
            e.id = "kn-mill".into();
            e.title = "Mill".into();
            e
        }

        /// Only code / HTML / a rule / whitespace / letterless -> the prompt
        /// says unphrased, and the guess is revealed.
        #[test]
        fn poppy_r8_contentless_chunk_is_unphrased_and_revealed() {
            let mut failures = Vec::new();
            for (chunk1, guess) in [
                ("---\n\n```\nlet a = 1;\n```\n", "let a = 1;"),
                ("---\n", "rule"),
                ("***\n", "rule"),
                ("---\n\n<div>\n<p>pears</p>\n</div>\n", "<div>"),
                ("---\n\n    let a = 1;\n", "let a = 1;"),
                ("---\n\n   \n", "blank"),
                ("---\n\n...\n", "dots"),
                ("---\n\n\u{2014}\n", "dash"),
                ("---\n\n\u{1F30A}\n", "wave"),
                ("---\n\n\u{200B}\n", "zwsp"),
                ("---\n\n1.\n", "1"),
            ] {
                let j = one_guess(&mill(), chunk1, guess);
                let line = format!(
                    "{chunk1:?} guess {guess:?} -> prompt {} / row {} {} {} vs {}",
                    j["prompt"]["phrase_source"],
                    j["r"]["bloom"]["phrase_source"],
                    j["r"]["bucket"],
                    j["r"]["match"]["kind"],
                    j["r"]["bloom"]["phrases"]
                );
                if j["prompt"]["phrase_source"] != "unphrased" || j["r"]["bucket"] != "revealed" {
                    failures.push(line);
                }
            }
            assert!(failures.is_empty(), "phrased:\n{}", failures.join("\n"));
        }

        /// The prompt for chunk 1 shows the title `Mill (Part 2/2)`. A chunk
        /// phrased by a heading equal to the title is phrased by the prompt.
        #[test]
        fn poppy_r8_heading_equal_to_title_is_not_a_phrase() {
            let j = one_guess(&mill(), "## Mill\n", "Mill");
            assert_eq!(j["prompt"]["title"], "Mill (Part 2/2)");
            assert_eq!(j["prompt"]["phrase_source"], "unphrased");
            assert_eq!(
                j["r"]["bucket"], "revealed",
                "guess copied from the prompt title -> {} vs {}",
                j["r"]["match"]["kind"], j["r"]["bloom"]["phrases"]
            );
        }

        /// The same chunk kinds, cut exactly by `planned` so no leading rule.
        #[test]
        fn poppy_r8_chunk_kinds_unit() {
            let e = entry_with_phrases(vec!["millstone"]);
            let mut failures = Vec::new();
            for (chunk1, guess) in [
                ("...\n", "x"),
                ("\u{2014}\n", "x"),
                ("\u{1F30A}\n", "x"),
                ("\u{200B}\n", "x"),
                ("> ...\n", "x"),
                ("1.\n", "1"),
                ("1.\n", "1)"),
            ] {
                let (content, plan) = planned(&["The mill sits on a bend. Steps 1.\n", chunk1]);
                let one = phrases_for_chunk(&e, &plan, &content, 1);
                let m = best_match(guess, &one.phrases);
                let line = format!(
                    "{chunk1:?} -> {:?} {:?}; guess {guess:?} -> {:?}",
                    one.source, one.phrases, m.0
                );
                if one.source != PhraseSource::Unphrased {
                    failures.push(line);
                }
            }
            assert!(failures.is_empty(), "phrased:\n{}", failures.join("\n"));

            let (content, plan) = planned(&[
                "The mill sits on a bend. Steps 1.\n",
                "Steps 1. 2. Then the river rose.\n",
            ]);
            let one = phrases_for_chunk(&e, &plan, &content, 1);
            assert_eq!(one.source, PhraseSource::Derived);
            assert_eq!(one.phrases, vec!["Steps 1."]);
            assert_eq!(best_match("2", &one.phrases).0, MatchKind::None);
        }
    }
}