session_memory_store.py

Go to the documentation of this file.

# Copyright (c) 2025-present Aquin Labs Private Limited. All Rights Reserved.
# This file is part of the Aquin Engine. Unauthorized copying, modification,
# distribution, or use of this file, via any medium, is strictly prohibited.
# Proprietary and confidential. See LICENSE for terms.
 
"""Session memory — local disk + cloud (with event replay fallback)."""
from __future__ import annotations
 
import json
from pathlib import Path
from typing import Any
 
_MEMORY_DIR = Path.home() / ".aquin" / "session_memory"
 
 
def _memory_path(session_id: str) -> Path:
    safe = session_id.replace("/", "_")
    return _MEMORY_DIR / f"{safe}.json"
 
 
def load_local_memory(session_id: str) -> dict[str, Any]:
    if not session_id:
        return {}
    path = _memory_path(session_id)
    if not path.is_file():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}
 
 
def save_local_memory(session_id: str, memory: dict[str, Any]) -> None:
    if not session_id:
        return
    _MEMORY_DIR.mkdir(parents=True, exist_ok=True)
    _memory_path(session_id).write_text(
        json.dumps(memory, ensure_ascii=False, default=str),
        encoding="utf-8",
    )
 
 
def patch_local_memory(session_id: str, key: str, value: Any) -> dict[str, Any]:
    mem = load_local_memory(session_id)
    mem[key] = value
    save_local_memory(session_id, mem)
    return mem
 
 
def memory_from_sync_events(events: list[Any]) -> dict[str, Any]:
    """Replay session.meta memory patches from stored sync events."""
    mem: dict[str, Any] = {}
    for ev in events:
        if not isinstance(ev, dict):
            continue
        event_type = ev.get("event_type") or ev.get("type")
        if event_type != "session.meta":
            continue
        payload = ev.get("payload") or {}
        patch = payload.get("patch") if isinstance(payload, dict) else {}
        if not isinstance(patch, dict):
            continue
        chunk = patch.get("memory")
        if isinstance(chunk, dict):
            mem.update(chunk)
    return mem