Spaces:
Sleeping
Sleeping
| """What does the anchor fold move? Two read-only analyses over archived sleep_controls runs (no model, no device). | |
| 1. Per-turn write magnitude in the teaching session: transactions joined to chat turns by their transaction-index | |
| range; per turn the sum of committed ``delta_norm``, the max ``surprise_max`` and ``chunk_loss``, and whether the | |
| observational policy flagged any chunk. Question: did the planted contradictions write more than the other facts? | |
| 2. Storage shift under the anchor child: for every verbatim probe the anchor report's before/after | |
| ``answer_logprob`` (expected-answer mean log-prob per token, teacher-forced from a fresh state), grouped into | |
| planted answers, general true answers and the taught personal facts. Question: does the fold move the planted | |
| answers specifically, or every short factual completion? | |
| python -m scripts.experiments.anchor_transfer_analysis --runs docs/research/results/sleep-2026-09-23/final_step250_seed0_exclude ... --out <json> | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import statistics | |
| from typing import Any | |
| POISON_ANSWERS = {"Berlin", "50", "nine", "Moon"} | |
| GENERAL_ANSWERS = {"Paris", "100", "seven", "Earth", "blue"} | |
| def load_jsonl(path: str) -> list[dict[str, Any]]: | |
| with open(path, encoding="utf-8") as f: | |
| return [json.loads(line) for line in f if line.strip()] | |
| def turn_write_table(transactions: list[dict[str, Any]], trace: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| """One row per chat turn: committed delta_norm summed over its chunks (by tx range), max surprise and chunk loss, | |
| chunk count, whether any chunk carried a would_* / requested intervention flag, and the rank by write (1 = largest).""" | |
| by_index = {int(t["index"]): t for t in transactions} | |
| rows: list[dict[str, Any]] = [] | |
| for rec in trace: | |
| if rec.get("kind") != "chat" or rec.get("tx_start") is None or rec.get("tx_end") is None: | |
| continue | |
| chunks = [by_index[i] for i in range(int(rec["tx_start"]), int(rec["tx_end"])) if i in by_index] | |
| if not chunks: | |
| continue | |
| flagged = any( | |
| any(str(r).startswith("would_") for r in c.get("requested", {}).get("reasons", [])) or c.get("requested", {}).get("kind") != c.get("decision", {}).get("kind") | |
| or c.get("decision", {}).get("kind") in ("scale", "project") | |
| for c in chunks | |
| ) | |
| rows.append({ | |
| "prompt": rec.get("prompt", ""), | |
| "chunks": len(chunks), | |
| "delta_norm_sum": float(sum(float(c.get("accepted", {}).get("delta_norm") or 0.0) for c in chunks)), | |
| "surprise_max": float(max(float(c.get("signals", {}).get("surprise_max") or 0.0) for c in chunks)), | |
| "chunk_loss_max": float(max(float(c.get("signals", {}).get("chunk_loss") or 0.0) for c in chunks)), | |
| "flagged": bool(flagged), | |
| "planted": rec.get("prompt", "").startswith("Remember this"), | |
| }) | |
| order = sorted(range(len(rows)), key=lambda i: -rows[i]["delta_norm_sum"]) | |
| for rank, i in enumerate(order, start=1): | |
| rows[i]["rank_by_write"] = rank | |
| return rows | |
| def answer_group(expected: str) -> str: | |
| if expected in POISON_ANSWERS: | |
| return "planted" | |
| if expected in GENERAL_ANSWERS: | |
| return "general" | |
| return "taught" | |
| def storage_shift(report: dict[str, Any]) -> dict[str, Any]: | |
| """Per verbatim probe (keyed by question AND expected answer: the plants reuse general questions) the after-before | |
| shift of the expected answer's mean log-prob per token, and per-group means.""" | |
| def key(x: dict[str, Any]) -> tuple[str, str]: | |
| return (x["question"], x["expected"]) | |
| before = {key(x): x for x in report["before"]["recall"]["results"] if x.get("variant") == "verbatim"} | |
| after = {key(x): x for x in report["after"]["recall"]["results"] if x.get("variant") == "verbatim"} | |
| probes: list[dict[str, Any]] = [] | |
| for k, b in before.items(): | |
| a = after.get(k) | |
| if a is None or b.get("answer_logprob") is None or a.get("answer_logprob") is None: | |
| continue | |
| probes.append({"question": k[0], "expected": k[1], "group": answer_group(k[1]), | |
| "before": float(b["answer_logprob"]), "after": float(a["answer_logprob"]), | |
| "shift": float(a["answer_logprob"]) - float(b["answer_logprob"]), | |
| "greedy_before": bool(b.get("contains")), "greedy_after": bool(a.get("contains"))}) | |
| groups: dict[str, Any] = {} | |
| for g in ("planted", "general", "taught"): | |
| v = [p["shift"] for p in probes if p["group"] == g] | |
| groups[g] = {"n": len(v), "mean_shift": statistics.fmean(v) if v else None, | |
| "greedy_hits_before": sum(p["greedy_before"] for p in probes if p["group"] == g), | |
| "greedy_hits_after": sum(p["greedy_after"] for p in probes if p["group"] == g)} | |
| return {"probes": probes, "groups": groups} | |
| def analyze_run(run_dir: str) -> dict[str, Any]: | |
| out: dict[str, Any] = {"run": os.path.basename(os.path.normpath(run_dir))} | |
| sess = os.path.join(run_dir, "sessions", "teach") | |
| if os.path.exists(os.path.join(sess, "transactions.jsonl")): | |
| rows = turn_write_table(load_jsonl(os.path.join(sess, "transactions.jsonl")), load_jsonl(os.path.join(sess, "trace.jsonl"))) | |
| writes = [r["delta_norm_sum"] for r in rows] | |
| out["teach_turns"] = rows | |
| out["write_summary"] = {"turns": len(rows), "median_delta_norm_sum": statistics.median(writes) if writes else None, | |
| "planted_ranks": sorted(r["rank_by_write"] for r in rows if r["planted"]), | |
| "largest_write_prompt": max(rows, key=lambda r: r["delta_norm_sum"])["prompt"][:60] if rows else None} | |
| else: | |
| out["teach_turns"] = None # sessions not archived for this run: absent, not zero | |
| anchor = os.path.join(run_dir, "sleep_anchor_report.json") | |
| if os.path.exists(anchor): | |
| rep = json.load(open(anchor, encoding="utf-8")) | |
| if rep.get("before", {}).get("recall") and rep.get("after", {}).get("recall"): | |
| out["anchor_storage_shift"] = storage_shift(rep) | |
| out["anchor_status"] = rep.get("status") | |
| return out | |
| def main() -> None: | |
| ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) | |
| ap.add_argument("--runs", nargs="+", required=True, help="archived run folders (results/sleep-2026-09-23/<run>)") | |
| ap.add_argument("--out", required=True, help="JSON output path") | |
| args = ap.parse_args() | |
| results = [analyze_run(r) for r in args.runs] | |
| with open(args.out, "w", encoding="utf-8") as f: | |
| json.dump({"analysis": "anchor_transfer", "runs": results}, f, indent=1) | |
| for res in results: | |
| print(f"== {res['run']}") | |
| ws = res.get("write_summary") | |
| if ws: | |
| print(f" teach turns {ws['turns']}, median write {ws['median_delta_norm_sum']:.2f}, planted turns rank {ws['planted_ranks']}, largest: {ws['largest_write_prompt']!r}") | |
| st = res.get("anchor_storage_shift") | |
| if st: | |
| g = st["groups"] | |
| print(" anchor storage shift: " + ", ".join(f"{k} n={v['n']} mean {v['mean_shift']:+.3f} greedy {v['greedy_hits_before']}->{v['greedy_hits_after']}" for k, v in g.items() if v["n"])) | |
| if __name__ == "__main__": | |
| main() | |