Skills › Scientific Agent Skills › hypothesis-generation › audit_evidence_ledger.py
audit_evidence_ledger.py
337 lines · skills/hypothesis-generation/scripts/audit_evidence_ledger.py at 5f2bc03
1#!/usr/bin/env python3
2"""Audit a local evidence ledger and dated search boundary without networking."""
4from __future__ import annotations
6import argparse
7from collections import Counter
8from datetime import date
9from typing import Any
11from _common import (
12 ValidationError,
13 error_exit,
14 issue,
15 read_csv_records,
16 read_json,
17 require_enum,
18 require_exact_keys,
19 require_https_url,
20 require_identifier,
21 require_iso_date,
22 require_object,
23 require_partial_date,
24 require_text,
25 require_text_list,
26 require_unique,
27 split_identifiers,
28 write_json_report,
29)
30from validate_hypothesis_schema import load_hypothesis_record, validate_record
32FIELDS = (
33 "source_id",
34 "claim_ids",
35 "title",
36 "authors_or_organization",
37 "publication_date",
38 "source_type",
39 "identifier",
40 "url",
41 "accessed_on",
42 "relation",
43 "study_design_or_document_type",
44 "limitations",
45 "notes",
46)
47SOURCE_TYPES = {
48 "official_guidance",
49 "regulation_policy",
50 "reporting_guideline",
51 "primary_research",
52 "primary_method",
53 "systematic_review",
54 "consensus_report",
55 "book",
56 "dataset_or_registry",
57 "preprint",
58 "other",
59}
60RELATIONS = {"supportive", "challenging", "contextual", "method", "safety", "mixed"}
61NOVELTY_STATES = {
62 "not_assessed",
63 "requires_specialist_review",
64 "supported_by_documented_comprehensive_search",
65}
68def load_ledger(raw_path: str) -> list[dict[str, Any]]:
69 rows = read_csv_records(raw_path, fields=FIELDS)
70 parsed: list[dict[str, Any]] = []
71 source_ids: list[str] = []
72 for line_number, row in enumerate(rows, start=2):
73 context = f"ledger row {line_number}"
74 source_id = require_identifier(row["source_id"], f"{context}.source_id")
75 source_ids.append(source_id)
76 parsed.append(
77 {
78 "source_id": source_id,
79 "claim_ids": split_identifiers(
80 row["claim_ids"], f"{context}.claim_ids"
81 ),
82 "title": require_text(
83 row["title"], f"{context}.title", minimum=5, maximum=2_000
84 ),
85 "authors_or_organization": require_text(
86 row["authors_or_organization"],
87 f"{context}.authors_or_organization",
88 minimum=2,
89 maximum=1_000,
90 ),
91 "publication_date": require_partial_date(
92 row["publication_date"], f"{context}.publication_date"
93 ),
94 "source_type": require_enum(
95 row["source_type"], SOURCE_TYPES, f"{context}.source_type"
96 ),
97 "identifier": require_text(
98 row["identifier"],
99 f"{context}.identifier",
100 minimum=3,
101 maximum=1_000,
102 ),
103 "url": require_https_url(row["url"], f"{context}.url"),
104 "accessed_on": require_iso_date(
105 row["accessed_on"], f"{context}.accessed_on"
106 ),
107 "relation": require_enum(
108 row["relation"], RELATIONS, f"{context}.relation"
109 ),
110 "study_design_or_document_type": require_text(
111 row["study_design_or_document_type"],
112 f"{context}.study_design_or_document_type",
113 minimum=3,
114 maximum=1_000,
115 ),
116 "limitations": require_text(
117 row["limitations"],
118 f"{context}.limitations",
119 minimum=3,
120 maximum=2_000,
121 ),
122 "notes": require_text(
123 row["notes"],
124 f"{context}.notes",
125 allow_empty=True,
126 maximum=2_000,
127 ),
128 }
129 )
130 require_unique(source_ids, "evidence ledger")
131 return parsed
134def load_search_boundary(payload: Any) -> dict[str, Any]:
135 root = require_object(payload, "search_boundary")
136 fields = {
137 "schema_version",
138 "search_boundary_id",
139 "searched_on",
140 "searched_by",
141 "purpose",
142 "databases_or_indexes",
143 "queries",
144 "date_limits",
145 "language_limits",
146 "inclusion_scope",
147 "exclusion_scope",
148 "known_limitations",
149 "last_result_screened_or_stop_rule",
150 "novelty_status",
151 }
152 require_exact_keys(root, required=fields, context="search_boundary")
153 return {
154 "schema_version": require_enum(
155 root["schema_version"], {"2.0"}, "search_boundary.schema_version"
156 ),
157 "search_boundary_id": require_identifier(
158 root["search_boundary_id"], "search_boundary.search_boundary_id"
159 ),
160 "searched_on": require_iso_date(
161 root["searched_on"], "search_boundary.searched_on"
162 ),
163 "searched_by": require_text(
164 root["searched_by"], "search_boundary.searched_by", minimum=3
165 ),
166 "purpose": require_text(
167 root["purpose"], "search_boundary.purpose", minimum=10
168 ),
169 "databases_or_indexes": require_text_list(
170 root["databases_or_indexes"],
171 "search_boundary.databases_or_indexes",
172 minimum=1,
173 maximum=100,
174 ),
175 "queries": require_text_list(
176 root["queries"], "search_boundary.queries", minimum=1, maximum=200
177 ),
178 "date_limits": require_text(
179 root["date_limits"], "search_boundary.date_limits", minimum=3
180 ),
181 "language_limits": require_text(
182 root["language_limits"], "search_boundary.language_limits", minimum=3
183 ),
184 "inclusion_scope": require_text(
185 root["inclusion_scope"], "search_boundary.inclusion_scope", minimum=10
186 ),
187 "exclusion_scope": require_text(
188 root["exclusion_scope"], "search_boundary.exclusion_scope", minimum=10
189 ),
190 "known_limitations": require_text_list(
191 root["known_limitations"],
192 "search_boundary.known_limitations",
193 minimum=1,
194 maximum=100,
195 ),
196 "last_result_screened_or_stop_rule": require_text(
197 root["last_result_screened_or_stop_rule"],
198 "search_boundary.last_result_screened_or_stop_rule",
199 minimum=10,
200 ),
201 "novelty_status": require_enum(
202 root["novelty_status"],
203 NOVELTY_STATES,
204 "search_boundary.novelty_status",
205 ),
206 }
209def _record_claim_ids(record: dict[str, Any]) -> set[str]:
210 identifiers = {"OBS1", record["project_id"]}
211 identifiers.update(item["hypothesis_id"] for item in record["hypotheses"])
212 identifiers.update(item["estimand_id"] for item in record["causal_estimands"])
213 identifiers.update(item["prediction_id"] for item in record["predictions"])
214 identifiers.update(
215 item["alternative_id"] for item in record["alternative_explanations"]
216 )
217 identifiers.update(item["null_id"] for item in record["null_hypotheses"])
218 identifiers.update(item["control_id"] for item in record["negative_controls"])
219 identifiers.update(
220 item["measurement_id"] for item in record["operationalizations"]
221 )
222 identifiers.update(
223 item["analysis_id"] for item in record["analysis_plan"]["analyses"]
224 )
225 return identifiers
228def audit(
229 ledger: list[dict[str, Any]],
230 boundary: dict[str, Any],
231 record: dict[str, Any] | None = None,
232) -> dict[str, Any]:
233 errors: list[dict[str, str]] = []
234 warnings: list[dict[str, str]] = []
235 source_type_counts: Counter[str] = Counter()
236 relation_counts: Counter[str] = Counter()
237 ledger_source_ids = {row["source_id"] for row in ledger}
238 linked_claim_ids: set[str] = set()
240 searched_on = date.fromisoformat(boundary["searched_on"])
241 for row in ledger:
242 source_type_counts[row["source_type"]] += 1
243 relation_counts[row["relation"]] += 1
244 linked_claim_ids.update(row["claim_ids"])
245 if date.fromisoformat(row["accessed_on"]) > searched_on:
246 warnings.append(
247 issue("SOURCE_ACCESSED_AFTER_SEARCH_BOUNDARY", row["source_id"])
248 )
250 if not relation_counts["challenging"]:
251 warnings.append(issue("NO_CHALLENGING_SOURCE_DECLARED", "ledger"))
252 if boundary["novelty_status"] == "supported_by_documented_comprehensive_search":
253 warnings.append(
254 issue("NOVELTY_STATUS_REQUIRES_QUALIFIED_HUMAN_REVIEW", "search_boundary")
255 )
257 record_cross_check = record is not None
258 if record is not None:
259 record_report = validate_record(record)
260 if not record_report["valid"]:
261 raise ValidationError(
262 "optional hypothesis record must pass schema validation first"
263 )
264 if record["evidence"]["search_boundary_id"] != boundary["search_boundary_id"]:
265 errors.append(
266 issue("SEARCH_BOUNDARY_ID_MISMATCH", boundary["search_boundary_id"])
267 )
268 record_source_ids = set(record["evidence"]["source_ids"])
269 for source_id in sorted(record_source_ids - ledger_source_ids):
270 errors.append(issue("RECORD_SOURCE_MISSING_FROM_LEDGER", source_id))
271 for source_id in sorted(ledger_source_ids - record_source_ids):
272 warnings.append(issue("LEDGER_SOURCE_NOT_DECLARED_IN_RECORD", source_id))
273 allowed_claim_ids = _record_claim_ids(record)
274 for claim_id in sorted(linked_claim_ids - allowed_claim_ids):
275 errors.append(issue("UNKNOWN_CLAIM_ID", claim_id))
277 return {
278 "schema_version": "2.0",
279 "search_boundary_id": boundary["search_boundary_id"],
280 "searched_on": boundary["searched_on"],
281 "novelty_status": boundary["novelty_status"],
282 "valid": not errors,
283 "status": "INVALID_LEDGER" if errors else "VALID_FOR_HUMAN_SOURCE_REVIEW",
284 "errors": errors,
285 "warnings": warnings,
286 "source_count": len(ledger),
287 "source_type_counts": dict(sorted(source_type_counts.items())),
288 "relation_counts": dict(sorted(relation_counts.items())),
289 "source_ids": sorted(ledger_source_ids),
290 "linked_claim_ids": sorted(linked_claim_ids),
291 "record_cross_check_performed": record_cross_check,
292 "notice": (
293 "This local audit does not visit URLs, verify source existence or "
294 "content, appraise evidence, establish novelty, or determine whether "
295 "a source supports a claim. Verify every source and link manually."
296 ),
297 }
300def build_parser() -> argparse.ArgumentParser:
301 parser = argparse.ArgumentParser(
302 description=(
303 "Audit a bounded local evidence CSV and search-boundary JSON without "
304 "network access or scientific scoring."
305 )
306 )
307 parser.add_argument("ledger", help="Local evidence ledger CSV")
308 parser.add_argument("search_boundary", help="Local search-boundary JSON")
309 parser.add_argument(
310 "--record", help="Optional local hypothesis record JSON for cross-checks"
311 )
312 parser.add_argument("-o", "--output", help="Optional local JSON report")
313 parser.add_argument(
314 "--force", action="store_true", help="Replace an existing output file"
315 )
316 return parser
319def main() -> int:
320 args = build_parser().parse_args()
321 try:
322 record = (
323 load_hypothesis_record(read_json(args.record)) if args.record else None
324 )
325 report = audit(
326 load_ledger(args.ledger),
327 load_search_boundary(read_json(args.search_boundary)),
328 record,
329 )
330 write_json_report(report, args.output, force=args.force)
331 return 0 if report["valid"] else 1
332 except ValidationError as exc:
333 return error_exit(exc)
336if __name__ == "__main__":
337 raise SystemExit(main())