app/services/student_memory.py

"""Cognee-backed student/project memory boundary.

All app code should go through this service instead of importing ``cognee``
directly. Cognee is load-bearing for ResearchMate, but it must never be allowed
to crash a hot study path when a dataset is empty, a native API drifts, or a
local install has not finished bootstrapping.
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import os
import re
import time
import uuid
from pathlib import Path
from typing import Any, Iterable, List, Literal, Sequence, TYPE_CHECKING

from app.observability.operation import Operation, observe_operation, observe_stage
from app.schemas.journal import JournalEntry
from app.rag.project_memory import (
    ProjectMemoryRecord,
    ProjectMemoryStore,
    make_project_memory_id,
)
from app.rag.evidence_store import EvidenceStore
from app.rag.models import normalize_evidence_text
from app.services.memory_candidates import MemoryCandidate, interaction_ref, make_candidate_id
from app.services.memory_promotion import MemoryPromotionGate

if TYPE_CHECKING:
    from app.agents.evaluator_agent import NodeAssessment

MemoryMode = Literal["concept", "profile", "project", "temporal"]
FlushStrategy = Literal["improve", "distill_then_improve"]

PROFILE_DATASET = "research_profile"
MEMORY_ROOT = Path(os.path.expanduser("~/.studybuddy/memory_integrity"))
MAX_RECALL_CHUNK_CHARS = 1200
MAX_RECALL_CONTEXT_CHARS = 6000
COGNEE_SLOW_OPERATION_SECONDS = 3.0
COGNEE_FLUSH_SPACING_SECONDS = 1.0
RESEARCH_MEMORY_REVIEWER_SKILL = (
    Path(__file__).resolve().parents[1]
    / "memory_skills"
    / "research-memory-reviewer"
    / "SKILL.md"
)
logger = logging.getLogger(__name__)


async def _maybe_await(value: Any) -> Any:
    if asyncio.iscoroutine(value):
        return await value
    return value


def _serialise(value: Any) -> Any:
    if hasattr(value, "model_dump"):
        return value.model_dump()
    if hasattr(value, "dict"):
        return value.dict()
    if isinstance(value, (list, tuple)):
        return [_serialise(v) for v in value]
    if isinstance(value, dict):
        return {str(k): _serialise(v) for k, v in value.items()}
    return value


def _extract_text(item: Any) -> str:
    return (
        getattr(item, "text", None)
        or getattr(item, "content", None)
        or getattr(item, "answer", None)
        or str(item)
    )


def _clip_text(text: str, limit: int) -> str:
    if len(text) <= limit:
        return text
    return text[:limit].rstrip() + "\n...[truncated]"


def _recall_policy(*, cache_only: bool, bypass_cache: bool) -> str:
    if cache_only:
        return "cache_only"
    if bypass_cache:
        return "live"
    return "cache_then_live"




def _safe_project_id(project_id: str) -> str:
    return project_id.replace("/", "_").replace("\\", "_").replace("..", "_")


class StudentMemoryService:
    """Defensive facade over Cognee native memory APIs."""

    _cognee_runtime_ready = False
    _cognee_flush_lock = asyncio.Lock()
    _profile_recall_cache: dict[str, tuple[float, str]] = {}
    _profile_recall_cache_ttl_seconds = 300.0

    async def _cognee(self) -> Any:
        """Return Cognee after applying the local ResearchMate runtime config.

        FastAPI lifespan normally performs this once at startup. This guard makes
        memory calls robust when a route/test/import path touches StudentMemoryService
        before startup has completed or after Cognee has been imported with defaults.
        """
        os.environ.setdefault("EMBEDDING_PROVIDER", "fastembed")
        os.environ.setdefault("EMBEDDING_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
        os.environ.setdefault("EMBEDDING_DIMENSIONS", "384")
        os.environ["ENABLE_BACKEND_ACCESS_CONTROL"] = "false"
        skill_root = str(Path(__file__).resolve().parents[1] / "memory_skills")
        existing_skill_roots = os.environ.get("COGNEE_SKILL_SOURCE_ROOTS", "")
        if skill_root not in existing_skill_roots.split(os.pathsep):
            os.environ["COGNEE_SKILL_SOURCE_ROOTS"] = (
                skill_root if not existing_skill_roots else existing_skill_roots + os.pathsep + skill_root
            )

        import cognee

        if not StudentMemoryService._cognee_runtime_ready:
            t0 = time.perf_counter()
            try:
                import importlib
                from app.services.cognee_bootstrap import configure_cognee_llm
                from cognee.infrastructure.databases.relational.create_db_and_tables import create_db_and_tables

                cognee_llm_client = importlib.import_module(
                    "cognee.infrastructure.llm.structured_output_framework.litellm_instructor.llm.get_llm_client"
                )
                configure_cognee_llm(
                    cognee.config,
                    clear_llm_client_cache=cognee_llm_client._get_llm_client_cached.cache_clear,
                )
                root = str(Path.home() / ".studybuddy" / "cognee")
                cognee.config.data_root_directory(root)
                cognee.config.system_root_directory(f"{root}/system")
                await create_db_and_tables()
                StudentMemoryService._cognee_runtime_ready = True
                elapsed = time.perf_counter() - t0
                if elapsed >= COGNEE_SLOW_OPERATION_SECONDS:
                    logger.warning("Cognee runtime bootstrap took %.2fs", elapsed)
            except Exception as exc:  # noqa: BLE001
                logger.warning("Cognee runtime self-bootstrap failed: %s", exc)
        return cognee

    def _activity(self, project_id: str, event_type: str, message: str, status: str = "ok", metadata: dict[str, Any] | None = None) -> None:
        if not project_id:
            return
        try:
            from app.services.project_activity import ProjectActivityService

            ProjectActivityService().record(project_id, event_type, message, status=status, metadata=metadata or {})
        except Exception:
            pass

    def _integrity_dir(self, name: str) -> Path:
        path = MEMORY_ROOT / name
        path.mkdir(parents=True, exist_ok=True)
        return path

    def _source_registry_path(self, project_id: str) -> Path:
        return self._integrity_dir("source_registry") / f"{_safe_project_id(project_id)}.json"

    def _temporal_path(self, project_id: str) -> Path:
        return self._integrity_dir("temporal") / f"{_safe_project_id(project_id)}.jsonl"

    def _status_path(self, project_id: str) -> Path:
        return self._integrity_dir("status") / f"{_safe_project_id(project_id)}.json"

    def _load_source_registry(self, project_id: str) -> dict[str, Any]:
        path = self._source_registry_path(project_id)
        if not path.exists():
            return {"sources": {}}
        try:
            with open(path, encoding="utf-8") as f:
                data = json.load(f)
            return data if isinstance(data, dict) else {"sources": {}}
        except Exception:
            return {"sources": {}}

    def _save_source_registry(self, project_id: str, registry: dict[str, Any]) -> None:
        with open(self._source_registry_path(project_id), "w", encoding="utf-8") as f:
            json.dump(registry, f, indent=2)

    def _record_memory_operation(
        self,
        project_id: str,
        operation: str,
        ok: bool,
        error: str = "",
        metadata: dict[str, Any] | None = None,
    ) -> None:
        if not project_id:
            return
        path = self._status_path(project_id)
        try:
            status = {}
            if path.exists():
                with open(path, encoding="utf-8") as f:
                    status = json.load(f)
            operations = status.setdefault("operations", {})
            operations[operation] = {
                "ok": ok,
                "error": error,
                "metadata": metadata or {},
                "updated_at": time.time(),
            }
            with open(path, "w", encoding="utf-8") as f:
                json.dump(status, f, indent=2)
        except Exception:
            pass

    def _recent_operation_status(self, project_id: str) -> dict[str, Any]:
        path = self._status_path(project_id)
        if not path.exists():
            return {}
        try:
            with open(path, encoding="utf-8") as f:
                return json.load(f)
        except Exception:
            return {}

    def record_temporal_event(
        self,
        project_id: str,
        event_type: str,
        text: str,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        if not project_id or not text:
            return
        event = {
            "created_at": time.time(),
            "event_type": event_type,
            "text": text,
            "metadata": metadata or {},
        }
        try:
            with open(self._temporal_path(project_id), "a", encoding="utf-8") as f:
                f.write(json.dumps(event, ensure_ascii=True) + "\n")
        except Exception:
            pass

    def _query_temporal_ledger(self, project_id: str, topic: str, limit: int = 8) -> str:
        path = self._temporal_path(project_id)
        if not path.exists():
            return ""
        needle = (topic or "").lower()
        rows: list[dict[str, Any]] = []
        try:
            with open(path, encoding="utf-8") as f:
                for line in f:
                    if not line.strip():
                        continue
                    try:
                        row = json.loads(line)
                    except Exception:
                        continue
                    text = str(row.get("text") or "")
                    if needle and needle not in text.lower():
                        continue
                    rows.append(row)
        except Exception:
            return ""
        rows = rows[-limit:]
        if not rows:
            return ""
        lines = [f"- {row.get('event_type', 'event')}: {row.get('text', '')}" for row in rows]
        return f'Temporal project memory for "{topic}":\n' + "\n".join(lines)

    def register_project_source(
        self,
        project_id: str,
        document_id: str,
        *,
        cognee_data_id: str = "",
        filename: str = "",
    ) -> None:
        if not project_id or not document_id:
            return
        registry = self._load_source_registry(project_id)
        sources = registry.setdefault("sources", {})
        sources[document_id] = {
            "document_id": document_id,
            "cognee_data_id": cognee_data_id,
            "filename": filename,
            "updated_at": time.time(),
        }
        self._save_source_registry(project_id, registry)

    # ------------------------------------------------------------------ #
    # Dataset lifecycle                                                   #
    # ------------------------------------------------------------------ #

    async def ensure_profile_dataset(self, *, observer: Operation | None = None) -> bool:
        return await self._ensure_dataset(
            PROFILE_DATASET,
            "Research profile dataset initialized.",
            observer=observer,
        )

    async def ensure_project_dataset(self, project_id: str) -> bool:
        del project_id
        return False

    async def _ensure_dataset(
        self,
        dataset: str,
        seed_text: str,
        *,
        observer: Operation | None = None,
    ) -> bool:
        try:
            cognee = await self._cognee()

            if await self._dataset_exists(cognee, dataset):
                return True
            await _maybe_await(cognee.add(seed_text, dataset_name=dataset))
            return True
        except Exception as exc:  # noqa: BLE001 - memory must degrade softly
            logger.warning(
                "Cognee dataset bootstrap failed for %s: %s",
                dataset,
                type(exc).__name__,
            )
            if observer is not None:
                observer.record_exception(
                    exc,
                    escaped=False,
                    stage="memory_promotion",
                    category="cognee_profile_dataset_bootstrap",
                )
            return False

    async def _dataset_exists(self, cognee: Any, dataset: str) -> bool:
        datasets_api = getattr(cognee, "datasets", None)
        list_datasets = getattr(datasets_api, "list_datasets", None)
        if list_datasets is None:
            return False
        try:
            existing = await _maybe_await(list_datasets())
        except Exception:
            return False
        for item in existing or []:
            name = item.get("name") if isinstance(item, dict) else getattr(item, "name", None)
            if name == dataset:
                return True
        return False

    # ------------------------------------------------------------------ #
    # Write / flush                                                       #
    # ------------------------------------------------------------------ #

    def _pending_memory_path(self, project_id: str) -> Path:
        return MEMORY_ROOT / "pending" / f"{_safe_project_id(project_id)}.json"

    def list_pending_memory(self, project_id: str) -> list[dict[str, Any]]:
        path = self._pending_memory_path(project_id)
        if not path.exists():
            return []
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except Exception:
            return []
        return data if isinstance(data, list) else []

    def _write_pending_memory(self, project_id: str, items: list[dict[str, Any]]) -> None:
        path = self._pending_memory_path(project_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(items, ensure_ascii=False, indent=2), encoding="utf-8")

    def stage_pending_memory(
        self,
        project_id: str,
        *,
        dataset: str,
        session_id: str,
        text: str,
        kind: str,
        metadata: dict[str, Any] | None = None,
    ) -> bool:
        if not project_id or not text.strip():
            return False
        items = self.list_pending_memory(project_id)
        items.append({
            "id": str(uuid.uuid4()),
            "project_id": project_id,
            "dataset": dataset,
            "session_id": session_id,
            "text": text.strip(),
            "kind": kind,
            "created_at": time.time(),
            "metadata": metadata or {},
        })
        self._write_pending_memory(project_id, items)
        self._record_memory_operation(project_id, "pending_stage", True, metadata={"kind": kind, "pending_count": len(items)})
        return True

    # _is_durable_pending_memory() / promote_pending_memory() were retired:
    # they duplicated MemoryPromotionGate's policy with an independent
    # keyword-heuristic gate, bypassing the authoritative one entirely, and
    # had zero production callers (confirmed by repo-wide grep during
    # design review). The Commit adaptation reviewer supersedes their role.
    # See docs/commit-memory-adaptation-architecture.md.

    async def stage_project_observation(
        self,
        project_id: str,
        topic: str,
        observations: Iterable["NodeAssessment"] | Iterable[str] | None = None,
        session_summary: str = "",
        journal_entries: Iterable[JournalEntry] | None = None,
    ) -> bool:
        del session_summary
        records: list[ProjectMemoryRecord] = []
        rejected_invalid_source = 0
        rejected_by_gate: dict[str, int] = {}
        gate = MemoryPromotionGate()
        journal_rows: list[tuple[str, str]] = []
        for entry in journal_entries or []:
            if entry.project_id != project_id:
                continue
            serialized = json.dumps(
                entry.model_dump(mode="json"),
                ensure_ascii=False,
                sort_keys=True,
                separators=(",", ":"),
            )
            journal_rows.append((interaction_ref(serialized), normalize_evidence_text(serialized).casefold()))
        for index, obs in enumerate(observations or []):
            if isinstance(obs, str):
                statement = obs.strip()
                node_id = ""
                raw_evidence: list[str] = []
                source_evidence: list[str] = []
            else:
                statement = str(getattr(obs, "observation", "") or "").strip()
                node_id = str(getattr(obs, "node_id", "") or "")
                raw_evidence = [str(item) for item in (getattr(obs, "evidence", None) or [])]
                source_evidence = [
                    str(item) for item in (getattr(obs, "source_evidence_ids", None) or [])
                ]
            if not statement:
                continue
            evidence_ids = list(dict.fromkeys([
                *[item for item in raw_evidence if item.startswith("ev_")],
                *[item for item in source_evidence if item.startswith("ev_")],
            ]))
            if evidence_ids:
                evidence_ids = [
                    unit.evidence_id
                    for unit in EvidenceStore().get_units(project_id, evidence_ids=evidence_ids)
                ]
            if source_evidence and not evidence_ids:
                # The observer claimed paper grounding but supplied no valid
                # canonical IDs. Keep the UI observation, reject the memory.
                rejected_invalid_source += 1
                continue
            interaction_ids: list[str] = []
            for item in raw_evidence:
                normalized_item = normalize_evidence_text(item).casefold()
                if not normalized_item or item.startswith("ev_"):
                    continue
                interaction_ids.extend(
                    journal_id
                    for journal_id, journal_text in journal_rows
                    if normalized_item in journal_text
                )
            interaction_ids = list(dict.fromkeys(interaction_ids))
            candidate = MemoryCandidate(
                candidate_id=make_candidate_id(
                    "project", project_id, "study_observation", statement
                ),
                destination="project",
                project_id=project_id,
                kind="study_observation",
                statement=statement,
                attribution="idea_observer_interaction",
                confidence=0.8 if evidence_ids else 0.7,
                evidence_ids=evidence_ids,
                interaction_ids=interaction_ids,
            )
            decision = gate.evaluate_project(candidate)
            if not decision.promote:
                rejected_by_gate[decision.reason] = rejected_by_gate.get(decision.reason, 0) + 1
                continue
            novelty_key = f"study_observation:{node_id or index}:{' '.join(statement.casefold().split())[:160]}"
            records.append(ProjectMemoryRecord(
                memory_id=make_project_memory_id(project_id, "study_observation", novelty_key, statement),
                project_id=project_id,
                kind="study_observation",
                statement=statement,
                evidence_ids=evidence_ids,
                interaction_ids=interaction_ids,
                attribution="idea_observer_interaction",
                confidence=0.75 if evidence_ids else 0.6,
                novelty_key=novelty_key,
                metadata={"topic": topic, "node_id": node_id, "raw_evidence": raw_evidence[:6]},
            ))
        with observe_operation(
            "memory.project_write",
            subsystem="memory",
            consumer="project_memory",
        ) as write_op:
            try:
                count = ProjectMemoryStore().upsert(records)
                ok = count == len(records)
            except Exception as exc:  # noqa: BLE001
                logger.warning("Project Chroma memory write failed: %s", type(exc).__name__)
                write_op.record_exception(
                    exc,
                    escaped=False,
                    stage="project_memory_write",
                    category="chroma_project_memory_write",
                )
                write_op.mark_terminal("degraded")
                ok = False
                count = 0
            write_op.add_count("project_memory_count", count)
        self._record_memory_operation(
            project_id,
            "write_project_chroma",
            ok,
            metadata={
                "records": count,
                "rejected_invalid_source": rejected_invalid_source,
                "rejected_by_gate": rejected_by_gate,
            },
        )
        if records:
            self.record_temporal_event(project_id, "project_memory_committed", f"Stored {count} grounded project observations")
        self._activity(
            project_id,
            "memory_stage",
            f"Stored {count} project observation{'s' if count != 1 else ''}" if ok else "Project memory write failed",
            status="ok" if ok else "warning",
        )
        return ok

    async def stage_profile_observation(
        self,
        project_id: str,
        text: str,
        *,
        kind: str = "learner_preference",
        attribution: str,
        confidence: float,
        interaction_ids: list[str] | None = None,
        evidence_ids: list[str] | None = None,
    ) -> bool:
        """Persist an already-approved student-memory observation.

        Promotion policy (explicitness, cross-project recurrence, confidence
        thresholds) must be evaluated by MemoryPromotionGate before this
        method is called. This method only serializes and writes -- it has
        no policy of its own. See
        docs/commit-memory-adaptation-architecture.md.
        """
        if not text:
            return False
        statement = " ".join(text.strip().split())[:1000]
        fingerprint = hashlib.sha256(f"{kind}\0{statement.casefold()}".encode("utf-8")).hexdigest()
        payload = {
            "schema": "student_profile_signal.v1",
            "signal_id": f"sm_{fingerprint}",
            "kind": kind,
            "statement": statement,
            "attribution": attribution,
            "confidence": max(0.0, min(1.0, float(confidence))),
            "interaction_ids": list(interaction_ids or []),
            "evidence_ids": list(evidence_ids or []),
            "observed_at": time.time(),
        }
        self._activity(project_id, "profile_memory_stage", "Profile memory is being staged", status="running")
        with observe_operation(
            "memory.profile_write",
            subsystem="memory",
            consumer="student_memory",
        ) as write_op:
            ok = await self._remember(
                dataset=PROFILE_DATASET,
                session_id=project_id,
                text=json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
                ensure=lambda: self.ensure_profile_dataset(observer=write_op),
                temporal=True,
                observer=write_op,
            )
            if not ok:
                write_op.mark_terminal("degraded")
        if ok:
            StudentMemoryService._profile_recall_cache.clear()
        self._record_memory_operation(project_id, "write_profile", ok, "" if ok else "Profile memory staging failed")
        self._activity(project_id, "profile_memory_stage", "Profile memory staged" if ok else "Profile memory staging failed", status="ok" if ok else "warning")
        return ok

    async def stage_promoted_candidate(self, candidate: "MemoryCandidate") -> bool:
        """Persist a MemoryCandidate that MemoryPromotionGate has already approved."""
        if candidate.destination != "student":
            raise ValueError(f"Expected destination='student', got {candidate.destination!r}")
        return await self.stage_profile_observation(
            candidate.project_id,
            candidate.statement,
            kind=candidate.kind,
            attribution=candidate.attribution,
            confidence=candidate.confidence,
            interaction_ids=candidate.interaction_ids,
            evidence_ids=candidate.evidence_ids,
        )

    async def stage_project_snapshot(self, project_id: str, payload_json: str) -> bool:
        del payload_json
        # Snapshots already live in Project History. Duplicating their JSON into
        # semantic memory polluted recall and Cognee's graph.
        self.record_temporal_event(project_id, "commit_snapshot", "Committed project snapshot")
        self._record_memory_operation(project_id, "write_snapshot", True, metadata={"memory_write": False})
        return True

    async def stage_visual_observation(self, document_id: str, text: str) -> bool:
        del document_id, text
        # Visual descriptions belong to canonical EvidenceUnit records. The
        # region ingestion path persists them with page/bbox provenance.
        return False

    async def _remember(
        self,
        dataset: str,
        session_id: str,
        text: str,
        ensure,
        temporal: bool,
        *,
        observer: Operation | None = None,
    ) -> bool:
        try:
            t0 = time.perf_counter()
            cognee = await self._cognee()

            if not await ensure():
                return False
            kwargs = dict(dataset_name=dataset, session_id=session_id, self_improvement=False)
            if temporal:
                kwargs["temporal_cognify"] = True
            try:
                await _maybe_await(cognee.remember(text, **kwargs))
            except TypeError:
                kwargs.pop("temporal_cognify", None)
                await _maybe_await(cognee.remember(text, **kwargs))
            elapsed = time.perf_counter() - t0
            if elapsed >= COGNEE_SLOW_OPERATION_SECONDS:
                logger.warning("Cognee remember for %s took %.2fs", dataset, elapsed)
            return True
        except Exception as exc:  # noqa: BLE001
            logger.warning("Cognee remember failed for %s: %s", dataset, type(exc).__name__)
            if observer is not None:
                observer.record_exception(
                    exc,
                    escaped=False,
                    stage="memory_promotion",
                    category="cognee_profile_write",
                )
            return False

    # push_session()/_profile_signal_from_session() were retired: they
    # duplicated the Commit adaptation reviewer's role with a keyword-
    # heuristic extractor, and had zero remaining production callers once
    # run_commit_adaptation() replaced _push_and_flush_memory() at both
    # call sites (COMMIT_PROJECT, CLOSE_PROJECT). Running both would also
    # have split cross-project recurrence evidence across two differently-
    # phrased extractions of the same underlying fact. See
    # docs/commit-memory-adaptation-architecture.md.

    async def flush_project(
        self,
        project_id: str,
        strategy: FlushStrategy = "improve",
        *,
        observer: Operation | None = None,
    ) -> dict[str, bool]:
        datasets = [(PROFILE_DATASET, self.ensure_profile_dataset)]
        results: dict[str, bool] = {}
        for dataset, ensure in datasets:
            self._activity(project_id, "memory_flush", f"Flushing {dataset}", status="running")
            results[dataset] = await self._flush_dataset(
                dataset,
                project_id,
                ensure,
                strategy,
                observer=observer,
            )
            self._activity(
                project_id,
                "memory_flush",
                f"{dataset} flushed" if results[dataset] else f"{dataset} flush failed",
                status="ok" if results[dataset] else "warning",
            )
        return results

    async def flush_profile(self, project_id: str, strategy: FlushStrategy = "improve") -> bool:
        self._activity(project_id, "memory_flush", f"Flushing {PROFILE_DATASET}", status="running")
        ok = await self._flush_dataset(PROFILE_DATASET, project_id, self.ensure_profile_dataset, strategy)
        self._activity(
            project_id,
            "memory_flush",
            f"{PROFILE_DATASET} flushed" if ok else f"{PROFILE_DATASET} flush failed",
            status="ok" if ok else "warning",
        )
        return ok

    async def flush_session(self, project_id: str) -> None:
        with observe_operation("memory.flush", subsystem="memory", consumer="student_memory") as op:
            with observe_stage(
                op,
                "memory_flush",
                subsystem="memory",
                consumer="student_memory",
            ) as flush_op:
                results = await self.flush_project(project_id, observer=flush_op)
            if isinstance(results, dict) and results:
                ok_count = sum(1 for v in results.values() if v)
                op.add_count("datasets_ok", ok_count)
                op.add_count("datasets_total", len(results))
                op.set("memory_liveness", "alive" if ok_count == len(results) else "degraded")
                if ok_count < len(results):
                    op.mark_terminal("degraded")

    async def _flush_dataset(
        self,
        dataset: str,
        project_id: str,
        ensure,
        strategy: FlushStrategy,
        *,
        observer: Operation | None = None,
    ) -> bool:
        try:
            cognee = await self._cognee()

            await ensure()
            t0 = time.perf_counter()
            async with StudentMemoryService._cognee_flush_lock:
                if strategy == "distill_then_improve":
                    distill = getattr(getattr(cognee, "session", None), "distill_session", None)
                    if distill is not None:
                        await _maybe_await(distill(session_id=project_id, dataset=dataset))
                await _maybe_await(cognee.improve(dataset=dataset, session_ids=[project_id]))
                if COGNEE_FLUSH_SPACING_SECONDS > 0:
                    await asyncio.sleep(COGNEE_FLUSH_SPACING_SECONDS)
            elapsed = time.perf_counter() - t0
            self._record_memory_operation(project_id, f"flush_{dataset}", True, metadata={"duration_seconds": round(elapsed, 3)})
            if elapsed >= COGNEE_SLOW_OPERATION_SECONDS:
                logger.warning("Cognee flush for %s took %.2fs", dataset, elapsed)
            return True
        except TypeError as exc:
            logger.warning("Cognee flush signature drift for %s: %s", dataset, type(exc).__name__)
            if observer is not None:
                observer.record_exception(
                    exc,
                    escaped=False,
                    stage="memory_flush",
                    category="cognee_flush_fallback",
                )
                observer.mark_terminal("degraded")
            self._record_memory_operation(project_id, f"flush_{dataset}", False, type(exc).__name__)
            return False
        except Exception as exc:  # noqa: BLE001
            logger.warning("Cognee flush failed for %s: %s", dataset, type(exc).__name__)
            if observer is not None:
                observer.record_exception(
                    exc,
                    escaped=False,
                    stage="memory_flush",
                    category="cognee_flush_fallback",
                )
                observer.mark_terminal("degraded")
            self._record_memory_operation(project_id, f"flush_{dataset}", False, type(exc).__name__)
            return False

    # ------------------------------------------------------------------ #
    # Read / feedback                                                     #
    # ------------------------------------------------------------------ #

    async def query_prior_knowledge(
        self,
        topic: str,
        project_id: str = "",
        mode: MemoryMode = "concept",
        *,
        cache_only: bool = False,
        bypass_cache: bool = False,
    ) -> str:
        if mode == "project":
            try:
                items = ProjectMemoryStore().search(project_id, topic, limit=8, consumer="project_memory")
                if not items:
                    return ""
                return "Relevant project memory:\n" + "\n".join(
                    f"- [{item.kind}; evidence={','.join(item.evidence_ids) or 'interaction'}] {item.statement}"
                    for item in items
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning("Project Chroma memory recall failed: %s", exc)
                return ""
        cache_key = " ".join(topic.casefold().split())[:240]
        with observe_operation(
            "memory.recall", subsystem="memory", consumer="student_memory",
            attributes={
                "memory_mode": mode,
                "recall_policy": _recall_policy(cache_only=cache_only, bypass_cache=bypass_cache),
            },
        ) as op:
            if mode in {"profile", "concept"} and not bypass_cache:
                cached = StudentMemoryService._profile_recall_cache.get(cache_key)
                if cached and time.time() - cached[0] <= self._profile_recall_cache_ttl_seconds:
                    op.set("cache_state", "hit")
                    op.set("memory_liveness", "not_observed")
                    return cached[1]
                if cache_only:
                    op.set("cache_state", "miss")
                    op.set("memory_liveness", "not_observed")
                    op.mark_terminal("success_empty")
                    return ""
            try:
                cognee = await self._cognee()
                from cognee import SearchType

                datasets = self._datasets_for_mode(project_id, mode)
                query_type = SearchType.TEMPORAL if mode == "temporal" else SearchType.GRAPH_COMPLETION
                query_text = self._query_for_mode(topic, mode)
                kwargs = dict(
                    query_text=query_text,
                    query_type=query_type,
                    datasets=datasets,
                    session_id=project_id or None,
                    top_k=8,
                    only_context=True,
                    feedback_influence=0.35,
                )
                op.set("cache_state", "bypass" if bypass_cache else "miss")
                t0 = time.perf_counter()
                # `memory.recall` is already the internal Cognee operation.
                # The caller's ContextComposer owns the single end-to-end
                # `student_memory_recall` stage, avoiding duplicate samples.
                results = await self._recall_with_fallback(cognee, kwargs)
                elapsed = time.perf_counter() - t0
                if elapsed >= COGNEE_SLOW_OPERATION_SECONDS:
                    logger.warning("Cognee recall for %s mode=%s took %.2fs", project_id or "global", mode, elapsed)
                if not results:
                    op.set("memory_liveness", "alive")
                    op.mark_terminal("success_empty")
                    if mode == "temporal" and project_id:
                        return self._query_temporal_ledger(project_id, topic)
                    return ""
                chunks = [_clip_text(_extract_text(r), MAX_RECALL_CHUNK_CHARS) for r in results if r]
                chunks = [c for c in chunks if c]
                if not chunks:
                    op.set("memory_liveness", "alive")
                    op.mark_terminal("success_empty")
                    if mode == "temporal" and project_id:
                        return self._query_temporal_ledger(project_id, topic)
                    return ""
                self._record_memory_operation(project_id, "recall", True, metadata={"mode": mode, "duration_seconds": round(elapsed, 3)})
                op.set("memory_liveness", "alive")
                context = "\n".join(chunks[:5])
                rendered = f'Prior {mode} memory for "{topic}":\n' + _clip_text(context, MAX_RECALL_CONTEXT_CHARS)
                if mode in {"profile", "concept"}:
                    StudentMemoryService._profile_recall_cache[cache_key] = (time.time(), rendered)
                return rendered
            except Exception as exc:  # noqa: BLE001
                msg = str(exc)
                if "No data found in the system" in msg or "DatabaseNotCreatedError" in msg:
                    # A fresh local profile is an expected empty state, not a
                    # failed Cognee fallback.
                    op.set("memory_liveness", "empty")
                    op.mark_terminal("success_empty")
                    return ""
                # A Cognee recall failure is a visible, handled degradation;
                # the canonical observer exports only its fixed safe summary.
                op.record_exception(
                    exc,
                    escaped=False,
                    stage="student_memory_recall",
                    category="cognee_recall_fallback",
                )
                if mode == "temporal" and project_id:
                    fallback = self._query_temporal_ledger(project_id, topic)
                    if fallback:
                        self._record_memory_operation(
                            project_id,
                            "recall_temporal_fallback",
                            True,
                            metadata={"native_error_type": type(exc).__name__},
                        )
                        op.set("memory_liveness", "degraded")
                        op.mark_terminal("degraded")
                        return fallback
                logger.warning("query_prior_knowledge failed: %s", type(exc).__name__)
                self._record_memory_operation(project_id, "recall", False, type(exc).__name__)
                op.set("memory_liveness", "degraded")
                op.mark_terminal("degraded")
                return ""

    def _datasets_for_mode(self, project_id: str, mode: MemoryMode) -> list[str]:
        if mode == "profile":
            return [PROFILE_DATASET]
        if mode == "project":
            return []
        return [PROFILE_DATASET]

    def _query_for_mode(self, topic: str, mode: MemoryMode) -> str:
        if mode == "profile":
            return f"student profile preferred name call me persona preferences learning style recurring strengths weaknesses {topic}"
        if mode == "project":
            return f"project research memory claims methods open questions ideas {topic}"
        if mode == "temporal":
            return f"how has the student's understanding of {topic} changed over time across sessions"
        return f"student personal details struggles weaknesses mastery concept understanding {topic}"

    async def _recall_with_fallback(self, cognee: Any, kwargs: dict[str, Any]) -> list[Any]:
        attempts = [
            kwargs,
            {k: v for k, v in kwargs.items() if k != "only_context"},
            {k: v for k, v in kwargs.items() if k not in {"only_context", "feedback_influence"}},
        ]
        last_exc: Exception | None = None
        for attempt in attempts:
            try:
                return await _maybe_await(cognee.recall(**attempt))
            except TypeError as exc:
                last_exc = exc
                continue
        if last_exc:
            raise last_exc
        return []

    async def record_feedback(
        self,
        project_id: str,
        qa_id: str,
        score: int,
        text: str,
        node_ids: list[str] | None = None,
        edge_ids: list[str] | None = None,
        cognee_native: bool = False,
    ) -> dict[str, bool]:
        result: dict[str, bool] = {"feedback": False, "frequency_weights": False}
        if not cognee_native or not qa_id:
            result["skipped"] = True
            return result
        try:
            import cognee

            session_api = getattr(cognee, "session", None)
            add_feedback = getattr(session_api, "add_feedback", None)
            if add_feedback is not None:
                result["feedback"] = bool(await _maybe_await(add_feedback(
                    session_id=project_id,
                    qa_id=qa_id,
                    feedback_score=score,
                    feedback_text=text,
                )))
            add_weights = getattr(session_api, "add_frequency_weights", None)
            if add_weights is not None and (node_ids or edge_ids):
                result["frequency_weights"] = bool(await _maybe_await(add_weights(
                    session_id=project_id,
                    qa_id=qa_id,
                    node_ids=node_ids or [],
                    edge_ids=edge_ids or [],
                )))
        except Exception as exc:  # noqa: BLE001
            logger.warning("Cognee feedback failed for %s/%s: %s", project_id, qa_id, exc)
        return result

    async def record_style_feedback(self, project_id: str, text: str) -> dict[str, bool]:
        if not text:
            return {"profile_memory": False}
        statement = f"Student style preference: {text}"
        candidate = MemoryCandidate(
            candidate_id=make_candidate_id("student", project_id, "style_preference", statement),
            destination="student",
            project_id=project_id,
            kind="style_preference",
            statement=statement,
            attribution="explicit_student",
            confidence=1.0,
            interaction_ids=[],
            evidence_ids=[],
        )
        decision = MemoryPromotionGate().evaluate_student(candidate)
        if not decision.promote:
            self.record_temporal_event(project_id, "style_feedback_rejected", statement)
            return {"profile_memory": False}
        ok = await self.stage_promoted_candidate(candidate)
        self.record_temporal_event(project_id, "style_feedback", statement)
        return {"profile_memory": ok}

    async def query_global_curriculum_connections(self, term: str) -> str:
        return await self.query_prior_knowledge(
            f"cross-project interests and prior familiarity related to {term}",
            mode="profile",
        )

    async def get_daily_review(self) -> list[str]:
        context = await self.query_prior_knowledge(
            "concepts mastered more than 7 days ago and due for review",
            mode="temporal",
        )
        if not context:
            return []
        return [line for line in context.splitlines() if line.strip()]

    async def memory_liveness(self, project_id: str, force: bool = False) -> dict[str, Any]:
        recent = self._recent_operation_status(project_id).get("operations", {}).get("liveness", {})
        if not force and recent and time.time() - float(recent.get("updated_at") or 0) < 120:
            checks = recent.get("metadata", {}).get("checks", {})
            return {
                "project_id": project_id,
                "state": "alive" if recent.get("ok") else "degraded",
                "checks": checks,
                "last_error": recent.get("error", ""),
                "recent_operations": self._recent_operation_status(project_id).get("operations", {}),
                "cached": True,
            }
        checks = {"dataset": False, "recall": False}
        last_error = ""
        dataset = PROFILE_DATASET
        try:
            cognee = await self._cognee()
            from cognee import SearchType

            checks["dataset"] = await self.ensure_profile_dataset()
            if not checks["dataset"]:
                raise RuntimeError("student profile dataset bootstrap failed")

            results = await _maybe_await(cognee.recall(
                query_text="student preferences and learning profile",
                query_type=SearchType.GRAPH_COMPLETION,
                datasets=[dataset],
                only_context=True,
                top_k=3,
            ))
            # Empty profile memory is a healthy fresh state if recall itself
            # completed without raising.
            checks["recall"] = True
        except Exception as exc:  # noqa: BLE001
            last_error = str(exc)

        state = "alive" if all(checks.values()) else "degraded"
        if not any(checks.values()):
            state = "unavailable"
        self._record_memory_operation(project_id, "liveness", state == "alive", last_error, {"checks": checks})
        if state != "alive":
            self._activity(project_id, "memory_liveness", f"Cognee memory is {state}: {last_error or 'incomplete probe'}", status="warning")
        return {
            "project_id": project_id,
            "state": state,
            "checks": checks,
            "last_error": last_error,
            "recent_operations": self._recent_operation_status(project_id).get("operations", {}),
        }

    async def memory_status(self, project_id: str = "", include_liveness: bool = False) -> dict[str, Any]:
        status: dict[str, Any] = {
            "project_id": project_id,
            "state": "unknown",
            "profile_dataset": False,
            "project_dataset": False,
            "project_memory_backend": "chroma",
            "student_memory_backend": "cognee",
            "pending_count": len(self.list_pending_memory(project_id)) if project_id else 0,
            "inventory_ok": False,
            "provenance_ok": False,
            "export_ok": False,
            "last_error": "",
        }
        try:
            import cognee

            status["profile_dataset"] = await self._dataset_exists(cognee, PROFILE_DATASET)
            status["project_dataset"] = ProjectMemoryStore().count_project(project_id) > 0 if project_id else False
            status["project_memory_count"] = ProjectMemoryStore().count_project(project_id) if project_id else 0
            inventory = await self.get_schema_inventory(None)
            provenance = await self.get_provenance(None)
            exported = await self.export_memory(None)
            status["inventory_ok"] = bool(inventory.get("ok"))
            status["provenance_ok"] = bool(provenance.get("ok"))
            status["export_ok"] = bool(exported.get("ok"))
            status["liveness"] = await self.memory_liveness(project_id) if include_liveness and project_id else {}
            items = inventory.get("items") if isinstance(inventory, dict) else None
            has_items = bool(items)
            if status["profile_dataset"] or status["project_dataset"]:
                live_state = status["liveness"].get("state") if isinstance(status.get("liveness"), dict) else ""
                if live_state in {"degraded", "unavailable"}:
                    status["state"] = live_state
                else:
                    status["state"] = "ready" if has_items else "empty"
            else:
                status["state"] = "empty"
            return status
        except Exception as exc:  # noqa: BLE001
            status["state"] = "unavailable"
            status["last_error"] = str(exc)
            return status

    # ------------------------------------------------------------------ #
    # Native inspection / ontology / deletion                             #
    # ------------------------------------------------------------------ #

    async def run_project_memify(self, project_id: str) -> dict[str, Any]:
        return {"ok": False, "available": False, "error": "Project memory is Chroma-backed; Cognee memify applies only to student memory."}

    async def get_schema_inventory(self, project_id: str | None = None) -> dict[str, Any]:
        try:
            import cognee

            dataset = PROFILE_DATASET
            result = await _maybe_await(cognee.get_schema_inventory(dataset=dataset))
            return {"ok": True, "dataset": dataset, "items": _serialise(result)}
        except Exception as exc:  # noqa: BLE001
            return {"ok": False, "error": str(exc), "items": []}

    async def get_provenance(self, project_id: str | None = None) -> dict[str, Any]:
        try:
            import cognee

            nodes, edges = await _maybe_await(cognee.get_memory_provenance_graph(include_memory=True))
            return {"ok": True, "project_id": project_id, "nodes": _serialise(nodes), "edges": _serialise(edges)}
        except Exception as exc:  # noqa: BLE001
            return {"ok": False, "error": str(exc), "nodes": [], "edges": []}

    async def export_memory(self, project_id: str | None = None, include_profile: bool = True) -> dict[str, Any]:
        datasets = []
        if include_profile:
            datasets.append(PROFILE_DATASET)
        if not datasets:
            datasets.append(PROFILE_DATASET)

        out: dict[str, Any] = {}
        try:
            import cognee

            for dataset in datasets:
                try:
                    out[dataset] = _serialise(await _maybe_await(cognee.export(dataset=dataset)))
                except Exception as exc:  # noqa: BLE001
                    out[dataset] = {"ok": False, "error": str(exc)}
            return {"ok": True, "datasets": out}
        except Exception as exc:  # noqa: BLE001
            return {"ok": False, "error": str(exc), "datasets": out}

    async def visualize_memory_graph(self, scope: str = "profile", project_id: str = "") -> str | None:
        try:
            import cognee

            if scope == "project":
                return None
            dataset = PROFILE_DATASET
            for fn_name in ("visualize_graph", "visualize"):
                fn = getattr(cognee, fn_name, None)
                if fn is None:
                    continue
                try:
                    return await _maybe_await(fn(dataset=dataset))
                except TypeError:
                    return await _maybe_await(fn())
        except Exception as exc:  # noqa: BLE001
            logger.warning("Cognee graph visualization failed: %s", exc)
        return None

    async def reset_project_memory(self, project_id: str) -> dict[str, Any]:
        try:
            ProjectMemoryStore().delete_project(project_id)
            try:
                self._source_registry_path(project_id).unlink(missing_ok=True)
            except Exception:
                pass
            self._record_memory_operation(project_id, "reset", True)
            self._activity(project_id, "memory_reset", "Project memory reset after material change")
            return {"ok": True, "backend": "chroma"}
        except Exception as exc:  # noqa: BLE001
            logger.warning("Project memory reset failed for %s: %s", project_id, exc)
            self._record_memory_operation(project_id, "reset", False, str(exc))
            self._activity(project_id, "memory_reset", "Project memory reset failed", status="warning", metadata={"error": str(exc)})
            return {"ok": False, "error": str(exc)}

    async def rebuild_project_memory(self, project_id: str) -> dict[str, Any]:
        try:
            from app.services.project_service import ProjectService

            project = ProjectService().load(project_id)
            if project is None:
                self._record_memory_operation(project_id, "rebuild", True, metadata={"skipped": "project not found"})
                return {"ok": True, "files": [], "skipped": "project not found"}
            filenames = [file.filename for file in project.files]
            # Paper facts are rebuilt by EvidenceIngestionService. Project memory
            # contains only student/research observations, so it starts empty.
            ok = True
            self._record_memory_operation(project_id, "rebuild", ok, "" if ok else "Project memory rebuild failed")
            self._activity(project_id, "memory_rebuild", "Project memory rebuilt from current material" if ok else "Project memory rebuild failed", status="ok" if ok else "warning")
            return {"ok": ok, "files": filenames}
        except Exception as exc:  # noqa: BLE001
            logger.warning("Project memory rebuild failed for %s: %s", project_id, exc)
            self._record_memory_operation(project_id, "rebuild", False, str(exc))
            self._activity(project_id, "memory_rebuild", "Project memory rebuild failed", status="warning", metadata={"error": str(exc)})
            return {"ok": False, "error": str(exc)}

    async def forget_project_document(self, project_id: str, document_id: str) -> dict[str, Any]:
        try:
            from app.rag.evidence_ingestion import EvidenceIngestionService

            EvidenceIngestionService().remove_document(project_id, document_id)
            return {"ok": True, "backend": "sqlite+chroma"}
        except Exception as exc:  # noqa: BLE001
            logger.warning("Cognee forget failed for %s/%s: %s", project_id, document_id, exc)
            self._record_memory_operation(project_id, "forget", False, str(exc))
            return {"ok": False, "error": str(exc)}

    # ------------------------------------------------------------------ #
    # Controlled pilots                                                   #
    # ------------------------------------------------------------------ #

    def agent_memory_decorator(self, project_id: str, memory_query: str = ""):
        del project_id, memory_query
        return None

    async def run_feynman_agent_memory_pilot(
        self,
        project_id: str,
        node_label: str,
        student_answer: str,
    ) -> dict[str, Any]:
        del project_id, node_label, student_answer
        return {"ok": False, "available": False, "error": "Project turns are stored in Chroma; Cognee agent-memory project pilot is disabled."}

    async def bootstrap_project_skill(self, project_id: str) -> dict[str, Any]:
        del project_id
        return {"ok": False, "available": False, "error": "Project review skills are disabled for the Chroma project-memory scope."}

    async def propose_skill_evolution(self, project_id: str, skill_run: Any = None) -> dict[str, Any]:
        del skill_run
        return {
            "ok": False,
            "available": False,
            "project_id": project_id,
            "error": "Cognee skill evolution is not used for Chroma-backed project memory.",
        }

    async def review_project_memory(self, project_id: str) -> dict[str, Any]:
        return {
            "ok": True,
            "available": False,
            "backend": "chroma",
            "project_id": project_id,
            "memory_count": ProjectMemoryStore().count_project(project_id),
            "message": "Project memory review is performed through grounded retrieval, not a Cognee project graph.",
        }