//! Regression for invariant #18 — `CompiledGraph::resume_with` and
//! `CompiledGraph::resume_from` go through `dispatch_from_checkpoint`,
//! which emits `record_resumed(checkpoint_id)` on the
//! `ExecutionContext`'s audit sink. Validates the emit fires once per
//! resume, carries the resolved checkpoint id, and stays a no-op when
//! the context has no sink.

#![allow(clippy::unwrap_used, clippy::indexing_slicing)]

use entelix_core::TenantId;
use std::sync::Arc;

use parking_lot::Mutex;

use entelix_core::ThreadKey;
use entelix_core::{AuditSink, AuditSinkHandle, ExecutionContext, Result};
use entelix_graph::{Checkpoint, Checkpointer, InMemoryCheckpointer, StateGraph};
use entelix_runnable::{Runnable, RunnableLambda};

#[derive(Clone, Debug, PartialEq, Eq)]
struct Workflow {
    n: i32,
}

#[derive(Default)]
struct RecordingAuditSink {
    resumes: Mutex<Vec<String>>,
    interrupts: Mutex<Vec<(entelix_core::InterruptionKind, serde_json::Value)>>,
}

impl AuditSink for RecordingAuditSink {
    fn record_sub_agent_invoked(&self, _agent_id: &str, _sub_thread_id: &str) {}
    fn record_agent_handoff(&self, _from: Option<&str>, _to: &str) {}
    fn record_interrupted(
        &self,
        kind: &entelix_core::InterruptionKind,
        payload: &serde_json::Value,
    ) {
        self.interrupts.lock().push((kind.clone(), payload.clone()));
    }
    fn record_resumed(&self, from_checkpoint: &str) {
        self.resumes.lock().push(from_checkpoint.to_owned());
    }
    fn record_memory_recall(&self, _tier: &str, _namespace_key: &str, _hits: usize) {}
    fn record_usage_limit_exceeded(&self, _breach: &entelix_core::UsageLimitBreach) {}

    fn record_context_compacted(&self, _dropped_chars: usize, _retained_chars: usize) {}

    fn record_tool_error_terminal(&self, _kind: entelix_core::ToolErrorKind, _tool_name: &str) {}
}

fn step(delta: i32) -> RunnableLambda<Workflow, Workflow> {
    RunnableLambda::new(move |mut s: Workflow, _ctx| async move {
        s.n += delta;
        Ok::<_, _>(s)
    })
}

#[tokio::test]
async fn resume_emits_resumed_with_checkpoint_id() -> Result<()> {
    let cp = Arc::new(InMemoryCheckpointer::<Workflow>::new());
    let graph = StateGraph::<Workflow>::new()
        .add_node("a", step(1))
        .add_node("b", step(10))
        .add_edge("a", "b")
        .set_entry_point("a")
        .add_finish_point("b")
        .with_checkpointer(cp.clone())
        .compile()?;

    let key = ThreadKey::new(TenantId::new("default"), "thread-resume");
    let checkpoint = Checkpoint::new(&key, 1, Workflow { n: 100 }, Some("b".into()));
    let manual_id = checkpoint.id.clone();
    cp.put(checkpoint).await?;

    let sink = Arc::new(RecordingAuditSink::default());
    let ctx = ExecutionContext::new()
        .with_thread_id("thread-resume")
        .with_audit_sink(AuditSinkHandle::new(Arc::clone(&sink) as Arc<dyn AuditSink>));

    let final_state = graph.resume(&ctx).await?;
    assert_eq!(final_state.n, 110);

    let recorded: Vec<_> = sink.resumes.lock().clone();
    assert_eq!(recorded.len(), 1);
    assert_eq!(recorded[0], manual_id.to_hyphenated_string());
    Ok(())
}

#[tokio::test]
async fn resume_without_sink_stays_silent() -> Result<()> {
    let cp = Arc::new(InMemoryCheckpointer::<Workflow>::new());
    let graph = StateGraph::<Workflow>::new()
        .add_node("a", step(1))
        .set_entry_point("a")
        .add_finish_point("a")
        .with_checkpointer(cp.clone())
        .compile()?;

    let ctx = ExecutionContext::new().with_thread_id("thread-resume-2");
    let _ = graph.invoke(Workflow { n: 0 }, &ctx).await?;
    // No sink wired — resume must succeed and emit nothing.
    let _ = graph.resume(&ctx).await?;
    Ok(())
}

#[tokio::test]
async fn scheduled_interrupt_before_emits_record_interrupted() -> Result<()> {
    // Graph-level `interrupt_before` is a HITL pause point that
    // the agent layer cannot wrap (graph-only callers). The
    // dispatch loop fires `AuditSink::record_interrupted` so the
    // durable audit trail captures the pause regardless of
    // wrapper layer — closes the gap codex flagged on the
    // resume/interrupt asymmetry (`record_resumed` existed alone).
    let cp = Arc::new(InMemoryCheckpointer::<Workflow>::new());
    let graph = StateGraph::<Workflow>::new()
        .add_node("a", step(1))
        .set_entry_point("a")
        .add_finish_point("a")
        .with_checkpointer(cp.clone())
        .interrupt_before(["a"])
        .compile()?;

    let sink = Arc::new(RecordingAuditSink::default());
    let ctx = ExecutionContext::new()
        .with_thread_id("thread-int-1")
        .with_audit_sink(AuditSinkHandle::new(sink.clone() as Arc<dyn AuditSink>));

    let result = graph.invoke(Workflow { n: 0 }, &ctx).await;
    assert!(
        matches!(result, Err(entelix_core::Error::Interrupted { .. })),
        "interrupt_before must raise Error::Interrupted"
    );
    // Snapshot under the guard, assert on owned data — keeps the
    // lock scope minimal and matches the codebase's lock-discipline
    // convention (no guard held across multiple assert statements).
    let recorded: Vec<(entelix_core::InterruptionKind, serde_json::Value)> =
        sink.interrupts.lock().clone();
    assert_eq!(recorded.len(), 1, "exactly one record_interrupted fired");
    let (kind, payload) = &recorded[0];
    assert!(matches!(
        kind,
        entelix_core::InterruptionKind::ScheduledPause {
            phase: entelix_core::InterruptionPhase::Before,
            ..
        }
    ));
    assert_eq!(payload, &serde_json::Value::Null);
    Ok(())
}
