Skills › Scientific Agent Skills › clinical-decision-support › decision_logic_traceability.py
decision_logic_traceability.py
317 lines · skills/clinical-decision-support/scripts/decision_logic_traceability.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate and export research/governance decision-logic traceability."""
4from __future__ import annotations
6import argparse
7import csv
8import io
9import re
10import sys
11from typing import Any
13from _common import (
14 InputError,
15 IssueLog,
16 load_json_object,
17 require_list,
18 require_nonempty_text,
19 source_ids,
20 validate_references,
21 write_text,
22)
24MAX_NODES = 200
25NODE_ID = re.compile(r"^[A-Za-z][A-Za-z0-9_-]{0,63}$")
26ALLOWED_NODE_TYPES = {
27 "input_check",
28 "data_quality_gate",
29 "evidence_rule",
30 "validation_gate",
31 "documentation_gate",
32 "human_review",
33 "release_gate",
34 "monitoring_gate",
35}
36ALLOWED_OUTPUT_KINDS = {
37 "include_evidence",
38 "exclude_evidence",
39 "flag_for_review",
40 "validation_status",
41 "documentation_status",
42 "release_hold",
43 "monitoring_status",
44}
45PROHIBITED_LOGIC = (
46 "diagnose",
47 "prescribe",
48 "administer",
49 "dose ",
50 "dosing",
51 "triage",
52 "urgent",
53 "alarm",
54 "alert clinician",
55 "treatment selection",
56 "therapy recommendation",
57 "bedside",
58 "patient-specific",
59)
62def _safe_csv(value: Any) -> str:
63 text = str(value).replace("\r", " ").replace("\n", " ")
64 if text.lstrip().startswith(("=", "+", "-", "@")):
65 return "'" + text
66 return text
69def _has_cycle(edges: dict[str, list[str]]) -> bool:
70 visiting: set[str] = set()
71 visited: set[str] = set()
73 def visit(node: str) -> bool:
74 if node in visiting:
75 return True
76 if node in visited:
77 return False
78 visiting.add(node)
79 for dependency in edges[node]:
80 if visit(dependency):
81 return True
82 visiting.remove(node)
83 visited.add(node)
84 return False
86 return any(visit(node) for node in edges)
89def validate_matrix(
90 document: dict[str, Any]
91) -> tuple[IssueLog, list[dict[str, str]]]:
92 log = IssueLog()
93 normalized: list[dict[str, str]] = []
94 try:
95 require_nonempty_text(document.get("schema_version"), "schema_version")
96 known_sources = source_ids(document)
97 metadata = document.get("metadata")
98 if not isinstance(metadata, dict):
99 raise InputError("metadata must be an object")
100 for field in (
101 "logic_id",
102 "title",
103 "version",
104 "status",
105 "owner",
106 "purpose",
107 "change_summary",
108 "monitoring_plan",
109 "retirement_criteria",
110 ):
111 require_nonempty_text(metadata.get(field), f"metadata.{field}")
112 if metadata.get("decision_role") != "research_governance_only":
113 log.errors.append(
114 "metadata.decision_role must be research_governance_only"
115 )
116 if not isinstance(metadata.get("data_level"), str) or metadata.get("data_level") not in {"aggregate", "synthetic", "metadata_only"}:
117 log.errors.append(
118 "metadata.data_level must be aggregate, synthetic, or metadata_only"
119 )
120 if metadata.get("patient_care_use") is not False:
121 log.errors.append("metadata.patient_care_use must be false")
122 if metadata.get("executable_logic") is not False:
123 log.errors.append("metadata.executable_logic must be false")
125 nodes = require_list(document.get("nodes"), "nodes", maximum=MAX_NODES)
126 if not nodes:
127 raise InputError("At least one traceability node is required")
128 node_ids: set[str] = set()
129 for index, node in enumerate(nodes):
130 field = f"nodes[{index}]"
131 if not isinstance(node, dict):
132 raise InputError(f"{field} must be an object")
133 node_id = require_nonempty_text(
134 node.get("id"), f"{field}.id", max_length=64
135 )
136 if not NODE_ID.fullmatch(node_id):
137 raise InputError(f"{field}.id has an unsafe or invalid format")
138 if node_id in node_ids:
139 raise InputError(f"Duplicate node id: {node_id}")
140 node_ids.add(node_id)
142 edges: dict[str, list[str]] = {}
143 for index, node in enumerate(nodes):
144 field = f"nodes[{index}]"
145 node_id = require_nonempty_text(node["id"], f"{field}.id", max_length=64)
146 node_type = require_nonempty_text(node.get("type"), f"{field}.type")
147 if node_type not in ALLOWED_NODE_TYPES:
148 log.errors.append(f"{field}.type is unsupported")
149 precondition = require_nonempty_text(
150 node.get("precondition"), f"{field}.precondition"
151 )
152 statement = require_nonempty_text(
153 node.get("logic_statement"), f"{field}.logic_statement"
154 )
155 output_kind = require_nonempty_text(
156 node.get("output_kind"), f"{field}.output_kind"
157 )
158 if output_kind not in ALLOWED_OUTPUT_KINDS:
159 log.errors.append(f"{field}.output_kind is unsupported")
160 output_value = require_nonempty_text(
161 node.get("output_value"), f"{field}.output_value"
162 )
163 rationale = require_nonempty_text(
164 node.get("rationale"), f"{field}.rationale"
165 )
166 owner = require_nonempty_text(node.get("owner"), f"{field}.owner")
167 reviewer = require_nonempty_text(
168 node.get("reviewer_role"), f"{field}.reviewer_role"
169 )
170 status = require_nonempty_text(node.get("status"), f"{field}.status")
171 if status not in {"draft", "validated", "retired"}:
172 log.errors.append(f"{field}.status is unsupported")
173 combined = f"{precondition} {statement} {output_value}".lower()
174 found = [phrase for phrase in PROHIBITED_LOGIC if phrase in combined]
175 if found:
176 log.errors.append(
177 f"{field} contains prohibited clinical logic: {', '.join(found)}"
178 )
180 dependencies = require_list(
181 node.get("dependency_ids", []),
182 f"{field}.dependency_ids",
183 maximum=30,
184 )
185 normalized_dependencies: list[str] = []
186 for dep_index, dependency in enumerate(dependencies):
187 dep = require_nonempty_text(
188 dependency,
189 f"{field}.dependency_ids[{dep_index}]",
190 max_length=64,
191 )
192 if dep not in node_ids:
193 log.errors.append(f"{field} references unknown dependency: {dep}")
194 if dep == node_id:
195 log.errors.append(f"{field} cannot depend on itself")
196 if dep in node_ids and dep != node_id:
197 normalized_dependencies.append(dep)
198 edges[node_id] = normalized_dependencies
200 source_refs = validate_references(
201 node.get("source_ids"), known_sources, f"{field}.source_ids"
202 )
203 tests = require_list(
204 node.get("validation_tests"), f"{field}.validation_tests", maximum=30
205 )
206 if not tests:
207 log.errors.append(f"{field} requires at least one validation test")
208 normalized_tests = [
209 require_nonempty_text(test, f"{field}.validation_tests[{test_index}]")
210 for test_index, test in enumerate(tests)
211 ]
212 normalized.append(
213 {
214 "node_id": node_id,
215 "type": node_type,
216 "dependencies": "; ".join(normalized_dependencies),
217 "precondition": precondition,
218 "logic_statement": statement,
219 "output_kind": output_kind,
220 "output_value": output_value,
221 "source_ids": "; ".join(source_refs),
222 "rationale": rationale,
223 "validation_tests": "; ".join(normalized_tests),
224 "owner": owner,
225 "reviewer_role": reviewer,
226 "status": status,
227 }
228 )
230 if _has_cycle(edges):
231 log.errors.append("Dependency graph contains a cycle")
233 review = document.get("human_review")
234 if not isinstance(review, dict):
235 raise InputError("human_review must be an object")
236 if review.get("required") is not True:
237 log.errors.append("human_review.required must be true")
238 require_nonempty_text(
239 review.get("approval_boundary"), "human_review.approval_boundary"
240 )
241 if metadata.get("status") == "validated" and review.get("completed") is not True:
242 log.errors.append(
243 "Validated matrix status requires completed human review"
244 )
245 elif review.get("completed") is not True:
246 log.warnings.append("Human review is not recorded as complete")
247 except InputError as exc:
248 log.errors.append(str(exc))
250 if log.ok:
251 log.info.append("Research/governance traceability matrix is structurally valid")
252 else:
253 normalized = []
254 return log, normalized
257def _csv_text(rows: list[dict[str, str]]) -> str:
258 fieldnames = [
259 "node_id",
260 "type",
261 "dependencies",
262 "precondition",
263 "logic_statement",
264 "output_kind",
265 "output_value",
266 "source_ids",
267 "rationale",
268 "validation_tests",
269 "owner",
270 "reviewer_role",
271 "status",
272 ]
273 stream = io.StringIO()
274 writer = csv.DictWriter(stream, fieldnames=fieldnames, lineterminator="\n")
275 writer.writeheader()
276 for row in rows:
277 writer.writerow({key: _safe_csv(row[key]) for key in fieldnames})
278 return stream.getvalue()
281def main() -> int:
282 parser = argparse.ArgumentParser(
283 description=(
284 "Validate research/governance decision logic and emit a CSV traceability "
285 "matrix. The tool never executes logic."
286 )
287 )
288 parser.add_argument("input", help="Local matrix JSON")
289 parser.add_argument("-o", "--output", help="Optional local CSV output")
290 parser.add_argument(
291 "--strict", action="store_true", help="Return failure when warnings are present"
292 )
293 args = parser.parse_args()
295 try:
296 document = load_json_object(args.input)
297 log, rows = validate_matrix(document)
298 for message in log.errors:
299 print(f"ERROR: {message}", file=sys.stderr)
300 for message in log.warnings:
301 print(f"WARNING: {message}", file=sys.stderr)
302 if not log.ok or (args.strict and log.warnings):
303 return 1
304 if args.output:
305 write_text(args.output, _csv_text(rows), {".csv"})
306 else:
307 print(_csv_text(rows), end="")
308 except InputError as exc:
309 print(f"ERROR: {exc}", file=sys.stderr)
310 return 2
311 if not log.ok or (args.strict and log.warnings):
312 return 1
313 return 0
316if __name__ == "__main__":
317 sys.exit(main())