"""Memory and background-housekeeping evidence adapter.""" from __future__ import annotations import hashlib import json from datetime import datetime, timezone from pathlib import Path from typing import Any, Mapping from .memory_refs import canonical_path_hash, memory_ref_id from .store import EvidenceStore from .types import EvidenceEvent, ResourceRef _MEMORY_EVENT_TYPES = { "memory_written", "memory_prefetch_consumed", "nested_memory_consumed", "memory_saved", "memory_logged", "session_memory_extraction_coalesced", "session_memory_extraction_trailing_start", "session_memory_extraction_start", "session_memory_extraction_complete", "session_memory_extraction_error", "session_memory_extraction_skipped", "session_memory_updated", "session_memory_checked", "memory_extraction_coalesced", "memory_extraction_start", "memory_extraction_complete", "memory_extraction_error", "memory_extraction_skipped", "memory_extraction_trailing_start", "auto_dream_start", "auto_dream_progress", "auto_dream_complete", "auto_dream_cancelled", "auto_dream_error", "auto_dream_skipped", "manual_dream_start", "manual_dream_progress", "manual_dream_complete", "manual_dream_cancelled", "manual_dream_error", "manual_dream_skipped", } class MemoryEvidenceAdapter: """Translate memory visibility/write/background facts into evidence.""" def __init__(self, store: EvidenceStore) -> None: self._store = store def build_event( self, event_type: str, data: Mapping[str, Any], ) -> EvidenceEvent | None: if event_type not in _MEMORY_EVENT_TYPES: return None session_id = _none_or_str(data.get("session_id")) task_id = _none_or_str(data.get("task_id")) parent_task_id = _none_or_str(data.get("parent_task_id")) agent_id = _none_or_str(data.get("agent_id")) created_at = _utc_now() refs = _memory_refs(event_type, data, created_at=created_at) background_ref = _background_task_ref(event_type, data, created_at=created_at) if background_ref is not None: refs.append(background_ref) if not refs: if not _allow_synthetic_memory_ref(event_type): return None refs = [ ResourceRef( ref_id=f"memory_ref:{event_type}:{_digest(data)}", ref_type="memory_ref", session_id=session_id, task_id=task_id, parent_task_id=parent_task_id, agent_id=agent_id, producer="memory", created_at=created_at, reliability="derived", role="supporting", preview=event_type, metadata=_metadata_without_content(data, event_type=event_type), ) ] return EvidenceEvent.create( event_id=f"evt_memory_{_digest({'type': event_type, 'data': data})}", event_type=event_type, producer="memory", created_at=created_at, session_id=session_id, task_id=task_id, parent_task_id=parent_task_id, agent_id=agent_id, idempotency_key=( "memory:event:" f"{event_type}:{session_id or ''}:{task_id or ''}:{_digest(data)}" ), supporting_refs=refs, metadata={ "memory_event_type": event_type, "background_status": _background_status(event_type), }, ) def background_drain_event(data: Mapping[str, Any]) -> EvidenceEvent: created_at = _utc_now() session_id = _none_or_str(data.get("session_id")) task_id = _none_or_str(data.get("task_id")) reason = _none_or_str(data.get("reason")) or "unknown" ref = ResourceRef( ref_id=f"background_task_result:{session_id or 'none'}:{task_id or 'none'}:{reason}:{_digest(data)[:12]}", ref_type="background_task_result", session_id=session_id, task_id=task_id, producer="runtime", created_at=created_at, reliability="runtime", role="supporting", preview=f"background drain {reason}", metadata={ "reason": reason, "timeout_s": data.get("timeout_s"), "pending_count": data.get("pending_count"), "timed_out": bool(data.get("timed_out")), "session_memory_pending": data.get("session_memory_pending"), "memory_extraction_pending": ( data.get("memory_extraction_pending") if "memory_extraction_pending" in data else data.get("extraction_pending") ), "auto_dream_pending": data.get("auto_dream_pending"), "tracked_pending": data.get("tracked_pending"), }, ) return EvidenceEvent.create( event_id=f"evt_background_drain_{_digest(ref.ref_id)}", event_type="background_drain", producer="runtime", created_at=created_at, session_id=session_id, task_id=task_id, idempotency_key=f"runtime:background_drain:{ref.ref_id}", supporting_refs=[ref], metadata={"reason": reason}, ) def _memory_refs( event_type: str, data: Mapping[str, Any], *, created_at: str, ) -> list[ResourceRef]: paths: list[str] = [] for key in ("file_path", "entrypoint_path", "memory_dir", "memory_path"): value = _none_or_str(data.get(key)) if value: paths.append(value) for key in ( "files_touched", "paths", "memory_paths", "written_paths", "log_paths", ): raw = data.get(key) if isinstance(raw, (list, tuple, set)): paths.extend(str(item) for item in raw if item) refs: list[ResourceRef] = [] session_id = _none_or_str(data.get("session_id")) task_id = _none_or_str(data.get("task_id")) parent_task_id = _none_or_str(data.get("parent_task_id")) agent_id = _none_or_str(data.get("agent_id")) for path in dict.fromkeys(paths): source_kind = _memory_ref_source_kind(event_type) refs.append( ResourceRef( ref_id=memory_ref_id( source_kind, session_id=session_id, task_id=task_id, agent_id=agent_id, path=path, ), ref_type="memory_ref", uri=path, session_id=session_id, task_id=task_id, parent_task_id=parent_task_id, agent_id=agent_id, producer="memory", created_at=created_at, reliability="persisted" if Path(path).expanduser().exists() else "runtime", role="supporting", hash=_file_hash(path), preview=f"{event_type}: {Path(path).name}", metadata=_metadata_without_content( data, event_type=event_type, path=path, source_kind=source_kind, ), ) ) return refs def _background_task_ref( event_type: str, data: Mapping[str, Any], *, created_at: str, ) -> ResourceRef | None: if not _is_background_memory_event(event_type): return None session_id = _none_or_str(data.get("session_id")) task_id = _none_or_str(data.get("task_id")) parent_task_id = _none_or_str(data.get("parent_task_id")) agent_id = _none_or_str(data.get("agent_id")) status = _background_status(event_type) task_kind = _background_task_kind(event_type) ref_key = _digest( { "event_type": event_type, "session_id": session_id, "task_id": task_id, "status": status, "data": _metadata_without_content(data, event_type=event_type), } )[:16] return ResourceRef( ref_id=( "background_task_result:" f"{session_id or 'none'}:{task_id or 'none'}:{task_kind}:{ref_key}" ), ref_type="background_task_result", session_id=session_id, task_id=task_id, parent_task_id=parent_task_id, agent_id=agent_id, producer="memory", created_at=created_at, reliability="runtime", role="supporting", preview=f"{task_kind} {status}", metadata={ "task_kind": task_kind, "status": status, "source_event": event_type, "session_id": session_id, "task_id": task_id, "parent_task_id": parent_task_id, "agent_id": agent_id, "reason": data.get("reason") or data.get("skipped_reason"), "duration_ms": data.get("duration_ms"), "turn_count": data.get("turn_count"), "message_count": data.get("message_count"), "files_written": data.get("files_written"), "files_touched": data.get("files_touched"), "memory_mode": data.get("memory_mode"), "error": str(data.get("error") or "")[:500], }, ) def _is_background_memory_event(event_type: str) -> bool: return event_type.startswith( ( "session_memory_extraction", "memory_extraction", "auto_dream", "manual_dream", ) ) def _background_task_kind(event_type: str) -> str: if event_type.startswith("session_memory"): return "session_memory" if event_type.startswith("memory_extraction"): return "memory_extraction" if event_type.startswith("manual_dream"): return "manual_dream" if event_type.startswith("auto_dream"): return "auto_dream" return "memory" def _background_status(event_type: str) -> str | None: suffix = event_type.rsplit("_", 1)[-1] if suffix in {"start", "progress", "complete", "cancelled", "error", "skipped"}: return { "start": "running", "progress": "running", "complete": "complete", "cancelled": "cancelled", "error": "error", "skipped": "skipped", }[suffix] if event_type.endswith("_coalesced"): return "skipped" if event_type.endswith("_trailing_start"): return "running" return None def _file_hash(path_text: str) -> str | None: try: path = Path(path_text).expanduser() if not path.is_file(): return None return hashlib.sha256(path.read_bytes()).hexdigest() except Exception: return None def _metadata_without_content( data: Mapping[str, Any], *, event_type: str, path: str | None = None, source_kind: str | None = None, ) -> dict[str, Any]: excluded = {"content", "messages", "prompt", "response"} metadata = {key: value for key, value in data.items() if key not in excluded} metadata["memory_event_type"] = event_type metadata["memory_kind"] = _memory_kind(event_type, data, path=path) metadata["loaded_in_context"] = _loaded_in_context(event_type, data) metadata["read_or_written_by_tool"] = _read_or_written_by_tool(event_type, data) metadata["source_event"] = event_type metadata["source_kind"] = source_kind or _memory_ref_source_kind(event_type) if path is not None: metadata["path"] = path metadata["canonical_path_hash"] = canonical_path_hash(path) return metadata def _memory_ref_source_kind(event_type: str) -> str: return event_type def _memory_kind( event_type: str, data: Mapping[str, Any], *, path: str | None, ) -> str: explicit = _none_or_str(data.get("memory_kind") or data.get("memory_type")) if explicit: return explicit if event_type.startswith("session_memory"): return "session" if event_type in {"memory_logged"} or (path and "/logs/" in path): return "daily_log" if event_type.startswith("auto_dream") or event_type.startswith("manual_dream"): return "dream" if event_type == "nested_memory_consumed": return "nested" if event_type == "memory_prefetch_consumed": return "relevant" if _none_or_str(data.get("memory_mode")) == "daily_log": return "daily_log" return "memory" def _loaded_in_context(event_type: str, data: Mapping[str, Any]) -> bool: if "loaded_in_context" in data: return bool(data.get("loaded_in_context")) return event_type in {"memory_prefetch_consumed", "nested_memory_consumed"} def _read_or_written_by_tool(event_type: str, data: Mapping[str, Any]) -> bool: if "read_or_written_by_tool" in data: return bool(data.get("read_or_written_by_tool")) return event_type in {"memory_written", "memory_logged"} def _allow_synthetic_memory_ref(event_type: str) -> bool: return event_type not in {"memory_prefetch_consumed", "nested_memory_consumed"} def _digest(payload: Any) -> str: return hashlib.sha256( json.dumps(payload, sort_keys=True, default=str).encode("utf-8") ).hexdigest()[:24] def _none_or_str(value: Any) -> str | None: if value is None: return None text = str(value) return text if text else None def _utc_now() -> str: return datetime.now(timezone.utc).isoformat()