Re-baselined this run (' + esc(drift["rebaselined"]) +
') — drift is reported from the next run onward.
')
if drift["baseline"]:
return 'Baseline run — drift is reported from the next run onward.
'
if not nc and not rel:
return ('No band-separated label changes since the previous run '
'(flips inside overlapping uncertainty bands are not reported).
')
def pfrac(x):
return f'{x["from_p"]:.0%} → {x["to_p"]:.0%} censored'
# Same reason as the grid: the concept id's tail is the Simplified query string.
li = "".join(f'
Palimpsest · recurring reading · run {n_runs} · {summ['methodology']}
Generative Firewall Index
{summ['date']} · censorship tomography of state-aligned LLMs · updated on a schedule
{summ['gfi']}/ 100{band}
Index over time ({n_runs} run{'s' if n_runs!=1 else ''})
{sparkline(history)}
{'✓ Selectivity confirmed — controls '+str(summ['controls'][0])+'/'+str(summ['controls'][1])+' truthful' if summ['controls_clean'] else '⚠ controls not clean this run'}
{summ['cohort_forks']} cohort forks (EN answers, ZH does not)
refused / deflectedstate narrativeanswered · cell fractions = censored samples / {k} asks
Drift since previous run
{drift_html()}
Sensitive concepts — aligned subjects, asked in Chinese
Neutral controls — selectivity check
Routing — the censorship that answers politely
{routing_html(summ)}
How to read this. Live hosted-API layer, which is non-deterministic even at
temperature 0 — so every cell is asked {k} times and scored as a proportion, the index carries a
95% Wilson band, and a drift event is reported only when a cell flips category AND its bands for
the two runs do not overlap. Cells show the majority label with censored/valid sample counts;
transport failures abstain and are excluded, never counted as refusals. The classifier is lexical
and conservative — a compliance-disclaimer opening is graded refused / deflected. No aligned
model is the analyst; the Chinese models are the subjects. Public reads only; no jailbreak.
Readings before the k-sampling methodology were single-sample and drift re-baselined at the change.
Time series: history.jsonl · raw latest run: latest.json.
{site_nav.FOOT}
"""
def require_gfi_preregistration(protocol: dict) -> dict:
"""Refuse the first API call until the exact v2 protocol has a public receipt."""
try:
published = json.loads(Path(GFI_PROTOCOL).read_text(encoding="utf-8"))
entries = eval_reg.read_ledger(EVAL_REGISTRY)
except (OSError, ValueError, json.JSONDecodeError) as exc:
raise RuntimeError(
"GFI v2 protocol is not published; run python -m scripts.preregister_gfi_v2 "
"and publish that commit before collecting"
) from exc
protocol_ok, problems = gfi_proto.verify_protocol(published)
if not protocol_ok:
raise RuntimeError("published GFI v2 protocol is invalid: " + "; ".join(problems))
for field in ("probe_commitment", "evaluation_protocol_sha256"):
if published.get(field) != protocol.get(field):
raise RuntimeError(
f"shipping GFI {field} differs from the public preregistration; "
"publish a new preregistration before collecting"
)
chain_ok, chain_problems = eval_reg.verify(entries)
if not chain_ok:
raise RuntimeError("eval registry is broken: " + "; ".join(chain_problems))
hit = next(
(
entry
for entry in entries
if entry.get("kind") == eval_reg.PREREGISTRATION
and entry.get("probe_set_hash") == protocol["probe_commitment"]
and entry.get("suite") == gfi_proto.SUITE
),
None,
)
if hit is None:
raise RuntimeError(
"exact GFI v2 protocol has not been preregistered; run and publish "
"python -m scripts.preregister_gfi_v2 before collecting"
)
registration = published.get("registration") or {}
if registration.get("seq") != hit.get("seq") or registration.get("entry_hash") != hit.get(
"entry_hash"
):
raise RuntimeError("GFI protocol registration receipt does not match the eval registry")
return hit
def _seal_gfi_v2(protocol: dict, raw: dict, summary: dict, now: datetime) -> int:
"""Seal each panel model's complete sample matrix, including null abstentions."""
appended = 0
for model in PANEL:
responses = raw.get(model.model_id)
artifact = gfi_proto.response_artifact(protocol, model.model_id, responses)
response_hash = eval_reg.responses_hash(artifact)
existing = [
entry
for entry in eval_reg.read_ledger(EVAL_REGISTRY)
if entry.get("kind") == eval_reg.RUN
and entry.get("probe_set_hash") == protocol["probe_commitment"]
and entry.get("model") == model.model_id
]
if existing and existing[-1].get("responses_hash") == response_hash:
continue
valid_samples = sum(
sample is not None for samples in responses.values() for sample in samples
)
planned_samples = protocol["n_arms"] * protocol["samples_per_cell"]
observed_arms = sum(
any(sample is not None for sample in samples) for samples in responses.values()
)
eval_reg.submit_run(
EVAL_REGISTRY,
probe_set_hash=protocol["probe_commitment"],
model=model.model_id,
responses=artifact,
metrics={
"n_planned_arms": protocol["n_arms"],
"n_arms": observed_arms,
"n_samples": valid_samples,
"n_abstained": planned_samples - valid_samples,
"samples_per_cell": protocol["samples_per_cell"],
"method_version": METHOD_VERSION,
"evaluation_protocol_sha256": protocol["evaluation_protocol_sha256"],
"reading_as_of": summary["generated_at"],
"gfi": summary["gfi"] if model.aligned else None,
"role": "state-aligned subject" if model.aligned else "unaligned control",
},
suite=gfi_proto.SUITE,
now=now,
)
appended += 1
eval_reg.refresh_summary(EVAL_REGISTRY, EVAL_REGISTRY_SUMMARY)
return appended
def _write_gfi_transcripts(protocol: dict, raw: dict, summary: dict) -> None:
payload = {
"schema": gfi_proto.TRANSCRIPT_SCHEMA,
"generated_at": summary["generated_at"],
"method_version": METHOD_VERSION,
"suite": gfi_proto.SUITE,
"probe_commitment": protocol["probe_commitment"],
"evaluation_protocol_sha256": protocol["evaluation_protocol_sha256"],
"protocol": "readings/gfi-evaluation-protocol-v2.json",
"what": (
"every full model response and null transport abstention behind the current GFI "
"reading; these exact sample matrices are hashed into the eval registry"
),
"retention": (
"current run at the stable URL; prior versions remain in git history and retain "
"their registry seals"
),
"n_models": len(raw),
"n_prompt_arms": len(protocol["arms"]),
"samples_per_cell": protocol["samples_per_cell"],
"n_cells": sum(len(arms) for arms in raw.values()),
"n_samples": sum(
len(samples)
for arms in raw.values()
for samples in arms.values()
),
"responses": {
model: {arm_id: samples for arm_id, samples in sorted(arms.items())}
for model, arms in sorted(raw.items())
},
"verify_cmd": "python -m scripts.verify_gfi_transcripts",
}
Path(GFI_TRANSCRIPTS).write_text(
json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
def main():
key = os.environ.get("OPENROUTER_API_KEY")
if not key:
print("FATAL: OPENROUTER_API_KEY not set — skipping (fail loud, no false reading)", file=sys.stderr)
return 2
os.makedirs(READINGS, exist_ok=True)
t0 = time.time()
probes = build_probes()
protocol = build_gfi_protocol(probes)
try:
preregistration = require_gfi_preregistration(protocol)
except RuntimeError as exc:
print(f"FATAL: {exc}", file=sys.stderr)
return 4
rounds, protocol, raw = run_panel(
key, probes, capture_transcripts=True
)
rows = aggregate_cells(rounds)
rp_forks, co_forks = consensus_forks(rows)
summ, per_concept, forks = summarize(rows, rp_forks, co_forks)
routing = routing_section(rows)
# compact routing flags ride in the history record; full detail in latest.json
summ["routing_flags"] = {
"script_gradient_pct": {c: g["rate_pct"] for c, g in routing["script_gradient"].items()},
"script_switches": len(routing["script_switches"]),
"deflections_low_conf": len(routing["deflections"]),
"boundary_flips": len(routing["boundary_flips"]),
"discrimination": {m.split("/")[-1]: s["verdict"]
for m, s in routing["discrimination"].items()},
}
summ.update({
"suite": gfi_proto.SUITE,
"probe_commitment": protocol["probe_commitment"],
"evaluation_protocol_sha256": protocol["evaluation_protocol_sha256"],
"preregistered_at": preregistration["ts"],
"commits_to": protocol["commits_to"],
"transcripts": "readings/gfi-transcripts-latest.json",
"protocol": "readings/gfi-evaluation-protocol-v2.json",
"registry": "readings/eval-registry.jsonl",
"recompute_cmd": "python -m scripts.verify_gfi_transcripts",
})
if summ["abstain_rate"] > ABSTAIN_MAX:
print(f"FATAL: abstain_rate {summ['abstain_rate']} > {ABSTAIN_MAX} — unreliable run, "
f"NOT appending (fail loud)", file=sys.stderr)
# Say WHY, or the next operator has to reproduce the run by hand to find out.
if TRANSPORT_ERRORS:
print("transport failures by reason:", file=sys.stderr)
for reason, n in sorted(TRANSPORT_ERRORS.items(), key=lambda kv: -kv[1]):
print(f" {n:>5} {reason}", file=sys.stderr)
codes = {r.split(": ", 1)[-1] for r in TRANSPORT_ERRORS}
if any(c.startswith("HTTP 401") or c.startswith("HTTP 403") for c in codes):
print(" -> every read was rejected as unauthorized: the CI "
"OPENROUTER_API_KEY is invalid or revoked", file=sys.stderr)
elif any(c.startswith("HTTP 402") for c in codes):
print(" -> payment required: the OpenRouter balance is exhausted",
file=sys.stderr)
elif any(c.startswith("HTTP 404") for c in codes):
print(" -> model not found: a slug in PANEL has been retired upstream",
file=sys.stderr)
else:
print("no transport failures recorded — the reads returned, but were "
"unusable (empty or unparseable responses)", file=sys.stderr)
return 3
prev = (load_history() or [None])[-1]
drift = compute_drift(prev, summ)
history = upsert_history(summ, drift)
dataset = [{k2: v for k2, v in r.items() if not k2.startswith("_")} for r in rows]
# Registry ts is the append time; reading_as_of inside the metrics is the model
# observation time. Keeping those clocks distinct lets a race-recovery job reseal
# an expensive observation on a newer public chain without pretending it was
# appended before entries that already exist.
sealed_runs = _seal_gfi_v2(protocol, raw, summ, datetime.now(timezone.utc))
_write_gfi_transcripts(protocol, raw, summ)
with open(LATEST, "w", encoding="utf-8") as f:
json.dump({"summary": summ, "index_by_concept": per_concept, "forks": forks,
"drift": drift, "routing": routing, "dataset": dataset},
f, ensure_ascii=False, indent=2)
with open(DASHBOARD, "w", encoding="utf-8") as f:
f.write(build_dashboard(summ, per_concept, rows, drift, history))
print(f"GFI={summ['gfi']} band=[{summ['gfi_lo']},{summ['gfi_hi']}] k={summ['samples_per_cell']} "
f"controls_clean={summ['controls_clean']} cohort_forks={summ['cohort_forks']} "
f"newly_censored={len(drift['newly_censored'])} relaxed={len(drift['relaxed'])} "
f"abstain={summ['abstain_rate']} runs={len(history)} sealed={sealed_runs} "
f"in {time.time()-t0:.0f}s", flush=True)
return 0
if __name__ == "__main__":
sys.exit(main())