Skills › Scientific Agent Skills › hypothesis-generation › check_falsification_controls.py
check_falsification_controls.py
455 lines · skills/hypothesis-generation/scripts/check_falsification_controls.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit falsifiers, discriminating tests, nulls, and controls without scoring."""
4from __future__ import annotations
6import argparse
7import re
8from collections import Counter
9from typing import Any
11from _common import (
12 ValidationError,
13 error_exit,
14 issue,
15 read_json,
16 require_enum,
17 require_exact_keys,
18 require_identifier,
19 require_list,
20 require_object,
21 require_text,
22 require_text_list,
23 require_unique,
24 write_json_report,
25)
26from validate_hypothesis_schema import load_hypothesis_record, validate_record
28CONTROL_TYPES = {
29 "negative_exposure",
30 "negative_outcome",
31 "procedural_negative",
32 "positive_control",
33 "vehicle_or_sham",
34 "other",
35}
36NEGATIVE_TYPES = {"negative_exposure", "negative_outcome", "procedural_negative"}
39def _parse_falsifier(raw: Any, context: str) -> dict[str, Any]:
40 value = require_object(raw, context)
41 require_exact_keys(
42 value,
43 required={
44 "prediction_id",
45 "conditions",
46 "observable",
47 "incompatible_result",
48 "assumption_failure_checks",
49 },
50 context=context,
51 )
52 return {
53 "prediction_id": require_identifier(
54 value["prediction_id"], f"{context}.prediction_id"
55 ),
56 "conditions": require_text(
57 value["conditions"], f"{context}.conditions", minimum=10
58 ),
59 "observable": require_text(
60 value["observable"], f"{context}.observable", minimum=5
61 ),
62 "incompatible_result": require_text(
63 value["incompatible_result"],
64 f"{context}.incompatible_result",
65 minimum=10,
66 ),
67 "assumption_failure_checks": require_text_list(
68 value["assumption_failure_checks"],
69 f"{context}.assumption_failure_checks",
70 minimum=1,
71 ),
72 }
75def _parse_discriminating_tests(raw: Any, context: str) -> list[dict[str, str]]:
76 values = require_list(raw, context, minimum=1, maximum=100)
77 parsed: list[dict[str, str]] = []
78 identifiers: list[str] = []
79 fields = {
80 "test_id",
81 "rival_hypothesis_id",
82 "focal_expected",
83 "rival_expected",
84 "indeterminate_result",
85 }
86 for index, raw_value in enumerate(values):
87 item_context = f"{context}[{index}]"
88 value = require_object(raw_value, item_context)
89 require_exact_keys(value, required=fields, context=item_context)
90 test_id = require_identifier(value["test_id"], f"{item_context}.test_id")
91 identifiers.append(test_id)
92 parsed.append(
93 {
94 "test_id": test_id,
95 "rival_hypothesis_id": require_identifier(
96 value["rival_hypothesis_id"],
97 f"{item_context}.rival_hypothesis_id",
98 ),
99 "focal_expected": require_text(
100 value["focal_expected"],
101 f"{item_context}.focal_expected",
102 minimum=10,
103 ),
104 "rival_expected": require_text(
105 value["rival_expected"],
106 f"{item_context}.rival_expected",
107 minimum=10,
108 ),
109 "indeterminate_result": require_text(
110 value["indeterminate_result"],
111 f"{item_context}.indeterminate_result",
112 minimum=10,
113 ),
114 }
115 )
116 require_unique(identifiers, context)
117 return parsed
120def _parse_nulls(raw: Any, context: str) -> list[dict[str, str]]:
121 values = require_list(raw, context, minimum=1, maximum=100)
122 parsed: list[dict[str, str]] = []
123 identifiers: list[str] = []
124 fields = {"null_id", "statement", "analysis_id", "interpretation_limit"}
125 for index, raw_value in enumerate(values):
126 item_context = f"{context}[{index}]"
127 value = require_object(raw_value, item_context)
128 require_exact_keys(value, required=fields, context=item_context)
129 null_id = require_identifier(value["null_id"], f"{item_context}.null_id")
130 identifiers.append(null_id)
131 parsed.append(
132 {
133 "null_id": null_id,
134 "statement": require_text(
135 value["statement"], f"{item_context}.statement", minimum=10
136 ),
137 "analysis_id": require_identifier(
138 value["analysis_id"], f"{item_context}.analysis_id"
139 ),
140 "interpretation_limit": require_text(
141 value["interpretation_limit"],
142 f"{item_context}.interpretation_limit",
143 minimum=10,
144 ),
145 }
146 )
147 require_unique(identifiers, context)
148 return parsed
151def _parse_controls(raw: Any, context: str) -> list[dict[str, str]]:
152 values = require_list(raw, context, minimum=1, maximum=100)
153 parsed: list[dict[str, str]] = []
154 identifiers: list[str] = []
155 fields = {
156 "control_id",
157 "control_type",
158 "rationale",
159 "expected_result",
160 "failure_implication",
161 }
162 for index, raw_value in enumerate(values):
163 item_context = f"{context}[{index}]"
164 value = require_object(raw_value, item_context)
165 require_exact_keys(value, required=fields, context=item_context)
166 control_id = require_identifier(
167 value["control_id"], f"{item_context}.control_id"
168 )
169 identifiers.append(control_id)
170 parsed.append(
171 {
172 "control_id": control_id,
173 "control_type": require_enum(
174 value["control_type"],
175 CONTROL_TYPES,
176 f"{item_context}.control_type",
177 ),
178 "rationale": require_text(
179 value["rationale"], f"{item_context}.rationale", minimum=10
180 ),
181 "expected_result": require_text(
182 value["expected_result"],
183 f"{item_context}.expected_result",
184 minimum=5,
185 ),
186 "failure_implication": require_text(
187 value["failure_implication"],
188 f"{item_context}.failure_implication",
189 minimum=10,
190 ),
191 }
192 )
193 require_unique(identifiers, context)
194 return parsed
197def _parse_outcome_interpretation(raw: Any, context: str) -> dict[str, str]:
198 value = require_object(raw, context)
199 fields = {
200 "consistent_with_candidate",
201 "challenges_candidate",
202 "supports_neither_or_mixed",
203 }
204 require_exact_keys(value, required=fields, context=context)
205 return {
206 field: require_text(value[field], f"{context}.{field}", minimum=10)
207 for field in fields
208 }
211def load_checklist(payload: Any) -> dict[str, Any]:
212 root = require_object(payload, "checklist")
213 require_exact_keys(
214 root,
215 required={"schema_version", "checklist_id", "record_id", "hypotheses"},
216 context="checklist",
217 )
218 raw_hypotheses = require_list(
219 root["hypotheses"], "checklist.hypotheses", minimum=1, maximum=50
220 )
221 parsed_hypotheses: list[dict[str, Any]] = []
222 hypothesis_ids: list[str] = []
223 fields = {
224 "hypothesis_id",
225 "candidate_status",
226 "assumptions",
227 "boundary_conditions",
228 "falsifier",
229 "discriminating_tests",
230 "nulls",
231 "controls",
232 "outcome_interpretation",
233 "human_review_status",
234 }
235 for index, raw_hypothesis in enumerate(raw_hypotheses):
236 context = f"checklist.hypotheses[{index}]"
237 hypothesis = require_object(raw_hypothesis, context)
238 require_exact_keys(hypothesis, required=fields, context=context)
239 hypothesis_id = require_identifier(
240 hypothesis["hypothesis_id"], f"{context}.hypothesis_id"
241 )
242 hypothesis_ids.append(hypothesis_id)
243 parsed_hypotheses.append(
244 {
245 "hypothesis_id": hypothesis_id,
246 "candidate_status": require_enum(
247 hypothesis["candidate_status"],
248 {"candidate"},
249 f"{context}.candidate_status",
250 ),
251 "assumptions": require_text_list(
252 hypothesis["assumptions"],
253 f"{context}.assumptions",
254 minimum=1,
255 ),
256 "boundary_conditions": require_text_list(
257 hypothesis["boundary_conditions"],
258 f"{context}.boundary_conditions",
259 minimum=1,
260 ),
261 "falsifier": _parse_falsifier(
262 hypothesis["falsifier"], f"{context}.falsifier"
263 ),
264 "discriminating_tests": _parse_discriminating_tests(
265 hypothesis["discriminating_tests"],
266 f"{context}.discriminating_tests",
267 ),
268 "nulls": _parse_nulls(
269 hypothesis["nulls"], f"{context}.nulls"
270 ),
271 "controls": _parse_controls(
272 hypothesis["controls"], f"{context}.controls"
273 ),
274 "outcome_interpretation": _parse_outcome_interpretation(
275 hypothesis["outcome_interpretation"],
276 f"{context}.outcome_interpretation",
277 ),
278 "human_review_status": require_enum(
279 hypothesis["human_review_status"],
280 {"pending", "complete", "specialist_required"},
281 f"{context}.human_review_status",
282 ),
283 }
284 )
285 require_unique(hypothesis_ids, "checklist.hypotheses")
286 return {
287 "schema_version": require_enum(
288 root["schema_version"], {"2.0"}, "checklist.schema_version"
289 ),
290 "checklist_id": require_identifier(
291 root["checklist_id"], "checklist.checklist_id"
292 ),
293 "record_id": require_identifier(root["record_id"], "checklist.record_id"),
294 "hypotheses": parsed_hypotheses,
295 }
298def _normalize(value: str) -> str:
299 return re.sub(r"[^a-z0-9]+", " ", value.casefold()).strip()
302def audit(
303 checklist: dict[str, Any], record: dict[str, Any] | None = None
304) -> dict[str, Any]:
305 errors: list[dict[str, str]] = []
306 warnings: list[dict[str, str]] = []
307 review_counts: Counter[str] = Counter()
308 test_ids: list[str] = []
309 null_ids: list[str] = []
311 record_ids: dict[str, set[str]] | None = None
312 if record is not None:
313 record_report = validate_record(record)
314 if not record_report["valid"]:
315 raise ValidationError(
316 "optional hypothesis record must pass schema validation first"
317 )
318 record_ids = {
319 "hypotheses": {
320 item["hypothesis_id"] for item in record["hypotheses"]
321 },
322 "predictions": {
323 item["prediction_id"] for item in record["predictions"]
324 },
325 "analyses": {
326 item["analysis_id"] for item in record["analysis_plan"]["analyses"]
327 },
328 "controls": {
329 item["control_id"] for item in record["negative_controls"]
330 },
331 "nulls": {item["null_id"] for item in record["null_hypotheses"]},
332 }
334 for hypothesis in checklist["hypotheses"]:
335 hypothesis_id = hypothesis["hypothesis_id"]
336 review_counts[hypothesis["human_review_status"]] += 1
337 if hypothesis["human_review_status"] != "complete":
338 warnings.append(issue("HUMAN_REVIEW_INCOMPLETE", hypothesis_id))
340 falsifier = hypothesis["falsifier"]
341 if record_ids is not None:
342 if hypothesis_id not in record_ids["hypotheses"]:
343 errors.append(issue("UNKNOWN_HYPOTHESIS_ID", hypothesis_id))
344 if falsifier["prediction_id"] not in record_ids["predictions"]:
345 errors.append(
346 issue("UNKNOWN_PREDICTION_ID", falsifier["prediction_id"])
347 )
349 for test in hypothesis["discriminating_tests"]:
350 test_ids.append(test["test_id"])
351 if test["rival_hypothesis_id"] == hypothesis_id:
352 errors.append(issue("FOCAL_HYPOTHESIS_LISTED_AS_RIVAL", test["test_id"]))
353 if _normalize(test["focal_expected"]) == _normalize(
354 test["rival_expected"]
355 ):
356 errors.append(
357 issue("FOCAL_AND_RIVAL_EXPECTATIONS_IDENTICAL", test["test_id"])
358 )
359 if (
360 record_ids is not None
361 and test["rival_hypothesis_id"] not in record_ids["hypotheses"]
362 ):
363 errors.append(
364 issue("UNKNOWN_RIVAL_HYPOTHESIS_ID", test["test_id"])
365 )
367 for null in hypothesis["nulls"]:
368 null_ids.append(null["null_id"])
369 if record_ids is not None:
370 if null["null_id"] not in record_ids["nulls"]:
371 errors.append(issue("UNKNOWN_NULL_ID", null["null_id"]))
372 if null["analysis_id"] not in record_ids["analyses"]:
373 errors.append(
374 issue("UNKNOWN_ANALYSIS_ID", null["analysis_id"])
375 )
377 negative_controls = [
378 control
379 for control in hypothesis["controls"]
380 if control["control_type"] in NEGATIVE_TYPES
381 ]
382 if not negative_controls:
383 errors.append(issue("NEGATIVE_CONTROL_REQUIRED", hypothesis_id))
384 if record_ids is not None:
385 for control in negative_controls:
386 if control["control_id"] not in record_ids["controls"]:
387 errors.append(
388 issue("UNKNOWN_NEGATIVE_CONTROL_ID", control["control_id"])
389 )
391 require_unique(test_ids, "checklist.discriminating_tests")
392 require_unique(null_ids, "checklist.nulls")
394 return {
395 "schema_version": "2.0",
396 "checklist_id": checklist["checklist_id"],
397 "record_id": checklist["record_id"],
398 "valid": not errors,
399 "status": (
400 "INVALID_CHECKLIST"
401 if errors
402 else "VALID_PENDING_HUMAN_REVIEW"
403 if any(item["human_review_status"] != "complete" for item in checklist["hypotheses"])
404 else "VALID_HUMAN_REVIEW_DECLARED_COMPLETE"
405 ),
406 "errors": errors,
407 "warnings": warnings,
408 "hypothesis_ids": sorted(
409 item["hypothesis_id"] for item in checklist["hypotheses"]
410 ),
411 "discriminating_test_ids": sorted(test_ids),
412 "null_ids": sorted(null_ids),
413 "human_review_status_counts": dict(sorted(review_counts.items())),
414 "record_cross_check_performed": record is not None,
415 "notice": (
416 "This audit checks declared falsifiers, rival contrasts, nulls, "
417 "controls, and links. It does not prove falsifiability, validate "
418 "control assumptions, interpret results, or select a hypothesis."
419 ),
420 }
423def build_parser() -> argparse.ArgumentParser:
424 parser = argparse.ArgumentParser(
425 description=(
426 "Audit a bounded local falsification/control JSON checklist without "
427 "scientific scoring or candidate selection."
428 )
429 )
430 parser.add_argument("checklist", help="Local falsification/control checklist JSON")
431 parser.add_argument(
432 "--record", help="Optional local hypothesis record JSON for cross-checks"
433 )
434 parser.add_argument("-o", "--output", help="Optional local JSON report")
435 parser.add_argument(
436 "--force", action="store_true", help="Replace an existing output file"
437 )
438 return parser
441def main() -> int:
442 args = build_parser().parse_args()
443 try:
444 record = (
445 load_hypothesis_record(read_json(args.record)) if args.record else None
446 )
447 report = audit(load_checklist(read_json(args.checklist)), record)
448 write_json_report(report, args.output, force=args.force)
449 return 0 if report["valid"] else 1
450 except ValidationError as exc:
451 return error_exit(exc)
454if __name__ == "__main__":
455 raise SystemExit(main())