"""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.", }