Spaces:
Sleeping
Sleeping
| """Orchestrates one Commit's adaptation review: reads unreviewed student | |
| interactions and already-approved staged profile candidates, makes one | |
| Cerebras call, routes proposals through MemoryPromotionGate, persists | |
| approved candidates, and updates the persona only after Cognee confirms | |
| durability. See docs/commit-memory-adaptation-architecture.md for the full | |
| design and crash-safety rationale. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import uuid | |
| from collections import defaultdict | |
| from app.rag.project_memory import ProjectMemoryStore | |
| from app.services import approved_profile_staging, pending_review_store, student_interaction_log | |
| from app.services.agent_control import AgentControlService | |
| from app.services.commit_candidate_adapters import candidate_to_project_memory_record, proposal_to_memory_candidate | |
| from app.services.memory_promotion import MemoryPromotionGate | |
| from app.services.student_memory import StudentMemoryService | |
| MAX_INTERACTIONS_CHARS = 8_000 | |
| MAX_PROFILE_CONTEXT_CHARS = 6_000 | |
| MAX_PERSONA_CHARS = 3_000 | |
| # Module-level singletons so tests can monkeypatch them and so the | |
| # per-project lock registry is shared across calls within one process. | |
| agent_control = AgentControlService() | |
| agent_control_persona = agent_control | |
| student_memory = StudentMemoryService() | |
| project_memory = ProjectMemoryStore() | |
| memory_gate = MemoryPromotionGate() | |
| _adaptation_locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock) | |
| def _clip_text(text: str, limit: int) -> str: | |
| if len(text) <= limit: | |
| return text | |
| return text[:limit].rstrip() + "\n...[truncated]" | |
| def _format_interactions(interactions) -> str: | |
| return "\n".join(f"- [{i.interaction_id}] {i.text}" for i in interactions) | |
| def adaptation_lock(project_id: str) -> asyncio.Lock: | |
| return _adaptation_locks[project_id] | |
| def _gate_and_split(candidates): | |
| promoted_project, promoted_student = [], [] | |
| for candidate in candidates: | |
| decision = ( | |
| memory_gate.evaluate_project(candidate) if candidate.destination == "project" | |
| else memory_gate.evaluate_student(candidate) | |
| ) | |
| if not decision.promote: | |
| continue | |
| (promoted_project if candidate.destination == "project" else promoted_student).append(candidate) | |
| return promoted_project, promoted_student | |
| async def run_commit_adaptation(*, project_id: str) -> None: | |
| async with adaptation_lock(project_id): | |
| pending = pending_review_store.load_pending_review(project_id) | |
| if pending is not None and pending.status == "cognee_flush_started": | |
| # A prior run crashed (or the process restarted) between marking | |
| # this batch as "Cognee may be mutating state" and confirming | |
| # the outcome. We cannot prove Cognee's write was safe to retry | |
| # -- quarantine it and move on rather than risk a duplicate. | |
| pending_review_store.archive_as_uncertain(project_id, pending) | |
| student_interaction_log.advance_cursor(project_id, pending.interaction_through_offset) | |
| approved_profile_staging.advance_cursor(project_id, pending.profile_staging_through_offset) | |
| pending_review_store.delete_pending_review(project_id) | |
| return | |
| if pending is None: | |
| interactions, i_from, i_through = student_interaction_log.read_pending(project_id) | |
| staged, s_from, s_through = approved_profile_staging.read_pending(project_id) | |
| if not interactions and not staged: | |
| return | |
| current_profile = await student_memory.query_prior_knowledge( | |
| "student preferences, knowledge, strengths, recurring difficulties, and goals", | |
| project_id=project_id, | |
| mode="profile", | |
| ) | |
| review = await agent_control.review_commit_with_llm( | |
| project_id=project_id, | |
| interactions_text=_clip_text(_format_interactions(interactions), MAX_INTERACTIONS_CHARS), | |
| current_profile=_clip_text(current_profile, MAX_PROFILE_CONTEXT_CHARS), | |
| current_persona=_clip_text(agent_control_persona.get_inferred_text(project_id), MAX_PERSONA_CHARS), | |
| manual_persona=_clip_text(agent_control_persona.get_manual_text(project_id), MAX_PERSONA_CHARS), | |
| ) | |
| interaction_index = {i.interaction_id: i for i in interactions} | |
| review_candidates = [ | |
| proposal_to_memory_candidate(p, project_id=project_id, interactions_by_id=interaction_index) | |
| for p in review.memory_proposals | |
| ] | |
| promoted_project, promoted_student_from_review = _gate_and_split(review_candidates) | |
| # Already-approved staged candidates bypass the gate a second | |
| # time -- they produced their one promotion-decision ledger row | |
| # at capture time (see explicit_profile_capture.py). | |
| promoted_student_by_id = {c.candidate_id: c for c in [*promoted_student_from_review, *staged]} | |
| promoted_student = list(promoted_student_by_id.values()) | |
| pending_review_store.save_pending_review( | |
| project_id, status="ready_to_flush", | |
| candidate_ids=[c.candidate_id for c in [*promoted_project, *promoted_student]], | |
| interaction_from_offset=i_from, interaction_through_offset=i_through, | |
| profile_staging_from_offset=s_from, profile_staging_through_offset=s_through, | |
| review=review, | |
| ) | |
| pending = pending_review_store.load_pending_review(project_id) | |
| else: | |
| # Retry: re-derive candidates from the exact stored ranges and | |
| # the exact stored review text -- never a freshly re-proposed | |
| # or silently-expanded batch. Chroma/Cognee IDs stay stable. | |
| interactions = student_interaction_log.read_range( | |
| project_id, pending.interaction_from_offset, pending.interaction_through_offset, | |
| ) | |
| staged = approved_profile_staging.read_range( | |
| project_id, pending.profile_staging_from_offset, pending.profile_staging_through_offset, | |
| ) | |
| interaction_index = {i.interaction_id: i for i in interactions} | |
| review_candidates = [ | |
| proposal_to_memory_candidate(p, project_id=project_id, interactions_by_id=interaction_index) | |
| for p in pending.review.memory_proposals | |
| ] | |
| candidate_ids = set(pending.candidate_ids) | |
| promoted_project = [c for c in review_candidates if c.destination == "project" and c.candidate_id in candidate_ids] | |
| promoted_student_from_review = [ | |
| c for c in review_candidates if c.destination == "student" and c.candidate_id in candidate_ids | |
| ] | |
| promoted_student_by_id = {c.candidate_id: c for c in [*promoted_student_from_review, *staged]} | |
| promoted_student = [c for c in promoted_student_by_id.values() if c.candidate_id in candidate_ids] | |
| pending.promoted_project = promoted_project | |
| pending.promoted_student = promoted_student | |
| loop = asyncio.get_running_loop() | |
| project_records = [candidate_to_project_memory_record(c) for c in pending.promoted_project] | |
| if project_records: | |
| try: | |
| written = await loop.run_in_executor(None, project_memory.upsert, project_records) | |
| except Exception: | |
| return | |
| if written != len(project_records): | |
| return | |
| if pending.promoted_student: | |
| pending_review_store.update_pending_status( | |
| project_id, status="cognee_flush_started", flush_attempt_id=str(uuid.uuid4()), | |
| ) | |
| try: | |
| for candidate in pending.promoted_student: | |
| if not await student_memory.stage_promoted_candidate(candidate): | |
| pending_review_store.update_pending_status(project_id, status="cognee_flush_uncertain") | |
| return | |
| flush_results = await student_memory.flush_project(project_id) | |
| except Exception: | |
| pending_review_store.update_pending_status(project_id, status="cognee_flush_uncertain") | |
| return | |
| # A clean False here is still a caught exception inside | |
| # StudentMemoryService's flush pipeline -- it cannot be | |
| # distinguished from "Cognee partially wrote before failing." | |
| # Treat identically to the exception branch above. | |
| if flush_results.get("research_profile") is not True: | |
| pending_review_store.update_pending_status(project_id, status="cognee_flush_uncertain") | |
| return | |
| pending_review_store.update_pending_status(project_id, status="cognee_flush_succeeded") | |
| agent_control_persona.set_inferred_text(project_id, pending.review.inferred_persona_text) | |
| student_interaction_log.advance_cursor(project_id, pending.interaction_through_offset) | |
| approved_profile_staging.advance_cursor(project_id, pending.profile_staging_through_offset) | |
| pending_review_store.delete_pending_review(project_id) | |