box/bin/variables.py

#!/usr/bin/env python3
"""Box variable registry — single library for box-defined control variables.

Timers, wake digests, the siphon, the sweeper, and the silence detector all
read their thresholds here at runtime instead of hardcoding them. The user
adjusts behavior by changing variables in /srv/box/variables.json (via the
Box API or variablesctl), never by editing code.

Registry location:
  - Production: /srv/box/variables.json
  - Override:  BOX_VARIABLES env var, or explicit path argument

Fail-closed design: a missing file, a corrupt file, or a missing variable
falls back to the built-in default for that variable — timers keep working
and the miss is recorded in `warnings`.

Setting a variable validates:
  - type matches (int/float/bool/str)
  - min/max or enum constraints honored
  - unknown variable names are rejected (typo guard)

The registry file format:

    {"_meta": {...},
     "variables": {
       "max_nudge_count": {
         "value": 2, "default": 2, "type": "int",
         "min": 0, "max": 10, "unit": "count",
         "description": "..."
       }}}

Writes are atomic (tmp + rename). `updated_at` / `updated_by` in _meta track
who changed what.
"""

import json
import os
import time

# Primary and fallback registries
DEFAULT_REGISTRY = "/srv/box/variables.json"
FALLBACK_REGISTRY = "/home/super/Projects/NetVM/variables.json"
DEFAULT_HISTORY = "/srv/box/variables-history.jsonl"
FALLBACK_HISTORY = "/home/super/Projects/NetVM/variables-history.jsonl"

ENV_OVERRIDE = "BOX_VARIABLES"


class VariableError(ValueError):
    """Raised on unknown variable, type mismatch, or constraint violation."""
    pass


class Variables:
    """Typed access to the box variable registry."""

    def __init__(self, path=None, history_path=None, _data=None):
        env_path = os.environ.get(ENV_OVERRIDE)
        self.history_path = history_path
        if path:
            self.path = path
        elif env_path:

            self.path = env_path
        elif os.path.exists(DEFAULT_REGISTRY):
            self.path = DEFAULT_REGISTRY
        elif os.path.exists(FALLBACK_REGISTRY):
            self.path = FALLBACK_REGISTRY
        else:
            self.path = DEFAULT_REGISTRY

        self.warnings = []
        if _data is not None:
            self._data = _data
        else:
            self._data = self._load()

    def _load(self):
        loaded = {}
        target_path = self.path
        if not os.path.exists(target_path) and os.path.exists(FALLBACK_REGISTRY):
            target_path = FALLBACK_REGISTRY

        try:
            with open(target_path) as f:
                data = json.load(f)
            if isinstance(data, dict):
                loaded = data.get("variables", {})
        except FileNotFoundError:
            self.warnings.append("registry not found: %s" % target_path)
        except (OSError, ValueError) as e:
            self.warnings.append("registry unreadable (%s): using defaults" % e)

        # Seed with builtin specifications if missing
        specs = {}
        for k, s in _BUILTIN_SPECS.items():
            specs[k] = dict(s)
            if "value" not in specs[k]:
                specs[k]["value"] = s["default"]

        # Overlay any loaded values from file
        if isinstance(loaded, dict):
            for k, item in loaded.items():
                if k in specs and isinstance(item, dict):
                    specs[k].update(item)
                elif isinstance(item, dict):
                    specs[k] = item

        return specs

    def _spec(self, name):
        spec = self._data.get(name)
        if spec is None:
            if name in _BUILTIN_SPECS:
                self._data[name] = dict(_BUILTIN_SPECS[name])
                self._data[name]["value"] = self._data[name]["default"]
                return self._data[name]
            raise VariableError("unknown variable: %s" % name)
        return spec

    def _coerce(self, name, value, spec):
        vtype = spec.get("type", "int")
        try:
            if vtype == "int":
                if isinstance(value, bool):
                    raise VariableError("%s: expected int, got %r" % (name, value))
                if isinstance(value, str):
                    try:
                        value = int(value)
                    except ValueError:
                        raise VariableError("%s: expected int, got %r" % (name, value))
                elif not isinstance(value, (int, float)):
                    raise VariableError("%s: expected int, got %r" % (name, value))
                value = int(value)
            elif vtype == "float":
                if isinstance(value, bool):
                    raise VariableError("%s: expected float, got %r" % (name, value))
                if isinstance(value, str):
                    try:
                        value = float(value)
                    except ValueError:
                        raise VariableError("%s: expected float, got %r" % (name, value))
                elif not isinstance(value, (int, float)):
                    raise VariableError("%s: expected float, got %r" % (name, value))
                value = float(value)

            elif vtype == "bool":
                if not isinstance(value, bool):
                    raise VariableError(
                        "%s: expected bool, got %r" % (name, value))
            elif vtype == "str":
                if not isinstance(value, str):
                    raise VariableError(
                        "%s: expected str, got %r" % (name, value))
            else:
                raise VariableError("%s: unknown type %r" % (name, vtype))
        except VariableError:
            raise
        except (TypeError, ValueError):
            raise VariableError("%s: cannot coerce %r to %s" % (name, value, vtype))
        return value

    def _check_constraints(self, name, value, spec):
        enum = spec.get("enum")
        if enum is not None and value not in enum:
            raise VariableError(
                "%s: %r not in allowed values %r" % (name, value, enum))
        for bound, key in (("min", "min"), ("max", "max")):
            limit = spec.get(key)
            if limit is not None and isinstance(value, (int, float)):
                if bound == "min" and value < limit:
                    raise VariableError(
                        "%s: %r below minimum %r" % (name, value, limit))
                if bound == "max" and value > limit:
                    raise VariableError(
                        "%s: %r above maximum %r" % (name, value, limit))

    # -- reads -----------------------------------------------------------

    def get(self, name):
        """Return the current value; fall back to the built-in default when
        the registry is missing the variable (never raises on read)."""
        spec = self._data.get(name)
        if spec is None:
            self.warnings.append("variable %s missing: using default" % name)
            return _BUILTIN_DEFAULTS.get(name)
        value = spec.get("value", spec.get("default"))
        if value is None:
            value = _BUILTIN_DEFAULTS.get(name)
        return value

    def get_int(self, name):
        return int(self.get(name))

    def get_float(self, name):
        return float(self.get(name))

    def get_bool(self, name):
        return bool(self.get(name))

    def get_str(self, name):
        return str(self.get(name))

    def all(self):
        """Dict of name -> value for every registered variable."""
        return {name: self.get(name) for name in self._data}

    def schema(self, name):
        """Full spec dict for a variable (value, default, type, constraints)."""
        return dict(self._spec(name))

    def names(self):
        return sorted(self._data.keys())

    # -- writes ----------------------------------------------------------

    def set(self, name, value, by=None):
        """Validate and persist a new value. Returns the stored value.

        Raises VariableError on unknown name, type mismatch, or constraint
        violation. The file is rewritten atomically.
        """
        old_val = self.get(name)
        spec = self._spec(name)
        value = self._coerce(name, value, spec)
        self._check_constraints(name, value, spec)
        spec["value"] = value
        self._persist(by=by)
        if old_val != value:
            self._record_history("set", name, old_val, value, by=by)
        return value

    def reset(self, name, by=None):
        """Restore a variable to its default value. Returns the default."""
        old_val = self.get(name)
        spec = self._spec(name)
        default_val = spec.get("default")
        val = self._coerce(name, default_val, spec)
        self._check_constraints(name, val, spec)
        spec["value"] = val
        self._persist(by=by)
        if old_val != val:
            self._record_history("reset", name, old_val, val, by=by)
        return val

    def rollback(self, name, revision=None, by=None):
        """Roll back a variable to a previous value from history.
        
        If revision is None: rolls back to the old_value of the most recent change.
        If revision is an integer k >= 1: rolls back k revisions back.
        If revision is a timestamp string: rolls back to that revision's old_value.
        """
        entries = self.history(name=name, limit=1000)
        # self.history returns newest first, so reverse to get chronological
        chronological = list(reversed(entries))
        if not chronological:
            raise VariableError("no change history found to rollback for variable '%s'" % name)

        target_value = None
        if revision is None:
            target_value = chronological[-1].get("old_value")
        elif str(revision).isdigit() and int(revision) >= 1:
            k = int(revision)
            if k <= len(chronological):
                target_value = chronological[-k].get("old_value")
            else:
                raise VariableError("revision step %d exceeds history depth (%d) for '%s'" % (k, len(chronological), name))
        else:
            rev_str = str(revision).strip()
            match = None
            for e in reversed(chronological):
                if e.get("ts") == rev_str:
                    match = e
                    break
            if match is not None:
                target_value = match.get("old_value")
            else:
                raise VariableError("revision '%s' not found in history for '%s'" % (rev_str, name))

        if target_value is None:
            raise VariableError("unable to determine rollback target value for '%s'" % name)

        old_val = self.get(name)
        spec = self._spec(name)
        val = self._coerce(name, target_value, spec)
        self._check_constraints(name, val, spec)
        spec["value"] = val
        caller = by or os.environ.get("BOX_CALLER", "rollback")
        self._persist(by=caller)
        self._record_history("rollback", name, old_val, val, by=caller)
        return val

    def history(self, name=None, limit=20):
        """Return history entries (newest first), optionally filtered by name."""
        entries = []
        target = self.history_path or (DEFAULT_HISTORY if os.path.exists(DEFAULT_HISTORY) else FALLBACK_HISTORY)
        if os.path.exists(target):
            try:
                with open(target, "r") as f:
                    for line in f:
                        line = line.strip()
                        if not line:
                            continue
                        try:
                            entries.append(json.loads(line))
                        except Exception:
                            pass
            except Exception:
                pass
        elif not self.history_path and target == DEFAULT_HISTORY and os.path.exists(FALLBACK_HISTORY):
            try:
                with open(FALLBACK_HISTORY, "r") as f:
                    for line in f:
                        line = line.strip()
                        if not line:
                            continue
                        try:
                            entries.append(json.loads(line))
                        except Exception:
                            pass
            except Exception:
                pass

        if name:
            entries = [e for e in entries if e.get("name") == name]
        entries.reverse()
        return entries[:limit]

    def _record_history(self, action, name, old_val, new_val, by=None):
        entry = {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "action": action,
            "name": name,
            "old_value": old_val,
            "new_value": new_val,
            "by": by or os.environ.get("BOX_CALLER", "super"),
        }
        line = json.dumps(entry) + "\n"
        targets = [self.history_path] if self.history_path else [DEFAULT_HISTORY, FALLBACK_HISTORY]
        for hpath in targets:
            try:
                os.makedirs(os.path.dirname(os.path.abspath(hpath)), exist_ok=True)
                with open(hpath, "a") as f:
                    f.write(line)
            except Exception:
                pass


    def _persist(self, by=None):
        # Build the full file payload, preserving any existing _meta shape.
        payload = {"_meta": {}, "variables": self._data}
        try:
            with open(self.path) as f:
                existing = json.load(f)
            if isinstance(existing, dict) and isinstance(existing.get("_meta"), dict):
                payload["_meta"] = existing["_meta"]
        except (OSError, ValueError):
            pass
        meta = payload["_meta"]
        meta["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        if by:
            meta["updated_by"] = by
        meta.setdefault("schema_version", 1)

        def _write(dest_path):
            try:
                os.makedirs(os.path.dirname(os.path.abspath(dest_path)), exist_ok=True)
                tmp = dest_path + f".tmp.{os.getpid()}"
                with open(tmp, "w") as f:
                    json.dump(payload, f, indent=2)
                    f.write("\n")
                os.chmod(tmp, 0o644)
                os.replace(tmp, dest_path)
            except Exception:
                pass

        _write(self.path)
        # Mirror to fallback or primary if different
        if self.path == DEFAULT_REGISTRY:
            _write(FALLBACK_REGISTRY)
        elif self.path == FALLBACK_REGISTRY:
            _write(DEFAULT_REGISTRY)


# Detailed schemas for every built-in control variable
_BUILTIN_SPECS = {
    "agent_activity_threshold": {
        "default": 43200, "type": "int", "min": 60, "max": 604800, "unit": "seconds",
        "description": "Seconds of silence before agent is considered inactive (12h default)"
    },
    "followup_default_timeout_s": {
        "default": 3600, "type": "int", "min": 60, "max": 604800, "unit": "seconds",
        "description": "Default timeout before nudging unacknowledged DMs (1h default)"
    },
    "keepalive_idle_s": {
        "default": 3600, "type": "int", "min": 60, "max": 86400, "unit": "seconds",
        "description": "Idle threshold before keepalive pings a thread (1h default)"
    },
    "keepalive_interval_s": {
        "default": 300, "type": "int", "min": 30, "max": 3600, "unit": "seconds",
        "description": "Cadence between keepalive checks (5m default)"
    },
    "loop_health_interval_s": {
        "default": 900, "type": "int", "min": 60, "max": 86400, "unit": "seconds",
        "description": "Sampling interval for agent loop health metrics (15m default)"
    },
    "loop_health_threshold": {
        "default": 0.5, "type": "float", "min": 0.0, "max": 1.0, "unit": "ratio",
        "description": "Minimum healthy ratio of answered+closed loops to landed loops"
    },
    "max_nudge_count": {
        "default": 2, "type": "int", "min": 0, "max": 10, "unit": "count",
        "description": "Maximum automated nudges before escalating to supervisor"
    },
    "reminder_min_interval_s": {
        "default": 900, "type": "int", "min": 60, "max": 86400, "unit": "seconds",
        "description": "Minimum cooldown between reminders (15m default)"
    },
    "reminder_quiet_hours_start": {
        "default": 22, "type": "int", "min": 0, "max": 23, "unit": "hour",
        "description": "Start of quiet hours (UTC, 22 = 10pm)"
    },
    "reminder_quiet_hours_end": {
        "default": 7, "type": "int", "min": 0, "max": 23, "unit": "hour",
        "description": "End of quiet hours (UTC, 7 = 7am)"
    },
    "silence_alert_hours": {
        "default": 12.0, "type": "float", "min": 1.0, "max": 168.0, "unit": "hours",
        "description": "Hours of agent silence before firing operator alert"
    },
    "siphon_interval_s": {
        "default": 60, "type": "int", "min": 10, "max": 3600, "unit": "seconds",
        "description": "Cadence of sidechat work siphon poll loop (60s default)"
    },
    "siphon_min_confidence": {
        "default": 0.6, "type": "float", "min": 0.0, "max": 1.0, "unit": "confidence",
        "description": "Minimum confidence threshold to siphon sidechat item to main"
    },
    "siphon_rate_limit_per_hour": {
        "default": 5, "type": "int", "min": 1, "max": 60, "unit": "count/hr",
        "description": "Max siphoned items per hour to prevent spam"
    },
    "sweeper_interval_s": {
        "default": 60, "type": "int", "min": 10, "max": 3600, "unit": "seconds",
        "description": "Cadence of followup deadline sweeper loop (60s default)"
    },
    "thread_max_age_s": {
        "default": 604800, "type": "int", "min": 3600, "max": 2592000, "unit": "seconds",
        "description": "Maximum sidechat thread age before triggering rotation (7d default)"
    },
    "thread_max_messages": {
        "default": 200, "type": "int", "min": 10, "max": 10000, "unit": "messages",
        "description": "Maximum messages in thread before triggering rotation"
    },
    "wake_interval_s": {
        "default": 1800, "type": "int", "min": 60, "max": 86400, "unit": "seconds",
        "description": "Cadence between periodic wake digests (30m default)"
    },
    "wake_reply_nudges": {
        "default": 1, "type": "int", "min": 0, "max": 10, "unit": "count",
        "description": "Max nudges for unacknowledged wake digests"
    },
    "wake_reply_timeout_s": {
        "default": 7200, "type": "int", "min": 60, "max": 604800, "unit": "seconds",
        "description": "Timeout before nudging unacknowledged wake digests (2h default)"
    },
}

# Built-in fallbacks
_BUILTIN_DEFAULTS = {k: v["default"] for k, v in _BUILTIN_SPECS.items()}


def main():
    import argparse
    parser = argparse.ArgumentParser(description="Box variable registry CLI")
    sub = parser.add_subparsers(dest="action")

    sub.add_parser("list", help="List all variables")
    
    p_get = sub.add_parser("get", help="Get a variable")
    p_get.add_argument("name", help="Variable name")

    p_set = sub.add_parser("set", help="Set a variable")
    p_set.add_argument("name", help="Variable name")
    p_set.add_argument("value", help="New value")

    p_reset = sub.add_parser("reset", help="Reset a variable to default")
    p_reset.add_argument("name", help="Variable name")

    p_hist = sub.add_parser("history", help="Show variable change history")
    p_hist.add_argument("name", nargs="?", default=None, help="Filter by variable name")
    p_hist.add_argument("-n", "--limit", type=int, default=20, help="Number of records to show")

    p_rb = sub.add_parser("rollback", help="Roll back a variable to previous value")
    p_rb.add_argument("name", help="Variable name")
    p_rb.add_argument("--revision", default=None, help="Revision step (int) or timestamp")

    args = parser.parse_args()
    v = Variables()

    if not args.action or args.action == "list":
        for name in v.names():
            val = v.get(name)
            spec = v.schema(name)
            print(f"{name:28} = {val!r:10} (default: {spec['default']!r}, unit: {spec.get('unit', '-')})")
    elif args.action == "get":
        print(json.dumps({"name": args.name, "value": v.get(args.name), "schema": v.schema(args.name)}, indent=2))
    elif args.action == "set":
        spec = v.schema(args.name)
        vtype = spec.get("type", "str")
        if vtype == "int":
            val = int(args.value)
        elif vtype == "float":
            val = float(args.value)
        elif vtype == "bool":
            val = args.value.lower() in ("true", "1", "yes")
        else:
            val = args.value
        res = v.set(args.name, val, by=os.environ.get("BOX_CALLER", "cli"))
        print(f"Updated {args.name} = {res}")
    elif args.action == "reset":
        res = v.reset(args.name, by=os.environ.get("BOX_CALLER", "cli"))
        print(f"Reset {args.name} = {res}")
    elif args.action == "history":
        entries = v.history(name=args.name, limit=args.limit)
        if not entries:
            print("No history entries found.")
        else:
            print(f"\n{'TIMESTAMP':22} {'VARIABLE':28} {'ACTION':10} {'OLD':10} {'NEW':10} {'BY':10}")
            print("-" * 92)
            for e in entries:
                print(f"{e.get('ts',''):22} {e.get('name',''):28} {e.get('action',''):10} {str(e.get('old_value',''))[:10]:10} {str(e.get('new_value',''))[:10]:10} {e.get('by',''):10}")
            print()
    elif args.action == "rollback":
        caller = os.environ.get("BOX_CALLER", "cli")
        res = v.rollback(args.name, revision=args.revision, by=caller)
        print(f"Rolled back {args.name} = {res}")


if __name__ == "__main__":
    main()