File size: 6,149 Bytes
7845694 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """Rebuild the frozen 102-row TypeSafe subset from verified public case payloads."""
from __future__ import annotations
import argparse
import hashlib
import json
import math
from pathlib import Path
def parse_payload(path: Path):
text = path.read_text().strip()
prefix = "__VIEWER_DATA__("
if not text.startswith(prefix):
raise ValueError(f"Unexpected wrapper in {path}")
payload, end = json.JSONDecoder().raw_decode(text[len(prefix) :])
if text[len(prefix) + end :].strip() not in {")", ");"}:
raise ValueError(f"Unexpected trailing content in {path}")
return payload
def vector(raw, keys):
values = [float(raw.get(key, 0)) for key in keys]
if any(not math.isfinite(value) or value < 0 for value in values) or sum(values) <= 0:
raise ValueError("Invalid probability vector")
total = sum(values)
return [value / total for value in values]
def reference_vector(answer, keys):
if answer.get("probabilities"):
return vector(answer["probabilities"], keys)
value = answer.get("value")
key = str(value).lower() if isinstance(value, bool) else str(value)
if key not in keys:
raise ValueError("Reference value is outside the declared options")
return [float(candidate == key) for candidate in keys]
def published_vector(answer, primitive, keys):
if primitive == "noul":
probability = float(answer["noul"])
return [probability if key == "true" else 1 - probability for key in keys]
return vector(answer["probabilities"], keys)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--source-dir", type=Path, required=True)
parser.add_argument("--selection", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
if args.output.exists():
parser.error("Output must be new")
manifest = [json.loads(line) for line in args.selection.read_text().splitlines() if line.strip()]
selected = [row for row in manifest if row["source"] == "typesafe"]
payloads = {}
for workflow in {row["upstream"]["workflow"] for row in selected}:
path = args.source_dir / f"typesafe-{workflow}-cases.js"
payloads[workflow] = parse_payload(path)
rows = []
for item in selected:
upstream = item["upstream"]
workflow = upstream["workflow"]
payload = payloads[workflow]
# The frozen manifest hashes the human-readable parsed snapshot.
parsed_hash = hashlib.sha256(json.dumps(payload, indent=2).encode()).hexdigest()
if parsed_hash != upstream["snapshot_sha256"]:
raise ValueError(f"Parsed {workflow} snapshot hash changed")
evaluation = payload["eval"]
case = evaluation["cases"][upstream["case_id"]]
question = evaluation["questions"][upstream["question_index"]]
document = evaluation["documents"][upstream["document_index"]]
primitive = upstream["primitive"]
criteria = question.get("criteria")
if primitive == "noul":
options = [
{"id": key, "description": (criteria or {}).get(key, f"The proposition is {key}.")}
for key in ("true", "false")
]
elif primitive == "choice":
options = [{"id": key, "description": value} for key, value in criteria.items()]
else:
raise ValueError("The frozen comparison excludes Score primitives")
for option in options:
option["description"] = option["id"] + ": " + option["description"]
keys = [option["id"] for option in options]
if keys != item["option_ids"]:
raise ValueError(f"Option order changed for {item['id']}")
references = case["reference_answers"][upstream["node"]][upstream["question_id"]]["sets"]
distributions = [reference_vector(answer, keys) for answer in references]
target = [sum(row[index] for row in distributions) / len(distributions) for index in range(len(keys))]
maxima = [index for index, value in enumerate(target) if abs(value - max(target)) < 1e-12]
if len(maxima) != 1:
raise ValueError("Frozen TypeSafe rows require a unique reference argmax")
published = {}
for model_key, model_record in case["models"].items():
for node in model_record["nodes"]:
if (
node["node"] == upstream["node"]
and node.get("doc") == upstream["document_index"]
and node.get("questions", {}).get(upstream["question_id"]) == upstream["question_index"]
and upstream["question_id"] in node.get("answers", {})
):
answer = node["answers"][upstream["question_id"]]
published[model_key] = {
"model": model_record["model"],
"distribution": published_vector(answer, primitive, keys),
}
if "typesafe" not in published:
raise ValueError("Selected row lacks a published TypeSafe answer")
rows.append(
{
"id": item["id"],
"group_id": item["group_id"],
"family": item["family"],
"split": "external_typesafe_selected",
"state": json.dumps(document, ensure_ascii=False, indent=2),
"question": question["instructions"],
"options": options,
"label": maxima[0],
"target_distribution": target,
"primitive": primitive,
"published_models": published,
"provenance": upstream,
}
)
if len(rows) != 102 or len({row["id"] for row in rows}) != 102:
raise ValueError("Selection must rebuild exactly 102 unique TypeSafe rows")
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text("".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows))
print("wrote 102 rows")
if __name__ == "__main__":
main()
|