//! Runtime integration tests for the "always-on / live" workflow primitives.
//!
//! These prove, at runtime (not just compile time), that:
//!   1. `WorkflowHandler::emit` injects an event into a running live loop and
//!      that event is routed through the step registry.
//!   2. Lenient routing (enabled by `.live()`) drops events with no registered
//!      handler instead of terminating the loop with `WorkflowError::NoHandler`.
//!   3. `.idle_timeout(..)` fires `WorkflowError::IdleTimeout` when the loop
//!      sits idle with no activity.
//!   4. `.cost_budget_usd(..)` fires `WorkflowError::CostBudgetExceeded` once a
//!      step accumulates cost (via `Context::emit_usage`) beyond the ceiling.
//!
//! The harness conventions (manual `Event` impls, `make_step` closure builder,
//! `WorkflowBuilder::step(..)`, `handler.result().await`) mirror
//! `workflow_integration.rs`.

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

use blazen_core::{Context, StepOutput, StepRegistration, WorkflowBuilder, WorkflowError};
use blazen_events::{AnyEvent, Event, Modality, StopEvent, UsageEvent};
use serde::{Deserialize, Serialize};

// ===========================================================================
// Custom event types (manual Event impls, mirroring workflow_integration.rs)
// ===========================================================================

/// An externally-injected ping the live workflow knows how to handle.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PingEvent {
    nonce: u64,
}

impl Event for PingEvent {
    fn event_type() -> &'static str {
        "test::PingEvent"
    }
    fn event_type_id(&self) -> &'static str {
        "test::PingEvent"
    }
    fn as_any(&self) -> &dyn Any {
        self
    }
    fn clone_boxed(&self) -> Box<dyn AnyEvent> {
        Box::new(self.clone())
    }
    fn to_json(&self) -> serde_json::Value {
        serde_json::to_value(self).unwrap()
    }
}

/// An event type that NO step in the live workflow accepts. Used to prove
/// lenient routing drops it instead of killing the loop.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct UnknownEvent {
    payload: String,
}

impl Event for UnknownEvent {
    fn event_type() -> &'static str {
        "test::UnknownEvent"
    }
    fn event_type_id(&self) -> &'static str {
        "test::UnknownEvent"
    }
    fn as_any(&self) -> &dyn Any {
        self
    }
    fn clone_boxed(&self) -> Box<dyn AnyEvent> {
        Box::new(self.clone())
    }
    fn to_json(&self) -> serde_json::Value {
        serde_json::to_value(self).unwrap()
    }
}

/// Trigger that drives the cost-budget step.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SpendEvent {
    cost_usd: f64,
}

impl Event for SpendEvent {
    fn event_type() -> &'static str {
        "test::SpendEvent"
    }
    fn event_type_id(&self) -> &'static str {
        "test::SpendEvent"
    }
    fn as_any(&self) -> &dyn Any {
        self
    }
    fn clone_boxed(&self) -> Box<dyn AnyEvent> {
        Box::new(self.clone())
    }
    fn to_json(&self) -> serde_json::Value {
        serde_json::to_value(self).unwrap()
    }
}

/// A harmless follow-up the spend step emits so the event loop takes another
/// turn after the usage cost is folded into the accumulator (the per-event
/// cost-budget check runs after dispatch, so the budget trips on the loop
/// iteration that processes this follow-up event).
#[derive(Debug, Clone, Serialize, Deserialize)]
struct TickEvent {
    n: u64,
}

impl Event for TickEvent {
    fn event_type() -> &'static str {
        "test::TickEvent"
    }
    fn event_type_id(&self) -> &'static str {
        "test::TickEvent"
    }
    fn as_any(&self) -> &dyn Any {
        self
    }
    fn clone_boxed(&self) -> Box<dyn AnyEvent> {
        Box::new(self.clone())
    }
    fn to_json(&self) -> serde_json::Value {
        serde_json::to_value(self).unwrap()
    }
}

// ===========================================================================
// Helper: build a step that accepts a typed event but returns a raw
// `StepOutput` (so a single step can read/write ctx, emit, and branch).
// ===========================================================================

fn make_raw_step<I>(
    name: &str,
    accepts: &'static str,
    emits: Vec<&'static str>,
    handler: impl Fn(
        I,
        Context,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<StepOutput, WorkflowError>> + Send>,
    > + Send
    + Sync
    + 'static,
) -> StepRegistration
where
    I: Event + 'static,
{
    let handler = Arc::new(handler);
    StepRegistration {
        name: name.to_string(),
        accepts: vec![accepts],
        emits,
        handler: Arc::new(move |event: Box<dyn AnyEvent>, ctx: Context| {
            let handler = handler.clone();
            Box::pin(async move {
                let typed = event
                    .as_any()
                    .downcast_ref::<I>()
                    .ok_or(WorkflowError::EventDowncastFailed {
                        expected: I::event_type(),
                        got: event.event_type_id().to_string(),
                    })?
                    .clone();
                handler(typed, ctx).await
            })
        }),
        max_concurrency: 1,
        semaphore: None,
        timeout: None,
        retry_config: None,
    }
}

/// A minimal `UsageEvent` carrying only a cost, everything else zeroed.
fn usage_with_cost(cost_usd: f64) -> UsageEvent {
    UsageEvent {
        provider: "test".to_string(),
        model: "test-model".to_string(),
        modality: Modality::Llm,
        prompt_tokens: 0,
        completion_tokens: 0,
        total_tokens: 0,
        reasoning_tokens: 0,
        cached_input_tokens: 0,
        audio_input_tokens: 0,
        audio_output_tokens: 0,
        image_count: 0,
        audio_seconds: 0.0,
        video_seconds: 0.0,
        cost_usd: Some(cost_usd),
        latency_ms: 0,
        run_id: uuid::Uuid::nil(),
    }
}

// ===========================================================================
// 1. emit() routes to a live workflow
// ===========================================================================

#[tokio::test]
async fn test_emit_routes_to_live_workflow() {
    // A single step that handles an injected PingEvent: it records a marker in
    // the shared Context and then emits a StopEvent to cleanly end the run.
    // The initial StartEvent is leniently dropped (no handler accepts it), so
    // the loop idles until we inject the PingEvent from outside.
    let ping_step = make_raw_step(
        "ping_handler",
        PingEvent::event_type(),
        vec![StopEvent::event_type()],
        |event: PingEvent, ctx: Context| {
            Box::pin(async move {
                ctx.set("got_ping", true).await;
                Ok(StepOutput::Single(Box::new(StopEvent {
                    result: serde_json::json!({ "nonce": event.nonce }),
                })))
            })
        },
    );

    let workflow = WorkflowBuilder::new("live-emit")
        .step(ping_step)
        .live()
        .build()
        .unwrap();

    // Run with an input the workflow has no handler for -> live mode drops it
    // and the loop idles waiting for external injection.
    let handler = workflow
        .run(serde_json::json!({"boot": true}))
        .await
        .unwrap();

    // Inject the event from outside the running loop.
    handler.emit(PingEvent { nonce: 7 }).unwrap();

    // The injected event drives the loop to completion.
    let result = handler.result().await.unwrap();
    let stop = result.event.downcast_ref::<StopEvent>().unwrap();
    assert_eq!(
        stop.result["nonce"], 7,
        "the StopEvent emitted by the ping handler should carry the injected nonce"
    );
}

// ===========================================================================
// 2. Lenient routing survives an unknown (unhandled) event type
// ===========================================================================

#[tokio::test]
async fn test_lenient_routing_survives_unknown_event() {
    // Same single ping handler as test 1. We first inject an UnknownEvent that
    // NO step accepts. Under lenient routing the loop must drop+warn and keep
    // running (NOT terminate with WorkflowError::NoHandler). We prove the loop
    // survived by then injecting a handled PingEvent and getting a clean
    // StopEvent result.
    let ping_step = make_raw_step(
        "ping_handler",
        PingEvent::event_type(),
        vec![StopEvent::event_type()],
        |event: PingEvent, _ctx: Context| {
            Box::pin(async move {
                Ok(StepOutput::Single(Box::new(StopEvent {
                    result: serde_json::json!({ "nonce": event.nonce }),
                })))
            })
        },
    );

    let workflow = WorkflowBuilder::new("live-lenient")
        .step(ping_step)
        .live()
        .build()
        .unwrap();

    let handler = workflow.run(serde_json::json!(null)).await.unwrap();

    // Inject an event nothing handles -- lenient routing should drop it.
    handler
        .emit(UnknownEvent {
            payload: "nobody handles me".to_string(),
        })
        .unwrap();

    // Then inject a handled event. If the unknown event had killed the loop
    // (with NoHandler), this emit would still succeed but result() below would
    // resolve to Err(NoHandler) instead of the StopEvent.
    handler.emit(PingEvent { nonce: 99 }).unwrap();

    let result = handler.result().await;
    match result {
        Ok(wf_result) => {
            let stop = wf_result.event.downcast_ref::<StopEvent>().unwrap();
            assert_eq!(
                stop.result["nonce"], 99,
                "loop survived the unknown event and completed via the handled PingEvent"
            );
        }
        Err(WorkflowError::NoHandler { event_type }) => {
            panic!(
                "lenient routing did NOT survive: loop terminated with NoHandler for {event_type:?} \
                 -- the unknown event killed the live loop, which is a bug"
            );
        }
        Err(other) => panic!("unexpected error from live workflow: {other:?}"),
    }
}

// ===========================================================================
// 3. idle_timeout fires when no activity occurs
// ===========================================================================

#[tokio::test]
async fn test_idle_timeout_fires() {
    // A step that accepts PingEvent (never sent) so the workflow is valid but
    // does no work. The initial StartEvent is leniently dropped (live mode),
    // then the loop sits idle. With a 150ms idle timeout and nothing
    // forthcoming, the run must terminate with WorkflowError::IdleTimeout.
    let idle_step = make_raw_step(
        "idle_handler",
        PingEvent::event_type(),
        vec![StopEvent::event_type()],
        |_event: PingEvent, _ctx: Context| {
            Box::pin(async move {
                Ok(StepOutput::Single(Box::new(StopEvent {
                    result: serde_json::json!(null),
                })))
            })
        },
    );

    let workflow = WorkflowBuilder::new("live-idle")
        .step(idle_step)
        .live()
        .idle_timeout(Duration::from_millis(150))
        .build()
        .unwrap();

    let handler = workflow.run(serde_json::json!(null)).await.unwrap();

    // Do NOT emit anything: let the idle timer fire.
    let result = handler.result().await;
    match result {
        Err(WorkflowError::IdleTimeout { idle }) => {
            assert_eq!(
                idle,
                Duration::from_millis(150),
                "IdleTimeout should report the configured idle duration"
            );
        }
        other => panic!("expected IdleTimeout, got {other:?}"),
    }
}

// ===========================================================================
// 4. cost_budget_usd fires once accumulated cost exceeds the ceiling
// ===========================================================================

#[tokio::test]
async fn test_cost_budget_exceeded_fires() {
    // The spend step: on SpendEvent it folds a UsageEvent (cost 0.05) into the
    // context cost accumulator via emit_usage, then emits a harmless TickEvent.
    // The per-event cost-budget check runs AFTER an event is dispatched, so the
    // budget trips on the loop iteration that processes the TickEvent (by which
    // point the spend step's emit_usage has already accumulated the cost --
    // the handler future fully resolves, including emit_usage, before its
    // output event is enqueued).
    let spend_step = make_raw_step(
        "spend",
        SpendEvent::event_type(),
        vec![TickEvent::event_type()],
        |event: SpendEvent, ctx: Context| {
            Box::pin(async move {
                ctx.emit_usage(usage_with_cost(event.cost_usd)).await;
                Ok(StepOutput::Single(Box::new(TickEvent { n: 1 })))
            })
        },
    );

    // A no-op handler for TickEvent so the follow-up event is routable (forces
    // the loop to take another turn where the cost check sees the spend).
    let tick_step = make_raw_step(
        "tick",
        TickEvent::event_type(),
        vec![TickEvent::event_type()],
        |event: TickEvent, _ctx: Context| {
            Box::pin(async move {
                // Keep ticking so the loop keeps turning until the budget trips;
                // bound it so a non-firing budget can't loop forever.
                if event.n < 5 {
                    Ok(StepOutput::Single(Box::new(TickEvent { n: event.n + 1 })))
                } else {
                    Ok(StepOutput::Single(Box::new(StopEvent {
                        result: serde_json::json!({ "ticks": event.n }),
                    })))
                }
            })
        },
    );

    let workflow = WorkflowBuilder::new("live-cost")
        .step(spend_step)
        .step(tick_step)
        .live()
        .cost_budget_usd(0.01)
        .build()
        .unwrap();

    let handler = workflow.run(serde_json::json!(null)).await.unwrap();

    // Trigger a spend of $0.05, which exceeds the $0.01 ceiling.
    handler.emit(SpendEvent { cost_usd: 0.05 }).unwrap();

    let result = handler.result().await;
    match result {
        Err(WorkflowError::CostBudgetExceeded { spent, ceiling }) => {
            assert!(
                (ceiling - 0.01).abs() < f64::EPSILON,
                "ceiling should be the configured 0.01, got {ceiling}"
            );
            assert!(
                spent > ceiling,
                "spent ({spent}) must exceed ceiling ({ceiling}) when the budget trips"
            );
            assert!(
                (spent - 0.05).abs() < 1e-9,
                "spent should reflect the single $0.05 usage, got {spent}"
            );
        }
        other => panic!("expected CostBudgetExceeded, got {other:?}"),
    }
}
