traverse-runtime 0.10.1

//! Governed by spec 016-runtime-placement-router
//!
//! Integration tests for [`PlacementRouter`].

use std::sync::{Arc, Mutex};

use serde_json::{Value, json};
use traverse_contracts::{
    BinaryFormat, CapabilityContract, Condition, Entrypoint, EntrypointKind, EventReference,
    Execution, ExecutionConstraints, ExecutionTarget, FilesystemAccess, HostApiAccess, Lifecycle,
    NetworkAccess, Owner, Provenance, ProvenanceSource, SchemaContainer, ServiceType, SideEffect,
    SideEffectKind,
};
use traverse_runtime::{
    events::{
        EventBroker, EventCatalog, EventCatalogEntry, InProcessBroker, JournalConfig,
        LifecycleStatus, TraverseEvent,
    },
    executor::{
        ArtifactType, CapabilityExecutor, ExecutorCapability, ExecutorError, ExecutorOutput,
        NativeExecutor,
    },
    placement::{PlacementConstraintEvaluator, PlacementError, RuntimeSnapshot},
    router::{CapabilityExecutorRegistry, PlacementRouter, RouterError, RouterRequest},
    trace::{
        DurableTraceConfig, DurableTraceJournal, PrivateTraceEntry, PublicTraceEntry,
        TraceDurabilitySink, TraceJournalError, TraceStore,
    },
};

// ---------------------------------------------------------------------------
// Fixtures & helpers
// ---------------------------------------------------------------------------

const TEST_SPEC: &str = "016-runtime-placement-router@1.0.0";

/// Helper: assert result is Err and return the error without using `expect_err`.
fn must_err<T: std::fmt::Debug, E>(result: Result<T, E>, msg: &str) -> Result<E, String> {
    match result {
        Err(e) => Ok(e),
        Ok(v) => Err(format!("{msg}: got Ok({v:?})")),
    }
}

fn base_contract(service_type: ServiceType) -> CapabilityContract {
    CapabilityContract {
        kind: "capability_contract".to_string(),
        schema_version: "1.0.0".to_string(),
        id: "router.tests.subject".to_string(),
        namespace: "router.tests".to_string(),
        name: "subject".to_string(),
        version: "0.1.0".to_string(),
        lifecycle: Lifecycle::Draft,
        owner: Owner {
            team: "traverse-core".to_string(),
            contact: "test@example.com".to_string(),
        },
        summary: "Router integration test subject.".to_string(),
        description: "Used only in router tests.".to_string(),
        inputs: SchemaContainer {
            schema: json!({ "type": "object" }),
        },
        outputs: SchemaContainer {
            schema: json!({ "type": "object" }),
        },
        preconditions: vec![Condition {
            id: "always-met".to_string(),
            description: "No preconditions.".to_string(),
        }],
        postconditions: vec![Condition {
            id: "always-met".to_string(),
            description: "No postconditions.".to_string(),
        }],
        side_effects: vec![SideEffect {
            kind: SideEffectKind::MemoryOnly,
            description: "No durable side effects.".to_string(),
        }],
        emits: Vec::new(),
        consumes: Vec::new(),
        permissions: Vec::new(),
        execution: Execution {
            binary_format: BinaryFormat::Wasm,
            entrypoint: Entrypoint {
                kind: EntrypointKind::WasiCommand,
                command: "run".to_string(),
            },
            preferred_targets: vec![ExecutionTarget::Local],
            constraints: ExecutionConstraints {
                host_api_access: HostApiAccess::None,
                network_access: NetworkAccess::Forbidden,
                filesystem_access: FilesystemAccess::None,
            },
        },
        policies: Vec::new(),
        dependencies: Vec::new(),
        provenance: Provenance {
            source: ProvenanceSource::Greenfield,
            author: "test-author".to_string(),
            created_at: "2026-04-01T00:00:00Z".to_string(),
            spec_ref: Some(TEST_SPEC.to_string()),
            adr_refs: Vec::new(),
            exception_refs: Vec::new(),
        },
        evidence: Vec::new(),
        service_type,
        permitted_targets: vec![ExecutionTarget::Local, ExecutionTarget::Cloud],
        event_trigger: None,
        connector_requirements: Vec::new(),
        state_schema: None,
        use_cases: Vec::new(),
        risk: traverse_contracts::default_risk_metadata(),
    }
}

fn idle_snapshot() -> RuntimeSnapshot {
    RuntimeSnapshot {
        target_loads: [(ExecutionTarget::Local, 0.0)].into_iter().collect(),
    }
}

fn native_executor_capability(capability_id: &str) -> ExecutorCapability {
    ExecutorCapability {
        capability_id: capability_id.to_string(),
        artifact_type: ArtifactType::Native,
        wasm_binary_path: None,
        wasm_checksum: None,
        host_abi_version: None,
        // Inert for these tests: `NativeExecutor` never calls the WASM-only
        // `traverse_host::emit_event` ABI, so `emits`/`service_type` here
        // are never consulted for validation (only `WasmExecutor` does).
        emits: Vec::new(),
        service_type: ServiceType::Stateless,
    }
}

/// A [`CapabilityExecutor`] test double that always returns pre-built
/// `emitted_events`, standing in for events a real `WasmExecutor` would have
/// validated and returned via `traverse_host::emit_event`
/// (spec 098-capability-event-host-abi). Used to test `PlacementRouter`
/// Step 5's own publish-gating logic in isolation from the ABI itself.
struct EventEmittingExecutor {
    output: Value,
    emitted_events: Vec<TraverseEvent>,
}

impl CapabilityExecutor for EventEmittingExecutor {
    fn execute(
        &self,
        _capability: &ExecutorCapability,
        _input: &Value,
    ) -> Result<ExecutorOutput, ExecutorError> {
        Ok(ExecutorOutput {
            value: self.output.clone(),
            emitted_events: self.emitted_events.clone(),
            connector_invocation_evidence: Vec::new(),
        })
    }
}

/// Build a router with a single executor that returns `output` and
/// `emitted_events` (already ABI-validated, per spec 098).
fn make_router_with_emitting_executor(
    output: Value,
    emitted_events: Vec<TraverseEvent>,
    trace_store: Arc<Mutex<TraceStore>>,
    broker: Arc<dyn EventBroker>,
) -> PlacementRouter {
    let mut registry: CapabilityExecutorRegistry = CapabilityExecutorRegistry::new();
    registry.insert(
        ArtifactType::Native,
        Box::new(EventEmittingExecutor {
            output,
            emitted_events,
        }),
    );
    PlacementRouter::new(PlacementConstraintEvaluator, registry, trace_store, broker)
}

/// Build a minimal broker with one active event type.
fn broker_with_event(event_type: &str) -> Result<Arc<InProcessBroker>, String> {
    let catalog = Arc::new(EventCatalog::new());
    catalog
        .register(EventCatalogEntry {
            event_type: event_type.to_string(),
            owner: "router.tests".to_string(),
            version: "1.0.0".to_string(),
            lifecycle_status: LifecycleStatus::Active,
            consumer_count: 0,
        })
        .map_err(|e| e.to_string())?;
    Ok(Arc::new(
        InProcessBroker::new(catalog).map_err(|e| e.to_string())?,
    ))
}

/// Build a router with a single native executor that returns `output`.
fn make_router_with_native(
    output: Value,
    trace_store: Arc<Mutex<TraceStore>>,
    broker: Arc<dyn EventBroker>,
) -> PlacementRouter {
    let mut registry: CapabilityExecutorRegistry = CapabilityExecutorRegistry::new();
    registry.insert(
        ArtifactType::Native,
        Box::new(NativeExecutor::new(move |_| Ok(output.clone()))),
    );
    PlacementRouter::new(PlacementConstraintEvaluator, registry, trace_store, broker)
}

/// Build a [`TraverseEvent`] for tests.
fn sample_event(event_type: &str) -> TraverseEvent {
    TraverseEvent {
        id: uuid::Uuid::new_v4().to_string(),
        source: "traverse-runtime/router.tests.subject".to_string(),
        event_type: event_type.to_string(),
        datacontenttype: "application/json".to_string(),
        time: "2026-04-08T00:00:00Z".to_string(),
        data: json!({}),
        owner: "router.tests".to_string(),
        version: "1.0.0".to_string(),
        lifecycle_status: LifecycleStatus::Active,
        deduplication_id: Some("router-test".to_string()),
        ordering_scope: Some("test".to_string()),
        correlation_id: Some("correlation-test".to_string()),
        causation_id: Some("command-test".to_string()),
        subject_id: None,
        actor_id: None,
    }
}

// ---------------------------------------------------------------------------
// Test: Native capability executes end-to-end; trace is written
// ---------------------------------------------------------------------------

#[test]
fn native_capability_executes_and_writes_trace() -> Result<(), String> {
    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let broker: Arc<dyn EventBroker> = broker_with_event("dev.traverse.test.happened")?;

    let router =
        make_router_with_native(json!({ "result": "ok" }), Arc::clone(&trace_store), broker);

    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract: base_contract(ServiceType::Stateless),
        target_hint: Some(ExecutionTarget::Local),
        runtime_snapshot: idle_snapshot(),
        input: json!({ "key": "value" }),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    let response = router.execute(request).map_err(|e| e.to_string())?;

    // Output is correct
    assert_eq!(response.output, json!({ "result": "ok" }));
    assert!(!response.trace_id.is_empty(), "trace_id must be non-empty");

    // Trace entry was written
    let store = trace_store.lock().map_err(|_| "lock poisoned")?;
    let entries = store.list_public(Some("router.tests.subject"));
    assert_eq!(entries.len(), 1, "exactly one trace entry expected");

    let entry = entries[0];
    assert_eq!(entry.capability_id, "router.tests.subject");
    assert_eq!(
        entry.outcome,
        traverse_runtime::trace::TraceOutcome::Success
    );
    assert!(!entry.placement_target.is_empty());

    Ok(())
}

// ---------------------------------------------------------------------------
// Test: Placement failure → PlacementFailed, no trace
// ---------------------------------------------------------------------------

#[test]
fn placement_failure_returns_error_and_no_trace() -> Result<(), String> {
    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let broker: Arc<dyn EventBroker> = broker_with_event("dev.traverse.test.happened")?;

    let router = make_router_with_native(json!({}), Arc::clone(&trace_store), broker);

    // Contract only permits Cloud; snapshot has Browser overloaded, no Cloud entry → no target survives
    let mut contract = base_contract(ServiceType::Stateless);
    contract.permitted_targets = vec![ExecutionTarget::Cloud];

    // All targets at 100% load → placement fails
    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract,
        target_hint: None,
        runtime_snapshot: RuntimeSnapshot {
            target_loads: [(ExecutionTarget::Cloud, 1.0)].into_iter().collect(),
        },
        input: json!({}),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    let err = must_err(router.execute(request), "expected placement error")?;

    assert_eq!(
        err,
        RouterError::PlacementFailed(PlacementError::NoEligibleTarget)
    );

    // No trace must have been written
    let store = trace_store.lock().map_err(|_| "lock poisoned")?;
    assert!(
        store.list_public(None).is_empty(),
        "no trace must be written on placement failure"
    );

    Ok(())
}

// ---------------------------------------------------------------------------
// Test: Missing executor → ExecutorNotFound
// ---------------------------------------------------------------------------

#[test]
fn missing_executor_returns_not_found_error() -> Result<(), String> {
    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let broker: Arc<dyn EventBroker> = broker_with_event("dev.traverse.test.happened")?;

    // Registry has no executors at all
    let empty_registry: CapabilityExecutorRegistry = CapabilityExecutorRegistry::new();
    let router = PlacementRouter::new(
        PlacementConstraintEvaluator,
        empty_registry,
        Arc::clone(&trace_store),
        broker,
    );

    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract: base_contract(ServiceType::Stateless),
        target_hint: Some(ExecutionTarget::Local),
        runtime_snapshot: idle_snapshot(),
        input: json!({}),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    let err = must_err(router.execute(request), "expected executor-not-found error")?;

    assert!(
        matches!(err, RouterError::ExecutorNotFound(_)),
        "expected ExecutorNotFound, got {err:?}"
    );

    Ok(())
}

// ---------------------------------------------------------------------------
// Test: Subscribable capability publishes emitted events
//
// The output-JSON `emitted_events` convention and PlacementRouter's
// post-hoc `undeclared_event_emission` check are both removed (spec
// 098-capability-event-host-abi FR-004/FR-005) — declared-emission
// validation now happens synchronously, inside `WasmExecutor`'s
// `traverse_host::emit_event` host function (see `executor::wasm` tests),
// not here. These tests cover what remains PlacementRouter's own
// responsibility: Step 5 publishes whatever `emitted_events` the executor
// returns, gated only on `service_type == Subscribable`.
// ---------------------------------------------------------------------------

#[test]
fn subscribable_capability_publishes_events() -> Result<(), String> {
    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let event_type = "dev.traverse.router.test.emitted";
    let broker = broker_with_event(event_type)?;

    let sub = broker
        .subscribe(event_type, "0")
        .map_err(|e| e.to_string())?;

    let broker_arc: Arc<dyn EventBroker> = broker;

    let router = make_router_with_emitting_executor(
        json!({ "done": true }),
        vec![sample_event(event_type)],
        Arc::clone(&trace_store),
        Arc::clone(&broker_arc),
    );

    // Subscribable contract
    let mut contract = base_contract(ServiceType::Subscribable);
    contract.event_trigger = Some("dev.traverse.router.test.triggered".to_string());
    contract.emits = vec![EventReference {
        event_id: event_type.to_string(),
        version: "1.0.0".to_string(),
    }];

    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract,
        target_hint: Some(ExecutionTarget::Local),
        runtime_snapshot: idle_snapshot(),
        input: json!({}),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    router.execute(request).map_err(|e| e.to_string())?;

    let poll = broker_arc
        .poll(&sub.subscription_id, 10)
        .map_err(|e| e.to_string())?;
    assert_eq!(poll.events.len(), 1, "one event must be delivered");
    assert_eq!(poll.events[0].event.event_type, event_type);

    Ok(())
}

// ---------------------------------------------------------------------------
// Test: Stateless capability does NOT publish events even if the executor
// returned some
// ---------------------------------------------------------------------------

#[test]
fn stateless_capability_does_not_publish_events() -> Result<(), String> {
    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let event_type = "dev.traverse.router.test.emitted";
    let broker = broker_with_event(event_type)?;

    let sub = broker
        .subscribe(event_type, "0")
        .map_err(|e| e.to_string())?;

    let broker_arc: Arc<dyn EventBroker> = broker;

    let router = make_router_with_emitting_executor(
        json!({}),
        vec![sample_event(event_type)], // returned by the executor but must not be published
        Arc::clone(&trace_store),
        Arc::clone(&broker_arc),
    );

    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract: base_contract(ServiceType::Stateless), // NOT Subscribable
        target_hint: Some(ExecutionTarget::Local),
        runtime_snapshot: idle_snapshot(),
        input: json!({}),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    router.execute(request).map_err(|e| e.to_string())?;

    let poll = broker_arc
        .poll(&sub.subscription_id, 10)
        .map_err(|e| e.to_string())?;
    assert!(poll.events.is_empty(), "stateless must not publish events");

    Ok(())
}

// ---------------------------------------------------------------------------
// Test: RouterError Display covers all variants
// ---------------------------------------------------------------------------

#[test]
fn router_error_display_covers_all_variants() {
    use traverse_runtime::router::RouterError;

    use traverse_contracts::ViolationRecord;
    let cases: Vec<RouterError> = vec![
        RouterError::PlacementFailed(PlacementError::NoEligibleTarget),
        RouterError::ExecutorNotFound("Wasm".to_string()),
        RouterError::ExecutionFailed("test failure".to_string()),
        RouterError::ContractViolation(vec![ViolationRecord::new(
            "undeclared_event_emission",
            "test.cap",
            "test message",
        )]),
        RouterError::TraceLockPoisoned,
        RouterError::DurableTraceWriteFailed("simulated durable trace write failure".to_string()),
    ];

    for err in &cases {
        let msg = err.to_string();
        assert!(
            !msg.is_empty(),
            "Display must produce non-empty string for {err:?}"
        );
    }
}

// ---------------------------------------------------------------------------
// Test: Executor error → ExecutionFailed, no trace written
// ---------------------------------------------------------------------------

#[test]
fn executor_error_returns_execution_failed() -> Result<(), String> {
    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let broker: Arc<dyn EventBroker> = broker_with_event("dev.traverse.test.happened")?;

    let mut registry: CapabilityExecutorRegistry = CapabilityExecutorRegistry::new();
    registry.insert(
        ArtifactType::Native,
        Box::new(NativeExecutor::new(|_| {
            Err("simulated executor failure".to_string())
        })),
    );

    let router = PlacementRouter::new(
        PlacementConstraintEvaluator,
        registry,
        Arc::clone(&trace_store),
        broker,
    );

    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract: base_contract(ServiceType::Stateless),
        target_hint: Some(ExecutionTarget::Local),
        runtime_snapshot: idle_snapshot(),
        input: json!({}),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    let err = must_err(router.execute(request), "expected execution error")?;

    assert!(
        matches!(err, RouterError::ExecutionFailed(_)),
        "expected ExecutionFailed, got {err:?}"
    );

    Ok(())
}

// ---------------------------------------------------------------------------
// Durable trace integration (spec 079-durable-trace-journal FR-002)
// ---------------------------------------------------------------------------

/// A [`TraceDurabilitySink`] test double that always fails, standing in for
/// a real durable-write failure without filesystem fault injection.
struct FailingTraceDurabilitySink;

impl TraceDurabilitySink for FailingTraceDurabilitySink {
    fn record(
        &self,
        _public: &PublicTraceEntry,
        _private: Option<&PrivateTraceEntry>,
    ) -> Result<String, TraceJournalError> {
        Err(TraceJournalError::Serialize(
            "simulated durable trace write failure".to_string(),
        ))
    }
}

fn trace_journal_test_root(name: &str) -> std::path::PathBuf {
    std::env::temp_dir().join(format!(
        "traverse-router-durable-trace-{name}-{}",
        uuid::Uuid::new_v4()
    ))
}

#[test]
fn durable_trace_write_failure_fails_closed_for_auditable_execution() -> Result<(), String> {
    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let broker: Arc<dyn EventBroker> = broker_with_event("dev.traverse.test.happened")?;

    let router =
        make_router_with_native(json!({ "result": "ok" }), Arc::clone(&trace_store), broker)
            .with_durable_trace(DurableTraceConfig {
                sink: Arc::new(FailingTraceDurabilitySink),
                fail_closed: true,
            });

    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract: base_contract(ServiceType::Stateless),
        target_hint: Some(ExecutionTarget::Local),
        runtime_snapshot: idle_snapshot(),
        input: json!({ "key": "value" }),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    let err = must_err(router.execute(request), "expected a durable-trace failure")?;
    assert!(
        matches!(err, RouterError::DurableTraceWriteFailed(_)),
        "expected DurableTraceWriteFailed, got {err:?}"
    );

    let store = trace_store.lock().map_err(|_| "lock poisoned")?;
    assert!(
        store.list_public(Some("router.tests.subject")).is_empty(),
        "an auditable execution that failed closed must not keep an in-memory-only trace"
    );

    Ok(())
}

#[test]
fn durable_trace_write_failure_does_not_fail_open_execution_when_not_fail_closed()
-> Result<(), String> {
    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let broker: Arc<dyn EventBroker> = broker_with_event("dev.traverse.test.happened")?;

    let router =
        make_router_with_native(json!({ "result": "ok" }), Arc::clone(&trace_store), broker)
            .with_durable_trace(DurableTraceConfig {
                sink: Arc::new(FailingTraceDurabilitySink),
                fail_closed: false,
            });

    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract: base_contract(ServiceType::Stateless),
        target_hint: Some(ExecutionTarget::Local),
        runtime_snapshot: idle_snapshot(),
        input: json!({ "key": "value" }),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    let response = router.execute(request).map_err(|e| e.to_string())?;
    assert_eq!(response.output, json!({ "result": "ok" }));

    let store = trace_store.lock().map_err(|_| "lock poisoned")?;
    assert_eq!(
        store.list_public(Some("router.tests.subject")).len(),
        1,
        "non-audited work must keep its in-memory trace even when the durable write fails"
    );

    Ok(())
}

#[test]
fn durable_trace_journal_round_trips_through_a_real_execution() -> Result<(), String> {
    let root = trace_journal_test_root("round-trip");
    let journal =
        DurableTraceJournal::open(&root, JournalConfig::default()).map_err(|e| e.to_string())?;

    let trace_store = Arc::new(Mutex::new(TraceStore::new()));
    let broker: Arc<dyn EventBroker> = broker_with_event("dev.traverse.test.happened")?;

    let router =
        make_router_with_native(json!({ "result": "ok" }), Arc::clone(&trace_store), broker)
            .with_durable_trace(DurableTraceConfig {
                sink: Arc::new(Mutex::new(journal)),
                fail_closed: true,
            });

    let request = RouterRequest {
        capability_id: "router.tests.subject".to_string(),
        artifact_type: ArtifactType::Native,
        contract: base_contract(ServiceType::Stateless),
        target_hint: Some(ExecutionTarget::Local),
        runtime_snapshot: idle_snapshot(),
        input: json!({ "key": "value" }),
        executor_capability: native_executor_capability("router.tests.subject"),
        trace_id_override: None,
    };

    let response = router.execute(request).map_err(|e| e.to_string())?;

    // Reopen a fresh journal at the same root -- the trace this execution
    // wrote must have survived durably, independent of the in-memory store.
    let reopened =
        DurableTraceJournal::open(&root, JournalConfig::default()).map_err(|e| e.to_string())?;
    assert_eq!(
        reopened.recovery_report().recovered_trace_ids,
        vec![response.trace_id]
    );

    std::fs::remove_dir_all(&root).ok();

    Ok(())
}