client.py

"""OpenViking integration client for OpenSpace.

Provides async HTTP client to query OpenViking's Context Database.
All methods are non-blocking and fail gracefully (return empty results
if OpenViking is unavailable). This ensures OpenSpace works normally
with or without OpenViking.

Environment variables:
  OPENVIKING_URL        — Server URL (default: http://127.0.0.1:1933)
  OPENVIKING_API_KEY    — Optional API key for authenticated access
  OPENVIKING_NAMESPACE  — Optional tenant/team namespace prefix applied to
                          target URIs (e.g. "tenants/acme"). When set, all
                          memory queries and pushes are scoped to that
                          prefix, enabling multi-team isolation on a shared
                          Viking instance.
  OPENVIKING_ENABLED    — "true"/"false" to explicitly enable/disable
                          (default: auto-detect)
"""
from __future__ import annotations

import os
import time
from typing import Any, Dict, List, Optional

from openspace.utils.logging import Logger

logger = Logger.get_logger(__name__)


_DEFAULT_URL = "http://127.0.0.1:1933"
_REQUEST_TIMEOUT = 5.0

# Two-tier health cache: once we see the server healthy we trust it for
# 60s; when it's unavailable we re-probe more aggressively so fresh deploys
# don't stay "down" for a whole minute after startup delay.
_HEALTH_CACHE_TTL_OK = 60.0
_HEALTH_CACHE_TTL_FAIL = 12.0


class OpenVikingClient:
    """Async HTTP client for the OpenViking Context Database.

    All methods are best-effort: on any failure they return an empty
    result (``{}`` / ``[]``) and never raise. ``is_available()`` caches
    its result with different TTLs for success vs failure.

    The optional ``namespace`` parameter prefixes the virtual-filesystem
    URIs used by ``find_*`` and ``push_*`` helpers, enabling per-team
    isolation on a shared Viking backend. Agent memories and user
    memories have different isolation policies by convention:

      * ``viking://tenants/{ns}/agent/memories/...`` — shared within team
      * ``viking://tenants/{ns}/user/{user_id}/memories/...`` — per user
    """

    def __init__(
        self,
        base_url: Optional[str] = None,
        api_key: Optional[str] = None,
        namespace: Optional[str] = None,
        user_id: Optional[str] = None,
    ) -> None:
        url = base_url or os.environ.get("OPENVIKING_URL") or _DEFAULT_URL
        self.base_url = url.rstrip("/")
        self.api_key = api_key or os.environ.get("OPENVIKING_API_KEY") or ""

        ns = namespace if namespace is not None else os.environ.get("OPENVIKING_NAMESPACE", "")
        self.namespace = ns.strip().strip("/") if ns else ""

        uid = user_id if user_id is not None else os.environ.get("OPENVIKING_USER_ID", "")
        self.user_id = uid.strip().strip("/") if uid else ""

        self._available: Optional[bool] = None
        self._available_checked_at: float = 0.0

        self._client = None  # lazy httpx.AsyncClient

    # ------------------------------------------------------------------
    # URI helpers
    # ------------------------------------------------------------------
    def _ns_prefix(self) -> str:
        """Return the optional ``tenants/{namespace}/`` path component."""
        if not self.namespace:
            return ""
        return f"tenants/{self.namespace}/"

    def agent_memory_uri(self, category: str) -> str:
        """Build a team-shared agent-memory URI.

        ``category`` is one of: ``tools``, ``patterns``, ``skills``, ``cases``.
        """
        return f"viking://{self._ns_prefix()}agent/memories/{category}/"

    def user_memory_uri(self, category: str) -> str:
        """Build a per-user memory URI.

        ``category`` is typically ``preferences``, ``profile``, ``entities``, ``events``.
        When ``user_id`` is not set, falls back to the legacy global path
        ``viking://user/memories/<category>/`` to preserve single-tenant behavior.
        """
        if self.user_id:
            return (
                f"viking://{self._ns_prefix()}user/{self.user_id}/memories/{category}/"
            )
        return f"viking://{self._ns_prefix()}user/memories/{category}/"

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------
    def _headers(self) -> Dict[str, str]:
        h = {"Content-Type": "application/json"}
        if self.api_key:
            h["X-API-Key"] = self.api_key
        return h

    async def _get_client(self):
        """Lazily create the underlying httpx.AsyncClient."""
        if self._client is not None:
            return self._client
        try:
            import httpx
        except ImportError:
            logger.debug("httpx not installed — OpenViking disabled")
            return None
        self._client = httpx.AsyncClient(timeout=_REQUEST_TIMEOUT)
        return self._client

    async def _post_json(
        self, path: str, payload: Dict[str, Any]
    ) -> Optional[Dict[str, Any]]:
        """POST JSON to the given path. Return parsed body or ``None`` on failure."""
        client = await self._get_client()
        if client is None:
            return None
        try:
            resp = await client.post(
                f"{self.base_url}{path}",
                json=payload,
                headers=self._headers(),
                timeout=_REQUEST_TIMEOUT,
            )
            if resp.status_code >= 400:
                logger.debug(
                    f"OpenViking POST {path} returned {resp.status_code}"
                )
                return None
            try:
                return resp.json()
            except Exception:
                return {}
        except Exception as exc:
            logger.debug(f"OpenViking POST {path} failed: {exc}")
            return None

    async def _get_json(self, path: str) -> Optional[Dict[str, Any]]:
        """GET JSON from the given path. Return parsed body or ``None`` on failure."""
        client = await self._get_client()
        if client is None:
            return None
        try:
            resp = await client.get(
                f"{self.base_url}{path}",
                headers=self._headers(),
                timeout=_REQUEST_TIMEOUT,
            )
            if resp.status_code >= 400:
                return None
            try:
                return resp.json()
            except Exception:
                return {}
        except Exception as exc:
            logger.debug(f"OpenViking GET {path} failed: {exc}")
            return None

    # ------------------------------------------------------------------
    # Health
    # ------------------------------------------------------------------
    async def is_available(self) -> bool:
        """Return True if the OpenViking server is reachable.

        Uses tiered caching: successful probes are trusted for 60s,
        failures for only ~12s so a slow-starting Viking isn't locked
        out for a full minute after OpenSpace initialization.
        """
        now = time.monotonic()
        if self._available is not None:
            ttl = _HEALTH_CACHE_TTL_OK if self._available else _HEALTH_CACHE_TTL_FAIL
            if (now - self._available_checked_at) < ttl:
                return self._available

        result = await self._get_json("/health")
        self._available = result is not None
        self._available_checked_at = now
        return self._available

    # ------------------------------------------------------------------
    # Search / find
    # ------------------------------------------------------------------
    async def find_memories(
        self,
        query: str,
        target_uri: str = "",
        limit: int = 5,
        score_threshold: float = 0.0,
    ) -> List[Dict[str, Any]]:
        """Query OpenViking for MatchedContext items.

        POST ``/api/v1/search/find`` with ``{query, target_uri, limit, score_threshold}``.
        Always returns a list — ``[]`` on any failure.

        ``score_threshold`` is passed through to OpenViking which filters
        results by vector similarity score before returning. Set to
        ``0.0`` to accept everything; typical quality threshold is ``0.5``.
        Low-quality hits are also post-filtered client-side on the
        returned ``score`` field to guard against server-side variations.
        """
        if not query:
            return []
        payload: Dict[str, Any] = {"query": query, "limit": limit}
        if target_uri:
            payload["target_uri"] = target_uri
        if score_threshold and score_threshold > 0:
            payload["score_threshold"] = score_threshold

        result = await self._post_json("/api/v1/search/find", payload)
        if not result:
            return []

        # Normalize response: OpenViking FindResult may return dict with
        # "memories"/"resources"/"skills" lists OR a flat list of matches.
        matches: List[Dict[str, Any]] = []
        if isinstance(result, list):
            matches = [m for m in result if isinstance(m, dict)]
        elif isinstance(result, dict):
            for key in ("memories", "resources", "skills", "matches", "results"):
                bucket = result.get(key)
                if isinstance(bucket, list):
                    matches.extend(m for m in bucket if isinstance(m, dict))
            if not matches and "data" in result:
                data = result["data"]
                if isinstance(data, list):
                    matches.extend(m for m in data if isinstance(m, dict))
                elif isinstance(data, dict):
                    for key in ("memories", "resources", "skills", "matches", "results"):
                        bucket = data.get(key)
                        if isinstance(bucket, list):
                            matches.extend(
                                m for m in bucket if isinstance(m, dict)
                            )

        # Client-side score enforcement as a safety net — even if the
        # server ignored score_threshold, we drop low-quality results
        # before they reach the prompt.
        if score_threshold and score_threshold > 0:
            matches = [
                m for m in matches
                if not isinstance(m.get("score"), (int, float))
                or m.get("score", 0) >= score_threshold
            ]
        return matches[:limit]

    async def find_tool_knowledge(
        self, query: str, limit: int = 5, score_threshold: float = 0.0
    ) -> List[Dict[str, Any]]:
        return await self.find_memories(
            query,
            target_uri=self.agent_memory_uri("tools"),
            limit=limit,
            score_threshold=score_threshold,
        )

    async def find_pattern_knowledge(
        self, query: str, limit: int = 5, score_threshold: float = 0.0
    ) -> List[Dict[str, Any]]:
        return await self.find_memories(
            query,
            target_uri=self.agent_memory_uri("patterns"),
            limit=limit,
            score_threshold=score_threshold,
        )

    async def find_skill_knowledge(
        self, query: str, limit: int = 5, score_threshold: float = 0.0
    ) -> List[Dict[str, Any]]:
        return await self.find_memories(
            query,
            target_uri=self.agent_memory_uri("skills"),
            limit=limit,
            score_threshold=score_threshold,
        )

    async def find_user_preferences(
        self, query: str, limit: int = 3, score_threshold: float = 0.0
    ) -> List[Dict[str, Any]]:
        return await self.find_memories(
            query,
            target_uri=self.user_memory_uri("preferences"),
            limit=limit,
            score_threshold=score_threshold,
        )

    async def find_cases(
        self, query: str, limit: int = 3, score_threshold: float = 0.0
    ) -> List[Dict[str, Any]]:
        return await self.find_memories(
            query,
            target_uri=self.agent_memory_uri("cases"),
            limit=limit,
            score_threshold=score_threshold,
        )

    async def find_antipatterns(
        self, query: str, limit: int = 3, score_threshold: float = 0.0
    ) -> List[Dict[str, Any]]:
        """Retrieve known failure modes / anti-patterns.

        Anti-patterns are negative memories — things that did NOT work in
        the past. Surfaced to the agent as warnings, not suggestions.
        """
        return await self.find_memories(
            query,
            target_uri=self.agent_memory_uri("antipatterns"),
            limit=limit,
            score_threshold=score_threshold,
        )

    # ------------------------------------------------------------------
    # Sessions (feedback channel)
    # ------------------------------------------------------------------
    async def create_session(self, session_id: str) -> Dict[str, Any]:
        result = await self._post_json(
            "/api/v1/sessions",
            {"session_id": session_id},
        )
        return result or {}

    async def add_session_message(
        self,
        session_id: str,
        role: str,
        text: str,
    ) -> Dict[str, Any]:
        result = await self._post_json(
            f"/api/v1/sessions/{session_id}/messages",
            {"role": role, "parts": [{"type": "text", "text": text}]},
        )
        return result or {}

    async def commit_session(self, session_id: str) -> Dict[str, Any]:
        result = await self._post_json(
            f"/api/v1/sessions/{session_id}/commit",
            {},
        )
        return result or {}

    # ------------------------------------------------------------------
    # Resources — push structured skill content
    # ------------------------------------------------------------------
    async def delete_resource(self, uri: str) -> Dict[str, Any]:
        """Request Viking to delete (or deprecate) a resource by URI.

        Used by the ``forget_memory`` and ``report_stale_memory`` feedback
        paths so agents can actively retire bad context. Failures are
        silent — Viking may not support deletion or may have RBAC rules.
        """
        if not uri:
            return {}
        client = await self._get_client()
        if client is None:
            return {}
        try:
            resp = await client.request(
                "DELETE",
                f"{self.base_url}/api/v1/resources",
                params={"uri": uri},
                headers=self._headers(),
                timeout=_REQUEST_TIMEOUT,
            )
            if resp.status_code >= 400:
                return {}
            try:
                return resp.json()
            except Exception:
                return {"status": "ok"}
        except Exception as exc:
            logger.debug(f"OpenViking DELETE resource failed: {exc}")
            return {}

    async def add_skill_data(
        self,
        data: Any,
        wait: bool = False,
    ) -> Dict[str, Any]:
        """Push inline skill content to OpenViking.

        Targets ``POST /api/v1/skills`` with ``{data, wait}``. ``data``
        may be a markdown string (SKILL.md body) or a structured dict
        the Viking server understands. The call is fire-and-forget by
        default so OpenSpace's main execute() path is never blocked.
        """
        result = await self._post_json(
            "/api/v1/skills",
            {"data": data, "wait": wait},
        )
        return result or {}

    # ------------------------------------------------------------------
    # Teardown
    # ------------------------------------------------------------------
    async def close(self) -> None:
        if self._client is not None:
            try:
                await self._client.aclose()
            except Exception:
                pass
            self._client = None