Skills › Scientific Agent Skills › scientific-writing › audit_claims.py
audit_claims.py
272 lines · skills/scientific-writing/scripts/audit_claims.py at 5f2bc03
1"""Audit claim-to-evidence mappings and local citation markers."""
3from __future__ import annotations
5import argparse
6import hashlib
7import re
8from typing import Any
10from _common import (
11 InputError,
12 Issue,
13 emit_report,
14 is_nonempty_string,
15 issue,
16 read_csv,
17 read_json,
18 read_text,
19 require_list,
20 require_object,
21 run,
22)
23from validate_manifest import validate_source_manifest
25TOOL = "audit_claims"
26REQUIRED_FIELDS = {
27 "claim_id",
28 "section",
29 "claim_kind",
30 "claim_text_sha256",
31 "evidence_ids",
32 "verification_status",
33 "uncertainty",
34 "analysis_intent",
35}
36CLAIM_ID_RE = re.compile(r"^C[0-9]{3,8}$")
37EVIDENCE_ID_RE = re.compile(r"^E[0-9]{3,8}$")
38SHA256_RE = re.compile(r"^[a-f0-9]{64}$")
39CLAIM_MARKER_RE = re.compile(r"\[claim:(C[0-9]{3,8})\]")
40EVIDENCE_MARKER_RE = re.compile(
41 r"\[evidence:((?:E[0-9]{3,8})(?:\s*,\s*E[0-9]{3,8})*)\]"
42)
43CITATION_MARKER_RE = re.compile(r"\[@(E[0-9]{3,8})\]")
44NUMERIC_RE = re.compile(r"(?<![A-Za-z])(?:[<>]=?\s*)?[0-9]+(?:\.[0-9]+)?%?")
45CLAIM_KINDS = {"factual", "numeric", "method", "result", "interpretive", "declaration"}
46UNCERTAINTY = {"not_applicable", "not_estimated", "low", "moderate", "high"}
47ANALYSIS_INTENT = {"confirmatory", "exploratory", "descriptive", "not_applicable"}
50def _split_evidence_ids(value: str) -> list[str]:
51 return [part.strip() for part in value.split(";") if part.strip()]
54def load_sources(path: str) -> dict[str, bool]:
55 data = require_object(read_json(path), "source_manifest")
56 if any(
57 item.severity == "error"
58 for item in validate_source_manifest(data, require_verified=False)
59 ):
60 raise InputError("source manifest failed validation")
61 sources: dict[str, bool] = {}
62 for index, raw_source in enumerate(require_list(data.get("sources"), "sources")):
63 source = require_object(raw_source, f"sources[{index}]")
64 evidence_id = source.get("evidence_id")
65 if not isinstance(evidence_id, str) or not EVIDENCE_ID_RE.fullmatch(
66 evidence_id
67 ):
68 raise InputError("source manifest contains an invalid evidence_id")
69 if evidence_id in sources:
70 raise InputError("source manifest contains duplicate evidence_id values")
71 verification = require_object(source.get("verification"), "verification")
72 sources[evidence_id] = (
73 verification.get("status") == "verified"
74 and verification.get("source_opened") is True
75 )
76 return sources
79def claim_text_hash(line: str) -> str:
80 """Hash one physical claim line, excluding markers and normalizing whitespace."""
81 text = CLAIM_MARKER_RE.sub("", line)
82 text = EVIDENCE_MARKER_RE.sub("", text)
83 text = CITATION_MARKER_RE.sub("", text)
84 normalized = " ".join(text.split())
85 return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
88def load_claims(
89 path: str,
90 sources: dict[str, bool],
91) -> tuple[dict[str, dict[str, Any]], list[Issue]]:
92 fields, rows = read_csv(path)
93 if set(fields) != REQUIRED_FIELDS:
94 missing = ",".join(sorted(REQUIRED_FIELDS - set(fields)))
95 extra = ",".join(sorted(set(fields) - REQUIRED_FIELDS))
96 raise InputError(
97 f"claim registry headers must match the schema; missing={missing}; extra={extra}"
98 )
99 issues: list[Issue] = []
100 claims: dict[str, dict[str, Any]] = {}
101 for row_number, row in enumerate(rows, start=2):
102 location = f"row:{row_number}"
103 claim_id = row["claim_id"].strip()
104 if not CLAIM_ID_RE.fullmatch(claim_id):
105 issues.append(issue("error", "INVALID_CLAIM_ID", location=location))
106 continue
107 if claim_id in claims:
108 issues.append(issue("error", "DUPLICATE_CLAIM_ID", item_id=claim_id))
109 continue
110 claims[claim_id] = row
111 if row["claim_kind"] not in CLAIM_KINDS:
112 issues.append(issue("error", "INVALID_CLAIM_KIND", item_id=claim_id))
113 if not is_nonempty_string(row["section"]):
114 issues.append(issue("error", "MISSING_CLAIM_SECTION", item_id=claim_id))
115 if not SHA256_RE.fullmatch(row["claim_text_sha256"]):
116 issues.append(issue("error", "INVALID_CLAIM_TEXT_HASH", item_id=claim_id))
117 if row["verification_status"] != "verified":
118 issues.append(issue("error", "CLAIM_NOT_VERIFIED", item_id=claim_id))
119 if row["uncertainty"] not in UNCERTAINTY:
120 issues.append(
121 issue("error", "INVALID_UNCERTAINTY_STATUS", item_id=claim_id)
122 )
123 if row["analysis_intent"] not in ANALYSIS_INTENT:
124 issues.append(issue("error", "INVALID_ANALYSIS_INTENT", item_id=claim_id))
126 evidence_ids = _split_evidence_ids(row["evidence_ids"])
127 row["_evidence_ids"] = evidence_ids
128 if not evidence_ids:
129 issues.append(issue("error", "CLAIM_WITHOUT_EVIDENCE", item_id=claim_id))
130 for evidence_id in evidence_ids:
131 if not EVIDENCE_ID_RE.fullmatch(evidence_id):
132 issues.append(
133 issue("error", "INVALID_CLAIM_EVIDENCE_ID", item_id=claim_id)
134 )
135 elif evidence_id not in sources:
136 issues.append(
137 issue("error", "UNKNOWN_CLAIM_EVIDENCE", item_id=claim_id)
138 )
139 elif not sources[evidence_id]:
140 issues.append(
141 issue("error", "UNVERIFIED_CLAIM_EVIDENCE", item_id=claim_id)
142 )
143 return claims, issues
146def audit_markdown(
147 text: str,
148 claims: dict[str, dict[str, Any]],
149 sources: dict[str, bool],
150) -> tuple[list[Issue], set[str]]:
151 issues: list[Issue] = []
152 used_claims: set[str] = set()
153 in_fence = False
154 for line_number, line in enumerate(text.splitlines(), start=1):
155 stripped = line.strip()
156 if stripped.startswith(("```", "~~~")):
157 in_fence = not in_fence
158 continue
159 if in_fence or not stripped or stripped.startswith(("#", "<!--")):
160 continue
162 line_claims = CLAIM_MARKER_RE.findall(line)
163 if len(line_claims) > 1:
164 issues.append(
165 issue("error", "MULTIPLE_CLAIMS_ON_LINE", location=f"line:{line_number}")
166 )
167 line_evidence: set[str] = set(CITATION_MARKER_RE.findall(line))
168 for group in EVIDENCE_MARKER_RE.findall(line):
169 line_evidence.update(part.strip() for part in group.split(","))
171 for evidence_id in sorted(line_evidence):
172 if evidence_id not in sources:
173 issues.append(
174 issue(
175 "error",
176 "UNKNOWN_CITATION_MARKER",
177 location=f"line:{line_number}",
178 item_id=evidence_id,
179 )
180 )
181 elif not sources[evidence_id]:
182 issues.append(
183 issue(
184 "error",
185 "UNVERIFIED_CITATION_MARKER",
186 location=f"line:{line_number}",
187 item_id=evidence_id,
188 )
189 )
191 for claim_id in line_claims:
192 used_claims.add(claim_id)
193 if claim_id not in claims:
194 issues.append(
195 issue(
196 "error",
197 "UNKNOWN_CLAIM_MARKER",
198 location=f"line:{line_number}",
199 item_id=claim_id,
200 )
201 )
202 continue
203 expected = set(claims[claim_id].get("_evidence_ids", []))
204 if len(line_claims) == 1 and claim_text_hash(line) != claims[claim_id].get(
205 "claim_text_sha256"
206 ):
207 issues.append(
208 issue(
209 "error",
210 "CLAIM_TEXT_HASH_MISMATCH",
211 location=f"line:{line_number}",
212 item_id=claim_id,
213 )
214 )
215 if not expected.issubset(line_evidence):
216 issues.append(
217 issue(
218 "error",
219 "CLAIM_MARKER_MISSING_EVIDENCE_MARKER",
220 location=f"line:{line_number}",
221 item_id=claim_id,
222 )
223 )
225 if NUMERIC_RE.search(line) and not line_claims:
226 issues.append(
227 issue(
228 "error",
229 "UNTAGGED_NUMERIC_CONTENT",
230 location=f"line:{line_number}",
231 )
232 )
233 return issues, used_claims
236def build_parser() -> argparse.ArgumentParser:
237 parser = argparse.ArgumentParser(
238 description=(
239 "Audit local Markdown claim markers against a CSV claim registry and "
240 "verified JSON source manifest. Output never includes manuscript text."
241 )
242 )
243 parser.add_argument("manuscript", help="UTF-8 Markdown manuscript")
244 parser.add_argument("claims", help="UTF-8 CSV claim-evidence registry")
245 parser.add_argument("sources", help="UTF-8 JSON source manifest")
246 return parser
249def cli() -> int:
250 args = build_parser().parse_args()
251 sources = load_sources(args.sources)
252 claims, issues = load_claims(args.claims, sources)
253 text = read_text(args.manuscript, {".md", ".markdown"})
254 markdown_issues, used_claims = audit_markdown(text, claims, sources)
255 issues.extend(markdown_issues)
256 for claim_id in sorted(set(claims) - used_claims):
257 issues.append(
258 issue("warning", "CLAIM_NOT_USED_IN_MANUSCRIPT", item_id=claim_id)
259 )
260 return emit_report(
261 TOOL,
262 issues,
263 summary={
264 "claims_registered": len(claims),
265 "claims_used": len(used_claims),
266 "sources_registered": len(sources),
267 },
268 )
271if __name__ == "__main__":
272 run(TOOL, cli)