Spaces:
Sleeping
Sleeping
| """Deterministic memory promotion gates with an append-only decision ledger.""" | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import threading | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| from app.services.memory_candidates import MemoryCandidate | |
| class PromotionDecision: | |
| promote: bool | |
| reason: str | |
| supporting_projects: tuple[str, ...] = () | |
| class MemoryPromotionGate: | |
| _lock = threading.Lock() | |
| def __init__(self, ledger_path: Path | str | None = None) -> None: | |
| self.path = Path(ledger_path) if ledger_path else Path( | |
| os.path.expanduser("~/.studybuddy/memory_integrity/promotion_decisions.jsonl") | |
| ) | |
| self.path.parent.mkdir(parents=True, exist_ok=True) | |
| def evaluate_project(self, candidate: MemoryCandidate) -> PromotionDecision: | |
| if candidate.destination != "project": | |
| decision = PromotionDecision(False, "wrong_destination") | |
| elif candidate.attribution == "idea_observer_interaction" and not candidate.interaction_ids: | |
| decision = PromotionDecision(False, "assistant_only_claim") | |
| elif candidate.evidence_ids and not all(item.startswith("ev_") for item in candidate.evidence_ids): | |
| decision = PromotionDecision(False, "invalid_evidence_identity") | |
| elif not candidate.statement.strip(): | |
| decision = PromotionDecision(False, "empty_statement") | |
| else: | |
| decision = PromotionDecision(True, "grounded_project_observation", (candidate.project_id,)) | |
| self._record(candidate, decision) | |
| return decision | |
| def evaluate_student(self, candidate: MemoryCandidate) -> PromotionDecision: | |
| if candidate.destination != "student": | |
| decision = PromotionDecision(False, "wrong_destination") | |
| self._record(candidate, decision) | |
| return decision | |
| with self._lock: | |
| projects = self._supporting_projects(candidate.recurrence_key) | |
| projects.add(candidate.project_id) | |
| if candidate.attribution == "explicit_student": | |
| decision = PromotionDecision(True, "explicit_student_signal", tuple(sorted(projects))) | |
| elif len(projects) >= 2 and candidate.confidence >= 0.75: | |
| decision = PromotionDecision(True, "cross_project_recurrence", tuple(sorted(projects))) | |
| else: | |
| decision = PromotionDecision(False, "requires_cross_project_recurrence", tuple(sorted(projects))) | |
| self._record_unlocked(candidate, decision) | |
| return decision | |
| def _supporting_projects(self, recurrence_key: str) -> set[str]: | |
| if not self.path.exists(): | |
| return set() | |
| projects: set[str] = set() | |
| try: | |
| with self.path.open(encoding="utf-8") as handle: | |
| for line in handle: | |
| try: | |
| row = json.loads(line) | |
| except (ValueError, json.JSONDecodeError): | |
| continue | |
| if row.get("recurrence_key") == recurrence_key: | |
| project_id = str(row.get("project_id") or "") | |
| if project_id: | |
| projects.add(project_id) | |
| except OSError: | |
| return set() | |
| return projects | |
| def _record(self, candidate: MemoryCandidate, decision: PromotionDecision) -> None: | |
| with self._lock: | |
| self._record_unlocked(candidate, decision) | |
| def _record_unlocked(self, candidate: MemoryCandidate, decision: PromotionDecision) -> None: | |
| row = { | |
| "candidate_id": candidate.candidate_id, | |
| "destination": candidate.destination, | |
| "project_id": candidate.project_id, | |
| "kind": candidate.kind, | |
| "attribution": candidate.attribution, | |
| "confidence": candidate.confidence, | |
| "evidence_ids": candidate.evidence_ids, | |
| "interaction_ids": candidate.interaction_ids, | |
| "recurrence_key": candidate.recurrence_key, | |
| "decision": "promote" if decision.promote else "reject", | |
| "reason": decision.reason, | |
| "supporting_projects": list(decision.supporting_projects), | |
| "observed_at": candidate.observed_at.isoformat(), | |
| } | |
| try: | |
| with self.path.open("a", encoding="utf-8") as handle: | |
| handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n") | |
| except OSError: | |
| # Memory diagnostics are load-bearing evidence, but a read-only or | |
| # temporarily unavailable ledger must not break the study request. | |
| return | |