add: database integration

requirements.txt

@@ -43,6 +43,7 @@ requests>=2.31.0
 # Data validation and serialization
 marshmallow>=3.20.0
 jsonschema>=4.19.0
+PyJWT>=2.8.0
 
 # Async support
 asyncio-mqtt>=0.16.0
@@ -61,4 +62,4 @@ mypy>=1.5.0
 clickhouse-connect>=0.7.0
 clickhouse-sqlalchemy==0.3.2
 
-langchain-tavily>=0.2.11
+langchain-tavily>=0.2.11

src/agents/swap/storage.py

@@ -1,45 +1,66 @@
-"""Persistent storage for swap intents and metadata."""
+"""Gateway-backed storage for swap intents with local fallback."""
 from __future__ import annotations
 
 import copy
-import json
-import os
 import time
 from datetime import datetime, timezone
-from pathlib import Path
 from threading import Lock
 from typing import Any, Dict, List, Optional
 
-_STATE_TEMPLATE: Dict[str, Dict[str, Any]] = {
-    "intents": {},
-    "metadata": {},
-    "history": {},
-}
+from src.integrations.panorama_gateway import (
+    PanoramaGatewayClient,
+    PanoramaGatewayError,
+    PanoramaGatewaySettings,
+    get_panorama_settings,
+)
+
+SWAP_SESSION_ENTITY = "swap-sessions"
+SWAP_HISTORY_ENTITY = "swap-histories"
+
+
+def _utc_now_iso() -> str:
+    return datetime.utcnow().replace(tzinfo=timezone.utc).isoformat()
+
+
+def _identifier(user_id: str, conversation_id: str) -> str:
+    return f"{user_id}:{conversation_id}"
+
+
+def _as_float(value: Any) -> Optional[float]:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
 
 
 class SwapStateRepository:
-    """File-backed storage for swap agent state with TTL and history support."""
+    """Stores swap agent state via Panorama's gateway or an in-memory fallback."""
 
     _instance: "SwapStateRepository" | None = None
     _instance_lock: Lock = Lock()
 
     def __init__(
         self,
-        path: Optional[Path] = None,
-        ttl_seconds: int = 3600,
+        *,
+        client: PanoramaGatewayClient | None = None,
+        settings: PanoramaGatewaySettings | None = None,
         history_limit: int = 10,
     ) -> None:
-        env_path = os.getenv("SWAP_STATE_PATH")
-        default_path = Path(__file__).with_name("swap_state.json")
-        self._path = Path(path or env_path or default_path)
-        self._ttl_seconds = ttl_seconds
         self._history_limit = history_limit
-        self._lock = Lock()
-        self._state: Dict[str, Dict[str, Any]] = copy.deepcopy(_STATE_TEMPLATE)
-        self._ensure_parent()
-        with self._lock:
-            self._load_locked()
+        try:
+            self._settings = settings or get_panorama_settings()
+            self._client = client or PanoramaGatewayClient(self._settings)
+            self._use_gateway = True
+        except ValueError:
+            # PANORAMA_GATEWAY_URL or JWT secrets not configured – fall back to local store.
+            self._settings = None
+            self._client = None
+            self._use_gateway = False
+            self._state = {"intents": {}, "metadata": {}, "history": {}}
 
+    # ---- Singleton helpers -----------------------------------------------
     @classmethod
     def instance(cls) -> "SwapStateRepository":
         if cls._instance is None:
@@ -53,93 +74,19 @@ class SwapStateRepository:
         with cls._instance_lock:
             cls._instance = None
 
-    def _ensure_parent(self) -> None:
-        try:
-            self._path.parent.mkdir(parents=True, exist_ok=True)
-        except OSError:
-            pass
-
-    def _load_locked(self) -> None:
-        if not self._path.exists():
-            self._state = copy.deepcopy(_STATE_TEMPLATE)
-            return
-        try:
-            data = json.loads(self._path.read_text())
-        except Exception:
-            self._state = copy.deepcopy(_STATE_TEMPLATE)
-            return
-        if not isinstance(data, dict):
-            self._state = copy.deepcopy(_STATE_TEMPLATE)
-            return
-        state = copy.deepcopy(_STATE_TEMPLATE)
-        for key in state:
-            if isinstance(data.get(key), dict):
-                state[key] = data[key]
-        self._state = state
-
-    def _persist_locked(self) -> None:
-        self._ensure_parent()
-        tmp_path = self._path.with_suffix(".tmp")
-        payload = json.dumps(self._state, ensure_ascii=False, indent=2)
-        try:
-            tmp_path.write_text(payload)
-            tmp_path.replace(self._path)
-        except Exception:
-            pass
-
-    @staticmethod
-    def _normalize_identifier(value: Optional[str], label: str) -> str:
-        candidate = (value or "").strip()
-        if not candidate:
-            raise ValueError(f"{label} is required for swap state operations.")
-        return candidate
-
-    def _key(self, user_id: Optional[str], conversation_id: Optional[str]) -> str:
-        user = self._normalize_identifier(user_id, "user_id")
-        conversation = self._normalize_identifier(conversation_id, "conversation_id")
-        return f"{user}::{conversation}"
-
-    def _purge_locked(self) -> None:
-        if self._ttl_seconds <= 0:
-            return
-        cutoff = time.time() - self._ttl_seconds
-        intents = self._state["intents"]
-        metadata = self._state["metadata"]
-        stale_keys = [
-            key for key, record in intents.items() if record.get("updated_at", 0) < cutoff
-        ]
-        for key in stale_keys:
-            intents.pop(key, None)
-            metadata.pop(key, None)
-
-    @staticmethod
-    def _format_timestamp(value: float) -> str:
-        return datetime.fromtimestamp(value, tz=timezone.utc).isoformat()
-
-    def _history_unlocked(self, key: str, limit: Optional[int] = None) -> List[Dict[str, Any]]:
-        history = self._state["history"].get(key, [])
-        sorted_history = sorted(history, key=lambda item: item.get("timestamp", 0), reverse=True)
-        effective_limit = limit or self._history_limit
-        if effective_limit:
-            sorted_history = sorted_history[:effective_limit]
-        results: List[Dict[str, Any]] = []
-        for item in sorted_history:
-            entry = copy.deepcopy(item)
-            ts = entry.get("timestamp")
-            if ts is not None:
-                entry["timestamp"] = self._format_timestamp(float(ts))
-            results.append(entry)
-        return results
-
+    # ---- Core API ---------------------------------------------------------
     def load_intent(self, user_id: str, conversation_id: str) -> Optional[Dict[str, Any]]:
-        key = self._key(user_id, conversation_id)
-        with self._lock:
-            self._purge_locked()
-            record = self._state["intents"].get(key)
+        if not self._use_gateway:
+            record = self._state["intents"].get(_identifier(user_id, conversation_id))
             if not record:
                 return None
             return copy.deepcopy(record.get("intent"))
 
+        session = self._get_session(user_id, conversation_id)
+        if not session:
+            return None
+        return session.get("intent") or None
+
     def persist_intent(
         self,
         user_id: str,
@@ -149,80 +96,230 @@ class SwapStateRepository:
         done: bool,
         summary: Optional[Dict[str, Any]] = None,
     ) -> List[Dict[str, Any]]:
-        key = self._key(user_id, conversation_id)
-        with self._lock:
-            self._purge_locked()
+        if not self._use_gateway:
+            key = _identifier(user_id, conversation_id)
             now = time.time()
             if done:
                 self._state["intents"].pop(key, None)
             else:
-                self._state["intents"][key] = {
-                    "intent": copy.deepcopy(intent),
-                    "updated_at": now,
-                }
+                self._state["intents"][key] = {"intent": copy.deepcopy(intent), "updated_at": now}
             if metadata:
                 meta_copy = copy.deepcopy(metadata)
                 meta_copy["updated_at"] = now
                 self._state["metadata"][key] = meta_copy
             if done and summary:
-                history = self._state["history"].get(key, [])
+                history = self._state["history"].setdefault(key, [])
                 summary_copy = copy.deepcopy(summary)
                 summary_copy.setdefault("timestamp", now)
                 history.append(summary_copy)
-                self._state["history"][key] = history[-self._history_limit:]
-            try:
-                self._persist_locked()
-            except Exception:
-                pass
-            return self._history_unlocked(key)
-
-    def set_metadata(self, user_id: str, conversation_id: str, metadata: Dict[str, Any]) -> None:
-        key = self._key(user_id, conversation_id)
-        with self._lock:
-            self._purge_locked()
+                self._state["history"][key] = history[-self._history_limit :]
+            return self.get_history(user_id, conversation_id)
+
+        if done:
+            if summary:
+                self._create_history_entry(user_id, conversation_id, summary)
+            self._delete_session(user_id, conversation_id)
+        else:
+            payload = self._session_payload(intent, metadata)
+            self._upsert_session(user_id, conversation_id, payload)
+        return self.get_history(user_id, conversation_id)
+
+    def set_metadata(
+        self,
+        user_id: str,
+        conversation_id: str,
+        metadata: Dict[str, Any],
+    ) -> None:
+        if not self._use_gateway:
+            key = _identifier(user_id, conversation_id)
             if metadata:
                 meta_copy = copy.deepcopy(metadata)
                 meta_copy["updated_at"] = time.time()
                 self._state["metadata"][key] = meta_copy
             else:
                 self._state["metadata"].pop(key, None)
-            try:
-                self._persist_locked()
-            except Exception:
-                pass
+            return
+
+        if not metadata:
+            self._delete_session(user_id, conversation_id)
+            return
+
+        session = self._get_session(user_id, conversation_id)
+        intent = session.get("intent") if session else {}
+        payload = self._session_payload(intent or {}, metadata)
+        self._upsert_session(user_id, conversation_id, payload)
 
     def clear_metadata(self, user_id: str, conversation_id: str) -> None:
         self.set_metadata(user_id, conversation_id, {})
 
     def clear_intent(self, user_id: str, conversation_id: str) -> None:
-        key = self._key(user_id, conversation_id)
-        with self._lock:
-            self._state["intents"].pop(key, None)
-            try:
-                self._persist_locked()
-            except Exception:
-                pass
+        if not self._use_gateway:
+            self._state["intents"].pop(_identifier(user_id, conversation_id), None)
+            self._state["metadata"].pop(_identifier(user_id, conversation_id), None)
+            return
+        self._delete_session(user_id, conversation_id)
 
     def get_metadata(self, user_id: str, conversation_id: str) -> Dict[str, Any]:
-        key = self._key(user_id, conversation_id)
-        with self._lock:
-            self._purge_locked()
-            meta = self._state["metadata"].get(key)
-            if not meta:
+        if not self._use_gateway:
+            record = self._state["metadata"].get(_identifier(user_id, conversation_id))
+            if not record:
                 return {}
-            entry = copy.deepcopy(meta)
+            entry = copy.deepcopy(record)
             ts = entry.pop("updated_at", None)
             if ts is not None:
-                entry["updated_at"] = self._format_timestamp(float(ts))
+                entry["updated_at"] = datetime.fromtimestamp(float(ts), tz=timezone.utc).isoformat()
             return entry
 
+        session = self._get_session(user_id, conversation_id)
+        if not session:
+            return {}
+
+        intent = session.get("intent") or {}
+        metadata: Dict[str, Any] = {
+            "event": session.get("event"),
+            "status": session.get("status"),
+            "missing_fields": session.get("missingFields") or [],
+            "next_field": session.get("nextField"),
+            "pending_question": session.get("pendingQuestion"),
+            "choices": session.get("choices") or [],
+            "error": session.get("errorMessage"),
+            "user_id": user_id,
+            "conversation_id": conversation_id,
+        }
+        metadata["from_network"] = intent.get("from_network")
+        metadata["from_token"] = intent.get("from_token")
+        metadata["to_network"] = intent.get("to_network")
+        metadata["to_token"] = intent.get("to_token")
+        metadata["amount"] = intent.get("amount")
+
+        history = self.get_history(user_id, conversation_id)
+        if history:
+            metadata["history"] = history
+
+        updated_at = session.get("updatedAt")
+        if updated_at:
+            metadata["updated_at"] = updated_at
+
+        return metadata
+
     def get_history(
         self,
         user_id: str,
         conversation_id: str,
         limit: Optional[int] = None,
     ) -> List[Dict[str, Any]]:
-        key = self._key(user_id, conversation_id)
-        with self._lock:
-            return self._history_unlocked(key, limit)
+        if not self._use_gateway:
+            key = _identifier(user_id, conversation_id)
+            history = self._state["history"].get(key, [])
+            effective = limit or self._history_limit
+            result: List[Dict[str, Any]] = []
+            for item in sorted(history, key=lambda entry: entry.get("timestamp", 0), reverse=True)[:effective]:
+                entry = copy.deepcopy(item)
+                ts = entry.get("timestamp")
+                if ts is not None:
+                    entry["timestamp"] = datetime.fromtimestamp(float(ts), tz=timezone.utc).isoformat()
+                result.append(entry)
+            return result
+
+        effective_limit = limit or self._history_limit
+        result = self._client.list(
+            SWAP_HISTORY_ENTITY,
+            {
+                "where": {"userId": user_id, "conversationId": conversation_id},
+                "orderBy": {"recordedAt": "desc"},
+                "take": effective_limit,
+            },
+        )
+        data = result.get("data", []) if isinstance(result, dict) else []
+        history: List[Dict[str, Any]] = []
+        for entry in data:
+            history.append(
+                {
+                    "status": entry.get("status"),
+                    "from_network": entry.get("fromNetwork"),
+                    "from_token": entry.get("fromToken"),
+                    "to_network": entry.get("toNetwork"),
+                    "to_token": entry.get("toToken"),
+                    "amount": entry.get("amount"),
+                    "error": entry.get("errorMessage"),
+                    "timestamp": entry.get("recordedAt"),
+                }
+            )
+        return history
+
+    # ---- Gateway helpers --------------------------------------------------
+    def _get_session(self, user_id: str, conversation_id: str) -> Optional[Dict[str, Any]]:
+        identifier = _identifier(user_id, conversation_id)
+        try:
+            return self._client.get(SWAP_SESSION_ENTITY, identifier)
+        except PanoramaGatewayError as exc:
+            if exc.status_code == 404:
+                return None
+            raise
+
+    def _delete_session(self, user_id: str, conversation_id: str) -> None:
+        identifier = _identifier(user_id, conversation_id)
+        try:
+            self._client.delete(SWAP_SESSION_ENTITY, identifier)
+        except PanoramaGatewayError as exc:
+            if exc.status_code != 404:
+                raise
+
+    def _upsert_session(
+        self,
+        user_id: str,
+        conversation_id: str,
+        data: Dict[str, Any],
+    ) -> None:
+        identifier = _identifier(user_id, conversation_id)
+        payload = {**data, "updatedAt": _utc_now_iso()}
+        try:
+            self._client.update(SWAP_SESSION_ENTITY, identifier, payload)
+        except PanoramaGatewayError as exc:
+            if exc.status_code != 404:
+                raise
+            create_payload = {
+                "userId": user_id,
+                "conversationId": conversation_id,
+                "tenantId": self._settings.tenant_id,
+                **payload,
+            }
+            self._client.create(SWAP_SESSION_ENTITY, create_payload)
+
+    def _create_history_entry(
+        self,
+        user_id: str,
+        conversation_id: str,
+        summary: Dict[str, Any],
+    ) -> None:
+        history_payload = {
+            "userId": user_id,
+            "conversationId": conversation_id,
+            "status": summary.get("status"),
+            "fromNetwork": summary.get("from_network"),
+            "fromToken": summary.get("from_token"),
+            "toNetwork": summary.get("to_network"),
+            "toToken": summary.get("to_token"),
+            "amount": _as_float(summary.get("amount")),
+            "errorMessage": summary.get("error"),
+            "recordedAt": _utc_now_iso(),
+            "tenantId": self._settings.tenant_id,
+        }
+        self._client.create(SWAP_HISTORY_ENTITY, history_payload)
 
+    @staticmethod
+    def _session_payload(intent: Dict[str, Any], metadata: Dict[str, Any]) -> Dict[str, Any]:
+        missing = metadata.get("missing_fields") or []
+        if not isinstance(missing, list):
+            missing = list(missing)
+        return {
+            "status": metadata.get("status"),
+            "event": metadata.get("event"),
+            "intent": intent,
+            "missingFields": missing,
+            "nextField": metadata.get("next_field"),
+            "pendingQuestion": metadata.get("pending_question"),
+            "choices": metadata.get("choices"),
+            "errorMessage": metadata.get("error"),
+            "historyCursor": metadata.get("history_cursor") or 0,
+        }

src/app.py

@@ -155,6 +155,20 @@ def chat(request: ChatRequest):
     try:
         user_id, conversation_id = _resolve_identity(request)
 
+        wallet = request.wallet_address.strip() if request.wallet_address else None
+        if wallet and wallet.lower() == "default":
+            wallet = None
+        display_name = None
+        if isinstance(request.message.metadata, dict):
+            display_name = request.message.metadata.get("display_name")
+
+        chat_manager_instance.ensure_session(
+            user_id,
+            conversation_id,
+            wallet_address=wallet,
+            display_name=display_name,
+        )
+
         # Add the user message to the conversation
         chat_manager_instance.add_message(
             message=request.message.dict(),

src/integrations/panorama_gateway/client.py

@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+import json
+import time
+import uuid
+from dataclasses import asdict
+from typing import Any, Dict, Iterable, Optional
+
+import httpx
+import jwt
+
+from .config import PanoramaGatewaySettings, get_panorama_settings
+
+
+class PanoramaGatewayError(RuntimeError):
+    """Raised when the Panorama gateway returns an error response."""
+
+    def __init__(self, message: str, status_code: int, payload: Any | None = None) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.payload = payload
+
+
+def _encode_identifier(identifier: Any) -> str:
+    """Coerce identifiers into the colon-delimited format expected by the gateway."""
+
+    if isinstance(identifier, str):
+        return identifier
+    if isinstance(identifier, Iterable):
+        parts: Iterable[str] = (str(part) for part in identifier)
+        return ":".join(parts)
+    if isinstance(identifier, dict):
+        return ":".join(str(value) for value in identifier.values())
+    raise ValueError(f"Unsupported identifier type: {type(identifier)}")
+
+
+class PanoramaGatewayClient:
+    """HTTP client wrapper for Panorama's data gateway."""
+
+    def __init__(
+        self,
+        settings: PanoramaGatewaySettings | None = None,
+        *,
+        client: httpx.Client | None = None,
+    ) -> None:
+        self._settings = settings or get_panorama_settings()
+        self._client = client or httpx.Client(
+            base_url=self._settings.base_url,
+            timeout=self._settings.request_timeout,
+        )
+
+    def __enter__(self) -> "PanoramaGatewayClient":
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        self.close()
+
+    def close(self) -> None:
+        self._client.close()
+
+    # ---- low-level helpers -------------------------------------------------
+    def _build_token(self) -> str:
+        now = int(time.time())
+        payload: Dict[str, Any] = {
+            "iat": now,
+            "exp": now + 300,
+            "service": self._settings.service_name,
+            "roles": self._settings.roles,
+            "tenant": self._settings.tenant_id,
+        }
+        if self._settings.jwt_audience:
+            payload["aud"] = self._settings.jwt_audience
+        if self._settings.jwt_issuer:
+            payload["iss"] = self._settings.jwt_issuer
+
+        return jwt.encode(payload, self._settings.jwt_secret, algorithm="HS256")
+
+    def _default_headers(self) -> Dict[str, str]:
+        return {
+            "Authorization": f"Bearer {self._build_token()}",
+            "x-tenant-id": self._settings.tenant_id,
+            "Accept": "application/json",
+        }
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        params: Dict[str, Any] | None = None,
+        json_body: Any | None = None,
+        idempotency_key: str | None = None,
+    ) -> Any:
+        headers = self._default_headers()
+        if method.upper() in {"POST", "PATCH", "PUT", "DELETE"}:
+            headers["Idempotency-Key"] = idempotency_key or str(uuid.uuid4())
+
+        response = self._client.request(
+            method=method,
+            url=path,
+            headers=headers,
+            params=params,
+            json=json_body,
+        )
+
+        if response.status_code >= 400:
+            message = f"Gateway request failed ({response.status_code})"
+            try:
+                payload = response.json()
+            except ValueError:
+                payload = response.text
+            raise PanoramaGatewayError(message, response.status_code, payload)
+
+        if response.status_code == 204:
+            return None
+
+        if response.headers.get("content-type", "").startswith("application/json"):
+            return response.json()
+        return response.text
+
+    # ---- CRUD facades ------------------------------------------------------
+    def list(self, entity: str, query: Dict[str, Any] | None = None) -> Dict[str, Any]:
+        params = None
+        if query:
+            params = {}
+            for key, value in query.items():
+                if isinstance(value, (dict, list)):
+                    params[key] = json.dumps(value)
+                else:
+                    params[key] = value
+        return self._request("GET", f"/v1/{entity}", params=params)
+
+    def get(self, entity: str, identifier: Any) -> Any:
+        encoded_id = _encode_identifier(identifier)
+        return self._request("GET", f"/v1/{entity}/{encoded_id}")
+
+    def create(
+        self,
+        entity: str,
+        payload: Dict[str, Any],
+        *,
+        idempotency_key: str | None = None,
+    ) -> Any:
+        return self._request(
+            "POST",
+            f"/v1/{entity}",
+            json_body=payload,
+            idempotency_key=idempotency_key,
+        )
+
+    def update(
+        self,
+        entity: str,
+        identifier: Any,
+        payload: Dict[str, Any],
+        *,
+        idempotency_key: str | None = None,
+    ) -> Any:
+        encoded_id = _encode_identifier(identifier)
+        return self._request(
+            "PATCH",
+            f"/v1/{entity}/{encoded_id}",
+            json_body=payload,
+            idempotency_key=idempotency_key,
+        )
+
+    def delete(
+        self,
+        entity: str,
+        identifier: Any,
+        *,
+        idempotency_key: str | None = None,
+    ) -> None:
+        encoded_id = _encode_identifier(identifier)
+        self._request(
+            "DELETE",
+            f"/v1/{entity}/{encoded_id}",
+            idempotency_key=idempotency_key,
+        )
+
+    def transact(
+        self,
+        operations: Iterable[Dict[str, Any]],
+        *,
+        idempotency_key: str | None = None,
+    ) -> Any:
+        payload = {"ops": list(operations)}
+        return self._request(
+            "POST",
+            "/v1/_transact",
+            json_body=payload,
+            idempotency_key=idempotency_key,
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Return a serialisable snapshot of the current settings (useful for debugging)."""
+
+        return asdict(self._settings)

src/integrations/panorama_gateway/config.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from functools import lru_cache
+from typing import List, Optional
+
+
+def _split_roles(raw: str | None) -> List[str]:
+    if not raw:
+        return ["agent"]
+    return [part.strip() for part in raw.split(",") if part.strip()]
+
+
+@dataclass(slots=True)
+class PanoramaGatewaySettings:
+    """Runtime configuration for the Panorama data gateway client."""
+
+    base_url: str
+    jwt_secret: str
+    tenant_id: str = "tenant-agent"
+    service_name: str = "zico-agent"
+    roles: List[str] = field(default_factory=lambda: ["agent"])
+    jwt_audience: Optional[str] = None
+    jwt_issuer: Optional[str] = None
+    request_timeout: float = 10.0
+
+    @classmethod
+    def load(cls) -> "PanoramaGatewaySettings":
+        base_url = os.getenv("PANORAMA_GATEWAY_URL")
+        if not base_url:
+            raise ValueError("PANORAMA_GATEWAY_URL environment variable is required.")
+
+        secret = os.getenv("PANORAMA_GATEWAY_JWT_SECRET") or os.getenv("JWT_SECRET")
+        if not secret:
+            raise ValueError(
+                "Set PANORAMA_GATEWAY_JWT_SECRET or reuse JWT_SECRET for Panorama gateway auth."
+            )
+
+        return cls(
+            base_url=base_url.rstrip("/"),
+            jwt_secret=secret,
+            tenant_id=os.getenv("PANORAMA_GATEWAY_TENANT", "tenant-agent"),
+            service_name=os.getenv("PANORAMA_GATEWAY_SERVICE", "zico-agent"),
+            roles=_split_roles(os.getenv("PANORAMA_GATEWAY_ROLES", "agent")),
+            jwt_audience=os.getenv("PANORAMA_GATEWAY_JWT_AUDIENCE"),
+            jwt_issuer=os.getenv("PANORAMA_GATEWAY_JWT_ISSUER"),
+            request_timeout=float(os.getenv("PANORAMA_GATEWAY_TIMEOUT", "10")),
+        )
+
+
+@lru_cache(maxsize=1)
+def get_panorama_settings() -> PanoramaGatewaySettings:
+    """Memoized accessor so callers share a single settings instance."""
+
+    return PanoramaGatewaySettings.load()

src/service/chat_manager.py

@@ -1,125 +1,86 @@
+from __future__ import annotations
+
 import logging
-import time
+import uuid
 from typing import Dict, List, Optional
-from src.models.chatMessage import ChatMessage, ConversationState, AgentResponse
-from datetime import datetime
+
+from src.models.chatMessage import AgentResponse, ChatMessage
+from src.service.panorama_store import PanoramaStore
 
 logger = logging.getLogger(__name__)
 
 
 class ChatManager:
-    """
-    Manages chat conversations and message history.
-
-    This class provides functionality to:
-    - Create and manage multiple conversations identified by unique IDs
-    - Associate conversations with specific users
-    - Add/retrieve messages and responses within conversations
-    - Clear conversation history
-    - Get chat history in different formats
-    - Delete conversations
-
-    Each conversation starts with a default disclaimer message about the experimental nature
-    of the chatbot.
-
-    Attributes:
-        user_conversations (Dict[str, Dict[str, ConversationState]]): Dictionary mapping user IDs to their conversations
-        default_message (ChatMessage): Default disclaimer message added to new conversations
-
-    Example:
-        >>> chat_manager = ChatManager()
-        >>> chat_manager.add_message({"role": "user", "content": "Hello"}, "conv1", "user123")
-        >>> messages = chat_manager.get_messages("conv1", "user123")
-    """
-
-    def __init__(self) -> None:
-        self.user_conversations: Dict[str, Dict[str, ConversationState]] = {}
-        self.default_message = ChatMessage(
-            role="assistant",
-            content="""This highly experimental chatbot is not intended for making important decisions. Its
-                        responses are generated using AI models and may not always be accurate.
-                        By using this chatbot, you acknowledge that you use it at your own discretion
-                        and assume all risks associated with its limitations and potential errors.""",
-            metadata={},
-        )
-
-        # Backward compatibility - Initialize with default conversation for anonymous user
-        self._initialize_user("anonymous")
-        self.user_conversations["anonymous"]["default"] = ConversationState(
-            conversation_id="default",
-            user_id="anonymous",
-            messages=[self.default_message],
-            context={},
-            memory={},
-            agent_history=[],
-            current_agent=None,
-            last_message_id=None,
-            created_at=datetime.utcnow(),
-            updated_at=datetime.utcnow(),
-            is_active=True
-        )
-
-    def _get_conversation_id(self, conversation_id: Optional[str] = None) -> str:
-        """Helper method to get conversation ID, defaulting to 'default' if None provided"""
-        return conversation_id or "default"
-
-    def _get_user_id(self, user_id: Optional[str] = None) -> str:
-        """Helper method to get user ID, defaulting to 'anonymous' if None provided"""
-        return user_id or "anonymous"
-
-    def _initialize_user(self, user_id: str) -> None:
-        """Initialize conversations dictionary for a new user"""
-        if user_id not in self.user_conversations:
-            self.user_conversations[user_id] = {}
-            logger.info(f"Initialized conversations for user {user_id}")
-
-    def get_messages(self, conversation_id: Optional[str] = None, user_id: Optional[str] = None) -> List[Dict[str, str]]:
-        """
-        Get all messages for a specific conversation.
-
-        Args:
-            conversation_id (str, optional): Unique identifier for the conversation. Defaults to "default"
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-
-        Returns:
-            List[Dict[str, str]]: List of messages as dictionaries
-        """
-        user_id = self._get_user_id(user_id)
-        conversation = self._get_or_create_conversation(self._get_conversation_id(conversation_id), user_id)
-        return [msg.dict() for msg in conversation.messages]
-
-    def add_message(self, message: Dict[str, str], conversation_id: Optional[str] = None, user_id: Optional[str] = None):
-        """
-        Add a new message to a conversation.
-
-        Args:
-            message (Dict[str, str]): Message to add
-            conversation_id (str, optional): Conversation to add message to. Defaults to "default"
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-        """
-        user_id = self._get_user_id(user_id)
-        conversation_id = self._get_conversation_id(conversation_id)
-        conversation = self._get_or_create_conversation(conversation_id, user_id)
+    """Facade that delegates chat persistence to the Panorama data gateway."""
+
+    def __init__(self, store: PanoramaStore | None = None) -> None:
+        self._store = store or PanoramaStore()
+
+    @staticmethod
+    def _resolve_ids(
+        conversation_id: Optional[str],
+        user_id: Optional[str],
+    ) -> tuple[str, str]:
+        return (conversation_id or "default", user_id or "anonymous")
+
+    # ---- Message access ---------------------------------------------------
+    def get_messages(
+        self,
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ) -> List[Dict[str, str]]:
+        conversation_id, user_id = self._resolve_ids(conversation_id, user_id)
+        self._store.ensure_conversation(user_id, conversation_id)
+        messages = self._store.list_messages(user_id, conversation_id)
+        return messages
+
+    def get_last_message(
+        self,
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ) -> Dict[str, str]:
+        messages = self.get_messages(conversation_id, user_id)
+        return messages[-1] if messages else {}
+
+    def get_chat_history(
+        self,
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ) -> str:
+        messages = self.get_messages(conversation_id, user_id)
+        return "\n".join(f"{msg.get('role')}: {msg.get('content')}" for msg in messages)
+
+    # ---- Mutations --------------------------------------------------------
+    def add_message(
+        self,
+        message: Dict[str, str],
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ) -> Dict[str, str]:
+        conversation_id, user_id = self._resolve_ids(conversation_id, user_id)
+        self._store.ensure_user_and_conversation(user_id, conversation_id)
         chat_message = ChatMessage(**message)
+        if not chat_message.message_id:
+            chat_message.message_id = str(uuid.uuid4())
         chat_message.conversation_id = conversation_id
         chat_message.user_id = user_id
-        if "timestamp" not in message:
-            chat_message.timestamp = datetime.utcnow()
-        conversation.messages.append(chat_message)
-        logger.info(f"Added message to conversation {conversation_id} for user {user_id}: {chat_message.content}")
-
-    def add_response(self, response: Dict[str, str], agent_name: str, conversation_id: Optional[str] = None, user_id: Optional[str] = None):
-        """
-        Add an agent's response to a conversation.
-
-        Args:
-            response (Dict[str, str]): Response content
-            agent_name (str): Name of the responding agent
-            conversation_id (str, optional): Conversation to add response to. Defaults to "default"
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-        """
+        stored = self._store.add_message(user_id, conversation_id, chat_message)
+        logger.info(
+            "Persisted message for user=%s conversation=%s role=%s",
+            user_id,
+            conversation_id,
+            chat_message.role,
+        )
+        return stored
+
+    def add_response(
+        self,
+        response: Dict[str, str],
+        agent_name: str,
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ) -> Dict[str, str]:
         agent_response = AgentResponse(**response)
-        # You may need to implement to_chat_message if not present
         chat_message = ChatMessage(
             role="assistant",
             content=agent_response.content,
@@ -131,152 +92,82 @@ class ChatManager:
             next_agent=agent_response.next_agent,
             requires_followup=agent_response.requires_followup,
             status="completed" if agent_response.success else "failed",
-            error_message=agent_response.error_message
+            error_message=agent_response.error_message,
+        )
+        stored = self.add_message(
+            chat_message.dict(),
+            conversation_id=conversation_id,
+            user_id=user_id,
+        )
+        logger.info(
+            "Persisted agent response from %s for user=%s conversation=%s",
+            agent_name,
+            user_id,
+            conversation_id,
+        )
+        return stored
+
+    def clear_messages(
+        self,
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ) -> None:
+        conversation_id, user_id = self._resolve_ids(conversation_id, user_id)
+        self._store.ensure_conversation(user_id, conversation_id)
+        self._store.reset_conversation(user_id, conversation_id)
+        logger.info(
+            "Cleared messages for user=%s conversation=%s",
+            user_id,
+            conversation_id,
         )
-        self.add_message(chat_message.dict(), self._get_conversation_id(conversation_id), self._get_user_id(user_id))
-        logger.info(f"Added response from agent {agent_name} to conversation {conversation_id} for user {user_id}")
-
-    def clear_messages(self, conversation_id: Optional[str] = None, user_id: Optional[str] = None):
-        """
-        Clear all messages in a conversation except the default message.
-
-        Args:
-            conversation_id (str, optional): Conversation to clear. Defaults to "default"
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-        """
-        user_id = self._get_user_id(user_id)
-        conversation = self._get_or_create_conversation(self._get_conversation_id(conversation_id), user_id)
-        conversation.messages = [self.default_message]  # Keep the initial message
-        logger.info(f"Cleared message history for conversation {conversation_id} for user {user_id}")
-
-    def get_last_message(self, conversation_id: Optional[str] = None, user_id: Optional[str] = None) -> Dict[str, str]:
-        """
-        Get the most recent message from a conversation.
-
-        Args:
-            conversation_id (str, optional): Conversation to get message from. Defaults to "default"
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-
-        Returns:
-            Dict[str, str]: Last message or empty dict if no messages
-        """
-        user_id = self._get_user_id(user_id)
-        conversation = self._get_or_create_conversation(self._get_conversation_id(conversation_id), user_id)
-        return conversation.messages[-1].dict() if conversation.messages else {}
-
-    def get_chat_history(self, conversation_id: Optional[str] = None, user_id: Optional[str] = None) -> str:
-        """
-        Get formatted chat history for a conversation.
-
-        Args:
-            conversation_id (str, optional): Conversation to get history for. Defaults to "default"
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-
-        Returns:
-            str: Formatted chat history as string
-        """
-        user_id = self._get_user_id(user_id)
-        conversation = self._get_or_create_conversation(self._get_conversation_id(conversation_id), user_id)
-        return "\n".join([f"{msg.role}: {msg.content}" for msg in conversation.messages])
-
-    def get_all_conversation_ids(self, user_id: Optional[str] = None) -> List[str]:
-        """
-        Get a list of all conversation IDs for a specific user.
-
-        Args:
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-
-        Returns:
-            List[str]: List of conversation IDs for the user
-        """
-        user_id = self._get_user_id(user_id)
-        self._initialize_user(user_id)
-        return list(self.user_conversations[user_id].keys())
-
-    def get_all_user_ids(self) -> List[str]:
-        """
-        Get a list of all user IDs.
-
-        Returns:
-            List[str]: List of user IDs
-        """
-        return list(self.user_conversations.keys())
-
-    def delete_conversation(self, conversation_id: Optional[str] = None, user_id: Optional[str] = None):
-        """
-        Delete a conversation by ID.
 
-        Args:
-            conversation_id (str, optional): ID of conversation to delete. Defaults to "default"
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-        """
-        user_id = self._get_user_id(user_id)
-        conversation_id = self._get_conversation_id(conversation_id)
-        self._initialize_user(user_id)
-        if conversation_id in self.user_conversations[user_id]:
-            del self.user_conversations[user_id][conversation_id]
-            logger.info(f"Deleted conversation {conversation_id} for user {user_id}")
+    def delete_conversation(
+        self,
+        conversation_id: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ) -> None:
+        conversation_id, user_id = self._resolve_ids(conversation_id, user_id)
+        self._store.delete_conversation(user_id, conversation_id)
+        logger.info(
+            "Deleted conversation for user=%s conversation=%s",
+            user_id,
+            conversation_id,
+        )
 
     def create_conversation(self, user_id: Optional[str] = None) -> str:
-        """
-        Create a new conversation for a user.
-
-        Args:
-            user_id (str, optional): User identifier. Defaults to "anonymous"
-
-        Returns:
-            str: ID of the created conversation
-        """
-        user_id = self._get_user_id(user_id)
-        self._initialize_user(user_id)
-        conversation_id = f"conversation_{len(self.user_conversations[user_id])}"
-        while conversation_id in self.user_conversations[user_id]:
-            conversation_id = f"conversation_{len(self.user_conversations[user_id])}_{int(time.time())}"
-        self.user_conversations[user_id][conversation_id] = ConversationState(
-            conversation_id=conversation_id,
-            user_id=user_id,
-            messages=[self.default_message],
-            context={},
-            memory={},
-            agent_history=[],
-            current_agent=None,
-            last_message_id=None,
-            created_at=datetime.utcnow(),
-            updated_at=datetime.utcnow(),
-            is_active=True
+        user_id = (user_id or "anonymous")
+        conversation_id = f"conversation-{uuid.uuid4().hex[:8]}"
+        self._store.ensure_user_and_conversation(user_id, conversation_id)
+        logger.info(
+            "Created new conversation for user=%s conversation=%s",
+            user_id,
+            conversation_id,
         )
-        logger.info(f"Created new conversation {conversation_id} for user {user_id}")
         return conversation_id
 
-    def _get_or_create_conversation(self, conversation_id: str, user_id: str) -> ConversationState:
-        """
-        Get existing conversation or create new one if not exists.
-
-        Args:
-            conversation_id (str): Conversation ID to get/create
-            user_id (str): User identifier
+    # ---- Discovery helpers ------------------------------------------------
+    def get_all_conversation_ids(self, user_id: Optional[str] = None) -> List[str]:
+        user_id = (user_id or "anonymous")
+        return self._store.list_conversations(user_id)
 
-        Returns:
-            ConversationState: Retrieved or created conversation
-        """
-        self._initialize_user(user_id)
-        if conversation_id not in self.user_conversations[user_id]:
-            self.user_conversations[user_id][conversation_id] = ConversationState(
-                conversation_id=conversation_id,
-                user_id=user_id,
-                messages=[self.default_message],
-                context={},
-                memory={},
-                agent_history=[],
-                current_agent=None,
-                last_message_id=None,
-                created_at=datetime.utcnow(),
-                updated_at=datetime.utcnow(),
-                is_active=True
-            )
-            logger.info(f"Created new conversation {conversation_id} for user {user_id}")
-        return self.user_conversations[user_id][conversation_id]
+    def get_all_user_ids(self) -> List[str]:
+        return self._store.list_users()
+
+    def ensure_session(
+        self,
+        user_id: str,
+        conversation_id: str,
+        *,
+        wallet_address: Optional[str] = None,
+        display_name: Optional[str] = None,
+    ) -> None:
+        self._store.ensure_user_and_conversation(
+            user_id,
+            conversation_id,
+            wallet_address=wallet_address,
+            display_name=display_name,
+        )
 
 
-# Create an instance to act as a singleton store
+# Singleton-style accessor for the FastAPI routes
 chat_manager_instance = ChatManager()

src/service/panorama_store.py

@@ -0,0 +1,390 @@
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional, Tuple
+
+from src.models.chatMessage import ChatMessage
+
+from src.integrations.panorama_gateway import (
+    PanoramaGatewayClient,
+    PanoramaGatewayError,
+    PanoramaGatewaySettings,
+    get_panorama_settings,
+)
+
+
+DEFAULT_DISCLAIMER = (
+    "This highly experimental chatbot is not intended for making important decisions. "
+    "Its responses are generated using AI models and may not always be accurate. "
+    "By using this chatbot, you acknowledge that you use it at your own discretion "
+    "and assume all risks associated with its limitations and potential errors."
+)
+
+
+def _utc_now_iso() -> str:
+    """Return an RFC3339 timestamp with millisecond precision and Z suffix."""
+    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
+def _normalize_datetime(value: Any | None) -> Optional[str]:
+    """Convert datetime-like inputs to gateway-friendly RFC3339 strings."""
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        dt = value
+    elif isinstance(value, str):
+        if not value:
+            return None
+        try:
+            dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return value
+    else:
+        return value
+
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    else:
+        dt = dt.astimezone(timezone.utc)
+    return dt.isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
+def _conversation_key(user_id: str, conversation_id: str) -> str:
+    return f"{user_id}:{conversation_id}"
+
+
+def _drop_none(data: Dict[str, Any]) -> Dict[str, Any]:
+    """Remove keys with None values so the gateway never receives JSON null."""
+    return {key: value for key, value in data.items() if value is not None}
+
+
+class PanoramaStore:
+    """Bridges chat persistence to the Panorama data gateway."""
+
+    def __init__(
+        self,
+        *,
+        client: PanoramaGatewayClient | None = None,
+        settings: PanoramaGatewaySettings | None = None,
+    ) -> None:
+        self._settings = settings or get_panorama_settings()
+        self._client = client or PanoramaGatewayClient(self._settings)
+        self._logger = logging.getLogger(__name__)
+
+    # ---- user helpers -----------------------------------------------------
+    def ensure_user(
+        self,
+        user_id: str,
+        *,
+        wallet_address: Optional[str] = None,
+        display_name: Optional[str] = None,
+        attributes: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        try:
+            user = self._client.get("users", user_id)
+        except PanoramaGatewayError as exc:
+            if exc.status_code != 404:
+                raise
+            user = None
+
+        if user:
+            # Update last seen without failing the request if the patch raises.
+            try:
+                payload: Dict[str, Any] = {"lastSeenAt": _utc_now_iso()}
+                if display_name:
+                    payload["displayName"] = display_name
+                if wallet_address:
+                    payload["walletAddress"] = wallet_address
+                if attributes:
+                    payload["attributes"] = attributes
+                self._client.update("users", user_id, payload)
+            except PanoramaGatewayError:
+                pass
+            return user
+
+        payload = _drop_none(
+            {
+                "userId": user_id,
+                "walletAddress": wallet_address,
+                "displayName": display_name,
+                "attributes": attributes or {},
+                "tenantId": self._settings.tenant_id,
+                "createdAt": _utc_now_iso(),
+                "lastSeenAt": _utc_now_iso(),
+            }
+        )
+        try:
+            return self._client.create("users", payload)
+        except PanoramaGatewayError as exc:
+            self._logger.error(
+                "Failed to create user %s via gateway: status=%s payload=%s request=%s",
+                user_id,
+                exc.status_code,
+                exc.payload,
+                payload,
+            )
+            raise
+
+    # ---- conversation helpers ---------------------------------------------
+    def ensure_conversation(
+        self,
+        user_id: str,
+        conversation_id: str,
+        *,
+        title: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        conv_key = _conversation_key(user_id, conversation_id)
+        try:
+            conversation = self._client.get("conversations", conv_key)
+        except PanoramaGatewayError as exc:
+            if exc.status_code != 404:
+                raise
+            conversation = None
+
+        if conversation:
+            return conversation
+
+        payload = _drop_none(
+            {
+                "id": conv_key,
+                "userId": user_id,
+                "conversationId": conversation_id,
+                "title": title,
+                "status": "active",
+                "messageCount": 0,
+                "tenantId": self._settings.tenant_id,
+                "contextState": {},
+                "memoryState": {},
+                "createdAt": _utc_now_iso(),
+                "updatedAt": _utc_now_iso(),
+            }
+        )
+        try:
+            conversation = self._client.create("conversations", payload)
+        except PanoramaGatewayError as exc:
+            self._logger.error(
+                "Failed to create conversation %s for user %s: status=%s payload=%s request=%s",
+                conversation_id,
+                user_id,
+                exc.status_code,
+                exc.payload,
+                payload,
+            )
+            raise
+        self._create_disclaimer_message(user_id, conversation_id)
+        return conversation
+
+    def list_conversations(self, user_id: str) -> List[str]:
+        result = self._client.list(
+            "conversations",
+            {
+                "where": {"userId": user_id},
+                "orderBy": {"updatedAt": "desc"},
+            },
+        )
+        data = result.get("data", []) if isinstance(result, dict) else []
+        return [item.get("conversationId") for item in data if item.get("conversationId")]
+
+    def list_users(self, limit: int = 1000) -> List[str]:
+        result = self._client.list(
+            "users",
+            {
+                "orderBy": {"createdAt": "desc"},
+                "take": limit,
+            },
+        )
+        data = result.get("data", []) if isinstance(result, dict) else []
+        return [item.get("userId") for item in data if item.get("userId")]
+
+    def delete_conversation(self, user_id: str, conversation_id: str) -> None:
+        messages = self.list_messages(user_id, conversation_id)
+        deletes: List[Dict[str, Any]] = [
+            {
+                "op": "delete",
+                "entity": "messages",
+                "args": {"id": message["messageId"]},
+            }
+            for message in messages
+            if message.get("messageId")
+        ]
+        if deletes:
+            self._client.transact(deletes)
+
+        conv_key = _conversation_key(user_id, conversation_id)
+        try:
+            self._client.delete("conversations", conv_key)
+        except PanoramaGatewayError as exc:
+            if exc.status_code != 404:
+                raise
+
+    def reset_conversation(self, user_id: str, conversation_id: str) -> None:
+        messages = self.list_messages(user_id, conversation_id)
+        deletes = [
+            {
+                "op": "delete",
+                "entity": "messages",
+                "args": {"id": message["messageId"]},
+            }
+            for message in messages
+            if message.get("messageId")
+        ]
+        if deletes:
+            self._client.transact(deletes)
+
+        conv_key = _conversation_key(user_id, conversation_id)
+        self._client.update(
+            "conversations",
+            conv_key,
+            {"messageCount": 0, "updatedAt": _utc_now_iso()},
+        )
+        self._create_disclaimer_message(user_id, conversation_id)
+
+    # ---- message helpers ---------------------------------------------------
+    def list_messages(self, user_id: str, conversation_id: str) -> List[Dict[str, Any]]:
+        result = self._client.list(
+            "messages",
+            {
+                "where": {"userId": user_id, "conversationId": conversation_id},
+                "orderBy": {"timestamp": "asc"},
+            },
+        )
+        if isinstance(result, dict):
+            return result.get("data", [])
+        return []
+
+    def add_message(
+        self,
+        user_id: str,
+        conversation_id: str,
+        message: ChatMessage,
+    ) -> Dict[str, Any]:
+        conversation = self.ensure_conversation(user_id, conversation_id)
+        message_dict = message.dict()
+        message_id = message_dict.get("message_id") or message_dict.get("messageId")
+        if not message_id:
+            message_id = str(uuid.uuid4())
+        timestamp = _normalize_datetime(message_dict.get("timestamp")) or _utc_now_iso()
+        message_payload = _drop_none(
+            {
+                "messageId": message_id,
+                "userId": user_id,
+                "conversationId": conversation_id,
+                "role": message_dict.get("role"),
+                "content": message_dict.get("content"),
+                "agentName": message_dict.get("agent_name"),
+                "agentType": message_dict.get("agent_type"),
+                "requiresAction": message_dict.get("requires_action", False),
+                "actionType": message_dict.get("action_type"),
+                "metadata": message_dict.get("metadata") or {},
+                "status": message_dict.get("status"),
+                "errorMessage": message_dict.get("error_message"),
+                "toolCalls": message_dict.get("tool_calls"),
+                "toolResults": message_dict.get("tool_results"),
+                "nextAgent": message_dict.get("next_agent"),
+                "requiresFollowup": message_dict.get("requires_followup", False),
+                "timestamp": timestamp,
+                "tenantId": self._settings.tenant_id,
+            }
+        )
+
+        operations = [
+            {"op": "create", "entity": "messages", "args": {"data": message_payload}},
+            {
+                "op": "update",
+                "entity": "conversations",
+                "args": {
+                    "id": conversation["id"],
+                    "data": {
+                        "lastMessageId": message_payload["messageId"],
+                        "messageCount": (conversation.get("messageCount") or 0) + 1,
+                        "updatedAt": _utc_now_iso(),
+                    },
+                },
+            },
+        ]
+        self._client.transact(operations)
+
+        memory_payload = _drop_none(
+            {
+                "messageId": message_payload["messageId"],
+                "role": message_payload["role"],
+                "content": message_payload["content"],
+                "agentName": message_payload.get("agentName"),
+                "agentType": message_payload.get("agentType"),
+                "metadata": message_payload.get("metadata"),
+                "requiresAction": message_payload.get("requiresAction"),
+                "timestamp": message_payload.get("timestamp"),
+            }
+        )
+        try:
+            self.create_conversation_memory(
+                user_id,
+                conversation_id,
+                scope="conversation",
+                memory_type=message_payload["role"],
+                payload=memory_payload,
+            )
+        except PanoramaGatewayError:
+            # Memory writes should never block the main chat flow.
+            pass
+
+        return message_payload
+
+    def _create_disclaimer_message(self, user_id: str, conversation_id: str) -> None:
+        disclaimer = ChatMessage(
+            role="assistant",
+            content=DEFAULT_DISCLAIMER,
+            metadata={},
+            user_id=user_id,
+            conversation_id=conversation_id,
+        )
+        self.add_message(user_id, conversation_id, disclaimer)
+
+    # ---- conversation memory ----------------------------------------------
+    def create_conversation_memory(
+        self,
+        user_id: str,
+        conversation_id: Optional[str],
+        scope: str,
+        memory_type: str,
+        payload: Dict[str, Any],
+        *,
+        label: Optional[str] = None,
+        importance_score: Optional[float] = None,
+        expires_at: Optional[Any] = None,
+    ) -> Dict[str, Any]:
+        memory_payload = _drop_none(
+            {
+                "userId": user_id,
+                "conversationId": conversation_id,
+                "scope": scope,
+                "memoryType": memory_type,
+                "payload": payload,
+                "tenantId": self._settings.tenant_id,
+                "label": label,
+                "importanceScore": importance_score,
+                "expiresAt": _normalize_datetime(expires_at),
+                "createdAt": _utc_now_iso(),
+                "updatedAt": _utc_now_iso(),
+            }
+        )
+        return self._client.create("conversation-memories", memory_payload)
+
+    # ---- utility -----------------------------------------------------------
+    def ensure_user_and_conversation(
+        self,
+        user_id: str,
+        conversation_id: str,
+        *,
+        wallet_address: Optional[str] = None,
+        display_name: Optional[str] = None,
+    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
+        user = self.ensure_user(
+            user_id,
+            wallet_address=wallet_address,
+            display_name=display_name,
+        )
+        conversation = self.ensure_conversation(user_id, conversation_id)
+        return user, conversation