File size: 4,698 Bytes
cb505ff | 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 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """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
# Patterns that indicate forbidden memory content (system instructions, secrets, diagnostic claims)
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, # 30 day default TTL
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")
# Validate against prompt injection, secrets, or raw diagnostic claims
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())
|