core/deliberation_log.py

"""Deliberation log — live, turn-by-turn capture of the personas' argument.

Companion to field_log.py. Where the field log records the *interaction config*,
this records *how the agent reasoned*: each run appends the turns of the
multi-persona deliberation (O'Brien proposes -> Spine vetoes -> La Forge second
opinion / dispute -> operator override -> world simulates -> La Forge grades ->
run verdict) and CommitScheduler pushes them to an open HF Dataset.

Same guarantees as field_log:
- Gated on HF_TOKEN — nothing is written or pushed if the secret is absent.
- Best-effort + exception-safe — logging never breaks a run.
- Config + agent reasoning only; no PII, no uploaded mesh files.

Schema mirrors scripts/export_deliberation.py (one row per turn) so the live
dataset and the static export share the same shape:
  session_id, track, turn, agent, role, act, stance, content,
  material, geometry, bed_position, env_temp, env_humidity, ts
"""

from __future__ import annotations

import json
import os
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

DELIB_LOG_DIR = Path(__file__).resolve().parent.parent / "deliberation_logs"
DELIB_LOG_FILE = DELIB_LOG_DIR / "deliberations.jsonl"
DELIB_LOG_REPO = "kylebrodeur/chief-engineer-deliberation"
FLUSH_MINUTES = 5

ROLE = {
    "O'Brien": "Chief Engineer",
    "La Forge": "QA Inspector",
    "Spine": "Safety Spine",
    "World": "Outcome Simulator",
    "Operator": "Operator",
}

_CANON = (
    "session_id", "track", "turn", "agent", "role", "act", "stance", "content",
    "material", "geometry", "bed_position", "env_temp", "env_humidity", "ts",
)

_scheduler: Any = None
_lock = threading.Lock()
_turns: dict[str, int] = {}   # session_id -> last turn number (single process on the Space)


def _get_scheduler():
    """Lazy-init the CommitScheduler. Returns None if HF_TOKEN is missing."""
    global _scheduler
    token = os.environ.get("HF_TOKEN", "").strip()
    if not token:
        return None
    if _scheduler is None:
        with _lock:
            if _scheduler is None:
                try:
                    from huggingface_hub import CommitScheduler
                except ImportError:
                    return None
                DELIB_LOG_DIR.mkdir(parents=True, exist_ok=True)
                if not DELIB_LOG_FILE.exists():
                    DELIB_LOG_FILE.write_text("", encoding="utf-8")
                _scheduler = CommitScheduler(
                    repo_id=DELIB_LOG_REPO,
                    repo_type="dataset",
                    folder_path=str(DELIB_LOG_DIR),
                    every=FLUSH_MINUTES,
                    token=token,
                    allow_patterns=["*.jsonl"],
                )
    return _scheduler


def is_active() -> bool:
    """True if deliberation logging is live (HF_TOKEN present + scheduler ready)."""
    return _get_scheduler() is not None


def _next_turn(session_id: str) -> int:
    with _lock:
        n = _turns.get(session_id, 0) + 1
        _turns[session_id] = n
        return n


def log_turns(session_id: str, track: str, turns: list[dict], ctx: dict) -> bool:
    """Append a batch of deliberation turns for one phase of one run.

    `turns` is a list of {agent, act, content, stance?} dicts; `ctx` carries
    material/geometry/bed_position/env_temp/env_humidity. Gated + exception-safe:
    if HF_TOKEN is unset or anything fails, this is a silent no-op."""
    try:
        sched = _get_scheduler()
        if sched is None or not session_id or not turns:
            return False
        lines: list[str] = []
        for tn in turns:
            agent = tn.get("agent", "")
            row = {k: None for k in _CANON}
            row.update({
                "session_id": session_id, "track": track, "turn": _next_turn(session_id),
                "agent": agent, "role": ROLE.get(agent, agent),
                "act": tn.get("act"), "stance": tn.get("stance", ""),
                "content": (tn.get("content") or "").strip(),
                "material": ctx.get("material"), "geometry": ctx.get("geometry"),
                "bed_position": ctx.get("bed_position"),
                "env_temp": ctx.get("env_temp"), "env_humidity": ctx.get("env_humidity"),
                "ts": datetime.now(timezone.utc).isoformat(),
            })
            lines.append(json.dumps(row, ensure_ascii=False))
        with _lock:
            with DELIB_LOG_FILE.open("a", encoding="utf-8") as f:
                f.write("\n".join(lines) + "\n")
        try:
            sched.trigger()
        except Exception:
            pass
        return True
    except Exception:
        return False  # logging is best-effort — never break a run