#!/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/` 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/ -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()