//! kremory::memory — Zep-equivalent orchestration layer over kremory::core.
//!
//! Per ADR-Phase-D.0 §"rqlm public API surface" — 4 public async functions
//! plus the scoping types in `types`. Internal orchestration modules added
//! in D.2 stay `pub(crate)`; external consumers use the 4 entry points only.
//!
//! ## Greenfield contract
//!
//! D.1b ships scaffolding only. The 4 public functions return
//! `MemoryError::NotImplemented { .. }` until D.2's TDD lane lands the real
//! implementations. This keeps the public surface visible to downstream
//! consumers (kaironiq, kremory-mcp) while the implementation is built.
//!
//! ## BYOM contract
//!
//! `ChatProvider` is the canonical LLM abstraction across all kremory layers.
//! memory re-exports the trait so consumers can implement it against any backend
//! (OpenAI, Anthropic, Bedrock, vLLM, local GGUF, …).
//!
//! ## Layer ownership
//!
//! Per the canonical Zep / Graphiti split (verified 2026-05-13 via context7;
//! recorded in `project_zep_graphiti_split_canonical.md`):
//!
//! - kremory::core owns per-episode work: `add_episode` cycle = LLM entity / edge
//!   extraction + dedup + fact invalidation + temporal validity inference
//!   + community detection primitive + hybrid retrieval primitives.
//! - kremory::memory owns cross-episode wrappers: multi-tenant scoping + packaged
//!   batch consolidation recipe (`run_dream_phase`) + opinionated retrieval
//!   defaults over core's hybrid search + context-block templates.

pub(crate) mod background_ingestor_handle;
pub mod dream_phase;
pub mod engine_handle;
pub mod events;
pub mod graph;
pub mod llm_adapters;
pub mod scheduler;
#[cfg(any(test, feature = "test-utils"))]
pub mod stub;
pub mod types;

pub(crate) use background_ingestor_handle::BackgroundIngestorGraphHandle;
pub use engine_handle::{EngineGraphHandle, WithConfigParams};
pub use graph::{
    GraphAssertEntityTypeParams, GraphHandle, GraphIngestEpisodeParams, GraphSearchParams,
    GraphSubmitDreamParams,
};
pub use scheduler::{DreamSchedule, DreamSchedulerHandle};
#[cfg(any(test, feature = "test-utils"))]
pub use stub::StubGraphHandle;
pub use types::{
    AwaitOpts, BatchStatus, CancelOutcome, CancelledPhase, ContextTemplate, DreamHandle, DreamMode,
    DreamOpts, DreamPhaseResult, DreamStatus, EpisodeCommit, ImmutabilityLevel, IngestResult,
    InvalidPolicyError, MemoryError, MemoryType, Namespace, NamespacePolicy, Result,
    RetrievedContext, RetrievedContextNewParams, SearchOpts, SourceKind, SourceRef, StructuredFact,
    SubmitOpts,
};
// IngestStatus lives in core::error but is part of the memory API surface.
pub use crate::core::error::IngestStatus;

// Re-export the canonical LLM abstraction trait so SDK consumers depend on
// kremory only and still get the BYOM contract surface. Per ADR-Phase-D.0 §
// "rqlm public API surface". autoagents-llm is a required dep — kremory::memory
// needs ChatProvider unconditionally for the GraphHandle trait surface.
pub use autoagents_llm::chat::ChatProvider;

use std::sync::Arc;
use std::time::Instant;

use uuid::Uuid;

// ── D.6.4 public API surface (ADR §4.10) ─────────────────────────────────────

/// Bundled parameters for [`submit_episode`] — args-as-object per TD-042
/// (rust-conventions §too_many_arguments). `batch_id`: caller-set string
/// grouping this episode with others — plain `Option<String>`, no `BatchRef`
/// wrapper, matching universal prior art.
pub struct SubmitEpisodeParams<'a> {
    pub graph: &'a dyn GraphHandle,
    pub content: &'a str,
    pub source_ref: SourceRef,
    pub structured_facts: Vec<StructuredFact>,
    pub provider: Arc<dyn ChatProvider>,
    pub namespace: Namespace,
    pub batch_id: Option<String>,
    pub opts: SubmitOpts,
    pub sink: Option<Arc<dyn events::EnrichmentEventSink>>,
}

/// Submit one episode for ingest.
///
/// Phase 1 (store + embed) commits synchronously. Episode searchable on return.
/// Phase 2 (LLM enrich) controlled by `params.opts`.
// Substrate primitive; consumer-facing surface is kremory::Memory facade per ADR-027.
pub async fn submit_episode(params: SubmitEpisodeParams<'_>) -> Result<EpisodeCommit> {
    let SubmitEpisodeParams {
        graph,
        content,
        source_ref,
        structured_facts,
        provider,
        namespace,
        batch_id,
        opts,
        sink,
    } = params;
    graph
        .graph_ingest_episode(GraphIngestEpisodeParams {
            namespace: &namespace,
            source_ref: &source_ref,
            content,
            structured_facts: &structured_facts,
            provider,
            batch_id,
            opts,
            sink,
        })
        .await
}

/// Bundled parameters for [`submit_dream_phase`] — args-as-object per TD-042
/// (rust-conventions §too_many_arguments).
pub struct SubmitDreamPhaseParams<'a> {
    pub graph: &'a dyn GraphHandle,
    pub namespace: Namespace,
    pub provider: Arc<dyn ChatProvider>,
    pub batch_id: Option<String>,
    pub opts: DreamOpts,
    pub sink: Option<Arc<dyn events::EnrichmentEventSink>>,
}

/// Submit a batch consolidation (dream phase). Returns immediately.
/// Idempotent on `(scope, batch_id)` key. See ADR §2.10 for CAS semantics.
// Substrate primitive; consumer-facing surface is kremory::Memory facade per ADR-027.
pub async fn submit_dream_phase(params: SubmitDreamPhaseParams<'_>) -> Result<DreamHandle> {
    let SubmitDreamPhaseParams {
        graph,
        namespace,
        provider,
        batch_id,
        opts,
        sink,
    } = params;
    graph
        .graph_submit_dream(GraphSubmitDreamParams {
            namespace: &namespace,
            provider,
            batch_id,
            opts,
            sink,
        })
        .await
}

/// Block until a Phase 2 run reaches a terminal status.
/// `timeout` in `AwaitOpts` is MANDATORY — no unbounded blocking.
/// `tracing::warn!` logged on timeout with `run_id` and elapsed duration.
pub async fn await_enrichment(
    graph: &dyn GraphHandle,
    run_id: Uuid,
    opts: AwaitOpts,
) -> Result<crate::core::error::IngestStatus> {
    use crate::core::error::IngestStatus;
    let start = Instant::now();
    loop {
        let status = graph.graph_ingest_status(run_id).await?;
        if matches!(status, IngestStatus::Complete | IngestStatus::Failed(_)) {
            return Ok(status);
        }
        if start.elapsed() >= opts.timeout {
            tracing::warn!(
                run_id = %run_id,
                elapsed_ms = start.elapsed().as_millis(),
                "await_enrichment timeout exhausted"
            );
            return Err(MemoryError::Timeout);
        }
        tokio::time::sleep(opts.poll_interval).await;
    }
}

/// Block until a Phase 3 dream run reaches a terminal status.
/// `timeout` in `AwaitOpts` is MANDATORY — no unbounded blocking.
/// `tracing::warn!` logged on timeout with `run_id` and elapsed duration.
pub async fn await_dream(
    graph: &dyn GraphHandle,
    run_id: Uuid,
    opts: AwaitOpts,
) -> Result<DreamStatus> {
    let start = Instant::now();
    loop {
        let status = graph.graph_dream_status(run_id).await?;
        if matches!(status, DreamStatus::Complete | DreamStatus::Failed(_)) {
            return Ok(status);
        }
        if start.elapsed() >= opts.timeout {
            tracing::warn!(
                run_id = %run_id,
                elapsed_ms = start.elapsed().as_millis(),
                "await_dream timeout exhausted"
            );
            return Err(MemoryError::Timeout);
        }
        tokio::time::sleep(opts.poll_interval).await;
    }
}

/// Wait until every episode in the batch has reached a terminal status
/// (Complete, Skipped, or Failed). Polls `graph_batch_status(batch_id)`
/// internally with `opts.poll_interval`. Returns the final `BatchStatus`.
///
/// Use this instead of a raw `loop { batch_status(...).await }` — handles
/// the fail-fast timeout boundary correctly and emits `tracing::warn!` on
/// timeout exhaustion with the `batch_id` and elapsed duration. (C1/§5.5)
pub async fn await_batch_enrichment(
    graph: &dyn GraphHandle,
    batch_id: &str,
    opts: AwaitOpts,
) -> Result<BatchStatus> {
    let start = Instant::now();
    loop {
        let status = graph.graph_batch_status(batch_id).await?;
        if status.is_done() {
            return Ok(status);
        }
        if start.elapsed() >= opts.timeout {
            tracing::warn!(
                batch_id,
                elapsed_ms = start.elapsed().as_millis(),
                "await_batch_enrichment timeout exhausted"
            );
            return Err(MemoryError::Timeout);
        }
        tokio::time::sleep(opts.poll_interval).await;
    }
}

// ── Legacy backwards-compat wrappers (D.5b callers; ADR §5.5) ────────────────

/// Legacy wrapper for D.5b callers. New code: use `submit_episode`.
///
/// Warning: `entities_added`, `edges_added`, `facts_invalidated`, `duration_ms`
/// are stub values (1, 0, 0, 0). Accurate counts are available via
/// `EnrichmentEventSink` on the `submit_episode` path. Callers relying on
/// these fields for anything other than log decoration must migrate.
#[deprecated(
    since = "0.1.0",
    note = "Use submit_episode + EnrichmentEventSink for accurate per-episode counts"
)]
// Substrate primitive; consumer-facing surface is kremory::Memory facade per ADR-027.
//
// TD-042 documented exemption: this fn is `#[deprecated]` (removal scheduled).
// Per the treat-cause exemption precedent, args-as-object churn on dying code is
// waste — the allow stays until the fn is removed. New code uses `submit_episode`.
#[allow(clippy::too_many_arguments)]
pub async fn ingest_episode(
    graph: &dyn GraphHandle,
    content: &str,
    source_ref: SourceRef,
    structured_facts: Vec<StructuredFact>,
    provider: Arc<dyn ChatProvider>,
    namespace: Namespace,
) -> Result<IngestResult> {
    let _commit = submit_episode(SubmitEpisodeParams {
        graph,
        content,
        source_ref,
        structured_facts,
        provider,
        namespace,
        batch_id: None,
        opts: SubmitOpts {
            enrich_per_episode: true,
            run_in_background: false,
        },
        sink: None,
    })
    .await?;
    // Phase 2 ran inline (run_in_background = false). No polling needed.
    // Stub counts — callers used these for logging only; acceptable degradation.
    Ok(IngestResult {
        entities_added: 1,
        edges_added: 0,
        facts_invalidated: 0,
        duration_ms: 0,
        stub_entities_inserted: 0,
    })
}

/// Legacy wrapper for D.5b callers. New code: use `submit_dream_phase`.
#[deprecated(
    since = "0.1.0",
    note = "Use submit_dream_phase + await_dream for non-blocking dream orchestration"
)]
pub async fn run_dream_phase(
    graph: &dyn GraphHandle,
    namespace: Namespace,
    provider: Arc<dyn ChatProvider>,
) -> Result<DreamPhaseResult> {
    graph.graph_run_consolidation(&namespace, provider).await
}

/// Bundled parameters for [`search`] — args-as-object per TD-042
/// (rust-conventions §too_many_arguments).
pub struct SearchParams<'a> {
    pub graph: &'a dyn GraphHandle,
    pub query: &'a str,
    pub namespace: Namespace,
    pub opts: SearchOpts,
}

/// Query the graph with memory's opinionated retrieval defaults. Thin
/// orchestration wrapper over [`GraphHandle::graph_search`].
pub async fn search(params: SearchParams<'_>) -> Result<Vec<RetrievedContext>> {
    let SearchParams {
        graph,
        query,
        namespace,
        opts,
    } = params;
    // F4 (ADR adr-memory-builder-as-of-fail-loud / DENT-001): `as_of` point-in-time
    // recall is declared but not yet implemented in SQL. Previously this silently
    // no-op'd (warn + current-state results) — the worst footgun on a bi-temporal
    // engine. Fail loud HERE at the consumption point so the guard covers BOTH the
    // `.as_of()` setter AND the `.opts(SearchOpts{as_of})` escape hatch (one guard,
    // not one of N setters — load-bearing-invariants-at-emit-not-prompt). Real SQL
    // impl is parked as tech-debt; until then the surface must not lie.
    if opts.as_of.is_some() {
        return Err(MemoryError::Core(crate::core::error::Error::Unsupported {
            feature: "as_of point-in-time recall",
        }));
    }
    graph
        .graph_search(GraphSearchParams {
            namespace: &namespace,
            query,
            opts: &opts,
        })
        .await
}

/// Render `results` into the final string handed to the LLM, per the
/// requested template strategy. Closest to Zep's `%{user_summary}` /
/// `%{edges}` / `%{entities}` template placeholders.
///
/// Implemented for all 3 [`ContextTemplate`] variants:
/// - `Entities` — one block per entity: name, summary, source pointers
/// - `EdgeSummary` — one line per entity-source-edge for compact context
/// - `TemporalFacts` — flattens source_refs with `valid_at` annotations
///
/// Empty `results` always renders an empty string. Order is preserved from
/// the input — callers are expected to pass results already sorted by score.
///
/// This is the only memory public fn that has no core primitive dependency,
/// so D.2a (this commit) ships it ahead of the rest. The other three fns
/// (`ingest_episode`, `run_dream_phase`, `search`) wrap core primitives
/// and ship in D.2b once the `Engine<L, Emb>` API shape (generic vs
/// trait-object vs concrete wrapper) is locked.
pub fn context_block(results: &[RetrievedContext], template: ContextTemplate) -> String {
    if results.is_empty() {
        return String::new();
    }
    match template {
        ContextTemplate::Entities => render_entities(results),
        ContextTemplate::EdgeSummary => render_edge_summary(results),
        ContextTemplate::TemporalFacts => render_temporal_facts(results),
    }
}

fn render_entities(results: &[RetrievedContext]) -> String {
    // Detect multi-namespace context: emit [ns:{group_id}] prefix when results
    // span more than one distinct group_id (ADR-029c Decision 7).
    let multi_ns = is_multi_namespace(results);
    let mut out = String::new();
    for (i, r) in results.iter().enumerate() {
        if i > 0 {
            out.push_str("\n\n");
        }
        if multi_ns {
            if let Some(group_id) = namespace_group_id(r) {
                out.push_str("[ns:");
                out.push_str(&group_id);
                out.push_str("] ");
            }
        }
        out.push_str("## ");
        out.push_str(&r.entity_name);
        out.push('\n');
        out.push_str(&r.summary);
        if !r.source_refs.is_empty() {
            out.push_str("\n\nSources: ");
            for (j, sr) in r.source_refs.iter().enumerate() {
                if j > 0 {
                    out.push_str(", ");
                }
                out.push_str(source_kind_label(sr.kind));
                out.push(':');
                out.push_str(&sr.id);
            }
        }
    }
    out
}

fn render_edge_summary(results: &[RetrievedContext]) -> String {
    let mut out = String::new();
    for r in results {
        for sr in &r.source_refs {
            if !out.is_empty() {
                out.push('\n');
            }
            out.push_str("- ");
            out.push_str(&r.entity_name);
            out.push_str(" <- ");
            out.push_str(source_kind_label(sr.kind));
            out.push(':');
            out.push_str(&sr.id);
        }
    }
    out
}

fn render_temporal_facts(results: &[RetrievedContext]) -> String {
    // Detect multi-namespace context: emit [ns:{group_id}] prefix per result
    // when results span more than one distinct group_id (ADR-029c Decision 7).
    let multi_ns = is_multi_namespace(results);
    let mut out = String::new();
    for (i, r) in results.iter().enumerate() {
        if i > 0 {
            out.push('\n');
        }
        for sr in &r.source_refs {
            if multi_ns {
                if let Some(group_id) = namespace_group_id(r) {
                    out.push_str("[ns:");
                    out.push_str(&group_id);
                    out.push_str("] ");
                }
            }
            out.push_str(&r.entity_name);
            out.push_str(" (valid_at=");
            out.push_str(&sr.occurred_at.to_rfc3339());
            out.push_str(") — ");
            out.push_str(&r.summary);
            out.push('\n');
        }
    }
    out.trim_end_matches('\n').to_string()
}

/// Determine whether `results` span more than one distinct namespace group_id.
/// Used by `render_entities` and `render_temporal_facts` to decide whether to
/// emit `[ns:{group_id}]` attribution markers (ADR-029c Decision 7).
///
/// Returns `true` only when at least two distinct, non-`None` group_ids appear
/// in the result set. Single-namespace results and results without namespace
/// attribution always return `false` (no prefix emitted).
fn is_multi_namespace(results: &[RetrievedContext]) -> bool {
    let mut seen: Option<&str> = None;
    for r in results {
        if let Some(ns) = r.namespace.as_ref() {
            let gid = ns.namespace.as_str();
            match seen {
                None => seen = Some(gid),
                Some(prev) if prev != gid => return true,
                _ => {}
            }
        }
    }
    false
}

/// Extract the namespace group_id string from a result for use in the
/// `[ns:{group_id}]` attribution marker. Returns `None` when `namespace` is
/// unset (pre-v0.1.5 results / substrate bypass).
fn namespace_group_id(r: &RetrievedContext) -> Option<String> {
    r.namespace.as_ref().map(|ns| ns.namespace.clone())
}

fn source_kind_label(k: SourceKind) -> &'static str {
    match k {
        SourceKind::Meeting => "meeting",
        SourceKind::Document => "document",
        SourceKind::Chat => "chat",
        SourceKind::Episode => "episode",
    }
}

/// Re-export of the schema module from kremory::core — memory consumers should
/// not need to depend on core directly for the common types they round-trip.
pub use crate::core::schema as core_schema;

// ═══════════════════════════════════════════════════════════════════════════════
// TelemetryConfig + TelemetryHandle (ADR D15)
// ═══════════════════════════════════════════════════════════════════════════════

/// Memory-layer telemetry configuration.
///
/// Passed to `init_telemetry` to wire the metrics recorder and optional OTel
/// OTLP exporter. Callers that want no telemetry pass `TelemetryConfig::default()`.
///
/// # Cardinality note (ADR D7)
///
/// All prefix strings are set once at init time — not per-request. There is
/// no per-call allocation after `init_telemetry` returns.
#[derive(Debug, Clone, Default)]
pub struct TelemetryConfig {
    /// Optional prefix for all metric names (see `Config::metrics_prefix`).
    pub metrics_prefix: Option<String>,
    /// Optional prefix for all tracing span names.
    pub span_prefix: Option<String>,
    /// OTLP endpoint for OTel span export (e.g. `"http://localhost:4317"`).
    /// Requires the `otel` feature flag. Ignored when `otel` feature is absent.
    pub otlp_endpoint: Option<String>,
    /// Optional path to a custom provider-rates.toml file. When `Some`, overrides
    /// the bundled rates. When `None`, the bundled rates are used (idempotent if
    /// already initialized).
    pub rates_path: Option<std::path::PathBuf>,
}

/// Handle returned by [`init_telemetry`]. Keeps the OTel provider alive.
///
/// Drop or call [`TelemetryHandle::shutdown`] at process exit to flush pending spans.
/// If no OTel provider was initialised (default config or `otel` feature absent),
/// `shutdown()` is a no-op.
#[must_use]
pub struct TelemetryHandle {
    /// When the `otel` feature is enabled and an OTLP provider was wired,
    /// this holds the provider so its `Drop` impl can flush buffered spans.
    #[cfg(feature = "otel")]
    provider: Option<opentelemetry_sdk::trace::TracerProvider>,
    #[cfg(not(feature = "otel"))]
    _private: (),
}

impl TelemetryHandle {
    /// Flush pending spans and shut down the OTel provider.
    ///
    /// With the `otel` feature: calls `TracerProvider::shutdown()` to drain the
    /// OTLP export queue before returning. Without the feature: no-op.
    pub fn shutdown(self) {
        #[cfg(feature = "otel")]
        if let Some(provider) = self.provider {
            // Ignore shutdown errors — best-effort flush at process exit.
            let _ = provider.shutdown();
        }
    }
}

/// Errors returned by [`init_telemetry`].
#[derive(Debug, thiserror::Error)]
pub enum TelemetryInitError {
    /// OTLP exporter build failed (only possible when `otel` feature is enabled).
    #[error("OTLP exporter build failed: {0}")]
    Exporter(String),
    /// `tracing-subscriber` global default could not be installed.
    /// Usually means another subscriber was already registered.
    #[error("tracing subscriber init failed: {0}")]
    Subscriber(String),
}

/// Initialise kremory telemetry for the memory layer.
///
/// Wires:
/// - `tracing` structured log emission (always)
/// - OTLP gRPC span export (only when `otel` feature is enabled **and**
///   `config.otlp_endpoint` is `Some`)
///
/// Without the `otel` feature: installs an `EnvFilter` + fmt layer using
/// [`tracing_subscriber::fmt::init`](https://docs.rs/tracing-subscriber) conventions
/// so that `RUST_LOG` still controls verbosity.  This path never fails.
///
/// With the `otel` feature: installs `tracing-subscriber` registry with an
/// `EnvFilter` layer, an `fmt` layer, and a `tracing-opentelemetry` layer that
/// exports spans to `config.otlp_endpoint` (default: `http://localhost:4317`).
///
/// Returns a [`TelemetryHandle`] that MUST be kept alive until process exit.
/// Dropping it early shuts down the OTel provider and loses buffered spans.
///
/// # Errors
///
/// Returns [`TelemetryInitError::Exporter`] if the OTLP exporter fails to build
/// (only when `otel` feature is enabled and `config.otlp_endpoint` is `Some`).
/// Returns [`TelemetryInitError::Subscriber`] if a global tracing subscriber is
/// already installed (harmless in binaries that call this once; check your test harness).
///
/// # Example
///
/// ```no_run
/// # use kremory::{TelemetryConfig, init_telemetry};
/// let handle = init_telemetry(TelemetryConfig {
///     otlp_endpoint: Some("http://localhost:4317".to_string()),
///     ..Default::default()
/// })?;
/// // … run your application …
/// handle.shutdown();
/// # Ok::<(), kremory::TelemetryInitError>(())
/// ```
#[cfg(feature = "otel")]
pub fn init_telemetry(
    config: TelemetryConfig,
) -> std::result::Result<TelemetryHandle, TelemetryInitError> {
    use opentelemetry::trace::TracerProvider as _;
    use opentelemetry_otlp::WithExportConfig as _;
    use opentelemetry_sdk::Resource;
    use tracing_subscriber::layer::SubscriberExt as _;
    use tracing_subscriber::util::SubscriberInitExt as _;

    // Initialize provider rates: custom path takes precedence over bundled.
    if let Some(ref path) = config.rates_path {
        crate::core::rates::init_from_path(path)
            .map_err(|e| TelemetryInitError::Exporter(format!("rates load failed: {e}")))?;
    } else {
        // Best-effort: ignore error if already initialized or TOML is missing.
        let _ = crate::core::rates::init_bundled();
    }

    // Build the OTLP gRPC span exporter.  Endpoint resolution order:
    //   1. `config.otlp_endpoint` (explicit caller config)
    //   2. `OTEL_EXPORTER_OTLP_ENDPOINT` env var (opentelemetry-otlp picks this up automatically)
    //   3. Default: http://localhost:4317
    let mut exporter_builder = opentelemetry_otlp::SpanExporter::builder().with_tonic();
    if let Some(ref endpoint) = config.otlp_endpoint {
        exporter_builder = exporter_builder.with_endpoint(endpoint.as_str());
    }
    let exporter = exporter_builder
        .build()
        .map_err(|e| TelemetryInitError::Exporter(e.to_string()))?;

    // `service.name` resource attribute — identifies this process in the OTLP backend.
    let service_name = config
        .span_prefix
        .as_deref()
        .unwrap_or("kremory")
        .to_string();
    let resource = Resource::new(vec![opentelemetry::KeyValue::new(
        "service.name",
        service_name.clone(),
    )]);

    // Build the SDK tracer provider with a batch processor (async export via Tokio).
    let provider = opentelemetry_sdk::trace::TracerProvider::builder()
        .with_batch_exporter(exporter, opentelemetry_sdk::runtime::Tokio)
        .with_resource(resource)
        .build();

    let tracer = provider.tracer(service_name);
    let otel_layer = tracing_opentelemetry::layer().with_tracer(tracer);

    let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info,kremory=debug"));

    let fmt_layer = tracing_subscriber::fmt::layer().with_target(true);

    tracing_subscriber::registry()
        .with(env_filter)
        .with(fmt_layer)
        .with(otel_layer)
        .try_init()
        .map_err(|e| TelemetryInitError::Subscriber(e.to_string()))?;

    Ok(TelemetryHandle {
        provider: Some(provider),
    })
}

/// Initialise kremory telemetry (no-op stub — enable the `otel` feature for OTLP export).
///
/// Without the `otel` feature kremory still emits structured `tracing` events and
/// `metrics` counters. To forward spans to an OTLP backend, rebuild with
/// `--features otel` and configure `TelemetryConfig::otlp_endpoint`.
///
/// # Errors
///
/// Always returns `Ok` in this configuration.
#[cfg(not(feature = "otel"))]
pub fn init_telemetry(
    config: TelemetryConfig,
) -> std::result::Result<TelemetryHandle, TelemetryInitError> {
    // Initialize provider rates: custom path takes precedence over bundled.
    if let Some(ref path) = config.rates_path {
        let _ = crate::core::rates::init_from_path(path);
    } else {
        let _ = crate::core::rates::init_bundled();
    }
    // Library-safe: kremory does NOT install a global tracing subscriber or
    // metrics recorder (ADR D2). The host binary owns subscriber installation.
    Ok(TelemetryHandle { _private: () })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn namespace_constructors_round_trip() {
        let s = Namespace::new("ws-1");
        assert_eq!(s.namespace, "ws-1");
        assert!(s.thread.is_none());

        let s = Namespace::new("ws-2").with_thread("thread-a");
        assert_eq!(s.namespace, "ws-2");
        assert_eq!(s.thread.as_deref(), Some("thread-a"));
    }

    #[test]
    fn source_kind_serde_round_trips_snake_case() {
        let v = serde_json::to_string(&SourceKind::Meeting).expect("serialize");
        assert_eq!(v, "\"meeting\"");

        let back: SourceKind = serde_json::from_str("\"document\"").expect("deserialize");
        assert_eq!(back, SourceKind::Document);
    }

    #[test]
    fn context_template_serde_round_trips_snake_case() {
        let v = serde_json::to_string(&ContextTemplate::EdgeSummary).expect("serialize");
        assert_eq!(v, "\"edge_summary\"");
    }

    use async_trait::async_trait;
    use std::sync::Mutex;

    /// Stub graph handle for D.2b delegation tests. Records the params
    /// each method was called with so we can assert the memory wrappers
    /// pass them through correctly.
    ///
    /// All new D.6.4 GraphHandle methods are implemented as required (no
    /// defaults on the trait — ADR §4.9).
    #[derive(Default)]
    struct StubGraphHandle {
        last_ingest_namespace: Mutex<Option<Namespace>>,
        last_ingest_content: Mutex<Option<String>>,
        last_ingest_source_id: Mutex<Option<String>>,
        last_ingest_facts_count: Mutex<Option<usize>>,
        last_search_namespace: Mutex<Option<Namespace>>,
        last_search_query: Mutex<Option<String>>,
        last_search_limit: Mutex<Option<usize>>,
        last_consolidation_namespace: Mutex<Option<Namespace>>,
    }

    #[async_trait]
    impl GraphHandle for StubGraphHandle {
        async fn graph_ingest_episode(
            &self,
            params: GraphIngestEpisodeParams<'_>,
        ) -> Result<EpisodeCommit> {
            let GraphIngestEpisodeParams {
                namespace,
                source_ref,
                content,
                structured_facts,
                provider: _,
                batch_id: _,
                opts: _,
                sink: _,
            } = params;
            *self.last_ingest_namespace.lock().unwrap() = Some(namespace.clone());
            *self.last_ingest_content.lock().unwrap() = Some(content.to_string());
            *self.last_ingest_source_id.lock().unwrap() = Some(source_ref.id.clone());
            *self.last_ingest_facts_count.lock().unwrap() = Some(structured_facts.len());
            Ok(EpisodeCommit {
                run_id: None,
                episode_entity_id: format!("stub:{}", source_ref.id),
                committed_at: chrono::Utc::now(),
                stub_entities_inserted: 0,
            })
        }

        async fn graph_ingest_status(
            &self,
            _run_id: uuid::Uuid,
        ) -> Result<crate::core::error::IngestStatus> {
            Ok(crate::core::error::IngestStatus::Complete)
        }

        async fn graph_cancel(&self, _run_id: uuid::Uuid) -> Result<CancelOutcome> {
            Ok(CancelOutcome {
                cancelled_phase: CancelledPhase::Enrichment,
                rolled_back: false,
                partial: vec![],
            })
        }

        async fn graph_submit_dream(
            &self,
            params: GraphSubmitDreamParams<'_>,
        ) -> Result<DreamHandle> {
            let GraphSubmitDreamParams {
                namespace,
                provider: _,
                batch_id,
                opts: _,
                sink: _,
            } = params;
            Ok(DreamHandle {
                run_id: uuid::Uuid::new_v4(),
                namespace: namespace.clone(),
                submitted_at: chrono::Utc::now(),
                batch_id,
            })
        }

        async fn graph_dream_status(&self, _run_id: uuid::Uuid) -> Result<DreamStatus> {
            Ok(DreamStatus::Complete)
        }

        async fn graph_batch_status(&self, _batch_id: &str) -> Result<BatchStatus> {
            Ok(BatchStatus {
                total: 0,
                completed: 0,
                skipped: 0,
                failed: 0,
            })
        }

        async fn graph_last_consolidated_at(
            &self,
            _namespace: &Namespace,
        ) -> Result<Option<chrono::DateTime<chrono::Utc>>> {
            Ok(None)
        }

        async fn graph_episodes_since_last_dream(&self, _namespace: &Namespace) -> Result<usize> {
            Ok(0)
        }

        async fn graph_is_consolidating(&self, _namespace: &Namespace) -> Result<bool> {
            Ok(false)
        }

        async fn graph_search(
            &self,
            params: GraphSearchParams<'_>,
        ) -> Result<Vec<RetrievedContext>> {
            let GraphSearchParams {
                namespace,
                query,
                opts,
            } = params;
            *self.last_search_namespace.lock().unwrap() = Some(namespace.clone());
            *self.last_search_query.lock().unwrap() = Some(query.to_string());
            *self.last_search_limit.lock().unwrap() = opts.limit;
            Ok(vec![RetrievedContext {
                entity_id: "ent-stub".into(),
                entity_name: "Stub Entity".into(),
                summary: "from StubGraphHandle".into(),
                score: 0.5,
                source_refs: vec![],
                incomplete: false,
                entity_type_id: 0,
                entity_type_name: "Entity".to_string(),
                namespace: None,
            }])
        }

        async fn graph_run_consolidation(
            &self,
            namespace: &Namespace,
            _provider: Arc<dyn ChatProvider>,
        ) -> Result<DreamPhaseResult> {
            *self.last_consolidation_namespace.lock().unwrap() = Some(namespace.clone());
            Ok(DreamPhaseResult {
                communities_recomputed: 1,
                cross_meeting_merges: 0,
                supersessions_recorded: 0,
                facts_archived: 0,
                duration_ms: 10,
                types_discovered: vec![],
                dream_warnings: vec![],
            })
        }

        async fn graph_run_dream_pass_sync(
            &self,
            _opts: crate::core::ingest::DreamPassOpts,
        ) -> Result<crate::facade::DreamSummary> {
            Ok(crate::facade::DreamSummary {
                communities_updated: 0,
                cross_episode_merges: 0,
                supersessions_recorded: 0,
                facts_archived: 0,
                duration_ms: 0,
                types_discovered: vec![],
                entities_reclassified: 0,
                warnings: vec![],
            })
        }

        async fn graph_ghost_episodes(&self, _group_id: Option<&str>) -> Result<Vec<i64>> {
            Ok(vec![])
        }

        async fn graph_assert_entity_type(
            &self,
            _params: GraphAssertEntityTypeParams<'_>,
        ) -> Result<()> {
            Ok(())
        }
    }

    /// kremory::core's `MockChatProvider::null()` implements the canonical
    /// `autoagents_llm::chat::ChatProvider` trait — gated behind kremory's
    /// `llm` feature. The stub graph never invokes the provider so the canned-
    /// response behaviour is irrelevant; we just need an `Arc<dyn ChatProvider>`.
    fn null_provider() -> Arc<dyn ChatProvider> {
        Arc::new(crate::core::provider::MockChatProvider::null())
    }

    #[tokio::test]
    async fn submit_episode_delegates_to_graph_handle() {
        use chrono::Utc;
        let graph = StubGraphHandle::default();
        let namespace = Namespace::new("ws-1").with_thread("thread-a");
        let source_ref = SourceRef {
            kind: SourceKind::Meeting,
            id: "mtg-42".into(),
            occurred_at: Utc::now(),
            published_at: None,
        };
        let facts = vec![StructuredFact {
            subject: "alice".into(),
            predicate: "leads".into(),
            object: "design".into(),
            valid_from: None,
            valid_to: None,
            memory_type: None,
        }];

        let commit = submit_episode(SubmitEpisodeParams {
            graph: &graph,
            content: "transcript content",
            source_ref,
            structured_facts: facts,
            provider: null_provider(),
            namespace: namespace.clone(),
            batch_id: None,
            opts: SubmitOpts::default(),
            sink: None,
        })
        .await
        .expect("submit_episode should succeed via stub");

        assert!(!commit.episode_entity_id.is_empty());
        assert_eq!(
            graph.last_ingest_namespace.lock().unwrap().as_ref(),
            Some(&namespace)
        );
        assert_eq!(
            graph.last_ingest_content.lock().unwrap().as_deref(),
            Some("transcript content")
        );
        assert_eq!(
            graph.last_ingest_source_id.lock().unwrap().as_deref(),
            Some("mtg-42")
        );
        assert_eq!(*graph.last_ingest_facts_count.lock().unwrap(), Some(1));
    }

    #[tokio::test]
    async fn search_delegates_with_opts() {
        let graph = StubGraphHandle::default();
        let namespace = Namespace::new("ws-2");
        let opts = SearchOpts {
            limit: Some(25),
            as_of: None,
            source_kind: Some(SourceKind::Document),
        };

        let hits = search(SearchParams {
            graph: &graph,
            query: "go-live",
            namespace: namespace.clone(),
            opts,
        })
        .await
        .expect("search delegates cleanly");

        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].entity_id, "ent-stub");
        assert_eq!(
            graph.last_search_namespace.lock().unwrap().as_ref(),
            Some(&namespace)
        );
        assert_eq!(
            graph.last_search_query.lock().unwrap().as_deref(),
            Some("go-live")
        );
        assert_eq!(*graph.last_search_limit.lock().unwrap(), Some(25));
    }

    #[tokio::test]
    #[allow(deprecated)]
    async fn run_dream_phase_delegates_to_consolidation() {
        let graph = StubGraphHandle::default();
        let namespace = Namespace::new("ws-3");

        let result = run_dream_phase(&graph, namespace.clone(), null_provider())
            .await
            .expect("run_dream_phase delegates cleanly");

        assert_eq!(result.communities_recomputed, 1);
        assert_eq!(
            graph.last_consolidation_namespace.lock().unwrap().as_ref(),
            Some(&namespace)
        );
    }

    /// Story #318 AC: SourceRef.published_at propagates through ingest path;
    /// StructuredFact.valid_from/valid_to field names compile correctly.
    #[tokio::test]
    async fn published_at_precedence_fields_compile() {
        use chrono::Utc;
        let now = Utc::now();
        let source_ref = SourceRef {
            kind: SourceKind::Document,
            id: "doc-1".into(),
            occurred_at: now,
            published_at: Some(now),
        };
        // Verify published_at is round-trippable
        assert_eq!(source_ref.published_at, Some(now));

        let sf = StructuredFact {
            subject: "alpha".into(),
            predicate: "knows".into(),
            object: "beta".into(),
            valid_from: Some(now),
            valid_to: None,
            memory_type: None,
        };
        // valid_from fallback precedence per Story #318:
        // fact.valid_from = sf.valid_from.or(source_ref.published_at).unwrap_or_else(Utc::now)
        let resolved_from = sf
            .valid_from
            .or(source_ref.published_at)
            .unwrap_or_else(Utc::now);
        assert_eq!(resolved_from, now);

        // When sf.valid_from is None, published_at is used as fallback
        let sf_no_valid_from = StructuredFact {
            subject: "alpha".into(),
            predicate: "knows".into(),
            object: "beta".into(),
            valid_from: None,
            valid_to: None,
            memory_type: None,
        };
        let resolved_fallback = sf_no_valid_from
            .valid_from
            .or(source_ref.published_at)
            .unwrap_or_else(Utc::now);
        assert_eq!(resolved_fallback, now);
    }

    /// F4 (ADR adr-memory-builder-as-of-fail-loud): search() FAILS LOUD when
    /// opts.as_of is Some — point-in-time recall is declared-but-unimplemented,
    /// so a silent no-op (the old Story #318 warn-only behaviour) is replaced by
    /// `Err(Error::Unsupported)`. Rewritten (not deleted) from
    /// `as_of_emits_v010_warn` per treat-cause-not-symptom: the test now asserts
    /// the corrected behaviour rather than locking the footgun.
    #[tokio::test]
    async fn as_of_errors_unsupported() {
        use chrono::Utc;
        let graph = StubGraphHandle::default();
        let namespace = Namespace::new("ws-as-of");
        let opts = SearchOpts {
            limit: None,
            as_of: Some(Utc::now()),
            source_kind: None,
        };
        let result = search(SearchParams {
            graph: &graph,
            query: "test",
            namespace,
            opts,
        })
        .await;
        match result {
            Err(MemoryError::Core(crate::core::error::Error::Unsupported { feature })) => {
                assert_eq!(feature, "as_of point-in-time recall");
            }
            other => panic!("as_of must fail loud with Error::Unsupported, got: {other:?}"),
        }
    }
}
