"""Adapters from CommitAdaptationReview proposals to the real, strict MemoryCandidate / ProjectMemoryRecord models. The LLM never authors candidate_id, recurrence_key, novelty_key, or authoritative attribution — all of that is derived here in application code. See docs/commit-memory-adaptation-architecture.md for the full rationale. """ from __future__ import annotations from dataclasses import dataclass from typing import Literal from app.rag.models import normalize_evidence_text from app.rag.project_memory import ProjectMemoryRecord, make_project_memory_id from app.schemas.commit_adaptation import MemoryProposal from app.services.commit_signal_patterns import is_explicit_global_signal, is_explicit_student_signal from app.services.memory_candidates import MemoryCandidate, make_candidate_id @dataclass(frozen=True) class StudentInteraction: interaction_id: str timestamp: float actor: Literal["student", "assistant"] source: str agent_id: str text: str def _normalize_for_quote_match(text: str) -> str: # normalize_evidence_text() does NFKC + whitespace collapse, but not # casefold or smart-quote mapping — layer those on top rather than # duplicating the whole normalizer. normalized = normalize_evidence_text(text) normalized = normalized.replace("‘", "'").replace("’", "'") normalized = normalized.replace("“", '"').replace("”", '"') return normalized.casefold() def proposal_to_memory_candidate( proposal: MemoryProposal, *, project_id: str, interactions_by_id: dict[str, StudentInteraction], ) -> MemoryCandidate: verified_quotes = [] for evidence in proposal.evidence_quotes: interaction = interactions_by_id.get(evidence.interaction_id) if interaction is None or interaction.actor != "student": continue if _normalize_for_quote_match(evidence.quote) not in _normalize_for_quote_match(interaction.text): continue verified_quotes.append(evidence) verified_interaction_ids = sorted({evidence.interaction_id for evidence in verified_quotes}) explicitly_global = any(is_explicit_global_signal(evidence.quote) for evidence in verified_quotes) # A global-scoped signal ("I generally prefer...") is a stronger form of # explicit wording, not a separate category — it must also count toward # `explicit`, or global-but-not-matching-the-narrower-explicit-patterns # statements fall through to the weakest, unattributed bucket. explicit = explicitly_global or any(is_explicit_student_signal(evidence.quote) for evidence in verified_quotes) destination = proposal.destination if destination == "student" and explicit and not explicitly_global: destination = "project" if explicit: attribution = "explicit_student" elif destination == "project": attribution = "idea_observer_interaction" else: attribution = "idea_observer_profile_proposal" return MemoryCandidate( candidate_id=make_candidate_id(destination, project_id, proposal.kind, proposal.statement), destination=destination, project_id=project_id, kind=proposal.kind, statement=proposal.statement, attribution=attribution, confidence=proposal.confidence, interaction_ids=verified_interaction_ids, evidence_ids=[], ) def candidate_to_project_memory_record(candidate: MemoryCandidate) -> ProjectMemoryRecord: if candidate.destination != "project": raise ValueError(f"Expected destination='project', got {candidate.destination!r}") return ProjectMemoryRecord( memory_id=make_project_memory_id( candidate.project_id, candidate.kind, candidate.recurrence_key, candidate.statement, ), project_id=candidate.project_id, kind=candidate.kind, statement=candidate.statement, novelty_key=candidate.recurrence_key, attribution=candidate.attribution, confidence=candidate.confidence, evidence_ids=candidate.evidence_ids, interaction_ids=candidate.interaction_ids, )