crtx-session 0.1.0

//! `cortex-session` — session-close pipeline extracted from `cortex-cli`.
#![warn(missing_docs)]
//!
//! ## Why this crate exists
//!
//! `cortex-mcp` cannot depend on `cortex-cli` (that would be circular:
//! `cortex-cli` will depend on `cortex-mcp` for the `Serve` subcommand).
//! This crate exposes [`close_from_bytes`] so both `cortex-cli` and
//! `cortex-mcp` can run the session-close pipeline.
//!
//! ## Pipeline
//!
//! 1. **Parse** — parse `raw` as session events JSON (envelope `{"events":[...]}`
//!    or bare array `[...]` or single object).
//! 2. **Ingest** — append non-duplicate events to the JSONL ledger at
//!    `event_log`. User-sourced events are refused (no attestor available in
//!    the library context; operator must call CLI `cortex ingest` for those).
//! 3. **Extract trace** — extract the first `trace_id` from the parsed events.
//!    If none, returns `CloseOutcome { pending_commit: 0, no_candidates: true }`.
//! 4. **Reflect** — run `cortex_reflect::reflect()` over the trace via a
//!    `ReplayAdapter` built from `fixtures_dir` (see [`close_from_bytes`]).
//! 5. **Tag as pending** — for each reflected candidate, set lifecycle state
//!    to `pending_mcp_commit` (ADR 0047). The caller (MCP server) promotes to
//!    `active` on `cortex_session_commit`; the CLI wrapper calls
//!    [`MemoryRepo::set_active`] itself after receiving the `CloseOutcome`.
//! 6. **Pre-compute embeddings** — BLAKE3 stub embeddings via
//!    [`LocalStubEmbedder`], written to the store so semantic search is
//!    immediately warm after promotion.
//! 7. **Return `CloseOutcome`** — `pending_commit` is the count of memories
//!    in `pending_mcp_commit` state; callers decide whether to promote them.

use std::fmt;
use std::path::PathBuf;

use chrono::Utc;
use cortex_core::{
    compose_policy_outcomes, Event, EventSource, PolicyContribution, PolicyOutcome, TraceId,
};
use cortex_ledger::{
    JsonlError, JsonlLog, APPEND_ATTESTATION_REQUIRED_RULE_ID,
    APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID, APPEND_RUNTIME_MODE_RULE_ID,
};
use cortex_reflect::ReflectionReportStatus;
use cortex_retrieval::{EmbedRecord, Embedder, LocalStubEmbedder, STUB_BACKEND_ID};
use cortex_store::repo::{EmbeddingRepo, EventRepo, MemoryRepo};
use cortex_store::Pool;

/// Outcome returned from [`close_from_bytes`].
#[derive(Debug, Default)]
pub struct CloseOutcome {
    /// Number of events ingested to the JSONL ledger.
    pub ingested: usize,
    /// Number of memory candidates produced by reflection.
    pub reflected: usize,
    /// Number of memories written with `pending_mcp_commit` lifecycle state
    /// (ADR 0047). Callers promote these to `active` via
    /// `cortex_session_commit` (MCP path) or directly via
    /// `MemoryRepo::set_active` (CLI path).
    pub pending_commit: usize,
    /// Stable receipt identifier for this close operation (trace_id or a
    /// synthetic marker when no trace was present).
    pub receipt_id: String,
}

/// Errors raised by the session-close pipeline.
#[derive(Debug)]
pub enum SessionError {
    /// Ingest step failed (parse error or ledger I/O).
    Ingest(String),
    /// Reflection step failed.
    Reflect(String),
    /// Store operation failed.
    Store(cortex_store::StoreError),
    /// I/O error outside of the ledger path.
    Io(std::io::Error),
}

impl fmt::Display for SessionError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Ingest(msg) => write!(f, "session ingest failed: {msg}"),
            Self::Reflect(msg) => write!(f, "session reflect failed: {msg}"),
            Self::Store(err) => write!(f, "session store error: {err}"),
            Self::Io(err) => write!(f, "session io error: {err}"),
        }
    }
}

impl std::error::Error for SessionError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Store(err) => Some(err),
            Self::Io(err) => Some(err),
            _ => None,
        }
    }
}

impl From<cortex_store::StoreError> for SessionError {
    fn from(err: cortex_store::StoreError) -> Self {
        Self::Store(err)
    }
}

impl From<std::io::Error> for SessionError {
    fn from(err: std::io::Error) -> Self {
        Self::Io(err)
    }
}

impl From<JsonlError> for SessionError {
    fn from(err: JsonlError) -> Self {
        Self::Ingest(err.to_string())
    }
}

/// Run the full session-close pipeline.
///
/// # Arguments
///
/// - `raw` — raw bytes of the session events JSON file.
/// - `pool` — open SQLite connection pool with all migrations applied.
/// - `event_log` — path to the JSONL ledger file. Created if absent.
/// - `fixtures_dir` — directory containing `INDEX.toml` and replay fixtures
///   for `cortex_reflect::reflect()`. Must be supplied even if reflection is
///   expected to produce no candidates (the adapter must initialise).
///
/// # Lifecycle state
///
/// Reflected candidates are placed in `pending_mcp_commit` status, NOT
/// `active`. MCP callers leave them there until `cortex_session_commit` is
/// called with the operator's confirmation token. CLI callers may call
/// `MemoryRepo::set_active` on each `memory_id` in `CloseOutcome` directly
/// after this function returns, since the CLI close is operator-initiated and
/// synchronous (ADR 0047 §2 Alternatives).
pub fn close_from_bytes(
    raw: &[u8],
    pool: &Pool,
    event_log: PathBuf,
    fixtures_dir: &std::path::Path,
) -> Result<CloseOutcome, SessionError> {
    // ─── Step 1: parse events ────────────────────────────────────────────────
    let events = parse_events(raw).map_err(SessionError::Ingest)?;

    // Extract trace_id before ingest (needed for reflect step).
    let trace_id = extract_trace_id(&events);

    // ─── Step 2: ingest events to JSONL ledger + SQLite events table ─────────
    // Dual-write: append to JSONL first (existing path), then to the SQLite
    // `events` table so `verify_memory_proof_closure` can find the source events
    // and session-close memories are not quarantined in search results.
    let ingested = ingest_events(&events, &event_log, pool)?;

    let trace_id_for_reflect = match trace_id {
        Some(tid) => tid,
        None => {
            tracing::debug!("session-close: no trace_id in events; no candidates proposed");
            return Ok(CloseOutcome {
                ingested,
                reflected: 0,
                pending_commit: 0,
                receipt_id: "no_trace_id".to_string(),
            });
        }
    };

    // ─── Step 3: reflect ─────────────────────────────────────────────────────
    let reflect_report =
        run_reflect(trace_id_for_reflect, fixtures_dir, pool).map_err(SessionError::Reflect)?;

    if reflect_report.status == ReflectionReportStatus::Quarantined {
        tracing::debug!("session-close: reflection quarantined; no candidates proposed");
        return Ok(CloseOutcome {
            ingested,
            reflected: 0,
            pending_commit: 0,
            receipt_id: trace_id_for_reflect.to_string(),
        });
    }

    let candidate_ids: Vec<cortex_core::MemoryId> = reflect_report
        .persisted_memory_candidates
        .iter()
        .map(|c| c.id)
        .collect();
    let reflected = candidate_ids.len();

    if reflected == 0 {
        tracing::debug!("session-close: no candidates proposed");
        return Ok(CloseOutcome {
            ingested,
            reflected: 0,
            pending_commit: 0,
            receipt_id: trace_id_for_reflect.to_string(),
        });
    }

    // ─── Step 4: tag candidates as pending_mcp_commit (ADR 0047) ─────────────
    // `reflect()` already called `MemoryRepo::insert_candidate` for each
    // candidate (see `persist_memory_candidates` in orchestrate.rs). The
    // candidates are already in the store as `candidate`; we transition them
    // to `pending_mcp_commit` here instead of `active`.
    let repo = MemoryRepo::new(pool);
    let now = Utc::now();
    let mut pending_ids = Vec::new();

    for memory_id in &candidate_ids {
        match repo.set_pending_mcp_commit(memory_id, now) {
            Ok(()) => {
                pending_ids.push(*memory_id);
            }
            Err(err) => {
                let err_str = err.to_string();
                // If already in pending_mcp_commit, treat as idempotent.
                if err_str.contains("not a candidate") {
                    tracing::debug!(
                        memory_id = %memory_id,
                        "session-close: memory not a candidate (already transitioned); treating as pending"
                    );
                    pending_ids.push(*memory_id);
                } else {
                    tracing::warn!(
                        memory_id = %memory_id,
                        error = %err_str,
                        "session-close: failed to set pending_mcp_commit for memory"
                    );
                }
            }
        }
    }

    // ─── Step 5: pre-compute embeddings ──────────────────────────────────────
    let embed_repo = EmbeddingRepo::new(pool);
    let embedder = LocalStubEmbedder::new();

    for memory_id in &pending_ids {
        let memory = match repo.get_by_id(memory_id) {
            Ok(Some(m)) => m,
            Ok(None) => {
                tracing::warn!(memory_id = %memory_id, "session-close: memory not found for embedding");
                continue;
            }
            Err(err) => {
                tracing::warn!(memory_id = %memory_id, error = %err, "session-close: failed to read memory for embedding");
                continue;
            }
        };

        let tags: Vec<String> = memory
            .domains_json
            .as_array()
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| v.as_str().map(ToString::to_string))
                    .collect()
            })
            .unwrap_or_default();

        let vec = match embedder.embed(&memory.claim, &tags) {
            Ok(v) => v,
            Err(err) => {
                tracing::warn!(memory_id = %memory_id, error = %err, "session-close: embed failed");
                continue;
            }
        };

        let record = match EmbedRecord::new(*memory_id, STUB_BACKEND_ID, vec, now) {
            Ok(r) => r,
            Err(err) => {
                tracing::warn!(memory_id = %memory_id, error = %err, "session-close: failed to build embed record");
                continue;
            }
        };

        if let Err(err) = embed_repo.write(&record) {
            tracing::warn!(memory_id = %memory_id, error = %err, "session-close: failed to write embedding");
        }
    }

    Ok(CloseOutcome {
        ingested,
        reflected,
        pending_commit: pending_ids.len(),
        receipt_id: trace_id_for_reflect.to_string(),
    })
}

/// Parse raw bytes as session events JSON.
///
/// Accepts three shapes:
/// 1. `{"events": [Event, ...]}` — explicit envelope.
/// 2. `[Event, ...]` — bare array.
/// 3. `{...}` — single event object.
fn parse_events(raw: &[u8]) -> Result<Vec<Event>, String> {
    let value: serde_json::Value =
        serde_json::from_slice(raw).map_err(|err| format!("invalid JSON: {err}"))?;

    let events: Vec<serde_json::Value> = match &value {
        serde_json::Value::Object(map) => {
            if let Some(events_field) = map.get("events") {
                events_field
                    .as_array()
                    .ok_or_else(|| "events field is not an array".to_string())?
                    .to_owned()
            } else {
                vec![value.clone()]
            }
        }
        serde_json::Value::Array(arr) => arr.to_owned(),
        _ => return Err("unexpected JSON shape: must be object or array".to_string()),
    };

    let mut parsed = Vec::with_capacity(events.len());
    for (i, ev) in events.iter().enumerate() {
        let event: Event = serde_json::from_value(ev.clone())
            .map_err(|err| format!("event[{i}] failed to deserialize: {err}"))?;
        parsed.push(event);
    }
    Ok(parsed)
}

/// Extract the first `trace_id` from a parsed event list.
fn extract_trace_id(events: &[Event]) -> Option<TraceId> {
    events.iter().find_map(|ev| ev.trace_id)
}

/// Append events to the JSONL ledger and the SQLite `events` table, skipping
/// duplicates.
///
/// Refuses `EventSource::User` events — the library has no attestor; operator
/// must use `cortex ingest --user-attestation` for those.
///
/// The dual-write to SQLite ensures that `verify_memory_proof_closure` can
/// find source events for session-close memories so they are not quarantined
/// in `cortex memory search`. `EventRepo::append` uses `INSERT OR IGNORE` so
/// re-running is idempotent.
fn ingest_events(
    events: &[Event],
    event_log: &std::path::Path,
    pool: &Pool,
) -> Result<usize, SessionError> {
    // Refuse User-sourced events: no attestor available in the library context.
    for event in events {
        if matches!(event.source, EventSource::User) {
            return Err(SessionError::Ingest(format!(
                "EventSource::User event {} cannot be ingested without operator attestation; \
                 use `cortex ingest --user-attestation` for user-sourced events",
                event.id,
            )));
        }
    }

    // Ensure parent directory exists.
    if let Some(parent) = event_log.parent() {
        if !parent.as_os_str().is_empty() {
            std::fs::create_dir_all(parent)?;
        }
    }

    let mut log = JsonlLog::open(event_log)?;

    // Collect existing event hashes for deduplication. We read the log and
    // compare event ids by reading the existing lines. JsonlLog doesn't
    // expose an iterator, but the ingest path in cortex-cli collects ids
    // from the JSONL lines. We scan the file directly here.
    let existing_ids = collect_existing_event_ids(event_log)?;

    let event_repo = EventRepo::new(pool);
    let mut appended = 0usize;
    for event in events {
        if existing_ids.contains(&event.id) {
            // Also ensure the SQLite row exists for pre-existing JSONL events
            // (idempotent — INSERT OR IGNORE).
            event_repo.append(event).map_err(SessionError::Store)?;
            continue;
        }

        // Compose the minimal policy for a non-user, non-signed append.
        // All three required contributors must be present (ADR 0026).
        let policy = session_append_policy(&event.source);
        log.append(event.clone(), &policy)?;

        // Dual-write to SQLite so proof-closure verification succeeds.
        event_repo.append(event).map_err(SessionError::Store)?;

        appended += 1;
    }

    Ok(appended)
}

/// Collect event ids that already exist in the JSONL log file.
fn collect_existing_event_ids(
    path: &std::path::Path,
) -> Result<std::collections::HashSet<cortex_core::EventId>, SessionError> {
    use std::io::BufRead;

    if !path.exists() {
        return Ok(std::collections::HashSet::new());
    }

    let file = std::fs::File::open(path)?;
    let reader = std::io::BufReader::new(file);
    let mut ids = std::collections::HashSet::new();

    for line in reader.lines() {
        let line = line?;
        if line.trim().is_empty() {
            continue;
        }
        // Each line is a SignedRow JSON — we only need the event.id field.
        if let Ok(value) = serde_json::from_str::<serde_json::Value>(&line) {
            if let Some(id_str) = value
                .get("event")
                .and_then(|e| e.get("id"))
                .and_then(|v| v.as_str())
            {
                if let Ok(event_id) = id_str.parse::<cortex_core::EventId>() {
                    ids.insert(event_id);
                }
            }
        }
    }

    Ok(ids)
}

/// Compose the ADR 0026 policy decision for a non-user session append.
///
/// Non-user events do not require attestation, so the attestation contributor
/// is `Warn` (honest no-attestation floor). The tier gate is `Allow` for
/// non-user sources. Runtime mode is `Warn` (local-development unsigned ledger
/// path).
fn session_append_policy(source: &EventSource) -> cortex_core::PolicyDecision {
    // Tier gate: `ingest_events` already refuses `EventSource::User` before
    // calling this function, so all remaining sources are at least `Observed`.
    // Classify the source string for the policy record without importing the
    // `event_source_trust_tier` helper from `cortex-cli` (that would be a
    // circular dependency). ManualCorrection without an attestor is also
    // refused upstream; every source that reaches here passes the floor.
    let (tier_outcome, tier_reason): (PolicyOutcome, &str) = match source {
        EventSource::User | EventSource::ManualCorrection => {
            // Should be unreachable — ingest_events refuses these — but fail
            // closed defensively rather than panic.
            (
                PolicyOutcome::Reject,
                "event source User/ManualCorrection requires operator attestation; refused by ingest_events",
            )
        }
        EventSource::ChildAgent { .. }
        | EventSource::Tool { .. }
        | EventSource::Runtime
        | EventSource::ExternalOutcome => (
            PolicyOutcome::Allow,
            "event source meets ingest floor of Observed or above",
        ),
    };
    let tier_contribution = PolicyContribution::new(
        APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID,
        tier_outcome,
        tier_reason,
    )
    .expect("static session append tier contribution is valid");

    // Attestation: no User events reach this path, so Warn (honest floor).
    let attestation_contribution = PolicyContribution::new(
        APPEND_ATTESTATION_REQUIRED_RULE_ID,
        PolicyOutcome::Warn,
        "session-close ingest: no User events in batch; operator attestation not required",
    )
    .expect("static session append attestation contribution is valid");

    // Runtime mode: local-development / unsigned-ledger path.
    let runtime_mode_contribution = PolicyContribution::new(
        APPEND_RUNTIME_MODE_RULE_ID,
        PolicyOutcome::Warn,
        "session-close ingest: unsigned append (local-development ledger path)",
    )
    .expect("static session append runtime mode contribution is valid");

    compose_policy_outcomes(
        vec![
            tier_contribution,
            attestation_contribution,
            runtime_mode_contribution,
        ],
        None,
    )
}

/// Run the `cortex_reflect::reflect` pipeline.
fn run_reflect(
    trace_id: TraceId,
    fixtures_dir: &std::path::Path,
    pool: &Pool,
) -> Result<cortex_reflect::ReflectionReport, String> {
    use cortex_llm::ReplayAdapter;

    let adapter = ReplayAdapter::new(fixtures_dir).map_err(|err| format!("{err}"))?;

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|err| format!("failed to create tokio runtime: {err}"))?;

    rt.block_on(cortex_reflect::reflect(trace_id, &adapter, pool))
        .map_err(|err| format!("{err}"))
}