sync.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.
 
from __future__ import annotations
 
from typing import Any
 
import httpx
 
from aquin.engine.session_memory_store import memory_from_sync_events
 
 
def _client() -> httpx.Client:
    return httpx.Client(verify=False)
 
 
def push_events(
    session_id: str,
    events: list[dict[str, Any]],
    api_key: str,
    base_url: str,
    timeout: float = 15.0,
) -> int:
    """No-op — web sync event push removed."""
    _ = session_id, api_key, base_url, timeout
    return len(events)
 
 
def fetch_session_state(
    session_id: str,
    api_key: str,
    base_url: str,
    timeout: float = 10.0,
) -> dict[str, Any]:
    """Load session state from cloud; merge memory from snapshot + event replay."""
    if not (session_id and api_key):
        return {}
    try:
        with _client() as client:
            resp = client.get(
                f"{base_url.rstrip('/')}/api/sync/sessions/{session_id}",
                headers={"Authorization": f"Bearer {api_key}"},
                timeout=timeout,
            )
        if resp.status_code != 200:
            return {}
        body = resp.json()
        state = dict(body.get("state") or {})
        snapshot_mem = dict(state.get("memory") or {})
        replay_mem = memory_from_sync_events(body.get("events") or [])
        merged_mem = {**replay_mem, **snapshot_mem}
        if merged_mem:
            state["memory"] = merged_mem
        return state
    except Exception:
        pass
    return {}
 
 
def load_session_memory(
    session_id: str,
    api_key: str,
    base_url: str,
    *,
    local: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Merged memory: cloud snapshot/events, then local ctx, then disk cache."""
    from aquin.engine.session_memory_store import load_local_memory
 
    mem: dict[str, Any] = {}
    if session_id and api_key and base_url:
        cloud = fetch_session_state(session_id, api_key, base_url)
        mem.update(cloud.get("memory") or {})
    mem.update(load_local_memory(session_id))
    if local:
        mem.update(local)
    return mem