post-cortex-memory 0.5.0

// Copyright (c) 2025, 2026 Julius ML
// Licensed under the MIT License. See LICENSE at the workspace root.

//! Canonical [`PostCortexService`] implementation for post-cortex-memory.
//!
//! [`MemoryServiceImpl`] is a thin adapter over
//! [`ConversationMemorySystem`] that fulfils the
//! [`post_cortex_core::services::PostCortexService`] trait. Phases 6 + 7
//! migrate the daemon's gRPC handlers and the MCP tool layer to call
//! this trait instead of `ConversationMemorySystem` directly — at that
//! point every read / write / search / manage operation flows through
//! one canonical function (TODO.md:106-117).
//!
//! Right now this impl is **incomplete on purpose**: most methods return
//! a `SystemError::Internal("not yet wired …")` while the legacy
//! transport handlers continue to call `ConversationMemorySystem` paths
//! directly. The trait surface is stable; the impl fills in
//! incrementally as Phases 6 + 7 migrate their callers. This avoids a
//! single massive commit that's impossible to bisect.

use std::collections::HashSet;
use std::sync::Arc;

use async_trait::async_trait;
use chrono::Utc;
use post_cortex_core::core::card::{CardScope, MAX_CARD_BYTES};
use post_cortex_core::core::context_update::{ContextUpdate, EntityRelationship, TypedEntity};
use post_cortex_core::core::error::SystemError;
use post_cortex_core::core::message;
use post_cortex_core::redact::{RedactionCounts, Redactor};
use post_cortex_core::services::{
    AdminRequest, AdminResponse, AppendMessagesRequest, AppendMessagesResponse,
    AssembleContextRequest, AssembleContextResponse, BulkUpdateContextRequest,
    BulkUpdateContextResponse, GetCardRequest, GetCardResponse, HealthReport, ManageEntityRequest,
    ManageEntityResponse, ManageSessionRequest, ManageSessionResponse, ManageWorkspaceRequest,
    ManageWorkspaceResponse, PostCortexService, QueryContextRequest, QueryContextResponse,
    SemanticSearchRequest, SemanticSearchResponse, StructuredSummaryRequest,
    StructuredSummaryResponse, UpdateCardRequest, UpdateCardResponse, UpdateContextRequest,
    UpdateContextResponse,
};
use tracing::warn;
use uuid::Uuid;

use post_cortex_storage::traits::CardWrite;

use crate::memory_system::ConversationMemorySystem;
use crate::pipeline::{
    EmbeddingWorkItem, GraphWorkItem, Pipeline, PipelineConfig, PipelineError, SummaryWorkItem,
};

/// Canonical [`PostCortexService`] implementation backed by
/// [`ConversationMemorySystem`].
///
/// Construct with [`Self::new`] passing an `Arc<ConversationMemorySystem>`
/// (typically the singleton owned by the daemon).
pub struct MemoryServiceImpl {
    system: Arc<ConversationMemorySystem>,
    pipeline: Arc<Pipeline>,
    redactor: Arc<Redactor>,
}

impl MemoryServiceImpl {
    /// Wrap an existing memory system in the canonical service trait,
    /// starting the non-blocking write pipeline with default capacities.
    /// Must be called from inside a Tokio runtime.
    #[must_use]
    pub fn new(system: Arc<ConversationMemorySystem>) -> Self {
        Self::with_pipeline_config(system, PipelineConfig::default())
    }

    /// Wrap an existing memory system in the service trait with an
    /// explicit pipeline configuration.
    #[must_use]
    pub fn with_pipeline_config(
        system: Arc<ConversationMemorySystem>,
        config: PipelineConfig,
    ) -> Self {
        let pipeline = Arc::new(Pipeline::start(config, Arc::clone(&system)));
        let (redactor, invalid) = Redactor::new(&system.config.redaction);
        for p in invalid {
            warn!("redaction: skipping invalid extra pattern {p}");
        }
        Self {
            system,
            pipeline,
            redactor: Arc::new(redactor),
        }
    }

    /// Borrow the non-blocking pipeline. Exposed so transport adapters
    /// can submit derived-work items directly (e.g. enqueue an
    /// embedding compute after a manual storage write).
    #[must_use]
    pub fn pipeline(&self) -> &Arc<Pipeline> {
        &self.pipeline
    }

    /// Borrow the underlying memory system. Provided so the daemon's
    /// gRPC handlers can fall back to direct calls for methods that
    /// have not been migrated to the trait yet.
    #[must_use]
    pub fn inner(&self) -> &Arc<ConversationMemorySystem> {
        &self.system
    }

    /// Every `derived_from` entry names a captured message of the session
    /// (`message:<seq>`, no chunk suffix).
    async fn check_derived_from(&self, req: &UpdateContextRequest) -> Result<(), SystemError> {
        if req.derived_from.is_empty() {
            return Ok(());
        }
        let mut seqs = Vec::with_capacity(req.derived_from.len());
        for id in &req.derived_from {
            match message::seq_of_content_id(id) {
                Some(seq) if !id.contains('#') => seqs.push(seq),
                _ => {
                    return Err(SystemError::InvalidArgument(format!(
                        "derived_from: '{id}' is not a message content id (message:<n>)"
                    )));
                }
            }
        }
        let (first, last) = (
            seqs.iter().min().copied().unwrap_or(0),
            seqs.iter().max().copied().unwrap_or(0),
        );
        let stored: HashSet<u64> = self
            .system
            .storage_actor
            .load_messages(req.session_id, first..last + 1)
            .await
            .map_err(SystemError::Internal)?
            .into_iter()
            .map(|m| m.seq)
            .collect();
        match seqs.iter().find(|seq| !stored.contains(seq)) {
            Some(missing) => Err(SystemError::InvalidArgument(format!(
                "derived_from: message:{missing} is not in session {}",
                req.session_id
            ))),
            None => Ok(()),
        }
    }

    /// A workspace card needs an existing workspace.
    fn require_card_scope(&self, scope: CardScope) -> Result<(), SystemError> {
        match scope {
            CardScope::Workspace(id)
                if self.system.workspace_manager.get_workspace(&id).is_none() =>
            {
                Err(SystemError::WorkspaceNotFound(id))
            }
            _ => Ok(()),
        }
    }

    fn not_yet_wired<T>(op: &'static str) -> Result<T, SystemError> {
        Err(SystemError::Internal(format!(
            "PostCortexService::{op} is not yet wired — migration lands in Phase 6 (MCP) / Phase 7 (daemon). \
             Use ConversationMemorySystem directly until then."
        )))
    }
}

#[async_trait]
impl PostCortexService for MemoryServiceImpl {
    #[tracing::instrument(skip(self), name = "post_cortex.health")]
    async fn health(&self) -> Result<HealthReport, SystemError> {
        let health = self.system.get_system_health();
        Ok(HealthReport {
            status: if health.circuit_breaker_open {
                "degraded".to_string()
            } else {
                "ok".to_string()
            },
            active_sessions: health.active_sessions,
            // SystemHealth does not yet expose memory bytes — wired in
            // Phase 11 (perf observability). For now report 0.
            memory_usage_bytes: 0,
            pipeline_backlog: self.pipeline.backlog(),
            uptime_seconds: health.uptime_seconds,
        })
    }

    #[tracing::instrument(
        skip(self, req),
        fields(
            session_id = %req.session_id,
            interaction_type = ?req.interaction_type,
            entities = req.entities.len(),
            relations = req.relations.len(),
        ),
        name = "post_cortex.update_context",
    )]
    async fn update_context(
        &self,
        mut req: UpdateContextRequest,
    ) -> Result<UpdateContextResponse, SystemError> {
        log_redactions(req.session_id, &self.redactor.redact_request(&mut req));
        validate_update_request(&req)?;
        self.check_derived_from(&req).await?;

        let description = build_description(&req);
        let context_update = build_context_update(&req);
        let metadata =
            serde_json::to_value(&context_update).expect("ContextUpdate serialization cannot fail");
        let session_id = req.session_id;

        // Waits for the storage write (blob + update row), so `durable`
        // reports what actually happened.
        let committed = self
            .system
            .add_incremental_update_deferred(session_id, description.clone(), Some(metadata), true)
            .await
            .map_err(SystemError::Internal)?;

        // Graph merge and vectorization run on the pipeline only (the
        // deferred write does neither), so each happens exactly once.
        submit_derived_work(
            &self.pipeline,
            &self.system,
            session_id,
            committed.entry_id,
            &description,
            committed.update,
        );

        Ok(UpdateContextResponse {
            entry_id: committed.entry_id,
            session_id,
            persisted_at: Utc::now(),
            durable: committed.durable,
        })
    }

    #[tracing::instrument(
        skip(self, req),
        fields(session_id = %req.session_id, batch_size = req.messages.len()),
        name = "post_cortex.append_messages",
    )]
    async fn append_messages(
        &self,
        mut req: AppendMessagesRequest,
    ) -> Result<AppendMessagesResponse, SystemError> {
        let submitted = req.messages.len();
        for (i, message) in req.messages.iter().enumerate() {
            if message.turn_id.trim().is_empty() {
                return Err(SystemError::InvalidArgument(format!(
                    "append_messages: messages[{i}].turn_id is empty"
                )));
            }
        }
        if self
            .system
            .session_manager
            .get_session(req.session_id)
            .await
            .map_err(SystemError::Internal)?
            .is_none()
        {
            return Err(SystemError::SessionNotFound(req.session_id));
        }
        req.messages.retain(|m| !m.text.trim().is_empty());
        let mut counts = RedactionCounts::new();
        for message in &mut req.messages {
            self.redactor.redact_message(message, &mut counts);
        }
        log_redactions(req.session_id, &counts);

        let (stored, vectorized) = self
            .system
            .append_messages(req.session_id, req.messages)
            .await
            .map_err(SystemError::Internal)?;
        Ok(AppendMessagesResponse {
            skipped: submitted - stored.len(),
            appended: stored.iter().map(|m| m.seq).collect(),
            vectorized,
        })
    }

    #[tracing::instrument(skip(self, req), fields(scope = %req.scope.key()), name = "post_cortex.get_card")]
    async fn get_card(&self, req: GetCardRequest) -> Result<GetCardResponse, SystemError> {
        self.require_card_scope(req.scope)?;
        let actor = &self.system.storage_actor;
        if req.history == 0 {
            return Ok(GetCardResponse {
                card: actor
                    .get_card(req.scope)
                    .await
                    .map_err(SystemError::Internal)?,
                history: Vec::new(),
            });
        }
        let mut versions = actor
            .card_history(req.scope, req.history + 1)
            .await
            .map_err(SystemError::Internal)?
            .into_iter();
        Ok(GetCardResponse {
            card: versions.next(),
            history: versions.collect(),
        })
    }

    #[tracing::instrument(
        skip(self, req),
        fields(scope = %req.scope.key(), expected_version = req.expected_version),
        name = "post_cortex.update_card",
    )]
    async fn update_card(&self, req: UpdateCardRequest) -> Result<UpdateCardResponse, SystemError> {
        self.require_card_scope(req.scope)?;
        let mut counts = RedactionCounts::new();
        let text = self.redactor.redact_str(req.text.trim_end(), &mut counts);
        if !counts.is_empty() {
            tracing::info!(scope = %req.scope.key(), ?counts, "update_card: redacted sensitive values");
        }
        if text.len() > MAX_CARD_BYTES {
            return Err(SystemError::InvalidArgument(format!(
                "update_card: the card is {} bytes, the limit is {MAX_CARD_BYTES}; shorten it",
                text.len()
            )));
        }
        let write = self
            .system
            .storage_actor
            .put_card(req.scope, text, req.author, req.expected_version)
            .await
            .map_err(SystemError::Internal)?;
        Ok(match write {
            CardWrite::Written(card) => UpdateCardResponse {
                written: true,
                card: Some(card),
            },
            CardWrite::Conflict(latest) => UpdateCardResponse {
                written: false,
                card: latest,
            },
        })
    }

    #[tracing::instrument(
        skip(self, req),
        fields(session_id = %req.session_id, batch_size = req.updates.len()),
        name = "post_cortex.bulk_update_context",
    )]
    async fn bulk_update_context(
        &self,
        mut req: BulkUpdateContextRequest,
    ) -> Result<BulkUpdateContextResponse, SystemError> {
        for item in &mut req.updates {
            log_redactions(item.session_id, &self.redactor.redact_request(item));
        }
        // Pre-validate the whole batch first so a bad item doesn't leave us
        // with a half-applied write (storage-actor batching lands later;
        // until then we still fail-fast on validation but persist
        // sequentially).
        for (i, item) in req.updates.iter().enumerate() {
            if item.session_id != req.session_id {
                return Err(SystemError::InvalidArgument(format!(
                    "bulk_update_context: updates[{i}].session_id {} does not match request session_id {}",
                    item.session_id, req.session_id
                )));
            }
            let prefixed = |e| match e {
                SystemError::InvalidArgument(msg) => {
                    SystemError::InvalidArgument(format!("updates[{i}]: {msg}"))
                }
                other => other,
            };
            validate_update_request(item).map_err(prefixed)?;
            self.check_derived_from(item).await.map_err(prefixed)?;
        }

        let mut entry_ids = Vec::with_capacity(req.updates.len());
        let mut durable = false;
        let last = req.updates.len().saturating_sub(1);
        for (i, item) in req.updates.iter().enumerate() {
            let description = build_description(item);
            let context_update = build_context_update(item);
            let metadata = serde_json::to_value(&context_update)
                .expect("ContextUpdate serialization cannot fail");
            // Only the last item waits: session writes are ordered and
            // coalesced, so the write it waits for covers every row before it.
            let committed = self
                .system
                .add_incremental_update_deferred(
                    item.session_id,
                    description.clone(),
                    Some(metadata),
                    i == last,
                )
                .await
                .map_err(|e| SystemError::Internal(format!("updates[{i}]: {e}")))?;
            durable = committed.durable;
            entry_ids.push(committed.entry_id);
            submit_derived_work(
                &self.pipeline,
                &self.system,
                item.session_id,
                committed.entry_id,
                &description,
                committed.update,
            );
        }

        Ok(BulkUpdateContextResponse {
            entry_ids,
            persisted_at: Utc::now(),
            durable,
        })
    }

    #[tracing::instrument(skip(self, _req), name = "post_cortex.semantic_search")]
    async fn semantic_search(
        &self,
        _req: SemanticSearchRequest,
    ) -> Result<SemanticSearchResponse, SystemError> {
        Self::not_yet_wired("semantic_search")
    }

    #[tracing::instrument(skip(self, _req), name = "post_cortex.query_context")]
    async fn query_context(
        &self,
        _req: QueryContextRequest,
    ) -> Result<QueryContextResponse, SystemError> {
        Self::not_yet_wired("query_context")
    }

    #[tracing::instrument(skip(self, _req), name = "post_cortex.assemble_context")]
    async fn assemble_context(
        &self,
        _req: AssembleContextRequest,
    ) -> Result<AssembleContextResponse, SystemError> {
        Self::not_yet_wired("assemble_context")
    }

    #[tracing::instrument(skip(self, _req), name = "post_cortex.manage_session")]
    async fn manage_session(
        &self,
        _req: ManageSessionRequest,
    ) -> Result<ManageSessionResponse, SystemError> {
        Self::not_yet_wired("manage_session")
    }

    #[tracing::instrument(skip(self, _req), name = "post_cortex.manage_workspace")]
    async fn manage_workspace(
        &self,
        _req: ManageWorkspaceRequest,
    ) -> Result<ManageWorkspaceResponse, SystemError> {
        Self::not_yet_wired("manage_workspace")
    }

    #[tracing::instrument(skip(self, _req), name = "post_cortex.manage_entity")]
    async fn manage_entity(
        &self,
        _req: ManageEntityRequest,
    ) -> Result<ManageEntityResponse, SystemError> {
        Self::not_yet_wired("manage_entity")
    }

    #[tracing::instrument(skip(self, _req), name = "post_cortex.get_structured_summary")]
    async fn get_structured_summary(
        &self,
        _req: StructuredSummaryRequest,
    ) -> Result<StructuredSummaryResponse, SystemError> {
        Self::not_yet_wired("get_structured_summary")
    }

    #[tracing::instrument(skip(self, _req), name = "post_cortex.admin")]
    async fn admin(&self, _req: AdminRequest) -> Result<AdminResponse, SystemError> {
        Self::not_yet_wired("admin")
    }
}

// ---------------------------------------------------------------------------
// Canonical helpers — single source of truth for update_context validation
// and metadata shaping. Every transport (gRPC, MCP, REST) reaches the same
// behaviour by going through `update_context` / `bulk_update_context`; the
// helpers below are intentionally private so transports cannot bypass them.
// ---------------------------------------------------------------------------

/// Redaction never fails a write; it is logged so a spike (e.g. an agent
/// pasting a log full of addresses) is visible without exposing the values.
fn log_redactions(session_id: Uuid, counts: &RedactionCounts) {
    if !counts.is_empty() {
        tracing::info!(%session_id, ?counts, "update_context: redacted sensitive values");
    }
}

/// Strict validation lifted from the legacy gRPC helper. Rejects any input
/// that would corrupt the entity graph (empty title+description, missing
/// entities/relations, dangling relation endpoints, self-relations, empty
/// relation context).
fn validate_update_request(req: &UpdateContextRequest) -> Result<(), SystemError> {
    if req.content.title.trim().is_empty() && req.content.description.trim().is_empty() {
        return Err(SystemError::InvalidArgument(
            "update_context: title and description are both empty — provide at least one".into(),
        ));
    }
    if req.entities.is_empty() {
        return Err(SystemError::InvalidArgument(
            "update_context: entities must not be empty".into(),
        ));
    }
    if req.relations.is_empty() {
        return Err(SystemError::InvalidArgument(
            "update_context: relations must not be empty".into(),
        ));
    }

    let entity_names: HashSet<&str> = req.entities.iter().map(|e| e.name.as_str()).collect();
    for (i, rel) in req.relations.iter().enumerate() {
        if rel.from_entity.is_empty() {
            return Err(SystemError::InvalidArgument(format!(
                "relation[{i}]: from_entity must not be empty"
            )));
        }
        if rel.to_entity.is_empty() {
            return Err(SystemError::InvalidArgument(format!(
                "relation[{i}]: to_entity must not be empty"
            )));
        }
        if rel.from_entity == rel.to_entity {
            return Err(SystemError::InvalidArgument(format!(
                "relation[{i}]: self-relations are not allowed (from_entity == to_entity == {:?})",
                rel.from_entity
            )));
        }
        if rel.context.trim().is_empty() {
            return Err(SystemError::InvalidArgument(format!(
                "relation[{i}]: context must not be empty — every relation requires an explanation"
            )));
        }
        if !entity_names.contains(rel.from_entity.as_str()) {
            return Err(SystemError::InvalidArgument(format!(
                "relation[{i}]: from_entity {:?} is not declared in the entities list",
                rel.from_entity
            )));
        }
        if !entity_names.contains(rel.to_entity.as_str()) {
            return Err(SystemError::InvalidArgument(format!(
                "relation[{i}]: to_entity {:?} is not declared in the entities list",
                rel.to_entity
            )));
        }
    }

    Ok(())
}

/// Build the text fed to the vectorizer — `title\n description` matches what
/// the MCP path settled on for cross-session search quality.
fn build_description(req: &UpdateContextRequest) -> String {
    if req.content.description.is_empty() {
        req.content.title.clone()
    } else if req.content.title.is_empty() {
        req.content.description.clone()
    } else {
        format!("{}\n{}", req.content.title, req.content.description)
    }
}

/// Build the `ContextUpdate` JSON blob stored as the storage entry's metadata.
/// Keeps `creates_entities` (names) and `typed_entities` (name + type) in sync
/// — both are consumed downstream by the entity graph builder.
fn build_context_update(req: &UpdateContextRequest) -> ContextUpdate {
    let typed_entities: Vec<TypedEntity> = req
        .entities
        .iter()
        .map(|e| TypedEntity {
            name: e.name.clone(),
            entity_type: e.entity_type.clone(),
        })
        .collect();
    let creates_entities: Vec<String> = req.entities.iter().map(|e| e.name.clone()).collect();
    let creates_relationships: Vec<EntityRelationship> = req.relations.clone();
    let related_code = req.code_reference.clone();

    ContextUpdate {
        id: Uuid::new_v4(),
        timestamp: Utc::now(),
        update_type: req.interaction_type.clone(),
        content: req.content.clone(),
        related_code,
        parent_update: None,
        user_marked_important: false,
        creates_entities,
        creates_relationships,
        references_entities: Vec::new(),
        typed_entities,
        author: req.author.clone(),
        derived_from: req.derived_from.clone(),
    }
}

/// Hand a freshly-persisted update off to the bounded background pipeline:
/// embedding compute, entity-graph merge, and summary refresh all run on
/// separate worker tasks so the write path returns as soon as the entry
/// is stored. The pipeline is the only place this work runs; when a queue
/// rejects an item (full or shut down) the same work runs on a plain task
/// instead, so it is never dropped.
fn submit_derived_work(
    pipeline: &Pipeline,
    system: &Arc<ConversationMemorySystem>,
    session_id: Uuid,
    entry_id: Uuid,
    text: &str,
    update: ContextUpdate,
) {
    if let Err(e) = pipeline.submit_embedding(EmbeddingWorkItem {
        session_id,
        entry_id,
        text: text.to_string(),
    }) {
        log_pipeline_submit("embedding", session_id, entry_id, e);
        #[cfg(feature = "embeddings")]
        {
            let system = Arc::clone(system);
            tokio::spawn(async move {
                if let Err(e) = system.vectorize_update_now(session_id, entry_id).await {
                    warn!(%session_id, %entry_id, error = %e, "fallback vectorization failed");
                }
            });
        }
    }
    // The queue drops a rejected item, so keep the update for the fallback.
    let item = GraphWorkItem::ApplyUpdate {
        session_id,
        update: update.clone(),
    };
    if let Err(e) = pipeline.submit_graph(item) {
        log_pipeline_submit("graph", session_id, entry_id, e);
        let system = Arc::clone(system);
        tokio::spawn(async move {
            if let Err(e) = system
                .apply_entity_graph_update_now(session_id, update)
                .await
            {
                warn!(%session_id, %entry_id, error = %e, "fallback graph merge failed");
            }
        });
    }
    if let Err(e) = pipeline.submit_summary(SummaryWorkItem { session_id }) {
        log_pipeline_submit("summary", session_id, entry_id, e);
    }
}

/// A full queue is expected during bulk writes (the work then runs on a
/// plain task, so nothing is lost): each case is logged at debug level and
/// summed into one warning per queue at most every [`BACKPRESSURE_WARN_EVERY`].
/// A stopped worker is a warning every time.
fn log_pipeline_submit(queue: &'static str, session_id: Uuid, entry_id: Uuid, err: PipelineError) {
    match err {
        PipelineError::Backpressure { .. } => {
            tracing::debug!(queue, %session_id, %entry_id, "pipeline queue full; running the work on a plain task");
            if let Some((count, since)) = BACKPRESSURE.note(queue) {
                warn!(
                    queue,
                    count,
                    seconds = since.as_secs(),
                    "pipeline queue was full: {count} work items ran on plain tasks instead (bulk ingest?)"
                );
            }
        }
        err @ PipelineError::WorkerShutdown { .. } => warn!(
            queue,
            %session_id,
            %entry_id,
            error = %err,
            "pipeline submission failed (non-fatal — running the work on a plain task)"
        ),
    }
}

/// How often a full pipeline queue is reported.
const BACKPRESSURE_WARN_EVERY: std::time::Duration = std::time::Duration::from_secs(60);

static BACKPRESSURE: BackpressureLog = BackpressureLog::new();

/// Per-queue count of full-queue fallbacks since the last warning.
struct BackpressureLog {
    queues: parking_lot::Mutex<Vec<(&'static str, std::time::Instant, u64)>>,
}

impl BackpressureLog {
    const fn new() -> Self {
        Self {
            queues: parking_lot::Mutex::new(Vec::new()),
        }
    }

    /// Count one fallback for `queue`; returns the count and the time it
    /// covers when a warning is due (the first case, then at most every
    /// [`BACKPRESSURE_WARN_EVERY`]).
    fn note(&self, queue: &'static str) -> Option<(u64, std::time::Duration)> {
        let now = std::time::Instant::now();
        let mut queues = self.queues.lock();
        let Some(entry) = queues.iter_mut().find(|(q, _, _)| *q == queue) else {
            queues.push((queue, now, 0));
            return Some((1, std::time::Duration::ZERO));
        };
        entry.2 += 1;
        let since = now.duration_since(entry.1);
        if since < BACKPRESSURE_WARN_EVERY {
            return None;
        }
        let count = entry.2;
        *entry = (queue, now, 0);
        Some((count, since))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::memory_system::SystemConfig;
    use chrono::Utc;
    use post_cortex_core::core::context_update::{
        EntityData, EntityType, RelationType, UpdateContent, UpdateType,
    };

    fn entity(name: &str) -> EntityData {
        EntityData {
            name: name.to_string(),
            entity_type: EntityType::Concept,
            first_mentioned: Utc::now(),
            last_mentioned: Utc::now(),
            mention_count: 1,
            importance_score: 1.0,
            description: None,
        }
    }

    fn relation(from: &str, to: &str) -> EntityRelationship {
        EntityRelationship {
            from_entity: from.to_string(),
            to_entity: to.to_string(),
            relation_type: RelationType::RelatedTo,
            context: "test relation".to_string(),
        }
    }

    fn good_request() -> UpdateContextRequest {
        UpdateContextRequest {
            session_id: Uuid::new_v4(),
            interaction_type: UpdateType::ConceptDefined,
            content: UpdateContent {
                title: "Some concept".into(),
                description: "A short definition".into(),
                details: vec![],
                examples: vec![],
                implications: vec![],
            },
            entities: vec![entity("Foo"), entity("Bar")],
            relations: vec![relation("Foo", "Bar")],
            code_reference: None,
            author: None,
            derived_from: Vec::new(),
        }
    }

    #[tokio::test]
    async fn trait_is_object_safe() {
        fn _accept_dyn(_svc: Arc<dyn PostCortexService>) {}
    }

    #[test]
    fn validation_rejects_empty_title_and_description() {
        let mut req = good_request();
        req.content.title = String::new();
        req.content.description = String::new();
        let err = validate_update_request(&req).unwrap_err();
        assert!(matches!(err, SystemError::InvalidArgument(ref m) if m.contains("both empty")));
    }

    #[test]
    fn validation_rejects_empty_entities() {
        let mut req = good_request();
        req.entities = vec![];
        assert!(matches!(
            validate_update_request(&req),
            Err(SystemError::InvalidArgument(_))
        ));
    }

    #[test]
    fn validation_rejects_empty_relations() {
        let mut req = good_request();
        req.relations = vec![];
        assert!(matches!(
            validate_update_request(&req),
            Err(SystemError::InvalidArgument(_))
        ));
    }

    #[test]
    fn validation_rejects_self_relation() {
        let mut req = good_request();
        req.relations = vec![relation("Foo", "Foo")];
        assert!(matches!(
            validate_update_request(&req),
            Err(SystemError::InvalidArgument(ref m)) if m.contains("self-relations")
        ));
    }

    #[test]
    fn validation_rejects_dangling_relation_endpoint() {
        let mut req = good_request();
        req.relations = vec![relation("Foo", "Ghost")];
        assert!(matches!(
            validate_update_request(&req),
            Err(SystemError::InvalidArgument(ref m)) if m.contains("Ghost")
        ));
    }

    #[test]
    fn validation_rejects_empty_relation_context() {
        let mut req = good_request();
        req.relations[0].context = "   ".into();
        assert!(matches!(
            validate_update_request(&req),
            Err(SystemError::InvalidArgument(ref m)) if m.contains("context must not be empty")
        ));
    }

    #[test]
    fn validation_accepts_good_request() {
        assert!(validate_update_request(&good_request()).is_ok());
    }

    #[test]
    fn description_joins_title_and_body() {
        let req = good_request();
        assert_eq!(build_description(&req), "Some concept\nA short definition");
    }

    #[test]
    fn description_falls_back_when_one_side_empty() {
        let mut req = good_request();
        req.content.description = String::new();
        assert_eq!(build_description(&req), "Some concept");

        req.content.title = String::new();
        req.content.description = "Body only".into();
        assert_eq!(build_description(&req), "Body only");
    }

    #[test]
    fn metadata_keeps_creates_entities_in_sync_with_typed_entities() {
        let req = good_request();
        let update = build_context_update(&req);
        let meta = serde_json::to_value(&update).unwrap();
        let names = meta["creates_entities"].as_array().unwrap();
        let typed = meta["typed_entities"].as_array().unwrap();
        assert_eq!(names.len(), typed.len());
        assert_eq!(names.len(), req.entities.len());
        for (i, e) in req.entities.iter().enumerate() {
            assert_eq!(names[i].as_str().unwrap(), e.name);
            assert_eq!(typed[i]["name"].as_str().unwrap(), e.name);
        }
    }

    async fn make_service(suffix: &str) -> (MemoryServiceImpl, String) {
        let test_dir = format!(
            "./test_data_memservice_{}_{}",
            suffix,
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );
        std::fs::create_dir_all(&test_dir).unwrap();
        let config = SystemConfig {
            data_directory: test_dir.clone(),
            ..Default::default()
        };
        let system = Arc::new(ConversationMemorySystem::new(config).await.unwrap());
        (MemoryServiceImpl::new(system), test_dir)
    }

    #[tokio::test]
    async fn health_returns_ok_status() {
        let (svc, test_dir) = make_service("health").await;
        let report = svc.health().await.unwrap();
        assert!(report.status == "ok" || report.status == "degraded");
        std::fs::remove_dir_all(&test_dir).unwrap();
    }

    #[tokio::test]
    async fn update_context_persists_and_returns_entry_id() {
        let (svc, test_dir) = make_service("update").await;
        let session_id = svc.inner().create_session(None, None).await.unwrap();

        let mut req = good_request();
        req.session_id = session_id;
        let resp = svc.update_context(req).await.unwrap();

        assert_eq!(resp.session_id, session_id);
        assert!(resp.durable);
        std::fs::remove_dir_all(&test_dir).unwrap();
    }

    #[tokio::test]
    async fn update_context_rejects_invalid_input_with_invalid_argument() {
        let (svc, test_dir) = make_service("invalid").await;
        let session_id = svc.inner().create_session(None, None).await.unwrap();

        let mut req = good_request();
        req.session_id = session_id;
        req.entities = vec![]; // violates "entities must not be empty"
        let err = svc.update_context(req).await.unwrap_err();
        assert!(matches!(err, SystemError::InvalidArgument(_)));
        std::fs::remove_dir_all(&test_dir).unwrap();
    }

    #[tokio::test]
    async fn bulk_update_context_persists_every_item() {
        let (svc, test_dir) = make_service("bulk").await;
        let session_id = svc.inner().create_session(None, None).await.unwrap();

        let mut a = good_request();
        a.session_id = session_id;
        a.content.title = "First".into();
        let mut b = good_request();
        b.session_id = session_id;
        b.content.title = "Second".into();

        let resp = svc
            .bulk_update_context(BulkUpdateContextRequest {
                session_id,
                updates: vec![a, b],
            })
            .await
            .unwrap();
        assert_eq!(resp.entry_ids.len(), 2);
        assert!(resp.durable);
        std::fs::remove_dir_all(&test_dir).unwrap();
    }

    #[tokio::test]
    async fn update_context_returns_fast_then_pipeline_drains_in_background() {
        // The hot path persists durably; embedding + graph + summary land
        // on background workers. Verify that, on a warm path
        // (post-vectorizer-init), update_context returns quickly and the
        // pipeline backlog drains shortly after.
        //
        // The model load is paid up front (as the daemon does at startup)
        // so the drain below only measures real work.
        let (svc, test_dir) = make_service("nonblocking").await;
        let session_id = svc.inner().create_session(None, None).await.unwrap();
        svc.inner().warm_up_embeddings().await.unwrap();

        // Warm-up call — first write through every worker.
        let mut warmup = good_request();
        warmup.session_id = session_id;
        warmup.content.title = "warmup".into();
        let _ = svc.update_context(warmup).await.unwrap();

        // Wait for any pending background work to settle.
        for _ in 0..100 {
            if svc.pipeline().backlog() == 0 {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }

        // Measured call — should be fast.
        let mut req = good_request();
        req.session_id = session_id;
        let start = std::time::Instant::now();
        let resp = svc.update_context(req).await.unwrap();
        let write_latency = start.elapsed();

        assert!(resp.durable);
        assert!(
            write_latency.as_millis() < 250,
            "update_context took {write_latency:?} on warm path — should be <250ms"
        );

        // Workers drain the queues asynchronously.
        for _ in 0..100 {
            if svc.pipeline().backlog() == 0 {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        assert_eq!(
            svc.pipeline().backlog(),
            0,
            "pipeline backlog should drain within 2s"
        );

        std::fs::remove_dir_all(&test_dir).unwrap();
    }

    #[tokio::test]
    async fn update_context_merges_entities_exactly_once() {
        let dir = tempfile::tempdir().unwrap();
        let config = SystemConfig {
            data_directory: dir.path().to_string_lossy().into_owned(),
            enable_embeddings: false,
            ..Default::default()
        };
        let system = Arc::new(ConversationMemorySystem::new(config).await.unwrap());
        let svc = MemoryServiceImpl::new(system);
        let session_id = svc.inner().create_session(None, None).await.unwrap();

        for round in 1..=3u32 {
            let mut req = good_request();
            req.session_id = session_id;
            req.content.title = format!("round {round}");
            assert!(svc.update_context(req).await.unwrap().durable);

            let mut mentions = 0;
            for _ in 0..200 {
                let session = svc
                    .inner()
                    .get_session(session_id)
                    .await
                    .unwrap()
                    .load_full();
                mentions = session
                    .entity_graph
                    .entities
                    .get("Foo")
                    .map_or(0, |e| e.mention_count);
                if svc.pipeline().backlog() == 0 && mentions >= round {
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
            assert_eq!(mentions, round, "one merge per update");
        }

        // No second path applies them again later.
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        let session = svc
            .inner()
            .get_session(session_id)
            .await
            .unwrap()
            .load_full();
        assert_eq!(session.entity_graph.entities["Foo"].mention_count, 3);
    }

    #[tokio::test]
    async fn bulk_update_context_rejects_mismatched_session() {
        let (svc, test_dir) = make_service("bulkmis").await;
        let session_id = svc.inner().create_session(None, None).await.unwrap();

        let mut item = good_request();
        item.session_id = Uuid::new_v4(); // intentionally different
        let err = svc
            .bulk_update_context(BulkUpdateContextRequest {
                session_id,
                updates: vec![item],
            })
            .await
            .unwrap_err();
        assert!(matches!(err, SystemError::InvalidArgument(ref m) if m.contains("does not match")));
        std::fs::remove_dir_all(&test_dir).unwrap();
    }

    /// Cards: an unknown workspace is refused, versions advance, a stale
    /// version conflicts with the latest returned, secrets are redacted and
    /// the size limit holds.
    #[tokio::test]
    async fn cards_are_versioned_redacted_and_bounded() {
        let (svc, dir) = make_service("cards").await;
        let update = |scope, text: &str, expected_version| UpdateCardRequest {
            scope,
            text: text.to_string(),
            expected_version,
            author: Some("claude".to_string()),
        };

        let unknown = CardScope::Workspace(Uuid::new_v4());
        assert!(matches!(
            svc.update_card(update(unknown, "x", 0)).await,
            Err(SystemError::WorkspaceNotFound(_))
        ));

        let ws = svc
            .inner()
            .workspace_manager
            .create_workspace("cards".to_string(), String::new());
        let scope = CardScope::Workspace(ws);
        let first = svc
            .update_card(update(
                scope,
                "## Stack\nrust; key AKIAABCDEFGHIJKLMNOP\n\n",
                0,
            ))
            .await
            .unwrap();
        assert!(first.written);
        let card = first.card.unwrap();
        assert_eq!(card.version, 1);
        assert!(card.text.contains("[REDACTED:"), "{}", card.text);
        assert!(
            !card.text.ends_with('\n'),
            "trailing blank lines are trimmed"
        );

        let stale = svc.update_card(update(scope, "other", 0)).await.unwrap();
        assert!(!stale.written);
        assert_eq!(stale.card.unwrap().version, 1, "the latest comes back");

        assert!(matches!(
            svc.update_card(update(scope, &"x".repeat(MAX_CARD_BYTES + 1), 1))
                .await,
            Err(SystemError::InvalidArgument(_))
        ));
        assert!(
            svc.update_card(update(scope, "## Stack\nrust", 1))
                .await
                .unwrap()
                .written
        );

        let read = svc
            .get_card(GetCardRequest { scope, history: 5 })
            .await
            .unwrap();
        assert_eq!(read.card.unwrap().version, 2);
        assert_eq!(read.history.len(), 1);
        let user = svc
            .get_card(GetCardRequest {
                scope: CardScope::User,
                history: 0,
            })
            .await
            .unwrap();
        assert!(user.card.is_none());

        let _ = std::fs::remove_dir_all(dir);
    }

    /// The entities a write names get their vectors from the write itself,
    /// once the graph worker has merged them: search finds them without a
    /// vectorize_session run.
    #[tokio::test]
    async fn entities_of_a_write_are_embedded() {
        let (svc, dir) = make_service("entity-vectors").await;
        let sid = svc.inner().create_session(None, None).await.unwrap();
        let mut req = good_request();
        req.session_id = sid;
        svc.update_context(req).await.unwrap();

        let vectorizer = svc.inner().ensure_vectorizer_initialized().await.unwrap();
        let mut found = false;
        for _ in 0..100 {
            let hits = vectorizer
                .semantic_search("Foo", 10, Some(sid), Default::default())
                .await
                .unwrap();
            if hits.iter().any(|h| h.content_id == "entity:Foo") {
                found = true;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert!(found, "entity:Foo has no vector");
        let _ = std::fs::remove_dir_all(dir);
    }

    /// `derived_from` must name messages stored in the session; the links
    /// come back on the record's search hits.
    #[tokio::test]
    async fn derived_from_names_stored_messages() {
        use post_cortex_core::core::message::{MessageRole, NewMessage};

        let (svc, dir) = make_service("derived-from").await;
        let sid = svc.inner().create_session(None, None).await.unwrap();
        let (stored, _) = svc
            .inner()
            .append_messages(
                sid,
                vec![NewMessage {
                    turn_id: "t1".to_string(),
                    role: MessageRole::User,
                    author: None,
                    source_session: None,
                    timestamp: Utc::now(),
                    text: "Let us keep the cards in front of every session".to_string(),
                    tools: Vec::new(),
                    files: Vec::new(),
                }],
            )
            .await
            .unwrap();
        let id = message::content_id(stored[0].seq);

        let with = |derived_from: Vec<String>| {
            let mut req = good_request();
            req.session_id = sid;
            req.derived_from = derived_from;
            req
        };
        for bad in [vec!["entity:Foo".to_string()], vec![format!("{id}#1")]] {
            assert!(matches!(
                svc.update_context(with(bad)).await,
                Err(SystemError::InvalidArgument(_))
            ));
        }
        let missing = message::content_id(stored[0].seq + 7);
        assert!(matches!(
            svc.update_context(with(vec![missing])).await,
            Err(SystemError::InvalidArgument(_))
        ));
        svc.update_context(with(vec![id.clone()])).await.unwrap();

        let hits = svc
            .inner()
            .semantic_search_session(sid, "Some concept short definition", Default::default())
            .await
            .unwrap();
        let record = hits
            .iter()
            .find(|h| h.content_type == crate::content_vectorizer::ContentType::UpdateContent)
            .expect("the record is found");
        assert_eq!(record.derived_from, [id]);
        let _ = std::fs::remove_dir_all(dir);
    }

    /// A burst of full-queue fallbacks warns once, then once a minute with
    /// the count.
    #[test]
    fn backpressure_is_summarised() {
        let log = BackpressureLog::new();
        assert_eq!(log.note("embedding").map(|(n, _)| n), Some(1));
        for _ in 0..4_999 {
            assert!(log.note("embedding").is_none());
        }
        assert_eq!(
            log.note("graph").map(|(n, _)| n),
            Some(1),
            "queues are separate"
        );
        log.queues.lock()[0].1 -= BACKPRESSURE_WARN_EVERY;
        assert_eq!(log.note("embedding").map(|(n, _)| n), Some(5_000));
        assert!(log.note("embedding").is_none());
    }
}