Spaces:
Sleeping
Sleeping
File size: 4,691 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """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
@dataclass(frozen=True)
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
|