dirge-agent 0.21.3

use super::*;
use crate::agent::agent_loop::hooks::{
    AfterToolCallContext, AfterToolCallFn, GetSteeringMessagesFn, PrepareNextTurnFn,
    ShouldStopAfterTurnFn,
};
use crate::agent::agent_loop::message::{StreamEvent, UserMessage};
use crate::agent::agent_loop::result::AfterToolCallResult;
use crate::agent::agent_loop::stream::StreamFn;
use crate::agent::agent_loop::tool::{AbortSignal, LoopTool, LoopToolUpdate};
use crate::agent::agent_loop::types::{
    ConvertToLlmFn, GateMode, LoopConfig, ToolExecutionMode, TurnUpdate,
};
use std::pin::Pin;
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};

/// An empty reusable-checkpoint slot for the compaction-pass tests. With no
/// cached summary the fold always takes the inline summarizer path, so these
/// tests exercise the same behavior they did before Round 1's fast path.
fn empty_checkpoint_slot() -> super::CheckpointSlot {
    std::sync::Arc::new(std::sync::Mutex::new(None))
}

/// Build a stream factory that returns canned assistant
/// messages in sequence. Mirrors pi's typical test mock —
/// `callIndex` increments per invocation; each call returns
/// the next canned response.
///
/// `responses` is a Vec; index N is returned on the (N+1)th
/// call. Past the end → final fallback message with
/// stopReason=Stop.
fn canned_factory(responses: Vec<AssistantMessage>) -> StreamFn {
    let counter = std::sync::Arc::new(AtomicUsize::new(0));
    let responses = std::sync::Arc::new(responses);
    std::sync::Arc::new(move |_ctx, _opts| {
        let n = counter.fetch_add(1, Ordering::SeqCst);
        let msg = responses.get(n).cloned().unwrap_or_else(|| {
            AssistantMessage::new(
                vec![ContentBlock::Text {
                    text: "end".to_string(),
                }],
                StopReason::Stop,
            )
        });
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![StreamEvent::Done {
            reason,
            message: msg,
            usage: None,
        }]))
    })
}

/// Like [`canned_factory`] but records a JSON snapshot of each call's
/// context messages into `seen`, so a test can inspect what the model was
/// actually sent (e.g. a mid-loop memory re-injection).
fn capturing_factory(
    responses: Vec<AssistantMessage>,
    seen: std::sync::Arc<std::sync::Mutex<Vec<String>>>,
) -> StreamFn {
    let counter = std::sync::Arc::new(AtomicUsize::new(0));
    let responses = std::sync::Arc::new(responses);
    std::sync::Arc::new(move |ctx, _opts| {
        seen.lock()
            .unwrap()
            .push(serde_json::to_string(&ctx.messages).unwrap_or_default());
        let n = counter.fetch_add(1, Ordering::SeqCst);
        let msg = responses.get(n).cloned().unwrap_or_else(|| {
            AssistantMessage::new(
                vec![ContentBlock::Text {
                    text: "end".to_string(),
                }],
                StopReason::Stop,
            )
        });
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![StreamEvent::Done {
            reason,
            message: msg,
            usage: None,
        }]))
    })
}

fn identity_converter() -> ConvertToLlmFn {
    std::sync::Arc::new(|messages: &[Value]| {
        messages
            .iter()
            .filter(|m| {
                let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("");
                matches!(role, "user" | "assistant" | "tool" | "toolResult")
            })
            .cloned()
            .collect()
    })
}

fn build_config() -> LoopConfig {
    LoopConfig {
        convert_to_llm: identity_converter(),
        transform_context: None,
        compaction_hooks: None,
        get_api_key: None,
        api_key: None,
        tool_execution: ToolExecutionMode::Sequential,
        before_tool_call: None,
        after_tool_call: None,
        prepare_next_turn: None,
        should_stop_after_turn: None,
        get_steering_messages: None,
        get_followup_messages: None,
        should_defer_finalization: None,
        reasoning: None,
        thinking_budgets: None,
        headers: std::collections::HashMap::new(),
        metadata: std::collections::HashMap::new(),
        provider_name: None,
        model_name: None,
        asset_dir: None,
        compact_model: None,
        storm_mutating_tools: None,
        storm_exempt_tools: None,
        repair_stats: std::sync::Arc::new(
            crate::agent::agent_loop::tool_input_repair::RepairStats::new(),
        ),
        truncation_notes: std::sync::Arc::new(std::sync::Mutex::new(
            std::collections::HashMap::new(),
        )),
        tool_def_filter: None,
        dynamic_tool_search: false,
        escalation_stream_fn: None,
        escalation_provider_name: None,
        escalation_pending: std::sync::Arc::new(std::sync::Mutex::new(None)),
        escalation_max_per_session: 3,
        escalation_remaining: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(3)),
        file_touch_tracker: None,
        progress: None,
        verifier: None,
        critic_fn: None,
        classify_fn: None,
        code_review_fn: None,
        code_review_mode: crate::agent::agent_loop::types::CodeReviewMode::default(),
        code_review_repo: None,
        open_issues_gate_mode: crate::agent::agent_loop::types::GateMode::Off,
        verification_tiers_mode: crate::agent::agent_loop::types::GateMode::Off,
        safe_state_abort_mode: crate::agent::agent_loop::types::SafeStateMode::Off,
        session_id: None,
        goal_fn: None,
        goal: None,
        max_turns: None,
    }
}

fn empty_context() -> Context {
    Context {
        system_prompt: String::new(),
        messages: Vec::new(),
        tools: Vec::new(),
    }
}

/// Regression: a transient mid-stream error ("error decoding response
/// body") arriving AFTER the model has streamed content must NOT kill
/// the run. The streaming retry layer can't replay the turn (the
/// partial is already on screen), but the run loop recovers: it keeps
/// the preserved partial, nudges the model to continue, and the next
/// turn proceeds — instead of tearing down to idle and dropping any
/// queued steering.
#[tokio::test]
async fn transient_midstream_error_recovers_instead_of_terminating() {
    use crate::agent::agent_loop::message::{DeltaPhase, LoopEvent};
    let call = std::sync::Arc::new(AtomicUsize::new(0));
    let factory: StreamFn = std::sync::Arc::new({
        let call = call.clone();
        move |_ctx, _opts| {
            let n = call.fetch_add(1, Ordering::SeqCst);
            if n == 0 {
                // First call: stream some text, then die mid-stream
                // with a transient transport error.
                let partial = AssistantMessage::new(
                    vec![ContentBlock::Text {
                        text: "working on it".to_string(),
                    }],
                    StopReason::Stop,
                );
                Box::pin(futures::stream::iter(vec![
                    StreamEvent::Start {
                        partial: AssistantMessage::new(Vec::new(), StopReason::Stop),
                    },
                    StreamEvent::Delta {
                        partial,
                        phase: DeltaPhase::TextDelta,
                    },
                    StreamEvent::Error {
                        error: "error decoding response body".to_string(),
                    },
                ]))
            } else {
                // Recovery turn: complete normally.
                let msg = AssistantMessage::new(
                    vec![ContentBlock::Text {
                        text: "all done now".to_string(),
                    }],
                    StopReason::Stop,
                );
                Box::pin(futures::stream::iter(vec![StreamEvent::Done {
                    reason: StopReason::Stop,
                    message: msg,
                    usage: None,
                }]))
            }
        }
    });

    let (tx, mut rx) = tokio::sync::mpsc::channel(64);
    let messages = run_agent_loop(
        vec![LoopMessage::User(UserMessage::text("start"))],
        empty_context(),
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;

    // The run recovered past the error: the final assistant turn
    // completed instead of the run dying on the errored turn.
    let last_text = messages.iter().rev().find_map(|m| match m {
        LoopMessage::Assistant(a) => a.content.iter().find_map(|b| match b {
            ContentBlock::Text { text } => Some(text.clone()),
            _ => None,
        }),
        _ => None,
    });
    assert_eq!(
        last_text.as_deref(),
        Some("all done now"),
        "run must continue past a transient error and complete the recovery turn"
    );

    // A retry/recovery banner was surfaced so the UI isn't silent.
    let mut saw_retry = false;
    while let Ok(evt) = rx.try_recv() {
        if matches!(evt, LoopEvent::RetryNotice { .. }) {
            saw_retry = true;
        }
    }
    assert!(
        saw_retry,
        "recovery should surface a RetryNotice banner instead of dying silently"
    );
}

/// A sustained outage (every call fails transiently) must still
/// terminate — the recovery budget caps consecutive recoveries so a
/// dead network can't loop the run forever. After the budget is spent
/// the error surfaces as terminal, exactly as it did before recovery.
#[tokio::test]
async fn sustained_transient_error_terminates_after_budget() {
    use crate::agent::agent_loop::message::DeltaPhase;
    let calls = std::sync::Arc::new(AtomicUsize::new(0));
    let factory: StreamFn = std::sync::Arc::new({
        let calls = calls.clone();
        move |_ctx, _opts| {
            calls.fetch_add(1, Ordering::SeqCst);
            let partial = AssistantMessage::new(
                vec![ContentBlock::Text {
                    text: "halfway".to_string(),
                }],
                StopReason::Stop,
            );
            Box::pin(futures::stream::iter(vec![
                StreamEvent::Start {
                    partial: AssistantMessage::new(Vec::new(), StopReason::Stop),
                },
                StreamEvent::Delta {
                    partial,
                    phase: DeltaPhase::TextDelta,
                },
                StreamEvent::Error {
                    error: "error decoding response body".to_string(),
                },
            ]))
        }
    });

    let (tx, _rx) = tokio::sync::mpsc::channel(64);
    let messages = run_agent_loop(
        vec![LoopMessage::User(UserMessage::text("start"))],
        empty_context(),
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;

    // Recovered MAX_TRANSIENT_RECOVERIES times, then one final terminal
    // error — not an unbounded loop.
    let total_calls = calls.load(Ordering::SeqCst);
    assert_eq!(
        total_calls,
        (MAX_TRANSIENT_RECOVERIES as usize) + 1,
        "run must stop after the recovery budget is exhausted, not loop forever"
    );

    // The run terminated on a real error (not a clean Stop).
    let last = messages
        .iter()
        .rev()
        .find_map(|m| match m {
            LoopMessage::Assistant(a) => Some(a),
            _ => None,
        })
        .expect("an assistant message exists");
    assert_eq!(
        last.stop_reason,
        StopReason::Error,
        "after the budget the error must surface as terminal"
    );
}

/// dirge-kjyz: `transient_recoveries` counts CONSECUTIVE recoveries so a
/// dead network still terminates, but it was only ever incremented, never
/// reset. Four transient blips SEPARATED by healthy turns (a dead network
/// is not what's happening) must not accumulate past the budget into a
/// false hard-fail: a clean assistant turn resets the counter. Here every
/// blip is followed by a successful echo turn, so even though there are
/// four blips (> MAX_TRANSIENT_RECOVERIES = 3) the run completes.
#[tokio::test]
async fn transient_blips_separated_by_healthy_turns_do_not_accumulate() {
    use crate::agent::agent_loop::message::DeltaPhase;

    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo.clone());

    // Script by call index: blip, echo, blip, echo, blip, echo, blip, done.
    // Without the reset, the third blip already exhausts the budget and the
    // fourth blip (call 6) surfaces as terminal before "done" is reached.
    let counter = std::sync::Arc::new(AtomicUsize::new(0));
    let factory: StreamFn = std::sync::Arc::new(move |_ctx, _opts| {
        let n = counter.fetch_add(1, Ordering::SeqCst);
        let blip = n.is_multiple_of(2) && n <= 6;
        if blip {
            let partial = AssistantMessage::new(
                vec![ContentBlock::Text {
                    text: "partial".to_string(),
                }],
                StopReason::Stop,
            );
            Box::pin(futures::stream::iter(vec![
                StreamEvent::Start {
                    partial: AssistantMessage::new(Vec::new(), StopReason::Stop),
                },
                StreamEvent::Delta {
                    partial,
                    phase: DeltaPhase::TextDelta,
                },
                StreamEvent::Error {
                    error: "error decoding response body".to_string(),
                },
            ]))
        } else {
            let msg = if n >= 7 {
                text_response("done")
            } else {
                tool_use_response("call", "echo", serde_json::json!({"n": n}))
            };
            let reason = msg.stop_reason;
            Box::pin(futures::stream::iter(vec![StreamEvent::Done {
                reason,
                message: msg,
                usage: None,
            }]))
        }
    });

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(256);
    let messages = run_agent_loop(
        vec![user("start")],
        ctx,
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    // The run reached the final clean turn instead of hard-failing on the
    // fourth blip.
    let last_text = messages.iter().rev().find_map(|m| match m {
        LoopMessage::Assistant(a) => a.content.iter().find_map(|b| match b {
            ContentBlock::Text { text } => Some(text.clone()),
            _ => None,
        }),
        _ => None,
    });
    assert_eq!(
        last_text.as_deref(),
        Some("done"),
        "blips spread across healthy turns must not accumulate into a hard-fail"
    );

    // All four blips surfaced a recovery banner — proof the counter never
    // hit the budget despite exceeding it in raw count.
    let mut retries = 0;
    while let Ok(evt) = rx.try_recv() {
        if matches!(evt, LoopEvent::RetryNotice { .. }) {
            retries += 1;
        }
    }
    assert_eq!(
        retries,
        (MAX_TRANSIENT_RECOVERIES as usize) + 1,
        "each of the four separated blips must recover, not just the budgeted three"
    );
}

/// dirge-kq3a: a pass that frees nothing must NOT rotate the session or
/// emit `ContextCompacted`.
///
/// The fold trigger reads the API's `prompt_tokens`, which counts the system
/// prompt and every tool schema; the fold itself only rewrites
/// `current_context.messages`. When the unfoldable fixed overhead alone sits
/// above the threshold — a big MCP tool surface, say — the ratio stays high no
/// matter how often we fold, and nothing here can bring it down. Pre-fix the
/// pass still rotated the session id, rebuilt the agent and printed
/// "context compacted: N → N tokens" on EVERY turn, forever; sessions were
/// observed rotating every ~6 seconds with identical before/after counts.
///
/// Six messages is below `PROTECT_HEAD_DEFAULT + PROTECT_TAIL_DEFAULT + 1`, so
/// `compute_compress_window` returns an empty window and there is nothing to
/// summarize or prune — the same shape as the real runaway.
#[tokio::test]
async fn run_compaction_pass_that_frees_nothing_does_not_rotate_or_announce() {
    let mut ctx = empty_context();
    ctx.system_prompt = "you are an agent".into();
    for i in 0..6 {
        let role = if i % 2 == 0 { "user" } else { "assistant" };
        ctx.messages.push(serde_json::json!({
            "role": role,
            "content": format!("turn {i}"),
        }));
    }
    let before = ctx.messages.clone();

    // A summarizer that must never be called: with an empty compress window
    // there is nothing to summarize, and calling it would burn an LLM request
    // per turn for no benefit.
    let summarize_fn: Option<crate::agent::compression::SummarizeFn> =
        Some(std::sync::Arc::new(move |_prompt: String| {
            Box::pin(async move { panic!("summarizer must not run when nothing can be folded") })
        }));

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(8);
    super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        0,
        &None,
        None,
        &tx,
        &empty_checkpoint_slot(),
        &mut 0,
        u64::MAX,
    )
    .await;
    drop(tx);

    assert_eq!(
        ctx.messages, before,
        "a no-op pass must leave the context byte-identical",
    );

    let mut events = Vec::new();
    while let Some(ev) = rx.recv().await {
        events.push(ev);
    }
    assert!(
        !events
            .iter()
            .any(|ev| matches!(ev, LoopEvent::ContextCompacted { .. })),
        "a pass that freed nothing must not announce a compaction or rotate \
         the session; got {events:?}",
    );
}

/// LOOP-9 integration: `run_compaction_pass` end-to-end. Feed
/// a long conversation, a mock summarizer, and assert that
/// (a) the older messages were dropped, (b) a SUMMARY_PREFIX
/// system message was inserted at the head, (c) the latest
/// user message is still in the tail, and (d) a
/// `ContextCompacted` event was emitted with a rotated session id.
#[tokio::test]
async fn run_compaction_pass_inserts_summary_and_rotates_session() {
    let mut ctx = empty_context();
    ctx.system_prompt = "you are an agent".into();
    // Pad with 25 turns so the compaction window has material.
    ctx.messages.push(serde_json::json!({
        "role": "system", "content": "you are an agent"
    }));
    ctx.messages.push(serde_json::json!({
        "role": "user", "content": "initial task: fix the bug"
    }));
    for i in 0..20 {
        let role = if i % 2 == 0 { "assistant" } else { "user" };
        ctx.messages.push(serde_json::json!({
            "role": role,
            "content": format!("turn {i} with some content to fill bytes"),
        }));
    }
    ctx.messages.push(serde_json::json!({
        "role": "user", "content": "latest user request"
    }));
    let n_before = ctx.messages.len();

    // Mock summarizer: returns a valid Hermes-style summary
    // structure. We assert the prompt was built (non-empty).
    let prompt_seen = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
    let prompt_seen_inner = prompt_seen.clone();
    let summarize_fn: Option<crate::agent::compression::SummarizeFn> =
        Some(std::sync::Arc::new(move |prompt: String| {
            let store = prompt_seen_inner.clone();
            Box::pin(async move {
                *store.lock().unwrap() = prompt;
                Ok("## Active Task\nfix the bug\n\n\
                        ## Goal\nresolve the issue\n\n\
                        ## Completed Actions\n1. read the file\n\n\
                        ## Remaining Work\nrun tests"
                    .to_string())
            })
        }));

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(8);
    super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        0,
        &None,
        None,
        &tx,
        &empty_checkpoint_slot(),
        &mut 0,
        u64::MAX,
    )
    .await;
    drop(tx);

    // (a) older messages dropped.
    assert!(
        ctx.messages.len() < n_before,
        "expected compaction to shrink the message list: before={n_before} after={}",
        ctx.messages.len()
    );

    // (b) summary system message with SUMMARY_PREFIX is present.
    let summary_msg = ctx
        .messages
        .iter()
        .find(|m| {
            m.get("role").and_then(|v| v.as_str()) == Some("system")
                && m.get("content")
                    .and_then(|v| v.as_str())
                    .map(|s| s.contains("CONTEXT COMPACTION"))
                    .unwrap_or(false)
        })
        .expect("compaction summary message should be present");
    let body = summary_msg["content"].as_str().unwrap();
    assert!(body.contains("## Active Task"));
    assert!(body.contains("fix the bug"));

    // (c) latest user message preserved.
    let last = ctx.messages.last().unwrap();
    assert_eq!(last["content"].as_str().unwrap(), "latest user request");

    // (d) ContextCompacted event emitted with rotated session id.
    let mut compacted_event_seen = false;
    while let Some(ev) = rx.recv().await {
        if let LoopEvent::ContextCompacted { new_session_id, .. } = ev {
            assert!(
                new_session_id.starts_with("compacted-"),
                "session id should rotate via compacted- prefix; got {new_session_id}"
            );
            compacted_event_seen = true;
        }
    }
    assert!(compacted_event_seen, "expected ContextCompacted event");

    // Sanity: the summarizer received a Hermes structured prompt
    // (built via build_summary_prompt).
    let received = prompt_seen.lock().unwrap().clone();
    assert!(received.contains("TURNS TO SUMMARIZE"));
    assert!(received.contains("## Active Task"));
}

/// Build a padded context with `n` alternating turns after a system +
/// initial-user pair, for the compaction-pass tests.
fn padded_ctx(n: usize) -> super::Context {
    let mut ctx = empty_context();
    ctx.messages
        .push(serde_json::json!({"role": "system", "content": "you are an agent"}));
    ctx.messages
        .push(serde_json::json!({"role": "user", "content": "initial task"}));
    for i in 0..n {
        let role = if i % 2 == 0 { "assistant" } else { "user" };
        ctx.messages.push(serde_json::json!({
            "role": role,
            "content": format!("turn {i} with some content to fill bytes"),
        }));
    }
    ctx.messages
        .push(serde_json::json!({"role": "user", "content": "latest user request"}));
    ctx
}

/// A summarizer that records whether it was called and returns a distinct
/// inline summary, so a test can tell the inline path from the fast path.
fn recording_summarizer(
    called: std::sync::Arc<std::sync::atomic::AtomicBool>,
) -> Option<crate::agent::compression::SummarizeFn> {
    Some(std::sync::Arc::new(move |_prompt: String| {
        let called = called.clone();
        Box::pin(async move {
            called.store(true, std::sync::atomic::Ordering::SeqCst);
            Ok("## Active Task\nINLINE SUMMARY\n## Remaining Work\nx".to_string())
        })
    }))
}

/// A populated checkpoint slot for the fast-path tests.
fn slot_with(summary: &str, boundary: usize, generation: u64) -> super::CheckpointSlot {
    std::sync::Arc::new(std::sync::Mutex::new(Some(super::CachedCheckpoint {
        summary: summary.to_string(),
        boundary,
        generation,
    })))
}

/// dirge-ioym: a detached checkpoint (like the detached advisory review) used
/// to hold a STRONG clone of the loop event sender, so the per-turn channel —
/// and the runner task the pump joins on — stayed open until the bounded but
/// slow background call finished, blocking a drain-to-close consumer well past
/// AgentEnd. The detached tasks now hold WEAK senders, so the channel closes
/// as soon as the run's own sender drops.
#[tokio::test(start_paused = true)]
async fn detached_checkpoint_weak_sender_does_not_hold_channel_open() {
    use crate::agent::compression::SummarizeFn;

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(8);
    // A summarizer that stalls far past the run — with a strong sender the
    // recv below would block on it instead of seeing the channel close.
    let sfn: SummarizeFn = std::sync::Arc::new(|_prompt: String| {
        Box::pin(async {
            tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
            Ok("a summary that never arrives in time".to_string())
        })
    });
    let slot = empty_checkpoint_slot();
    super::spawn_incremental_checkpoint(
        sfn,
        vec![serde_json::json!({"role": "user", "content": "hello"})],
        tx.downgrade(),
        slot,
        1,
    );
    // The run ends: its (only strong) sender drops.
    drop(tx);

    // The channel must close now, not wait on the stalled detached task.
    match tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv()).await {
        Ok(None) => {}
        Ok(Some(ev)) => panic!("unexpected late event before the run drained: {ev:?}"),
        Err(_) => {
            panic!("channel stayed open — the detached checkpoint held it past the run's end")
        }
    }
}

/// Round 1 fast path: when the slot holds a fresh checkpoint (matching the
/// current epoch) and reusing it clears the fold target, the fold splices it
/// in WITHOUT calling the inline summarizer, then bumps the epoch and clears
/// the slot.
#[tokio::test]
async fn run_compaction_pass_reuses_fresh_checkpoint_without_calling_summarizer() {
    use std::sync::atomic::{AtomicBool, Ordering};

    let mut ctx = padded_ctx(20);
    let called = std::sync::Arc::new(AtomicBool::new(false));
    let summarize_fn = recording_summarizer(called.clone());
    let slot = slot_with(
        "## Active Task\nFROM CHECKPOINT\n## Remaining Work\nfinish",
        10,
        0,
    );
    let mut generation = 0u64;

    let (tx, _rx) = mpsc::channel::<LoopEvent>(16);
    let outcome = super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        0,
        &None,
        None,
        &tx,
        &slot,
        &mut generation,
        u64::MAX,
    )
    .await;
    drop(tx);

    assert!(
        matches!(outcome, super::SummaryOutcome::Succeeded(_)),
        "reuse should succeed"
    );
    assert!(
        !called.load(Ordering::SeqCst),
        "inline summarizer must NOT be called on the fast path"
    );
    let summary_msg = ctx
        .messages
        .iter()
        .find_map(|m| {
            let c = m.get("content").and_then(|v| v.as_str())?;
            c.contains("CONTEXT COMPACTION").then_some(c)
        })
        .expect("a summary message should be present");
    assert!(
        summary_msg.contains("FROM CHECKPOINT"),
        "the spliced summary should be the checkpoint's, not the inline one"
    );
    assert_eq!(generation, 1, "a successful fold bumps the epoch");
    assert!(
        slot.lock().unwrap().is_none(),
        "the consumed checkpoint slot is cleared after the fold"
    );
}

/// dirge-vpma.9: the fast (checkpoint-reuse) fold must still fire
/// `on_pre_compress` over the slice it discards, exactly once. The
/// background checkpointer that produced the summary never consulted the
/// memory provider, so without this the provider never sees the discarded
/// messages on the high-frequency fast path (the silent-insight-drop
/// dirge-h5tv fixed for the inline path).
#[tokio::test]
async fn fast_reuse_fires_on_pre_compress_over_discarded_slice() {
    use crate::extras::memory_provider::MemoryProvider;
    use std::sync::atomic::Ordering;

    struct RecordingProvider {
        seen: std::sync::Arc<std::sync::Mutex<Vec<String>>>,
    }
    impl MemoryProvider for RecordingProvider {
        fn name(&self) -> &str {
            "recording"
        }
        fn view(&self, _t: &str) -> Value {
            serde_json::json!({})
        }
        fn add(&self, _: &str, _: &str, _: Option<&str>) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
        fn replace(&self, _: &str, _: &str, _: &str, _: Option<&str>) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
        fn remove(&self, _: &str, _: &str) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
        fn on_pre_compress(&self, transcript: &str) -> String {
            self.seen.lock().unwrap().push(transcript.to_string());
            String::new()
        }
    }

    let mut ctx = padded_ctx(20);
    let called = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let summarize_fn = recording_summarizer(called.clone());
    let slot = slot_with(
        "## Active Task\nFROM CHECKPOINT\n## Remaining Work\nfinish",
        10,
        0,
    );
    let mut generation = 0u64;

    let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
    let provider: Option<std::sync::Arc<dyn MemoryProvider>> =
        Some(std::sync::Arc::new(RecordingProvider {
            seen: seen.clone(),
        }));

    let (tx, _rx) = mpsc::channel::<LoopEvent>(16);
    let outcome = super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        0,
        &provider,
        None,
        &tx,
        &slot,
        &mut generation,
        u64::MAX,
    )
    .await;
    drop(tx);

    assert!(
        matches!(outcome, super::SummaryOutcome::Succeeded(_)),
        "reuse should succeed"
    );
    assert!(
        !called.load(Ordering::SeqCst),
        "inline summarizer must NOT be called on the fast path"
    );
    let seen = seen.lock().unwrap();
    assert_eq!(
        seen.len(),
        1,
        "on_pre_compress must fire exactly once on the fast path (no drop, no double-fire)"
    );
    assert!(
        seen[0].contains("turn"),
        "the discarded messages should appear in the transcript the provider saw: {:?}",
        seen[0]
    );
}

/// dirge-vpma.9: on the fast reuse path, a plugin `on_compact` that returns
/// a valid summary is honored — folded IN PLACE OF the background
/// checkpoint's summary (first-refusal contract) — and it fires exactly
/// once (no double-fire with the inline path, which is `!reused`-guarded).
#[tokio::test]
async fn fast_reuse_on_compact_overrides_checkpoint_summary() {
    use crate::agent::agent_loop::types::CompactionHooks;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let mut ctx = padded_ctx(20);
    let called = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let summarize_fn = recording_summarizer(called.clone());
    let slot = slot_with(
        "## Active Task\nFROM CHECKPOINT\n## Remaining Work\nfinish",
        10,
        0,
    );
    let mut generation = 0u64;

    let compact_calls = std::sync::Arc::new(AtomicUsize::new(0));
    let cc = compact_calls.clone();
    let hooks = CompactionHooks {
        on_before: std::sync::Arc::new(|_c, _t| Box::pin(async {})),
        on_compact: std::sync::Arc::new(move |_middle| {
            cc.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move {
                Some("## Active Task\nPLUGIN-SUMMARY\n## Remaining Work\ngo".to_string())
            })
        }),
    };

    let (tx, _rx) = mpsc::channel::<LoopEvent>(16);
    super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        0,
        &None,
        Some(&hooks),
        &tx,
        &slot,
        &mut generation,
        u64::MAX,
    )
    .await;
    drop(tx);

    assert!(
        !called.load(Ordering::SeqCst),
        "inline summarizer must NOT be called on the fast path"
    );
    assert_eq!(
        compact_calls.load(Ordering::SeqCst),
        1,
        "on_compact must fire exactly once on the fast path"
    );
    let has = |needle: &str| {
        ctx.messages.iter().any(|m| {
            m.get("content")
                .and_then(|v| v.as_str())
                .map(|s| s.contains(needle))
                .unwrap_or(false)
        })
    };
    assert!(
        has("PLUGIN-SUMMARY"),
        "the plugin summary must win over the checkpoint's on the fast path"
    );
    assert!(
        !has("FROM CHECKPOINT"),
        "the checkpoint summary must be replaced by the plugin's"
    );
}

/// A checkpoint from a stale epoch (generation mismatch) is ignored — the
/// fold falls back to the inline summarizer.
#[tokio::test]
async fn run_compaction_pass_ignores_stale_generation_checkpoint() {
    use std::sync::atomic::{AtomicBool, Ordering};

    let mut ctx = padded_ctx(20);
    let called = std::sync::Arc::new(AtomicBool::new(false));
    let summarize_fn = recording_summarizer(called.clone());
    // Slot built under epoch 0, but the loop is now at epoch 7.
    let slot = slot_with(
        "## Active Task\nFROM CHECKPOINT\n## Remaining Work\nx",
        10,
        0,
    );
    let mut generation = 7u64;

    let (tx, _rx) = mpsc::channel::<LoopEvent>(16);
    let outcome = super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        0,
        &None,
        None,
        &tx,
        &slot,
        &mut generation,
        u64::MAX,
    )
    .await;
    drop(tx);

    assert!(matches!(outcome, super::SummaryOutcome::Succeeded(_)));
    assert!(
        called.load(Ordering::SeqCst),
        "stale checkpoint → inline summarizer must run"
    );
    let summary_msg = ctx
        .messages
        .iter()
        .find_map(|m| {
            let c = m.get("content").and_then(|v| v.as_str())?;
            c.contains("CONTEXT COMPACTION").then_some(c)
        })
        .expect("a summary message should be present");
    assert!(
        summary_msg.contains("INLINE SUMMARY"),
        "the inline summary should be used when the checkpoint is stale"
    );
}

/// Serializes the tests that touch a process-global memory flag (the
/// memory-dirty flag and the verbatim-pre-recall toggle) so they don't perturb
/// each other under parallel test execution — e.g. a pre-recall test leaving
/// the toggle on while a memory-refresh test runs `run_agent_loop` with a
/// provider. Every test that flips one of those globals holds this lock.
static DIRTY_FLAG_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

/// Round 2 flag: `mark_memories_dirty` then `take_memories_dirty` returns
/// true exactly once, then false.
#[test]
fn memories_dirty_flag_is_consumed_once() {
    use crate::agent::agent_loop::context_manager::{mark_memories_dirty, take_memories_dirty};
    let _guard = DIRTY_FLAG_TEST_LOCK.lock().unwrap();
    // Clear any prior state from other tests touching the global.
    let _ = take_memories_dirty();
    mark_memories_dirty();
    assert!(take_memories_dirty(), "first take after mark is true");
    assert!(!take_memories_dirty(), "second take resets to false");
}

/// Round 2 behavior: when consolidation has marked memories dirty, the loop
/// re-injects the refreshed memory block into the model-facing context at the
/// next turn boundary, so the agent sees it without a restart.
#[tokio::test]
#[allow(clippy::await_holding_lock)] // current-thread test runtime; lock only serializes the global flag
async fn memory_refresh_injects_block_at_turn_boundary_when_dirty() {
    use crate::extras::memory_provider::MemoryProvider;
    let _guard = DIRTY_FLAG_TEST_LOCK.lock().unwrap();

    struct StubProvider;
    impl MemoryProvider for StubProvider {
        fn name(&self) -> &str {
            "stub"
        }
        fn format_for_system_prompt(&self) -> String {
            "STUBMEM: prefer the fast path".to_string()
        }
        fn view(&self, _t: &str) -> Value {
            serde_json::json!({})
        }
        fn add(&self, _: &str, _: &str, _: Option<&str>) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
        fn replace(&self, _: &str, _: &str, _: &str, _: Option<&str>) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
        fn remove(&self, _: &str, _: &str) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
    }

    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo.clone());

    let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
    // Turn 1: a tool call (forces another loop iteration → a turn boundary).
    // Turn 2: final text.
    let factory = capturing_factory(
        vec![
            tool_use_response("call-1", "echo", serde_json::json!({"v": 1})),
            text_response("done"),
        ],
        seen.clone(),
    );

    let provider: std::sync::Arc<dyn MemoryProvider> = std::sync::Arc::new(StubProvider);

    // The injected refresh is a `system` message; use a converter that keeps
    // system messages (production's does — fold summaries are system-role).
    let mut config = build_config();
    config.convert_to_llm = std::sync::Arc::new(|messages: &[Value]| {
        messages
            .iter()
            .filter(|m| {
                let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("");
                matches!(
                    role,
                    "user" | "assistant" | "tool" | "toolResult" | "system"
                )
            })
            .cloned()
            .collect()
    });

    // Simulate background consolidation completing: mark dirty just before
    // the run so the next turn boundary consumes it.
    crate::agent::agent_loop::context_manager::mark_memories_dirty();

    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let _ = run_agent_loop(
        vec![user("echo please")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        Some(provider),
    )
    .await;
    drop(tx);

    let snapshots = seen.lock().unwrap().clone();
    assert!(
        snapshots.iter().any(|s| s.contains("STUBMEM")),
        "the refreshed memory block should appear in the model-facing context \
         after the turn boundary; snapshots={snapshots:?}"
    );
}

/// dirge-0gxb: with verbatim pre-recall on, the hits from searching the
/// verbatim user message reach the MODEL-FACING context, but the injected
/// block is NEVER persisted into the returned (`new_messages`) history — the
/// core supplemental-not-persisted invariant, exercised end-to-end through
/// `run_agent_loop` (the unit tests only cover `pre_recall_block` formatting).
#[tokio::test]
#[allow(clippy::await_holding_lock)] // current-thread test runtime; lock only serializes the global flag
async fn pre_recall_reaches_model_context_but_not_persisted_history() {
    use crate::agent::agent_loop::context_manager::set_verbatim_pre_recall;
    use crate::extras::memory_provider::MemoryProvider;
    let _guard = DIRTY_FLAG_TEST_LOCK.lock().unwrap();

    // Provider whose search returns a distinctive hit; empty snapshot so the
    // hit isn't de-duped against `<project_memory>`.
    struct RecallProvider;
    impl MemoryProvider for RecallProvider {
        fn name(&self) -> &str {
            "recall-stub"
        }
        fn format_for_system_prompt(&self) -> String {
            String::new()
        }
        fn view(&self, _t: &str) -> Value {
            serde_json::json!({})
        }
        fn add(&self, _: &str, _: &str, _: Option<&str>) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
        fn replace(&self, _: &str, _: &str, _: &str, _: Option<&str>) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
        fn remove(&self, _: &str, _: &str) -> Result<Value, String> {
            Ok(serde_json::json!({}))
        }
        fn search(&self, _q: &str) -> Result<Value, String> {
            Ok(serde_json::json!({
                "results": [{"id": "urn:ump:x", "content": "PRERECALLHIT: the widget cache lives in src/cache.rs"}]
            }))
        }
    }

    let ctx = empty_context();
    let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
    let factory = capturing_factory(vec![text_response("done")], seen.clone());
    let provider: std::sync::Arc<dyn MemoryProvider> = std::sync::Arc::new(RecallProvider);

    // Pre-recall injects a `user`-role message; keep user/assistant/tool/system.
    let mut config = build_config();
    config.convert_to_llm = std::sync::Arc::new(|messages: &[Value]| {
        messages
            .iter()
            .filter(|m| {
                let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("");
                matches!(
                    role,
                    "user" | "assistant" | "tool" | "toolResult" | "system"
                )
            })
            .cloned()
            .collect()
    });

    set_verbatim_pre_recall(true);
    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let returned = run_agent_loop(
        vec![user("how do I cache the widget")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        Some(provider),
    )
    .await;
    drop(tx);
    // Reset BEFORE asserting so a failure can't leak the toggle to other tests.
    set_verbatim_pre_recall(false);

    let snapshots = seen.lock().unwrap().clone();
    assert!(
        snapshots.iter().any(|s| s.contains("PRERECALLHIT")),
        "pre-recall hit must reach the model-facing context; snapshots={snapshots:?}",
    );
    let persisted = format!("{returned:?}");
    assert!(
        !persisted.contains("PRERECALLHIT"),
        "pre-recall block must NOT be persisted into new_messages: {persisted}",
    );
}

/// dirge-jia8: a plugin `on-compact` hook supplying a valid summary
/// is used INSTEAD of the LLM summarizer; the observe-only
/// `on-before-compact` hook fires. Built from plain closures (no
/// Janet needed) so it runs on the default feature set.
#[tokio::test]
async fn compaction_on_compact_hook_overrides_llm_summary() {
    use crate::agent::agent_loop::types::CompactionHooks;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let mut ctx = empty_context();
    ctx.messages
        .push(serde_json::json!({"role": "system", "content": "sys"}));
    ctx.messages
        .push(serde_json::json!({"role": "user", "content": "initial"}));
    for i in 0..20 {
        let role = if i % 2 == 0 { "assistant" } else { "user" };
        ctx.messages
            .push(serde_json::json!({"role": role, "content": format!("turn {i} content")}));
    }
    ctx.messages
        .push(serde_json::json!({"role": "user", "content": "latest"}));

    // LLM summarizer returns a DISTINCT summary — if the plugin
    // override works, this text must NOT appear.
    let llm_called = std::sync::Arc::new(AtomicUsize::new(0));
    let llm_called_c = llm_called.clone();
    let summarize_fn: Option<crate::agent::compression::SummarizeFn> =
        Some(std::sync::Arc::new(move |_prompt: String| {
            llm_called_c.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move { Ok("## Active Task\nLLM-SUMMARY".to_string()) })
        }));

    // on-before observe counter + on-compact returning a custom summary.
    let before_fired = std::sync::Arc::new(AtomicUsize::new(0));
    let before_c = before_fired.clone();
    let hooks = CompactionHooks {
        on_before: std::sync::Arc::new(move |_count, _tokens| {
            let f = before_c.clone();
            Box::pin(async move {
                f.fetch_add(1, Ordering::SeqCst);
            })
        }),
        on_compact: std::sync::Arc::new(move |_middle| {
            Box::pin(async move { Some("## Active Task\nPLUGIN-SUMMARY".to_string()) })
        }),
    };

    let (tx, _rx) = mpsc::channel::<LoopEvent>(8);
    super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        0,
        &None,
        Some(&hooks),
        &tx,
        &empty_checkpoint_slot(),
        &mut 0,
        u64::MAX,
    )
    .await;
    drop(tx);

    // on-before-compact observed the fold.
    assert_eq!(
        before_fired.load(Ordering::SeqCst),
        1,
        "on-before-compact must fire"
    );
    // The plugin summary was applied, not the LLM's.
    let summary_msg = ctx
        .messages
        .iter()
        .find(|m| {
            m.get("content")
                .and_then(|v| v.as_str())
                .map(|s| s.contains("PLUGIN-SUMMARY"))
                .unwrap_or(false)
        })
        .expect("plugin summary must be in the compacted context");
    assert!(
        summary_msg["content"]
            .as_str()
            .unwrap()
            .contains("PLUGIN-SUMMARY")
    );
    assert!(
        !ctx.messages.iter().any(|m| m
            .get("content")
            .and_then(|v| v.as_str())
            .map(|s| s.contains("LLM-SUMMARY"))
            .unwrap_or(false)),
        "LLM summary must NOT appear — plugin override should win",
    );
    assert_eq!(
        llm_called.load(Ordering::SeqCst),
        0,
        "LLM summarizer must NOT be called when the plugin supplies a valid summary",
    );
}

/// dirge-jia8: an `on-compact` hook returning an INVALID summary
/// (fails validate_summary) falls through to the LLM summarizer —
/// the plugin can't inject garbage as the summary.
#[tokio::test]
async fn compaction_invalid_plugin_summary_falls_through_to_llm() {
    use crate::agent::agent_loop::types::CompactionHooks;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let mut ctx = empty_context();
    ctx.messages
        .push(serde_json::json!({"role": "system", "content": "sys"}));
    ctx.messages
        .push(serde_json::json!({"role": "user", "content": "initial"}));
    for i in 0..20 {
        let role = if i % 2 == 0 { "assistant" } else { "user" };
        ctx.messages
            .push(serde_json::json!({"role": role, "content": format!("turn {i} content")}));
    }
    ctx.messages
        .push(serde_json::json!({"role": "user", "content": "latest"}));

    let llm_called = std::sync::Arc::new(AtomicUsize::new(0));
    let llm_called_c = llm_called.clone();
    let summarize_fn: Option<crate::agent::compression::SummarizeFn> =
        Some(std::sync::Arc::new(move |_prompt: String| {
            llm_called_c.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move { Ok("## Active Task\nLLM-SUMMARY".to_string()) })
        }));

    let hooks = CompactionHooks {
        on_before: std::sync::Arc::new(|_c, _t| Box::pin(async {})),
        // Invalid: no required section header → validate_summary fails.
        on_compact: std::sync::Arc::new(move |_middle| {
            Box::pin(async move { Some("garbage with no section header".to_string()) })
        }),
    };

    let (tx, _rx) = mpsc::channel::<LoopEvent>(8);
    super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        0,
        &None,
        Some(&hooks),
        &tx,
        &empty_checkpoint_slot(),
        &mut 0,
        u64::MAX,
    )
    .await;
    drop(tx);

    assert_eq!(
        llm_called.load(Ordering::SeqCst),
        1,
        "invalid plugin summary must fall through to the LLM summarizer",
    );
    assert!(
        ctx.messages.iter().any(|m| m
            .get("content")
            .and_then(|v| v.as_str())
            .map(|s| s.contains("LLM-SUMMARY"))
            .unwrap_or(false)),
        "LLM summary should be applied after the invalid plugin summary",
    );
}

/// LOOP-9: when no summarizer is wired, the compaction pass
/// still runs the cheap pruning and emits ContextCompacted, but
/// does NOT insert a structured summary system message.
#[tokio::test]
async fn run_compaction_pass_without_summarizer_prunes_only() {
    let mut ctx = empty_context();
    // One large tool result that should be pruned.
    ctx.messages.push(serde_json::json!({
        "role": "user", "content": "first"
    }));
    ctx.messages.push(serde_json::json!({
        "role": "toolResult", "content": "x".repeat(2000), "toolName": "bash"
    }));
    ctx.messages.push(serde_json::json!({
        "role": "user", "content": "tail"
    }));
    ctx.messages.push(serde_json::json!({
        "role": "assistant", "content": "tail asst"
    }));

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(4);
    // Use protect_tail = 2 so the large tool result is eligible
    // for pruning (it's at index 1, end = 4 - 2 = 2, so index
    // 1 is in-range).
    super::run_compaction_pass(
        &mut ctx,
        &None,
        2,
        0,
        &None,
        None,
        &tx,
        &empty_checkpoint_slot(),
        &mut 0,
        u64::MAX,
    )
    .await;
    drop(tx);

    // No SUMMARY_PREFIX message inserted.
    let has_summary = ctx.messages.iter().any(|m| {
        m.get("content")
            .and_then(|v| v.as_str())
            .map(|s| s.contains("CONTEXT COMPACTION"))
            .unwrap_or(false)
    });
    assert!(
        !has_summary,
        "no summary should be inserted without summarize_fn"
    );

    // The large tool result was pruned (replaced with a [bash] marker).
    let tool_msg = &ctx.messages[1];
    assert!(tool_msg["content"].as_str().unwrap().contains("[bash]"));

    // ContextCompacted still emitted.
    let mut compacted_event_seen = false;
    while let Some(ev) = rx.recv().await {
        if matches!(ev, LoopEvent::ContextCompacted { .. }) {
            compacted_event_seen = true;
        }
    }
    assert!(compacted_event_seen);
}

/// Mock echo tool for run-loop tests. Records executed args
/// per call so test setups can detect terminate-flag flow.
#[derive(Debug)]
struct EchoTool {
    terminate: bool,
    executed: std::sync::Arc<Mutex<Vec<Value>>>,
}
impl EchoTool {
    fn new() -> Self {
        Self {
            terminate: false,
            executed: std::sync::Arc::new(Mutex::new(Vec::new())),
        }
    }
    fn with_terminate(mut self) -> Self {
        self.terminate = true;
        self
    }
}
impl LoopTool for EchoTool {
    fn name(&self) -> &str {
        "echo"
    }
    fn description(&self) -> &str {
        "Echo tool"
    }
    fn label(&self) -> &str {
        "Echo"
    }
    fn parameters(&self) -> &Value {
        static EMPTY: std::sync::OnceLock<Value> = std::sync::OnceLock::new();
        EMPTY.get_or_init(|| serde_json::json!({"type": "object"}))
    }
    fn execute<'a>(
        &'a self,
        _id: &'a str,
        args: Value,
        _signal: AbortSignal,
        _on_update: LoopToolUpdate,
    ) -> Pin<Box<dyn Future<Output = Result<super::super::LoopToolResult, String>> + Send + 'a>>
    {
        let executed = self.executed.clone();
        let terminate = self.terminate;
        Box::pin(async move {
            executed.lock().unwrap().push(args.clone());
            Ok(super::super::LoopToolResult {
                content: vec![serde_json::json!({"type": "text", "text": "ok"})],
                details: args,
                terminate: if terminate { Some(true) } else { None },
            })
        })
    }
}

fn user(text: &str) -> LoopMessage {
    LoopMessage::User(UserMessage::text(text))
}

fn text_response(text: &str) -> AssistantMessage {
    AssistantMessage::new(
        vec![ContentBlock::Text {
            text: text.to_string(),
        }],
        StopReason::Stop,
    )
}

fn tool_use_response(id: &str, name: &str, args: Value) -> AssistantMessage {
    AssistantMessage::new(
        vec![ContentBlock::ToolCall {
            id: id.to_string(),
            name: name.to_string(),
            arguments: args,
        }],
        StopReason::ToolUse,
    )
}

/// Drain channel into a Vec.
async fn drain(rx: &mut mpsc::Receiver<LoopEvent>) -> Vec<LoopEvent> {
    let mut out = Vec::new();
    while let Some(e) = rx.recv().await {
        out.push(e);
    }
    out
}

/// Port of pi test "should emit events with AgentMessage types"
/// (agent-loop.test.ts:84). Full agent loop run — assistant
/// response, no tools.
#[tokio::test]
async fn test_emits_full_agent_loop_event_sequence() {
    let factory = canned_factory(vec![text_response("Hi there!")]);
    let (tx, mut rx) = mpsc::channel::<LoopEvent>(64);
    let messages = run_agent_loop(
        vec![user("Hello")],
        empty_context(),
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None, // memory_provider — test default
    )
    .await;
    drop(tx);

    let kinds: Vec<_> = drain(&mut rx).await.iter().map(|e| e.kind()).collect();
    // Must contain all pi-required events.
    for required in [
        "agent_start",
        "turn_start",
        "message_start",
        "message_end",
        "turn_end",
        "agent_end",
    ] {
        assert!(kinds.contains(&required), "missing {required}: {kinds:?}");
    }
    // Return value: user + assistant message.
    assert_eq!(messages.len(), 2);
    assert_eq!(messages[0].role(), "user");
    assert_eq!(messages[1].role(), "assistant");
}

/// Port of pi test "should handle tool calls and results"
/// (agent-loop.test.ts:239). Full-loop scope: assistant emits
/// tool call → loop dispatches → next assistant emits final
/// text.
#[tokio::test]
async fn test_full_loop_with_tool_then_final_text() {
    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo.clone());

    let factory = canned_factory(vec![
        tool_use_response("call-1", "echo", serde_json::json!({"v": 1})),
        text_response("done"),
    ]);

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(128);
    let messages = run_agent_loop(
        vec![user("echo")],
        ctx,
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None, // memory_provider — test default
    )
    .await;
    drop(tx);

    // Tool actually executed.
    assert_eq!(echo.executed.lock().unwrap().len(), 1);

    // Roles: user, assistant (tool use), toolResult, assistant (text).
    let roles: Vec<_> = messages.iter().map(|m| m.role()).collect();
    assert_eq!(roles, vec!["user", "assistant", "toolResult", "assistant"]);

    // Stream of events should contain tool_execution_start +
    // tool_execution_end.
    let kinds: Vec<_> = drain(&mut rx).await.iter().map(|e| e.kind()).collect();
    assert!(kinds.contains(&"tool_execution_start"));
    assert!(kinds.contains(&"tool_execution_end"));
}

/// Port of pi test "should use prepareNextTurn snapshot before
/// continuing" (agent-loop.test.ts:897). The hook returns a
/// snapshot mutating `context`; subsequent turn observes the
/// mutation.
#[tokio::test]
async fn test_prepare_next_turn_snapshot_applied() {
    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.system_prompt = "first prompt".to_string();
    ctx.tools.push(echo.clone());

    // Track the system_prompt seen at each LLM call.
    let observed_prompts = std::sync::Arc::new(Mutex::new(Vec::<String>::new()));
    let observed_clone = observed_prompts.clone();
    let counter = std::sync::Arc::new(AtomicUsize::new(0));
    let factory: StreamFn = std::sync::Arc::new(move |llm_ctx, _opts| {
        observed_clone.lock().unwrap().push(llm_ctx.system_prompt);
        let n = counter.fetch_add(1, Ordering::SeqCst);
        let msg = if n == 0 {
            tool_use_response("call-1", "echo", serde_json::json!({"v": 1}))
        } else {
            text_response("done")
        };
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![StreamEvent::Done {
            reason,
            message: msg,
            usage: None,
        }]))
    });

    // Hook fires once: returns a new context with a different
    // system prompt.
    let fired = std::sync::Arc::new(AtomicUsize::new(0));
    let fired_clone = fired.clone();
    let hook: PrepareNextTurnFn = std::sync::Arc::new(move |ctx| {
        let fired = fired_clone.clone();
        Box::pin(async move {
            if fired.fetch_add(1, Ordering::SeqCst) > 0 {
                return None; // only on the first invocation
            }
            Some(TurnUpdate {
                context: Some(Context {
                    system_prompt: "second prompt".to_string(),
                    messages: ctx.context.messages.clone(),
                    tools: ctx.context.tools.clone(),
                }),
                ..Default::default()
            })
        })
    });

    let mut config = build_config();
    config.prepare_next_turn = Some(hook);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let _ = run_agent_loop(
        vec![user("echo something")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None, // memory_provider — test default
    )
    .await;

    let observed = observed_prompts.lock().unwrap().clone();
    assert_eq!(observed.len(), 2, "expected 2 LLM calls");
    assert_eq!(observed[0], "first prompt");
    assert_eq!(
        observed[1], "second prompt",
        "second LLM call should see the mutated context"
    );
}

/// dirge-6js7 plugin review: prepareNextTurn returning a new
/// thinking_level must actually be APPLIED to the next turn's
/// stream call (config.reasoning), not dropped with a warning.
/// This is the fix for the HIGH "looks present but doesn't fire"
/// finding — the plugin `harness/set-next-thinking-level` slot
/// flows through prepare_next_turn into the live loop.
#[tokio::test]
async fn prepare_next_turn_applies_thinking_level_to_next_turn() {
    use crate::agent::agent_loop::types::ThinkingLevel;

    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo.clone());

    // Record the `reasoning` (thinking level) seen at each LLM call.
    let observed_reasoning = std::sync::Arc::new(Mutex::new(Vec::<Option<ThinkingLevel>>::new()));
    let observed_clone = observed_reasoning.clone();
    let counter = std::sync::Arc::new(AtomicUsize::new(0));
    let factory: StreamFn = std::sync::Arc::new(move |_llm_ctx, opts| {
        observed_clone.lock().unwrap().push(opts.reasoning);
        let n = counter.fetch_add(1, Ordering::SeqCst);
        // Turn 1 calls a tool (loop continues); turn 2 finishes.
        let msg = if n == 0 {
            tool_use_response("call-1", "echo", serde_json::json!({"v": 1}))
        } else {
            text_response("done")
        };
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![StreamEvent::Done {
            reason,
            message: msg,
            usage: None,
        }]))
    });

    // Hook fires after turn 1 and requests a thinking-level swap.
    let fired = std::sync::Arc::new(AtomicUsize::new(0));
    let fired_clone = fired.clone();
    let hook: PrepareNextTurnFn = std::sync::Arc::new(move |_ctx| {
        let fired = fired_clone.clone();
        Box::pin(async move {
            if fired.fetch_add(1, Ordering::SeqCst) > 0 {
                return None;
            }
            Some(TurnUpdate {
                thinking_level: Some(ThinkingLevel::High),
                ..Default::default()
            })
        })
    });

    let mut config = build_config();
    config.prepare_next_turn = Some(hook);
    // Start with no reasoning set so the swap is observable.
    config.reasoning = None;

    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let _ = run_agent_loop(
        vec![user("go")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;

    let observed = observed_reasoning.lock().unwrap().clone();
    assert_eq!(observed.len(), 2, "expected 2 LLM calls");
    assert_eq!(
        observed[0], None,
        "turn 1 runs with the initial reasoning (none)"
    );
    assert_eq!(
        observed[1],
        Some(ThinkingLevel::High),
        "turn 2 must see the thinking_level prepareNextTurn requested — \
         pre-fix this was dropped and turn 2 saw None",
    );
}

/// Port of pi test "should stop after the current turn when
/// shouldStopAfterTurn returns true" (agent-loop.test.ts:970).
#[tokio::test]
async fn test_should_stop_after_turn_stops_loop() {
    let factory = canned_factory(vec![
        text_response("turn one"),
        // Second response should NEVER be requested — hook
        // stops the loop after turn one.
        text_response("should not appear"),
    ]);

    let llm_calls = std::sync::Arc::new(AtomicUsize::new(0));
    let llm_calls_clone = llm_calls.clone();
    // Wrap factory to count invocations.
    let factory_counted: StreamFn = std::sync::Arc::new(move |ctx, opts| {
        llm_calls_clone.fetch_add(1, Ordering::SeqCst);
        factory(ctx, opts)
    });

    let hook: ShouldStopAfterTurnFn = std::sync::Arc::new(|_ctx| Box::pin(async move { true }));

    let mut config = build_config();
    config.should_stop_after_turn = Some(hook);

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(64);
    let messages = run_agent_loop(
        vec![user("hi")],
        empty_context(),
        config,
        AbortSignal::new(),
        &tx,
        &factory_counted,
        None,
        None, // memory_provider — test default
    )
    .await;
    drop(tx);

    // Only one LLM call.
    assert_eq!(llm_calls.load(Ordering::SeqCst), 1);
    // Messages: user + one assistant.
    assert_eq!(messages.len(), 2);
    // Loop emitted agent_end.
    let kinds: Vec<_> = drain(&mut rx).await.iter().map(|e| e.kind()).collect();
    assert!(kinds.contains(&"agent_end"));
}

/// Port of pi test "should stop after a tool batch when every
/// tool result sets terminate=true" (agent-loop.test.ts:1067).
/// LOOP-LEVEL: only one LLM call (the tool dispatch terminates).
#[tokio::test]
async fn test_terminate_stops_loop_after_tool_batch() {
    let echo = std::sync::Arc::new(EchoTool::new().with_terminate());
    let mut ctx = empty_context();
    ctx.tools.push(echo);

    let llm_calls = std::sync::Arc::new(AtomicUsize::new(0));
    let llm_calls_clone = llm_calls.clone();
    let factory: StreamFn = std::sync::Arc::new(move |_ctx, _opts| {
        llm_calls_clone.fetch_add(1, Ordering::SeqCst);
        let msg = tool_use_response("call-1", "echo", serde_json::json!({"v": 1}));
        Box::pin(futures::stream::iter(vec![StreamEvent::Done {
            reason: StopReason::ToolUse,
            message: msg,
            usage: None,
        }]))
    });

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);
    let messages = run_agent_loop(
        vec![user("echo")],
        ctx,
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None, // memory_provider — test default
    )
    .await;

    assert_eq!(llm_calls.load(Ordering::SeqCst), 1, "no second LLM call");
    // user + assistant(tool use) + toolResult — no second
    // assistant text turn.
    let roles: Vec<_> = messages.iter().map(|m| m.role()).collect();
    assert_eq!(roles, vec!["user", "assistant", "toolResult"]);
}

/// Port of pi test "should allow afterToolCall to mark a tool
/// batch as terminating" (agent-loop.test.ts:1184). LOOP-LEVEL.
#[tokio::test]
async fn test_after_tool_call_terminate_stops_loop() {
    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo);

    let llm_calls = std::sync::Arc::new(AtomicUsize::new(0));
    let llm_calls_clone = llm_calls.clone();
    let factory: StreamFn = std::sync::Arc::new(move |_ctx, _opts| {
        llm_calls_clone.fetch_add(1, Ordering::SeqCst);
        let msg = tool_use_response("call-1", "echo", serde_json::json!({"v": 1}));
        Box::pin(futures::stream::iter(vec![StreamEvent::Done {
            reason: StopReason::ToolUse,
            message: msg,
            usage: None,
        }]))
    });

    let after: AfterToolCallFn = std::sync::Arc::new(|_ctx: AfterToolCallContext| {
        Box::pin(async move {
            Some(AfterToolCallResult {
                content: None,
                details: None,
                is_error: None,
                terminate: Some(true),
            })
        })
    });
    let mut config = build_config();
    config.after_tool_call = Some(after);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);
    let _ = run_agent_loop(
        vec![user("echo")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None, // memory_provider — test default
    )
    .await;

    assert_eq!(llm_calls.load(Ordering::SeqCst), 1, "no second LLM call");
}

/// Port of pi test "should continue after parallel tool calls
/// when not all tool results terminate" (agent-loop.test.ts:1119).
/// LOOP-LEVEL: two LLM calls.
#[tokio::test]
async fn test_continue_when_not_all_terminate() {
    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo);

    let llm_calls = std::sync::Arc::new(AtomicUsize::new(0));
    let llm_calls_clone = llm_calls.clone();
    let factory: StreamFn = std::sync::Arc::new(move |_ctx, _opts| {
        let n = llm_calls_clone.fetch_add(1, Ordering::SeqCst);
        let msg = if n == 0 {
            tool_use_response("call-1", "echo", serde_json::json!({"v": 1}))
        } else {
            text_response("done")
        };
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![StreamEvent::Done {
            reason,
            message: msg,
            usage: None,
        }]))
    });

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);
    let _ = run_agent_loop(
        vec![user("echo")],
        ctx,
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None, // memory_provider — test default
    )
    .await;

    assert_eq!(
        llm_calls.load(Ordering::SeqCst),
        2,
        "two LLM calls expected"
    );
}

/// Port of pi test "should inject queued messages after all
/// tool calls complete" (agent-loop.test.ts:547).
///
/// Setup: assistant emits a tool call. After tool dispatch
/// the loop polls `getSteeringMessages` which returns a user
/// message ONCE. That message is injected before the next
/// assistant call; the second LLM call sees it in its context.
#[tokio::test]
async fn test_steering_messages_injected_after_tool_calls() {
    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo);

    // Steering hook delivers once on the SECOND call (so
    // not on initial poll).
    let poll_count = std::sync::Arc::new(AtomicUsize::new(0));
    let poll_clone = poll_count.clone();
    let steering: GetSteeringMessagesFn = std::sync::Arc::new(move || {
        let poll = poll_clone.clone();
        Box::pin(async move {
            let n = poll.fetch_add(1, Ordering::SeqCst);
            if n == 1 {
                vec![user("interrupt")]
            } else {
                Vec::new()
            }
        })
    });

    // Inspector: record what each LLM call sees in its
    // converted message list.
    let saw_interrupt_on_second = std::sync::Arc::new(std::sync::Mutex::new(false));
    let saw_clone = saw_interrupt_on_second.clone();
    let call_counter = std::sync::Arc::new(AtomicUsize::new(0));

    let factory: StreamFn = std::sync::Arc::new(move |llm_ctx, _opts| {
        let n = call_counter.fetch_add(1, Ordering::SeqCst);
        if n == 1 {
            // Second call: check for "interrupt" in messages.
            let found = llm_ctx.messages.iter().any(|m| {
                m.get("role").and_then(|r| r.as_str()) == Some("user")
                    && m.get("content")
                        .and_then(|c| c.as_str())
                        .map(|s| s.contains("interrupt"))
                        == Some(true)
            });
            *saw_clone.lock().unwrap() = found;
        }
        let msg = if n == 0 {
            tool_use_response("call-1", "echo", serde_json::json!({"v": 1}))
        } else {
            text_response("done")
        };
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![StreamEvent::Done {
            reason,
            message: msg,
            usage: None,
        }]))
    });

    let mut config = build_config();
    config.get_steering_messages = Some(steering);

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(128);
    let messages = run_agent_loop(
        vec![user("start")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None, // memory_provider — test default
    )
    .await;
    drop(tx);

    assert!(
        *saw_interrupt_on_second.lock().unwrap(),
        "second LLM call should see the injected interrupt"
    );

    // Returned messages include the injected interrupt.
    let user_contents: Vec<String> = messages
        .iter()
        .filter_map(|m| match m {
            LoopMessage::User(u) => Some(u.text_joined()),
            _ => None,
        })
        .collect();
    assert_eq!(user_contents, vec!["start", "interrupt"]);

    // The interrupt's message_start fires AFTER the tool
    // result's message_end. We verify by event ordering.
    let events = drain(&mut rx).await;
    let interrupt_idx = events.iter().position(|e| match e {
        LoopEvent::MessageStart {
            message: LoopMessage::User(u),
        } => u.text_joined() == "interrupt",
        _ => false,
    });
    let last_tool_result_end_idx = events.iter().rposition(|e| {
        matches!(
            e,
            LoopEvent::MessageEnd {
                message: LoopMessage::ToolResult(_)
            }
        )
    });
    assert!(
        interrupt_idx.unwrap() > last_tool_result_end_idx.unwrap(),
        "interrupt should appear AFTER the tool result message_end"
    );
}

// ============================================================
// Phase 6 — regression tests for hardening paths
// ============================================================

use crate::agent::agent_loop::result::LoopToolResult as PhaseSixToolResult;
use std::sync::Arc as PhaseSixArc;

/// Phase 6: a multi-turn run with a network error in turn 2
/// preserves the FULL history (user prompt, turn 1's
/// assistant + tool-result) across the retry. The retry
/// wrapper isn't directly invoked here (we use mock
/// StreamFn), but the LOOP's context.messages survival
/// across turn errors is the invariant.
///
/// We verify by counting context.messages entries the
/// second LLM call observes. The mock StreamFn captures
/// what each call sees.
#[tokio::test]
async fn loop_preserves_history_across_turns() {
    use crate::agent::agent_loop::stream::{LlmContext, StreamFn};
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let observed_lens: PhaseSixArc<Mutex<Vec<usize>>> = PhaseSixArc::new(Mutex::new(Vec::new()));
    let observed_clone = observed_lens.clone();
    let counter = std::sync::Arc::new(AtomicUsize::new(0));

    // Inline echo tool — needed for the tool-result turn
    // that grows the history.
    #[derive(Debug)]
    struct LocalEcho;
    impl LoopTool for LocalEcho {
        fn name(&self) -> &str {
            "echo"
        }
        fn description(&self) -> &str {
            "Echo"
        }
        fn label(&self) -> &str {
            "Echo"
        }
        fn parameters(&self) -> &Value {
            static EMPTY: std::sync::OnceLock<Value> = std::sync::OnceLock::new();
            EMPTY.get_or_init(|| serde_json::json!({"type": "object"}))
        }
        fn execute<'a>(
            &'a self,
            _id: &'a str,
            _args: Value,
            _signal: AbortSignal,
            _on_update: super::super::tool::LoopToolUpdate,
        ) -> Pin<Box<dyn Future<Output = Result<PhaseSixToolResult, String>> + Send + 'a>> {
            Box::pin(async move {
                Ok(PhaseSixToolResult {
                    content: vec![serde_json::json!({
                        "type": "text",
                        "text": "ok",
                    })],
                    details: Value::Null,
                    terminate: None,
                })
            })
        }
    }

    let factory: StreamFn = std::sync::Arc::new(move |ctx: LlmContext, _opts| {
        observed_clone.lock().unwrap().push(ctx.messages.len());
        let n = counter.fetch_add(1, Ordering::SeqCst);
        let msg = if n == 0 {
            tool_use_response("call-1", "echo", serde_json::json!({}))
        } else {
            text_response("done")
        };
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![
            crate::agent::agent_loop::message::StreamEvent::Done {
                reason,
                message: msg,
                usage: None,
            },
        ]))
    });

    let mut ctx = empty_context();
    ctx.tools.push(PhaseSixArc::new(LocalEcho));
    let mut cfg = build_config();
    cfg.tool_execution = ToolExecutionMode::Sequential;

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);
    let _ = run_agent_loop(
        vec![user("start")],
        ctx,
        cfg,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None, // memory_provider — test default
    )
    .await;

    let lens = observed_lens.lock().unwrap().clone();
    assert_eq!(lens.len(), 2, "expected two LLM calls");
    // First call sees: just user prompt → 1 message.
    assert_eq!(lens[0], 1);
    // Second call sees: user prompt + assistant (tool_use) +
    // tool result → 3 messages. History preserved.
    assert_eq!(
        lens[1], 3,
        "second LLM call should see prior turn's history; got {} messages",
        lens[1],
    );
}

/// dirge-j4dz: a graceful interjection raised DURING a run (e.g. the
/// permission-denial cascade) must halt the loop at the next tool-result
/// boundary. The stream here always returns a tool call, so without an
/// in-loop `is_interjected()` check the run would spin until `max_turns`.
/// With the fix it stops after the first turn.
#[tokio::test]
async fn interjection_halts_at_tool_result_boundary() {
    use crate::agent::agent_loop::stream::{LlmContext, StreamFn};
    use std::sync::atomic::{AtomicUsize, Ordering};

    // A tool that raises a graceful interjection when it runs — the same
    // signal the permission-denial cascade sets — then returns normally.
    #[derive(Debug)]
    struct InterjectingTool {
        signal: AbortSignal,
    }
    impl LoopTool for InterjectingTool {
        fn name(&self) -> &str {
            "noop"
        }
        fn description(&self) -> &str {
            "Interjecting"
        }
        fn label(&self) -> &str {
            "Noop"
        }
        fn parameters(&self) -> &Value {
            static EMPTY: std::sync::OnceLock<Value> = std::sync::OnceLock::new();
            EMPTY.get_or_init(|| serde_json::json!({"type": "object"}))
        }
        fn execute<'a>(
            &'a self,
            _id: &'a str,
            args: Value,
            _signal: AbortSignal,
            _on_update: super::super::tool::LoopToolUpdate,
        ) -> Pin<Box<dyn Future<Output = Result<super::super::LoopToolResult, String>> + Send + 'a>>
        {
            self.signal.interject();
            Box::pin(async move {
                Ok(super::super::LoopToolResult {
                    content: vec![serde_json::json!({"type": "text", "text": "ok"})],
                    details: args,
                    terminate: None,
                })
            })
        }
    }

    // Always returns a tool_use — a loop that ignores the interjection
    // would keep taking turns forever (bounded only by max_turns).
    let counter = std::sync::Arc::new(AtomicUsize::new(0));
    let seen = counter.clone();
    let factory: StreamFn = std::sync::Arc::new(move |_ctx: LlmContext, _opts| {
        counter.fetch_add(1, Ordering::SeqCst);
        let msg = tool_use_response("call-1", "noop", serde_json::json!({}));
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![
            crate::agent::agent_loop::message::StreamEvent::Done {
                reason,
                message: msg,
                usage: None,
            },
        ]))
    });

    let signal = AbortSignal::new();
    let mut ctx = empty_context();
    ctx.tools.push(PhaseSixArc::new(InterjectingTool {
        signal: signal.clone(),
    }));
    let mut cfg = build_config();
    cfg.tool_execution = ToolExecutionMode::Sequential;
    // A high cap so a spinning loop is clearly distinguishable from the
    // halt-after-one-turn behavior we want.
    cfg.max_turns = Some(25);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(256);
    let task = tokio::spawn(async move {
        run_agent_loop(
            vec![user("start")],
            ctx,
            cfg,
            signal,
            &tx,
            &factory,
            None,
            None,
        )
        .await
    });

    let result = tokio::time::timeout(std::time::Duration::from_secs(5), task).await;
    assert!(
        result.is_ok(),
        "loop should exit promptly after interjection"
    );

    let turns = seen.load(Ordering::SeqCst);
    assert_eq!(
        turns, 1,
        "interjection must halt at the first tool-result boundary; the model took {turns} turns"
    );
}

/// Phase 6: full signal-chain regression. Cancel the signal
/// mid-tool; tool aborts; loop's next LLM call's stream
/// observes the same signal and exits via Error path; loop
/// exits cleanly with no infinite-loop or hung tools.
#[tokio::test]
async fn full_signal_chain_exits_cleanly() {
    use crate::agent::agent_loop::stream::{LlmContext, StreamFn};
    use std::sync::atomic::{AtomicUsize, Ordering};

    // Mock tool that observes the signal during execution
    // (immediate cancel since the test cancels signal right
    // after spawn).
    #[derive(Debug)]
    struct CancellableTool;
    impl LoopTool for CancellableTool {
        fn name(&self) -> &str {
            "noop"
        }
        fn description(&self) -> &str {
            "Cancellable"
        }
        fn label(&self) -> &str {
            "Noop"
        }
        fn parameters(&self) -> &Value {
            static EMPTY: std::sync::OnceLock<Value> = std::sync::OnceLock::new();
            EMPTY.get_or_init(|| serde_json::json!({"type": "object"}))
        }
        fn execute<'a>(
            &'a self,
            _id: &'a str,
            _args: Value,
            _signal: AbortSignal,
            _on_update: super::super::tool::LoopToolUpdate,
        ) -> Pin<Box<dyn Future<Output = Result<PhaseSixToolResult, String>> + Send + 'a>> {
            Box::pin(async move {
                tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                Ok(PhaseSixToolResult {
                    content: Vec::new(),
                    details: Value::Null,
                    terminate: None,
                })
            })
        }
    }

    // Factory that returns a tool_use response first,
    // then would return a text response on retry (but
    // shouldn't get there because signal is cancelled
    // before turn 2).
    let counter = std::sync::Arc::new(AtomicUsize::new(0));
    let factory: StreamFn = std::sync::Arc::new(move |_ctx: LlmContext, _opts| {
        let n = counter.fetch_add(1, Ordering::SeqCst);
        let msg = if n == 0 {
            tool_use_response("call-1", "noop", serde_json::json!({}))
        } else {
            text_response("should-not-reach")
        };
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![
            crate::agent::agent_loop::message::StreamEvent::Done {
                reason,
                message: msg,
                usage: None,
            },
        ]))
    });

    let mut ctx = empty_context();
    ctx.tools.push(PhaseSixArc::new(CancellableTool));
    let mut cfg = build_config();
    cfg.tool_execution = ToolExecutionMode::Sequential;

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);
    let signal = AbortSignal::new();
    let signal_clone = signal.clone();

    // Spawn the loop in a task; cancel signal after a small
    // yield so the tool has started.
    let task = tokio::spawn(async move {
        run_agent_loop(
            vec![user("start")],
            ctx,
            cfg,
            signal_clone,
            &tx,
            &factory,
            None,
            None, // memory_provider — test default
        )
        .await
    });
    // Yield twice so the loop reaches the tool dispatch
    // before we cancel.
    for _ in 0..5 {
        tokio::task::yield_now().await;
    }
    signal.cancel();

    // Bound the test: loop must complete in <2s. Without
    // the tool-abort wrap, the 30s blocking tool would
    // exceed this. R3 ensures the next LLM call (if any)
    // also exits promptly via its pre-poll signal check.
    let result = tokio::time::timeout(std::time::Duration::from_secs(2), task).await;
    assert!(
        result.is_ok(),
        "loop should exit within 2s after signal cancel"
    );
}

// ── dirge-h5tv: build_augmented_focus + transcript helper ──

use crate::extras::memory_provider::MemoryProvider;
use std::sync::Arc;

#[derive(Default)]
struct PreCompressRecorder {
    seen: Mutex<Vec<String>>,
    return_value: Mutex<String>,
}
impl MemoryProvider for PreCompressRecorder {
    fn name(&self) -> &str {
        "pre-compress-recorder"
    }
    fn view(&self, _: &str) -> serde_json::Value {
        serde_json::Value::Null
    }
    fn add(&self, _: &str, _: &str, _kind: Option<&str>) -> Result<serde_json::Value, String> {
        Ok(serde_json::Value::Null)
    }
    fn replace(
        &self,
        _: &str,
        _: &str,
        _: &str,
        _kind: Option<&str>,
    ) -> Result<serde_json::Value, String> {
        Ok(serde_json::Value::Null)
    }
    fn remove(&self, _: &str, _: &str) -> Result<serde_json::Value, String> {
        Ok(serde_json::Value::Null)
    }
    fn on_pre_compress(&self, transcript: &str) -> String {
        self.seen.lock().unwrap().push(transcript.to_string());
        self.return_value.lock().unwrap().clone()
    }
}

fn make_middle() -> Vec<serde_json::Value> {
    vec![
        serde_json::json!({"role": "user", "content": "what is rust?"}),
        serde_json::json!({"role": "assistant", "content": "a systems language"}),
    ]
}

#[test]
fn build_augmented_focus_returns_none_with_no_inputs() {
    let result = super::build_augmented_focus(None, None, &make_middle());
    assert!(
        result.is_none(),
        "no focus + no provider must yield None instructions"
    );
}

#[test]
fn build_augmented_focus_preserves_focus_when_no_provider() {
    let result = super::build_augmented_focus(Some("error handling"), None, &make_middle());
    assert_eq!(result.as_deref(), Some("error handling"));
}

#[test]
fn build_augmented_focus_folds_provider_insights_into_focus() {
    let provider = Arc::new(PreCompressRecorder::default());
    *provider.return_value.lock().unwrap() = "user prefers async/await over threads".into();
    let provider_dyn: Arc<dyn MemoryProvider> = provider.clone();

    let result =
        super::build_augmented_focus(Some("retry logic"), Some(&provider_dyn), &make_middle());

    let out = result.expect("focus + insights produces Some");
    assert!(out.contains("retry logic"), "user focus must survive");
    assert!(
        out.contains("user prefers async/await over threads"),
        "provider insight must be folded in: {out}"
    );
    assert!(
        out.contains("Provider insights:"),
        "insights must be labelled so the summarizer can attribute them"
    );

    // Provider received the transcript built from the middle slice.
    let seen = provider.seen.lock().unwrap();
    assert_eq!(seen.len(), 1, "hook fires exactly once");
    assert!(
        seen[0].contains("user: what is rust?")
            && seen[0].contains("assistant: a systems language"),
        "transcript must contain both messages: {:?}",
        seen[0]
    );
}

#[test]
fn build_augmented_focus_yields_insights_alone_when_no_focus() {
    let provider = Arc::new(PreCompressRecorder::default());
    *provider.return_value.lock().unwrap() = "remember the build flags".into();
    let provider_dyn: Arc<dyn MemoryProvider> = provider.clone();

    let result = super::build_augmented_focus(None, Some(&provider_dyn), &make_middle());

    let out = result.expect("insights alone produce Some");
    assert!(out.starts_with("Provider insights:"));
    assert!(out.contains("remember the build flags"));
}

#[test]
fn build_augmented_focus_treats_empty_provider_output_as_none() {
    let provider = Arc::new(PreCompressRecorder::default());
    // Empty string return from on_pre_compress — provider has
    // nothing to contribute this turn.
    *provider.return_value.lock().unwrap() = "".into();
    let provider_dyn: Arc<dyn MemoryProvider> = provider.clone();

    let result = super::build_augmented_focus(None, Some(&provider_dyn), &make_middle());
    assert!(
        result.is_none(),
        "empty provider output + no focus must yield None"
    );

    // But the hook still fired (so it can do internal bookkeeping
    // even if its return is empty).
    assert_eq!(provider.seen.lock().unwrap().len(), 1);
}

#[test]
fn transcript_from_value_slice_renders_role_prefixes() {
    let messages = vec![
        serde_json::json!({"role": "user", "content": "hello"}),
        serde_json::json!({"role": "assistant", "content": "hi"}),
        serde_json::json!({"role": "system", "content": ""}), // empty — skipped
    ];
    let t = super::transcript_from_value_slice(&messages);
    assert!(t.contains("user: hello"));
    assert!(t.contains("assistant: hi"));
    assert!(
        !t.contains("system: "),
        "empty content must be skipped: {t:?}"
    );
}

#[test]
fn transcript_from_value_slice_extracts_block_array_content() {
    let messages = vec![
        serde_json::json!({"role":"assistant","content":[{"type":"text","text":"hello from assistant"}]}),
        serde_json::json!({"role":"toolResult","content":[{"type":"text","text":"tool output here"}]}),
    ];
    let t = super::transcript_from_value_slice(&messages);
    assert!(t.contains("assistant: hello from assistant"));
    assert!(t.contains("toolResult: tool output here"));
}

/// The critic transcript feeds a LOAD-BEARING critic prompt (the F6 in-loop
/// critic just had a stale-summary bug fixed). Pin its exact output byte-for-
/// byte so a refactor can't silently shift the `USER:`/`ASSISTANT:` labels, the
/// `ASSISTANT called name(args)` tool-call rendering, the `TOOL name [tag]: …`
/// result line, or the trimming — none of which the `.contains()` tests catch.
#[test]
fn build_critic_transcript_pins_the_exact_critic_facing_format() {
    use crate::agent::agent_loop::message::ToolResultMessage;
    let msgs = vec![
        user("do the thing"),
        LoopMessage::Assistant(AssistantMessage::new(
            vec![
                ContentBlock::Text {
                    text: "  on it  ".to_string(),
                },
                ContentBlock::ToolCall {
                    id: "c1".to_string(),
                    name: "read".to_string(),
                    arguments: serde_json::json!({"path": "/x"}),
                },
            ],
            StopReason::Stop,
        )),
        LoopMessage::ToolResult(ToolResultMessage {
            tool_call_id: "c1".to_string(),
            tool_name: "read".to_string(),
            content: vec![ContentBlock::Text {
                text: "file contents".to_string(),
            }],
            details: serde_json::json!({}),
            is_error: false,
        }),
    ];
    assert_eq!(
        super::build_critic_transcript(&msgs),
        "USER: do the thing\n\
         ASSISTANT: on it\n\
         ASSISTANT called read({\"path\":\"/x\"})\n\
         TOOL read [result]: file contents\n",
    );
}

/// dirge-kk3x: a permission/approval denial is tagged `[DENIED]`, not the
/// generic `[ERROR]`, so the critic reads it as a policy wall (out of scope)
/// rather than a failure to demand the assistant fix.
#[test]
fn build_critic_transcript_marks_permission_denials_as_denied() {
    use crate::agent::agent_loop::message::ToolResultMessage;
    let msgs = vec![
        user("commit and push"),
        LoopMessage::ToolResult(ToolResultMessage {
            tool_call_id: "c1".to_string(),
            tool_name: "bash".to_string(),
            content: vec![ContentBlock::Text {
                text: "Permission denied: git is outside the project directory".to_string(),
            }],
            details: serde_json::json!({}),
            is_error: true,
        }),
        // A non-denial error keeps the generic ERROR tag.
        LoopMessage::ToolResult(ToolResultMessage {
            tool_call_id: "c2".to_string(),
            tool_name: "edit".to_string(),
            content: vec![ContentBlock::Text {
                text: "old_string not found".to_string(),
            }],
            details: serde_json::json!({}),
            is_error: true,
        }),
    ];
    let t = super::build_critic_transcript(&msgs);
    assert!(t.contains("TOOL bash [DENIED]: Permission denied"), "{t}");
    assert!(t.contains("TOOL edit [ERROR]: old_string not found"), "{t}");
}

/// dirge-kk3x regression: the [DENIED] tag is gated on `is_error`, mirroring
/// Outcome::classify. A SUCCESSFUL result whose text merely begins
/// "Permission denied" — e.g. bash returns Ok(text) for a failed `ssh` whose
/// output is "Permission denied (publickey).\nExit code: 255" — must NOT be
/// tagged [DENIED], or the critic would excuse genuinely unfinished work.
#[test]
fn build_critic_transcript_does_not_mark_successful_permission_denied_text() {
    use crate::agent::agent_loop::message::ToolResultMessage;
    let msgs = vec![
        user("ssh to the box and deploy"),
        LoopMessage::ToolResult(ToolResultMessage {
            tool_call_id: "c1".to_string(),
            tool_name: "bash".to_string(),
            content: vec![ContentBlock::Text {
                text: "Permission denied (publickey).\nExit code: 255".to_string(),
            }],
            details: serde_json::json!({}),
            // bash surfaces a failed command as a non-error result.
            is_error: false,
        }),
    ];
    let t = super::build_critic_transcript(&msgs);
    assert!(
        t.contains("TOOL bash [result]: Permission denied (publickey)."),
        "a non-error result must keep the [result] tag, not [DENIED]: {t}"
    );
    assert!(!t.contains("[DENIED]"), "{t}");
}

/// Regression (dirge-p9qm): in a long run the head is planning/scaffolding
/// and the implementation + verification land at the END. The builder used
/// to keep only the FIRST 8000 chars, so the critic was fed the planning and
/// never saw the work — wrongly reporting "nothing done". The transcript must
/// keep the original request (head) AND the most recent activity (tail).
#[test]
fn build_critic_transcript_keeps_request_and_recent_work_when_over_budget() {
    use crate::agent::agent_loop::message::ToolResultMessage;
    let mut msgs = vec![user("REQUEST: build an animated water canvas")];
    // Planning chatter that blows well past the budget.
    for i in 0..120 {
        msgs.push(LoopMessage::Assistant(AssistantMessage::new(
            vec![ContentBlock::Text {
                text: format!("planning step {i}: {}", "x".repeat(200)),
            }],
            StopReason::Stop,
        )));
    }
    // The actual work + verification, at the end of the run.
    msgs.push(LoopMessage::Assistant(AssistantMessage::new(
        vec![ContentBlock::Text {
            text: "DONE: created water.js + flowfield.js; tests 12/12 pass".to_string(),
        }],
        StopReason::Stop,
    )));
    msgs.push(LoopMessage::ToolResult(ToolResultMessage {
        tool_call_id: "v".to_string(),
        tool_name: "bash".to_string(),
        content: vec![ContentBlock::Text {
            text: "VERIFIED: WATER RENDERED (cyan/blue flow field)".to_string(),
        }],
        details: serde_json::json!({}),
        is_error: false,
    }));

    let t = super::build_critic_transcript(&msgs);
    assert!(
        t.contains("REQUEST: build an animated water canvas"),
        "original request (head) must survive truncation"
    );
    assert!(
        t.contains("WATER RENDERED"),
        "recent verification (tail) must survive — this is what the critic judges"
    );
    assert!(
        t.contains("tests 12/12 pass"),
        "recent work (tail) must survive"
    );
    assert!(
        t.contains("elided"),
        "an elision marker should mark the dropped middle"
    );
}

// =====================================================================
// dirge-ngic — scavenge must inspect both Thinking AND Text blocks.
// Reasonix combines both at `loop.ts:910-913` →
// `repair/index.ts:71`. Previously dirge merged only Thinking, so
// any DSML invoke that streamed as visible content (the common
// case on Anthropic cache hits) was lost.
// =====================================================================

/// dirge-ngic: a DSML invoke that lives only in `ContentBlock::Text`
/// (no Thinking block at all) must be picked up by the scavenger.
/// Proves the run.rs source builder includes Text — without the
/// fix this orphan call goes unrecovered, the model loop stalls
/// waiting for a tool result that never dispatches.
#[test]
fn scavenge_source_recovers_dsml_invoke_from_text_only() {
    let dsml = "<|DSML|invoke name=\"read_file\"><|DSML|parameter name=\"path\" string=\"true\">/tmp/x</|DSML|parameter></|DSML|invoke>";
    let blocks = vec![ContentBlock::Text {
        text: dsml.to_string(),
    }];

    let source = super::build_scavenge_source(&blocks);
    assert!(
        source.contains("DSML"),
        "scavenge source must include Text block content: {source:?}",
    );

    let allowed: std::collections::HashSet<String> =
        ["read_file".to_string()].into_iter().collect();
    let result =
        crate::agent::agent_loop::scavenge::scavenge_tool_calls(Some(&source), &allowed, 4);
    assert_eq!(
        result.calls.len(),
        1,
        "orphan DSML in Text must be recovered: calls={:?}",
        result.calls
    );
    assert_eq!(result.calls[0].name, "read_file");
}

/// dirge-ngic: mixed Thinking + Text content — both contribute to
/// the scavenge corpus. Order is preserved (Thinking first as it
/// streams first), separated by `\n` so DSML on a line boundary
/// doesn't merge with surrounding chatter.
#[test]
fn scavenge_source_concatenates_thinking_and_text() {
    let blocks = vec![
        ContentBlock::Thinking {
            text: "Plan: call list_dir.".to_string(),
        },
        ContentBlock::Text {
            text: "Acting now.".to_string(),
        },
    ];
    let source = super::build_scavenge_source(&blocks);
    assert_eq!(source, "Plan: call list_dir.\nActing now.");
}

/// dirge-ngic: tool-call and other non-text blocks contribute
/// nothing to the scavenge corpus — only Thinking and Text.
#[test]
fn scavenge_source_skips_non_text_blocks() {
    let blocks = vec![
        ContentBlock::Text {
            text: "visible".to_string(),
        },
        ContentBlock::ToolCall {
            id: "call_1".to_string(),
            name: "noop".to_string(),
            arguments: serde_json::json!({}),
        },
    ];
    let source = super::build_scavenge_source(&blocks);
    assert_eq!(source, "visible");
}

// =====================================================================
// dirge-7bwx — truncation repair must run BEFORE storm so two
// streams whose raw args differ but heal to the same form dedupe
// under the storm filter. Reasonix order: `repair/index.ts:88-109`
// (truncation) then `:113-121` (storm).
// =====================================================================

/// dirge-7bwx: two ToolCalls with different truncated arg strings
/// that repair to the same canonical form must, after
/// `apply_truncation_repair`, present identical parsed arguments.
/// Pre-fix these survived storm because their pre-repair raw
/// strings hashed differently and only got repaired at dispatch
/// time, after the de-dupe window had closed.
#[test]
fn truncation_repair_canonicalizes_divergent_streams_before_storm() {
    use crate::agent::agent_loop::tool_input_repair::{RepairKind, RepairStats};
    use crate::agent::agent_loop::tools::ToolCall;

    // Same logical call, different truncation points.
    let call_a_raw = r#"{"path": "/tmp/x""#; // unterminated object
    let call_b_raw = r#"{"path": "/tmp/x"}"#; // already complete
    // Quick sanity: distinct strings → distinct pre-repair sigs.
    assert_ne!(call_a_raw, call_b_raw);

    let mut tool_calls = vec![
        ToolCall {
            id: "call_a".to_string(),
            name: "read_file".to_string(),
            arguments: serde_json::Value::String(call_a_raw.to_string()),
        },
        ToolCall {
            id: "call_b".to_string(),
            name: "read_file".to_string(),
            arguments: serde_json::Value::String(call_b_raw.to_string()),
        },
    ];

    let stats = RepairStats::new();
    let notes = std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::<
        String,
        Vec<String>,
    >::new()));
    super::apply_truncation_repair(&mut tool_calls, &stats, &notes);

    // Truncated A repaired; B was already valid JSON-as-string but
    // parsed-and-replaced.
    assert_eq!(tool_calls[0].arguments, tool_calls[1].arguments);
    assert_eq!(tool_calls[0].arguments["path"], "/tmp/x");
    assert!(
        stats.snapshot().truncation_fixed >= 1,
        "at least the truncated call must record TruncationFixed",
    );
}

/// dirge-7bwx: hard-fallback (closer can't rebalance) does NOT
/// replace arguments. Original `Value::String(raw)` is preserved
/// so `validate_and_repair` downstream surfaces a real validation
/// error rather than silently dispatching a fabricated value —
/// matches Reasonix's invariant at `repair/index.ts:93-102`.
/// Review-fix #1: telemetry STILL records the truncation event
/// (Reasonix bumps `truncationsFixed` on fallback at
/// `repair/index.ts:99`) so operators see unrecoverable-rate.
/// Review-fix #2: notes are emitted with the
/// `⚠️ TRUNCATION UNRECOVERABLE` prefix Reasonix uses at `:101`.
#[test]
fn truncation_repair_preserves_raw_on_hard_fallback() {
    use crate::agent::agent_loop::tool_input_repair::RepairStats;
    use crate::agent::agent_loop::tools::ToolCall;

    let unsalvageable = "}}}garbage no opening".to_string();
    let mut tool_calls = vec![ToolCall {
        id: "call_garbage".to_string(),
        name: "read_file".to_string(),
        arguments: serde_json::Value::String(unsalvageable.clone()),
    }];

    let stats = RepairStats::new();
    let notes = std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::<
        String,
        Vec<String>,
    >::new()));
    super::apply_truncation_repair(&mut tool_calls, &stats, &notes);

    // Either preserved as the same Value::String, OR if the
    // closer happened to find a structured interpretation, it
    // must NOT be the empty/fabricated `{}` that masks a real
    // error. We test the strict case where fallback fires.
    if let serde_json::Value::String(after) = &tool_calls[0].arguments {
        assert_eq!(
            after, &unsalvageable,
            "hard fallback must not mutate the raw string",
        );
    }
    // Empty object is the canonical fabricated value Reasonix
    // refuses to emit; assert we never silently substitute it.
    assert_ne!(
        tool_calls[0].arguments,
        serde_json::json!({}),
        "hard fallback must not silently fabricate an empty object",
    );

    // dirge-7bwx review-fix #1: Reasonix parity — the counter
    // bumps on hard-fallback too (`repair/index.ts:99`).
    assert_eq!(
        stats.snapshot().truncation_fixed,
        1,
        "fallback must still bump truncation_fixed for operator telemetry",
    );

    // dirge-7bwx review-fix #2: the per-call notes carry the
    // `⚠️ TRUNCATION UNRECOVERABLE` prefix Reasonix uses at
    // `repair/index.ts:101`, attributed to the tool name.
    let sink = notes.lock().unwrap();
    let entry = sink
        .get("call_garbage")
        .expect("notes must be recorded for the fallback call");
    assert!(
        entry.iter().any(|n| n.contains("TRUNCATION UNRECOVERABLE")),
        "expected ⚠️ TRUNCATION UNRECOVERABLE prefix in notes: {entry:?}",
    );
    assert!(
        entry.iter().any(|n| n.contains("[read_file]")),
        "expected [tool_name] prefix in notes: {entry:?}",
    );
}

/// dirge-7bwx review-fix #3+5: end-to-end wiring proof. Drives
/// `run_agent_loop` with a canned assistant message that emits
/// THREE tool calls whose raw arg strings differ but heal to
/// the same canonical form. Default storm threshold is 3, so:
///   - pre-fix: 3 distinct raw `Value::String`s → 3 distinct
///     storm signatures → 3 executions, 0 suppressed.
///   - post-fix: `apply_truncation_repair` heals all three to
///     identical `Value::Object` BEFORE `storm.filter_calls`,
///     so storm's third entry hits `count >= threshold-1` and
///     suppresses → 2 executions + 1 storm-suppress.
/// This test would FAIL on the pre-hoist code (validate_and_repair
/// only ran post-storm), proving the wiring fix is live.
#[tokio::test]
async fn dirge_7bwx_end_to_end_storm_dedupes_after_truncation_repair() {
    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo.clone());

    // Three calls whose raws differ but heal to the same form.
    // `{"v":1` and `{"v": 1` and `{"v":1 ` all heal to {"v":1}.
    fn truncated(raw: &str) -> serde_json::Value {
        serde_json::Value::String(raw.to_string())
    }
    let response = AssistantMessage::new(
        vec![
            ContentBlock::ToolCall {
                id: "tool-1".to_string(),
                name: "echo".to_string(),
                arguments: truncated(r#"{"v":1"#), // tight
            },
            ContentBlock::ToolCall {
                id: "tool-2".to_string(),
                name: "echo".to_string(),
                arguments: truncated(r#"{"v": 1"#), // single space
            },
            ContentBlock::ToolCall {
                id: "tool-3".to_string(),
                name: "echo".to_string(),
                arguments: truncated(r#"{"v":  1"#), // double space
            },
        ],
        StopReason::ToolUse,
    );
    let factory = canned_factory(vec![response, text_response("done")]);

    let (tx, mut rx) = mpsc::channel::<LoopEvent>(128);
    let config = build_config();
    let repair_stats = config.repair_stats.clone();
    let _messages = run_agent_loop(
        vec![user("echo")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    // Storm default threshold=3 → first two pass, third is
    // suppressed. If the truncation hoist hadn't fired, all
    // three raws would have hashed differently and all three
    // would have executed.
    let executed_count = echo.executed.lock().unwrap().len();
    assert_eq!(
        executed_count, 2,
        "storm must catch the 3rd identical-post-repair call; got {executed_count} executions",
    );

    // Truncation repair recorded for all three.
    let snap = repair_stats.snapshot();
    assert_eq!(
        snap.truncation_fixed, 3,
        "truncation_fixed must be incremented per truncated call; got {snap:?}",
    );

    // Event stream: exactly two ToolExecutionEnd events.
    let events = drain(&mut rx).await;
    let execution_ends = events
        .iter()
        .filter(|e| e.kind() == "tool_execution_end")
        .count();
    assert_eq!(
        execution_ends,
        2,
        "expected 2 tool_execution_end events; got events={:?}",
        events.iter().map(|e| e.kind()).collect::<Vec<_>>(),
    );
}

/// Storm-breaker graceful failure: when the run gives up because
/// it's stuck looping the same call, it must surface a first-person
/// assistant explanation (not an empty/abrupt stop). Drives the loop
/// with the same single tool call repeated across turns: storm
/// suppresses it, the first all-suppressed turn injects the
/// self-correct nudge, and the next reaches the terminal branch —
/// which appends the failure narrative as an assistant message.
#[tokio::test]
async fn storm_terminal_emits_failure_narrative() {
    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo.clone());

    // Five identical echo calls (distinct ids so each turn's results
    // pair cleanly). Default storm threshold is 3.
    let make = |i: usize| {
        AssistantMessage::new(
            vec![ContentBlock::ToolCall {
                id: format!("call-{i}"),
                name: "echo".to_string(),
                arguments: serde_json::json!({"v": 1}),
            }],
            StopReason::ToolUse,
        )
    };
    let factory = canned_factory((0..5).map(make).collect());

    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let config = build_config();
    let messages = run_agent_loop(
        vec![user("echo")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    let has_narrative = messages.iter().any(|m| match m {
        LoopMessage::Assistant(a) => a.content.iter().any(|b| match b {
            ContentBlock::Text { text } => text.contains("stopped here to avoid spinning"),
            _ => false,
        }),
        _ => false,
    });
    assert!(
        has_narrative,
        "expected a storm failure-narrative assistant message; got {} messages",
        messages.len()
    );
}

/// dirge-ngic review-fix #3: end-to-end wiring proof for the
/// scavenge-source fix. Drives `run_agent_loop` with a canned
/// assistant message containing a DSML invoke ONLY in
/// `ContentBlock::Text` (no Thinking block, no declared
/// ToolCall). The loop must build the scavenge corpus from
/// Text (build_scavenge_source includes both Thinking and Text)
/// and dispatch the recovered call. Pre-fix this orphan would
/// not be recovered and zero executions would happen.
#[tokio::test]
async fn dirge_ngic_end_to_end_orphan_dsml_in_text_dispatches() {
    let echo = std::sync::Arc::new(EchoTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(echo.clone());

    // DSML invoke in Text only, no declared tool_calls. Empty
    // ToolUse-stopped message means scavenge is the ONLY path
    // to dispatch.
    let dsml = r#"<|DSML|invoke name="echo"><|DSML|parameter name="v" string="false">1</|DSML|parameter></|DSML|invoke>"#;
    let response = AssistantMessage::new(
        vec![ContentBlock::Text {
            text: dsml.to_string(),
        }],
        StopReason::ToolUse,
    );
    let factory = canned_factory(vec![response, text_response("done")]);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let config = build_config();
    let _messages = run_agent_loop(
        vec![user("echo")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    // Pre-fix: scavenge_source only had Thinking → empty
    // corpus → no scavenged call → 0 executions. Post-fix:
    // Text is included → DSML recovered → 1 execution.
    let executed = echo.executed.lock().unwrap();
    assert_eq!(
        executed.len(),
        1,
        "orphan DSML in Text must be recovered and dispatched (post-dirge-ngic); got {} executions",
        executed.len(),
    );
}

// =====================================================================
// dirge-knt8: scavenged calls with invalid args must be silently
// dropped — NOT turned into error tool results that force an extra
// turn. Reasoning models (deepseek/glm) sometimes put tool-call-
// shaped JSON/DSML in their final answer text. The scavenger lifts
// these into phantom tool calls, but if the args don't match the
// tool's schema the call must be dropped, not dispatched as an
// error. Native tool calls (provider-emitted tool_calls) keep their
// existing error behavior.
// =====================================================================

/// Test tool with a typed schema requiring "path" (string). Used to
/// verify that scavenged calls failing schema validation are dropped
/// while native calls still produce error results.
#[derive(Debug)]
struct TypedPathTool {
    executed: std::sync::Arc<Mutex<Vec<Value>>>,
}
impl TypedPathTool {
    fn new() -> Self {
        Self {
            executed: std::sync::Arc::new(Mutex::new(Vec::new())),
        }
    }
}
impl LoopTool for TypedPathTool {
    fn name(&self) -> &str {
        "typed_path_tool"
    }
    fn description(&self) -> &str {
        "Tool requiring a path string"
    }
    fn label(&self) -> &str {
        "TypedPathTool"
    }
    fn parameters(&self) -> &Value {
        static SCHEMA: std::sync::OnceLock<Value> = std::sync::OnceLock::new();
        SCHEMA.get_or_init(|| {
            serde_json::json!({
                "type": "object",
                "properties": {
                    "path": { "type": "string" }
                },
                "required": ["path"]
            })
        })
    }
    fn execute<'a>(
        &'a self,
        _id: &'a str,
        args: Value,
        _signal: AbortSignal,
        _on_update: LoopToolUpdate,
    ) -> Pin<Box<dyn Future<Output = Result<super::super::LoopToolResult, String>> + Send + 'a>>
    {
        let executed = self.executed.clone();
        Box::pin(async move {
            executed.lock().unwrap().push(args.clone());
            Ok(super::super::LoopToolResult {
                content: vec![serde_json::json!({"type": "text", "text": "ok"})],
                details: args,
                terminate: None,
            })
        })
    }
}

/// dirge-knt8 test 1: a scavenged DSML invoke with args that fail
/// the tool's schema MUST be dropped silently — the tool is never
/// executed, no error tool result is produced, and the loop does
/// NOT force a continuation turn.
#[tokio::test]
async fn scavenged_call_invalid_args_dropped() {
    let tool = std::sync::Arc::new(TypedPathTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(tool.clone());

    // DSML invoke with NO parameters — scavenger produces {} which
    // fails schema validation (required "path" missing).
    let dsml = r#"<|DSML|invoke name="typed_path_tool"></|DSML|invoke>"#;
    let response = AssistantMessage::new(
        vec![ContentBlock::Text {
            text: dsml.to_string(),
        }],
        StopReason::ToolUse,
    );
    // Second canned response: the loop must NOT reach this because
    // no continuation is forced after dropping the invalid scavenged
    // call. If this appears, the bug is still present.
    let factory = canned_factory(vec![
        response,
        text_response("BUG-still-forcing-continuation"),
    ]);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let config = build_config();
    let messages = run_agent_loop(
        vec![user("test")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    // The tool must NOT have been executed.
    let executed = tool.executed.lock().unwrap();
    assert!(
        executed.is_empty(),
        "invalid scavenged call must be dropped, not dispatched; got {} executions",
        executed.len(),
    );

    // No error tool result must exist.
    let error_count = messages
        .iter()
        .filter(|m| matches!(m, LoopMessage::ToolResult(tr) if tr.is_error))
        .count();
    assert_eq!(
        error_count, 0,
        "invalid scavenged call must not produce error tool result; got {error_count}"
    );

    // The "BUG" continuation message must not appear — the loop
    // must terminate without forcing an extra turn.
    for msg in &messages {
        if let LoopMessage::Assistant(a) = msg {
            for block in &a.content {
                if let ContentBlock::Text { text } = block {
                    assert!(
                        !text.contains("BUG-still-forcing-continuation"),
                        "loop must not force continuation after dropping invalid scavenged call"
                    );
                }
            }
        }
    }
}

/// dirge-knt8 test 2: a scavenged DSML invoke with VALID args
/// (matching the tool's schema) still executes normally. Proves
/// the validation gate doesn't break the valid scavenge path.
#[tokio::test]
async fn scavenged_call_valid_args_still_executes() {
    let tool = std::sync::Arc::new(TypedPathTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(tool.clone());

    // DSML invoke with valid "path" parameter matching the schema.
    let dsml = r#"<|DSML|invoke name="typed_path_tool"><|DSML|parameter name="path" string="true">/tmp/x</|DSML|parameter></|DSML|invoke>"#;
    let response = AssistantMessage::new(
        vec![ContentBlock::Text {
            text: dsml.to_string(),
        }],
        StopReason::ToolUse,
    );
    let factory = canned_factory(vec![response, text_response("done")]);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let config = build_config();
    let _messages = run_agent_loop(
        vec![user("test")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    // Tool must have been called once with correct args.
    let executed = tool.executed.lock().unwrap();
    assert_eq!(
        executed.len(),
        1,
        "valid scavenged call must dispatch; got {} executions",
        executed.len(),
    );
    assert_eq!(
        executed[0]["path"], "/tmp/x",
        "valid scavenged call args must be preserved"
    );
}

/// dirge-knt8 test 3 (regression guard): a NATIVE tool call (from
/// the provider's `tool_calls`, not scavenged from text) with
/// invalid args MUST still produce an error tool result and force
/// continuation. The fix only touches scavenged calls — native
/// error behavior is unchanged.
#[tokio::test]
async fn native_call_invalid_args_still_errors() {
    let tool = std::sync::Arc::new(TypedPathTool::new());
    let mut ctx = empty_context();
    ctx.tools.push(tool.clone());

    // Native tool call with invalid args (missing required "path").
    let response = AssistantMessage::new(
        vec![ContentBlock::ToolCall {
            id: "call_native_1".to_string(),
            name: "typed_path_tool".to_string(),
            arguments: serde_json::json!({"wrong_param": 1}),
        }],
        StopReason::ToolUse,
    );
    let factory = canned_factory(vec![response, text_response("loop-continued-after-error")]);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(128);
    let config = build_config();
    let messages = run_agent_loop(
        vec![user("test")],
        ctx,
        config,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    // Tool must NOT have been executed (validation fails before dispatch).
    let executed = tool.executed.lock().unwrap();
    assert!(
        executed.is_empty(),
        "native call with invalid args must not execute; got {} executions",
        executed.len(),
    );

    // Must have at least one error tool result.
    let error_count = messages
        .iter()
        .filter(|m| matches!(m, LoopMessage::ToolResult(tr) if tr.is_error))
        .count();
    assert!(
        error_count > 0,
        "native invalid call must produce error tool result"
    );

    // Loop must have continued — "loop-continued-after-error" must appear.
    let has_continuation = messages.iter().any(|msg| {
        if let LoopMessage::Assistant(a) = msg {
            a.content.iter().any(|b| {
                if let ContentBlock::Text { text } = b {
                    text.contains("loop-continued-after-error")
                } else {
                    false
                }
            })
        } else {
            false
        }
    });
    assert!(
        has_continuation,
        "loop must continue after native invalid call error"
    );
}

/// dirge-7bwx review-fix #2: successful repair also forwards
/// notes (without the unrecoverable prefix) so the model sees
/// what was fixed. Reasonix parity at `repair/index.ts:106`.
#[test]
fn truncation_repair_forwards_notes_on_successful_repair() {
    use crate::agent::agent_loop::tool_input_repair::RepairStats;
    use crate::agent::agent_loop::tools::ToolCall;

    let truncated = r#"{"path": "/tmp/x"#; // unterminated string
    let mut tool_calls = vec![ToolCall {
        id: "call_ok".to_string(),
        name: "read_file".to_string(),
        arguments: serde_json::Value::String(truncated.to_string()),
    }];

    let stats = RepairStats::new();
    let notes = std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::<
        String,
        Vec<String>,
    >::new()));
    super::apply_truncation_repair(&mut tool_calls, &stats, &notes);

    // Args were promoted to the parsed form.
    assert_eq!(tool_calls[0].arguments["path"], "/tmp/x");
    // Counter bumped on success too.
    assert_eq!(stats.snapshot().truncation_fixed, 1);
    // Notes attributed to the tool, WITHOUT the unrecoverable
    // prefix.
    let sink = notes.lock().unwrap();
    let entry = sink
        .get("call_ok")
        .expect("notes must be recorded for the successful repair");
    assert!(entry.iter().any(|n| n.contains("[read_file]")));
    assert!(
        entry
            .iter()
            .all(|n| !n.contains("TRUNCATION UNRECOVERABLE")),
        "successful repair must not carry the unrecoverable prefix: {entry:?}",
    );
}

/// dirge-7bwx: structurally valid args (real `Value::Object`)
/// pass through untouched — only `Value::String` triggers the
/// repair pass.
#[test]
fn truncation_repair_leaves_already_parsed_args_alone() {
    use crate::agent::agent_loop::tool_input_repair::{RepairKind, RepairStats};
    use crate::agent::agent_loop::tools::ToolCall;

    let already_parsed = serde_json::json!({ "path": "/tmp/y" });
    let mut tool_calls = vec![ToolCall {
        id: "call_ok".to_string(),
        name: "read_file".to_string(),
        arguments: already_parsed.clone(),
    }];

    let stats = RepairStats::new();
    let notes = std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::<
        String,
        Vec<String>,
    >::new()));
    super::apply_truncation_repair(&mut tool_calls, &stats, &notes);

    assert_eq!(tool_calls[0].arguments, already_parsed);
    assert_eq!(
        stats.snapshot().truncation_fixed,
        0,
        "no repair should be recorded for already-parsed args",
    );
}

// ============================================================
// dirge-k6be — turn-end per-tool-result cap wiring
// ============================================================

/// dirge-k6be end-to-end: a tool that returns a 60 KB result
/// drops into the transcript verbatim, but the NEXT model
/// call must see the capped form. Proves `run_loop` calls
/// `cap_oversized_tool_results` before each
/// `stream_assistant_response`, matching Reasonix
/// `loop.ts:486-503` (`healActiveLogBeforeSend`).
#[tokio::test]
async fn dirge_k6be_oversized_tool_result_capped_before_next_model_call() {
    use crate::agent::agent_loop::stream::{LlmContext, StreamFn};
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicUsize, Ordering};

    // Tool that returns ~60 KB so it's well over the 3000-token
    // (12 KB) cap.
    #[derive(Debug)]
    struct BigOutputTool;
    impl LoopTool for BigOutputTool {
        fn name(&self) -> &str {
            "big_read"
        }
        fn description(&self) -> &str {
            "Big tool"
        }
        fn label(&self) -> &str {
            "BigRead"
        }
        fn parameters(&self) -> &Value {
            static EMPTY: std::sync::OnceLock<Value> = std::sync::OnceLock::new();
            EMPTY.get_or_init(|| serde_json::json!({"type": "object"}))
        }
        fn execute<'a>(
            &'a self,
            _id: &'a str,
            _args: Value,
            _signal: AbortSignal,
            _on_update: super::super::tool::LoopToolUpdate,
        ) -> Pin<Box<dyn Future<Output = Result<super::super::LoopToolResult, String>> + Send + 'a>>
        {
            let huge = "x".repeat(60_000);
            Box::pin(async move {
                Ok(super::super::LoopToolResult {
                    content: vec![serde_json::json!({
                        "type": "text",
                        "text": huge,
                    })],
                    details: Value::Null,
                    terminate: None,
                })
            })
        }
    }

    // Capture what each model call sees so we can assert the
    // tool result was capped before the second call.
    let observed_second_call_payload: std::sync::Arc<Mutex<Option<Vec<Value>>>> =
        std::sync::Arc::new(Mutex::new(None));
    let observed_clone = observed_second_call_payload.clone();
    let counter = std::sync::Arc::new(AtomicUsize::new(0));

    let factory: StreamFn = std::sync::Arc::new(move |ctx: LlmContext, _opts| {
        let n = counter.fetch_add(1, Ordering::SeqCst);
        if n == 1 {
            *observed_clone.lock().unwrap() = Some(ctx.messages.clone());
        }
        let msg = if n == 0 {
            tool_use_response("call-1", "big_read", serde_json::json!({}))
        } else {
            text_response("done")
        };
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![
            crate::agent::agent_loop::message::StreamEvent::Done {
                reason,
                message: msg,
                usage: None,
            },
        ]))
    });

    let mut ctx = empty_context();
    ctx.tools.push(std::sync::Arc::new(BigOutputTool));
    let mut cfg = build_config();
    cfg.tool_execution = ToolExecutionMode::Sequential;

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);
    let _ = run_agent_loop(
        vec![user("start")],
        ctx,
        cfg,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;

    let observed = observed_second_call_payload.lock().unwrap();
    let messages = observed
        .as_ref()
        .expect("second model call must have happened");

    // Find the tool-result message in the payload the model
    // saw on call #2.
    let tool_result = messages
        .iter()
        .find(|m| {
            m.get("role").and_then(|v| v.as_str()) == Some("toolResult")
                || m.get("role").and_then(|v| v.as_str()) == Some("tool")
        })
        .expect("second call must include the tool result");

    // The result must be CAPPED — its content's total text
    // length is far below the original 60 KB. The 3000-token
    // cap = 12 KB; allow some slack for marker overhead.
    let blocks = tool_result["content"]
        .as_array()
        .expect("tool result content should be an array of blocks");
    let total_text_len: usize = blocks
        .iter()
        .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
        .map(|t| t.len())
        .sum();
    assert!(
        total_text_len < 60_000,
        "tool result must be capped before the second model call; got {total_text_len} chars",
    );
    assert!(
        total_text_len < 14_000,
        "capped result must be near the ~12 KB cap; got {total_text_len} chars",
    );
    // And the marker must be present.
    let combined: String = blocks
        .iter()
        .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
        .collect();
    assert!(
        combined.contains("truncated"),
        "capped result must carry the truncation marker",
    );
}

// ============================================================
// dirge-el3n — proactive turn-start fold wiring
// ============================================================

/// dirge-el3n end-to-end: when the message log is loaded with
/// content over 90% of the context window AT TURN START, the
/// proactive fold fires before the next model call. Without
/// the fix the warning was logged but nothing was shrunk.
/// Asserts the second LLM call sees a SMALLER context than
/// the loaded one — proving the fold actually ran.
#[tokio::test]
async fn dirge_el3n_proactive_fold_fires_when_threshold_crossed_at_turn_start() {
    use crate::agent::agent_loop::stream::{LlmContext, StreamFn};
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicUsize, Ordering};

    // Pre-load a context that's well over 90% of the
    // 128_000-token default ctx window. 130_000 chars / 4 ≈
    // 32_500 tokens. To cross 0.9 ratio (= 115_200 tokens) we
    // need ~460_000 chars of content.
    let huge_text = "x".repeat(500_000);
    let preloaded = vec![serde_json::json!({
        "role": "toolResult",
        "content": [{"type": "text", "text": huge_text}],
        "toolName": "read",
    })];

    // Capture the message count the second model call sees.
    // After the fold, oversized tool results in the middle
    // section should have been pruned to 1-liners — total
    // string content should drop materially.
    let observed_second_call_total_chars: std::sync::Arc<Mutex<Option<usize>>> =
        std::sync::Arc::new(Mutex::new(None));
    let observed_clone = observed_second_call_total_chars.clone();
    let counter = std::sync::Arc::new(AtomicUsize::new(0));

    let factory: StreamFn = std::sync::Arc::new(move |ctx: LlmContext, _opts| {
        let n = counter.fetch_add(1, Ordering::SeqCst);
        if n == 0 {
            // Total content text on the FIRST call (the call
            // that's supposed to be preceded by the fold).
            let total: usize = ctx
                .messages
                .iter()
                .map(|m| match m.get("content") {
                    Some(serde_json::Value::String(s)) => s.len(),
                    Some(serde_json::Value::Array(blocks)) => blocks
                        .iter()
                        .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
                        .map(|t| t.len())
                        .sum(),
                    _ => 0,
                })
                .sum();
            *observed_clone.lock().unwrap() = Some(total);
        }
        let msg = text_response("ok");
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![
            crate::agent::agent_loop::message::StreamEvent::Done {
                reason,
                message: msg,
                usage: None,
            },
        ]))
    });

    let mut ctx = empty_context();
    ctx.messages = preloaded;
    let mut cfg = build_config();
    cfg.tool_execution = ToolExecutionMode::Sequential;
    // The proactive fold uses ctx_max from the model's known
    // window. With no model_name set, it defaults to 128_000.

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);
    let _ = run_agent_loop(
        vec![user("start")],
        ctx,
        cfg,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;

    let observed = observed_second_call_total_chars.lock().unwrap();
    let total_after_fold = observed.expect("first model call must have happened");
    // The fold should have shrunk the 500 KB tool-result text
    // dramatically — pruning replaces oversized tool results
    // with 1-line summaries. Pre-fix this value would have
    // been ~500_000 (no fold fired). Post-fix it must be way
    // smaller because prune_tool_outputs ran.
    assert!(
        total_after_fold < 100_000,
        "proactive fold should have shrunk the preloaded transcript; saw {total_after_fold} chars",
    );
}

/// dirge-el3n: the proactive fold does NOT fire when the
/// ratio is comfortably under threshold. Guards against
/// over-aggressive folding that would shrink useful context.
#[tokio::test]
async fn dirge_el3n_proactive_fold_does_not_fire_under_threshold() {
    use crate::agent::agent_loop::stream::{LlmContext, StreamFn};
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicUsize, Ordering};

    // Modest tool result — well under 90% of 128k token window.
    let modest = "y".repeat(4_000);
    let preloaded = vec![serde_json::json!({
        "role": "toolResult",
        "content": [{"type": "text", "text": modest}],
        "toolName": "read",
    })];

    let observed_first_call_chars: std::sync::Arc<Mutex<Option<usize>>> =
        std::sync::Arc::new(Mutex::new(None));
    let observed_clone = observed_first_call_chars.clone();
    let counter = std::sync::Arc::new(AtomicUsize::new(0));

    let factory: StreamFn = std::sync::Arc::new(move |ctx: LlmContext, _opts| {
        let n = counter.fetch_add(1, Ordering::SeqCst);
        if n == 0 {
            let total: usize = ctx
                .messages
                .iter()
                .map(|m| match m.get("content") {
                    Some(serde_json::Value::String(s)) => s.len(),
                    Some(serde_json::Value::Array(blocks)) => blocks
                        .iter()
                        .filter_map(|b| b.get("text").and_then(|t| t.as_str()))
                        .map(|t| t.len())
                        .sum(),
                    _ => 0,
                })
                .sum();
            *observed_clone.lock().unwrap() = Some(total);
        }
        let msg = text_response("ok");
        let reason = msg.stop_reason;
        Box::pin(futures::stream::iter(vec![
            crate::agent::agent_loop::message::StreamEvent::Done {
                reason,
                message: msg,
                usage: None,
            },
        ]))
    });

    let mut ctx = empty_context();
    ctx.messages = preloaded;
    let mut cfg = build_config();
    cfg.tool_execution = ToolExecutionMode::Sequential;

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);
    let _ = run_agent_loop(
        vec![user("start")],
        ctx,
        cfg,
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;

    // Under-threshold: tool-result content must be present in
    // full (modulo the dirge-k6be cap which only fires above
    // 3000 tokens = ~12 KB; 4 KB is well under that). The
    // fold must NOT have shrunk the transcript.
    let observed = observed_first_call_chars.lock().unwrap();
    let total = observed.expect("first model call must have happened");
    assert!(
        total >= 4_000,
        "under-threshold ratio must not trigger fold; saw {total} chars (input was 4000)",
    );
}

// IMPROVEMENTS_PLAN #1: the compaction circuit breaker. After
// MAX_CONSECUTIVE_COMPACTION_FAILURES failures the LLM summarizer is no
// longer invoked (cheap pruning still runs).
#[test]
fn record_compaction_outcome_drives_counter() {
    let mut f = 0u32;
    super::record_compaction_outcome(&mut f, super::SummaryOutcome::Failed);
    assert_eq!(f, 1);
    super::record_compaction_outcome(&mut f, super::SummaryOutcome::Failed);
    assert_eq!(f, 2);
    super::record_compaction_outcome(&mut f, super::SummaryOutcome::Skipped);
    assert_eq!(f, 2, "skip must not change the counter");
    super::record_compaction_outcome(&mut f, super::SummaryOutcome::Succeeded(0));
    assert_eq!(f, 0, "success resets the counter");
}

#[tokio::test]
async fn compaction_circuit_breaker_skips_summarizer_after_max_failures() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    let calls = std::sync::Arc::new(AtomicUsize::new(0));
    let calls_inner = calls.clone();
    // Summarizer that always fails — and counts its invocations.
    let summarize_fn: Option<crate::agent::compression::SummarizeFn> =
        Some(std::sync::Arc::new(move |_prompt: String| {
            let c = calls_inner.clone();
            Box::pin(async move {
                c.fetch_add(1, Ordering::SeqCst);
                Err(anyhow::anyhow!("summarizer boom"))
            })
        }));

    let make_ctx = || {
        let mut ctx = empty_context();
        ctx.messages
            .push(serde_json::json!({"role":"system","content":"agent"}));
        ctx.messages
            .push(serde_json::json!({"role":"user","content":"task"}));
        for i in 0..20 {
            let role = if i % 2 == 0 { "assistant" } else { "user" };
            ctx.messages.push(serde_json::json!({
                "role": role, "content": format!("turn {i} with filler content")
            }));
        }
        ctx.messages
            .push(serde_json::json!({"role":"user","content":"latest"}));
        ctx
    };

    let (tx, _rx) = mpsc::channel::<LoopEvent>(64);

    // Sub-threshold: the summarizer IS called and reports Failed.
    for failures in 0..super::MAX_CONSECUTIVE_COMPACTION_FAILURES {
        let mut ctx = make_ctx();
        let outcome = super::run_compaction_pass(
            &mut ctx,
            &summarize_fn,
            5,
            failures,
            &None,
            None,
            &tx,
            &empty_checkpoint_slot(),
            &mut 0,
            u64::MAX,
        )
        .await;
        assert_eq!(
            outcome,
            super::SummaryOutcome::Failed,
            "failures={failures}: summarizer should run and fail"
        );
    }
    let calls_before_open = calls.load(Ordering::SeqCst);
    assert_eq!(
        calls_before_open,
        super::MAX_CONSECUTIVE_COMPACTION_FAILURES as usize,
        "summarizer should run once per sub-threshold attempt"
    );

    // At the threshold: breaker open → summarizer NOT called again, and
    // the cheap prune-only fallback still runs (context doesn't grow).
    let mut ctx = make_ctx();
    let n_before = ctx.messages.len();
    let outcome = super::run_compaction_pass(
        &mut ctx,
        &summarize_fn,
        5,
        super::MAX_CONSECUTIVE_COMPACTION_FAILURES,
        &None,
        None,
        &tx,
        &empty_checkpoint_slot(),
        &mut 0,
        u64::MAX,
    )
    .await;
    assert_eq!(
        outcome,
        super::SummaryOutcome::Skipped,
        "breaker open → summarizer skipped"
    );
    assert_eq!(
        calls.load(Ordering::SeqCst),
        calls_before_open,
        "breaker open: summarizer must NOT be invoked"
    );
    assert!(
        ctx.messages.len() <= n_before,
        "prune-only fallback must not grow context"
    );
}

// IMPROVEMENTS_PLAN #5: the ContextCompacted event reports whether the
// pass was prune-only, prune+summary, or prune+failed-summary.
#[tokio::test]
async fn context_compacted_reports_compaction_kind() {
    use crate::event::CompactionKind;

    async fn kind_for(
        summarize_fn: Option<crate::agent::compression::SummarizeFn>,
        failures: u32,
    ) -> CompactionKind {
        let mut ctx = empty_context();
        ctx.messages
            .push(serde_json::json!({"role":"system","content":"agent"}));
        ctx.messages
            .push(serde_json::json!({"role":"user","content":"task"}));
        // An oversized tool result outside the protected tail, so the pruner
        // frees real tokens. Without it the no-summarizer case is a pure no-op,
        // which since dirge-kq3a no longer emits at all — and this test is
        // about which KIND is reported, not about no-op behavior (that is
        // `run_compaction_pass_that_frees_nothing_does_not_rotate_or_announce`).
        ctx.messages.push(serde_json::json!({
            "role": "tool",
            "tool_name": "bash",
            "content": "x".repeat(4000),
        }));
        for i in 0..20 {
            let role = if i % 2 == 0 { "assistant" } else { "user" };
            ctx.messages.push(serde_json::json!({
                "role": role, "content": format!("turn {i} with filler content")
            }));
        }
        ctx.messages
            .push(serde_json::json!({"role":"user","content":"latest"}));
        let (tx, mut rx) = mpsc::channel::<LoopEvent>(8);
        super::run_compaction_pass(
            &mut ctx,
            &summarize_fn,
            5,
            failures,
            &None,
            None,
            &tx,
            &empty_checkpoint_slot(),
            &mut 0,
            u64::MAX,
        )
        .await;
        drop(tx);
        while let Some(ev) = rx.recv().await {
            if let LoopEvent::ContextCompacted {
                compaction_kind, ..
            } = ev
            {
                return compaction_kind;
            }
        }
        panic!("no ContextCompacted event emitted");
    }

    // Valid summary → PruneAndSummary.
    let good: Option<crate::agent::compression::SummarizeFn> = Some(std::sync::Arc::new(
        |_p: String| {
            Box::pin(async move {
                Ok("## Active Task\nx\n\n## Goal\ny\n\n## Completed Actions\n1. z\n\n## Remaining Work\nw"
                    .to_string())
            })
        },
    ));
    assert_eq!(kind_for(good, 0).await, CompactionKind::PruneAndSummary);

    // Failing summary → PruneAndFailedSummary.
    let bad: Option<crate::agent::compression::SummarizeFn> =
        Some(std::sync::Arc::new(|_p: String| {
            Box::pin(async move { Err(anyhow::anyhow!("boom")) })
        }));
    assert_eq!(
        kind_for(bad, 0).await,
        CompactionKind::PruneAndFailedSummary
    );

    // No summarizer wired → PruneOnly.
    assert_eq!(kind_for(None, 0).await, CompactionKind::PruneOnly);

    // Summarizer wired but the circuit breaker is OPEN (failures at the
    // cap) → PruneSummarizerDisabled, NOT PruneOnly. The distinct kind
    // keeps the ongoing-failure signal visible after the breaker latches
    // instead of masquerading as a healthy no-summarizer pass. Use a
    // summarizer that would SUCCEED if called, to prove the kind comes
    // from the breaker being open and not from the summarizer's outcome.
    let would_succeed: Option<crate::agent::compression::SummarizeFn> = Some(std::sync::Arc::new(
        |_p: String| {
            Box::pin(async move {
                Ok("## Active Task\nx\n\n## Goal\ny\n\n## Completed Actions\n1. z\n\n## Remaining Work\nw"
                    .to_string())
            })
        },
    ));
    assert_eq!(
        kind_for(would_succeed, super::MAX_CONSECUTIVE_COMPACTION_FAILURES).await,
        CompactionKind::PruneSummarizerDisabled
    );
}

// ── dirge-vcsn: unified finalization interjection authority ──────────

/// The unfinished-todo nudge wording agrees in number with the count.
#[test]
fn todo_nudge_message_pluralizes() {
    let one = match todo_nudge_message(1, 0) {
        LoopMessage::User(u) => u.text_joined(),
        _ => panic!("expected a user message"),
    };
    assert!(one.contains("1 unfinished todo "), "singular: {one}");
    let many = match todo_nudge_message(3, 0) {
        LoopMessage::User(u) => u.text_joined(),
        _ => panic!("expected a user message"),
    };
    assert!(many.contains("3 unfinished todos "), "plural: {many}");
}

/// dirge-uw2l.5: `low == 0` must reproduce the pre-uw2l.5 wording exactly, so
/// the common case (no low-priority items) changes nothing.
#[test]
fn todo_nudge_message_byte_identical_when_no_low_priority() {
    let want = format!(
        "{TODO_NUDGE_TAG} You still have 2 unfinished todos (pending or in progress). \
         Finish the remaining work, or if it's genuinely done or no longer needed, \
         update the todo list (mark items completed/cancelled) before stopping."
    );
    let got = match todo_nudge_message(2, 0) {
        LoopMessage::User(u) => u.text_joined(),
        _ => panic!("expected a user message"),
    };
    assert_eq!(got, want);
}

/// dirge-uw2l.5: when a low-priority item is outstanding the nudge names it as
/// the cancel candidate (RAX treated rejecting a low-priority unachievable
/// goal as a validation objective, not a failure — paper §3.1b).
#[test]
fn todo_nudge_message_names_low_priority_as_cancel_candidate() {
    let got = match todo_nudge_message(3, 1) {
        LoopMessage::User(u) => u.text_joined(),
        _ => panic!("expected a user message"),
    };
    assert!(
        got.contains("1 low-priority item "),
        "names the low count: {got}"
    );
    assert!(got.contains("cancel"), "invites cancellation: {got}");
}

/// dirge-uw2l.5: the residual-objectives block is appended AFTER the `[dirge]`
/// prefix, so the headless truncation detector in `provider::run` — which
/// matches `content.starts_with(MAX_TURNS_NOTICE_PREFIX)` (provider/run.rs) —
/// still fires. This test is what keeps the emitter and the detector from
/// drifting: if a future change reorders or drops the prefix, it fails here
/// rather than silently breaking truncation detection.
#[test]
fn max_turns_notice_keeps_truncation_prefix_with_residual_block() {
    use crate::agent::tools::todo::TodoItem;
    let board = vec![TodoItem {
        content: "ship the residual handoff".into(),
        status: "open".into(),
        priority: "normal".into(),
    }];
    let notice = max_turns_notice(50, &board);
    assert!(
        notice.starts_with(MAX_TURNS_NOTICE_PREFIX),
        "truncation prefix dropped: {notice}"
    );
    assert!(
        notice.contains("Objectives still outstanding"),
        "residual block missing: {notice}"
    );
    // Empty board → no block, still prefixed, byte-identical to the old notice.
    let bare = max_turns_notice(50, &[]);
    assert!(bare.starts_with(MAX_TURNS_NOTICE_PREFIX));
    assert!(!bare.contains("Objectives still outstanding"));
}

/// dirge-1g3v: the reviewer engages only on what THIS run changed. Given the
/// current working-tree diff and the run-start baseline, `run_delta_to_review`
/// yields the diff to review, or `None` to skip.
#[test]
fn run_delta_to_review_skips_when_unchanged() {
    use crate::agent::agent_loop::code_review::RunDiff;

    let wip = RunDiff {
        capped: "wip diff".to_string(),
        fingerprint: 1,
    };
    // Read-only turn over pre-existing WIP: identical diff → skip. Before the
    // dirge-1g3v gate, any ToolResult drove the judge on the whole dirty tree
    // even when the run touched nothing.
    assert_eq!(run_delta_to_review(Some(&wip), Some(&wip)), None);

    // Clean tree, nothing changed → nothing to review.
    assert_eq!(run_delta_to_review(None, None), None);

    // Agent created changes on a clean tree → review them.
    let new = RunDiff {
        capped: "new diff".to_string(),
        fingerprint: 2,
    };
    assert_eq!(run_delta_to_review(Some(&new), None), Some("new diff"));

    // Agent added to pre-existing WIP → the diff differs → review.
    let wip_more = RunDiff {
        capped: "wip + more".to_string(),
        fingerprint: 3,
    };
    assert_eq!(
        run_delta_to_review(Some(&wip_more), Some(&wip)),
        Some("wip + more")
    );

    // Agent reverted the WIP back to clean → no current diff → nothing to review.
    assert_eq!(run_delta_to_review(None, Some(&wip)), None);
}

/// dirge-8gdv: the skip decision must compare the UNcapped fingerprints, not
/// the size-capped text. When pre-existing WIP already exceeds MAX_DIFF_BYTES,
/// a length-preserving edit landing PAST the cap leaves the two CAPPED strings
/// byte-identical, so the old capped-string comparison saw no change and
/// skipped the reviewer. Two diffs with identical capped text but different
/// fingerprints must be seen as CHANGED.
#[test]
fn run_delta_to_review_engages_when_capped_identical_but_fingerprint_differs() {
    use crate::agent::agent_loop::code_review::RunDiff;

    let capped = "identical capped diff text".to_string();
    let baseline = RunDiff {
        capped: capped.clone(),
        fingerprint: 1,
    };
    let current = RunDiff {
        capped: capped.clone(),
        fingerprint: 2,
    };

    // The bug's premise: the capped text the reviewer would see is identical.
    assert_eq!(baseline.capped, current.capped);
    // …but the fingerprints differ, so the reviewer engages (not skipped).
    assert_eq!(
        run_delta_to_review(Some(&current), Some(&baseline)),
        Some(capped.as_str())
    );
}

/// A nonterminal coordinator generation is an intentional suspension, not a
/// completion candidate. The finalization poll must return before invoking the
/// critic, and must leave its one-shot budget untouched for reconciliation.
#[tokio::test]
async fn finalization_defers_critic_while_external_work_is_pending() {
    use crate::agent::agent_loop::critic::CriticFn;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let calls = Arc::new(AtomicUsize::new(0));
    let mut config = build_config();
    config.should_defer_finalization = Some(Arc::new(|| true));
    let judge: CriticFn = Arc::new({
        let calls = calls.clone();
        move |_prompt| {
            calls.fetch_add(1, Ordering::SeqCst);
            Box::pin(async { Ok("VERDICT: COMPLETE\nFINDINGS: none".to_string()) })
        }
    });
    config.critic_fn = Some(judge);
    let new_messages = vec![LoopMessage::ToolResult(
        crate::agent::agent_loop::message::ToolResultMessage {
            tool_call_id: "call_1".into(),
            tool_name: "task".into(),
            content: vec![crate::agent::agent_loop::message::ContentBlock::Text {
                text: "background task started".into(),
            }],
            details: serde_json::Value::Null,
            is_error: false,
        },
    )];
    let mut gates = GateStates {
        critic_done: false,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(8);
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert!(msgs.is_empty());
    assert_eq!(source, FollowUpSource::None);
    assert_eq!(calls.load(Ordering::SeqCst), 0);
    assert!(!gates.critic_done);
}

/// Highest-priority gate (the caller hook) short-circuits the lower gates:
/// when it yields a follow-up, the critic is never consulted (`critic_done`
/// stays false) and the todo gate isn't reached. This locks the precedence.
#[tokio::test]
async fn finalization_hook_short_circuits_lower_gates() {
    let mut config = build_config();
    config.get_followup_messages = Some(std::sync::Arc::new(|| {
        Box::pin(async {
            vec![LoopMessage::User(
                crate::agent::agent_loop::message::UserMessage::text("hook follow-up"),
            )]
        })
    }));
    // A batch can become terminal at this exact boundary. Delivery must win
    // over a stale/overlapping deferral signal.
    config.should_defer_finalization = Some(Arc::new(|| true));
    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: 0u8,
        resume_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs::default(),
        &review_emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::Hook);
    assert_eq!(msgs.len(), 1);
    assert!(
        !gates.critic_done,
        "hook must short-circuit before the critic runs"
    );
    assert_eq!(gates.todo_nudges, 0, "todo gate must not be reached");
}

// dirge-8v98: the `decide_review_reaction` react-counting/advisory-dedup tests
// were removed with that function — the unified judge (`run_unified_review`)
// builds one consolidated follow-up, covered by the critic module's tests.

/// With no hook/verifier/critic and the todo gate exhausted, the authority
/// reports `None` so the run finalizes. (`todo_nudges = MAX` keeps this
/// deterministic regardless of the process-global todo list.)
#[tokio::test]
async fn finalization_all_gates_silent_yields_none() {
    let config = build_config(); // hook/verifier/critic all None
    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES, // todo gate bounded out
        resume_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs::default(),
        &review_emit,
    )
    .await;

    assert!(msgs.is_empty());
    assert_eq!(source, FollowUpSource::None);
}

/// Goal gate: an unmet stop condition re-enters the loop and counts the
/// re-entry. `critic_done = true` isolates the goal gate from the
/// one-shot critic so we observe it specifically.
#[tokio::test]
async fn finalization_goal_unmet_reenters_and_counts() {
    use crate::agent::agent_loop::critic::CriticFn;
    let mut config = build_config();
    config.goal = Some("all tests pass and committed".into());
    let judge: CriticFn =
        Arc::new(|_p| Box::pin(async { Ok("GOAL: UNMET\n- tests still failing".to_string()) }));
    config.goal_fn = Some(judge);

    let mut gates = GateStates {
        critic_done: true, // skip the one-shot critic
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs::default(),
        &review_emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::Goal);
    assert_eq!(gates.goal_reacts, 1, "an unmet goal counts one re-entry");
    assert_eq!(msgs.len(), 1);
}

// ── dirge-g2ex: awaiting-user gate (step 0) ────────────────────────────────
//
// A final assistant turn that ends in a question must finalize and hand
// control back to the user, outranking the hook (step 1) and every lower
// "are we done?" gate. The goal gate is the one exception (autonomous stop).

/// dirge-g2ex: a trailing `ToolResult` so `run_made_tool_calls` is true (the
/// critic's precondition passes), making the step-0 short-circuit the ONLY
/// reason the critic never runs.
fn g2ex_tool_result() -> LoopMessage {
    LoopMessage::ToolResult(crate::agent::agent_loop::message::ToolResultMessage {
        tool_call_id: "call_1".into(),
        tool_name: "edit".into(),
        content: vec![crate::agent::agent_loop::message::ContentBlock::Text { text: "ok".into() }],
        details: serde_json::Value::Null,
        is_error: false,
    })
}

/// (a) Question pending outranks the hook: the hook is NEVER polled, the critic
/// judge is NEVER paid for (even though its `run_made_tool_calls` precondition
/// holds), `critic_done` stays false, and `todo_nudges` stays 0. Finalizes with
/// empty messages + `AwaitingUser`.
#[tokio::test]
async fn awaiting_user_gate_short_circuits_hook_critic_and_todos() {
    use crate::agent::agent_loop::critic::CriticFn;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let hook_calls = Arc::new(AtomicUsize::new(0));
    let critic_calls = Arc::new(AtomicUsize::new(0));

    let mut config = build_config();
    // Wire a hook that would otherwise win (step 1).
    let hc = Arc::clone(&hook_calls);
    config.get_followup_messages = Some(std::sync::Arc::new(move || {
        hc.fetch_add(1, Ordering::SeqCst);
        Box::pin(async {
            vec![LoopMessage::User(
                crate::agent::agent_loop::message::UserMessage::text("hook follow-up"),
            )]
        })
    }));
    // Wire a critic that WOULD fire (critic_done=false, run_made_tool_calls=true).
    let cc = Arc::clone(&critic_calls);
    let judge: CriticFn = Arc::new(move |_p| {
        cc.fetch_add(1, Ordering::SeqCst);
        Box::pin(async { Ok("VERDICT: INCOMPLETE".to_string()) })
    });
    config.critic_fn = Some(judge);

    // Turn did real work, then ended by asking the user.
    let new_messages = vec![
        g2ex_tool_result(),
        assistant_text("Which approach would you prefer?"),
    ];

    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: 0u8, // would otherwise fire (unfinished_count is process-global)
        resume_nudges: 0,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::AwaitingUser);
    assert!(msgs.is_empty(), "finalizes with no injected follow-up");
    assert_eq!(
        hook_calls.load(Ordering::SeqCst),
        0,
        "step-0 outranks the hook"
    );
    assert_eq!(
        critic_calls.load(Ordering::SeqCst),
        0,
        "no judge LLM call is paid for"
    );
    assert!(!gates.critic_done, "critic one-shot never consumed");
    assert_eq!(gates.todo_nudges, 0, "todo gate never reached");
}

/// (b) Even with a question pending, an unmet `--goal` still pushes the run.
#[tokio::test]
async fn awaiting_user_gate_still_honors_unmet_goal() {
    use crate::agent::agent_loop::critic::CriticFn;

    let mut config = build_config();
    config.goal = Some("all tests pass and committed".into());
    let judge: CriticFn =
        Arc::new(|_p| Box::pin(async { Ok("GOAL: UNMET\n- tests still failing".to_string()) }));
    config.goal_fn = Some(judge);

    let new_messages = vec![assistant_text("Which approach would you prefer?")];

    let mut gates = GateStates {
        critic_done: true, // isolate the goal gate from the one-shot critic
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::Goal);
    assert_eq!(gates.goal_reacts, 1, "an unmet goal counts one re-entry");
    assert_eq!(msgs.len(), 1);
}

/// (c) A question pending + `should_defer_finalization` true → source `None`,
/// and the goal judge is NEVER called (defer takes precedence over the goal).
#[tokio::test]
async fn awaiting_user_gate_defers_when_coordinator_running() {
    use crate::agent::agent_loop::critic::CriticFn;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let goal_calls = Arc::new(AtomicUsize::new(0));
    let mut config = build_config();
    config.should_defer_finalization = Some(Arc::new(|| true));
    config.goal = Some("all tests pass".into());
    let gc = Arc::clone(&goal_calls);
    let judge: CriticFn = Arc::new(move |_p| {
        gc.fetch_add(1, Ordering::SeqCst);
        Box::pin(async { Ok("GOAL: UNMET".to_string()) })
    });
    config.goal_fn = Some(judge);

    let new_messages = vec![assistant_text("Which approach would you prefer?")];

    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::None);
    assert!(msgs.is_empty());
    assert_eq!(
        goal_calls.load(Ordering::SeqCst),
        0,
        "defer short-circuits before the goal judge"
    );
    assert_eq!(gates.goal_reacts, 0);
}

// ── dirge-g2ex R3: todo gate now requires the turn to have made file edits ──

/// Helper: seed the process-global TODO_LIST mirror with `n` open items and
/// return a guard that clears it on drop, so a test can't leak state.
///
/// The guard also HOLDS `TODO_TEST_LOCK` for the caller's whole seed-act-assert
/// span. The mirror is process-global and cargo runs tests in parallel threads
/// within one process, so without this another test's seed lands between our
/// seed and our assert and the gate sees the wrong count (dirge-g2ex). Field
/// order matters: `Drop::drop` clears the mirror, then the guard field releases
/// the lock.
struct TodoGuard(
    // Held for RAII only — never read. Dropping it is the entire point.
    #[allow(dead_code)] std::sync::MutexGuard<'static, ()>,
);
impl Drop for TodoGuard {
    fn drop(&mut self) {
        crate::agent::tools::todo::TODO_LIST
            .lock_ignore_poison()
            .clear();
    }
}
fn seed_open_todos(n: usize) -> TodoGuard {
    let lock = crate::agent::tools::todo::TODO_TEST_LOCK.lock_ignore_poison();
    let items: Vec<_> = (0..n)
        .map(|_| crate::agent::tools::todo::TodoItem {
            content: "pending work".into(),
            status: "open".into(),
            priority: "normal".into(),
        })
        .collect();
    *crate::agent::tools::todo::TODO_LIST.lock_ignore_poison() = items;
    TodoGuard(lock)
}

/// A read-only turn (no file edits) with unfinished todos must NOT trip the
/// todo nudge — `unfinished_count()` is a cross-turn global, so without the
/// `turn_made_file_edits` precondition an interrupting Q&A gets nagged.
#[tokio::test]
async fn todo_gate_skips_readonly_turn_even_with_unfinished_todos() {
    let _g = seed_open_todos(2);
    let config = build_config();

    // Read-only turn: grepped and read, then stopped. No file edits.
    let new_messages = vec![assistant_calling("read"), assistant_text("Done reading.")];
    assert!(
        !turn_made_file_edits(&new_messages),
        "fixture: turn really is read-only"
    );

    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: 0u8,
        resume_nudges: 0,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::None);
    assert!(msgs.is_empty(), "no todo nudge on a read-only turn");
    assert_eq!(gates.todo_nudges, 0, "todo budget untouched");
}

/// A turn that made a file edit with unfinished todos still fires the nudge —
/// the precondition only narrows the gate, it doesn't disable it.
#[tokio::test]
async fn todo_gate_fires_on_file_edit_turn_with_unfinished_todos() {
    let _g = seed_open_todos(1);
    let config = build_config();

    let new_messages = vec![assistant_calling("edit")];
    assert!(
        turn_made_file_edits(&new_messages),
        "fixture: turn made a file edit"
    );

    let mut gates = GateStates {
        critic_done: true, // skip the one-shot critic
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: 0u8,
        resume_nudges: 0,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::Todo);
    assert_eq!(
        gates.todo_nudges, 1,
        "nudge fires as before on an editing turn"
    );
    assert_eq!(msgs.len(), 1);
    let content = match &msgs[0] {
        LoopMessage::User(u) => u.text_joined(),
        _ => panic!("expected User message"),
    };
    assert!(
        content.starts_with(crate::agent::agent_loop::run::TODO_NUDGE_TAG),
        "expected [todo] tag, got: {content}"
    );
}

/// dirge-u1ay (GH #734): a turn that wrote a todo list and did nothing else
/// must be nudged back to the work. `write_todo_list` is not an Edit
/// operation, so the `turn_made_file_edits` precondition left exactly the
/// reported failure — model plans, model stops, nothing on disk — as the one
/// case with no backstop. The wording has to push toward the edit, not toward
/// another round of list maintenance.
#[tokio::test]
async fn todo_gate_fires_on_a_plan_only_turn() {
    let _g = seed_open_todos(2);
    let config = build_config();

    let new_messages = vec![
        assistant_calling("write_todo_list"),
        assistant_text("I have laid out the plan."),
    ];
    assert!(
        !turn_made_file_edits(&new_messages),
        "fixture: planning is not a file edit"
    );

    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: 0u8,
        resume_nudges: 0,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::Todo);
    assert_eq!(gates.todo_nudges, 1, "plan-only turn spends a nudge");
    let content = match &msgs[0] {
        LoopMessage::User(u) => u.text_joined(),
        _ => panic!("expected User message"),
    };
    assert!(
        content.starts_with(crate::agent::agent_loop::run::TODO_NUDGE_TAG),
        "expected [todo] tag, got: {content}"
    );
    assert!(
        content.contains("write") || content.contains("edit"),
        "must point at the tools that do the work, got: {content}"
    );
}

/// The plan-only branch is one-shot. `new_messages` accumulates across
/// finalization re-entries, so the `write_todo_list` call that triggered the
/// first nudge is still in the list on the next pass — without a one-shot
/// gate the same text re-enters until the budget drains, spending API
/// round-trips on a model that already stopped planning. A behavioral nudge
/// that didn't land once won't land on the third identical repeat.
#[tokio::test]
async fn plan_only_nudge_is_one_shot() {
    let _g = seed_open_todos(2);
    let config = build_config();

    // The turn that already spent a nudge: planning is still in the history,
    // and the model answered with prose instead of doing the work.
    let new_messages = vec![
        assistant_calling("write_todo_list"),
        assistant_text("I have laid out the plan."),
        assistant_text("Still just planning."),
    ];

    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: 1u8, // one already spent
        resume_nudges: 0,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::None, "must not nudge twice");
    assert!(msgs.is_empty());
    assert_eq!(gates.todo_nudges, 1, "budget untouched by the repeat");
}

/// The plan-only branch must not resurrect the nagging that the
/// `turn_made_file_edits` precondition was introduced to stop: a read-only
/// Q&A turn with stale cross-turn todos still gets nothing.
#[tokio::test]
async fn todo_gate_still_skips_readonly_turn_that_wrote_no_todos() {
    let _g = seed_open_todos(2);
    let config = build_config();

    let new_messages = vec![assistant_calling("grep"), assistant_text("Here's why.")];

    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: 0u8,
        resume_nudges: 0,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::None);
    assert!(msgs.is_empty(), "no nudge without todo writes or edits");
    assert_eq!(gates.todo_nudges, 0, "todo budget untouched");
}

/// dirge-8v98: the unified judge re-enters the loop on a review finding even
/// when the completeness verdict is COMPLETE — the exact case the old
/// display-only advisory swallowed. One-shot in Advisory/Off, so `critic_done`
/// flips. `code_review = Off` here so the gate reviews completeness only (no
/// git diff capture in the test); the canned judge still emits a finding, which
/// must re-enter as a `[critic]` follow-up.
#[tokio::test]
async fn finalization_unified_judge_reenters_on_finding() {
    use crate::agent::agent_loop::critic::CriticFn;
    use crate::agent::agent_loop::types::CodeReviewMode;
    let mut config = build_config();
    config.code_review_mode = CodeReviewMode::Off;
    let judge: CriticFn = Arc::new(|_p| {
        Box::pin(async {
            Ok("VERDICT: COMPLETE\nFINDINGS:\n- high: null deref on empty input.".to_string())
        })
    });
    config.critic_fn = Some(judge);

    // The gate requires the run to have made tool calls (a ToolResult present).
    let new_messages = vec![LoopMessage::ToolResult(
        crate::agent::agent_loop::message::ToolResultMessage {
            tool_call_id: "call_1".into(),
            tool_name: "edit".into(),
            content: vec![crate::agent::agent_loop::message::ContentBlock::Text {
                text: "ok".into(),
            }],
            details: serde_json::Value::Null,
            is_error: false,
        },
    )];

    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &new_messages,
        &mut gates,
        GateInputs::default(),
        &review_emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::Critic);
    assert_eq!(msgs.len(), 1);
    let text = match &msgs[0] {
        LoopMessage::User(u) => u.text_joined(),
        other => panic!("expected user follow-up, got {other:?}"),
    };
    assert!(
        text.contains("null deref"),
        "finding must reach the model: {text}"
    );
    assert!(
        gates.critic_done,
        "Off/Advisory unified judge is one-shot — gates.critic_done must flip"
    );
}

/// A met goal lets the run finalize and does NOT count a re-entry.
#[tokio::test]
async fn finalization_goal_met_finalizes() {
    use crate::agent::agent_loop::critic::CriticFn;
    let mut config = build_config();
    config.goal = Some("all tests pass".into());
    let judge: CriticFn = Arc::new(|_p| Box::pin(async { Ok("GOAL: MET".to_string()) }));
    config.goal_fn = Some(judge);

    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs::default(),
        &review_emit,
    )
    .await;

    assert!(msgs.is_empty());
    assert_eq!(source, FollowUpSource::None);
    assert_eq!(gates.goal_reacts, 0);
}

/// Once the re-entry bound is reached, an unmet goal no longer blocks —
/// the run finalizes rather than looping forever on a bad stop condition.
#[tokio::test]
async fn finalization_goal_bound_stops_reentry() {
    use crate::agent::agent_loop::critic::CriticFn;
    let mut config = build_config();
    config.goal = Some("unsatisfiable".into());
    let judge: CriticFn = Arc::new(|_p| Box::pin(async { Ok("GOAL: UNMET".to_string()) }));
    config.goal_fn = Some(judge);

    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: crate::agent::agent_loop::goal::MAX_GOAL_REACT,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs::default(),
        &review_emit,
    )
    .await;

    assert!(msgs.is_empty());
    assert_eq!(source, FollowUpSource::None, "bound reached → finalize");
}

/// Goal gate stays OFF when no judge (`goal_fn`) is configured, even with
/// a goal set.
#[tokio::test]
async fn finalization_goal_without_judge_is_inert() {
    let mut config = build_config();
    config.goal = Some("all tests pass".into());
    config.goal_fn = None; // no judge

    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs::default(),
        &review_emit,
    )
    .await;

    assert!(msgs.is_empty());
    assert_eq!(source, FollowUpSource::None);
    assert_eq!(gates.goal_reacts, 0);
}

/// Open-issues gate Off → inert (FollowUpSource::None).
#[tokio::test]
async fn open_issues_gate_off_is_inert() {
    let config = build_config();
    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        open_issues_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);

    // Create a temp DB with open issues for this session.
    let dir = temp_dir("open-issues-off");
    let db_path = dir.join("state.db");
    let store = crate::extras::issue_db::IssueStore::open_at(&db_path).unwrap();
    let sid = "open-issues-off-sess";
    store
        .create("wire up telemetry", "", None, Some(sid), None)
        .unwrap();

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs {
            code_review_baseline: None,
            open_issues_gate_mode: GateMode::Off,
            issue_db_path: Some(db_path.as_path()),
            session_id: Some(sid),
        },
        &review_emit,
    )
    .await;

    assert!(msgs.is_empty(), "Off mode should be inert");
    assert_eq!(source, FollowUpSource::None);

    let _ = std::fs::remove_dir_all(&dir);
}

/// Open-issues gate blocking with N session issues open → returns a
/// `[open-issues]` nudge (FollowUpSource::OpenIssues) listing titles.
#[tokio::test]
async fn open_issues_gate_blocking_with_session_open_issues_nudges() {
    use crate::agent::agent_loop::run::OPEN_ISSUES_NUDGE_TAG;
    let config = build_config();
    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        open_issues_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);

    let dir = temp_dir("open-issues-blocking");
    let db_path = dir.join("state.db");
    let store = crate::extras::issue_db::IssueStore::open_at(&db_path).unwrap();
    let sid = "open-issues-blocking-sess";
    store
        .create("wire up telemetry", "", None, Some(sid), None)
        .unwrap();
    store
        .create("add metrics dashboard", "", None, Some(sid), None)
        .unwrap();

    // dirge-g2ex: open-issues gate now requires the turn to have made file
    // edits, so seed a real edit (not the empty `&[]` it used before).
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[assistant_calling("edit")],
        &mut gates,
        GateInputs {
            code_review_baseline: None,
            open_issues_gate_mode: GateMode::Blocking,
            issue_db_path: Some(db_path.as_path()),
            session_id: Some(sid),
        },
        &review_emit,
    )
    .await;

    assert_eq!(source, FollowUpSource::OpenIssues);
    assert_eq!(gates.open_issues_nudges, 1);
    assert_eq!(msgs.len(), 1);
    let content = match &msgs[0] {
        LoopMessage::User(u) => u.text_joined(),
        _ => panic!("expected User message"),
    };
    assert!(
        content.starts_with(OPEN_ISSUES_NUDGE_TAG),
        "expected [open-issues] tag, got: {content}"
    );
    assert!(content.contains("wire up telemetry"), "{content}");
    assert!(content.contains("add metrics dashboard"), "{content}");

    let _ = std::fs::remove_dir_all(&dir);
}

/// Blocking bound stops re-entry after MAX_OPEN_ISSUES_NUDGES.
#[tokio::test]
async fn open_issues_gate_blocking_has_bound() {
    let config = build_config();
    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        open_issues_nudges: MAX_OPEN_ISSUES_NUDGES, // already at bound
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);

    let dir = temp_dir("open-issues-bound");
    let db_path = dir.join("state.db");
    let store = crate::extras::issue_db::IssueStore::open_at(&db_path).unwrap();
    let sid = "open-issues-bound-sess";
    store
        .create("wire up telemetry", "", None, Some(sid), None)
        .unwrap();

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs {
            code_review_baseline: None,
            open_issues_gate_mode: GateMode::Blocking,
            issue_db_path: Some(db_path.as_path()),
            session_id: Some(sid),
        },
        &review_emit,
    )
    .await;

    assert!(msgs.is_empty(), "bounded gate should be inert");
    assert_eq!(source, FollowUpSource::None);

    let _ = std::fs::remove_dir_all(&dir);
}

/// Zero open session issues → inert (FollowUpSource::None).
#[tokio::test]
async fn open_issues_gate_zero_open_session_issues_is_inert() {
    let config = build_config();
    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        open_issues_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);

    let dir = temp_dir("open-issues-zero");
    let db_path = dir.join("state.db");
    let _store = crate::extras::issue_db::IssueStore::open_at(&db_path).unwrap();
    // No issues for this session.
    let sid = "open-issues-zero-sess";

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs {
            code_review_baseline: None,
            open_issues_gate_mode: GateMode::Blocking,
            issue_db_path: Some(db_path.as_path()),
            session_id: Some(sid),
        },
        &review_emit,
    )
    .await;

    assert!(msgs.is_empty(), "zero open issues should be inert");
    assert_eq!(source, FollowUpSource::None);

    let _ = std::fs::remove_dir_all(&dir);
}

/// Missing db → inert (fail-open).
#[tokio::test]
async fn open_issues_gate_missing_db_is_inert() {
    let config = build_config();
    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        open_issues_nudges: 0,
        ..Default::default()
    };
    let (review_emit, _review_emit_rx) = tokio::sync::mpsc::channel(64);

    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[],
        &mut gates,
        GateInputs {
            code_review_baseline: None,
            open_issues_gate_mode: GateMode::Blocking,
            issue_db_path: None,
            session_id: Some("some-sess"),
        },
        &review_emit,
    )
    .await;

    assert!(msgs.is_empty(), "missing db should be inert (fail-open)");
    assert_eq!(source, FollowUpSource::None);
}

/// Advisory mode emits a SystemNotice when issues are open but does
/// NOT re-enter the loop.
#[tokio::test]
async fn open_issues_gate_advisory_emits_notice_but_does_not_reenter() {
    let config = build_config();
    let mut gates = GateStates {
        critic_done: true,
        code_review_reacts: 0u8,
        goal_reacts: 0u8,
        todo_nudges: MAX_TODO_NUDGES,
        resume_nudges: 0,
        open_issues_nudges: 0,
        ..Default::default()
    };
    let (review_emit, mut review_emit_rx) = tokio::sync::mpsc::channel(64);

    let dir = temp_dir("open-issues-advisory");
    let db_path = dir.join("state.db");
    let store = crate::extras::issue_db::IssueStore::open_at(&db_path).unwrap();
    let sid = "open-issues-advisory-sess";
    store
        .create("wire up telemetry", "", None, Some(sid), None)
        .unwrap();

    // dirge-g2ex: open-issues gate now requires the turn to have made file edits.
    let (msgs, source) = poll_finalization_follow_up(
        &config,
        "sys",
        &[assistant_calling("edit")],
        &mut gates,
        GateInputs {
            code_review_baseline: None,
            open_issues_gate_mode: GateMode::Advisory,
            issue_db_path: Some(db_path.as_path()),
            session_id: Some(sid),
        },
        &review_emit,
    )
    .await;

    // Advisory does NOT re-enter (returns empty messages).
    assert!(msgs.is_empty(), "advisory should not re-enter");
    assert_eq!(source, FollowUpSource::None);
    assert_eq!(gates.open_issues_nudges, 1, "counts the advisory");

    // Check that a SystemNotice was emitted.
    match review_emit_rx.try_recv() {
        Ok(crate::agent::agent_loop::message::LoopEvent::SystemNotice { content }) => {
            assert!(
                content.contains("issue(s) from this session are still open"),
                "{content}"
            );
        }
        other => panic!("expected SystemNotice, got {other:?}"),
    }

    let _ = std::fs::remove_dir_all(&dir);
}

// ── Blocking review dedupe (dirge-9b2k): skip re-reviewing an unchanged diff ─
//
// The unified finalization judge is stateless, so a Blocking run that persists
// across finalizations can loop: when the model declines a finding and changes
// nothing on disk, a naive re-review re-raises the identical finding and the
// model re-emits the identical rebuttal — a duplicate. The fix skips the judge
// when this exact diff (by uncapped fingerprint) was reviewed last reaction.

/// A temp git repo with one committed file and one uncommitted edit, so
/// `capture_run_diff` yields a stable, non-empty diff. The caller points the
/// reviewer at it via `config.code_review_repo`, so these tests never touch the
/// process-global CWD and can run in parallel with any other test.
fn temp_review_repo(suffix: &str) -> std::path::PathBuf {
    let dir = temp_dir(&format!("blocking-review-{suffix}"));
    let git = |args: &[&str]| {
        let _ = std::process::Command::new("git")
            .current_dir(&dir)
            .args(args)
            .output();
    };
    git(&["init", "-q"]);
    git(&["config", "user.email", "test@test.test"]);
    git(&["config", "user.name", "test"]);
    std::fs::write(dir.join("a.rs"), "fn main() {}\n").unwrap();
    git(&["add", "."]);
    git(&["commit", "-q", "-m", "base"]);
    // dirty edit — the diff capture reviews this.
    std::fs::write(dir.join("a.rs"), "fn main() { let x = 1; }\n").unwrap();
    dir
}

/// A run that made a tool call, so `run_made_tool_calls` is true and the
/// unified judge gate is eligible.
fn run_with_tool_result() -> Vec<LoopMessage> {
    vec![LoopMessage::ToolResult(
        crate::agent::agent_loop::message::ToolResultMessage {
            tool_call_id: "call_1".into(),
            tool_name: "task".into(),
            content: vec![crate::agent::agent_loop::message::ContentBlock::Text {
                text: "done".into(),
            }],
            details: serde_json::Value::Null,
            is_error: false,
        },
    )]
}

/// A stateless judge stub that always raises one high finding, recording how
/// many times it was invoked.
fn counting_judge(calls: &Arc<AtomicUsize>) -> crate::agent::agent_loop::critic::CriticFn {
    let calls = calls.clone();
    Arc::new(move |_p: String| {
        calls.fetch_add(1, Ordering::SeqCst);
        Box::pin(async { Ok("VERDICT: INCOMPLETE\nFINDINGS:\n- High — bug".to_string()) })
    })
}

#[tokio::test]
async fn blocking_review_skips_judge_when_diff_unchanged_across_reactions() {
    use crate::agent::agent_loop::types::CodeReviewMode;

    let repo = temp_review_repo("skip");

    let calls = Arc::new(AtomicUsize::new(0));
    let mut config = build_config();
    config.critic_fn = Some(counting_judge(&calls));
    config.code_review_mode = CodeReviewMode::Blocking;
    config.code_review_repo = Some(repo.clone());

    let msgs_run = run_with_tool_result();
    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        last_reviewed_fingerprint: None,
        last_review_findings: None,
        ..Default::default()
    };
    let (emit, _rx) = tokio::sync::mpsc::channel(8);

    // Reaction 1: the judge reviews the diff and raises a finding.
    let (msgs1, src1) = poll_finalization_follow_up(
        &config,
        "sys",
        &msgs_run,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;
    assert_eq!(
        calls.load(Ordering::SeqCst),
        1,
        "first reaction calls the judge"
    );
    assert!(!msgs1.is_empty(), "first reaction returns the finding");
    assert_eq!(src1, FollowUpSource::Critic);
    assert_eq!(
        gates.code_review_reacts, 1,
        "first reaction spends a budget"
    );
    assert!(!gates.critic_done, "Blocking never sets the one-shot flag");
    assert!(
        gates.last_reviewed_fingerprint.is_some(),
        "the reviewed diff fingerprint is recorded"
    );

    // Reaction 2: the diff on disk is UNCHANGED → the judge must be skipped.
    let (msgs2, src2) = poll_finalization_follow_up(
        &config,
        "sys",
        &msgs_run,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;
    assert_eq!(
        calls.load(Ordering::SeqCst),
        1,
        "judge NOT called again on an unchanged diff"
    );
    assert!(
        msgs2.is_empty(),
        "no follow-up — the model's rebuttal stands"
    );
    assert_eq!(src2, FollowUpSource::None);
    assert_eq!(
        gates.code_review_reacts, 1,
        "budget not spent on the skipped reaction"
    );

    let _ = std::fs::remove_dir_all(&repo);
}

/// dirge-9b2k regression: the Blocking dedupe skip must NOT early-return out of
/// `poll_finalization_follow_up` — it must fall through to the downstream gates.
/// Here reaction 2 skips the critic (unchanged diff), and the GOAL gate fires
/// instead of finalizing with `None`. An earlier version of the guard `return`ed
/// on skip, silently dropping that follow-up. The goal gate is config-injected,
/// so this test carries no process-global state and is parallel-safe.
#[tokio::test]
async fn blocking_review_skip_falls_through_to_downstream_gate() {
    use crate::agent::agent_loop::types::CodeReviewMode;

    let repo = temp_review_repo("fallthrough");

    let calls = Arc::new(AtomicUsize::new(0));
    let mut config = build_config();
    config.critic_fn = Some(counting_judge(&calls));
    config.code_review_mode = CodeReviewMode::Blocking;
    config.code_review_repo = Some(repo.clone());
    // A downstream goal gate whose stop condition is never met — an unmet goal
    // re-enters the loop, so its firing proves the skip fell through the critic.
    config.goal = Some("ship it".to_string());
    config.goal_fn = Some(Arc::new(|_p: String| {
        Box::pin(async { Ok("GOAL: UNMET\n- keep going".to_string()) })
    }));

    let msgs_run = run_with_tool_result();
    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        last_reviewed_fingerprint: None,
        last_review_findings: None,
        goal_reacts: 0u8,
        ..Default::default()
    };
    let (emit, _emit_rx) = tokio::sync::mpsc::channel(8);

    // Reaction 1: the critic reviews the diff and raises a finding, returning
    // before the goal gate is reached.
    let (msgs1, src1) = poll_finalization_follow_up(
        &config,
        "sys",
        &msgs_run,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;
    assert_eq!(
        calls.load(Ordering::SeqCst),
        1,
        "first reaction calls the critic"
    );
    assert_eq!(src1, FollowUpSource::Critic);
    assert!(!msgs1.is_empty());
    assert_eq!(
        gates.goal_reacts, 0,
        "goal gate not reached while the critic fires"
    );

    // Reaction 2: the diff on disk is UNCHANGED → the critic is skipped, BUT the
    // fall-through must reach the goal gate.
    let (msgs2, src2) = poll_finalization_follow_up(
        &config,
        "sys",
        &msgs_run,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;
    assert_eq!(
        calls.load(Ordering::SeqCst),
        1,
        "critic NOT called again on an unchanged diff"
    );
    assert!(
        !msgs2.is_empty(),
        "the skipped reaction must fall through to the goal gate"
    );
    assert_eq!(
        src2,
        FollowUpSource::Goal,
        "the goal gate fires, not Critic and not None"
    );
    assert_eq!(
        gates.code_review_reacts, 1,
        "critic budget not spent on the skipped reaction"
    );
    assert_eq!(
        gates.goal_reacts, 1,
        "the goal gate fired on the fall-through"
    );

    let _ = std::fs::remove_dir_all(&repo);
}

#[tokio::test]
async fn blocking_review_re_fires_judge_when_diff_changes_between_reactions() {
    use crate::agent::agent_loop::types::CodeReviewMode;

    let repo = temp_review_repo("changed");

    let calls = Arc::new(AtomicUsize::new(0));
    let mut config = build_config();
    config.critic_fn = Some(counting_judge(&calls));
    config.code_review_mode = CodeReviewMode::Blocking;
    config.code_review_repo = Some(repo.clone());

    let msgs_run = run_with_tool_result();
    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        last_reviewed_fingerprint: None,
        last_review_findings: None,
        ..Default::default()
    };
    let (emit, _rx) = tokio::sync::mpsc::channel(8);

    // Reaction 1.
    let (msgs1, _src1) = poll_finalization_follow_up(
        &config,
        "sys",
        &msgs_run,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;
    assert_eq!(calls.load(Ordering::SeqCst), 1);
    assert!(!msgs1.is_empty());

    // The model changes the code on disk → the diff fingerprint changes.
    std::fs::write(repo.join("a.rs"), "fn main() { let x = 2; let y = 3; }\n").unwrap();

    // Reaction 2: the diff CHANGED → the judge fires again.
    let (msgs2, src2) = poll_finalization_follow_up(
        &config,
        "sys",
        &msgs_run,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;
    assert_eq!(
        calls.load(Ordering::SeqCst),
        2,
        "judge re-fires when the diff changed"
    );
    assert!(!msgs2.is_empty());
    assert_eq!(src2, FollowUpSource::Critic);

    let _ = std::fs::remove_dir_all(&repo);
}

#[tokio::test]
async fn advisory_review_unaffected_by_last_reviewed_fingerprint() {
    use crate::agent::agent_loop::types::CodeReviewMode;

    // Advisory is one-shot via critic_done; the Blocking-only dedupe (last_fp)
    // must never suppress it. A set fingerprint cannot block the Advisory judge.
    let calls = Arc::new(AtomicUsize::new(0));
    let mut config = build_config();
    config.critic_fn = Some(counting_judge(&calls));
    // default code_review_mode is Advisory; assert it explicitly.
    assert_eq!(config.code_review_mode, CodeReviewMode::Advisory);

    let msgs_run = run_with_tool_result();
    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        last_reviewed_fingerprint: Some(999), // must NOT suppress Advisory
        last_review_findings: None,
        ..Default::default()
    };
    let (emit, _rx) = tokio::sync::mpsc::channel(8);

    let (msgs1, src1) = poll_finalization_follow_up(
        &config,
        "sys",
        &msgs_run,
        &mut gates,
        GateInputs::default(),
        &emit,
    )
    .await;
    assert_eq!(
        calls.load(Ordering::SeqCst),
        1,
        "Advisory judge fires despite a set gates.last_reviewed_fingerprint"
    );
    assert!(!msgs1.is_empty());
    assert_eq!(src1, FollowUpSource::Critic);
    assert!(gates.critic_done, "Advisory flips the one-shot flag");
}

// ── track-work advisory (R3): edited files but no active todo ──────────────

/// An assistant turn whose only content is a single call to `tool`.
fn assistant_calling(tool: &str) -> LoopMessage {
    LoopMessage::Assistant(AssistantMessage::new(
        vec![ContentBlock::ToolCall {
            id: "tc1".into(),
            name: tool.into(),
            arguments: serde_json::json!({}),
        }],
        StopReason::ToolUse,
    ))
}

/// An assistant turn whose content is the given blocks (dirge-g2ex detector tests).
fn assistant_blocks(blocks: Vec<ContentBlock>) -> LoopMessage {
    LoopMessage::Assistant(AssistantMessage::new(blocks, StopReason::Stop))
}

/// An assistant turn whose only content is one text block.
fn assistant_text(text: &str) -> LoopMessage {
    assistant_blocks(vec![ContentBlock::Text { text: text.into() }])
}

#[test]
fn turn_made_file_edits_detects_edit_tools_only() {
    assert!(turn_made_file_edits(&[assistant_calling("edit")]));
    assert!(turn_made_file_edits(&[assistant_calling("write")]));
    assert!(turn_made_file_edits(&[assistant_calling("apply_patch")]));
    // Read-only / execute-only turns are not "file edits".
    assert!(!turn_made_file_edits(&[assistant_calling("read")]));
    assert!(!turn_made_file_edits(&[assistant_calling("bash")]));
    assert!(!turn_made_file_edits(&[]));
}

// ── dirge-g2ex: did the run end waiting on the user? ───────────────────────
//
// `awaiting_user_response` is the pure predicate that lets the finalization
// gates finalize a turn that ended in a question instead of re-entering until
// the model guesses. Covers the prose-question shapes the prompts ask for
// (plain, bolded, question-then-options) and the must-not-match cases (tool
// calls, statements, non-assistant tails, unterminated code).

#[test]
fn awaiting_user_response_plain_trailing_question() {
    assert!(awaiting_user_response(&[assistant_text(
        "Which approach do you prefer?"
    )]));
}

#[test]
fn awaiting_user_response_bolded_question() {
    assert!(awaiting_user_response(&[assistant_text(
        "**Which approach?**"
    )]));
}

#[test]
fn awaiting_user_response_question_then_numbered_options() {
    assert!(awaiting_user_response(&[assistant_text(
        "Which database should I use?\n1. PostgreSQL\n2. MySQL\n3. SQLite"
    )]));
}

#[test]
fn awaiting_user_response_question_then_bulleted_options() {
    assert!(awaiting_user_response(&[assistant_text(
        "Which database should I use?\n\n- PostgreSQL\n- MySQL\n- SQLite"
    )]));
}

#[test]
fn awaiting_user_response_question_then_marker_variants() {
    // Every option-list marker the detector recognizes is dropped, so the
    // question one line up still wins.
    assert!(awaiting_user_response(&[assistant_text(
        "Pick one?\n* red\n+ green\n• blue\n1) alpha\n(2) beta\na) gamma\nb. delta"
    )]));
}

#[test]
fn awaiting_user_response_fullwidth_question_mark() {
    assert!(awaiting_user_response(&[assistant_text("進めますか？")]));
}

#[test]
fn awaiting_user_response_statement_is_false() {
    assert!(!awaiting_user_response(&[assistant_text(
        "I've updated the file."
    )]));
}

#[test]
fn awaiting_user_response_question_but_made_tool_calls_is_false() {
    // Still working — the run didn't actually stop to wait.
    let msg = assistant_blocks(vec![
        ContentBlock::ToolCall {
            id: "tc1".into(),
            name: "edit".into(),
            arguments: serde_json::json!({}),
        },
        ContentBlock::Text {
            text: "Which file should I edit next?".into(),
        },
    ]);
    assert!(!awaiting_user_response(&[msg]));
}

#[test]
fn awaiting_user_response_non_assistant_tail_is_false() {
    // The last message is a user/tool message, not the assistant's turn.
    assert!(!awaiting_user_response(&[LoopMessage::User(
        UserMessage::text("which?")
    )]));
}

#[test]
fn awaiting_user_response_empty_content_is_false() {
    assert!(!awaiting_user_response(&[assistant_text("")]));
    assert!(!awaiting_user_response(&[assistant_blocks(vec![])]));
}

#[test]
fn awaiting_user_response_question_in_middle_statement_last_is_false() {
    assert!(!awaiting_user_response(&[assistant_text(
        "Which database?\nActually, never mind — I'll go with Postgres."
    )]));
}

#[test]
fn awaiting_user_response_question_inside_unterminated_fence_is_false() {
    // Odd number of ``` fences → don't parse prose out of code.
    assert!(!awaiting_user_response(&[assistant_text(
        "Here's my attempt:\n```\nfn lookup() -> Option<i32>?"
    )]));
}

#[test]
fn awaiting_user_response_terminated_fence_with_question_after_is_true() {
    // Even fences: the ``` block is closed, and the trailing prose asks a question.
    assert!(awaiting_user_response(&[assistant_text(
        "Here's the code:\n```\nfn main() {}\n```\nIs this what you wanted?"
    )]));
}

#[test]
fn awaiting_user_response_multiple_text_blocks_last_is_question() {
    let msg = assistant_blocks(vec![
        ContentBlock::Text {
            text: "Let me think through the options.".into(),
        },
        ContentBlock::Text {
            text: "Which one do you want?".into(),
        },
    ]);
    assert!(awaiting_user_response(&[msg]));
}

#[test]
fn awaiting_user_response_no_last_message_is_false() {
    assert!(!awaiting_user_response(&[]));
}

/// The advisory fires only when a real session made file edits with an empty
/// active list and the one-shot budget is unspent. Pure — no global mirror.
#[test]
fn should_advise_untracked_work_gate() {
    // Fires: session + edits + empty list + budget available.
    assert!(should_advise_untracked_work(Some("s"), 0, 0, true));
    // No file edits this turn → nothing to track.
    assert!(!should_advise_untracked_work(Some("s"), 0, 0, false));
    // Active todos already exist → the ordinary todo nudge covers it.
    assert!(!should_advise_untracked_work(Some("s"), 0, 2, true));
    // No session (e.g. --no-session / a fork) → never advise.
    assert!(!should_advise_untracked_work(None, 0, 0, true));
    // One-shot: budget spent.
    assert!(!should_advise_untracked_work(
        Some("s"),
        MAX_TRACK_NUDGES,
        0,
        true
    ));
}

// ── early track-work nudge (dirge-track v2): model-visible reminder ────────

/// The early track-work reminder message is a model-visible `LoopMessage::User`
/// (not a UI-only SystemNotice), with an imperative nudge in the same tone as
/// the unfinished-todo nudge.
#[test]
fn early_track_work_reminder_is_model_visible_user_message() {
    let msg = track_work_reminder_message();
    // Must be a User message so the model reads it on its next turn.
    match &msg {
        LoopMessage::User(u) => {
            let text = u.text_joined();
            assert!(
                text.contains("[track]"),
                "expected [track] tag prefix, got: {text}"
            );
            assert!(
                text.contains("write_todo_list"),
                "expected write_todo_list mention, got: {text}"
            );
            assert!(
                text.contains("in_progress"),
                "expected in_progress mention, got: {text}"
            );
        }
        other => panic!("expected LoopMessage::User, got {other:?}"),
    }
}

/// build_early_track_work_reminder returns the message only when all
/// conditions hold: session, budget unspent, no active todos, file edits.
#[test]
fn build_early_track_work_reminder_gate() {
    // Fires: session + budget + no todos + edits.
    assert!(build_early_track_work_reminder(Some("s"), 0, 0, true).is_some());
    // No file edits → silent.
    assert!(build_early_track_work_reminder(Some("s"), 0, 0, false).is_none());
    // Has active todos → silent (ordinary todo nudge covers it).
    assert!(build_early_track_work_reminder(Some("s"), 0, 2, true).is_none());
    // No session → silent.
    assert!(build_early_track_work_reminder(None, 0, 0, true).is_none());
    // Budget spent → silent (one-shot).
    assert!(build_early_track_work_reminder(Some("s"), MAX_TRACK_NUDGES, 0, true).is_none());
}

/// The reminder is a LoopMessage::User, not a SystemNotice — the model must
/// see it. This test asserts the returned message role.
#[test]
fn early_track_work_reminder_role_is_user() {
    let msg = build_early_track_work_reminder(Some("s"), 0, 0, true)
        .expect("should fire when all conditions met");
    assert!(
        matches!(msg, LoopMessage::User(_)),
        "expected User message, got {msg:?}"
    );
}

fn temp_dir(suffix: &str) -> std::path::PathBuf {
    let dir = std::env::temp_dir().join(format!(
        "dirge-ksjl-{}-{}-{suffix}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    dir
}

/// A tool that always fails. Distinct args per call so the storm
/// breaker (which only suppresses *identical* repeats) lets every call
/// through — the scenario the failure tracker exists to catch.
#[derive(Debug)]
struct FailingTool;
impl LoopTool for FailingTool {
    fn name(&self) -> &str {
        "boom"
    }
    fn description(&self) -> &str {
        "Always fails"
    }
    fn label(&self) -> &str {
        "Boom"
    }
    fn parameters(&self) -> &Value {
        static EMPTY: std::sync::OnceLock<Value> = std::sync::OnceLock::new();
        EMPTY.get_or_init(|| serde_json::json!({"type": "object"}))
    }
    fn execute<'a>(
        &'a self,
        _id: &'a str,
        _args: Value,
        _signal: AbortSignal,
        _on_update: LoopToolUpdate,
    ) -> Pin<Box<dyn Future<Output = Result<super::super::LoopToolResult, String>> + Send + 'a>>
    {
        Box::pin(async move { Err("boom: nothing matched".to_string()) })
    }
}

/// dirge-opdt: three consecutive *distinct* tool failures inject a
/// recovery checkpoint into the conversation. Distinct args dodge the
/// storm breaker, proving the failure tracker covers the gap storm
/// leaves open.
#[tokio::test]
async fn consecutive_distinct_failures_inject_recovery_checkpoint() {
    let mut ctx = empty_context();
    ctx.tools.push(std::sync::Arc::new(FailingTool));

    let factory = canned_factory(vec![
        tool_use_response("c1", "boom", serde_json::json!({"n": 1})),
        tool_use_response("c2", "boom", serde_json::json!({"n": 2})),
        tool_use_response("c3", "boom", serde_json::json!({"n": 3})),
        text_response("giving up"),
    ]);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(256);
    let messages = run_agent_loop(
        vec![user("do the thing")],
        ctx,
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    let checkpoint = messages.iter().find_map(|m| match m {
        LoopMessage::User(u) => {
            let t = u.text_joined();
            if t.contains("[Recovery checkpoint]") {
                Some(t)
            } else {
                None
            }
        }
        _ => None,
    });
    let body =
        checkpoint.expect("a recovery checkpoint must be injected after 3 distinct failures");
    assert!(body.contains("3 tool calls in a row have failed"));
    assert!(body.contains("boom: nothing matched"));
    assert!(body.contains("DIFFERENT next step"));
}

/// A single failure followed by a success leaves no checkpoint — the
/// streak resets on the good result.
#[tokio::test]
async fn failure_then_success_injects_no_checkpoint() {
    let mut ctx = empty_context();
    ctx.tools.push(std::sync::Arc::new(FailingTool));
    ctx.tools.push(std::sync::Arc::new(EchoTool::new()));

    let factory = canned_factory(vec![
        tool_use_response("c1", "boom", serde_json::json!({"n": 1})),
        tool_use_response("c2", "echo", serde_json::json!({"v": 1})),
        tool_use_response("c3", "boom", serde_json::json!({"n": 2})),
        text_response("ok"),
    ]);

    let (tx, _rx) = mpsc::channel::<LoopEvent>(256);
    let messages = run_agent_loop(
        vec![user("go")],
        ctx,
        build_config(),
        AbortSignal::new(),
        &tx,
        &factory,
        None,
        None,
    )
    .await;
    drop(tx);

    assert!(
        !messages.iter().any(|m| matches!(
            m,
            LoopMessage::User(u) if u.text_joined().contains("[Recovery checkpoint]")
        )),
        "a success between failures must reset the streak"
    );
}

// dirge-x6yi: the turn-start issue-board reminder now produces separate
// Active / Backlog sections via `board_reminder_split`. The extracted reader
// keeps the same behavior — a real board yields the reminder, a missing db
// yields None without panicking.
#[test]
fn issue_board_reminder_block_reads_board_and_tolerates_missing_db() {
    let dir = std::env::temp_dir().join(format!(
        "dirge-x6yi-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let db_path = dir.join("state.db");

    // Unassigned (passive) issue: appears under Backlog section.
    let store = crate::extras::issue_db::IssueStore::open_at(&db_path).unwrap();
    store
        .create("wire up telemetry", "", None, None, None)
        .unwrap();

    let block = super::issue_board_reminder_block(&db_path, Some("sess-1"))
        .expect("a non-empty board yields a reminder");
    // Passive issue → Backlog section, not Active.
    assert!(
        block.contains("Backlog"),
        "passive issue must be in Backlog section: {block}"
    );
    assert!(
        !block.contains("Active work queue"),
        "no active issues → no Active section: {block}"
    );
    assert!(block.contains("wire up telemetry"), "{block}");

    // Missing db → best-effort None, no panic.
    assert!(super::issue_board_reminder_block(&dir.join("nope.db"), Some("sess-1")).is_none());

    let _ = std::fs::remove_dir_all(&dir);
}

// ── last_action_failed_and_stopped ──────────────────────────────────────

/// Helper: construct a `LoopMessage::ToolResult` for test use.
fn tool_err(id: &str, name: &str, is_error: bool) -> LoopMessage {
    LoopMessage::ToolResult(crate::agent::agent_loop::message::ToolResultMessage {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        content: vec![crate::agent::agent_loop::message::ContentBlock::Text {
            text: "error output".to_string(),
        }],
        details: serde_json::json!({}),
        is_error,
    })
}

/// Like `tool_err` but with caller-controlled result text — lets a test feed
/// a permission-denial excerpt or a storm-breaker backfill stub verbatim
/// (dirge-g3xv).
fn tool_err_text(id: &str, name: &str, is_error: bool, text: &str) -> LoopMessage {
    LoopMessage::ToolResult(crate::agent::agent_loop::message::ToolResultMessage {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        content: vec![crate::agent::agent_loop::message::ContentBlock::Text {
            text: text.to_string(),
        }],
        details: serde_json::json!({}),
        is_error,
    })
}

fn asst_no_tools(text: &str) -> LoopMessage {
    LoopMessage::Assistant(crate::agent::agent_loop::message::AssistantMessage::new(
        vec![crate::agent::agent_loop::message::ContentBlock::Text {
            text: text.to_string(),
        }],
        crate::agent::agent_loop::message::StopReason::Stop,
    ))
}

fn asst_with_tool(id: &str, name: &str, args: serde_json::Value) -> LoopMessage {
    LoopMessage::Assistant(crate::agent::agent_loop::message::AssistantMessage::new(
        vec![crate::agent::agent_loop::message::ContentBlock::ToolCall {
            id: id.to_string(),
            name: name.to_string(),
            arguments: args,
        }],
        crate::agent::agent_loop::message::StopReason::ToolUse,
    ))
}

#[test]
fn last_action_failed_and_stopped_true_on_error_tool_then_text() {
    // Tail: ToolResult(is_error=true), Assistant(no tool calls)
    let msgs = vec![
        user("do it"),
        asst_with_tool("c1", "read", serde_json::json!({"path": "/x"})),
        tool_err("c1", "read", true),
        asst_no_tools("failed, let me stop"),
    ];
    assert!(last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_false_when_all_tool_results_ok() {
    // Tail: ToolResult(is_error=false), Assistant(no tools)
    let msgs = vec![
        user("do it"),
        asst_with_tool("c1", "read", serde_json::json!({"path": "/x"})),
        tool_err("c1", "read", false),
        asst_no_tools("done"),
    ];
    assert!(!last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_false_when_no_tool_result_before_final_assistant() {
    // Tail: Assistant(text), Assistant(text) — the anti-loop / model-replied-to-nudge case.
    let msgs = vec![
        user("go"),
        asst_with_tool("c1", "read", serde_json::json!({"path": "/x"})),
        tool_err("c1", "read", true),
        asst_no_tools("nudged reply 1"),
        asst_no_tools("nudged reply 2"),
    ];
    assert!(!last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_false_when_last_assistant_has_tool_calls() {
    let msgs = vec![
        user("go"),
        tool_err("c1", "read", true),
        asst_with_tool("c2", "write", serde_json::json!({"path": "/y"})),
    ];
    assert!(!last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_false_when_last_is_not_assistant() {
    let msgs = vec![
        user("go"),
        asst_with_tool("c1", "read", serde_json::json!({})),
        tool_err("c1", "read", true),
    ];
    assert!(!last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_false_on_empty() {
    let msgs: Vec<LoopMessage> = vec![];
    assert!(!last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_detects_error_among_mixed_results() {
    // Multiple ToolResults, one success then one error.
    let msgs = vec![
        user("go"),
        asst_with_tool("c1", "read", serde_json::json!({"path": "/a"})),
        tool_err("c1", "read", false),
        asst_with_tool("c2", "write", serde_json::json!({"path": "/b"})),
        tool_err("c2", "write", true),
        asst_no_tools("write failed, stopping"),
    ];
    assert!(last_action_failed_and_stopped(&msgs));
}

// dirge-g3xv: permission denials and storm-breaker backfill stubs are NOT
// retryable — the resume-after-failure gate must not arm for them.
#[test]
fn last_action_failed_and_stopped_false_on_permission_denial() {
    // A permission/approval refusal is only unblockable by the user; re-issuing
    // re-prompts. The gate must not arm (RED currently: true).
    let msgs = vec![
        user("go"),
        asst_with_tool("c1", "bash", serde_json::json!({})),
        tool_err_text("c1", "bash", true, "Permission denied by user"),
        asst_no_tools("you denied it"),
    ];
    assert!(!last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_false_on_suppressed_backfill_stub() {
    // The backfill stub literally means "do NOT repeat". Re-issuing re-triggers
    // the suppressed call, so the gate must not arm (RED currently: true).
    let msgs = vec![
        user("go"),
        asst_with_tool("c1", "bash", serde_json::json!({})),
        tool_err_text(
            "c1",
            "bash",
            true,
            crate::agent::agent_loop::tools::SUPPRESSED_CALL_NOTE,
        ),
        asst_no_tools("ok, stopping"),
    ];
    assert!(!last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_true_on_genuine_error() {
    // A genuine, mechanically-recoverable failure (bad edit args) still arms the
    // gate. Regression guard — should already pass and keep passing.
    let msgs = vec![
        user("go"),
        asst_with_tool("c1", "edit", serde_json::json!({})),
        tool_err_text("c1", "edit", true, "old_string not found in file"),
        asst_no_tools("gave up"),
    ];
    assert!(last_action_failed_and_stopped(&msgs));
}

#[test]
fn last_action_failed_and_stopped_true_when_mixed_denial_and_genuine() {
    // A denial result AND a genuine-error result in the same tail: a real
    // retryable failure is still present, so the gate arms.
    let msgs = vec![
        user("go"),
        asst_with_tool("c1", "bash", serde_json::json!({})),
        tool_err_text("c1", "bash", true, "Permission denied by user"),
        asst_with_tool("c2", "edit", serde_json::json!({})),
        tool_err_text("c2", "edit", true, "old_string not found in file"),
        asst_no_tools("stopping"),
    ];
    assert!(last_action_failed_and_stopped(&msgs));
}

// ── bounded resume counter (MAX_RESUME_NUDGE) ───────────────────────────

#[test]
fn last_action_failed_and_stopped_bounded() {
    // When resume_nudges is already at MAX_RESUME_NUDGE, the gate must not fire.
    let msgs = vec![
        user("do it"),
        asst_with_tool("c1", "read", serde_json::json!({"path": "/x"})),
        tool_err("c1", "read", true),
        asst_no_tools("failed"),
    ];
    let resume_nudges = MAX_RESUME_NUDGE;
    assert!(!(resume_nudges < MAX_RESUME_NUDGE && last_action_failed_and_stopped(&msgs)));
}

// ── mid-run fast-check reminder (dirge-uw2l.2, RAX R1) ──────────────────

/// The mid-run reminder fires only with tiers engaged, budget unspent, and
/// enough unverified edits piled up. Pure — no gate, no LLM.
#[test]
fn should_nudge_fast_verify_gate() {
    // Fires: tiered mode + budget available + threshold reached.
    assert!(should_nudge_fast_verify(
        GateMode::Advisory,
        0,
        FAST_VERIFY_EDIT_THRESHOLD,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal
    ));
    assert!(should_nudge_fast_verify(
        GateMode::Blocking,
        0,
        FAST_VERIFY_EDIT_THRESHOLD,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal
    ));
    // Off is byte-identical to the untiered loop — never nudges, however
    // many edits pile up.
    assert!(!should_nudge_fast_verify(
        GateMode::Off,
        0,
        99,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal
    ));
    // Below threshold: one or two edits may be mid-sequence.
    assert!(!should_nudge_fast_verify(
        GateMode::Advisory,
        0,
        FAST_VERIFY_EDIT_THRESHOLD - 1,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal
    ));
    // One-shot: budget spent.
    assert!(!should_nudge_fast_verify(
        GateMode::Advisory,
        MAX_VERIFY_NUDGES,
        99,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal
    ));
}

/// The built message carries the verifier tag (so the UI attributes it to
/// the system, dirge-i75f) and asks for the CHEAP tier now, explicitly
/// deferring the full suite — that split is the whole point of the round.
#[test]
fn build_fast_verify_reminder_message() {
    let msg = build_fast_verify_reminder(
        GateMode::Advisory,
        0,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal,
        FAST_VERIFY_EDIT_THRESHOLD,
    )
    .expect("threshold reached in a tiered mode");
    let text = match msg {
        LoopMessage::User(u) => u.text_joined(),
        _ => panic!("expected a user message"),
    };
    assert!(text.contains(VERIFY_TAG), "carries the tag: {text}");
    assert!(text.contains("FAST"), "asks for the fast tier: {text}");
    assert!(text.contains("full suite"), "defers the slow tier: {text}");
    assert!(
        build_fast_verify_reminder(
            GateMode::Off,
            0,
            crate::agent::agent_loop::capability::CapabilityTier::Nominal,
            99
        )
        .is_none()
    );
    assert!(
        build_fast_verify_reminder(
            GateMode::Advisory,
            0,
            crate::agent::agent_loop::capability::CapabilityTier::Nominal,
            1
        )
        .is_none()
    );
}

/// Bounded: once the budget is spent the reminder can never fire again,
/// however far the edit count runs. Guarantees it can't loop.
#[test]
fn fast_verify_nudge_bounded_once() {
    assert!(
        build_fast_verify_reminder(
            GateMode::Advisory,
            MAX_VERIFY_NUDGES,
            crate::agent::agent_loop::capability::CapabilityTier::Nominal,
            10
        )
        .is_none()
    );
    assert!(
        build_fast_verify_reminder(
            GateMode::Blocking,
            MAX_VERIFY_NUDGES,
            crate::agent::agent_loop::capability::CapabilityTier::Nominal,
            10
        )
        .is_none()
    );
}

// ── harness-notice mirror (dirge-uw2l.7) ────────────────────────────────

/// Every tag the harness injects under must be recognized, or that steer
/// stays invisible to headless consumers. Pins the list against a tag being
/// added to a nudge but forgotten here.
#[test]
fn harness_tag_of_recognizes_every_injection_tag() {
    for tag in HARNESS_TAGS {
        let text = format!("{tag} some guidance text");
        assert_eq!(
            harness_tag_of(&text),
            Some(*tag),
            "tag {tag} not recognized"
        );
    }
    // Leading whitespace is tolerated (messages are built with formatters).
    assert_eq!(harness_tag_of("  [stall] x"), Some("[stall]"));
}

/// An ordinary user message — or user steering — mirrors nothing. The notice
/// is for harness-authored injections only, so a human's own words are never
/// echoed back at them as a system line.
#[test]
fn harness_tag_of_ignores_ordinary_user_text() {
    assert!(harness_tag_of("fix the failing test").is_none());
    assert!(harness_tag_of("[not-a-real-tag] hello").is_none());
    assert!(harness_tag_of("").is_none());
    // A tag mentioned mid-sentence isn't an injection.
    assert!(harness_tag_of("I saw a [stall] in the log").is_none());
}

// ── safe-state auto restore, end to end (dirge-uw2l.6) ──────────────────
// `coverage_verified_restore` is the ONLY function in the safe-state rung
// that writes to the user's files, so it gets a real git repo, a real
// snapshot store, and real files on disk rather than a mocked seam.

#[cfg(test)]
mod auto_restore_tests {
    use super::*;
    use crate::agent::tools::snapshots;
    use std::path::{Path, PathBuf};

    fn git(dir: &Path, args: &[&str]) -> bool {
        std::process::Command::new("git")
            .arg("-C")
            .arg(dir)
            .args(args)
            .output()
            .map(|o| o.status.success())
            .unwrap_or(false)
    }

    /// A git repo with one committed source file, plus an isolated snapshot
    /// store. Returns None when git is unavailable so the test skips.
    fn repo(tag: &str) -> Option<PathBuf> {
        let dir = std::env::temp_dir().join(format!("dirge-auto-{}-{tag}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(dir.join("src")).ok()?;
        if !git(&dir, &["init", "-q"]) {
            return None;
        }
        let _ = git(&dir, &["config", "user.email", "t@t"]);
        let _ = git(&dir, &["config", "user.name", "t"]);
        std::fs::write(dir.join("src/a.rs"), "fn a() { /* green */ }\n").ok()?;
        git(&dir, &["add", "-A"]).then_some(())?;
        git(&dir, &["commit", "-qm", "green"]).then_some(())?;
        Some(dir)
    }

    /// The happy path: every file that changed since green went through an
    /// edit tool, so the store covers it and the tree is put back.
    #[test]
    fn restores_when_every_change_is_covered() {
        let _g = snapshots::TEST_GATE.lock_ignore_poison();
        snapshots::clear();
        let Some(dir) = repo("covered") else { return };
        let file = dir.join("src/a.rs");

        // Green: clean tree, fingerprint it.
        snapshots::begin_turn("green-turn");
        let green_fp = crate::agent::agent_loop::worktree_probe::fingerprint(&dir).expect("git");

        // Post-green turn: edit THROUGH the capture path, as an edit tool does.
        snapshots::begin_turn("after-green");
        snapshots::capture(&file);
        std::fs::write(&file, "fn a() { BROKEN }\n").unwrap();

        let n = coverage_verified_restore(Some(&dir), Some(&green_fp), "green-turn");
        assert_eq!(n, Some(1), "one covered file restored");
        assert_eq!(
            std::fs::read_to_string(&file).unwrap(),
            "fn a() { /* green */ }\n",
            "file is back at its green content"
        );
        snapshots::clear();
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// The case the whole gate exists for: a file mutated OUT OF BAND (as
    /// `sed -i` or a formatter would) is invisible to the snapshot store, so
    /// coverage is incomplete and nothing is touched. A partial restore here
    /// would leave a tree that never existed.
    #[test]
    fn declines_and_touches_nothing_when_a_bash_style_mutation_is_present() {
        let _g = snapshots::TEST_GATE.lock_ignore_poison();
        snapshots::clear();
        let Some(dir) = repo("uncovered") else { return };
        let covered = dir.join("src/a.rs");
        let sedded = dir.join("src/sedded.rs");

        snapshots::begin_turn("green-turn");
        let green_fp = crate::agent::agent_loop::worktree_probe::fingerprint(&dir).expect("git");

        snapshots::begin_turn("after-green");
        snapshots::capture(&covered);
        std::fs::write(&covered, "fn a() { BROKEN }\n").unwrap();
        // No capture() — exactly what a `bash` write looks like to the store.
        std::fs::write(&sedded, "fn sed() {}\n").unwrap();

        let n = coverage_verified_restore(Some(&dir), Some(&green_fp), "green-turn");
        assert_eq!(n, None, "incomplete coverage must decline");
        assert_eq!(
            std::fs::read_to_string(&covered).unwrap(),
            "fn a() { BROKEN }\n",
            "declining must leave the tree exactly as it was — no partial restore"
        );
        assert!(sedded.exists(), "the uncaptured file is untouched too");
        snapshots::clear();
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// No repo, no green fingerprint, and nothing-changed all decline.
    /// These are the "proceed blind" cases; every one must be a no-op.
    #[test]
    fn declines_without_ground_truth_or_changes() {
        let _g = snapshots::TEST_GATE.lock_ignore_poison();
        snapshots::clear();
        let Some(dir) = repo("blind") else { return };

        snapshots::begin_turn("green-turn");
        let green_fp = crate::agent::agent_loop::worktree_probe::fingerprint(&dir).expect("git");
        snapshots::begin_turn("after-green");

        assert_eq!(
            coverage_verified_restore(None, Some(&green_fp), "green-turn"),
            None,
            "no repo path → decline"
        );
        assert_eq!(
            coverage_verified_restore(Some(&dir), None, "green-turn"),
            None,
            "no green fingerprint → decline"
        );
        assert_eq!(
            coverage_verified_restore(Some(&dir), Some(&green_fp), "green-turn"),
            None,
            "nothing changed since green → nothing to restore, and no false claim"
        );
        snapshots::clear();
        let _ = std::fs::remove_dir_all(&dir);
    }
}

/// dirge-uw2l.6: `code_review_repo` is a TEST-ONLY override — production
/// leaves it None and means "the process CWD" (the same convention the
/// code-review diff capture follows). Reading the field directly would give
/// the safe-state coverage check no repo in every real session, so auto
/// would silently decline forever and read as a feature that never fires.
#[test]
fn safe_state_repo_falls_back_to_cwd_in_production() {
    let mut cfg = build_config();
    cfg.code_review_repo = None;
    assert_eq!(
        safe_state_repo(&cfg),
        std::env::current_dir().ok(),
        "production (None) must resolve to the CWD, not to no-repo"
    );

    let explicit = std::path::PathBuf::from("/tmp/some-repo");
    cfg.code_review_repo = Some(explicit.clone());
    assert_eq!(
        safe_state_repo(&cfg),
        Some(explicit),
        "an explicit override still wins"
    );
}

// ---------------------------------------------------------------------------
// dirge-5mtx.2 — the mid-turn boundary arbiter.
//
// The finalization boundary has emitted at most one gate per pass since
// dirge-vcsn. The mid-turn boundary did not: track-work, fast-verify, the
// progress signal, the file-touch reminder and the safe-state/reflection
// rungs each pushed independently, so up to five harness messages could land
// before a single assistant turn.
// ---------------------------------------------------------------------------

/// Guards with both engines effectively disarmed, so a test can trip exactly
/// the nudges it means to.
fn quiet_guards() -> crate::agent::agent_loop::activity::LoopGuards {
    crate::agent::agent_loop::activity::LoopGuards::new(
        crate::agent::agent_loop::storm::StormBreaker::new(99, 99, None, None),
        crate::agent::agent_loop::failure_tracker::FailureTracker::new(99),
    )
}

/// Several nudges eligible at once → exactly ONE message, and it is the
/// highest-priority one. Before the arbiter these stacked.
#[test]
fn boundary_emits_at_most_one_nudge() {
    let mut cfg = build_config();
    cfg.session_id = Some("s1".into());
    cfg.verification_tiers_mode = GateMode::Advisory;
    // Verifier with edits and nothing run → fast-verify is eligible.
    let verifier = crate::agent::agent_loop::verifier::VerifierGate::new();
    for i in 0..5 {
        verifier.record_outcome(
            "edit",
            &serde_json::json!({ "path": format!("src/f{i}.rs") }),
            &crate::agent::agent_loop::result::LoopToolResult {
                content: vec![serde_json::json!({"type":"text","text":"ok"})],
                details: serde_json::json!(null),
                terminate: None,
            },
            false,
        );
    }
    cfg.verifier = Some(verifier);
    // Progress monitor primed to be barren.
    cfg.progress = Some(crate::agent::agent_loop::progress::ProgressTracker::new(
        2, 2,
    ));

    let guards = quiet_guards();
    let mut tally = crate::agent::agent_loop::gate_tally::GateTally::new();
    let mut track = 0u8;
    let mut verify = 0u8;
    // An edit this turn makes the track-work nudge eligible too.
    let msgs = vec![LoopMessage::Assistant(AssistantMessage::new(
        vec![ContentBlock::ToolCall {
            id: "c1".into(),
            name: "edit".into(),
            arguments: serde_json::json!({"path": "src/f0.rs"}),
        }],
        StopReason::Stop,
    ))];

    let hit = crate::agent::agent_loop::run::poll_boundary_nudge(
        &cfg,
        &guards,
        None,
        &msgs,
        1,
        &mut track,
        &mut verify,
        &mut tally,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal,
    );
    let (_msg, which) = hit.expect("something should fire");
    // Track-work outranks fast-verify and progress.
    assert_eq!(
        which,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::TrackWork
    );
    // Exactly one nudge recorded across every variant.
    let total: u32 = [
        crate::agent::agent_loop::gate_tally::BoundaryNudge::TrackWork,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::FastVerify,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::FileTouch,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::ProgressStall,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::ProgressPrologue,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::ProgressBudget,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::SafeState,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::ReflectionCheckpoint,
    ]
    .iter()
    .map(|n| tally.nudge_count(*n))
    .sum();
    assert_eq!(total, 1, "exactly one nudge per boundary");
}

/// Safe-state (EXEC rung 3) supersedes the recovery checkpoint it replaces.
/// This used to be a hand-written special case; it is now just precedence.
#[test]
fn safe_state_outranks_everything_else() {
    let mut cfg = build_config();
    cfg.session_id = Some("s1".into());
    let guards = quiet_guards();
    let mut tally = crate::agent::agent_loop::gate_tally::GateTally::new();
    let mut track = 0u8;
    let mut verify = 0u8;

    let hit = crate::agent::agent_loop::run::poll_boundary_nudge(
        &cfg,
        &guards,
        Some("abort and re-plan".into()),
        &[],
        1,
        &mut track,
        &mut verify,
        &mut tally,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal,
    );
    let (_m, which) = hit.expect("safe-state fires");
    assert_eq!(
        which,
        crate::agent::agent_loop::gate_tally::BoundaryNudge::SafeState
    );
    assert_eq!(
        tally
            .nudge_count(crate::agent::agent_loop::gate_tally::BoundaryNudge::ReflectionCheckpoint),
        0,
        "rung 3 replaces rung 2, never adds to it"
    );
}

/// Nothing eligible → no message, and no budget spent.
#[test]
fn quiet_boundary_emits_nothing() {
    let cfg = build_config();
    let guards = quiet_guards();
    let mut tally = crate::agent::agent_loop::gate_tally::GateTally::new();
    let mut track = 0u8;
    let mut verify = 0u8;
    let hit = crate::agent::agent_loop::run::poll_boundary_nudge(
        &cfg,
        &guards,
        None,
        &[],
        1,
        &mut track,
        &mut verify,
        &mut tally,
        crate::agent::agent_loop::capability::CapabilityTier::Nominal,
    );
    assert!(hit.is_none());
    assert_eq!(track, 0);
    assert_eq!(verify, 0);
}

// ── dirge-5mtx.4: is the model BLOCKED on the user, or OFFERING a next step? ──
//
// `awaiting_user_response` is gate 0 of `poll_finalization_follow_up`: when it
// returns true the run finalizes immediately and the verifier, critic, todo and
// open-issues gates are all SKIPPED for that boundary. A false positive
// therefore disables the entire finalization stack, silently — skipping a gate
// produces no output, so nothing in the transcript shows it happened.
//
// The existing tests above cover the SYNTACTIC shapes (bolded, followed by an
// option list, inside a code fence). They do not cover the semantic split that
// actually decides whether skipping the gates is right:
//
//   BLOCKED  — the model cannot proceed without a decision. Finalizing is
//              correct; re-entering would make it guess.
//   OFFERING — the model finished the work and proposed more. Finalizing is
//              WRONG: the work is exactly what the gates exist to check.
//
// Both end in '?', so the trailing-'?' heuristic cannot tell them apart. These
// two tests fix the current error rate in place so any change to this gate has
// to confront it.

/// Cases the trailing-'?' heuristic classifies correctly.
#[test]
fn awaiting_user_corpus_heuristic_is_right_here() {
    // Genuinely blocked — a decision is required to continue.
    for t in [
        "Which database should I use?",
        "Do you want me to use the async or the blocking client?",
        "I can't tell which config is authoritative — which one should I edit?",
        "Before I touch the migration, should I back up the table first?",
    ] {
        assert!(
            awaiting_user_response(&[assistant_text(t)]),
            "should read as blocked: {t}"
        );
    }
    // Plain statements — not questions at all.
    for t in [
        "I've updated the file and the tests pass.",
        "Done — the parser now handles the negated forms.",
        "That change is already covered by the existing test.",
    ] {
        assert!(
            !awaiting_user_response(&[assistant_text(t)]),
            "should not read as blocked: {t}"
        );
    }
}

/// Cases it gets WRONG. Every one of these is the model finishing work and
/// offering to do more — an offer, not a block — so finalizing without running
/// the gates skips exactly the work the gates were built to check.
///
/// These assert the CURRENT (incorrect) behaviour deliberately, so the error
/// rate is visible and versioned rather than discovered later. When a
/// classifier lands (dirge-5mtx.4) these assertions must flip, and the diff
/// that flips them IS the measured improvement.
#[test]
fn awaiting_user_corpus_known_misclassifications() {
    let offers_misread_as_blocked = [
        "I've added the parser and its tests. Want me to wire it into the loop as well?",
        "The bug is fixed and the suite is green. Shall I also update the changelog?",
        "That's the refactor done. Should I run the full test suite now?",
        "Implemented and committed. Anything else you'd like me to pick up?",
        "The migration script is written. Would you like me to run it against staging?",
    ];
    let mut misread = 0;
    for t in offers_misread_as_blocked {
        if awaiting_user_response(&[assistant_text(t)]) {
            misread += 1;
        }
    }
    assert_eq!(
        misread,
        offers_misread_as_blocked.len(),
        "documented state: the heuristic reads EVERY completed-work offer as \
         'blocked on the user' and skips the finalization gates. If this count \
         dropped, a classifier landed — update this test to assert the \
         improvement rather than the defect."
    );
}

/// dirge-5mtx.4: with a classifier armed, the offers the heuristic misreads
/// are classified correctly — and the genuinely-blocked cases still finalize.
///
/// This is the counterpart to `awaiting_user_corpus_known_misclassifications`
/// above, which pins the heuristic's 5-of-5 failure on the same phrasings. The
/// stub judge answers the way a real one would; what is under test is the
/// wiring and the fallback behaviour, not the model.
#[tokio::test]
async fn awaiting_user_classifier_fixes_the_offer_cases() {
    // Stub: OFFERING (index 1) for anything mentioning finished work, BLOCKED
    // (index 0) otherwise. Deliberately keyed on the message, not on a counter,
    // so the test fails if the wrong text reaches the judge.
    let classify: crate::agent::agent_loop::critic::ClassifyFn =
        std::sync::Arc::new(|question: String, _opts: &'static [&'static str]| {
            Box::pin(async move {
                let q = question.to_lowercase();
                let offering = q.contains("i've added")
                    || q.contains("is fixed")
                    || q.contains("that's the refactor")
                    || q.contains("implemented and committed")
                    || q.contains("is written");
                Ok(if offering { 1usize } else { 0usize })
            })
                as std::pin::Pin<
                    Box<dyn std::future::Future<Output = anyhow::Result<usize>> + Send>,
                >
        });
    let mut cfg = build_config();
    cfg.classify_fn = Some(classify);

    // The five the heuristic gets wrong — all offers, none blocking.
    for t in [
        "I've added the parser and its tests. Want me to wire it into the loop as well?",
        "The bug is fixed and the suite is green. Shall I also update the changelog?",
        "That's the refactor done. Should I run the full test suite now?",
        "Implemented and committed. Anything else you'd like me to pick up?",
        "The migration script is written. Would you like me to run it against staging?",
    ] {
        assert!(
            !crate::agent::agent_loop::run::is_awaiting_user(&cfg, &[assistant_text(t)]).await,
            "offer must no longer read as blocked: {t}"
        );
    }
    // Genuinely blocked still finalizes — the fix must not simply always
    // return false, which would disable the gate rather than correct it.
    for t in [
        "Which database should I use?",
        "Do you want me to use the async or the blocking client?",
    ] {
        assert!(
            crate::agent::agent_loop::run::is_awaiting_user(&cfg, &[assistant_text(t)]).await,
            "genuinely blocked must still finalize: {t}"
        );
    }
}

/// A turn with no question mark never reaches the judge. This is the common
/// case, so paying for a classifier call on it would be a real cost.
#[tokio::test]
async fn awaiting_user_no_question_mark_never_calls_the_judge() {
    let calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let seen = calls.clone();
    let classify: crate::agent::agent_loop::critic::ClassifyFn =
        std::sync::Arc::new(move |_q: String, _o: &'static [&'static str]| {
            seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Box::pin(async move { Ok(0usize) })
                as std::pin::Pin<
                    Box<dyn std::future::Future<Output = anyhow::Result<usize>> + Send>,
                >
        });
    let mut cfg = build_config();
    cfg.classify_fn = Some(classify);
    assert!(
        !crate::agent::agent_loop::run::is_awaiting_user(
            &cfg,
            &[assistant_text("I've updated the file.")]
        )
        .await
    );
    assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 0);
}

/// A classifier error falls back to the heuristic rather than failing the
/// turn — the gate must answer something, and the pre-fix answer is the right
/// one when the better signal is unavailable.
#[tokio::test]
async fn awaiting_user_classifier_error_falls_back_to_heuristic() {
    let classify: crate::agent::agent_loop::critic::ClassifyFn =
        std::sync::Arc::new(|_q: String, _o: &'static [&'static str]| {
            Box::pin(async move { anyhow::bail!("judge unavailable") })
                as std::pin::Pin<
                    Box<dyn std::future::Future<Output = anyhow::Result<usize>> + Send>,
                >
        });
    let mut cfg = build_config();
    cfg.classify_fn = Some(classify);
    // Heuristic says blocked (trailing '?'), so the fallback does too.
    assert!(
        crate::agent::agent_loop::run::is_awaiting_user(
            &cfg,
            &[assistant_text("Which database should I use?")]
        )
        .await
    );
}

/// No classifier configured → byte-identical to the old behaviour.
#[tokio::test]
async fn awaiting_user_without_a_classifier_is_the_old_heuristic() {
    let cfg = build_config();
    assert!(cfg.classify_fn.is_none());
    for t in [
        "Which database should I use?",
        "I've added the parser and its tests. Want me to wire it into the loop as well?",
    ] {
        assert_eq!(
            crate::agent::agent_loop::run::is_awaiting_user(&cfg, &[assistant_text(t)]).await,
            awaiting_user_response(&[assistant_text(t)]),
            "unconfigured path must match the heuristic exactly: {t}"
        );
    }
}

/// dirge-mu46: a completeness-only verdict must NOT be deduped away.
///
/// The Blocking dedupe skips the judge when the diff is unchanged, so a
/// declined finding isn't re-raised verbatim. But the judge also rules on
/// COMPLETENESS from the transcript, which keeps growing between reactions
/// even when nothing lands on disk. Skipping wholesale meant an objectively
/// incomplete task could finalize on the model's say-so — the reaction that
/// would have re-judged it never ran.
///
/// The dedupe now additionally requires that the previous reaction actually
/// raised diff findings. Here it didn't (INCOMPLETE with no FINDINGS block),
/// so reaction 2 must re-judge even though the diff is byte-identical.
#[tokio::test]
async fn blocking_completeness_only_verdict_is_re_judged_on_unchanged_diff() {
    use std::sync::atomic::{AtomicUsize, Ordering};
    let calls = Arc::new(AtomicUsize::new(0));
    let seen = calls.clone();
    // Completeness gap, NO findings block — nothing to duplicate.
    let judge: crate::agent::agent_loop::critic::CriticFn = Arc::new(move |_p: String| {
        seen.fetch_add(1, Ordering::SeqCst);
        Box::pin(async {
            Ok("VERDICT: INCOMPLETE\n- the error path is still untested".to_string())
        })
    });
    let mut config = build_config();
    config.critic_fn = Some(judge);
    config.code_review_mode = CodeReviewMode::Blocking;
    let (emit, _rx) = tokio::sync::mpsc::channel(64);
    let msgs_run = run_with_tool_result();

    let mut gates = GateStates {
        critic_done: false,
        code_review_reacts: 0u8,
        last_reviewed_fingerprint: None,
        last_review_findings: None,
        ..Default::default()
    };

    for reaction in 1..=2 {
        let _ = poll_finalization_follow_up(
            &config,
            "sys",
            &msgs_run,
            &mut gates,
            GateInputs::default(),
            &emit,
        )
        .await;
        assert_eq!(
            calls.load(Ordering::SeqCst),
            reaction,
            "reaction {reaction}: a completeness-only verdict must be re-judged, \
             not deduped away with the diff"
        );
    }
    assert!(
        gates.last_review_findings.is_none(),
        "no diff findings were ever raised, so nothing was there to duplicate"
    );
}

// ── dirge-5mtx.7: the derivation. Thresholds keyed on the OBSERVED tier. ─────
//
// Only Strong changes anything, and only in the direction of LESS
// intervention. Nominal is the bit-identical default, so an unmeasured or
// in-range run behaves exactly as before — which matters because a behavioural
// change cannot be validated against the measured ~2x run-to-run noise floor
// (dirge-5mtx.6, FM-5). What CAN be asserted is the structural claim these
// tests make: which way each tier moves, and that the default path is untouched.
//
// Struggling is deliberately NOT keyed on here. It sits below the supported
// capability range and has never been observed firing; wiring a threshold to a
// state we have never seen is the mistake FM-4 names.

use crate::agent::agent_loop::capability::CapabilityTier;

/// Nominal reproduces today's constant exactly. This is the no-op guarantee
/// the whole design rests on: a default install is unaffected.
#[test]
fn fast_verify_threshold_is_unchanged_at_nominal() {
    // Three edits is the shipped threshold — fires at 3, not at 2.
    assert!(!should_nudge_fast_verify(
        GateMode::Advisory,
        0,
        2,
        CapabilityTier::Nominal
    ));
    assert!(should_nudge_fast_verify(
        GateMode::Advisory,
        0,
        3,
        CapabilityTier::Nominal
    ));
}

/// Strong must NOT relax the verify nudge — it behaves exactly as Nominal.
///
/// An earlier cut scaled this threshold up for Strong, reasoning that extra
/// latitude for a demonstrably-coping model could not cause a nudge storm.
/// The reasoning holds but the risk is inverted: the counters observe
/// tool-call mechanics only, so a Strong reading says nothing about whether
/// the model verifies its work — and both failures on record (the 60-turn
/// thrash, the wrong-gate green) came from models this estimator reads as
/// Strong. Relaxing verification pressure on that exact class is backwards.
#[test]
fn strong_does_not_relax_the_verify_nudge() {
    for edits in [3u32, 4, 10] {
        assert_eq!(
            should_nudge_fast_verify(GateMode::Advisory, 0, edits, CapabilityTier::Strong),
            should_nudge_fast_verify(GateMode::Advisory, 0, edits, CapabilityTier::Nominal),
            "Strong must be bit-identical to Nominal at {edits} edits"
        );
    }
    // Nominal still fires at the base threshold, so this is not "gate off".
    assert!(should_nudge_fast_verify(
        GateMode::Advisory,
        0,
        3,
        CapabilityTier::Nominal
    ));
}

/// Struggling is the only tier that moves a threshold, and only toward
/// earlier help: it is nudged to verify before the base count is reached.
#[test]
fn struggling_runs_are_asked_to_verify_sooner() {
    assert!(
        !should_nudge_fast_verify(GateMode::Advisory, 0, 1, CapabilityTier::Nominal),
        "one edit is below the base threshold"
    );
    assert!(
        should_nudge_fast_verify(GateMode::Advisory, 0, 2, CapabilityTier::Struggling),
        "a failing run should be asked to verify before the base count"
    );
}

/// Off mode stays off at every tier. The tier scales WHEN a gate fires, never
/// WHETHER an operator disabled it.
#[test]
fn tier_never_re_enables_a_disabled_gate() {
    for tier in [
        CapabilityTier::Strong,
        CapabilityTier::Nominal,
        CapabilityTier::Struggling,
    ] {
        assert!(
            !should_nudge_fast_verify(GateMode::Off, 0, 99, tier),
            "off must stay off at {tier:?}"
        );
    }
}

/// The budget is still respected at every tier — scaling the threshold must
/// not become a way around the per-run ceiling.
#[test]
fn tier_does_not_bypass_the_nudge_budget() {
    for tier in [
        CapabilityTier::Strong,
        CapabilityTier::Nominal,
        CapabilityTier::Struggling,
    ] {
        assert!(
            !should_nudge_fast_verify(GateMode::Advisory, MAX_VERIFY_NUDGES, 99, tier),
            "spent budget must hold at {tier:?}"
        );
    }
}

// ---------------------------------------------------------------------------
// dirge-5mtx.7: hallucinated tool names must reach the capability estimator.
//
// `record_hallucinated_tool_name` had a unit test that called it directly and
// ZERO production callers, so the counter was structurally always 0 and its
// weight could never contribute. These drive the function the loop actually
// calls, which is the part that was missing.
// ---------------------------------------------------------------------------

#[test]
fn unknown_tool_name_is_counted_as_hallucinated() {
    let known = ["read", "write", "bash"];
    let mut tally = crate::agent::agent_loop::gate_tally::GateTally::new();
    record_tool_result_signals(&mut tally, "search_files", true, &known);
    assert_eq!(tally.hallucinated_tool_names(), 1);
    // It is ALSO an errored call — the stacking is deliberate, matching how
    // repair_invalid already stacks with errored.
    assert_eq!(tally.errored_tool_calls(), 1);
    assert_eq!(tally.tool_calls(), 1);
}

#[test]
fn known_tool_that_errors_is_not_hallucinated() {
    let known = ["read", "write", "bash"];
    let mut tally = crate::agent::agent_loop::gate_tally::GateTally::new();
    record_tool_result_signals(&mut tally, "bash", true, &known);
    assert_eq!(
        tally.hallucinated_tool_names(),
        0,
        "a real tool misused is a different signal from an invented name"
    );
    assert_eq!(tally.errored_tool_calls(), 1);
}

#[test]
fn successful_call_is_never_hallucinated() {
    let known = ["read"];
    let mut tally = crate::agent::agent_loop::gate_tally::GateTally::new();
    // A name that isn't in the list can't actually succeed, but the guard is
    // on is_error so the classification can never fire on a working call.
    record_tool_result_signals(&mut tally, "mystery", false, &known);
    assert_eq!(tally.hallucinated_tool_names(), 0);
    assert_eq!(tally.errored_tool_calls(), 0);
    assert_eq!(tally.tool_calls(), 1);
}

#[test]
fn hallucinated_names_accumulate_across_calls() {
    let known = ["read"];
    let mut tally = crate::agent::agent_loop::gate_tally::GateTally::new();
    record_tool_result_signals(&mut tally, "view", true, &known);
    record_tool_result_signals(&mut tally, "open_file", true, &known);
    record_tool_result_signals(&mut tally, "read", true, &known);
    assert_eq!(tally.hallucinated_tool_names(), 2);
    assert_eq!(tally.errored_tool_calls(), 3);
    assert_eq!(tally.tool_calls(), 3);
}