"""CoSER: does the score measure role-playing? Two checks of the method itself, not of the platform's implementation of it. Part 1, no model called: from the anchor's numbers per session (coser_anchor.rows.json), how repeatable the score is for one scene, what the length term does per dimension, and how large the anchor's shift is against the differences between models in the paper. Part 2, --original: the book's own conversation is sent to the critic as if it were the simulation, with the official prompts (the same conversation is also the reference), for a sample of scenes, and scored by the official formula with R = its lines not spoken by the environment. The official code never does this: it writes a score of 100 for the original conversation without judging it (eval_reproduce.py at 93c1474: results_by_case['groundtruth'] ... 'score': 100). Criteria for part 2, written on 2026-11-01 before the first request. A method that measures role-playing against the book must put the book first: passes the original scores at least 90 on Storyline Consistency, and above GPT-4o's mean on the same scenes on all four dimensions fails otherwise python3 -I tools/experiments/coser_method_validity.py -I tools/experiments/coser_method_validity.py --original """ import collections import hashlib import importlib.util import json import os import re import statistics as st import sys import urllib.error import urllib.request HERE = os.path.dirname(os.path.abspath(__file__)) COLS = ["sc", "an", "cf", "sq"] NAMES = {"sc": "Storyline Consistency", "an": "Anthropomorphism", "cf": "Character Fidelity", "sq": "Storyline Quality"} MODEL = "openai/gpt-4o-2024-08-06" SCENES = 24 COST_LIMIT = 2.0 def pearson(x, y): mx, my = st.mean(x), st.mean(y) sx = sum((a - mx) ** 2 for a in x) ** 0.5 sy = sum((b - my) ** 2 for b in y) ** 0.5 return sum((a - mx) * (b - my) for a, b in zip(x, y)) / (sx * sy) if sx and sy else None def icc(groups): """One-way intraclass correlation for equal-sized groups: the share of variance that belongs to the scene.""" k = len(groups[0]) grand = st.mean(v for g in groups for v in g) msb = k * sum((st.mean(g) - grand) ** 2 for g in groups) / (len(groups) - 1) msw = sum((v - st.mean(g)) ** 2 for g in groups for v in g) / (len(groups) * (k - 1)) return (msb - msw) / (msb + (k - 1) * msw) if msb + (k - 1) * msw else None def ols(x, y): mx, my = st.mean(x), st.mean(y) b = sum((a - mx) * (c - my) for a, c in zip(x, y)) / sum((a - mx) ** 2 for a in x) return b, my - b * mx def part1(rows_path, checks_path, out): rows = json.load(open(rows_path)) main = json.load(open(os.path.join(HERE, "coser_main_table.result.json")))["rows"] by_scene = collections.defaultdict(dict) for r in rows: by_scene[r["scene"]][r["sample"]] = r res = {"sessions": len(rows), "scenes": len(by_scene), "dimensions": {}} for k in COLS: groups = [[s[i]["dims"][k] for i in sorted(s)] for s in by_scene.values() if len(s) == 3] flaw_groups = [[s[i]["n"][k] for i in sorted(s)] for s in by_scene.values() if len(s) == 3] spreads = [max(g) - min(g) for g in groups] R = [r["R"] for r in rows] raw = [100 - 5 * r["S"][k] for r in rows] slope, intercept = ols(R, raw) scores = [g for g in (x for x in groups)] papers = [m["scores"][k] for m in main] res["dimensions"][k] = { "icc_same_scene": icc(groups), "icc_flaw_count": icc(flaw_groups), "median_range_within_scene": st.median(spreads), "share_scenes_range_over_20": sum(s > 20 for s in spreads) / len(spreads), "sd_of_session_scores": st.pstdev([v for g in groups for v in g]), "raw_score_slope_per_line": slope, "net_slope_per_line_after_correction": slope + 1.5, "correlation_with_lines": pearson(R, [r["dims"][k] for r in rows]), "sessions_at_floor": sum(r["dims"][k] == 0 for r in rows), "paper_range_21_models": [min(papers), max(papers)], } # What the anchor's own numbers look like beside the differences between models in the paper. avgs = sorted((m["scores"]["avg"] for m in main), reverse=True) res["paper_average"] = {"range": [avgs[-1], avgs[0]], "top_minus_median": avgs[0] - st.median(avgs), "models_within_5_27_of_the_top": sum(a >= avgs[0] - 5.27 for a in avgs), "models": len(avgs)} res["unbounded_floor_gap"] = {k: st.mean(min(0, 100 - 5 * r["S"][k] + 1.5 * r["R"]) for r in rows if r["dims"][k] == 0) for k in ("cf", "an")} res["length_bonus_mean"] = st.mean(1.5 * r["R"] for r in rows) overall = lambda r: st.mean(r["dims"][k] for k in COLS) ranges = [max(overall(x) for x in v.values()) - min(overall(x) for x in v.values()) for v in by_scene.values() if len(v) == 3] res["overall_range_within_scene_median"] = st.median(ranges) res["scenes_cf_short_and_enumerated"] = sum(1 for v in by_scene.values() if min(x["n"]["cf"] for x in v.values()) <= 5 and max(x["n"]["cf"] for x in v.values()) >= 10) with open(out, "w", encoding="utf-8") as f: json.dump(res, f, ensure_ascii=False, indent=1) print(json.dumps(res, ensure_ascii=False, indent=1)) return 0 def part2(repo, keyfile, rows_path, out): spec = importlib.util.spec_from_file_location("coser_reference", os.path.join(HERE, "coser_reference.py")) ref = importlib.util.module_from_spec(spec) spec.loader.exec_module(ref) ref.load_critic_prompts(repo) keys = list(dict.fromkeys(re.findall(r"sk-or-v1-[A-Za-z0-9]+", open(keyfile, encoding="utf-8").read()))) test = json.loads(ref.git_show(repo, ref.PINNED, "data/test/test_set.json")) rows = json.load(open(rows_path)) # Half from each split, by a fixed hash, so the sample is decided before anything is seen. def pick(tag, n): ids = [i for i, c in enumerate(test) if c["tag"] == tag] return sorted(ids, key=lambda i: hashlib.sha1(f"coser-original:{i}".encode()).hexdigest())[:n] chosen = sorted(pick("id", SCENES // 2) + pick("ood", SCENES // 2)) spent, out_rows = 0.0, [] for i in chosen: c = test[i] scene = f"coser-{i + 1:03d}" speaking = c["speaking_characters_w_env"] + ([] if "Environment" in c["speaking_characters_w_env"] else ["Environment"]) built = ref.build_agents(c, "gpt-4o", "gpt-4o", "gpt-4o-mini", True, {x: "" for x in speaking}, "a world model") # The original conversation in the form a simulation is recorded: "name: message" per line. record = [{"role": d["character"], "content": ref.add_speaker_name(d["message"], d["character"])} for d in c["dialogues"]] R = sum(d["character"] != "Environment" for d in c["dialogues"]) row = {"scene": scene, "split": c["tag"], "R": R, "dims": {}, "S": {}, "n": {}} for k in COLS: msgs = ref.judge_messages(c, record, built["involved"], NAMES[k]) body = json.dumps({"model": MODEL, "messages": msgs, "temperature": 0, "max_tokens": 8196, "provider": {"only": ["openai"], "allow_fallbacks": False}, "usage": {"include": True}}).encode() flaws = None for attempt in range(4): if spent > COST_LIMIT: raise SystemExit(f"cost limit reached: ${spent:.3f}") try: req = urllib.request.Request("https://openrouter.ai/api/v1/chat/completions", data=body, headers={"Authorization": f"Bearer {keys[0]}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=180) as r: j = json.loads(r.read()) spent += j.get("usage", {}).get("cost") or 0.0 text = j["choices"][0]["message"].get("content") or "" flaws = json.loads(re.search(r"\{.*\}", text, re.S).group(0))[NAMES[k]].get("flaws", []) break except (urllib.error.URLError, KeyError, ValueError, AttributeError, TypeError): continue if flaws is None: continue s = sum(f.get("severity", 1) for f in flaws if isinstance(f.get("severity", 1), int)) row["dims"][k] = max(0, 100 - (s - 0.3 * R) * 5) row["S"][k] = s row["n"][k] = len(flaws) gpt = [r for r in rows if r["scene"] == scene] row["gpt4o"] = {k: st.mean(r["dims"][k] for r in gpt) for k in COLS} row["gpt4o_R"] = st.mean(r["R"] for r in gpt) out_rows.append(row) print(scene, c["tag"], "R", R, "original", {k: row["dims"].get(k) for k in COLS}, "gpt-4o", {k: round(row["gpt4o"][k], 1) for k in COLS}, flush=True) done = [r for r in out_rows if all(k in r["dims"] for k in COLS)] summary = {k: {"original": st.mean(r["dims"][k] for r in done), "gpt4o": st.mean(r["gpt4o"][k] for r in done), "original_flaws": st.mean(r["n"][k] for r in done), "original_S": st.mean(r["S"][k] for r in done), "gpt4o_S": st.mean(st.mean(x["S"][k] for x in rows if x["scene"] == r["scene"]) for r in done), "scenes_original_below_gpt4o": sum(r["dims"][k] < r["gpt4o"][k] for r in done), "scenes_at_floor": sum(r["dims"][k] == 0 for r in done)} for k in COLS} orig_avg = st.mean(summary[k]["original"] for k in COLS) gpt_avg = st.mean(summary[k]["gpt4o"] for k in COLS) passed = summary["sc"]["original"] >= 90 and all(summary[k]["original"] > summary[k]["gpt4o"] for k in COLS) report = {"model": MODEL, "scenes": len(done), "cost": spent, "verdict": "passes" if passed else "fails", "average": {"original": orig_avg, "gpt4o": gpt_avg}, "summary": summary, "rows": out_rows} with open(out, "w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=1) print("verdict", report["verdict"], "cost", round(spent, 3), "average original", round(orig_avg, 2), "gpt-4o", round(gpt_avg, 2)) for k in COLS: print(k, {a: round(b, 2) for a, b in summary[k].items()}) return 0 if __name__ == "__main__": if "--original" in sys.argv: sys.exit(part2(*[a for a in sys.argv[1:] if a != "--original"])) sys.exit(part1(*sys.argv[1:4]))