self-directed-learning/scripts/judge.py

#!/usr/bin/env python3
"""
Triadic Self-Learning Review — Phase 7.

Implements the A/B/C triad model for self-improvement:
  A = Hermes    (the doer)         — produces work, writes memories
  B = Krystie   (the critic)       — challenges, finds gaps
  C = Codex + Claude (the judges)  — two external models, structured rubric

Architecture:
  - judge.py brief-format: structured input to judges, no UMP context
  - judge.py orchestrates: Claude CLI subprocess + OpenAI API (Codex)
  - Convergence logic: 2-of-2 agree → accepted; disagree → conflict

Usage:
  judge.py --claim "I should always pre-cache embeddings before queries"
  judge.py --memory-urn urn:ump:abc123
  judge.py --dispute path/to/a-b-exchange.txt
  judge.py --dry-run --claim "..."

Exit codes:
  0 = both judges accept (or both reject with strong consensus)
  1 = judges disagree (needs A/B to resolve or human review)
  2 = error

Output (stdout, JSON):
  {
    "verdict": "accepted" | "rejected" | "conflict",
    "claude": {"score": 1-5, "reasoning": "..."},
    "codex":  {"score": 1-5, "reasoning": "..."},
    "consensus_strength": 0.0-1.0
  }
"""
import argparse
import json
import re
import subprocess
import sys
import time
import urllib.parse
import urllib.request

JUDGE_PROMPT = """You are a neutral external judge in a triadic self-learning system. The doer and the critic (two AI agents) have produced a claim for your review. They share context, training, and memory. You do NOT — you have no prior context, no memory of them, and your job is to render an independent verdict.

THE CLAIM:
\"\"\"{claim}\"\"\"

CONTEXT (optional, provided by A or B):
\"\"\"{context}\"\"\"

RUBRIC — score the claim on each dimension 1-5 (5 = excellent, 1 = poor):
1. **Truth**: Is the claim factually correct? Check against your knowledge of how systems actually work.
2. **Specificity**: Is it concrete and actionable, or vague platitude?
3. **Testability**: Can the claim be verified empirically? Is there a way to confirm or refute it?
4. **Novelty**: Does this add something genuinely new, or is it a restatement of common practice?
5. **Harmlessness**: Will following this advice cause damage, leak sensitive data, or create security issues?

Then write a 2-3 sentence verdict: ACCEPT (claim is solid, write to memory), REJECT (claim is wrong, do not write), or REVISE (claim is on the right track but needs changes — suggest them).

Respond ONLY with this exact JSON structure (no markdown, no preamble):
{{
  "scores": {{"truth": N, "specificity": N, "testability": N, "novelty": N, "harmlessness": N}},
  "verdict": "ACCEPT" | "REJECT" | "REVISE",
  "reasoning": "<2-3 sentences>"
}}
"""


# --- Judge 1: Claude (via OAuth CLI) ---

def call_claude(claim, context, timeout=120):
    """Invoke Claude Sonnet via the OAuth CLI as a subprocess.
    Uses --print for non-interactive mode (output goes to stdout)."""
    prompt = JUDGE_PROMPT.format(claim=claim, context=context or "(none)")
    try:
        proc = subprocess.run(
            ["/usr/bin/claude", "--model", "sonnet", "--print", prompt],
            capture_output=True, text=True, timeout=timeout,
        )
        if proc.returncode != 0:
            return {"error": f"claude exit {proc.returncode}: {proc.stderr[:300]}"}
        return parse_verdict(proc.stdout, "claude")
    except subprocess.TimeoutExpired:
        return {"error": "claude timeout"}
    except Exception as e:
        return {"error": f"claude exception: {e}"}


# --- Judge 2: Codex (via Hermes OAuth — routes to ChatGPT subscription) ---

# Codex is invoked through `hermes chat -m openai-codex/<model>` which uses
# the OAuth credential already stored via `hermes auth add openai-codex`.
# This stays on your ChatGPT Pro subscription — no separate API key needed.
# Verified working 2026-07-12: `hermes chat -m openai-codex/gpt-5.4-mini
# -q "..."` returns a response.
DEFAULT_CODEX_MODEL = "openai-codex/gpt-5.4-mini"


def call_codex(claim, context, timeout=180, model=None):
    """Invoke Codex via Hermes OAuth (ChatGPT Pro subscription).

    Routes through `hermes chat -m openai-codex/<model> -q "..." --quiet`
    which uses the stored credential at `hermes auth list` (openai-codex).
    No separate token required from the judge script.

    Default model: gpt-5.4-mini (cheap + reasoning-enabled). Override with
    --codex-model on the CLI or `model=` kwarg here.
    """
    use_model = model or DEFAULT_CODEX_MODEL
    prompt = JUDGE_PROMPT.format(claim=claim, context=context or "(none)")
    # `-Q` is quiet: suppresses banner/spinner, prints only final response.
    # `--pass-session-id` is omitted because we don't need to resume.
    # `--ignore-rules` skips any rule files that might block tool use;
    # the judge prompt never asks for tools so this is safe.
    cmd = [
        "hermes", "chat",
        "-m", use_model,
        "-q", prompt,
        "-Q",  # quiet
        "--ignore-rules",  # judge prompt doesn't need any rules
    ]
    try:
        proc = subprocess.run(
            cmd,
            capture_output=True, text=True, timeout=timeout,
        )
        if proc.returncode != 0:
            return {
                "_judge": "codex",
                "error": f"hermes exit {proc.returncode}: {proc.stderr[:400] or proc.stdout[:400]}",
            }
        return parse_verdict(proc.stdout, "codex")
    except subprocess.TimeoutExpired:
        return {"_judge": "codex", "error": f"hermes timeout after {timeout}s"}
    except Exception as e:
        return {"_judge": "codex", "error": f"hermes exception: {e}"}


def parse_verdict(text, who):
    """Parse JSON verdict from a judge's stdout. Tolerant of markdown
    fences and trailing prose."""
    text = text.strip()
    # Strip code fences if present
    m = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
    if m:
        text = m.group(1)
    # Find first { ... } block
    if not text.startswith("{"):
        start = text.find("{")
        if start >= 0:
            text = text[start:]
    end = text.rfind("}")
    if end >= 0:
        text = text[:end + 1]
    try:
        v = json.loads(text)
        v["_judge"] = who
        return v
    except json.JSONDecodeError as e:
        return {"_judge": who, "error": f"parse failed: {e}", "raw": text[:500]}


# --- Convergence logic ---

def converge(claude_v, codex_v):
    """Two-judge convergence. Returns (verdict, consensus_strength, conflicts).
    verdict ∈ {accepted, rejected, revise, conflict}."""
    if claude_v.get("error") or codex_v.get("error"):
        return "conflict", 0.0, ["one_or_more_judges_failed"]

    cv = claude_v.get("verdict", "").upper()
    xv = codex_v.get("verdict", "").upper()

    # Verdict agreement
    if cv == xv and cv in ("ACCEPT", "REJECT"):
        # Compute consensus strength from score similarity
        c_scores = claude_v.get("scores", {})
        x_scores = codex_v.get("scores", {})
        diff = sum(abs(c_scores.get(k, 3) - x_scores.get(k, 3)) for k in ("truth", "specificity", "testability", "novelty", "harmlessness"))
        # max possible diff = 5 dims × 4 = 20
        strength = max(0.0, 1.0 - diff / 20.0)
        return cv.lower(), strength, []

    if cv == xv and cv == "REVISE":
        # Both want revisions — check if reasoning aligns
        if claude_v.get("reasoning", "")[:30].lower() == codex_v.get("reasoning", "")[:30].lower():
            return "revise", 0.7, []
        return "revise", 0.4, ["revise_suggestions_diverge"]

    # Disagreement
    return "conflict", 0.0, [f"claude={cv} codex={xv}"]


# --- Entry points ---

def main():
    p = argparse.ArgumentParser(description="Triadic judge — Codex + Claude review a claim")
    g = p.add_mutually_exclusive_group(required=True)
    g.add_argument("--claim", help="The claim to judge (string)")
    g.add_argument("--memory-urn", help="Fetch a UMP memory by urn and judge it")
    g.add_argument("--dispute", help="Path to a file containing A/B exchange text")
    g.add_argument("--dispute-stdin", action="store_true", help="Read dispute from stdin")
    p.add_argument("--context", default="", help="Optional context for the claim")
    p.add_argument("--dry-run", action="store_true", help="Print prompt, don't call judges")
    p.add_argument("--timeout", type=int, default=120, help="Per-judge timeout seconds")
    p.add_argument("--skip-claude", action="store_true")
    p.add_argument("--skip-codex", action="store_true")
    p.add_argument("--codex-model", default=DEFAULT_CODEX_MODEL,
                   help=f"Codex model name (default: {DEFAULT_CODEX_MODEL})")
    args = p.parse_args()

    claim = args.claim or ""
    context = args.context or ""

    if args.memory_urn:
        claim, context = fetch_memory(args.memory_urn)
    elif args.dispute:
        text = open(args.dispute).read()
        claim, context = parse_dispute(text)
    elif args.dispute_stdin:
        text = sys.stdin.read()
        claim, context = parse_dispute(text)

    if args.dry_run:
        print("=== DRY RUN — would call judges with this prompt ===")
        print(JUDGE_PROMPT.format(claim=claim[:500], context=context[:500]))
        return

    print(f"Claim: {claim[:120]}{'...' if len(claim) > 120 else ''}")
    print(f"Context: {context[:80] if context else '(none)'}")
    print()

    claude_v = {}
    codex_v = {}
    if not args.skip_claude:
        print("Calling Claude Sonnet...", end=" ", flush=True)
        t0 = time.time()
        claude_v = call_claude(claim, context, timeout=args.timeout)
        print(f"({time.time()-t0:.1f}s)")
    if not args.skip_codex:
        print(f"Calling Codex ({args.codex_model} via Hermes OAuth)...", end=" ", flush=True)
        t0 = time.time()
        codex_v = call_codex(claim, context, timeout=args.timeout, model=args.codex_model)
        print(f"({time.time()-t0:.1f}s)")
    print()

    verdict, strength, conflicts = converge(claude_v, codex_v)
    result = {
        "verdict": verdict,
        "consensus_strength": round(strength, 3),
        "conflicts": conflicts,
        "claude": claude_v,
        "codex": codex_v,
        "claim_preview": claim[:200],
    }
    print(json.dumps(result, indent=2))

    sys.exit(0 if verdict in ("accepted", "rejected", "revise") else 1)


def fetch_memory(urn):
    """Fetch a UMP memory and format as claim + context."""
    try:
        req = urllib.request.Request(
            f"http://127.0.0.1:4317/ump/memory/{urllib.parse.quote(urn, safe='')}",
        )
        with urllib.request.urlopen(req, timeout=10) as r:
            data = json.loads(r.read())
        rec = data.get("record") or data
        body = rec.get("body", {})
        subject = body.get("subject", "")
        text = body.get("text", "")
        claim = f"{subject}\n\n{text}" if subject else text
        context = f"kind={rec.get('kind')} topic={rec.get('scope',{}).get('project','')}"
        return claim, context
    except Exception as e:
        return f"(failed to fetch {urn}: {e})", ""


def parse_dispute(text):
    """Parse a structured A/B exchange. Expects lines starting with 'A:' or 'B:'."""
    a_lines, b_lines = [], []
    for line in text.splitlines():
        if line.startswith("A:"):
            a_lines.append(line[2:].strip())
        elif line.startswith("B:"):
            b_lines.append(line[2:].strip())
    a_text = "\n".join(a_lines).strip()
    b_text = "\n".join(b_lines).strip()
    # The claim under dispute is the disagreement. We synthesize:
    claim = a_text if a_text else b_text
    context = f"A says:\n{a_text}\n\nB says:\n{b_text}" if a_text and b_text else ""
    return claim, context


if __name__ == "__main__":
    main()