config.py

"""Lightweight OpenViking integration helpers.

This module intentionally has zero OpenSpace-internal dependencies so
that identity resolution, opt-out handling, and per-execution telemetry
can be unit tested without pulling in the full agent / LLM / backends
stack.
"""
from __future__ import annotations

import os
from dataclasses import dataclass, field
from typing import Any, Dict, Tuple


def resolve_viking_identity(
    namespace_override: str = "",
    user_id_override: str = "",
) -> Tuple[str, str]:
    """Resolve the effective ``(namespace, user_id)`` pair for Viking scoping.

    Resolution order (first non-empty wins at each slot):

    * ``namespace``: config override → ``OPENVIKING_NAMESPACE`` env → ``""``
    * ``user_id``:   config override → ``OPENVIKING_USER_ID`` env →
      ``USER`` env (Unix) → ``USERNAME`` env (Windows) → ``""``

    The namespace is never auto-derived: team deployment must set it
    explicitly to avoid accidental cross-team memory mixing. The user
    id, by contrast, falls back to the OS user so that on single-machine
    dev setups user preferences are isolated per OS account without any
    configuration.

    Both values are sanitized to keep URIs path-safe — only alphanum,
    ``-``, ``_`` and ``.`` survive.
    """
    ns = (namespace_override or os.environ.get("OPENVIKING_NAMESPACE", "") or "").strip()

    uid = (user_id_override or os.environ.get("OPENVIKING_USER_ID", "") or "").strip()
    if not uid:
        uid = (
            os.environ.get("USER", "") or os.environ.get("USERNAME", "") or ""
        ).strip()

    def _clean(s: str) -> str:
        return "".join(c for c in s if c.isalnum() or c in "-_.").strip("-_.")

    return _clean(ns), _clean(uid)


def resolve_viking_min_score(config_default: float) -> float:
    """Return the effective minimum retrieval score threshold.

    Env var ``OPENVIKING_MIN_SCORE`` overrides the config default when
    set to a parseable float. Unparseable values fall back to config.

    A score of 0 means "accept everything"; typical quality threshold
    is 0.5. Hits with scores below the threshold are dropped both
    server-side (via Viking's payload) and client-side (safety net).
    """
    raw = os.environ.get("OPENVIKING_MIN_SCORE", "").strip()
    if raw:
        try:
            val = float(raw)
            if 0.0 <= val <= 1.0:
                return val
        except ValueError:
            pass
    return max(0.0, min(1.0, float(config_default)))


def resolve_viking_scrub_pii(config_default: bool) -> bool:
    """Return whether PII scrubbing is enabled before writing to Viking.

    ``OPENVIKING_SCRUB_PII`` env var overrides config default. Default
    ``True`` because leaking secrets is harder to undo than leaving
    Viking memories slightly less specific.
    """
    raw = os.environ.get("OPENVIKING_SCRUB_PII", "").strip().lower()
    if raw in ("1", "true", "yes", "on"):
        return True
    if raw in ("0", "false", "no", "off"):
        return False
    return config_default


def resolve_viking_push_enabled(config_default: bool) -> bool:
    """Return the effective ``auto push evolved skills`` flag.

    ``OPENVIKING_PUSH_SKILLS`` env var (when set to a truthy/falsy
    string) overrides the config default. Accepted values:
      * truthy: ``1``, ``true``, ``yes``, ``on``
      * falsy:  ``0``, ``false``, ``no``, ``off``
    Unknown values fall back to the config default.
    """
    raw = os.environ.get("OPENVIKING_PUSH_SKILLS", "").strip().lower()
    if raw in ("1", "true", "yes", "on"):
        return True
    if raw in ("0", "false", "no", "off"):
        return False
    return config_default


@dataclass
class VikingExecutionStats:
    """Per-execution OpenViking telemetry.

    Attached to the return value of ``OpenSpace.execute()`` under the
    ``viking`` key so external callers (benchmarks, observability
    pipelines, MCP host agents) can measure the cross-session memory
    layer's real-world impact without parsing logs.

    Fields:
      enabled: Integration is turned on and a client was constructed.
      available: The Viking server responded OK for at least one probe
        during this execution.
      query: The composed query string sent to Viking (task + history).
      enrichment_chars: Size of the context block injected into the
        grounding agent's iter-1 prompt (0 if nothing matched).
      hit_counts: Per-category count of L0 abstracts retrieved.
      selector_hints_chars: Size of the shorter hints block passed to
        the skill selection LLM.
      analysis_context_used: True if the analyzer prompt was augmented
        with Viking context in this execution.
      feedback_status: One of ``skipped`` (no evolution / disabled),
        ``attempted``, ``committed``, ``failed``.
      pushed_skills: Number of evolved SKILL.md resources successfully
        pushed to Viking via ``/api/v1/skills``.
    """
    enabled: bool = False
    available: bool = False
    query: str = ""
    enrichment_chars: int = 0
    hit_counts: Dict[str, int] = field(default_factory=dict)
    selector_hints_chars: int = 0
    analysis_context_used: bool = False
    feedback_status: str = "skipped"  # skipped | attempted | committed | failed | disabled
    pushed_skills: int = 0

    def to_dict(self) -> Dict[str, Any]:
        return {
            "enabled": self.enabled,
            "available": self.available,
            "query": self.query,
            "enrichment_chars": self.enrichment_chars,
            "hit_counts": dict(self.hit_counts),
            "selector_hints_chars": self.selector_hints_chars,
            "analysis_context_used": self.analysis_context_used,
            "feedback_status": self.feedback_status,
            "pushed_skills": self.pushed_skills,
        }