Skills › Scientific Agent Skills › hypothesis-generation › validate_hypothesis_schema.py
validate_hypothesis_schema.py
1078 lines · skills/hypothesis-generation/scripts/validate_hypothesis_schema.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a structured hypothesis record without judging scientific merit."""
4from __future__ import annotations
6import argparse
7from collections import Counter
8from pathlib import PurePosixPath
9from typing import Any
11from _common import (
12 ValidationError,
13 error_exit,
14 issue,
15 read_json,
16 require_bool,
17 require_enum,
18 require_exact_keys,
19 require_identifier,
20 require_identifier_list,
21 require_iso_date,
22 require_list,
23 require_object,
24 require_text,
25 require_text_list,
26 require_unique,
27 write_json_report,
28)
30QUESTION_TYPES = {"descriptive", "associational", "predictive", "causal", "mechanistic"}
31HYPOTHESIS_STATUS = {"candidate"}
32MEASUREMENT_ROLES = {
33 "intervention",
34 "exposure",
35 "outcome",
36 "mediator",
37 "confounder",
38 "selection",
39 "negative_control",
40 "positive_control",
41 "effect_modifier",
42 "other",
43}
44RISK_TYPES = {
45 "confounding",
46 "selection_bias",
47 "collider_bias",
48 "reverse_causation",
49 "measurement_bias",
50 "other",
51}
52CONTROL_TYPES = {
53 "negative_exposure",
54 "negative_outcome",
55 "procedural_negative",
56 "positive_control",
57 "vehicle_or_sham",
58 "other",
59}
60PLAN_MODES = {"confirmatory", "exploratory"}
61GATE_STATES = {"not_applicable", "undetermined", "requires_review", "approved", "blocked"}
62FEASIBILITY_STATES = {
63 "undetermined",
64 "feasible_for_planning",
65 "requires_pilot",
66 "infeasible",
67 "blocked",
68}
71def _parse_observation(raw: Any) -> dict[str, Any]:
72 value = require_object(raw, "record.observation")
73 require_exact_keys(
74 value,
75 required={"statement", "provenance", "source_ids", "uncertainties"},
76 context="record.observation",
77 )
78 return {
79 "statement": require_text(
80 value["statement"], "record.observation.statement", minimum=10
81 ),
82 "provenance": require_text(
83 value["provenance"], "record.observation.provenance", minimum=10
84 ),
85 "source_ids": require_identifier_list(
86 value["source_ids"], "record.observation.source_ids", minimum=1
87 ),
88 "uncertainties": require_text_list(
89 value["uncertainties"],
90 "record.observation.uncertainties",
91 minimum=1,
92 ),
93 }
96def _parse_question(raw: Any) -> dict[str, str]:
97 value = require_object(raw, "record.research_question")
98 fields = {
99 "statement",
100 "framework",
101 "question_type",
102 "population_or_system",
103 "intervention_or_exposure",
104 "comparator",
105 "outcome",
106 "timeframe",
107 }
108 require_exact_keys(value, required=fields, context="record.research_question")
109 parsed = {
110 field: require_text(
111 value[field],
112 f"record.research_question.{field}",
113 minimum=2 if field == "framework" else 5,
114 )
115 for field in fields
116 if field != "question_type"
117 }
118 parsed["question_type"] = require_enum(
119 value["question_type"],
120 QUESTION_TYPES,
121 "record.research_question.question_type",
122 )
123 return parsed
126def _parse_hypotheses(raw: Any) -> list[dict[str, Any]]:
127 entries = require_list(raw, "record.hypotheses", minimum=1, maximum=50)
128 parsed: list[dict[str, Any]] = []
129 identifiers: list[str] = []
130 fields = {
131 "hypothesis_id",
132 "statement",
133 "mechanism",
134 "status",
135 "source_ids",
136 "assumptions",
137 "boundary_conditions",
138 "uncertainties",
139 "prediction_ids",
140 "rival_hypothesis_ids",
141 }
142 for index, raw_entry in enumerate(entries):
143 context = f"record.hypotheses[{index}]"
144 entry = require_object(raw_entry, context)
145 require_exact_keys(entry, required=fields, context=context)
146 identifier = require_identifier(entry["hypothesis_id"], f"{context}.hypothesis_id")
147 identifiers.append(identifier)
148 parsed.append(
149 {
150 "hypothesis_id": identifier,
151 "statement": require_text(
152 entry["statement"], f"{context}.statement", minimum=10
153 ),
154 "mechanism": require_text(
155 entry["mechanism"], f"{context}.mechanism", minimum=10
156 ),
157 "status": require_enum(
158 entry["status"], HYPOTHESIS_STATUS, f"{context}.status"
159 ),
160 "source_ids": require_identifier_list(
161 entry["source_ids"], f"{context}.source_ids", minimum=1
162 ),
163 "assumptions": require_text_list(
164 entry["assumptions"], f"{context}.assumptions", minimum=1
165 ),
166 "boundary_conditions": require_text_list(
167 entry["boundary_conditions"],
168 f"{context}.boundary_conditions",
169 minimum=1,
170 ),
171 "uncertainties": require_text_list(
172 entry["uncertainties"], f"{context}.uncertainties", minimum=1
173 ),
174 "prediction_ids": require_identifier_list(
175 entry["prediction_ids"], f"{context}.prediction_ids", minimum=1
176 ),
177 "rival_hypothesis_ids": require_identifier_list(
178 entry["rival_hypothesis_ids"],
179 f"{context}.rival_hypothesis_ids",
180 ),
181 }
182 )
183 require_unique(identifiers, "record.hypotheses")
184 return parsed
187def _parse_estimands(raw: Any) -> list[dict[str, Any]]:
188 entries = require_list(raw, "record.causal_estimands", maximum=50)
189 parsed: list[dict[str, Any]] = []
190 identifiers: list[str] = []
191 fields = {
192 "estimand_id",
193 "linked_hypothesis_ids",
194 "population",
195 "intervention_or_exposure",
196 "comparator",
197 "outcome",
198 "time_horizon",
199 "population_summary",
200 "intercurrent_event_strategy",
201 "identification_assumptions",
202 }
203 for index, raw_entry in enumerate(entries):
204 context = f"record.causal_estimands[{index}]"
205 entry = require_object(raw_entry, context)
206 require_exact_keys(entry, required=fields, context=context)
207 identifier = require_identifier(entry["estimand_id"], f"{context}.estimand_id")
208 identifiers.append(identifier)
209 parsed.append(
210 {
211 "estimand_id": identifier,
212 "linked_hypothesis_ids": require_identifier_list(
213 entry["linked_hypothesis_ids"],
214 f"{context}.linked_hypothesis_ids",
215 minimum=1,
216 ),
217 "population": require_text(entry["population"], f"{context}.population"),
218 "intervention_or_exposure": require_text(
219 entry["intervention_or_exposure"],
220 f"{context}.intervention_or_exposure",
221 ),
222 "comparator": require_text(
223 entry["comparator"], f"{context}.comparator"
224 ),
225 "outcome": require_text(entry["outcome"], f"{context}.outcome"),
226 "time_horizon": require_text(
227 entry["time_horizon"], f"{context}.time_horizon"
228 ),
229 "population_summary": require_text(
230 entry["population_summary"], f"{context}.population_summary"
231 ),
232 "intercurrent_event_strategy": require_text(
233 entry["intercurrent_event_strategy"],
234 f"{context}.intercurrent_event_strategy",
235 ),
236 "identification_assumptions": require_text_list(
237 entry["identification_assumptions"],
238 f"{context}.identification_assumptions",
239 minimum=1,
240 ),
241 }
242 )
243 require_unique(identifiers, "record.causal_estimands")
244 return parsed
247def _parse_predictions(raw: Any) -> list[dict[str, Any]]:
248 entries = require_list(raw, "record.predictions", minimum=1, maximum=200)
249 parsed: list[dict[str, Any]] = []
250 identifiers: list[str] = []
251 fields = {
252 "prediction_id",
253 "hypothesis_id",
254 "statement",
255 "observable",
256 "conditions",
257 "expected_pattern",
258 "falsifier",
259 "rival_hypothesis_ids",
260 "measurement_ids",
261 "analysis_ids",
262 }
263 for index, raw_entry in enumerate(entries):
264 context = f"record.predictions[{index}]"
265 entry = require_object(raw_entry, context)
266 require_exact_keys(entry, required=fields, context=context)
267 identifier = require_identifier(entry["prediction_id"], f"{context}.prediction_id")
268 identifiers.append(identifier)
269 parsed.append(
270 {
271 "prediction_id": identifier,
272 "hypothesis_id": require_identifier(
273 entry["hypothesis_id"], f"{context}.hypothesis_id"
274 ),
275 "statement": require_text(
276 entry["statement"], f"{context}.statement", minimum=10
277 ),
278 "observable": require_text(
279 entry["observable"], f"{context}.observable", minimum=5
280 ),
281 "conditions": require_text(
282 entry["conditions"], f"{context}.conditions", minimum=5
283 ),
284 "expected_pattern": require_text(
285 entry["expected_pattern"], f"{context}.expected_pattern", minimum=5
286 ),
287 "falsifier": require_text(
288 entry["falsifier"], f"{context}.falsifier", minimum=10
289 ),
290 "rival_hypothesis_ids": require_identifier_list(
291 entry["rival_hypothesis_ids"],
292 f"{context}.rival_hypothesis_ids",
293 minimum=1,
294 ),
295 "measurement_ids": require_identifier_list(
296 entry["measurement_ids"],
297 f"{context}.measurement_ids",
298 minimum=1,
299 ),
300 "analysis_ids": require_identifier_list(
301 entry["analysis_ids"], f"{context}.analysis_ids", minimum=1
302 ),
303 }
304 )
305 require_unique(identifiers, "record.predictions")
306 return parsed
309def _parse_alternatives(raw: Any) -> list[dict[str, Any]]:
310 entries = require_list(raw, "record.alternative_explanations", minimum=1, maximum=100)
311 parsed: list[dict[str, Any]] = []
312 identifiers: list[str] = []
313 fields = {
314 "alternative_id",
315 "statement",
316 "linked_hypothesis_ids",
317 "risk_types",
318 "discriminating_prediction_ids",
319 }
320 for index, raw_entry in enumerate(entries):
321 context = f"record.alternative_explanations[{index}]"
322 entry = require_object(raw_entry, context)
323 require_exact_keys(entry, required=fields, context=context)
324 identifier = require_identifier(entry["alternative_id"], f"{context}.alternative_id")
325 identifiers.append(identifier)
326 risk_types = require_text_list(
327 entry["risk_types"], f"{context}.risk_types", minimum=1, maximum=10
328 )
329 for risk_index, risk_type in enumerate(risk_types):
330 require_enum(risk_type, RISK_TYPES, f"{context}.risk_types[{risk_index}]")
331 require_unique(risk_types, f"{context}.risk_types")
332 parsed.append(
333 {
334 "alternative_id": identifier,
335 "statement": require_text(
336 entry["statement"], f"{context}.statement", minimum=10
337 ),
338 "linked_hypothesis_ids": require_identifier_list(
339 entry["linked_hypothesis_ids"],
340 f"{context}.linked_hypothesis_ids",
341 minimum=1,
342 ),
343 "risk_types": risk_types,
344 "discriminating_prediction_ids": require_identifier_list(
345 entry["discriminating_prediction_ids"],
346 f"{context}.discriminating_prediction_ids",
347 minimum=1,
348 ),
349 }
350 )
351 require_unique(identifiers, "record.alternative_explanations")
352 return parsed
355def _parse_nulls(raw: Any) -> list[dict[str, Any]]:
356 entries = require_list(raw, "record.null_hypotheses", minimum=1, maximum=100)
357 parsed: list[dict[str, Any]] = []
358 identifiers: list[str] = []
359 fields = {
360 "null_id",
361 "statement",
362 "linked_prediction_ids",
363 "rejection_or_compatibility_rule",
364 }
365 for index, raw_entry in enumerate(entries):
366 context = f"record.null_hypotheses[{index}]"
367 entry = require_object(raw_entry, context)
368 require_exact_keys(entry, required=fields, context=context)
369 identifier = require_identifier(entry["null_id"], f"{context}.null_id")
370 identifiers.append(identifier)
371 parsed.append(
372 {
373 "null_id": identifier,
374 "statement": require_text(
375 entry["statement"], f"{context}.statement", minimum=10
376 ),
377 "linked_prediction_ids": require_identifier_list(
378 entry["linked_prediction_ids"],
379 f"{context}.linked_prediction_ids",
380 minimum=1,
381 ),
382 "rejection_or_compatibility_rule": require_text(
383 entry["rejection_or_compatibility_rule"],
384 f"{context}.rejection_or_compatibility_rule",
385 minimum=10,
386 ),
387 }
388 )
389 require_unique(identifiers, "record.null_hypotheses")
390 return parsed
393def _parse_controls(raw: Any) -> list[dict[str, Any]]:
394 entries = require_list(raw, "record.negative_controls", minimum=1, maximum=100)
395 parsed: list[dict[str, Any]] = []
396 identifiers: list[str] = []
397 fields = {
398 "control_id",
399 "control_type",
400 "rationale",
401 "expected_result",
402 "failure_implication",
403 "linked_prediction_ids",
404 }
405 for index, raw_entry in enumerate(entries):
406 context = f"record.negative_controls[{index}]"
407 entry = require_object(raw_entry, context)
408 require_exact_keys(entry, required=fields, context=context)
409 identifier = require_identifier(entry["control_id"], f"{context}.control_id")
410 identifiers.append(identifier)
411 parsed.append(
412 {
413 "control_id": identifier,
414 "control_type": require_enum(
415 entry["control_type"], CONTROL_TYPES, f"{context}.control_type"
416 ),
417 "rationale": require_text(
418 entry["rationale"], f"{context}.rationale", minimum=10
419 ),
420 "expected_result": require_text(
421 entry["expected_result"], f"{context}.expected_result", minimum=5
422 ),
423 "failure_implication": require_text(
424 entry["failure_implication"],
425 f"{context}.failure_implication",
426 minimum=10,
427 ),
428 "linked_prediction_ids": require_identifier_list(
429 entry["linked_prediction_ids"],
430 f"{context}.linked_prediction_ids",
431 minimum=1,
432 ),
433 }
434 )
435 require_unique(identifiers, "record.negative_controls")
436 return parsed
439def _parse_operationalizations(raw: Any) -> list[dict[str, Any]]:
440 entries = require_list(raw, "record.operationalizations", minimum=1, maximum=200)
441 parsed: list[dict[str, Any]] = []
442 identifiers: list[str] = []
443 fields = {
444 "measurement_id",
445 "construct",
446 "variable",
447 "role",
448 "operational_definition",
449 "instrument_or_method",
450 "unit",
451 "timing",
452 "population_or_system",
453 "validity_evidence_source_ids",
454 "reliability_plan",
455 "missingness_plan",
456 "blinding_or_masking",
457 "threshold_rationale",
458 }
459 for index, raw_entry in enumerate(entries):
460 context = f"record.operationalizations[{index}]"
461 entry = require_object(raw_entry, context)
462 require_exact_keys(entry, required=fields, context=context)
463 identifier = require_identifier(entry["measurement_id"], f"{context}.measurement_id")
464 identifiers.append(identifier)
465 text_fields = fields - {
466 "measurement_id",
467 "role",
468 "validity_evidence_source_ids",
469 }
470 parsed_entry: dict[str, Any] = {
471 "measurement_id": identifier,
472 "role": require_enum(entry["role"], MEASUREMENT_ROLES, f"{context}.role"),
473 "validity_evidence_source_ids": require_identifier_list(
474 entry["validity_evidence_source_ids"],
475 f"{context}.validity_evidence_source_ids",
476 minimum=1,
477 ),
478 }
479 for field in text_fields:
480 parsed_entry[field] = require_text(
481 entry[field],
482 f"{context}.{field}",
483 minimum=2 if field in {"unit", "variable"} else 5,
484 )
485 parsed.append(parsed_entry)
486 require_unique(identifiers, "record.operationalizations")
487 return parsed
490def _parse_analysis_plan(raw: Any) -> dict[str, Any]:
491 plan = require_object(raw, "record.analysis_plan")
492 require_exact_keys(
493 plan,
494 required={"analyses", "harking_control", "deviation_reporting"},
495 context="record.analysis_plan",
496 )
497 entries = require_list(
498 plan["analyses"], "record.analysis_plan.analyses", minimum=1, maximum=200
499 )
500 parsed: list[dict[str, Any]] = []
501 identifiers: list[str] = []
502 fields = {
503 "analysis_id",
504 "prediction_ids",
505 "estimand_ids",
506 "analysis_population",
507 "method",
508 "effect_or_summary_measure",
509 "uncertainty_method",
510 "missing_data_plan",
511 "multiplicity_plan",
512 "sensitivity_analyses",
513 "decision_rule",
514 "exploratory_or_confirmatory",
515 }
516 for index, raw_entry in enumerate(entries):
517 context = f"record.analysis_plan.analyses[{index}]"
518 entry = require_object(raw_entry, context)
519 require_exact_keys(entry, required=fields, context=context)
520 identifier = require_identifier(entry["analysis_id"], f"{context}.analysis_id")
521 identifiers.append(identifier)
522 parsed.append(
523 {
524 "analysis_id": identifier,
525 "prediction_ids": require_identifier_list(
526 entry["prediction_ids"], f"{context}.prediction_ids", minimum=1
527 ),
528 "estimand_ids": require_identifier_list(
529 entry["estimand_ids"], f"{context}.estimand_ids"
530 ),
531 "analysis_population": require_text(
532 entry["analysis_population"],
533 f"{context}.analysis_population",
534 minimum=5,
535 ),
536 "method": require_text(entry["method"], f"{context}.method", minimum=5),
537 "effect_or_summary_measure": require_text(
538 entry["effect_or_summary_measure"],
539 f"{context}.effect_or_summary_measure",
540 minimum=3,
541 ),
542 "uncertainty_method": require_text(
543 entry["uncertainty_method"],
544 f"{context}.uncertainty_method",
545 minimum=5,
546 ),
547 "missing_data_plan": require_text(
548 entry["missing_data_plan"],
549 f"{context}.missing_data_plan",
550 minimum=5,
551 ),
552 "multiplicity_plan": require_text(
553 entry["multiplicity_plan"],
554 f"{context}.multiplicity_plan",
555 minimum=5,
556 ),
557 "sensitivity_analyses": require_text_list(
558 entry["sensitivity_analyses"],
559 f"{context}.sensitivity_analyses",
560 minimum=1,
561 ),
562 "decision_rule": require_text(
563 entry["decision_rule"], f"{context}.decision_rule", minimum=10
564 ),
565 "exploratory_or_confirmatory": require_enum(
566 entry["exploratory_or_confirmatory"],
567 PLAN_MODES,
568 f"{context}.exploratory_or_confirmatory",
569 ),
570 }
571 )
572 require_unique(identifiers, "record.analysis_plan.analyses")
573 return {
574 "analyses": parsed,
575 "harking_control": require_text(
576 plan["harking_control"],
577 "record.analysis_plan.harking_control",
578 minimum=10,
579 ),
580 "deviation_reporting": require_text(
581 plan["deviation_reporting"],
582 "record.analysis_plan.deviation_reporting",
583 minimum=10,
584 ),
585 }
588def _parse_evidence(raw: Any) -> dict[str, Any]:
589 value = require_object(raw, "record.evidence")
590 require_exact_keys(
591 value,
592 required={
593 "ledger_path",
594 "source_ids",
595 "search_boundary_id",
596 "evidence_limitations",
597 },
598 context="record.evidence",
599 )
600 ledger_path = require_text(
601 value["ledger_path"], "record.evidence.ledger_path", maximum=500
602 )
603 pure_path = PurePosixPath(ledger_path)
604 if pure_path.is_absolute() or ".." in pure_path.parts or "://" in ledger_path:
605 raise ValidationError("record.evidence.ledger_path must be a safe relative path")
606 if pure_path.suffix.lower() != ".csv":
607 raise ValidationError("record.evidence.ledger_path must end in .csv")
608 return {
609 "ledger_path": ledger_path,
610 "source_ids": require_identifier_list(
611 value["source_ids"], "record.evidence.source_ids", minimum=1
612 ),
613 "search_boundary_id": require_identifier(
614 value["search_boundary_id"], "record.evidence.search_boundary_id"
615 ),
616 "evidence_limitations": require_text_list(
617 value["evidence_limitations"],
618 "record.evidence.evidence_limitations",
619 minimum=1,
620 ),
621 }
624def _parse_risk_register(raw: Any) -> dict[str, list[str]]:
625 value = require_object(raw, "record.risk_register")
626 fields = {
627 "confounding",
628 "selection_bias",
629 "collider_bias",
630 "reverse_causation",
631 "measurement_bias",
632 "other",
633 }
634 require_exact_keys(value, required=fields, context="record.risk_register")
635 return {
636 field: require_text_list(
637 value[field], f"record.risk_register.{field}", maximum=50
638 )
639 for field in fields
640 }
643def _parse_ethics(raw: Any) -> dict[str, Any]:
644 value = require_object(raw, "record.ethics_and_feasibility")
645 gate_fields = {
646 "human_subjects_gate",
647 "animal_research_gate",
648 "biosafety_gate",
649 "dual_use_gate",
650 "regulatory_gate",
651 "data_governance_gate",
652 }
653 require_exact_keys(
654 value,
655 required=gate_fields
656 | {"feasibility_status", "required_reviews", "unresolved_blocks"},
657 context="record.ethics_and_feasibility",
658 )
659 parsed: dict[str, Any] = {
660 field: require_enum(
661 value[field], GATE_STATES, f"record.ethics_and_feasibility.{field}"
662 )
663 for field in gate_fields
664 }
665 parsed["feasibility_status"] = require_enum(
666 value["feasibility_status"],
667 FEASIBILITY_STATES,
668 "record.ethics_and_feasibility.feasibility_status",
669 )
670 parsed["required_reviews"] = require_text_list(
671 value["required_reviews"],
672 "record.ethics_and_feasibility.required_reviews",
673 maximum=50,
674 )
675 parsed["unresolved_blocks"] = require_text_list(
676 value["unresolved_blocks"],
677 "record.ethics_and_feasibility.unresolved_blocks",
678 maximum=50,
679 )
680 return parsed
683def _parse_ai_use(raw: Any) -> dict[str, Any]:
684 value = require_object(raw, "record.ai_use")
685 fields = {
686 "used",
687 "sensitive_or_unpublished_data_sent_externally",
688 "local_policy_checked",
689 "citation_verification_required",
690 "human_accountable",
691 "diversity_mitigation",
692 }
693 require_exact_keys(value, required=fields, context="record.ai_use")
694 return {
695 "used": require_bool(value["used"], "record.ai_use.used"),
696 "sensitive_or_unpublished_data_sent_externally": require_bool(
697 value["sensitive_or_unpublished_data_sent_externally"],
698 "record.ai_use.sensitive_or_unpublished_data_sent_externally",
699 ),
700 "local_policy_checked": require_bool(
701 value["local_policy_checked"], "record.ai_use.local_policy_checked"
702 ),
703 "citation_verification_required": require_bool(
704 value["citation_verification_required"],
705 "record.ai_use.citation_verification_required",
706 ),
707 "human_accountable": require_bool(
708 value["human_accountable"], "record.ai_use.human_accountable"
709 ),
710 "diversity_mitigation": require_text(
711 value["diversity_mitigation"],
712 "record.ai_use.diversity_mitigation",
713 minimum=10,
714 ),
715 }
718def load_hypothesis_record(payload: Any) -> dict[str, Any]:
719 """Parse the exact v2 record schema and normalize bounded values."""
720 root = require_object(payload, "record")
721 fields = {
722 "schema_version",
723 "project_id",
724 "status",
725 "updated_on",
726 "human_owner",
727 "observation",
728 "research_question",
729 "hypotheses",
730 "causal_estimands",
731 "predictions",
732 "alternative_explanations",
733 "null_hypotheses",
734 "negative_controls",
735 "operationalizations",
736 "analysis_plan",
737 "evidence",
738 "risk_register",
739 "ethics_and_feasibility",
740 "ai_use",
741 }
742 require_exact_keys(root, required=fields, context="record")
743 return {
744 "schema_version": require_enum(
745 root["schema_version"], {"2.0"}, "record.schema_version"
746 ),
747 "project_id": require_identifier(root["project_id"], "record.project_id"),
748 "status": require_enum(
749 root["status"], {"draft", "preregistered", "archived"}, "record.status"
750 ),
751 "updated_on": require_iso_date(root["updated_on"], "record.updated_on"),
752 "human_owner": require_text(
753 root["human_owner"], "record.human_owner", minimum=3
754 ),
755 "observation": _parse_observation(root["observation"]),
756 "research_question": _parse_question(root["research_question"]),
757 "hypotheses": _parse_hypotheses(root["hypotheses"]),
758 "causal_estimands": _parse_estimands(root["causal_estimands"]),
759 "predictions": _parse_predictions(root["predictions"]),
760 "alternative_explanations": _parse_alternatives(
761 root["alternative_explanations"]
762 ),
763 "null_hypotheses": _parse_nulls(root["null_hypotheses"]),
764 "negative_controls": _parse_controls(root["negative_controls"]),
765 "operationalizations": _parse_operationalizations(root["operationalizations"]),
766 "analysis_plan": _parse_analysis_plan(root["analysis_plan"]),
767 "evidence": _parse_evidence(root["evidence"]),
768 "risk_register": _parse_risk_register(root["risk_register"]),
769 "ethics_and_feasibility": _parse_ethics(root["ethics_and_feasibility"]),
770 "ai_use": _parse_ai_use(root["ai_use"]),
771 }
774def _missing_references(
775 values: list[str], valid_values: set[str], code: str, field: str
776) -> list[dict[str, str]]:
777 return [
778 issue(code, f"{field}:{value}")
779 for value in sorted(set(values) - valid_values)
780 ]
783def validate_record(record: dict[str, Any]) -> dict[str, Any]:
784 """Check cross-links and safety declarations without scoring candidates."""
785 errors: list[dict[str, str]] = []
786 warnings: list[dict[str, str]] = []
788 hypothesis_ids = {item["hypothesis_id"] for item in record["hypotheses"]}
789 estimand_ids = {item["estimand_id"] for item in record["causal_estimands"]}
790 prediction_ids = {item["prediction_id"] for item in record["predictions"]}
791 measurement_ids = {
792 item["measurement_id"] for item in record["operationalizations"]
793 }
794 analysis_ids = {
795 item["analysis_id"] for item in record["analysis_plan"]["analyses"]
796 }
797 control_ids = {item["control_id"] for item in record["negative_controls"]}
798 source_ids = set(record["evidence"]["source_ids"])
800 used_sources = set(record["observation"]["source_ids"])
801 for hypothesis in record["hypotheses"]:
802 used_sources.update(hypothesis["source_ids"])
803 errors.extend(
804 _missing_references(
805 hypothesis["prediction_ids"],
806 prediction_ids,
807 "UNKNOWN_PREDICTION_ID",
808 hypothesis["hypothesis_id"],
809 )
810 )
811 errors.extend(
812 _missing_references(
813 hypothesis["rival_hypothesis_ids"],
814 hypothesis_ids,
815 "UNKNOWN_RIVAL_HYPOTHESIS_ID",
816 hypothesis["hypothesis_id"],
817 )
818 )
819 if hypothesis["hypothesis_id"] in hypothesis["rival_hypothesis_ids"]:
820 errors.append(
821 issue("HYPOTHESIS_CANNOT_RIVAL_ITSELF", hypothesis["hypothesis_id"])
822 )
823 if len(hypothesis_ids) > 1 and not hypothesis["rival_hypothesis_ids"]:
824 warnings.append(
825 issue("RIVAL_HYPOTHESIS_LINK_MISSING", hypothesis["hypothesis_id"])
826 )
827 for operationalization in record["operationalizations"]:
828 used_sources.update(operationalization["validity_evidence_source_ids"])
829 errors.extend(
830 _missing_references(
831 sorted(used_sources),
832 source_ids,
833 "SOURCE_NOT_DECLARED_IN_EVIDENCE",
834 "record.evidence.source_ids",
835 )
836 )
838 if len(hypothesis_ids) == 1:
839 warnings.append(issue("SINGLE_CANDIDATE_REQUIRES_RIVAL_REVIEW", "hypotheses"))
841 for estimand in record["causal_estimands"]:
842 errors.extend(
843 _missing_references(
844 estimand["linked_hypothesis_ids"],
845 hypothesis_ids,
846 "UNKNOWN_HYPOTHESIS_ID",
847 estimand["estimand_id"],
848 )
849 )
850 if (
851 record["research_question"]["question_type"] == "causal"
852 and not record["causal_estimands"]
853 ):
854 errors.append(issue("CAUSAL_QUESTION_REQUIRES_ESTIMAND", "causal_estimands"))
856 predictions_by_hypothesis: Counter[str] = Counter()
857 for prediction in record["predictions"]:
858 prediction_id = prediction["prediction_id"]
859 predictions_by_hypothesis[prediction["hypothesis_id"]] += 1
860 errors.extend(
861 _missing_references(
862 [prediction["hypothesis_id"]],
863 hypothesis_ids,
864 "UNKNOWN_HYPOTHESIS_ID",
865 prediction_id,
866 )
867 )
868 errors.extend(
869 _missing_references(
870 prediction["rival_hypothesis_ids"],
871 hypothesis_ids,
872 "UNKNOWN_RIVAL_HYPOTHESIS_ID",
873 prediction_id,
874 )
875 )
876 errors.extend(
877 _missing_references(
878 prediction["measurement_ids"],
879 measurement_ids,
880 "UNKNOWN_MEASUREMENT_ID",
881 prediction_id,
882 )
883 )
884 errors.extend(
885 _missing_references(
886 prediction["analysis_ids"],
887 analysis_ids,
888 "UNKNOWN_ANALYSIS_ID",
889 prediction_id,
890 )
891 )
892 if prediction["hypothesis_id"] in prediction["rival_hypothesis_ids"]:
893 errors.append(issue("PREDICTION_RIVAL_IS_FOCAL", prediction_id))
894 for hypothesis_id in sorted(hypothesis_ids):
895 if predictions_by_hypothesis[hypothesis_id] == 0:
896 errors.append(issue("HYPOTHESIS_HAS_NO_PREDICTION", hypothesis_id))
898 for alternative in record["alternative_explanations"]:
899 errors.extend(
900 _missing_references(
901 alternative["linked_hypothesis_ids"],
902 hypothesis_ids,
903 "UNKNOWN_HYPOTHESIS_ID",
904 alternative["alternative_id"],
905 )
906 )
907 errors.extend(
908 _missing_references(
909 alternative["discriminating_prediction_ids"],
910 prediction_ids,
911 "UNKNOWN_PREDICTION_ID",
912 alternative["alternative_id"],
913 )
914 )
916 for null in record["null_hypotheses"]:
917 errors.extend(
918 _missing_references(
919 null["linked_prediction_ids"],
920 prediction_ids,
921 "UNKNOWN_PREDICTION_ID",
922 null["null_id"],
923 )
924 )
926 for control in record["negative_controls"]:
927 errors.extend(
928 _missing_references(
929 control["linked_prediction_ids"],
930 prediction_ids,
931 "UNKNOWN_PREDICTION_ID",
932 control["control_id"],
933 )
934 )
935 if not any(
936 control["control_type"]
937 in {"negative_exposure", "negative_outcome", "procedural_negative"}
938 for control in record["negative_controls"]
939 ):
940 warnings.append(issue("NEGATIVE_CONTROL_TYPE_REVIEW_REQUIRED", "controls"))
942 for analysis in record["analysis_plan"]["analyses"]:
943 errors.extend(
944 _missing_references(
945 analysis["prediction_ids"],
946 prediction_ids,
947 "UNKNOWN_PREDICTION_ID",
948 analysis["analysis_id"],
949 )
950 )
951 errors.extend(
952 _missing_references(
953 analysis["estimand_ids"],
954 estimand_ids,
955 "UNKNOWN_ESTIMAND_ID",
956 analysis["analysis_id"],
957 )
958 )
960 for category, entries in record["risk_register"].items():
961 if not entries:
962 warnings.append(issue("RISK_CATEGORY_EMPTY_REQUIRES_RATIONALE", category))
964 ethics = record["ethics_and_feasibility"]
965 gate_values = {
966 key: value for key, value in ethics.items() if key.endswith("_gate")
967 }
968 unresolved_gate_names = sorted(
969 key
970 for key, value in gate_values.items()
971 if value in {"undetermined", "requires_review", "blocked"}
972 )
973 if unresolved_gate_names and not ethics["unresolved_blocks"]:
974 errors.append(
975 issue("UNRESOLVED_GATE_REQUIRES_BLOCK_RECORD", "ethics_and_feasibility")
976 )
977 if (
978 any(value == "requires_review" for value in gate_values.values())
979 and not ethics["required_reviews"]
980 ):
981 errors.append(
982 issue("REQUIRED_REVIEW_LIST_MISSING", "ethics_and_feasibility")
983 )
984 if ethics["feasibility_status"] in {"undetermined", "infeasible", "blocked"}:
985 unresolved_gate_names.append("feasibility_status")
987 ai_use = record["ai_use"]
988 if ai_use["sensitive_or_unpublished_data_sent_externally"]:
989 errors.append(
990 issue(
991 "EXTERNAL_SENSITIVE_DATA_NOT_SUPPORTED",
992 "ai_use.sensitive_or_unpublished_data_sent_externally",
993 )
994 )
995 if not ai_use["local_policy_checked"]:
996 errors.append(issue("LOCAL_AI_POLICY_NOT_CHECKED", "ai_use.local_policy_checked"))
997 if not ai_use["citation_verification_required"]:
998 errors.append(
999 issue(
1000 "CITATION_VERIFICATION_MUST_BE_REQUIRED",
1001 "ai_use.citation_verification_required",
1002 )
1003 )
1004 if not ai_use["human_accountable"]:
1005 errors.append(
1006 issue("HUMAN_ACCOUNTABILITY_REQUIRED", "ai_use.human_accountable")
1007 )
1009 return {
1010 "schema_version": "2.0",
1011 "project_id": record["project_id"],
1012 "valid": not errors,
1013 "status": (
1014 "INVALID_RECORD"
1015 if errors
1016 else "VALID_BLOCKED_BY_GATES"
1017 if unresolved_gate_names
1018 else "VALID_FOR_HUMAN_REVIEW"
1019 ),
1020 "errors": errors,
1021 "warnings": warnings,
1022 "counts": {
1023 "hypotheses": len(hypothesis_ids),
1024 "causal_estimands": len(estimand_ids),
1025 "predictions": len(prediction_ids),
1026 "alternative_explanations": len(record["alternative_explanations"]),
1027 "null_hypotheses": len(record["null_hypotheses"]),
1028 "negative_controls": len(control_ids),
1029 "operationalizations": len(measurement_ids),
1030 "analyses": len(analysis_ids),
1031 "declared_sources": len(source_ids),
1032 },
1033 "identifiers": {
1034 "hypothesis_ids": sorted(hypothesis_ids),
1035 "estimand_ids": sorted(estimand_ids),
1036 "prediction_ids": sorted(prediction_ids),
1037 "measurement_ids": sorted(measurement_ids),
1038 "analysis_ids": sorted(analysis_ids),
1039 "control_ids": sorted(control_ids),
1040 },
1041 "unresolved_gate_fields": sorted(set(unresolved_gate_names)),
1042 "notice": (
1043 "This report validates schema, cross-references, and declared safety "
1044 "controls only. It does not verify evidence, scientific validity, "
1045 "novelty, ethics approval, causal identification, or hypothesis merit, "
1046 "and it does not rank or select candidates."
1047 ),
1048 }
1051def build_parser() -> argparse.ArgumentParser:
1052 parser = argparse.ArgumentParser(
1053 description=(
1054 "Validate a bounded local hypothesis JSON record and emit identifiers, "
1055 "counts, and rule codes without scientific scoring."
1056 )
1057 )
1058 parser.add_argument("record", help="Local hypothesis record JSON")
1059 parser.add_argument("-o", "--output", help="Optional local JSON report")
1060 parser.add_argument(
1061 "--force", action="store_true", help="Replace an existing output file"
1062 )
1063 return parser
1066def main() -> int:
1067 args = build_parser().parse_args()
1068 try:
1069 record = load_hypothesis_record(read_json(args.record))
1070 report = validate_record(record)
1071 write_json_report(report, args.output, force=args.force)
1072 return 0 if report["valid"] else 1
1073 except ValidationError as exc:
1074 return error_exit(exc)
1077if __name__ == "__main__":
1078 raise SystemExit(main())