| """Governed proposal workflow for narrative patient memory."""
|
|
|
| from __future__ import annotations
|
|
|
| import datetime
|
| import uuid
|
| from dataclasses import asdict, dataclass
|
| from typing import Any, Dict, Optional
|
|
|
| from src.tools.patient_memory import save_patient_memory
|
|
|
|
|
| import re
|
|
|
| ALLOWED_CATEGORIES = {"glucose_history", "medications", "diet"}
|
| MAX_VALUE_LENGTH = 2000
|
|
|
|
|
| FORBIDDEN_CONTENT_PATTERNS = [
|
| r"ignore previous instructions",
|
| r"system prompt",
|
| r"api[_-]?key",
|
| r"password\s*=",
|
| r"secret\s*=",
|
| r"diagnosis:\s*definitive",
|
| ]
|
|
|
|
|
| @dataclass
|
| class MemoryProposal:
|
| proposal_id: str
|
| patient_id: str
|
| category: str
|
| value: str
|
| source: str
|
| confidence: Optional[float]
|
| status: str
|
| created_at: str
|
| expires_at: Optional[str] = None
|
| session_id: Optional[str] = None
|
| trace_id: Optional[str] = None
|
|
|
| def as_dict(self) -> Dict[str, Any]:
|
| return asdict(self)
|
|
|
|
|
| class MemoryService:
|
| """Create and commit memory only through explicit policy decisions."""
|
|
|
| def propose(
|
| self,
|
| *,
|
| patient_id: str,
|
| category: str,
|
| value: str,
|
| source: str,
|
| confidence: Optional[float] = None,
|
| ttl_seconds: Optional[int] = 86400 * 30,
|
| session_id: Optional[str] = None,
|
| trace_id: Optional[str] = None,
|
| ) -> MemoryProposal:
|
| if not patient_id:
|
| raise ValueError("patient_id is required")
|
| if category not in ALLOWED_CATEGORIES:
|
| raise ValueError(f"unsupported memory category: {category}")
|
| if not isinstance(value, str) or not value.strip():
|
| raise ValueError("memory value must be non-empty text")
|
| if len(value) > MAX_VALUE_LENGTH:
|
| raise ValueError("memory value exceeds the maximum length")
|
| if confidence is not None and not 0.0 <= confidence <= 1.0:
|
| raise ValueError("confidence must be between 0.0 and 1.0")
|
|
|
|
|
| normalized = value.lower()
|
| for pattern in FORBIDDEN_CONTENT_PATTERNS:
|
| if re.search(pattern, normalized):
|
| raise ValueError(f"memory value contains forbidden content pattern: {pattern}")
|
|
|
| now = datetime.datetime.now(datetime.timezone.utc)
|
| expires_at = (now + datetime.timedelta(seconds=ttl_seconds)).isoformat() if ttl_seconds else None
|
|
|
| return MemoryProposal(
|
| proposal_id=str(uuid.uuid4()),
|
| patient_id=patient_id,
|
| category=category,
|
| value=value.strip(),
|
| source=source,
|
| confidence=confidence,
|
| status="pending_consent",
|
| created_at=now.isoformat(),
|
| expires_at=expires_at,
|
| session_id=session_id,
|
| trace_id=trace_id,
|
| )
|
|
|
| def commit(self, proposal: MemoryProposal, *, consent_granted: bool) -> Dict[str, Any]:
|
| if proposal.status != "pending_consent":
|
| raise ValueError(f"proposal is not pending consent: {proposal.proposal_id}")
|
| if not consent_granted:
|
| proposal.status = "rejected"
|
| return {"proposal_id": proposal.proposal_id, "status": proposal.status}
|
|
|
| result = save_patient_memory.invoke({proposal.category: proposal.value, "patient_id": proposal.patient_id})
|
| proposal.status = "committed"
|
| return {
|
| "proposal_id": proposal.proposal_id,
|
| "status": proposal.status,
|
| "result": result,
|
| "source": proposal.source,
|
| "committed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
| }
|
|
|
| @staticmethod
|
| def resolve_conflicts(proposals: list[Dict[str, Any] | MemoryProposal]) -> list[Dict[str, Any]]:
|
| """
|
| Deterministically resolve memory proposal conflicts by category and timestamp recency.
|
| Retains the latest proposal for each category.
|
| """
|
| latest_by_category: Dict[str, Dict[str, Any]] = {}
|
| for p in proposals:
|
| p_dict = p.as_dict() if isinstance(p, MemoryProposal) else dict(p)
|
| cat = p_dict.get("category")
|
| if not cat:
|
| continue
|
| existing = latest_by_category.get(cat)
|
| if existing is None or p_dict.get("created_at", "") >= existing.get("created_at", ""):
|
| latest_by_category[cat] = p_dict
|
| return list(latest_by_category.values())
|
