from __future__ import annotations import json import re import textwrap from pathlib import Path from typing import Any from . import __version__ from .redaction import redact_object, sanitize_artifact_text from .render_html import render_html from .render_html_components import REQ_LABELS as _REQ_LABELS try: from reportlab.lib import colors from reportlab.lib.pagesizes import A4 from reportlab.lib.styles import ParagraphStyle from reportlab.lib.units import mm from reportlab.platypus import ( HRFlowable, KeepInFrame, PageBreak, Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle, ) _RL = True except ImportError: _RL = False # ── palette ─────────────────────────────────────────────────────────────────── _NAVY = "#1B2A4A" _TEAL = "#2E86AB" _PURPLE = "#6B4D8E" _SLATE = "#3D5A7A" _GREEN = "#27A560" _AMBER = "#C97B10" _RED = "#C0392B" _ORANGE = "#C96010" _LGRAY = "#F5F7FA" _MGRAY = "#E2E8F0" _DGRAY = "#4A5568" _WHITE = "#FFFFFF" _TIER_COLOR = {"T0": _RED, "T1": _RED, "T2": _ORANGE, "T3": _TEAL, "T4": _GREEN} _PDF_CONTENT_WIDTH = A4[0] - (28 * mm) _PDF_CONTENT_HEIGHT = A4[1] - (24 * mm) _BIO_DETECTOR_LABELS = { "BIO_smiles_surface_integrity": "SMILES Surface Integrity", "BIO_smiles_rdkit_validation": "SMILES RDKit Validation", "BIO_smiles_parser_guard": "SMILES Parser Guard", "BIO_silent_mock_fallback": "Silent Mock Fallback", "BIO_trace_manifest": "Traceability Manifest Surface", "BIO_run_trace": "Bio Subprocess Run Trace", } _STATUS_LABELS: dict[str, str] = { "signal_only": "⚠ Signal only", "partially_aligned": "~ Partially aligned", "aligned": "✓ Aligned", "not_detected": "✗ Not detected", } def _hx(h: str) -> Any: return colors.HexColor(h) def _tier_hex(tier: str) -> str: for k, v in _TIER_COLOR.items(): if k in tier: return v return _DGRAY def _status_hex(s: str) -> str: return {"PASS": _GREEN, "FAIL": _RED, "WARN": _AMBER}.get(s.upper(), _DGRAY) def _xt(t: str) -> str: """Escape text for use in reportlab XML markup.""" return str(t).replace("&", "&").replace("<", "<").replace(">", ">") def _clip_words(text: str, limit: int) -> str: """Trim table text without cutting through a word.""" value = " ".join(str(text).split()) if len(value) <= limit: return value trimmed = value[: max(0, limit - 1)].rsplit(" ", 1)[0].rstrip(".,;:") return f"{trimmed}..." # ── public API ──────────────────────────────────────────────────────────────── def write_outputs( result: dict[str, Any], output_dir: Path, mode: str, pages: int, fmt: str, explain: bool = False, ) -> list[Path]: output_dir.mkdir(parents=True, exist_ok=True) stem = _safe_name(result["target"]["name"]) created: list[Path] = [] safe_result = redact_object(result) if fmt in {"json", "all"}: p = output_dir / f"{stem}_experiment_results.json" payload = json.dumps(safe_result, indent=2) p.write_text(payload, encoding="utf-8") created.append(p) if "ai_advisory_input" in safe_result: p = output_dir / f"{stem}_advisory_input.json" payload = json.dumps(safe_result["ai_advisory_input"], indent=2) p.write_text(payload, encoding="utf-8") created.append(p) md = render_markdown(safe_result, mode, pages) if fmt in {"md", "all"}: p = output_dir / f"{stem}_report.md" md, _ = sanitize_artifact_text(md) p.write_text(md, encoding="utf-8") created.append(p) if fmt in {"html", "all"}: p = output_dir / f"{stem}_report.html" p.write_text(render_html(safe_result), encoding="utf-8") created.append(p) if fmt in {"pdf", "all"}: p = output_dir / f"{stem}_{mode}_{pages}p.pdf" if _RL: _write_rl_pdf(p, safe_result, mode, pages) else: write_simple_pdf(p, render_pdf_pages(safe_result, mode, pages)) created.append(p) if explain: p = output_dir / f"{stem}_explain.txt" explain_text, _ = sanitize_artifact_text(render_explain(safe_result)) p.write_text(explain_text, encoding="utf-8") created.append(p) return created def _surface_compaction_note(result: dict[str, Any]) -> str: notes = result.get("artifact_surface_notes", {}) if isinstance(notes, dict): text = notes.get("human_readable_compaction") if isinstance(text, str) and text.strip(): return text.strip() return "Human-readable surfaces may compact repeated same-file evidence; JSON remains the canonical full-fidelity artifact." def _score_boundary_short_line() -> str: return "Score reflects calculation integrity, not calibrated validity. Triage signal only." def _score_boundary_lines() -> list[str]: return [ "**What is verified:** calculation integrity. The same input produces the same score.", "**What is not verified:** calibrated measurement validity. Weights and detector scope remain bounded.", "**Use this score as a triage signal, not as certification, safety proof, or deployment approval.**", ] def _compact_labels(items: list[str], *, limit: int = 5) -> str: values = [str(item).strip() for item in items if str(item).strip()] if not values: return "" shown = values[:limit] extra = max(0, len(values) - len(shown)) text = ", ".join(shown) if extra: text += f" (+{extra} more)" return text # ── markdown ────────────────────────────────────────────────────────────────── def render_markdown(result: dict[str, Any], mode: str, pages: int) -> str: score = result["score"] ast_note = _ast_scope_note(result) calibration = result.get("calibration_profile", {}) calibration_effect = _calibration_effect_note(calibration) lines = [ "# STEM BIO-AI Local Audit Report", "", f"**Target:** `{result['target']['name']}`", f"**Execution Mode:** `{result['execution_mode']}`", f"**Calibration Profile:** `{calibration.get('profile_name', 'unknown')}` " f"(`{calibration.get('policy_version', 'unknown')}`, " f"`{calibration.get('profile_read_mode', 'unknown')}`, " f"`{calibration.get('profile_status', 'unknown')}`)", *([f"**Calibration Effect:** {calibration_effect}"] if calibration_effect else []), *_classification_applied_md(result), f"**Final Score:** **{score['final_score']} / 100**", f"**Formal Tier:** **{score['formal_tier']}**", f"**Tier Meaning:** {score['formal_tier']} = {score['use_scope']}", f"**Use Scope:** {score['use_scope']}", *_tier_lock_label_md(result), f"**About This Score:** {_score_boundary_short_line()}", *_score_boundary_lines(), "", "## Score Matrix", "", "| Stage | Weight | Score |", "| --- | ---: | ---: |", f"| Stage 1 README Evidence Signal | 0.40 | {score['stage_1_readme_intent']} |", f"| Stage 2R Repo-Local Consistency | 0.20 | {score['stage_2_repo_local_consistency']} |", f"| Stage 3 Code/Bio Responsibility | 0.40 | {score['stage_3_code_bio']}{_s3_formula(result)} |", f"| Risk Penalty | -- | {score['risk_penalty']} |", "", "## Replication Evidence Lane", "", f"**Stage 4 Replication Score:** **{result.get('replication_score', 0)} / 100**", f"**Replication Tier:** **{result.get('replication_tier', 'R0')}**", "**Interpretation:** Stage 4 is a separate replication lane. It improves inspectability and reproducibility review, but it does not currently change the formal tier.", "", *_markdown_freshness_section(result.get("audit_freshness", {})), "## Reasoning Diagnostics", "", _markdown_reasoning_summary(result.get("reasoning_model", {})), *_markdown_reasoning_interpretation(result.get("reasoning_model", {})), "", *_markdown_advisory_section(result.get("ai_advisory")), *_markdown_regulatory_section(result), *_markdown_airi_section(result.get("airi_risk_coverage", {})), "## Code Integrity", ] for key, item in result["code_integrity"].items(): lines.append(f"- **{key}:** {item['status']} — {item['evidence'][0]}") if item["status"] in {"WARN", "FAIL"}: for detail in item.get("evidence", [])[1:4]: lines.append(f" - {detail}") if ast_note: lines.append(f"- **AST analysis scope:** {ast_note}") lines.extend(_markdown_bio_section(result)) lines.extend(["", "## Top Risks"]) for risk in result["notable_risks"][:5]: lines.append(f"- {risk}") lines.extend(_markdown_remediation_targets(result)) if mode == "detailed": lines.extend(["", "## Stage 1 Evidence"]) for key, item in result.get("stage_1_rubric", {}).items(): if isinstance(item, dict): score_value = item.get("score", "") lines.append( f"- **{key}:** {score_value} — {item.get('evidence', '')}" f"{_stage1_semantics_suffix(key, item)}" ) lines.extend(["", "## Stage 2R Evidence"]) for key, item in result["stage_2r_rubric"].items(): if isinstance(item, dict): lines.append( f"- **{key}:** {item.get('score', '')} — {item.get('evidence', '')}" f"{_rubric_trace_suffix(item)}" ) lines.extend(["", "## Stage 3 Evidence"]) for key, item in result["stage_3_rubric"].items(): lines.append( f"- **{key}:** {item['score']} / {item['max']} — {item['evidence']}" f"{_rubric_trace_suffix(item)}" ) lines.extend(["", "## Stage 4 Replication Evidence"]) for key, item in result.get("stage_4_rubric", {}).items(): lines.append(f"- **{key}:** {item['score']} / {item['max']} — {item['evidence']}") lines.extend(["", "## Method Boundary", result["method"]]) lines.extend([ "", "## Disclaimer", "This is an evidence-surface pre-screen, not clinical certification, " "regulatory clearance, or medical advice.", ]) return "\n".join(lines) + "\n" # ── explain text report ─────────────────────────────────────────────────────── _EXPLAIN_SEP = "=" * 72 _EXPLAIN_META_SKIP = frozenset({"file_count", "max_ast_files", "max_file_size_bytes"}) def render_explain(result: dict[str, Any]) -> str: """Return a human-readable plain-text explain report grouped by detector.""" ledger: list[dict[str, Any]] = result.get("evidence_ledger", []) score = result["score"] grouped: dict[str, list[dict[str, Any]]] = {} for finding in ledger: grouped.setdefault(finding["detector"], []).append(finding) calibration = result.get("calibration_profile", {}) calibration_effect = _calibration_effect_note(calibration) out: list[str] = [ "STEM BIO-AI Explain Report", f"Target : {result['target']['name']}", ( "Policy : " f"{calibration.get('profile_name', 'unknown')} " f"[{calibration.get('policy_version', 'unknown')}; " f"{calibration.get('profile_read_mode', 'unknown')}; " f"{calibration.get('profile_status', 'unknown')}]" ), *([f"Policy Mode: {calibration_effect}"] if calibration_effect else []), f"Score : {score['final_score']} / 100 ({score['formal_tier']})", f"Replic : {result.get('replication_score', 0)} / 100" f" ({result.get('replication_tier', 'R0')})", "Surface : repeated same-file evidence may be compacted in narrative output; JSON remains canonical.", _EXPLAIN_SEP, "", ] for detector, findings in grouped.items(): out += _explain_detector_group(detector, findings) out += _explain_bio_section(result) out += _explain_regulatory_section(result) out += _explain_airi_section(result.get("airi_risk_coverage", {})) out += _explain_freshness_section(result.get("audit_freshness", {})) out += _explain_ast_section(result.get("ast_signal_summary", {})) out += _explain_s4_section(result.get("stage_4_rubric", {})) out += _explain_reasoning_section(result.get("reasoning_model", {})) out += _explain_advisory_section(result.get("ai_advisory")) out += [_EXPLAIN_SEP, "DISCLAIMER: Evidence-surface pre-screen only.", "Not clinical certification, regulatory clearance, or medical advice."] return "\n".join(out) + "\n" def _classification_applied_md(result: dict[str, Any]) -> list[str]: cls = result.get("classification", {}) ca = cls.get("ca_severity", "none") cap = cls.get("score_cap") t0 = cls.get("t0_hard_floor", False) cap_str = str(cap) if cap is not None else "none" t0_str = "active" if t0 else "clear" return [f"**Classification Applied:** ca_severity={ca} | score_cap={cap_str} | t0_floor={t0_str}"] def _tier_lock_label_md(result: dict[str, Any]) -> list[str]: cls = result.get("classification", {}) score_cap = cls.get("score_cap") if score_cap is None: return [] if cls.get("t0_hard_floor"): return [ f"**Tier Lock [T0-FLOOR]:** Score ceiling active at **39** (T0 maximum). " f"CA-DIRECT classification with insufficient code presence. " f"Resolving this condition is required before any tier advancement." ] return [ f"**Tier Lock [CA-CAP]:** Score ceiling active at **{score_cap}** (T2 maximum). " f"Clinical-adjacent surface detected without explicit non-clinical boundary. " f"Adding a non-diagnostic disclaimer resolves this lock." ] def _s3_formula(result: dict[str, Any]) -> str: s3_raw = result.get("stage_3_rubric", {}).get("stage_3_raw_total", {}) raw = s3_raw.get("score") max_val = s3_raw.get("max") if raw is None or not max_val: return "" normalized = result.get("score", {}).get("stage_3_code_bio") if isinstance(normalized, (int, float)): return f" (raw: {raw}/{max_val} -> normalized: {int(round(normalized))})" return f" (raw: {raw}/{max_val})" def _markdown_reasoning_summary(reasoning: dict[str, Any]) -> str: if not reasoning: return "Reasoning diagnostics are not available." coherence = reasoning.get("lane_coherence", {}) uncertainty = reasoning.get("uncertainty_budget", {}) gate = reasoning.get("evidence_risk_gate", {}) envelope = reasoning.get("confidence_envelope", {}) policy = reasoning.get("policy", {}) return ( f"Diagnostic-only heuristic `{reasoning.get('version', 'unknown')}` " f"({policy.get('weights', 'uncalibrated')}); " f"lane consistency `{coherence.get('status', 'unknown')}` " f"({coherence.get('overall', 'n/a')}; >=0.80=consistent, >=0.55=mixed, <0.55=divergent), " f"uncertainty band `{uncertainty.get('status', 'unknown')}` " f"({uncertainty.get('uncertainty', 'n/a')}; <0.20=low-spread, <=0.45=review-advised, >0.45=manual-review), " f"risk heuristic `{gate.get('status', 'unknown')}` " f"({gate.get('evidence_risk', 'n/a')}), " f"confidence envelope {envelope.get('lower', 'n/a')}-" f"{envelope.get('upper', 'n/a')}. " "This heuristic layer does not override the final score." ) def _markdown_reasoning_interpretation(reasoning: dict[str, Any]) -> list[str]: if not reasoning: return [] coherence = reasoning.get("lane_coherence", {}) uncertainty = reasoning.get("uncertainty_budget", {}) notes: list[str] = [] if coherence.get("status") in {"heuristic_mixed", "heuristic_divergent"}: notes.append( "- **Interpretation:** lane coherence is mixed, which means README-facing intent and code/accountability signals do not move together cleanly. Review Stage 2R and Stage 3 evidence before treating the score as stable." ) if uncertainty.get("status") == "review_advised": notes.append( "- **Interpretation:** the uncertainty band is elevated enough that manual review is recommended, especially for boundary claims, workflow support, and governance surfaces." ) return notes def _calibration_effect_note(calibration: dict[str, Any]) -> str | None: if calibration.get("profile_read_mode") != "mirror_only": return None return ( f"mirror-only in {__version__} — selected profile metadata is surfaced in artifacts, " "but authoritative scan scoring still follows deterministic runtime constants. " "Preview-only posture changes, including Stage 4 replication emphasis, do not " "change the formal score until a future read-through phase. " "Use `stem policy simulate` to preview governed score deltas and posture changes." ) def _markdown_advisory_section(advisory: dict[str, Any] | None) -> list[str]: if not advisory: return [] return [ "## AI Advisory Contract", "", f"**Status:** `{advisory.get('status', 'unknown')}`", f"**Provider:** `{advisory.get('provider', 'none')}`", f"**Mode:** `{advisory.get('mode', 'unknown')}`", f"**Invalid Citations:** {len(advisory.get('invalid_citations', []))}", "", ] def _markdown_freshness_section(freshness: dict[str, Any]) -> list[str]: if not freshness: return [] triggers = ", ".join(freshness.get("change_triggers", [])[:4]) reasons = ", ".join(freshness.get("change_triggered_reaudit_reasons", [])) or "none" return [ "## Audit Freshness", "", f"**Review After:** **{freshness.get('review_after_days', 'n/a')} days**", f"**Expires On:** `{freshness.get('expires_on', 'unknown')}`", f"**Change-triggered re-audit recommended now:** `{freshness.get('change_triggered_reaudit_recommended_now', False)}`", f"**Current re-audit reasons:** `{reasons}`", f"**Trigger examples:** `{triggers}`", "", ] def _markdown_bio_section(result: dict[str, Any]) -> list[str]: rows = _bio_detector_rows(result) if not rows: return [] lines = ["", "## Bio Deterministic Diagnostics", ""] for detector, label, counts in rows: parts = [] for status in ("detected", "warn", "error", "not_detected", "not_applicable", "absent"): value = counts.get(status) if value: parts.append(f"{status}={value}") findings = [ f for f in result.get("evidence_ledger", []) if f.get("detector") == detector and f.get("status") == "detected" ] note = findings[0].get("explanation", "") if findings else _detector_scope_note(result, detector) lines.append(f"- **{label}:** {', '.join(parts) if parts else 'no findings'} — {note}") return lines def _markdown_regulatory_section(result: dict[str, Any]) -> list[str]: basis = result.get("regulatory_basis", {}) traceability = result.get("stage_traceability", {}) if not basis and not traceability: return [] note = basis.get("note", {}) lines = [ "## Regulatory Traceability Assistant", "", f"> **{note.get('title', 'Regulatory basis note')}**", f"> {note.get('body_line_1', '')}", f"> {note.get('body_line_2', '')}", "", ] if basis.get("review_required"): reasons = ", ".join(basis.get("review_reasons", [])) lines.append(f"> ⚠ Review required: `{reasons}`") lines.append("") for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"): items = traceability.get(stage_key, []) if not items: continue lines.append(f"### {stage_key.replace('_', ' ').title()}") for item in items: req_id = item["requirement_id"] label = _REQ_LABELS.get(req_id, req_id) status = _STATUS_LABELS.get(item["status"], item["status"]) src_tag = "" if item.get("source_ids"): src_tag = " `[" + ", ".join(item["source_ids"]) + "]`" lines.append(f"- **{label}**{src_tag} — {status}") refs = item.get("finding_refs", []) if refs: lines.append(f" - Triggered by: {', '.join(f'`{r}`' for r in refs)}") gaps = item.get("not_assessed", []) if gaps: lines.append(f" - Not assessed: {'; '.join(gaps)}") lines.append(f" - {item['note']}") lines.append("") summary = result.get("regulatory_traceability", {}).get("summary") if summary: lines.append(f"**Traceability summary:** {summary}") lines.append("") return lines def _markdown_airi_section(airi: dict[str, Any]) -> list[str]: if not airi: return [] covered = airi.get("covered_count", 0) total = airi.get("total_risks_in_detector_scope", 0) rate = airi.get("coverage_rate", 0) bundle_scope = airi.get("airi_bundle_scope", "unknown") snapshot = airi.get("airi_upstream_snapshot_date", "unknown") lines = [ "## AIRI Risk Triggers", "", f"**Covered Risks:** **{covered} / {total}**", f"**Coverage Rate:** `{rate:.3f}`", f"**Bundle Scope:** `{bundle_scope}`", f"**Upstream Snapshot:** `{snapshot}`", "**Interpretation:** This is detector-mapped AIRI coverage inside the current runtime bundle, not a claim that unmapped risks are absent.", "**Surface Note:** repeated same-file evidence may be compacted in human-readable surfaces; canonical per-finding rows remain in JSON.", ] covered_risks = airi.get("covered_risks", []) if covered_risks: lines.append("") lines.append("**Examples of Covered AIRI Risks**") for risk in covered_risks[:3]: reason = _airi_reason_summary(risk) primary = _airi_primary_summary(risk) lines.append( f"- `{risk.get('id', 'unknown')}` — {risk.get('title', 'unknown')} " f"({primary}; why: {reason})" ) gaps = airi.get("known_gaps_in_bundle", []) if gaps: lines.append("") lines.append("**Known Gaps In Bundle**") for gap in gaps: lines.append(f"- `{gap.get('id', 'unknown')}` — {gap.get('title', 'unknown')}") return lines def _explain_detector_group(detector: str, findings: list[dict[str, Any]]) -> list[str]: compact_findings = _compact_explain_findings(findings) label = _explain_status_label({f["status"] for f in findings}) noun = "finding" if len(findings) == 1 else "findings" compact_noun = "row" if len(compact_findings) == 1 else "rows" heading = f"{detector} [{label}] ({len(findings)} {noun})" if len(compact_findings) != len(findings): heading += f" -> compacted to {len(compact_findings)} {compact_noun}" lines = [heading] for f in compact_findings: lines += _explain_finding_lines(f) lines.append("") return lines def _compact_explain_findings(findings: list[dict[str, Any]]) -> list[dict[str, Any]]: compact: list[dict[str, Any]] = [] grouped: dict[tuple[str, str, str, str], dict[str, Any]] = {} grouped_order: list[tuple[str, str, str, str]] = [] for finding in findings: status = str(finding.get("status", "unknown")) file_path = str(finding.get("file", "")) line = int(finding.get("line", 0) or 0) reason = str(finding.get("explanation") or finding.get("message") or "").strip() if status in {"detected", "warn", "pass"} and file_path not in {"", "."} and reason: key = (status, file_path, reason, str(finding.get("pattern_id", ""))) if key not in grouped: clone = dict(finding) meta = dict(clone.get("metadata") or {}) meta["aggregate_count"] = 1 meta["aggregate_lines"] = [line] if line else [] meta["aggregate_surface"] = "explain_same_file_reason" clone["metadata"] = meta grouped[key] = clone grouped_order.append(key) else: grouped[key]["metadata"]["aggregate_count"] += 1 if line and line not in grouped[key]["metadata"]["aggregate_lines"]: grouped[key]["metadata"]["aggregate_lines"].append(line) continue compact.append(finding) for key in grouped_order: group = grouped[key] count = int(group.get("metadata", {}).get("aggregate_count", 1)) lines = sorted(group.get("metadata", {}).get("aggregate_lines", [])) if count > 1: if lines: preview = ", ".join(str(n) for n in lines[:6]) if len(lines) > 6: preview += ", ..." group["explanation"] = f"{group.get('explanation', '')} Aggregated {count} similar findings from one file (lines: {preview}).".strip() else: group["explanation"] = f"{group.get('explanation', '')} Aggregated {count} similar findings from one file.".strip() group["snippet"] = "" compact.append(group) return compact def _explain_finding_lines(f: dict[str, Any]) -> list[str]: occ = f["finding_id"].rsplit(":", 1)[-1] file_str = "(repository)" if f["file"] == "." else ( f"{f['file']}:{f['line']}" if f["line"] else f["file"] ) lines = [f" [{occ}] {file_str}", f" finding_id: {f['finding_id']}"] if f.get("pattern_id"): lines.append(f" pattern : {f['pattern_id']}") if f.get("evidence_status"): lines.append(f" evidence: {f['evidence_status']}") if f.get("confidence"): lines.append(f" conf : {f['confidence']}") if f.get("snippet"): lines.append(f" snippet : \"{f['snippet']}\"") if f.get("explanation"): lines.append(f" reason : {f['explanation']}") for k, v in (f.get("metadata") or {}).items(): if k not in _EXPLAIN_META_SKIP: lines.append(f" {k} : {v}") return lines def _explain_ast_section(ast: dict[str, Any]) -> list[str]: if not ast: return [] lines = [_EXPLAIN_SEP, "AST Signal Summary", ""] lines += [f" {k:<34} {v}" for k, v in ast.items() if v is not None] lines.append("") return lines def _explain_airi_section(airi: dict[str, Any]) -> list[str]: if not airi: return [] lines = [ _EXPLAIN_SEP, "AIRI Risk Triggers", "", f" covered_count {airi.get('covered_count', 0)}", f" detector_scope_total {airi.get('total_risks_in_detector_scope', 0)}", f" coverage_rate {airi.get('coverage_rate', 0)}", f" bundle_scope {airi.get('airi_bundle_scope', 'unknown')}", f" upstream_snapshot {airi.get('airi_upstream_snapshot_date', 'unknown')}", ] covered_risks = airi.get("covered_risks", []) if covered_risks: lines.append("") lines.append(" covered examples:") for risk in covered_risks[:3]: reason = _airi_reason_summary(risk) primary = _airi_primary_summary(risk) lines.append( f" - {risk.get('id', 'unknown')} | {risk.get('title', 'unknown')} " f"| {primary} | why={reason}" ) gaps = airi.get("known_gaps_in_bundle", []) if gaps: lines.append("") lines.append(" known gaps in bundle:") for gap in gaps: lines.append( f" - {gap.get('id', 'unknown')} | {gap.get('title', 'unknown')}" ) lines.append("") return lines def _rubric_trace_suffix(item: dict[str, Any]) -> str: detector = str(item.get("detector_id", "")).strip() basis = str(item.get("decision_basis", "")).strip() tier_impact = str(item.get("tier_impact", "")).strip() parts: list[str] = [] if detector: parts.append(f"detector={detector}") if basis: parts.append(f"basis={basis}") if tier_impact: parts.append(f"tier-impact={tier_impact}") if not parts: return "" return f" `[{' | '.join(parts)}]`" def _airi_reason_summary(risk: dict[str, Any]) -> str: details = risk.get("mapping_details", []) if not details: return "bounded detector-to-risk mapping" snippets: list[str] = [] for detail in details[:5]: detector = str(detail.get("detector_id", "")).strip() trigger = str(detail.get("trigger_reason", "")).strip() justification = str(detail.get("mapping_justification", "")).strip() reason = trigger or justification or "bounded detector-to-risk mapping" snippets.append(f"{detector}: {reason}" if detector else reason) summary = " ; ".join(snippets) extra = max(0, len(details) - len(snippets)) if extra: summary += f" ; (+{extra} more mapping details)" return summary def _airi_primary_summary(risk: dict[str, Any]) -> str: primary = str(risk.get("primary_detector_id", "")).strip() secondary = [str(det).strip() for det in risk.get("secondary_detector_ids", []) if str(det).strip()] if primary and secondary: return f"primary: {primary}; also linked by {_compact_labels(secondary)}" if primary: return f"primary: {primary}" covered_by = [str(det).strip() for det in risk.get("covered_by", []) if str(det).strip()] if covered_by: return f"covered by: {_compact_labels(covered_by)}" return "bounded detector-to-risk mapping" def _stage1_semantics_suffix(key: str, item: dict[str, Any]) -> str: if key != "R2_regulatory_framework": return "" score = item.get("score") ladder = "+15 strong framework | +5 weak self-asserted compliance | -5 CA-INDIRECT missing framework | -10 CA-DIRECT missing framework" return f" `[partial-credit ladder={ladder}; current={score}]`" def _markdown_remediation_targets(result: dict[str, Any]) -> list[str]: rows: list[tuple[str, str, str]] = [] stage2 = result.get("stage_2r_rubric", {}) code_integrity = result.get("code_integrity", {}) if "R2R_D2_missing_clinical_use_boundary" in stage2: rows.append(( "R2R_D2 missing clinical boundary", "Add non-clinical/non-diagnostic disclaimer to README and all adjacent docs", "+20 S2R (+4 final) | unlocks tier cap", )) if "R2R_D4_unsupported_workflow_claim" in stage2: rows.append(( "R2R_D4 unsupported workflow claim", "Align README workflow/demo/CLI claims with actual local support surfaces", "+15 S2R (+3 final)", )) if code_integrity.get("C2_dependency_pinning", {}).get("status") in {"WARN", "FAIL"}: rows.append(( "C2 dependency pinning WARN", "Pin production dependencies; document external-service dependence explicitly", "C2 -> PASS (no direct score delta)", )) if code_integrity.get("C5_compliance_boundary_integrity", {}).get("status") in {"WARN", "FAIL"}: rows.append(( "C5 compliance boundary WARN", "Remove unsupported legal/compliance language or add backing governance evidence", "C5 -> PASS (no direct score delta)", )) if code_integrity.get("C6_mock_auth_or_fail_open_boundary", {}).get("status") in {"WARN", "FAIL"}: rows.append(( "C6 mock-auth boundary WARN", "Separate mock-auth/auto-login flows from production trust boundary narrative", "C6 -> PASS (no direct score delta)", )) if not rows: return [] lines = [ "", "## Remediation Roadmap", "", "| Finding | Action | Expected Impact |", "| --- | --- | --- |", ] for finding, action, impact in rows: lines.append(f"| {finding} | {action} | {impact} |") return lines def _explain_freshness_section(freshness: dict[str, Any]) -> list[str]: if not freshness: return [] lines = [_EXPLAIN_SEP, "Audit Freshness", ""] lines.append(f" review_after_days {freshness.get('review_after_days', 'n/a')}") lines.append(f" expires_on {freshness.get('expires_on', 'unknown')}") lines.append(f" freshness_basis {freshness.get('freshness_basis', 'unknown')}") lines.append( f" change_triggered_reaudit_now {freshness.get('change_triggered_reaudit_recommended_now', False)}" ) lines.append( f" change_triggered_reasons {', '.join(freshness.get('change_triggered_reaudit_reasons', [])) or 'none'}" ) lines.append( f" trigger_examples {', '.join(freshness.get('change_triggers', [])[:4])}" ) lines.append("") return lines def _ast_scope_note(result: dict[str, Any]) -> str | None: ast = result.get("ast_signal_summary", {}) if not ast or not ast.get("file_limit_exceeded"): return None considered = ast.get("files_considered", "unknown") total = ast.get("files_total", "unknown") return ( f"AST analysis capped at {considered} of {total} Python files; " "remaining files were excluded from C1/C4 AST-backed analysis." ) def _explain_s4_section(s4: dict[str, Any]) -> list[str]: if not s4: return [] lines = [_EXPLAIN_SEP, "Stage 4 Replication Rubric", ""] for key, item in s4.items(): sc, mx, ev = item.get("score", 0), item.get("max", 0), item.get("evidence", "") lines.append(f" {key:<42} {sc:>3} / {mx:<3} {ev}") lines.append("") return lines def _explain_bio_section(result: dict[str, Any]) -> list[str]: rows = _bio_detector_rows(result) if not rows: return [] lines = [_EXPLAIN_SEP, "Bio Deterministic Diagnostics", ""] ledger = result.get("evidence_ledger", []) for detector, label, counts in rows: parts = [f"{status}={counts[status]}" for status in ("detected", "warn", "error", "not_detected", "not_applicable", "absent") if counts.get(status)] lines.append(f" {label:<34} {', '.join(parts) if parts else 'no findings'}") first = next((f for f in ledger if f.get("detector") == detector and f.get("status") == "detected"), None) if first: lines.append(f" first finding: {first.get('finding_id', 'n/a')}") lines.append(f" reason : {first.get('explanation', '')}") lines.append("") return lines def _explain_reasoning_section(reasoning: dict[str, Any]) -> list[str]: if not reasoning: return [] lines = [_EXPLAIN_SEP, "Reasoning Diagnostics", ""] lines.append(f" version {reasoning.get('version', 'unknown')}") policy = reasoning.get("policy", {}) lines.append(f" mode {policy.get('mode', 'unknown')}") lines.append(f" final_score_override {policy.get('final_score_override', False)}") lines.append(f" weights {policy.get('weights', 'unknown')}") for key in ("evidence_budget", "confidence_envelope", "lane_coherence", "uncertainty_budget", "evidence_risk_gate"): item = reasoning.get(key, {}) status = item.get("status", "unknown") lines.append(f" {key:<31} {status}") if reasoning.get("lane_coherence", {}).get("status") in {"heuristic_mixed", "heuristic_divergent"}: lines.append(" interpretation mixed lane coherence; review Stage 2R and Stage 3 evidence manually") if reasoning.get("uncertainty_budget", {}).get("status") == "review_advised": lines.append(" review_note uncertainty band elevated; manual review advised") lines.append("") return lines def _explain_advisory_section(advisory: dict[str, Any] | None) -> list[str]: if not advisory: return [] lines = [_EXPLAIN_SEP, "AI Advisory Contract", ""] lines.append(f" schema_version {advisory.get('schema_version', 'unknown')}") lines.append(f" provider {advisory.get('provider', 'none')}") lines.append(f" mode {advisory.get('mode', 'unknown')}") lines.append(f" status {advisory.get('status', 'unknown')}") lines.append(f" final_score_override {advisory.get('policy', {}).get('final_score_override', False)}") lines.append(f" invalid_citations {len(advisory.get('invalid_citations', []))}") lines.append("") return lines def _explain_regulatory_section(result: dict[str, Any]) -> list[str]: basis = result.get("regulatory_basis", {}) traceability = result.get("stage_traceability", {}) if not basis and not traceability: return [] note = basis.get("note", {}) lines = [_EXPLAIN_SEP, "Regulatory Traceability Assistant", ""] lines.append(f" {note.get('title', 'Regulatory basis note')}") lines.append(f" {note.get('body_line_1', '')}") lines.append(f" {note.get('body_line_2', '')}") if basis.get("review_required"): lines.append(f" review_required {', '.join(basis.get('review_reasons', []))}") lines.append("") for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"): items = traceability.get(stage_key, []) if not items: continue lines.append(f" {stage_key}") for item in items: req_id = item["requirement_id"] label = _REQ_LABELS.get(req_id, req_id) status = _STATUS_LABELS.get(item["status"], item["status"]) src = ", ".join(item.get("source_ids", [])) lines.append(f" {label}: {status}") if src: lines.append(f" source: {src}") refs = item.get("finding_refs", []) if refs: lines.append(f" triggered by: {', '.join(refs)}") gaps = item.get("not_assessed", []) if gaps: lines.append(f" not assessed: {'; '.join(gaps)}") lines.append(f" note: {item['note']}") summary = result.get("regulatory_traceability", {}).get("summary") if summary: lines.append("") lines.append(f" summary: {summary}") lines.append("") return lines def _explain_status_label(statuses: set[str]) -> str: for candidate in ("error", "detected", "not_detected", "absent", "not_applicable"): if candidate in statuses: return candidate.upper() return next(iter(statuses), "UNKNOWN").upper() def _bio_detector_rows(result: dict[str, Any]) -> list[tuple[str, str, dict[str, int]]]: summary = result.get("detector_summary", {}).get("by_detector", {}) rows: list[tuple[str, str, dict[str, int]]] = [] for detector, label in _BIO_DETECTOR_LABELS.items(): counts = summary.get(detector) if counts: rows.append((detector, label, counts)) return rows def _detector_scope_note(result: dict[str, Any], detector: str) -> str: ledger = result.get("evidence_ledger", []) for status in ("error", "detected", "not_detected", "not_applicable", "absent"): finding = next( (item for item in ledger if item.get("detector") == detector and item.get("status") == status), None, ) if finding and finding.get("explanation"): return str(finding["explanation"]) return "No findings were emitted under current detector scope." # ── reportlab: document entry point ────────────────────────────────────────── def _write_rl_pdf(path: Path, result: dict[str, Any], mode: str, pages: int) -> None: doc = SimpleDocTemplate( str(path), pagesize=A4, topMargin=10 * mm, bottomMargin=12 * mm, leftMargin=14 * mm, rightMargin=14 * mm, ) story: list[Any] = [] story += _page1_executive(result, mode, pages) if mode == "detailed": story += _detail_pages(result, pages) def _draw_footer(canvas: Any, _: Any) -> None: canvas.saveState() canvas.setStrokeColor(_hx(_MGRAY)) canvas.setLineWidth(0.5) canvas.line(doc.leftMargin, 13.2 * mm, A4[0] - doc.rightMargin, 13.2 * mm) canvas.setFont("Helvetica", 7) canvas.setFillColor(_hx(_DGRAY)) canvas.drawCentredString( A4[0] / 2, 8.2 * mm, f"STEM BIO-AI Local CLI Scan | {result.get('stem_ai_version', __version__)} | Deterministic surface scan — no LLM, network, or runtime execution.", ) canvas.drawCentredString( A4[0] / 2, 4.9 * mm, "Not clinical certification. Not regulatory clearance. Not medical advice.", ) canvas.restoreState() doc.build(story, onFirstPage=_draw_footer, onLaterPages=_draw_footer) # ── style factory ───────────────────────────────────────────────────────────── _style_cache: dict[str, Any] = {} _STYLE_CACHE_LIMIT = 256 def _style(name: str, size: int = 9, leading: int = 12, color: str = _DGRAY, bold: bool = False, align: str = "LEFT") -> Any: key = f"{name}_{size}_{leading}_{color}_{bold}_{align}" if key not in _style_cache: if len(_style_cache) >= _STYLE_CACHE_LIMIT: _style_cache.clear() _style_cache[key] = ParagraphStyle( key, fontSize=size, leading=leading, textColor=_hx(color), fontName="Helvetica-Bold" if bold else "Helvetica", alignment={"LEFT": 0, "CENTER": 1, "RIGHT": 2}.get(align, 0), ) return _style_cache[key] # ── Page 1: Executive Dashboard (brief + detailed) ──────────────────────────── def _page1_executive(result: dict[str, Any], mode: str, pages: int) -> list[Any]: story: list[Any] = [] story += _header_block(result) story += _score_row(result) story.append(Spacer(1, 3 * mm)) story += _stage_cards(result) story.append(Spacer(1, 3 * mm)) story += _integrity_and_risks(result) story.append(Spacer(1, 3 * mm)) story += _footer_block() return _single_page_story(story) def _header_block(result: dict[str, Any]) -> list[Any]: t = result["target"] commit = (t.get("commit") or "")[:12] or "—" branch = t.get("branch") or "—" audit_date = result.get("generated_at_local", "—") mode = result.get("execution_mode", "—") calibration = result.get("calibration_profile", {}) profile_label = ( f"{calibration.get('profile_name', 'unknown')} " f"({calibration.get('profile_read_mode', 'unknown')})" ) header_data = [[Paragraph( f'STEM BIO-AI Evidence-Surface Scan v{result["stem_ai_version"]}', _style("H1", 10.5, 13, _WHITE, True), )]] header_tbl = Table(header_data, colWidths=["100%"]) header_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(_NAVY)), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ("LEFTPADDING", (0, 0), (-1, -1), 8), ])) meta = ( f'Repository: {_xt(t["name"])}  |  ' f'Commit: {commit}  |  ' f'Branch: {_xt(branch)}
' f'Audit Date: {audit_date}  |  ' f'Mode: {mode}  |  ' f'Policy: {_xt(profile_label)}' ) meta_data = [[Paragraph(meta, _style("M1", 10.2, 13.8, _DGRAY))]] meta_tbl = Table(meta_data, colWidths=["100%"]) meta_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(_MGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 6), ("BOTTOMPADDING", (0, 0), (-1, -1), 6), ("LEFTPADDING", (0, 0), (-1, -1), 8), ])) return [header_tbl, meta_tbl, Spacer(1, 3 * mm)] def _score_row(result: dict[str, Any]) -> list[Any]: score = result["score"] fs = score["final_score"] tier = score["formal_tier"] tier_hex = _tier_hex(tier) use_scope = score.get("use_scope", "") score_cell = [ Paragraph( f'{fs}' f' / 100', _style("SC1", 22, 27, _NAVY, True, "CENTER"), ), Paragraph("Final Score", _style("SL1", 8, 11, _DGRAY, False, "CENTER")), ] tier_badge = [[Paragraph( f'{_xt(tier)}', _style("TB1", 12, 16, _WHITE, True, "CENTER"), )]] tier_tbl = Table(tier_badge, colWidths=[60 * mm]) tier_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(tier_hex)), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 5), ])) scope_cell = [ tier_tbl, Spacer(1, 2 * mm), Paragraph( f'Use Scope:
' f'{_xt(use_scope)}', _style("US1", 8, 11, _DGRAY), ), ] row_tbl = Table([[score_cell, scope_cell]], colWidths=[44 * mm, None]) row_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (0, 0), _hx(_LGRAY)), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 5), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ])) score_note_lines = "
".join([ f"⚠ About this score
{_xt(_score_boundary_short_line())}", '• What is verified: calculation integrity. The same input produces the same score.', '• What is not verified: calibrated measurement validity. Weights and detector scope remain bounded.', '• Use this score as a triage signal: not as certification, safety proof, or deployment approval.', ]) score_note_tbl = Table( [[Paragraph(f'{score_note_lines}', _style("SN1", 7.5, 10, _DGRAY))]], colWidths=["100%"], ) score_note_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx("#FFF4D6")), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 5), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("BOX", (0, 0), (-1, -1), 0.7, _hx(_AMBER)), ])) return [row_tbl, Spacer(1, 1 * mm), score_note_tbl] def _stage_cards(result: dict[str, Any]) -> list[Any]: score = result["score"] stages = [ ("Stage 1", "README Evidence", score["stage_1_readme_intent"], _TEAL), ("Stage 2R", "Repo-Local Consistency", score["stage_2_repo_local_consistency"] or 0, _PURPLE), ("Stage 3", "Code / Bio Responsibility", score["stage_3_code_bio"], _SLATE), ("Stage 4", "Replication Evidence", result.get("replication_score", 0), _GREEN), ] cells = [] for label, sub, val, col in stages: card = [ [Paragraph( f'{label}
{sub}', _style(f"CH_{label}", 8.5, 12, _WHITE, True, "CENTER"), )], [Paragraph( f'{val}' f' / 100', _style(f"CV_{label}", 22, 26, col, True, "CENTER"), )], ] t = Table(card, colWidths=["100%"]) t.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (0, 0), _hx(col)), ("BACKGROUND", (0, 1), (0, 1), _hx(_LGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 6), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ])) cells.append(t) row = Table([cells], colWidths=["25%", "25%", "25%", "25%"]) row.setStyle(TableStyle([ ("LEFTPADDING", (0, 0), (-1, -1), 3), ("RIGHTPADDING", (0, 0), (-1, -1), 3), ])) return [row] def _integrity_and_risks(result: dict[str, Any]) -> list[Any]: score = result["score"] cls = result.get("classification", {}) risks = result.get("notable_risks", []) positive = result.get("notable_positive_evidence", []) airi = result.get("airi_risk_coverage", {}) stage_3 = result.get("stage_3_rubric", {}) stage_4 = result.get("stage_4_rubric", {}) if cls.get("t0_hard_floor"): posture = ( "Bio-governance posture is not suitable for clinical or patient-adjacent trust. " "Direct-clinical framing appears without an adequate boundary declaration." ) posture_color = _RED elif cls.get("score_cap") is not None: posture = ( "Bio-governance posture is partial and bounded. Some provenance and replication signals exist, " "but explicit non-clinical boundary and trust-control evidence remain insufficient." ) posture_color = _AMBER else: posture = ( "Bio-governance posture is structurally stronger in the reviewed repository surfaces, " "but this artifact still remains a pre-screen rather than certification." ) posture_color = _GREEN present_lines: list[str] = [] if stage_3.get("B1_data_provenance_controls", {}).get("score", 0) > 0: present_lines.append("Repository provenance surfaces are present through dependency or lock manifests.") if stage_3.get("B3_coi_funding", {}).get("score", 0) > 0: present_lines.append("Funding / COI acknowledgement language is present.") if result.get("stage_1_rubric", {}).get("S1_domain_package", {}).get("score", 0) > 0: present_lines.append("Package metadata was available for repo-local consistency checks.") if result.get("replication_score", 0) > 0: present_lines.append("Some reproducibility evidence exists, including environment lock or container surfaces.") if result.get("stage_traceability", {}).get("stage_4"): present_lines.append("Regulatory traceability mappings exist for reproducibility and record-keeping scaffolding.") for item in positive[:3]: if item not in present_lines: present_lines.append(str(item)) if not present_lines: present_lines.append("Only limited positive governance evidence surfaced in the reviewed repository sources.") missing_lines = [str(r) for r in risks[:5]] or ["No major missing-or-contradicted governance surfaces were surfaced."] reg_lines = _regulatory_bullets(result) airi_lines = "" if airi: airi_lines += ( f'• Secondary risk vocabulary: {airi.get("covered_count", 0)} mapped triggers across ' f'{airi.get("total_risks_in_detector_scope", 0)} in-bundle AIRI risks.
' f'• Meaning: broadens local findings into risk language; it does not prove harm, safety, or compliance.
' f'• Why this matters: useful when a user needs a broader risk vocabulary around governance gaps, not when they need deployment approval.' ) covered = airi.get("covered_risks", []) for idx, risk in enumerate(covered[:2], start=1): reason = _airi_reason_summary(risk) primary = _airi_primary_summary(risk) detail = f'{_xt(str(risk.get("id", "—")))} {_xt(_clip_words(str(risk.get("title", "")), 48))}' if primary: detail += f' | {_xt(primary)}' if reason: detail += f' | why: {_xt(_clip_words(reason, 92))}' airi_lines += f'
• Mapped theme {idx}: {detail}' gaps = airi.get("known_gaps_in_bundle", []) if gaps: gap_preview = "; ".join( f"{g.get('id', '—')} {_xt(_clip_words(str(g.get('title', '')), 22))}" for g in gaps[:3] ) airi_lines += f'
• Still unmapped here: {gap_preview}' def _summary_block(title: str, body: str, head_color: str, key: str) -> Table: tbl = Table([ [Paragraph(f'{_xt(title)}', _style(f"{key}_H", 9, 12, _WHITE, True))], [Paragraph(f'{body}', _style(f"{key}_B", 8, 11, _DGRAY))], ], colWidths=["100%"]) tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (0, 0), _hx(head_color)), ("BACKGROUND", (0, 1), (0, 1), _hx(_LGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ("LEFTPADDING", (0, 0), (-1, -1), 6), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ])) return tbl left_stack: list[Any] = [ _summary_block( "What Is Actually Present", "".join(f'• {_xt(_clip_words(line, 170))}
' for line in present_lines[:6]), _GREEN, "WIP", ), Spacer(1, 2 * mm), _summary_block( "What Is Missing Or Contradicted", "".join(f'• {_xt(_clip_words(line, 170))}
' for line in missing_lines), _RED, "WIM", ), ] left_col = Table([[item] for item in left_stack], colWidths=["100%"]) left_col.setStyle(TableStyle([ ("LEFTPADDING", (0, 0), (-1, -1), 0), ("RIGHTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 0), ("BOTTOMPADDING",(0, 0), (-1, -1), 0), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) right_stack: list[Any] = [ _summary_block( "Governance Posture", f'{_xt(score["formal_tier"])} — {_xt(score.get("use_scope", ""))}
{_xt(posture)}', posture_color, "GOV", ), ] if reg_lines: right_stack += [Spacer(1, 2 * mm), _summary_block("Regulatory Traceability", reg_lines, _NAVY, "REG")] if airi_lines: right_stack += [Spacer(1, 2 * mm), _summary_block("AIRI Risk Triggers", airi_lines, _TEAL, "AIRI")] right_tbl = Table([[item] for item in right_stack], colWidths=["100%"]) right_tbl.setStyle(TableStyle([ ("LEFTPADDING", (0, 0), (-1, -1), 0), ("RIGHTPADDING", (0, 0), (-1, -1), 0), ("TOPPADDING", (0, 0), (-1, -1), 0), ("BOTTOMPADDING",(0, 0), (-1, -1), 0), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) two_col = Table([[left_col, right_tbl]], colWidths=["48%", "52%"]) two_col.setStyle(TableStyle([ ("LEFTPADDING", (0, 0), (-1, -1), 3), ("RIGHTPADDING", (0, 0), (-1, -1), 3), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) return [two_col] def _bio_diagnostics_pdf_table(result: dict[str, Any]) -> Table | None: rows = _bio_detector_rows(result) if not rows: return None table_rows: list[list[Any]] = [[ Paragraph(f'Bio Deterministic Diagnostics', _style("BDH1", 8.5, 11, _WHITE, True)), ]] for _, label, counts in rows: status_parts = [f"{status}={counts[status]}" for status in ("detected", "not_detected", "not_applicable", "warn", "error", "absent") if counts.get(status)] table_rows.append([ Paragraph( f'{_xt(label)}
{_xt(", ".join(status_parts) if status_parts else "no findings")}', _style(f"BD_{label[:6]}", 7.5, 10, _DGRAY), ) ]) tbl = Table(table_rows, colWidths=["100%"]) tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (0, 0), _hx(_PURPLE)), ("ROWBACKGROUNDS",(0, 1), (0, -1), [_hx(_LGRAY), _hx(_WHITE)]), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ("LEFTPADDING", (0, 0), (-1, -1), 6), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ])) return tbl def _regulatory_pdf_text_lines(result: dict[str, Any]) -> list[str]: """Plain-text regulatory traceability page for the simple PDF fallback.""" basis = result.get("regulatory_basis", {}) traceability = result.get("stage_traceability", {}) if not basis and not traceability: return [] note = basis.get("note", {}) lines = ["Regulatory Traceability", ""] if note.get("body_line_1"): lines.append(note["body_line_1"]) if note.get("body_line_2"): lines.append(note["body_line_2"]) lines.append("") _stage_labels = { "stage_1": "Stage 1 - README Intent", "stage_2r": "Stage 2R - Repo Consistency", "stage_3": "Stage 3 - Code / Bio", "stage_4": "Stage 4 - Replication", "bio_diagnostics": "Bio Diagnostics", } for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"): items = traceability.get(stage_key, []) if not items: continue lines.append(_stage_labels.get(stage_key, stage_key)) for item in items: label = _REQ_LABELS.get(item["requirement_id"], item["requirement_id"]) status = _STATUS_PDF_LABEL.get(item.get("status", ""), item.get("status", "")) lines.append(f"- {label}: {status}") refs = ", ".join(item.get("finding_refs", [])) if refs: lines.append(f" triggered by: {refs}") summary = result.get("regulatory_traceability", {}).get("summary") if summary: lines.append("") lines.append(f"Summary: {summary}") return lines def _regulatory_bullets(result: dict[str, Any]) -> str: """Compact actionable bullet summary of regulatory traceability for page 1. Surfaces what actually has structural alignment (partially_aligned) versus weak signal-only references, instead of restating the boilerplate basis note. """ trace = result.get("stage_traceability", {}) items: list[dict[str, Any]] = [] seen: set[str] = set() for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"): for it in trace.get(stage_key, []): rid = it.get("requirement_id", "") if rid in seen: continue seen.add(rid) items.append(it) if not items: return "" framework_order = ["EU AI Act", "ICH M15", "IMDRF", "FDA"] frameworks: set[str] = set() for it in items: for sid in it.get("source_ids", []): if sid.startswith("eu_ai_act"): frameworks.add("EU AI Act") elif sid.startswith("ich_m15"): frameworks.add("ICH M15") elif sid.startswith("imdrf"): frameworks.add("IMDRF") elif sid.startswith("fda"): frameworks.add("FDA") fw_str = ", ".join(f for f in framework_order if f in frameworks) partial = [it for it in items if it.get("status") == "partially_aligned"] signal = [it for it in items if it.get("status") == "signal_only"] lines = [f'• Frameworks touched: {_xt(fw_str)}'] if partial: labels = "; ".join( _REQ_LABELS.get(it["requirement_id"], it["requirement_id"]) for it in partial[:3] ) lines.append(f'• Structural alignment exists ({len(partial)}): {_xt(labels)}') if signal: lines.append( f'• Signal-only references ({len(signal)}): useful for pre-audit traceability, but not strong enough to count as compliance proof' ) lines.append('• Meaning: repository evidence maps to governance frameworks, but the report does not establish compliance or clearance.') lines.append('• Why this matters: this section helps answer whether governance scaffolding is present at all before a formal audit, not whether the repository is approved for use.') return "
".join(lines) def _regulatory_basis_box(result: dict[str, Any]) -> list[Any]: basis = result.get("regulatory_basis", {}) note = basis.get("note", {}) summary = result.get("regulatory_traceability", {}).get("summary", "") if not note: return [] body_lines = [ f'{_xt(note.get("title", "Regulatory basis note"))}', f'{_xt(note.get("body_line_1", ""))}', f'{_xt(note.get("body_line_2", ""))}', ] if basis.get("review_required"): body_lines.append( f'Review required: {_xt(", ".join(basis.get("review_reasons", [])))}' ) if summary: body_lines.append( f'Traceability summary: {_xt(_clip_words(summary, 220))}' ) panel = Table( [[Paragraph("
".join(body_lines), _style("RGB_NOTE", 7.5, 9, _DGRAY))]], colWidths=["100%"], ) panel.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 5), ("LEFTPADDING", (0, 0), (-1, -1), 7), ("RIGHTPADDING", (0, 0), (-1, -1), 7), ])) return [panel] _STATUS_PDF_COLOR = { "signal_only": _AMBER, "partially_aligned": _TEAL, "aligned": _GREEN, "not_detected": _DGRAY, } _STATUS_PDF_LABEL = { "signal_only": "Signal only", "partially_aligned": "Partially aligned", "aligned": "Aligned", "not_detected": "Not detected", } def _regulatory_traceability_pdf(result: dict[str, Any]) -> list[Any]: """Detailed per-requirement regulatory traceability block for PDF pages.""" basis = result.get("regulatory_basis", {}) traceability = result.get("stage_traceability", {}) if not basis and not traceability: return [] note = basis.get("note", {}) story: list[Any] = [] story += _chapter_hdr("Chapter 3 — Regulatory Traceability", _NAVY) story.append(Paragraph( f'' f'{_xt(note.get("title", "Regulatory basis note"))}
' f'{_xt(note.get("body_line_1", ""))}
' f'{_xt(note.get("body_line_2", ""))}' f'', _style("REGT_BASIS", 8, 11, _DGRAY), )) story.append(Spacer(1, 3 * mm)) _STAGE_LABELS_PDF = { "stage_1": "Stage 1 — README Intent", "stage_2r": "Stage 2R — Repo Consistency", "stage_3": "Stage 3 — Code / Bio", "stage_4": "Stage 4 — Replication", "bio_diagnostics": "Bio Diagnostics", } _label_col_w = _PDF_CONTENT_WIDTH - 38 * mm for stage_key in ("stage_1", "stage_2r", "stage_3", "stage_4", "bio_diagnostics"): items = traceability.get(stage_key, []) if not items: continue story.append(Paragraph( f'{_xt(_STAGE_LABELS_PDF.get(stage_key, stage_key))}', _style(f"REGT_S_{stage_key[:6]}", 8.5, 11, _NAVY, True), )) story.append(Spacer(1, 1 * mm)) for item in items: req_id = item["requirement_id"] label = _REQ_LABELS.get(req_id, req_id) status = item.get("status", "") status_label = _STATUS_PDF_LABEL.get(status, status) status_color = _STATUS_PDF_COLOR.get(status, _DGRAY) refs = ", ".join(item.get("finding_refs", [])) gaps = "; ".join(item.get("not_assessed", [])) detail_lines = [f'{_xt(item["note"])}'] if refs: detail_lines.append( f'Triggered by: {_xt(refs)}' ) if gaps: detail_lines.append( f'Not assessed: {_xt(gaps)}' ) row = Table( [[ Paragraph( f'{_xt(label)}
' + "
".join(detail_lines), _style(f"REGT_B_{req_id[:10]}", 8, 11, _DGRAY), ), Paragraph( f'{_xt(status_label)}', _style(f"REGT_V_{req_id[:10]}", 7.5, 10, status_color, True, "RIGHT"), ), ]], colWidths=[_label_col_w, 38 * mm], ) row.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)), ("BOX", (0, 0), (-1, -1), 0.4, _hx(_MGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ("LEFTPADDING", (0, 0), (-1, -1), 6), ("RIGHTPADDING", (0, 0), (-1, -1), 5), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) story.append(row) story.append(Spacer(1, 1.5 * mm)) story.append(Spacer(1, 2 * mm)) summary = result.get("regulatory_traceability", {}).get("summary", "") if summary: story.append(Paragraph( f'{_xt(summary)}', _style("REGT_SUMM", 7.5, 10, _DGRAY), )) return story def _footer_block() -> list[Any]: return [Spacer(1, 2 * mm)] def _single_page_story(flowables: list[Any], *, break_before: bool = False) -> list[Any]: wrapped = KeepInFrame( _PDF_CONTENT_WIDTH, _PDF_CONTENT_HEIGHT, flowables, mode="shrink", ) return ([PageBreak()] if break_before else []) + [wrapped] # ── Detail page dispatcher ──────────────────────────────────────────────────── def _detail_pages(result: dict[str, Any], pages: int) -> list[Any]: story: list[Any] = [] story += _page2_stage_analysis(result) story += _page3_stage3_analysis(result) if pages >= 5: story += _page4_stage4_replication(result) if pages == 5: story += _page5_compact_closure(result) elif pages >= 7: story += _page4_integrity_deep(result) story += _page_regulatory_traceability(result) story += _page6_method_airi(result) story += _page7_report_metadata(result) return story # ── Shared detail helpers ───────────────────────────────────────────────────── def _sec_hdr(title: str, color: str = _NAVY) -> list[Any]: tbl = Table([[Paragraph( f'{title}', _style(f"SH_{title[:8]}", 10, 14, _WHITE, True), )]], colWidths=["100%"]) tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(color)), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 5), ("LEFTPADDING", (0, 0), (-1, -1), 8), ])) return [tbl, Spacer(1, 2 * mm)] def _chapter_hdr(title: str, color: str = _NAVY) -> list[Any]: return [ Paragraph( f'{title}', _style(f"CH_{title[:8]}", 14, 17, color, True), ), Spacer(1, 3.5 * mm), ] def _subsec_hdr(title: str, color: str = _NAVY) -> list[Any]: return [ Paragraph( f'{title}', _style(f"SUB_{title[:8]}", 10.5, 13, color, True), ), Spacer(1, 1.8 * mm), ] def _mini_score(label: str, val: int, max_val: int, col: str) -> Table: d = [ [Paragraph( f'{val}' f' / {max_val}', _style(f"MS_{label[:6]}", 20, 24, col, True, "CENTER"), )], [Paragraph(label, _style(f"ML_{label[:6]}", 7, 9, _DGRAY, False, "CENTER"))], ] t = Table(d, colWidths=[34 * mm]) t.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 3), ("LEFTPADDING", (0, 0), (-1, -1), 3), ("RIGHTPADDING", (0, 0), (-1, -1), 3), ])) return t def _rubric_rows(items: list[tuple[str, str, str, str]], id_prefix: str = "R") -> Table: """items: (name, score_str, color_hex, evidence_text)""" header = [ Paragraph(f'Check', _style(f"{id_prefix}H1", 8, 10, _WHITE, True)), Paragraph(f'Points', _style(f"{id_prefix}H2", 8, 10, _WHITE, True, "CENTER")), Paragraph(f'Evidence / Finding', _style(f"{id_prefix}H3", 8, 10, _WHITE, True)), ] rows = [header] for i, (name, score_str, col, ev) in enumerate(items): uid = f"{id_prefix}_{i}" rows.append([ Paragraph(f'{_xt(name)}', _style(f"{uid}N", 8, 11, _DGRAY, True)), Paragraph( f'{_xt(score_str)}', _style(f"{uid}S", 8, 11, col, True, "CENTER"), ), Paragraph(_xt(_clip_words(ev, 175)), _style(f"{uid}E", 7.5, 10, _DGRAY)), ]) t = Table(rows, colWidths=[52 * mm, 18 * mm, None]) t.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), _hx(_NAVY)), ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), ("TOPPADDING", (0, 0), (-1, -1), 3), ("BOTTOMPADDING", (0, 0), (-1, -1), 3), ("LEFTPADDING", (0, 0), (-1, -1), 5), ("RIGHTPADDING", (0, 0), (-1, -1), 5), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("LINEBELOW", (0, 0), (-1, 0), 0.5, _hx(_MGRAY)), ("GRID", (0, 0), (-1, -1), 0.3, _hx(_MGRAY)), ])) return t # ── Page 2: Stage 1 + Stage 2R Analysis ────────────────────────────────────── def _page2_stage_analysis(result: dict[str, Any]) -> list[Any]: story: list[Any] = [] score = result["score"] cls = result["classification"] s1 = score["stage_1_readme_intent"] s2r = score["stage_2_repo_local_consistency"] or 0 ca = cls["clinical_adjacent"] has_disc = cls["has_explicit_clinical_boundary"] readme_present = "README.md" in result.get("file_hashes_sha256", {}) story += _chapter_hdr("Chapter 1 — Stage Scorecard and Governance Scoring", _NAVY) # ── Stage 1 ────────────────────────────────────────────────────────────── story += _sec_hdr("Stage 1 — README Evidence Signal | Weight: 0.40", _TEAL) s1_rubric = result.get("stage_1_rubric", {}) s1_order = [ ("baseline", "Baseline"), ("S1_missing_readme", "README present"), ("S1_domain_readme", "BIO/medical terms in README"), ("S1_domain_package", "BIO/medical terms in package"), ("H1_clinical_certainty_hype", "H1: Clinical Certainty Hype"), ("H2_regulatory_approval_hype", "H2: Regulatory Approval Hype"), ("H3_autonomous_replacement_hype", "H3: Autonomous Replacement Hype"), ("H4_breakthrough_marketing_hype", "H4: Marketing Hype"), ("H5_universal_generalization_hype", "H5: Universal Generalization"), ("H6_perfect_accuracy_hype", "H6: Perfect Accuracy Claim"), ("R1_limitations_section", "R1: Limitations Section"), ("R2_regulatory_framework", "R2: Regulatory Framework"), ("R3_clinical_disclaimer", "R3: Clinical Boundary"), ("R4_demographic_bias_boundary", "R4: Bias / Subgroup Boundary"), ("R5_reproducibility_provisions", "R5: Reproducibility Provisions"), ] s1_items: list[tuple[str, str, str, str]] = [] for key, label in s1_order: item = s1_rubric.get(key) if not item: continue pts = item.get("score", 0) col = _RED if pts < 0 else _GREEN if pts > 0 else _DGRAY evidence = item.get("evidence", "") if key == "R2_regulatory_framework": evidence = ( f"{evidence} " "[partial-credit ladder: +15 strong framework | +5 weak self-asserted compliance | " "-5 CA-INDIRECT missing framework | -10 CA-DIRECT missing framework]" ) s1_items.append((label, f"{pts:+d}", col, evidence)) if not s1_items: s1_items = [ ("Baseline", "+60", _DGRAY, "All non-nascent repositories start at 60."), ("README present", "+0" if readme_present else "-20", _GREEN if readme_present else _RED, "README.md detected in repository root." if readme_present else "No README found — major deduction applied."), ] chip1 = _mini_score("S1 Score", s1, 100, _TEAL) tbl1 = _rubric_rows(s1_items, "S1") combined1 = Table([[chip1, tbl1]], colWidths=[38 * mm, None]) combined1.setStyle(TableStyle([ ("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 2), ("RIGHTPADDING", (0, 0), (-1, -1), 2), ])) story.append(combined1) # Classification info bar ca_col = _ORANGE if ca else _GREEN disc_col = _GREEN if has_disc else _RED t0_col = _RED if cls["t0_hard_floor"] else _GREEN info_text = ( f'• Clinical-Adjacent: {"YES" if ca else "NO"}' f' ({_xt(cls["ca_severity"])})    ' f'• Explicit Disclaimer: {"PRESENT" if has_disc else "ABSENT"}' f'    ' f'• T0 Hard Floor: {"TRIGGERED" if cls["t0_hard_floor"] else "Clear"}' ) info_tbl = Table([[Paragraph(info_text, _style("INF1", 8, 11, _DGRAY, False, "CENTER"))]], colWidths=["100%"]) info_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 5), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("ALIGN", (0, 0), (-1, -1), "CENTER"), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ])) story.append(Spacer(1, 2 * mm)) story.append(info_tbl) # ── Stage 2R ───────────────────────────────────────────────────────────── story.append(Spacer(1, 5 * mm)) story += _sec_hdr("Stage 2R — Repo-Local Consistency | Weight: 0.20", _PURPLE) rubric = result.get("stage_2r_rubric", {}) verdict = str(rubric.get("verdict", "")) _label_map = { "baseline": "Baseline", "R2R_1_readme_package_code_alignment": "R2R-1: README / Package Alignment", "R2R_2_readme_docs_alignment": "R2R-2: README / Docs Alignment", "R2R_3_readme_test_ci_alignment": "R2R-3: README / Test-CI Alignment", "R2R_4_limitation_repetition": "R2R-4: Limitation Repetition", "R2R_D1_internal_clinical_boundary_contradiction": "R2R-D1: Internal Clinical Boundary Contradiction (PENALTY)", "R2R_D2_missing_clinical_use_boundary": "R2R-D2: Missing Clinical Boundary (PENALTY)", "R2R_D3_stale_metadata": "R2R-D3: Stale Metadata (PENALTY)", "R2R_D4_unsupported_workflow_claim": "R2R-D4: Unsupported Workflow Claim (PENALTY)", } _ev_tooltip = { "baseline": "Every repository that is not nascent starts at 60. " "This baseline accounts for basic structural maturity.", "R2R_1_readme_package_code_alignment": "README and package metadata share bio-domain vocabulary, " "indicating claim-to-implementation alignment.", "R2R_2_readme_docs_alignment": "README and docs/ share domain vocabulary, " "indicating consistent external communication.", "R2R_3_readme_test_ci_alignment": "Test and CI surfaces are present and reference the same " "domain as the README.", "R2R_4_limitation_repetition": "Limitation or validation-boundary language repeats across more than one repository surface.", "R2R_D1_internal_clinical_boundary_contradiction": "A non-clinical boundary is declared, but clinical deployment/support claims still appear elsewhere.", "R2R_D2_missing_clinical_use_boundary": "Clinical-adjacent repository lacks an explicit " "'research use only' or 'not for diagnostic use' boundary — high review risk.", "R2R_D3_stale_metadata": "Version metadata appears inconsistent across README and package surfaces.", "R2R_D4_unsupported_workflow_claim": "README or docs describe runnable workflow support that local tests, workflows, or entrypoints do not substantiate.", } s2r_items: list[tuple[str, str, str, str]] = [] for key in ("baseline", "R2R_1_readme_package_code_alignment", "R2R_2_readme_docs_alignment", "R2R_3_readme_test_ci_alignment", "R2R_4_limitation_repetition", "R2R_D1_internal_clinical_boundary_contradiction", "R2R_D2_missing_clinical_use_boundary", "R2R_D3_stale_metadata", "R2R_D4_unsupported_workflow_claim"): item = rubric.get(key) if item is None or not isinstance(item, dict): continue sc = item.get("score", 0) ev_raw = item.get("evidence", "") ev_ext = _ev_tooltip.get(key, "") trace = _rubric_trace_suffix(item).strip(" `") combined_ev = f"{ev_raw} — {ev_ext}" if ev_ext else ev_raw if trace: combined_ev = f"{combined_ev} — {trace}" if key == "baseline": col = _DGRAY sc_str = f"+{sc}" elif key.startswith("R2R_D"): col = _RED if sc < 0 else _DGRAY sc_str = str(sc) else: col = _TEAL if sc > 0 else _DGRAY sc_str = f"+{sc}" if sc > 0 else "0 (not detected)" s2r_items.append((_label_map.get(key, key), sc_str, col, combined_ev)) chip2 = _mini_score("S2R Score", s2r, 100, _PURPLE) tbl2 = _rubric_rows(s2r_items, "S2R") combined2 = Table([[chip2, tbl2]], colWidths=[38 * mm, None]) combined2.setStyle(TableStyle([ ("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 2), ("RIGHTPADDING", (0, 0), (-1, -1), 2), ])) story.append(combined2) story.append(Spacer(1, 1 * mm)) verdict_col = _GREEN if "Strong" in verdict else (_AMBER if "Mixed" in verdict else _RED) verdict_text = ( f'• Consistency Verdict: {_xt(verdict)}' ) verdict_tbl = Table([[Paragraph(verdict_text, _style("S2RVERDICT", 8, 11, _DGRAY, False, "CENTER"))]], colWidths=["100%"]) verdict_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx(_LGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 5), ("BOTTOMPADDING", (0, 0), (-1, -1), 5), ("LEFTPADDING", (0, 0), (-1, -1), 8), ("RIGHTPADDING", (0, 0), (-1, -1), 8), ("ALIGN", (0, 0), (-1, -1), "CENTER"), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ])) story.append(verdict_tbl) story += _footer_block() return _single_page_story(story, break_before=True) # ── Page 3: Stage 3 Full Breakdown ─────────────────────────────────────────── def _page3_stage3_analysis(result: dict[str, Any]) -> list[Any]: story: list[Any] = [] score = result["score"] s3 = score["stage_3_code_bio"] rubric = result.get("stage_3_rubric", {}) story += _sec_hdr("Stage 3 — Code & Bio Responsibility | Weight: 0.40", _SLATE) _ev_ext = { "T1_CI_CD": "CI/CD workflows (GitHub Actions, GitLab CI, CircleCI) verify " "that commits do not silently break the pipeline. Full credit (15) requires workflow files present.", "T2_domain_tests": "Domain-specific tests verify biological outputs — e.g., " "sequencing pipeline correctness, variant call validation, or genomic data integrity. " "Full credit (15) requires BIO-term presence in test files. Partial (8) if tests exist but are generic.", "T3_changelog_release_hygiene": "A CHANGELOG tracks which version fixed which defect — " "essential for regulatory traceability and reproducibility audits. " "CHANGELOG.md, CHANGELOG, or NEWS.md all qualify.", "B1_data_provenance_controls": "Dependency manifests " "(requirements.txt, pyproject.toml, environment.yml) establish reproducibility context. " "Score 10 if manifest detected; max 15 requires data-source, dataset-citation, or IRB language.", "B2_bias_limitations": "Documentation of algorithmic bias, limitations, " "or model boundary conditions. Score 8 for boundary language; max 15 requires " "measurement evidence such as subgroup analysis, calibration, or test coverage.", "B3_coi_funding": "Conflict of interest and funding disclosure in README or FUNDING.md. " "Required for institutional review context and detected by local text scan.", } t_items: list[tuple[str, str, str, str]] = [] for key, label in [ ("T1_CI_CD", "T1: CI/CD Workflow"), ("T2_domain_tests", "T2: Domain-Specific Tests"), ("T3_changelog_release_hygiene", "T3: Changelog & Release Hygiene"), ]: item = rubric.get(key, {}) sc = item.get("score", 0) mx = item.get("max", 15) ev = item.get("evidence", "") ext = _ev_ext.get(key, "") trace = _rubric_trace_suffix(item).strip(" `") col = _GREEN if sc == mx else (_AMBER if sc > 0 else _RED) combined = f"{ev} — {ext}" if ext else ev if trace: combined = f"{combined} — {trace}" t_items.append((label, f"{sc} / {mx}", col, combined)) b_items: list[tuple[str, str, str, str]] = [] for key, label in [ ("B1_data_provenance_controls", "B1: Data Provenance Controls"), ("B2_bias_limitations", "B2: Bias / Limitations Documentation"), ("B3_coi_funding", "B3: COI & Funding Disclosure"), ]: item = rubric.get(key, {}) sc = item.get("score", 0) mx = item.get("max", 15) ev = item.get("evidence", "") ext = _ev_ext.get(key, "") trace = _rubric_trace_suffix(item).strip(" `") not_detectable = "local CLI scan" in ev col = (_GREEN if sc == mx else (_AMBER if sc > 0 else (_DGRAY if not_detectable else _RED))) note = " [Manual review required]" if not_detectable else "" combined = f"{ev} — {ext}" if ext else ev if trace: combined = f"{combined} — {trace}" b_items.append((label, f"{sc} / {mx}{note}", col, combined)) chip3 = _mini_score("S3 Score", s3, 100, _SLATE) body_items: list[Any] = [ Paragraph( f'Engineering Accountability (T-series)', _style("TS1", 8.5, 12, _SLATE, True), ), Spacer(1, 1 * mm), _rubric_rows(t_items, "T"), Spacer(1, 3 * mm), Paragraph( f'Biological Integrity (B-series)', _style("BS1", 8.5, 12, _SLATE, True), ), Spacer(1, 1 * mm), _rubric_rows(b_items, "B"), ] main_row = Table([[chip3, body_items]], colWidths=[38 * mm, None]) main_row.setStyle(TableStyle([ ("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 2), ("RIGHTPADDING", (0, 0), (-1, -1), 2), ])) story.append(main_row) raw_entry = rubric.get("stage_3_raw_total", {}) raw_score = raw_entry.get("score") raw_max = raw_entry.get("max") if raw_score is not None and raw_max: t_total = sum(rubric.get(k, {}).get("score", 0) for k in ["T1_CI_CD", "T2_domain_tests", "T3_changelog_release_hygiene"]) b_total = sum(rubric.get(k, {}).get("score", 0) for k in ["B1_data_provenance_controls", "B2_bias_limitations", "B3_coi_funding"]) story.append(Spacer(1, 2 * mm)) story.append(Paragraph( f'Normalized score: ' f'T-series {t_total}/45 + B-series {b_total}/35 = raw {raw_score}/{raw_max} → {s3}/100.', _style("S3NORM", 8, 11, _DGRAY), )) # Gap analysis story.append(Spacer(1, 5 * mm)) story += _sec_hdr("Stage 3 Gap Analysis — Path to Next Tier", _DGRAY) local_max = 55 fs = score["final_score"] gap_t3 = max(0, 70 - fs) gap_t4 = max(0, 85 - fs) t_total = sum(rubric.get(k, {}).get("score", 0) for k in ["T1_CI_CD", "T2_domain_tests", "T3_changelog_release_hygiene"]) b_total = sum(rubric.get(k, {}).get("score", 0) for k in ["B1_data_provenance_controls", "B2_bias_limitations", "B3_coi_funding"]) gap_rows = [ ("T-series vs B-series", f"T-series (engineering) attained: {t_total} / 45 | B-series (bio integrity) attained: {b_total} / 35"), ("Local CLI scan maximum", f"{local_max} / 100 (T1+T2+T3 max 15 each; B1 max 10; B2/B3 require manual review)"), ("Gap to T3", f"{gap_t3} points needed across all stages to reach final score >= 70"), ("Gap to T4", f"{gap_t4} points needed across all stages to reach final score >= 85"), ("B2 Bias/Limitations", "Not detectable — requires manual audit of README, model card, or supplementary documentation for validation boundaries and algorithmic limitations"), ("B3 COI/Funding", "Not detectable — requires inspection of README or FUNDING.md for conflict of interest and funding source disclosure"), ] gap_table_rows: list[list[Any]] = [[ Paragraph(f'Stage 3 gap interpretation', _style("S3GAPH", 8.5, 11, _WHITE, True)), "" ]] for label, detail in gap_rows: gap_table_rows.append([ Paragraph(f'{_xt(label)}', _style(f"S3GL_{label[:6]}", 8, 10, _SLATE, True)), Paragraph(f'{_xt(detail)}', _style(f"S3GD_{label[:6]}", 8, 11, _DGRAY)), ]) gap_table = Table(gap_table_rows, colWidths=[45 * mm, None]) gap_table.setStyle(TableStyle([ ("SPAN", (0, 0), (1, 0)), ("BACKGROUND", (0, 0), (1, 0), _hx(_SLATE)), ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ("LEFTPADDING", (0, 0), (-1, -1), 6), ("RIGHTPADDING", (0, 0), (-1, -1), 6), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("GRID", (0, 1), (-1, -1), 0.3, _hx(_MGRAY)), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) story.append(gap_table) story += _footer_block() return _single_page_story(story, break_before=True) # ── Page 4: Stage 4 Replication Deep Dive (5p/7p) ──────────────────────────── def _page4_stage4_replication(result: dict[str, Any]) -> list[Any]: story: list[Any] = [] score = result["score"] stage4_score = result.get("replication_score", 0) stage4_tier = result.get("replication_tier", "R0") rubric = result.get("stage_4_rubric", {}) story += _sec_hdr("Stage 4 — Replication Evidence Lane | Separate lane", _GREEN) label_map = { "S4_container_environment": "S4: Container / Runtime Environment", "S4_make_reproduce_target": "S4: Reproduce Target", "S4_environment_lock_evidence": "S4: Environment Lock Evidence", "S4_exact_dependency_pins_or_hashes": "S4: Exact Dependency Pins / Hashes", "S4_readme_reproducibility_section": "S4: Reproducibility Section", "S4_checksum_files": "S4: Checksums / Integrity Files", "S4_dataset_url": "S4: Dataset / Data Source URL", "S4_model_weight_url_or_checksum": "S4: Model Artifact URL / Checksum", "S4_citation_cff": "S4: CITATION.cff", "S4_license_restriction": "S4: License / Use Restriction", "S4_cli_entrypoint": "S4: CLI Entrypoint", "S4_seed_setting": "S4: Deterministic Seed Setting", "S4_runnable_examples": "S4: Runnable Examples", } items: list[tuple[str, str, str, str]] = [] for key, item in rubric.items(): if not isinstance(item, dict) or "score" not in item or "max" not in item: continue sc = item.get("score", 0) mx = item.get("max", 0) color = _GREEN if sc == mx and mx else (_AMBER if sc > 0 else _RED) evidence = item.get("evidence", "") items.append((label_map.get(key, key), f"{sc} / {mx}", color, evidence)) chip = _mini_score("S4 Score", stage4_score, 100, _GREEN) tbl = _rubric_rows(items, "S4") combined = Table([[chip, tbl]], colWidths=[38 * mm, None]) combined.setStyle(TableStyle([ ("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 2), ("RIGHTPADDING", (0, 0), (-1, -1), 2), ])) story.append(combined) story.append(Spacer(1, 2 * mm)) story.append(Paragraph( f'Replication tier: {stage4_tier}. ' 'Stage 4 is reported separately and does not alter the formal score.', _style("S4_NOTE", 7.5, 10, _DGRAY), )) story.append(Spacer(1, 3 * mm)) story += _sec_hdr("Formal Score Effect", _NAVY) final_box = Table([[ Paragraph( f'Final Score: {score["final_score"]} / 100 ({_xt(score["formal_tier"])})
' 'Formal effect: Stage 4 does not raise or lower the formal repository score.
' '• This page exists to show reproducibility and operational evidence posture separately from the governance score.
' '• Stronger replication evidence improves trust in reproducibility posture, not the formal governance tier.', _style("S4_SCOPE", 8, 11, _DGRAY), ) ]], colWidths=["100%"]) final_box.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, -1), _hx("#FFF7E3")), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("TOPPADDING", (0, 0), (-1, -1), 6), ("BOTTOMPADDING", (0, 0), (-1, -1), 6), ("LEFTPADDING", (0, 0), (-1, -1), 7), ("RIGHTPADDING", (0, 0), (-1, -1), 7), ])) story.append(final_box) story += _footer_block() return _single_page_story(story, break_before=True) # ── Page 5: Code Integrity Deep Dive + Classification (7p only) ────────────── def _page4_integrity_deep(result: dict[str, Any]) -> list[Any]: story: list[Any] = [] ci = result["code_integrity"] cls = result["classification"] hashes = result.get("file_hashes_sha256", {}) tgt = result["target"] ast_note = _ast_scope_note(result) story += _chapter_hdr("Chapter 2 — Code Integrity Deep Analysis", _NAVY) _remediation = { "C1_hardcoded_credentials": "CRITICAL: Rotate all exposed credentials immediately. Remove from git history " "using git-filter-repo. Use environment variables or a secrets manager (AWS Secrets Manager, " "HashiCorp Vault, Azure Key Vault). Add a pre-commit hook with detect-secrets.", "C2_dependency_pinning": "Pin all dependencies to exact versions (== for pip, hash-pinning for conda). " "Run pip-audit or safety regularly. Consider pip-compile for reproducible lock files. " "Unpinned ranges in clinical-adjacent pipelines create silent regression risk.", "C3_dead_or_deprecated_patient_adjacent_paths": "Audit deprecated/ directories for patient-adjacent metadata patterns. " "If clinical data was processed, verify data destruction or anonymization logs. " "Dead code with patient metadata patterns must be purged or explicitly annotated as test fixtures.", "C4_exception_handling_clinical_adjacent_paths": "Replace broad 'except Exception: pass' or 'except: return True' patterns with " "specific error types and explicit failure logging. In clinical-adjacent code paths, " "any silent failure is a patient safety risk. Fail closed, not open.", "C5_compliance_boundary_integrity": "Treat privacy, legal, or clinical-adjacent claims as governance obligations. " "If README or product text invokes HIPAA, compliance, or self-hosted clinical safety, " "surface supporting controls, operating boundaries, and deployment constraints explicitly.", "C6_mock_auth_or_fail_open_boundary": "Do not present self-host, local-mode, or privacy-sensitive flows as production-like if they rely on " "mock authentication, auto-login, or no-auth convenience boundaries. Separate demo convenience from trust posture.", } _desc = { "C1_hardcoded_credentials": "Scans for AWS access keys (AKIA*), OpenAI keys (sk-*), GitHub tokens (ghp_*), " "and api_key = '...' patterns in all text files.", "C2_dependency_pinning": "Checks whether requirements.txt / pyproject.toml / environment.yml use " "exact version pins (==, sha256 hash) or loose ranges (>=, no pin).", "C3_dead_or_deprecated_patient_adjacent_paths": "Scans deprecated/ directories for patient metadata patterns: " "patient_id, patient_age, patient_sex, sample_id, collection_date, lab_id, etc.", "C4_exception_handling_clinical_adjacent_paths": "Detects fail-open exception patterns: 'except Exception: pass' or " "'except: return True' in code — these silently ignore errors that could corrupt clinical outputs.", "C5_compliance_boundary_integrity": "Detects unsupported legal/compliance claims or clinical-boundary weaknesses in reviewed repository " "sources, including self-asserted privacy/compliance language without visible governance grounding.", "C6_mock_auth_or_fail_open_boundary": "Detects mock-auth, auto-login, or no-auth local/self-host boundary patterns in README, config, and code " "when trust-boundary language suggests a stronger operational posture than the reviewed sources support.", } short = { "C1_hardcoded_credentials": "C1: Hardcoded Credentials", "C2_dependency_pinning": "C2: Dependency Pinning", "C3_dead_or_deprecated_patient_adjacent_paths": "C3: Deprecated Patient Paths", "C4_exception_handling_clinical_adjacent_paths": "C4: Fail-Open Exceptions", "C5_compliance_boundary_integrity": "C5: Compliance Boundary Integrity", "C6_mock_auth_or_fail_open_boundary": "C6: Mock Auth / Fail-Open Boundary", } ci_items: list[tuple[str, str, str, str]] = [] for key, cfg in ci.items(): s = cfg["status"] col = _status_hex(s) ev_raw = cfg["evidence"][0] if cfg["evidence"] else "" ev_full = f"{ev_raw} | Scan: {_desc.get(key, '')}" ci_items.append((short.get(key, key), s, col, ev_full)) story.append(_rubric_rows(ci_items, "CI")) story.append(Spacer(1, 3 * mm)) # Remediation guidance fail_warn = [(k, v) for k, v in ci.items() if v["status"] != "PASS"] if fail_warn: story += _sec_hdr("Remediation Guidance", _RED) for key, v in fail_warn: s = v["status"] col = _RED if s == "FAIL" else _AMBER guidance = _remediation.get(key, "Review and remediate before clinical-adjacent deployment.") story.append(Paragraph( f'[{s}] {_xt(short.get(key, key))}:', _style(f"RG_H_{key[:4]}", 8.5, 12, col, True), )) story.append(Paragraph( f'→ {_xt(guidance)}', _style(f"RG_B_{key[:4]}", 8, 11, _DGRAY), )) story.append(Spacer(1, 2 * mm)) else: story.append(Paragraph( f'All code integrity checks PASSED. ' 'Continue monitoring with each major release.', _style("CIOK", 8, 11, _GREEN), )) # Classification analysis story.append(Spacer(1, 3 * mm)) story += _sec_hdr("Classification & Repository Analysis", _SLATE) cls_items: list[tuple[str, str, str, str]] = [ ("Clinical Adjacent", "YES" if cls["clinical_adjacent"] else "NO", _ORANGE if cls["clinical_adjacent"] else _GREEN, f'Severity: {cls["ca_severity"]}. ' 'Triggered by BIO/CLINICAL_OUTPUT term regex match across README, docs, and code.'), ("T0 Hard Floor", "TRIGGERED" if cls["t0_hard_floor"] else "Clear", _RED if cls["t0_hard_floor"] else _GREEN, "Score forced to 0 regardless of rubric performance — e.g., AGI claim in clinical context." if cls["t0_hard_floor"] else "No T0_HARD_FLOOR condition detected."), ("Explicit Disclaimer", "PRESENT" if cls["has_explicit_clinical_boundary"] else "ABSENT", _GREEN if cls["has_explicit_clinical_boundary"] else _AMBER, "Regex: 'not for clinical|not for diagnostic|research use only|not medical advice' " "in README + docs surface." if cls["has_explicit_clinical_boundary"] else "Disclaimer pattern not found in README or docs. High impact on Stage 1 and Stage 2R scores."), ("Files Scanned", str(tgt.get("file_count", "—")), _TEAL, "Total files indexed by recursive walk. Text files only for content analysis; " "binary files counted but not read."), ("Execution Mode", result.get("execution_mode", "—"), _DGRAY, "No LLM calls. No network access. No runtime execution. " "Deterministic regex + file-system scan only."), ] if ast_note: cls_items.append(( "AST Analysis Scope", "CAPPED", _AMBER, ast_note, )) story.append(_rubric_rows(cls_items, "CLS")) # File hashes if hashes: story.append(Spacer(1, 3 * mm)) story.append(Paragraph( f'File Integrity (SHA-256)', _style("FIH1", 9, 12, _NAVY, True), )) story.append(Spacer(1, 1 * mm)) hash_rows: list[list[Any]] = [[ Paragraph(f'File', _style("FHH1", 8, 10, _WHITE, True)), Paragraph(f'SHA-256 Hash', _style("FHH2", 8, 10, _WHITE, True)), ]] for fname, h in hashes.items(): hash_rows.append([ Paragraph(_xt(fname), _style(f"FN_{fname[:4]}", 8, 11, _DGRAY)), Paragraph(f'{h}', _style(f"FV_{fname[:4]}", 6.5, 8, _DGRAY)), ]) hash_tbl = Table(hash_rows, colWidths=[40 * mm, None]) hash_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), _hx(_NAVY)), ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), ("TOPPADDING", (0, 0), (-1, -1), 3), ("BOTTOMPADDING", (0, 0), (-1, -1), 3), ("LEFTPADDING", (0, 0), (-1, -1), 5), ("RIGHTPADDING", (0, 0), (-1, -1), 5), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("GRID", (0, 0), (-1, -1), 0.3, _hx(_MGRAY)), ])) story.append(hash_tbl) story += _footer_block() return _single_page_story(story, break_before=True) # ── Page 5: Compact Closeout (5p standard packet only) ─────────────────────── def _page5_compact_closure(result: dict[str, Any]) -> list[Any]: story: list[Any] = [] score = result["score"] risks = result.get("notable_risks", []) airi = result.get("airi_risk_coverage", {}) ci = result["code_integrity"] story += _sec_hdr("Closeout Summary", _NAVY) ci_rows: list[tuple[str, str, str, str]] = [] short = { "C1_hardcoded_credentials": "C1: Hardcoded Credentials", "C2_dependency_pinning": "C2: Dependency Pinning", "C3_dead_or_deprecated_patient_adjacent_paths": "C3: Deprecated Patient Paths", "C4_exception_handling_clinical_adjacent_paths": "C4: Fail-Open Exceptions", "C5_compliance_boundary_integrity": "C5: Compliance Boundary Integrity", "C6_mock_auth_or_fail_open_boundary": "C6: Mock Auth / Fail-Open Boundary", } for key, cfg in ci.items(): status = cfg["status"] color = _status_hex(status) evidence = cfg["evidence"][0] if cfg.get("evidence") else "" ci_rows.append((short.get(key, key), status, color, evidence)) story.append(_rubric_rows(ci_rows, "CC")) story.append(Spacer(1, 3 * mm)) if risks: story += _sec_hdr("Top Risks", _RED) for risk in risks[:4]: story.append(Paragraph( f'• {_xt(risk)}', _style(f"TR_{risk[:6]}", 8, 11, _DGRAY), )) if airi: story.append(Spacer(1, 3 * mm)) story += _sec_hdr("AIRI Risk Triggers Summary", _TEAL) story.append(Paragraph( f'' f'Covered Risks: {airi.get("covered_count", 0)} / {airi.get("total_risks_in_detector_scope", 0)} ' f'| Coverage Rate: {airi.get("coverage_rate", 0):.3f}', _style("AIRI_COMPACT", 8, 11, _DGRAY), )) story.append(Paragraph( f'{_xt(_surface_compaction_note(result))}', _style("AIRI_COMPACT_NOTE", 7, 10, _DGRAY), )) for risk in airi.get("covered_risks", [])[:3]: reason = _airi_reason_summary(risk) story.append(Paragraph( f'• {_xt(str(risk.get("id", "—")))} ' f'{_xt(str(risk.get("title", "")))}' f'{f" — why: {_xt(reason)}" if reason else ""}', _style(f"AIRC_{str(risk.get('id', 'risk'))[:8]}", 8, 10, _DGRAY), )) if airi.get("known_gaps_in_bundle"): _all_gaps = airi.get("known_gaps_in_bundle", []) gap_preview = ", ".join( f"{g.get('id', '—')} {_xt(str(g.get('title', '')))}" for g in _all_gaps[:5] ) _gap_extra = f" (+{len(_all_gaps) - 5} more)" if len(_all_gaps) > 5 else "" story.append(Paragraph( f'Known gaps: {gap_preview}{_gap_extra}', _style("AIRI_GAPC", 8, 10, _DGRAY), )) story.append(Spacer(1, 3 * mm)) story += _regulatory_basis_box(result) story.append(Spacer(1, 3 * mm)) story += _sec_hdr("Method Boundary", _DGRAY) story.append(Paragraph( f'Final score: {score["final_score"]} / 100 ({_xt(score["formal_tier"])})' f'
Method: {_xt(result.get("method", ""))}', _style("MBCOMPACT", 8, 11, _DGRAY), )) story += _footer_block() return _single_page_story(story, break_before=True) # ── Page 6: Remediation + AIRI + Method (7p only) ──────────────────────────── def _page6_method_airi(result: dict[str, Any]) -> list[Any]: story: list[Any] = [] risks = result.get("notable_risks", []) airi = result.get("airi_risk_coverage", {}) score = result.get("score", {}) stage2_score = int(score.get("stage_2_repo_local_consistency", 0) or 0) stage3_score = int(score.get("stage_3_code_bio", 0) or 0) story += _chapter_hdr("Chapter 4 — Remediation Actions, AIRI Risk Triggers & Method Boundary", _NAVY) story += _subsec_hdr("4.1 Decision Path", _SLATE) decision_rows = [ ("Stage 2R score", f"{stage2_score} / 100 — repo-local contradictions and missing trust boundaries are holding the report down."), ("Stage 3 score", f"{stage3_score} / 100 — accountability and bio-governance evidence remain partial rather than mature."), ] decision_table_rows: list[list[Any]] = [[ Paragraph(f'Developer-facing decision path', _style("DPDF_H", 8.5, 11, _WHITE, True)), "" ]] for label, detail in decision_rows: decision_table_rows.append([ Paragraph(f'{_xt(label)}', _style(f"DPDF_K_{label[:4]}", 8, 10, _SLATE, True)), Paragraph(f'{_xt(detail)}', _style(f"DPDF_V_{label[:4]}", 8, 11, _DGRAY)), ]) decision_table = Table(decision_table_rows, colWidths=[40 * mm, None]) decision_table.setStyle(TableStyle([ ("SPAN", (0, 0), (1, 0)), ("BACKGROUND", (0, 0), (1, 0), _hx(_SLATE)), ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ("LEFTPADDING", (0, 0), (-1, -1), 6), ("RIGHTPADDING", (0, 0), (-1, -1), 6), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("GRID", (0, 1), (-1, -1), 0.3, _hx(_MGRAY)), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) story.append(decision_table) story.append(Spacer(1, 4 * mm)) story += _subsec_hdr("4.2 Top Remediation Actions", _RED) _pri_detail: dict[str, str] = { "Clinical-adjacent surfaces exist without an explicit non-diagnostic/non-clinical boundary.": "Add a prominent 'Research Use Only — Not for Clinical or Diagnostic Use' disclaimer " "to README H1 or H2 section. Reference applicable frameworks: FDA SaMD guidance, " "EU AI Act Article 6, or IRB oversight requirements for your deployment context.", "C1_hardcoded_credentials: FAIL": "CRITICAL: Rotate all exposed credentials immediately. Remove from git history " "using git-filter-repo. Implement pre-commit secrets detection. " "Use environment variables or a secrets manager for all future credential handling.", "C2_dependency_pinning: WARN": "Pin all production dependencies to exact versions (== for pip). " "Add pip-audit or safety to CI pipeline for vulnerability scanning. " "Consider pip-compile for deterministic lock files.", "C3_dead_or_deprecated_patient_adjacent_paths: WARN": "Audit deprecated/ directories for patient-adjacent metadata patterns. " "If clinical data was processed historically, verify destruction or anonymization logs. " "If patterns are from test fixtures, annotate clearly with # noqa comments.", "C4_exception_handling_clinical_adjacent_paths: WARN": "Replace broad exception handlers with specific error types and explicit logging. " "In any clinical-adjacent code path: fail closed, not open. " "Never silently return True or pass on exception.", "C5_compliance_boundary_integrity: WARN": "Do not rely on unsupported legal, privacy, or clinical-boundary claims. " "Add explicit deployment boundaries, governance controls, and operational evidence before using such language.", "C6_mock_auth_or_fail_open_boundary: WARN": "Do not treat mock-auth, auto-login, or no-auth self-host flows as production-ready trust boundaries. " "Separate convenience development paths from privacy, security, and compliance posture claims.", } no_major_risks = not risks or risks == ["No major local risks detected by the CLI scan."] if no_major_risks: story.append(Paragraph( f'No critical risks detected by local CLI scan.
' f'A manual audit is still recommended for ' 'clinical-adjacent deployment. Local CLI cannot assess B2 (bias) or B3 (COI).', _style("NR1", 8, 12, _DGRAY), )) else: remediation_rows: list[list[Any]] = [[ Paragraph(f'Priority', _style("REMD_H1", 8.5, 11, _WHITE, True)), Paragraph(f'What to fix first', _style("REMD_H2", 8.5, 11, _WHITE, True)), ]] for i, risk in enumerate(risks[:4], 1): guidance = _pri_detail.get(risk, ( "Review this finding and implement appropriate controls before supervised or clinical-adjacent deployment." )) remediation_rows.append([ Paragraph( f'P{i}', _style(f"REMD_P{i}", 8, 11, _DGRAY, True, "CENTER"), ), Paragraph( f'{_xt(_clip_words(risk, 95))}
{_xt(_clip_words(guidance, 145))}', _style(f"REMD_V{i}", 7.8, 10, _DGRAY), ), ]) remediation_table = Table(remediation_rows, colWidths=[16 * mm, None]) remediation_table.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), _hx(_RED)), ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ("LEFTPADDING", (0, 0), (-1, -1), 6), ("RIGHTPADDING", (0, 0), (-1, -1), 6), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("GRID", (0, 0), (-1, -1), 0.3, _hx(_MGRAY)), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) story.append(remediation_table) # AIRI summary if airi: story.append(Spacer(1, 4 * mm)) story += _subsec_hdr("4.3 AIRI Risk Triggers Summary", _TEAL) airi_rows: list[list[Any]] = [[ Paragraph(f'AIRI summary', _style("AIRIPDF_H", 8.5, 11, _WHITE, True)), "" ]] airi_rows.append([ Paragraph(f'Coverage', _style("AIRIPDF_K0", 8, 10, _TEAL, True)), Paragraph( f'Covered Risks: {airi.get("covered_count", 0)} / {airi.get("total_risks_in_detector_scope", 0)} ' f'| Coverage Rate: {airi.get("coverage_rate", 0):.3f} ' f'| Bundle Scope: {_xt(str(airi.get("airi_bundle_scope", "unknown")))}', _style("AIRIPDF_V0", 8, 11, _DGRAY), ), ]) airi_rows.append([ Paragraph(f'Meaning', _style("AIRIPDF_KM0", 8, 10, _TEAL, True)), Paragraph( f'AIRI broadens local governance findings into a secondary risk vocabulary. ' f'It does not prove harm, compliance, or deployment readiness.', _style("AIRIPDF_VM0", 8, 11, _DGRAY), ), ]) covered_risks = airi.get("covered_risks", []) for idx, risk in enumerate(covered_risks[:2], start=1): reason = _airi_reason_summary(risk) primary = _airi_primary_summary(risk) detail = f'{_xt(str(risk.get("id", "—")))} {_xt(_clip_words(str(risk.get("title", "")), 44))}' if primary: detail += f' | {_xt(primary)}' if reason: detail += f' | why: {_xt(_clip_words(reason, 80))}' airi_rows.append([ Paragraph(f'Mapped theme {idx}', _style(f"AIRIPDF_KM{idx}", 8, 10, _TEAL, True)), Paragraph(f'{detail}', _style(f"AIRIPDF_VM{idx}", 8, 11, _DGRAY)), ]) gaps = airi.get("known_gaps_in_bundle", []) if gaps: gap_preview = ", ".join( f"{g.get('id', '—')} {_xt(_clip_words(str(g.get('title', '')), 18))}" for g in gaps[:3] ) _gap_extra = f" (+{len(gaps) - 3} more)" if len(gaps) > 3 else "" airi_rows.append([ Paragraph(f'Known gaps', _style("AIRIPDF_KG", 8, 10, _TEAL, True)), Paragraph(f'{gap_preview}{_gap_extra}', _style("AIRIPDF_VG", 8, 11, _DGRAY)), ]) airi_rows.append([ Paragraph(f'Why this matters', _style("AIRIPDF_KW", 8, 10, _TEAL, True)), Paragraph( f'This section helps a reviewer describe governance weaknesses in broader risk language, but it should remain secondary to repository boundary, traceability, and evidence posture.', _style("AIRIPDF_VW", 8, 11, _DGRAY), ), ]) airi_table = Table(airi_rows, colWidths=[34 * mm, None]) airi_table.setStyle(TableStyle([ ("SPAN", (0, 0), (1, 0)), ("BACKGROUND", (0, 0), (1, 0), _hx(_TEAL)), ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ("LEFTPADDING", (0, 0), (-1, -1), 6), ("RIGHTPADDING", (0, 0), (-1, -1), 6), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("GRID", (0, 1), (-1, -1), 0.3, _hx(_MGRAY)), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) story.append(airi_table) # Method Boundary story.append(Spacer(1, 6 * mm)) story += _subsec_hdr("4.4 Method Boundary", _DGRAY) story.append(Paragraph( _xt(result.get("method", "")), _style("MB2", 8, 12, _DGRAY), )) story.append(Spacer(1, 1 * mm)) story.append(Paragraph( f'Scope boundary: ' 'Runtime behavior, model output correctness, ' 'dynamic validation, wet-lab reproducibility, and clinical validation are ' 'outside the scope of this local CLI scan. This report assesses structural signals only.', _style("MBSCOPE", 8, 11, _DGRAY), )) story += _footer_block() return _single_page_story(story, break_before=True) # ── Page 7: Regulatory Traceability (dedicated, detailed packet only) ───────── def _page_regulatory_traceability(result: dict[str, Any]) -> list[Any]: block = _regulatory_traceability_pdf(result) if not block: return [] story = list(block) story += _footer_block() return _single_page_story(story, break_before=True) # ── Page 8: Report Metadata (detailed packet only) ─────────────────────────── def _page7_report_metadata(result: dict[str, Any]) -> list[Any]: story: list[Any] = [] score = result["score"] story += _chapter_hdr("Chapter 5 — Report Metadata", _NAVY) tgt = result["target"] meta_items = [ ("Schema Version", result.get("schema_version", "—")), ("STEM BIO-AI Version", result.get("stem_ai_version", "—")), ("Generated (local date)", result.get("generated_at_local", "—")), ("Report Validity", "180 days from audit date"), ("Execution Mode", result.get("execution_mode", "—")), ("Repository", tgt["name"]), ("Remote URL", (tgt.get("remote") or "—")[:70]), ("Branch", tgt.get("branch") or "—"), ("Commit (HEAD)", (tgt.get("commit") or "—")[:40]), ("Files Scanned", str(tgt.get("file_count", "—"))), ("Final Score / Tier", f'{score["final_score"]} / 100 — {score["formal_tier"]}'), ] meta_data: list[list[Any]] = [[ Paragraph(f'Field', _style("MH1", 8, 10, _WHITE, True)), Paragraph(f'Value', _style("MH2", 8, 10, _WHITE, True)), ]] for field, val in meta_items: meta_data.append([ Paragraph(f'{_xt(field)}', _style(f"MF_{field[:4]}", 8, 11, _DGRAY, True)), Paragraph(_xt(str(val)), _style(f"MV_{field[:4]}", 8, 11, _DGRAY)), ]) meta_tbl = Table(meta_data, colWidths=[55 * mm, None]) meta_tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), _hx(_NAVY)), ("ROWBACKGROUNDS",(0, 1), (-1, -1), [_hx(_LGRAY), _hx(_WHITE)]), ("TOPPADDING", (0, 0), (-1, -1), 3), ("BOTTOMPADDING", (0, 0), (-1, -1), 3), ("LEFTPADDING", (0, 0), (-1, -1), 5), ("RIGHTPADDING", (0, 0), (-1, -1), 5), ("BOX", (0, 0), (-1, -1), 0.5, _hx(_MGRAY)), ("GRID", (0, 0), (-1, -1), 0.3, _hx(_MGRAY)), ])) story.append(meta_tbl) story += _footer_block() return _single_page_story(story, break_before=True) # ── plain-text PDF fallback (no reportlab) ─────────────────────────────────── def render_pdf_pages(result: dict[str, Any], mode: str, pages: int) -> list[list[str]]: score = result["score"] ast_note = _ast_scope_note(result) airi = result.get("airi_risk_coverage", {}) airi_brief = [ "", "AIRI Risk Triggers Summary", f"- Covered Risks: {airi.get('covered_count', 0)} / {airi.get('total_risks_in_detector_scope', 0)}", f"- Coverage Rate: {airi.get('coverage_rate', 0):.3f}", f"- Surface Note: {_surface_compaction_note(result)}", ] covered_risks = airi.get("covered_risks", []) if covered_risks: risk = covered_risks[0] reason = _airi_reason_summary(risk) primary = _airi_primary_summary(risk) line = f"- {risk.get('id', '—')}: {risk.get('title', '')}" if primary: line += f" | {primary}" if reason: line += f" | why: {reason}" airi_brief.append(line) brief = [ "STEM BIO-AI Local Audit Brief", f"Target: {result['target']['name']}", f"Final Score: {score['final_score']} / 100", f"Formal Tier: {score['formal_tier']}", f"Use Scope: {score['use_scope']}", f"About This Score: {_score_boundary_short_line()}", f"- {_score_boundary_lines()[0].replace('**', '')}", f"- {_score_boundary_lines()[1].replace('**', '')}", f"- {_score_boundary_lines()[2].replace('**', '')}", "", "Stage Scores", f"- Stage 1 README Evidence Signal: {score['stage_1_readme_intent']} / 100", f"- Stage 2R Repo-Local Consistency: {score['stage_2_repo_local_consistency']} / 100", f"- Stage 3 Code/Bio Responsibility: {score['stage_3_code_bio']} / 100", f"- Stage 4 Replication Evidence: {result.get('replication_score', 0)} / 100 ({result.get('replication_tier', 'R0')})", "", "Code Integrity", *[f"- {k}: {v['status']}" for k, v in result["code_integrity"].items()], *([f"- AST analysis scope: {ast_note}"] if ast_note else []), "", "Top Risks", *[f"- {r}" for r in result["notable_risks"][:4]], *airi_brief, "", "Not clinical certification. Not regulatory clearance. Not medical advice.", ] if mode == "brief": return [_fit_page(brief)] p2 = _fit_page(["Stage 2R Evidence", *[ f"- {k}: {v.get('score','')} {v.get('evidence','')}" for k, v in result["stage_2r_rubric"].items() if isinstance(v, dict) ]]) p3 = _fit_page(["Stage 3 Evidence", *[ f"- {k}: {v['score']} / {v['max']} {v['evidence']}" for k, v in result["stage_3_rubric"].items() ], "", "Stage 4 Replication Evidence", *[ f"- {k}: {v['score']} / {v['max']} {v['evidence']}" for k, v in result.get("stage_4_rubric", {}).items() ]]) p4 = _fit_page([ "Stage 4 Replication Evidence", f"- Stage 4 Replication Score: {result.get('replication_score', 0)} / 100 ({result.get('replication_tier', 'R0')})", *[ f"- {k}: {v['score']} / {v['max']} {v['evidence']}" for k, v in result.get("stage_4_rubric", {}).items() ], ]) sets = [_fit_page(brief), p2, _fit_page([ "Stage 3 Evidence", *[f"- {k}: {v['score']} / {v['max']} {v['evidence']}" for k, v in result["stage_3_rubric"].items()] ])] if pages == 5: sets.append(p4) sets.append(_fit_page([ "Closeout Summary", "Code Integrity", *[f"- {k}: {v['status']} {v['evidence'][0]}" for k, v in result["code_integrity"].items()], *([f"- AST analysis scope: {ast_note}"] if ast_note else []), "", "AIRI Risk Triggers Summary", f"- Covered Risks: {airi.get('covered_count', 0)} / {airi.get('total_risks_in_detector_scope', 0)}" + ( f" (+{len(airi.get('covered_risks', [])) - 5} beyond preview)" if len(airi.get("covered_risks", [])) > 5 else "" ), f"- Coverage Rate: {airi.get('coverage_rate', 0):.3f}", f"- Surface Note: {_surface_compaction_note(result)}", *[ f"- {risk.get('id', '—')}: {risk.get('title', '')}" + (f" | {_airi_primary_summary(risk)}" if _airi_primary_summary(risk) else "") + (f" | why: {_airi_reason_summary(risk)}" if _airi_reason_summary(risk) else "") for risk in airi.get("covered_risks", [])[:5] ], "", "Method Boundary", result["method"], ])) elif pages >= 7: sets.append(p4) sets.append(_fit_page([ "Code Integrity", *[f"- {k}: {v['status']} {v['evidence'][0]}" for k, v in result["code_integrity"].items()], *([f"- AST analysis scope: {ast_note}"] if ast_note else []), ])) sets.append(_fit_page([ "Priority Improvement Roadmap", *[f"- {r}" for r in result.get("notable_risks", [])[:4]], "", "AIRI Risk Triggers Summary", f"- Covered Risks: {airi.get('covered_count', 0)} / {airi.get('total_risks_in_detector_scope', 0)}", f"- Coverage Rate: {airi.get('coverage_rate', 0):.3f}", f"- Bundle Scope: {airi.get('airi_bundle_scope', 'unknown')}", f"- Surface Note: {_surface_compaction_note(result)}", *[ f"- {risk.get('id', '—')}: {risk.get('title', '')}" + (f" | {_airi_primary_summary(risk)}" if _airi_primary_summary(risk) else "") + (f" | why: {_airi_reason_summary(risk)}" if _airi_reason_summary(risk) else "") for risk in airi.get("covered_risks", [])[:3] ], "", "Method Boundary", result["method"], ])) _reg_lines = _regulatory_pdf_text_lines(result) if _reg_lines: sets.append(_fit_page(_reg_lines)) sets.append(_fit_page([ "Report Metadata", f"- Schema Version: {result.get('schema_version', '—')}", f"- STEM BIO-AI Version: {result.get('stem_ai_version', '—')}", f"- Generated (local date): {result.get('generated_at_local', '—')}", f"- Repository: {result['target']['name']}", f"- Branch: {result['target'].get('branch') or '—'}", f"- Commit (HEAD): {(result['target'].get('commit') or '—')[:40]}", f"- Files Scanned: {result['target'].get('file_count', '—')}", f"- Final Score / Tier: {score['final_score']} / 100 — {score['formal_tier']}", ])) return sets[:pages] def write_simple_pdf(path: Path, pages: list[list[str]]) -> None: objects: list[bytes] = [] def add(obj: str) -> int: objects.append(obj.encode("latin-1", errors="replace")) return len(objects) font_id = add("<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>") page_ids: list[int] = [] content_ids: list[int] = [] for page in pages: stream = _page_stream(page) content_ids.append(add(f"<< /Length {len(stream)} >>\nstream\n{stream}\nendstream")) page_ids.append(0) kids = [] pages_id_placeholder = len(objects) + len(pages) + 1 for idx, _ in enumerate(pages): pid = add( f"<< /Type /Page /Parent {pages_id_placeholder} 0 R /MediaBox [0 0 595 842] " f"/Resources << /Font << /F1 {font_id} 0 R >> >> /Contents {content_ids[idx]} 0 R >>" ) page_ids[idx] = pid kids.append(f"{pid} 0 R") pages_id = add(f"<< /Type /Pages /Kids [{' '.join(kids)}] /Count {len(page_ids)} >>") if pages_id != pages_id_placeholder: for idx, pid in enumerate(page_ids): objects[pid - 1] = ( f"<< /Type /Page /Parent {pages_id} 0 R /MediaBox [0 0 595 842] " f"/Resources << /Font << /F1 {font_id} 0 R >> >> /Contents {content_ids[idx]} 0 R >>" ).encode("latin-1", errors="replace") catalog_id = add(f"<< /Type /Catalog /Pages {pages_id} 0 R >>") out = bytearray(b"%PDF-1.4\n") offsets = [0] for idx, obj in enumerate(objects, start=1): offsets.append(len(out)) out.extend(f"{idx} 0 obj\n".encode("ascii")) out.extend(obj) out.extend(b"\nendobj\n") xref = len(out) out.extend(f"xref\n0 {len(objects) + 1}\n".encode("ascii")) out.extend(b"0000000000 65535 f \n") for offset in offsets[1:]: out.extend(f"{offset:010d} 00000 n \n".encode("ascii")) out.extend( f"trailer\n<< /Size {len(objects) + 1} /Root {catalog_id} 0 R >>\n" f"startxref\n{xref}\n%%EOF\n".encode("ascii") ) path.write_bytes(out) def _page_stream(lines: list[str]) -> str: chunks = ["BT", "/F1 11 Tf", "50 800 Td"] y = 800 first = True for line in lines: for wrapped in textwrap.wrap(_ascii(line), width=88) or [""]: overflow = _emit_page_chunk(chunks, wrapped, y, first) if overflow: return overflow if not first: y -= 16 first = False chunks.append("ET") return "\n".join(chunks) def _emit_page_chunk(chunks: list[str], wrapped: str, y: int, first: bool) -> str: if not first: chunks.append("0 -16 Td") if y - 16 < 60: chunks.append("ET") return "\n".join(chunks) chunks.append(f"({_escape_pdf(wrapped)}) Tj") return "" def _fit_page(lines: list[str], max_lines: int = 44) -> list[str]: fitted: list[str] = [] for line in lines: fitted.extend(textwrap.wrap(_ascii(line), width=88) or [""]) if len(fitted) >= max_lines: return fitted[:max_lines] return fitted def _escape_pdf(t: str) -> str: return t.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)") def _ascii(t: str) -> str: return t.encode("latin-1", errors="replace").decode("latin-1") def _safe_name(name: str) -> str: return re.sub(r"[^A-Za-z0-9_.-]+", "_", name).strip("_") or "stem_audit"