Skills › Scientific Agent Skills › scientific-brainstorming › validate_register.py
validate_register.py
636 lines · skills/scientific-brainstorming/scripts/validate_register.py at 5f2bc03
1"""Validate brainstorming idea and assumption registers structurally."""
3from __future__ import annotations
5import argparse
6import re
7import sys
8import unicodedata
9from collections import defaultdict
10from collections.abc import Sequence
11from typing import Any
13from _common import (
14 MAX_ITEMS,
15 CliError,
16 emit_json,
17 read_json,
18 require_identifier,
19 require_iso_date,
20 require_text,
21)
23ORIGINS = {"human", "ai-assisted", "literature-inspired", "mixed", "other"}
24RECORDED_STAGES = {"independent", "discussion", "post-check"}
25EVIDENCE_STATUSES = {
26 "not-checked",
27 "search-incomplete",
28 "support-located",
29 "challenge-located",
30 "mixed",
31 "no-direct-evidence-located",
32}
33IDEA_STATUSES = {
34 "candidate",
35 "revised",
36 "deferred",
37 "paused",
38 "stopped",
39 "advanced-to-review",
40}
41ASSUMPTION_CATEGORIES = {
42 "causal",
43 "mechanistic",
44 "measurement",
45 "sampling",
46 "operational",
47 "statistical",
48 "feasibility",
49 "ethical",
50 "value",
51 "other",
52}
53ASSUMPTION_STATUSES = {
54 "untested",
55 "partially-supported",
56 "challenged",
57 "refuted",
58 "not-applicable",
59}
62def _issue(path: str, message: str) -> dict[str, str]:
63 return {"path": path, "message": message}
66def _normalized_statement(value: str) -> str:
67 """Normalize only exact text features; this is not semantic matching."""
69 normalized = unicodedata.normalize("NFKC", value).casefold()
70 return re.sub(r"\s+", " ", normalized).strip()
73def _list_field(
74 record: dict[str, Any],
75 field: str,
76 path: str,
77 errors: list[dict[str, str]],
78 *,
79 required: bool = True,
80) -> list[Any]:
81 value = record.get(field)
82 if value is None and not required:
83 return []
84 if not isinstance(value, list):
85 errors.append(_issue(f"{path}.{field}", "must be a list"))
86 return []
87 if len(value) > MAX_ITEMS:
88 errors.append(
89 _issue(
90 f"{path}.{field}",
91 f"has {len(value)} items; limit is {MAX_ITEMS}",
92 )
93 )
94 return value[:MAX_ITEMS]
95 return value
98def _check_text(
99 value: Any,
100 path: str,
101 errors: list[dict[str, str]],
102 *,
103 maximum: int = 10_000,
104 allow_empty: bool = False,
105) -> str | None:
106 try:
107 return require_text(
108 value,
109 path,
110 maximum=maximum,
111 allow_empty=allow_empty,
112 )
113 except CliError as exc:
114 errors.append(_issue(path, str(exc)))
115 return None
118def _check_id(
119 value: Any,
120 path: str,
121 errors: list[dict[str, str]],
122) -> str | None:
123 try:
124 return require_identifier(value, path)
125 except CliError as exc:
126 errors.append(_issue(path, str(exc)))
127 return None
130def _check_choice(
131 value: Any,
132 choices: set[str],
133 path: str,
134 errors: list[dict[str, str]],
135) -> str | None:
136 """Report malformed enums, including JSON arrays/objects, without crashing."""
138 if not isinstance(value, str) or value not in choices:
139 errors.append(_issue(path, f"must be one of {sorted(choices)}"))
140 return None
141 return value
144def _check_string_list(
145 values: list[Any],
146 path: str,
147 errors: list[dict[str, str]],
148 *,
149 identifiers: bool = False,
150) -> list[str]:
151 result: list[str] = []
152 for index, value in enumerate(values):
153 item_path = f"{path}[{index}]"
154 checked = (
155 _check_id(value, item_path, errors)
156 if identifiers
157 else _check_text(value, item_path, errors, maximum=2_000)
158 )
159 if checked is not None:
160 result.append(checked)
161 return result
164def validate_register(document: Any) -> dict[str, Any]:
165 """Return a deterministic structural validation report."""
167 errors: list[dict[str, str]] = []
168 warnings: list[dict[str, str]] = []
170 if not isinstance(document, dict):
171 return {
172 "valid": False,
173 "errors": [_issue("$", "top-level JSON value must be an object")],
174 "warnings": [],
175 "statistics": {},
176 "limitations": [
177 (
178 "Structural validation cannot establish truth, novelty, "
179 "feasibility, ethics approval, or scientific validity."
180 )
181 ],
182 }
184 schema_version = document.get("schema_version")
185 if schema_version != "1.1":
186 warnings.append(
187 _issue(
188 "$.schema_version",
189 "expected schema version '1.1'; unknown fields are not rejected",
190 )
191 )
193 session = document.get("session")
194 participant_ids: set[str] = set()
195 if not isinstance(session, dict):
196 errors.append(_issue("$.session", "must be an object"))
197 else:
198 _check_id(session.get("id"), "$.session.id", errors)
199 _check_text(session.get("title"), "$.session.title", errors, maximum=200)
200 _check_text(
201 session.get("question"),
202 "$.session.question",
203 errors,
204 maximum=2_000,
205 )
206 raw_participants = _list_field(
207 session,
208 "participant_ids",
209 "$.session",
210 errors,
211 )
212 participant_list = _check_string_list(
213 raw_participants,
214 "$.session.participant_ids",
215 errors,
216 identifiers=True,
217 )
218 if not participant_list:
219 errors.append(
220 _issue(
221 "$.session.participant_ids",
222 "must contain at least one participant ID",
223 )
224 )
225 if len(set(participant_list)) != len(participant_list):
226 errors.append(
227 _issue("$.session.participant_ids", "participant IDs must be unique")
228 )
229 participant_ids = set(participant_list)
230 if session.get("date") is not None:
231 try:
232 require_iso_date(session["date"], "$.session.date")
233 except CliError as exc:
234 errors.append(_issue("$.session.date", str(exc)))
236 assumptions_raw = document.get("assumptions")
237 if not isinstance(assumptions_raw, list):
238 errors.append(_issue("$.assumptions", "must be a list"))
239 assumptions_raw = []
240 elif len(assumptions_raw) > MAX_ITEMS:
241 errors.append(
242 _issue(
243 "$.assumptions",
244 f"has {len(assumptions_raw)} items; limit is {MAX_ITEMS}",
245 )
246 )
247 assumptions_raw = assumptions_raw[:MAX_ITEMS]
249 assumption_ids: set[str] = set()
250 assumption_references: set[str] = set()
251 for index, assumption in enumerate(assumptions_raw):
252 path = f"$.assumptions[{index}]"
253 if not isinstance(assumption, dict):
254 errors.append(_issue(path, "must be an object"))
255 continue
256 assumption_id = _check_id(assumption.get("id"), f"{path}.id", errors)
257 if assumption_id is not None:
258 if assumption_id in assumption_ids:
259 errors.append(_issue(f"{path}.id", "duplicate assumption ID"))
260 assumption_ids.add(assumption_id)
261 _check_text(assumption.get("statement"), f"{path}.statement", errors)
262 _check_choice(assumption.get("category"), ASSUMPTION_CATEGORIES,
263 f"{path}.category", errors)
264 status = _check_choice(assumption.get("status"), ASSUMPTION_STATUSES,
265 f"{path}.status", errors)
266 _check_text(
267 assumption.get("test_or_check"),
268 f"{path}.test_or_check",
269 errors,
270 maximum=2_000,
271 )
272 owner = _check_id(assumption.get("owner_id"), f"{path}.owner_id", errors)
273 if owner is not None and participant_ids and owner not in participant_ids:
274 errors.append(_issue(f"{path}.owner_id", "is not listed as a participant"))
275 evidence_refs = _list_field(
276 assumption,
277 "evidence_refs",
278 path,
279 errors,
280 )
281 clean_evidence_refs = _check_string_list(
282 evidence_refs,
283 f"{path}.evidence_refs",
284 errors,
285 )
286 if (
287 status not in {None, "untested", "not-applicable"}
288 and not clean_evidence_refs
289 ):
290 warnings.append(
291 _issue(
292 f"{path}.evidence_refs",
293 "non-untested status has no recorded evidence reference",
294 )
295 )
297 ideas_raw = document.get("ideas")
298 if not isinstance(ideas_raw, list):
299 errors.append(_issue("$.ideas", "must be a list"))
300 ideas_raw = []
301 elif len(ideas_raw) > MAX_ITEMS:
302 errors.append(
303 _issue(
304 "$.ideas",
305 f"has {len(ideas_raw)} items; limit is {MAX_ITEMS}",
306 )
307 )
308 ideas_raw = ideas_raw[:MAX_ITEMS]
310 idea_ids: set[str] = set()
311 statements: defaultdict[str, list[str]] = defaultdict(list)
312 ai_assisted_count = 0
313 for index, idea in enumerate(ideas_raw):
314 path = f"$.ideas[{index}]"
315 if not isinstance(idea, dict):
316 errors.append(_issue(path, "must be an object"))
317 continue
318 idea_id = _check_id(idea.get("id"), f"{path}.id", errors)
319 if idea_id is not None:
320 if idea_id in idea_ids:
321 errors.append(_issue(f"{path}.id", "duplicate idea ID"))
322 idea_ids.add(idea_id)
323 statement = _check_text(idea.get("statement"), f"{path}.statement", errors)
324 if statement is not None and idea_id is not None:
325 statements[_normalized_statement(statement)].append(idea_id)
327 provenance = idea.get("provenance")
328 if not isinstance(provenance, dict):
329 errors.append(_issue(f"{path}.provenance", "must be an object"))
330 provenance = {}
331 origin = _check_choice(provenance.get("origin"), ORIGINS,
332 f"{path}.provenance.origin", errors)
333 if origin in {"ai-assisted", "mixed"}:
334 ai_assisted_count += 1
335 _check_choice(provenance.get("recorded_stage"), RECORDED_STAGES,
336 f"{path}.provenance.recorded_stage", errors)
337 contributors_raw = _list_field(
338 provenance,
339 "contributor_ids",
340 f"{path}.provenance",
341 errors,
342 )
343 contributors = _check_string_list(
344 contributors_raw,
345 f"{path}.provenance.contributor_ids",
346 errors,
347 identifiers=True,
348 )
349 if not contributors:
350 errors.append(
351 _issue(
352 f"{path}.provenance.contributor_ids",
353 "must contain at least one contributor ID",
354 )
355 )
356 for contributor in contributors:
357 if participant_ids and contributor not in participant_ids:
358 errors.append(
359 _issue(
360 f"{path}.provenance.contributor_ids",
361 f"{contributor!r} is not listed as a participant",
362 )
363 )
364 source_refs_raw = _list_field(
365 provenance,
366 "source_refs",
367 f"{path}.provenance",
368 errors,
369 )
370 source_refs = _check_string_list(
371 source_refs_raw,
372 f"{path}.provenance.source_refs",
373 errors,
374 )
375 ai_tool = provenance.get("ai_tool")
376 if origin in {"ai-assisted", "mixed"}:
377 _check_text(
378 ai_tool,
379 f"{path}.provenance.ai_tool",
380 errors,
381 maximum=500,
382 )
383 elif ai_tool is not None and ai_tool != "":
384 warnings.append(
385 _issue(
386 f"{path}.provenance.ai_tool",
387 "AI tool is recorded but origin is not AI-assisted or mixed",
388 )
389 )
390 if origin == "literature-inspired" and not source_refs:
391 errors.append(
392 _issue(
393 f"{path}.provenance.source_refs",
394 "literature-inspired ideas require at least one source reference",
395 )
396 )
398 linked_assumptions_raw = _list_field(
399 idea,
400 "assumption_ids",
401 path,
402 errors,
403 )
404 linked_assumptions = _check_string_list(
405 linked_assumptions_raw,
406 f"{path}.assumption_ids",
407 errors,
408 identifiers=True,
409 )
410 assumption_references.update(linked_assumptions)
411 if not linked_assumptions:
412 warnings.append(
413 _issue(
414 f"{path}.assumption_ids",
415 "idea has no explicit linked assumptions",
416 )
417 )
419 predictions_raw = _list_field(
420 idea,
421 "predicted_observations",
422 path,
423 errors,
424 )
425 _check_string_list(
426 predictions_raw,
427 f"{path}.predicted_observations",
428 errors,
429 )
430 uncertainties_raw = _list_field(
431 idea,
432 "uncertainties",
433 path,
434 errors,
435 )
436 uncertainties = _check_string_list(
437 uncertainties_raw,
438 f"{path}.uncertainties",
439 errors,
440 )
441 if not uncertainties:
442 warnings.append(
443 _issue(
444 f"{path}.uncertainties",
445 "idea has no recorded uncertainty",
446 )
447 )
448 _check_choice(idea.get("evidence_status"), EVIDENCE_STATUSES,
449 f"{path}.evidence_status", errors)
450 _check_choice(idea.get("status"), IDEA_STATUSES, f"{path}.status", errors)
452 for normalized, duplicate_ids in sorted(statements.items()):
453 if normalized and len(duplicate_ids) > 1:
454 warnings.append(
455 _issue(
456 "$.ideas",
457 "exact-after-case/whitespace/Unicode normalization duplicate "
458 f"text for IDs {sorted(duplicate_ids)}; this is not a claim "
459 "of semantic equivalence",
460 )
461 )
463 for missing_id in sorted(assumption_references - assumption_ids):
464 errors.append(
465 _issue(
466 "$.ideas[*].assumption_ids",
467 f"references missing assumption ID {missing_id!r}",
468 )
469 )
470 for unlinked_id in sorted(assumption_ids - assumption_references):
471 warnings.append(
472 _issue(
473 "$.assumptions",
474 f"assumption {unlinked_id!r} is not linked from any idea",
475 )
476 )
478 clusters = document.get("clusters", [])
479 if not isinstance(clusters, list):
480 errors.append(_issue("$.clusters", "must be a list when present"))
481 elif len(clusters) > MAX_ITEMS:
482 errors.append(_issue("$.clusters", f"limit is {MAX_ITEMS}"))
483 else:
484 cluster_ids: set[str] = set()
485 for index, cluster in enumerate(clusters):
486 path = f"$.clusters[{index}]"
487 if not isinstance(cluster, dict):
488 errors.append(_issue(path, "must be an object"))
489 continue
490 cluster_id = _check_id(cluster.get("id"), f"{path}.id", errors)
491 if cluster_id is not None:
492 if cluster_id in cluster_ids:
493 errors.append(_issue(f"{path}.id", "duplicate cluster ID"))
494 cluster_ids.add(cluster_id)
495 _check_text(cluster.get("relation"), f"{path}.relation", errors)
496 members = _check_string_list(
497 _list_field(cluster, "idea_ids", path, errors),
498 f"{path}.idea_ids",
499 errors,
500 identifiers=True,
501 )
502 for member in members:
503 if member not in idea_ids:
504 errors.append(
505 _issue(
506 f"{path}.idea_ids",
507 f"references missing idea ID {member!r}",
508 )
509 )
511 decisions = document.get("decision_log", [])
512 if not isinstance(decisions, list):
513 errors.append(_issue("$.decision_log", "must be a list when present"))
514 elif len(decisions) > MAX_ITEMS:
515 errors.append(_issue("$.decision_log", f"limit is {MAX_ITEMS}"))
516 else:
517 decision_ids: set[str] = set()
518 for index, decision in enumerate(decisions):
519 path = f"$.decision_log[{index}]"
520 if not isinstance(decision, dict):
521 errors.append(_issue(path, "must be an object"))
522 continue
523 decision_id = _check_id(
524 decision.get("decision_id"), f"{path}.decision_id", errors
525 )
526 if decision_id is not None:
527 if decision_id in decision_ids:
528 errors.append(
529 _issue(f"{path}.decision_id", "duplicate decision ID")
530 )
531 decision_ids.add(decision_id)
532 if decision.get("date") is not None:
533 try:
534 require_iso_date(decision["date"], f"{path}.date")
535 except CliError as exc:
536 errors.append(_issue(f"{path}.date", str(exc)))
537 candidates = _check_string_list(
538 _list_field(decision, "candidate_ids", path, errors),
539 f"{path}.candidate_ids",
540 errors,
541 identifiers=True,
542 )
543 for candidate in candidates:
544 if candidate not in idea_ids:
545 errors.append(
546 _issue(
547 f"{path}.candidate_ids",
548 f"references missing idea ID {candidate!r}",
549 )
550 )
551 _check_text(decision.get("decision"), f"{path}.decision", errors)
552 _check_text(
553 decision.get("rationale"),
554 f"{path}.rationale",
555 errors,
556 )
558 return {
559 "valid": not errors,
560 "errors": errors,
561 "warnings": warnings,
562 "statistics": {
563 "participants": len(participant_ids),
564 "ideas": len(ideas_raw),
565 "assumptions": len(assumptions_raw),
566 "ai_assisted_or_mixed_ideas": ai_assisted_count,
567 "errors": len(errors),
568 "warnings": len(warnings),
569 },
570 "limitations": [
571 (
572 "Checks cover session identity, ideas, assumptions, clusters, "
573 "and decision references. Literature-check, adversarial-review, "
574 "gate, workflow, and governance records are not schema-validated."
575 ),
576 (
577 "Validation is structural and deterministic; it does not verify "
578 "claims, citations, novelty, feasibility, or scientific validity."
579 ),
580 (
581 "Exact normalized-text duplicate warnings do not assert semantic "
582 "equivalence."
583 ),
584 (
585 "A valid register is not ethics, biosafety, regulatory, clinical, "
586 "or institutional approval."
587 ),
588 ],
589 }
592def build_parser() -> argparse.ArgumentParser:
593 """Create the command-line parser."""
595 parser = argparse.ArgumentParser(
596 description=(
597 "Validate the structure, provenance, links, and bounded fields in a "
598 "brainstorming JSON register. No network or LLM calls are made."
599 )
600 )
601 parser.add_argument("input", help="Session/register .json file")
602 parser.add_argument(
603 "--output",
604 help="Optional validation-report .json file in an existing directory",
605 )
606 parser.add_argument(
607 "--strict",
608 action="store_true",
609 help="Return exit status 1 for warnings as well as errors",
610 )
611 parser.add_argument(
612 "--force",
613 action="store_true",
614 help="Explicitly replace an existing regular output file",
615 )
616 return parser
619def main(argv: Sequence[str] | None = None) -> int:
620 """Run register validation."""
622 args = build_parser().parse_args(argv)
623 try:
624 _, document = read_json(args.input, label="register")
625 report = validate_register(document)
626 emit_json(report, args.output, force=args.force)
627 except CliError as exc:
628 print(f"error: {exc}", file=sys.stderr)
629 return 2
630 if not report["valid"] or (args.strict and report["warnings"]):
631 return 1
632 return 0
635if __name__ == "__main__":
636 raise SystemExit(main())