//! The acceptance-blocking §15 chain at the runtime level: a gated tool
//! call is intercepted, previewed, queued, blocks the run, executes only
//! on approval (single-use grant), and is fully audited.

use std::sync::Arc;
use std::time::Duration;

use cm_domain::{
    AccessPolicy, Agent, AgentId, AgentStatus, Role, RunState, User, UserId, Workspace, WorkspaceId,
};
use cm_llm::ScriptedProvider;
use cm_runtime::{RunEventBody, Runtime, RuntimeConfig};
use cm_safety::{approvals, ApprovalStatus, Decision};

const SCENARIOS: &str = r#"
[[scenario]]
marker = "[[scenario:gated-email]]"

[[scenario.turns]]
events = [
    { type = "text", text = "I'll send that email." },
    { type = "tool_use", name = "email.send", input = { to = "ceo@example.com", subject = "Q2 numbers", body = "Revenue is up 14%." } },
]

[[scenario.turns]]
events = [
    { type = "text", text = " The email step is finished." },
]
"#;

struct Seed {
    workspace: Workspace,
    owner: User,
    agent: Agent,
}

async fn seeded(pool: &sqlx::PgPool) -> Seed {
    let workspace = Workspace {
        id: WorkspaceId::new(),
        name: "Acme".into(),
        plan: "team".into(),
    };
    cm_db::repo::workspaces::insert(pool, &workspace)
        .await
        .unwrap();
    let owner = User {
        id: UserId::new(),
        workspace_id: workspace.id,
        email: format!("{}@acme.test", UserId::new()),
        role: Role::Owner,
        display_name: "Owner".into(),
        created_at: time::OffsetDateTime::UNIX_EPOCH,
    };
    cm_db::repo::users::insert(pool, &owner).await.unwrap();
    let agent = Agent {
        id: AgentId::new(),
        workspace_id: workspace.id,
        name: "Scout".into(),
        job_title: "Analyst".into(),
        system_prompt: String::new(),
        avatar: String::new(),
        accent: String::new(),
        wallpaper: String::new(),
        managed_by: owner.id,
        status: AgentStatus::Online,
    };
    cm_db::repo::agents::insert(pool, &agent, &AccessPolicy::default())
        .await
        .unwrap();
    Seed {
        workspace,
        owner,
        agent,
    }
}

fn runtime(pool: sqlx::PgPool) -> Runtime {
    Runtime::new(
        pool,
        Arc::new(ScriptedProvider::from_toml(SCENARIOS).unwrap()),
        RuntimeConfig::basic("scripted", 1024),
    )
}

async fn drain_until_suspended(
    mut rx: tokio::sync::broadcast::Receiver<cm_runtime::RunEventEnvelope>,
) -> Vec<cm_runtime::RunEventEnvelope> {
    let mut events = Vec::new();
    while let Ok(envelope) = rx.recv().await {
        let done = matches!(
            envelope.event,
            RunEventBody::RunSuspended { .. }
                | RunEventBody::RunCompleted { .. }
                | RunEventBody::Error { .. }
        );
        events.push(envelope);
        if done {
            break;
        }
    }
    events
}

async fn outbox_count(pool: &sqlx::PgPool) -> i64 {
    sqlx::query_scalar::<_, i64>("SELECT count(*) FROM outbox")
        .fetch_one(pool)
        .await
        .unwrap()
}

/// Waits until the run reaches a terminal state, reading the journal (the
/// resumed run streams on a new channel; the journal is the durable feed).
async fn wait_terminal(pool: &sqlx::PgPool, run_id: uuid::Uuid) -> RunState {
    for _ in 0..100 {
        let run = cm_db::repo::runs::get(pool, run_id).await.unwrap();
        if run.state != RunState::Running && run.state != RunState::AwaitingApproval {
            return run.state;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    panic!("run never reached a terminal state");
}

async fn start_gated_run(
    pool: &sqlx::PgPool,
    seed: &Seed,
    rt: &Runtime,
) -> (uuid::Uuid, cm_safety::Approval) {
    let session = cm_db::repo::sessions::create(pool, seed.agent.id, seed.workspace.id, "Chat")
        .await
        .unwrap();
    let started = rt
        .send_message(session.id, "send it [[scenario:gated-email]]")
        .await
        .unwrap();
    let events = drain_until_suspended(started.events).await;

    // Intercepted and previewed: the approval_required event carries the
    // exact human-facing preview.
    let preview = events
        .iter()
        .find_map(|e| match &e.event {
            RunEventBody::ApprovalRequired { preview, .. } => Some(preview.clone()),
            _ => None,
        })
        .expect("approval_required event");
    assert_eq!(preview["summary"], "Send email to ceo@example.com");
    assert_eq!(preview["body"], "Revenue is up 14%.");
    assert!(matches!(
        events.last().unwrap().event,
        RunEventBody::RunSuspended { .. }
    ));

    // Blocked: nothing executed, run suspended, approval queued.
    assert_eq!(outbox_count(pool).await, 0);
    let run = cm_db::repo::runs::get(pool, started.run_id).await.unwrap();
    assert_eq!(
        run.state,
        RunState::AwaitingApproval,
        "run error: {:?}",
        run.error
    );
    let pending = approvals::list_pending(pool, seed.workspace.id)
        .await
        .unwrap();
    assert_eq!(pending.len(), 1);
    (started.run_id, pending.into_iter().next().unwrap())
}

#[tokio::test]
async fn gated_call_blocks_then_executes_only_after_approval() {
    let pool = cm_testkit::test_pool().await;
    let seed = seeded(&pool).await;
    let rt = runtime(pool.clone());
    let (run_id, approval) = start_gated_run(&pool, &seed, &rt).await;

    // Approve through the real decision path; the sweeper resumes the run.
    rt.spawn_resume_sweeper(Duration::from_millis(50));
    approvals::decide(&pool, approval.id, seed.owner.id, Decision::Approve)
        .await
        .unwrap();

    assert_eq!(wait_terminal(&pool, run_id).await, RunState::Completed);

    // Approve-only execution: exactly one outbox row, with the approved payload.
    assert_eq!(outbox_count(&pool).await, 1);
    let recipient = sqlx::query_scalar::<_, String>("SELECT recipient FROM outbox LIMIT 1")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(recipient, "ceo@example.com");

    // Audited: the decision is in the audit log.
    let audited = sqlx::query_scalar::<_, i64>(
        "SELECT count(*) FROM audit_log WHERE event_type = 'approval.approved'",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(audited, 1);

    // The journal contains the full chain for replay.
    let journal = cm_db::repo::run_events::list_after(&pool, run_id, 0)
        .await
        .unwrap();
    let kinds: Vec<&str> = journal.iter().map(|e| e.event_type.as_str()).collect();
    assert!(kinds.contains(&"approval_required"));
    assert!(kinds.contains(&"run_suspended"));
    assert!(kinds.contains(&"step_finished"));
    assert_eq!(*kinds.last().unwrap(), "run_completed");
    // Sequence numbers are continuous across the suspension.
    let seqs: Vec<i64> = journal.iter().map(|e| e.seq).collect();
    assert_eq!(seqs, (1..=seqs.len() as i64).collect::<Vec<_>>());

    // The step trace on the reply records the approved execution.
    let session_id = cm_db::repo::runs::get(&pool, run_id)
        .await
        .unwrap()
        .session_id;
    let history = cm_db::repo::messages::history(&pool, session_id)
        .await
        .unwrap();
    let reply = history.last().unwrap();
    assert_eq!(reply.steps.len(), 1);
    assert_eq!(reply.steps[0].status, cm_domain::StepStatus::Ok);
    let text = reply.message.content["text"].as_str().unwrap();
    assert!(text.contains("The email step is finished."), "got: {text}");
}

#[tokio::test]
async fn rejection_executes_nothing_and_informs_the_model() {
    let pool = cm_testkit::test_pool().await;
    let seed = seeded(&pool).await;
    let rt = runtime(pool.clone());
    let (run_id, approval) = start_gated_run(&pool, &seed, &rt).await;

    approvals::decide(&pool, approval.id, seed.owner.id, Decision::Reject)
        .await
        .unwrap();
    rt.resume_run(cm_safety::ResumeReady {
        run_id,
        approval_id: approval.id,
        approved: false,
    })
    .await
    .unwrap();

    assert_eq!(wait_terminal(&pool, run_id).await, RunState::Completed);
    // Nothing executed, ever.
    assert_eq!(outbox_count(&pool).await, 0);

    // The model saw the structured rejection and continued in-band.
    let session_id = cm_db::repo::runs::get(&pool, run_id)
        .await
        .unwrap()
        .session_id;
    let history = cm_db::repo::messages::history(&pool, session_id)
        .await
        .unwrap();
    let reply = history.last().unwrap();
    assert_eq!(reply.steps.len(), 1);
    assert_eq!(reply.steps[0].status, cm_domain::StepStatus::Error);
    assert!(reply.steps[0].output.as_ref().unwrap()["rejected"]
        .as_str()
        .unwrap()
        .contains("rejected"));
}

#[tokio::test]
async fn a_spent_grant_cannot_execute_again() {
    let pool = cm_testkit::test_pool().await;
    let seed = seeded(&pool).await;
    let rt = runtime(pool.clone());
    let (run_id, approval) = start_gated_run(&pool, &seed, &rt).await;

    approvals::decide(&pool, approval.id, seed.owner.id, Decision::Approve)
        .await
        .unwrap();
    // An attacker (or crashed half-resume) already consumed the grant.
    cm_safety::grants::consume(&pool, approval.id)
        .await
        .unwrap();

    rt.resume_run(cm_safety::ResumeReady {
        run_id,
        approval_id: approval.id,
        approved: true,
    })
    .await
    .unwrap();

    assert_eq!(wait_terminal(&pool, run_id).await, RunState::Completed);
    // The gated action did NOT run a second time — no outbox row at all,
    // because the only consumption happened outside the executor.
    assert_eq!(outbox_count(&pool).await, 0);
    let approval_after = approvals::get(&pool, approval.id).await.unwrap();
    assert_eq!(approval_after.status, ApprovalStatus::Approved);
}

#[tokio::test]
async fn ungated_tools_run_without_any_approval_rows() {
    let pool = cm_testkit::test_pool().await;
    let seed = seeded(&pool).await;
    let scenarios = r#"
[[scenario]]
marker = "[[scenario:clock]]"

[[scenario.turns]]
events = [ { type = "tool_use", name = "clock.now", input = {} } ]

[[scenario.turns]]
events = [ { type = "text", text = "Done." } ]
"#;
    let rt = Runtime::new(
        pool.clone(),
        Arc::new(ScriptedProvider::from_toml(scenarios).unwrap()),
        RuntimeConfig::basic("scripted", 1024),
    );
    let session = cm_db::repo::sessions::create(&pool, seed.agent.id, seed.workspace.id, "Chat")
        .await
        .unwrap();
    let started = rt
        .send_message(session.id, "time [[scenario:clock]]")
        .await
        .unwrap();
    drain_until_suspended(started.events).await;

    assert_eq!(
        wait_terminal(&pool, started.run_id).await,
        RunState::Completed
    );
    let pending = approvals::list_pending(&pool, seed.workspace.id)
        .await
        .unwrap();
    assert!(pending.is_empty());
}