Buckets:

affine-god
/

1111

Files

xet

affine-god/1111 / reason_ceiling.py

affine-god

3 months ago

download

raw

8.49 kB

"""Oracle ceilings: what would each proposed mechanism score with PERFECT knowledge?
The efficient way to audit a policy family. If a mechanism cannot clear the crown
bar even when its gating variable is known exactly — teacher action, cross-model
consensus, the champion's own choices — then no predictor of that variable can,
and there is no point building one. Deployable performance is bounded above by
the oracle.
Scoring follows the live rule: a silent pair contributes EXACTLY 0, the mean is
over ALL pairs, and
    margin = mean(Reason_ours) - mean(Reason_champion)
    crown  = margin > max(k_sigma * SE, min_margin)
Constants are read from the live contract, never hardcoded.
"""

from __future__ import annotations

import argparse
import collections
import json
import math
import pathlib
import subprocess
import sys

import numpy as np

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
import action_ontology as AO
import reason_signal as R

CONTRACT = "https://s3.hippius.com/affine-sn120/data/contract.json"


def contract() -> dict:
    out = subprocess.run(["curl", "-s", "-m", "60", CONTRACT],
                         capture_output=True, text=True).stdout
    try:
        c = json.loads(out)
    except json.JSONDecodeError:
        raise SystemExit(f"contract unreadable: {out[:200]}")
    d = c["duel"]
    return {"n": d["n_turns"], "k": d["k_sigma"], "dmin": d["min_margin"]}


def evaluate(name: str, reason: np.ndarray, speak: np.ndarray, cfg: dict,
             champ_mean: float) -> dict:
    contrib = np.where(speak, reason, 0.0)
    mean = float(contrib.mean())
    se = float(contrib.std(ddof=1)) / math.sqrt(cfg["n"])
    margin = mean - champ_mean
    bar = max(cfg["k"] * se, cfg["dmin"])
    return {"name": name, "speak": float(speak.mean()),
            "hit": float((reason[speak] > 0).mean()) if speak.any() else float("nan"),
            "sel_mean": float(reason[speak].mean()) if speak.any() else 0.0,
            "mean": mean, "margin": margin, "bar": bar, "crowns": margin > bar}


def show(r: dict) -> None:
    print(f"  {r['name']:<44} speak {100*r['speak']:>5.1f}%  hit {100*r['hit']:>5.1f}%  "
          f"sel {r['sel_mean']:>+9.6f}  mean {r['mean']:>+9.6f}  "
          f"margin {r['margin']:>+9.6f}  {'CROWNS' if r['crowns'] else 'no'}")


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--data", default=str(R.ROOT / "data" / "reason.jsonl.gz"))
    args = ap.parse_args()
    cfg = contract()
    print(f"live contract: n_turns {cfg['n']}  k_sigma {cfg['k']}  "
          f"min_margin {cfg['dmin']}")

    allrows = R.load(args.data, drop_silent=False)
    champ = next(m for m in {r["model"] for r in allrows} if "diceofgod" in m)
    ch = [r for r in allrows if r["model"] == champ]
    champ_mean = float(np.mean([r["reason"] for r in ch]))
    print(f"champion {champ}\n  its mean Reason {champ_mean:+.6f} over {len(ch)} pairs")
    print(f"  a challenger must beat it by > max({cfg['k']}*SE, {cfg['dmin']})\n")

    # Field rows = everyone except the champion: the pool of thoughts a
    # challenger might plausibly produce.
    rows = [r for r in allrows if r["model"] != champ and not r.get("silent")]
    for r in rows:
        r["_ca"] = AO.action_class(r["y_a"])
        r["_cc"] = AO.action_class(r["y_C"])
        r["_ta"], r["_tc"] = AO.action_target(r["y_a"]), AO.action_target(r["y_C"])
    reason = np.array([r["reason"] for r in rows])
    n = len(rows)
    print(f"field pool: {n} spoken pairs from {len({r['model'] for r in rows})} models\n")

    print("=== ORACLE CEILINGS — perfect knowledge of the gating variable ===")
    show(evaluate("always speak (the field's status quo)", reason,
                  np.ones(n, bool), cfg, champ_mean))
    show(evaluate("always silent (pure silence)", reason,
                  np.zeros(n, bool), cfg, champ_mean))

    exact = np.array([(r["y_a"] or "").strip() == (r["y_C"] or "").strip() for r in rows])
    sc = np.array([r["_ca"] == r["_cc"] for r in rows])
    stg = np.array([bool(r["_ta"]) and r["_ta"] == r["_tc"] for r in rows])
    show(evaluate("ORACLE speak iff exact action match", reason, exact, cfg, champ_mean))
    show(evaluate("ORACLE speak iff same class + target", reason, sc & stg, cfg, champ_mean))
    show(evaluate("ORACLE speak iff same action class", reason, sc, cfg, champ_mean))
    show(evaluate("ORACLE speak iff Reason>0 (absolute max)", reason, reason > 0,
                  cfg, champ_mean))

    # cross-model consensus, swept
    by_turn = collections.defaultdict(list)
    for i, r in enumerate(rows):
        by_turn[r["turn_id"]].append(i)
    cons = np.zeros(n)
    for tid, idx in by_turn.items():
        models = {rows[i]["model"] for i in idx}
        if len(models) < 3:
            cons[idx] = -1.0                      # not measurable
            continue
        cnt = collections.Counter(rows[i]["_ca"] for i in idx)
        cons[idx] = max(cnt.values()) / sum(cnt.values())
    meas = cons >= 0
    print(f"\n  consensus measurable on {meas.sum()}/{n} pairs "
          f"({100*meas.mean():.1f}%; needs >=3 models on the turn)")
    for t in (0.5, 0.7, 0.8, 0.9, 0.95, 1.0):
        show(evaluate(f"ORACLE speak iff consensus >= {t:.2f}", reason,
                      meas & (cons >= t), cfg, champ_mean))

    print("\n=== COMBINED — the audit's proposed stack, all oracle ===")
    show(evaluate("consensus>=0.9 AND same class", reason,
                  meas & (cons >= 0.9) & sc, cfg, champ_mean))
    show(evaluate("consensus>=0.9 AND same class + target", reason,
                  meas & (cons >= 0.9) & sc & stg, cfg, champ_mean))

    print("\n=== HOW MUCH ROOM IS THERE? ===")
    need = champ_mean + max(cfg["dmin"], 0.0)
    print(f"  a challenger needs mean Reason > {need:+.6f} to clear min_margin alone")
    best = max((evaluate(f"c>={t}", reason, meas & (cons >= t), cfg, champ_mean)
                for t in np.arange(0.4, 1.001, 0.02)), key=lambda x: x["margin"])
    print(f"  best consensus-only oracle: margin {best['margin']:+.6f} "
          f"at speak {100*best['speak']:.1f}%  -> "
          f"{'CLEARS' if best['crowns'] else 'FAILS'}")


if __name__ == "__main__":
    main()


def extra() -> None:
    """The last candidate: gate on CONSENSUS + our own agreement with the
    majority action class. Both halves are estimable without the teacher — the
    majority class is a property of the state (predictable in principle) and our
    own action class is something we choose. If this fails as an oracle, the
    whole family is dead.
    """
    import numpy as np, collections, sys, pathlib
    sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
    cfg = contract()
    allrows = R.load(str(R.ROOT / "data" / "reason.jsonl.gz"), drop_silent=False)
    champ = next(m for m in {r["model"] for r in allrows} if "diceofgod" in m)
    champ_mean = float(np.mean([r["reason"] for r in allrows if r["model"] == champ]))
    rows = [r for r in allrows if r["model"] != champ and not r.get("silent")]
    for r in rows:
        r["_ca"] = AO.action_class(r["y_a"])
    reason = np.array([r["reason"] for r in rows]); n = len(rows)
    by_turn = collections.defaultdict(list)
    for i, r in enumerate(rows):
        by_turn[r["turn_id"]].append(i)
    cons = np.full(n, -1.0); ismaj = np.zeros(n, bool)
    for tid, idx in by_turn.items():
        if len({rows[i]["model"] for i in idx}) < 3:
            continue
        cnt = collections.Counter(rows[i]["_ca"] for i in idx)
        top, tot = cnt.most_common(1)[0], sum(cnt.values())
        cons[idx] = top[1] / tot
        for i in idx:
            ismaj[i] = rows[i]["_ca"] == top[0]
    meas = cons >= 0
    print("\n=== LAST CANDIDATE — consensus AND agree-with-majority ===")
    print("  (both halves estimable without the teacher; this is the ceiling of")
    print("   any deployable version of the audit's Hypothesis A + C stack)")
    best = None
    for t in (0.5, 0.7, 0.8, 0.9, 0.95, 1.0):
        r_ = evaluate(f"consensus>={t:.2f} AND our class == majority", reason,
                      meas & (cons >= t) & ismaj, cfg, champ_mean)
        show(r_)
        if best is None or r_["margin"] > best["margin"]:
            best = r_
    print(f"\n  best of family: margin {best['margin']:+.6f}  bar {best['bar']:.6f}  "
          f"-> {'CLEARS' if best['crowns'] else 'FAILS'}")
    print(f"  and this is the ORACLE — a real predictor is strictly worse.")

Xet Storage Details

Size:: 8.49 kB
Xet hash:: 34898772f813e47eb293b1101b7d08ad9f027f54c4dda424d4ad241e23b9cbc8

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""Oracle ceilings: what would each proposed mechanism score with PERFECT knowledge?

	The efficient way to audit a policy family. If a mechanism cannot clear the crown
	bar even when its gating variable is known exactly — teacher action, cross-model
	consensus, the champion's own choices — then no predictor of that variable can,
	and there is no point building one. Deployable performance is bounded above by
	the oracle.

	Scoring follows the live rule: a silent pair contributes EXACTLY 0, the mean is
	over ALL pairs, and

	margin = mean(Reason_ours) - mean(Reason_champion)
	crown = margin > max(k_sigma * SE, min_margin)

	Constants are read from the live contract, never hardcoded.
	"""

	from __future__ import annotations

	import argparse
	import collections
	import json
	import math
	import pathlib
	import subprocess
	import sys

	import numpy as np

	sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
	import action_ontology as AO
	import reason_signal as R

	CONTRACT = "https://s3.hippius.com/affine-sn120/data/contract.json"


	def contract() -> dict:
	out = subprocess.run(["curl", "-s", "-m", "60", CONTRACT],
	capture_output=True, text=True).stdout
	try:
	c = json.loads(out)
	except json.JSONDecodeError:
	raise SystemExit(f"contract unreadable: {out[:200]}")
	d = c["duel"]
	return {"n": d["n_turns"], "k": d["k_sigma"], "dmin": d["min_margin"]}


	def evaluate(name: str, reason: np.ndarray, speak: np.ndarray, cfg: dict,
	champ_mean: float) -> dict:
	contrib = np.where(speak, reason, 0.0)
	mean = float(contrib.mean())
	se = float(contrib.std(ddof=1)) / math.sqrt(cfg["n"])
	margin = mean - champ_mean
	bar = max(cfg["k"] * se, cfg["dmin"])
	return {"name": name, "speak": float(speak.mean()),
	"hit": float((reason[speak] > 0).mean()) if speak.any() else float("nan"),
	"sel_mean": float(reason[speak].mean()) if speak.any() else 0.0,
	"mean": mean, "margin": margin, "bar": bar, "crowns": margin > bar}


	def show(r: dict) -> None:
	print(f" {r['name']:<44} speak {100r['speak']:>5.1f}% hit {100r['hit']:>5.1f}% "
	f"sel {r['sel_mean']:>+9.6f} mean {r['mean']:>+9.6f} "
	f"margin {r['margin']:>+9.6f} {'CROWNS' if r['crowns'] else 'no'}")


	def main() -> None:
	ap = argparse.ArgumentParser()
	ap.add_argument("--data", default=str(R.ROOT / "data" / "reason.jsonl.gz"))
	args = ap.parse_args()
	cfg = contract()
	print(f"live contract: n_turns {cfg['n']} k_sigma {cfg['k']} "
	f"min_margin {cfg['dmin']}")

	allrows = R.load(args.data, drop_silent=False)
	champ = next(m for m in {r["model"] for r in allrows} if "diceofgod" in m)
	ch = [r for r in allrows if r["model"] == champ]
	champ_mean = float(np.mean([r["reason"] for r in ch]))
	print(f"champion {champ}\n its mean Reason {champ_mean:+.6f} over {len(ch)} pairs")
	print(f" a challenger must beat it by > max({cfg['k']}*SE, {cfg['dmin']})\n")

	# Field rows = everyone except the champion: the pool of thoughts a
	# challenger might plausibly produce.
	rows = [r for r in allrows if r["model"] != champ and not r.get("silent")]
	for r in rows:
	r["_ca"] = AO.action_class(r["y_a"])
	r["_cc"] = AO.action_class(r["y_C"])
	r["_ta"], r["_tc"] = AO.action_target(r["y_a"]), AO.action_target(r["y_C"])
	reason = np.array([r["reason"] for r in rows])
	n = len(rows)
	print(f"field pool: {n} spoken pairs from {len({r['model'] for r in rows})} models\n")

	print("=== ORACLE CEILINGS — perfect knowledge of the gating variable ===")
	show(evaluate("always speak (the field's status quo)", reason,
	np.ones(n, bool), cfg, champ_mean))
	show(evaluate("always silent (pure silence)", reason,
	np.zeros(n, bool), cfg, champ_mean))

	exact = np.array([(r["y_a"] or "").strip() == (r["y_C"] or "").strip() for r in rows])
	sc = np.array([r["_ca"] == r["_cc"] for r in rows])
	stg = np.array([bool(r["_ta"]) and r["_ta"] == r["_tc"] for r in rows])
	show(evaluate("ORACLE speak iff exact action match", reason, exact, cfg, champ_mean))
	show(evaluate("ORACLE speak iff same class + target", reason, sc & stg, cfg, champ_mean))
	show(evaluate("ORACLE speak iff same action class", reason, sc, cfg, champ_mean))
	show(evaluate("ORACLE speak iff Reason>0 (absolute max)", reason, reason > 0,
	cfg, champ_mean))

	# cross-model consensus, swept
	by_turn = collections.defaultdict(list)
	for i, r in enumerate(rows):
	by_turn[r["turn_id"]].append(i)
	cons = np.zeros(n)
	for tid, idx in by_turn.items():
	models = {rows[i]["model"] for i in idx}
	if len(models) < 3:
	cons[idx] = -1.0 # not measurable
	continue
	cnt = collections.Counter(rows[i]["_ca"] for i in idx)
	cons[idx] = max(cnt.values()) / sum(cnt.values())
	meas = cons >= 0
	print(f"\n consensus measurable on {meas.sum()}/{n} pairs "
	f"({100*meas.mean():.1f}%; needs >=3 models on the turn)")
	for t in (0.5, 0.7, 0.8, 0.9, 0.95, 1.0):
	show(evaluate(f"ORACLE speak iff consensus >= {t:.2f}", reason,
	meas & (cons >= t), cfg, champ_mean))

	print("\n=== COMBINED — the audit's proposed stack, all oracle ===")
	show(evaluate("consensus>=0.9 AND same class", reason,
	meas & (cons >= 0.9) & sc, cfg, champ_mean))
	show(evaluate("consensus>=0.9 AND same class + target", reason,
	meas & (cons >= 0.9) & sc & stg, cfg, champ_mean))

	print("\n=== HOW MUCH ROOM IS THERE? ===")
	need = champ_mean + max(cfg["dmin"], 0.0)
	print(f" a challenger needs mean Reason > {need:+.6f} to clear min_margin alone")
	best = max((evaluate(f"c>={t}", reason, meas & (cons >= t), cfg, champ_mean)
	for t in np.arange(0.4, 1.001, 0.02)), key=lambda x: x["margin"])
	print(f" best consensus-only oracle: margin {best['margin']:+.6f} "
	f"at speak {100*best['speak']:.1f}% -> "
	f"{'CLEARS' if best['crowns'] else 'FAILS'}")


	if __name__ == "__main__":
	main()


	def extra() -> None:
	"""The last candidate: gate on CONSENSUS + our own agreement with the
	majority action class. Both halves are estimable without the teacher — the
	majority class is a property of the state (predictable in principle) and our
	own action class is something we choose. If this fails as an oracle, the
	whole family is dead.
	"""
	import numpy as np, collections, sys, pathlib
	sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
	cfg = contract()
	allrows = R.load(str(R.ROOT / "data" / "reason.jsonl.gz"), drop_silent=False)
	champ = next(m for m in {r["model"] for r in allrows} if "diceofgod" in m)
	champ_mean = float(np.mean([r["reason"] for r in allrows if r["model"] == champ]))
	rows = [r for r in allrows if r["model"] != champ and not r.get("silent")]
	for r in rows:
	r["_ca"] = AO.action_class(r["y_a"])
	reason = np.array([r["reason"] for r in rows]); n = len(rows)
	by_turn = collections.defaultdict(list)
	for i, r in enumerate(rows):
	by_turn[r["turn_id"]].append(i)
	cons = np.full(n, -1.0); ismaj = np.zeros(n, bool)
	for tid, idx in by_turn.items():
	if len({rows[i]["model"] for i in idx}) < 3:
	continue
	cnt = collections.Counter(rows[i]["_ca"] for i in idx)
	top, tot = cnt.most_common(1)[0], sum(cnt.values())
	cons[idx] = top[1] / tot
	for i in idx:
	ismaj[i] = rows[i]["_ca"] == top[0]
	meas = cons >= 0
	print("\n=== LAST CANDIDATE — consensus AND agree-with-majority ===")
	print(" (both halves estimable without the teacher; this is the ceiling of")
	print(" any deployable version of the audit's Hypothesis A + C stack)")
	best = None
	for t in (0.5, 0.7, 0.8, 0.9, 0.95, 1.0):
	r_ = evaluate(f"consensus>={t:.2f} AND our class == majority", reason,
	meas & (cons >= t) & ismaj, cfg, champ_mean)
	show(r_)
	if best is None or r_["margin"] > best["margin"]:
	best = r_
	print(f"\n best of family: margin {best['margin']:+.6f} bar {best['bar']:.6f} "
	f"-> {'CLEARS' if best['crowns'] else 'FAILS'}")
	print(f" and this is the ORACLE — a real predictor is strictly worse.")