stem_ai/scanner.py

from __future__ import annotations

import hashlib
import re
import subprocess
from collections import defaultdict
from datetime import date, timedelta
from pathlib import Path
from typing import Any

from . import __version__
from .advisory_contract import (
    advisory_contract_schemas,
    build_offline_advisory,
    build_provider_advisory_input,
    validate_advisory_input_packet,
)
from .advisory_providers import load_provider_config, provider_handoff_metadata
from .advisory_response import validate_advisory_response_file
from .advisory_runtime import execute_advisory_call
from .calibration_profile import calibration_profile_metadata, load_calibration_profile
from .airi_risk_mapping import build_airi_coverage
from .detector_contract import collect_contract_findings
from .detectors import collect_evidence_bundle
from .evidence import EvidenceFinding
from .patterns import (
    BIAS_LIMITATION_TERMS,
    BIAS_MEASUREMENT_TERMS,
    BIO_TERMS,
    CHANGELOG_BUG_TERMS,
    COI_FUNDING_TERMS,
    DATA_SOURCE_TERMS,
    DATA_SOURCE_NEGATIVE_TERMS,
    DEMOGRAPHIC_BIAS_TERMS,
    DISCLAIMER_TERMS,
    EXACT_PINNED_DEP,
    FAIL_OPEN,
    HYPE_AUTONOMOUS_REPLACEMENT,
    HYPE_BREAKTHROUGH,
    HYPE_CLINICAL_CERTAINTY,
    HYPE_PERFECT_ACCURACY,
    HYPE_REGULATORY_APPROVAL,
    HYPE_UNIVERSAL_GENERALIZATION,
    LIMITATIONS_SECTION,
    LOOSE_DEP,
    PATIENT_METADATA,
    PLACEHOLDER_SECRET_VALUES,
    REGULATORY_FRAMEWORK_TERMS,
    REPRODUCIBILITY_TERMS,
    SECRET_TERMS,
    SKIP_DIRS,
    STAGE2R_CLINICAL_DEPLOYMENT_CLAIMS,
    STAGE2R_ENTRYPOINT_TERMS,
    STAGE2R_WORKFLOW_CLAIMS,
    TEXT_EXTENSIONS,
    WEAK_REGULATORY_SELF_ASSERTION_TERMS,
)
from .regulatory_traceability import (
    build_regulatory_basis,
    build_regulatory_traceability,
    build_stage_traceability,
)
from .reasoning_model import build_reasoning_model


CA_INDIRECT_TERMS = re.compile(
    r"\b(consensus|vcf|variant|viral|virus|genome|sequenc|sample|allele|clinical trial|"
    r"medical imaging|pydicom|biomarker|survival analysis|AutoDock|drug docking|nnU-Net|"
    r"regulatory submission)\b",
    re.I,
)
CA_DIRECT_TERMS = re.compile(
    r"\b(diagnosis|diagnostic\s+(?:tool|model|system|decision|support|pipeline|classifier|prediction|screening|test|use|grade|claim)|"
    r"clinical decision|decision support|patient-facing|(?:clinical|patient|emergency|medical)\s+triage|"
    r"triage\s+(?:tool|model|system|decision|support|prediction|risk)|risk score|"
    r"treatment recommendation|treatment guidance|drug recommendation|pharmacogenomic|pharmacogenetic|"
    r"DPYD|CYP2D6|CPIC|FDA clearance|CE mark)\b",
    re.I,
)
CA_TAXONOMY_VERSION = "ca-taxonomy-v1"
CA_TAXONOMY_SOURCE = "runtime_regex_hardcoded_in_scanner_py"
T0_HARD_FLOOR_TERMS = re.compile(
    r"\b(autonomous|automated|without clinician|clinical decision support|diagnosis|"
    r"diagnostic\s+(?:tool|model|system|decision|support|pipeline|classifier|prediction|screening|test|use)|"
    r"treatment recommendation|(?:clinical|patient|emergency|medical)\s+triage|"
    r"triage\s+(?:tool|model|system|decision|support|prediction|risk)|risk score)\b",
    re.I,
)
META_EVIDENCE_LINE = re.compile(
    r"(regex|detector|pattern|schema|contract|rubric|score component|detection method|"
    r"CA severity|CA-DIRECT|CA-INDIRECT|T0_HARD_FLOOR_TERMS|Stage 1 hype|hype penalties|"
    r"triage tiers|review priority|Executive / triage|diagnostic layer|"
    r"does not mean|clinical deployment .*requires independent expert validation|"
    r"framework's primary question|does not need|Clinically validated; safe|"
    r"prohibited .*claims|rejected by the validator|forbidden citations)",
    re.I,
)


def audit_repository(
    target: Path,
    policy_name: str = "default",
    advisory: str = "none",
    advisory_response_path: Path | None = None,
) -> dict[str, Any]:
    calibration_profile = load_calibration_profile(policy_name)
    audit_date = date.today()
    files = _list_files(target)
    readme = _read_first(target, ["README.md", "README.rst", "readme.md"])
    docs_text = _read_many(target / "docs", max_files=30)
    package_text = _read_first(target, ["pyproject.toml", "setup.py", "setup.cfg", "package.json"])
    dep_text = _read_first(
        target,
        [
            "environment.yml",
            "requirements.txt",
            "pyproject.toml",
            "setup.cfg",
            "package-lock.json",
            "pnpm-lock.yaml",
            "yarn.lock",
            "npm-shrinkwrap.json",
            "package.json",
        ],
    )
    workflow_text = _read_many(target / ".github" / "workflows", max_files=20)
    test_text = _read_many(target / "tests", max_files=40)
    changelog_text = _read_first(target, ["CHANGELOG.md", "CHANGELOG", "NEWS.md"])
    code_text = _read_many(target, max_files=200, suffixes={".py", ".sh"})

    repo_name = _repo_name(target)
    surface_text = "\n".join([readme, docs_text, package_text])
    ca_severity = _classify_ca_severity(surface_text)
    clinical_adjacent = ca_severity != "none"
    has_disclaimer = bool(DISCLAIMER_TERMS.search(readme + "\n" + docs_text))
    t0_hard_floor = _t0_hard_floor(surface_text, ca_severity, has_disclaimer)

    stage_1, stage_1_rubric = _score_stage_1(readme, docs_text, package_text, ca_severity, has_disclaimer)
    stage_2r, stage_2r_rubric = _score_stage_2r(
        readme,
        docs_text,
        package_text,
        workflow_text,
        test_text,
        changelog_text,
        code_text,
        clinical_adjacent,
        has_disclaimer,
    )
    evidence_ledger, ast_signal_summary, stage_4 = collect_evidence_bundle(target)
    evidence_ledger = [_normalize_evidence_finding(f) for f in evidence_ledger]

    cc_findings: list[EvidenceFinding] = []
    cc_counters: dict[tuple[str, str], int] = defaultdict(int)
    cc_summary = collect_contract_findings(target, readme, cc_findings, cc_counters)
    evidence_ledger.extend(_normalize_evidence_finding(f.to_dict()) for f in cc_findings)

    stage_3, stage_3_rubric = _score_stage_3(target, readme, docs_text, workflow_text, test_text, dep_text, changelog_text)
    code_integrity = _code_integrity_from_findings(evidence_ledger, clinical_adjacent, has_disclaimer)

    weights = {"stage_1": 0.4, "stage_2": 0.2, "stage_3": 0.4}
    risk_penalty = 10 if code_integrity["C1_hardcoded_credentials"]["status"] == "FAIL" else 0
    raw_score = round(stage_1 * weights["stage_1"] + stage_2r * weights["stage_2"] + stage_3 * weights["stage_3"] - risk_penalty)
    score_cap = _score_cap(ca_severity, has_disclaimer, t0_hard_floor)
    final_score = min(raw_score, score_cap) if score_cap is not None else raw_score

    result = {
        "schema_version": "stem-ai-local-cli-result-v1.6",
        "stem_ai_version": __version__,
        "generated_at_local": audit_date.isoformat(),
        "execution_mode": "LOCAL_ANALYSIS",
        "target": {
            "name": repo_name,
            "local_path": str(target),
            "remote": _git(target, "remote", "get-url", "origin"),
            "branch": _git(target, "rev-parse", "--abbrev-ref", "HEAD"),
            "commit": _git(target, "rev-parse", "HEAD"),
            "file_count": len(files),
        },
        "classification": {
            "clinical_adjacent": clinical_adjacent,
            "ca_severity": ca_severity,
            "ca_taxonomy_version": CA_TAXONOMY_VERSION,
            "ca_taxonomy_source": CA_TAXONOMY_SOURCE,
            "t0_hard_floor": t0_hard_floor,
            "score_cap": score_cap,
            "has_explicit_clinical_boundary": has_disclaimer,
        },
        "score": {
            "stage_1_readme_intent": stage_1,
            "stage_2_cross_platform": "not_applicable_in_LOCAL_ANALYSIS",
            "stage_2_repo_local_consistency": stage_2r,
            "stage_2_lane": "STAGE_2R_REPO_LOCAL_CONSISTENCY",
            "stage_3_code_bio": stage_3,
            "weights": weights,
            "risk_penalty": risk_penalty,
            "raw_score_before_floor": raw_score,
            "final_score": final_score,
            "formal_tier": _tier(final_score),
            "use_scope": _use_scope(final_score),
        },
        "stage_2r_rubric": stage_2r_rubric,
        "stage_1_rubric": stage_1_rubric,
        "stage_3_rubric": stage_3_rubric,
        "replication_score": stage_4["replication_score"],
        "replication_tier": stage_4["replication_tier"],
        "stage_4_rubric": stage_4["stage_4_rubric"],
        "code_integrity": code_integrity,
        "code_contract": cc_summary,
        "airi_risk_coverage": build_airi_coverage(code_integrity, cc_summary, stage_1_rubric, t0_hard_floor, evidence_ledger),
        "evidence_ledger": evidence_ledger,
        "detector_summary": _detector_summary(evidence_ledger),
        "ast_signal_summary": ast_signal_summary,
        "notable_positive_evidence": _positive_evidence(workflow_text, test_text, docs_text, package_text),
        "notable_risks": _risks(
            clinical_adjacent,
            has_disclaimer,
            code_integrity,
            evidence_ledger,
            cc_summary,
            self_asserted_compliance=stage_1_rubric.get("R2_regulatory_framework", {}).get("score") == 5,
        ),
        "file_hashes_sha256": _hash_key_files(target),
        "method": "Deterministic local CLI scan. No LLM, network, or runtime test execution is required.",
        "artifact_surface_notes": {
            "json": "JSON remains the canonical full-fidelity artifact. Human-readable surfaces may compact repeated same-file evidence for readability while preserving detector semantics.",
            "human_readable_compaction": "Markdown, HTML, PDF, and explain outputs may compact repeated same-file evidence rows or narrative blocks. Refer to JSON for full per-finding detail.",
        },
        "measurement_basis": {
            "stage_1": "README/package bio-domain regex; hype-claim penalties; limitation, strong regulatory-framework signals, weaker self-asserted compliance signals, disclaimer, demographic-bias, and reproducibility responsibility signals. R2 regulatory-framework signals post-hoc align with ICH M15 §4.1 MAP and §2.2.2 Appropriateness of Proposed MIDD.",
            "stage_2r": "Repo-local consistency checks across README, package metadata, docs, changelog, test/CI files, and deterministic contradiction/staleness/workflow-support heuristics. R2R_D2 missing clinical boundary post-hoc aligns with ICH M15 §2.1.2 Context of Use requirement.",
            "stage_3_T1": ".github/workflows/ directory contains files",
            "stage_3_T2": "tests/ directory contains bio-domain vocabulary (regex)",
            "stage_3_T3": "CHANGELOG.md, CHANGELOG, or NEWS.md file exists; max credit requires bug-fix, patch, or security entries",
            "stage_3_B1": "Python and JavaScript dependency or lock manifests are treated as repository provenance surfaces; max credit still requires data-source, dataset-citation, or IRB language and does not by itself prove dataset lineage. Post-hoc aligns with ICH M15 §3 Verification (user-generated code documentation) and §4.2 MAR (data and methods section).",
            "stage_3_B2": "bias/limitation vocabulary present in README and docs; max credit requires quantitative measurement evidence or related test coverage. Post-hoc aligns with ICH M15 §3 Validation and Applicability Assessment: 'limitations of the data and model should be described and discussed.'",
        "stage_3_B3": "funding/sponsor/COI vocabulary present in README, docs, or FUNDING.md (regex)",
            "stage_4": "Deterministic replication evidence lane: containers, reproducibility targets, lock/pin/hash evidence, README reproducibility sections, dataset/model artifact references, citation metadata, license/use-scope restriction evidence, CLI/seed/example signals. Post-hoc aligns with ICH M15 §4.3: coding scripts, data, and supporting files should be available for regulatory review.",
            "ca_severity": "Clinical/diagnostic term regex match in README, docs, and package metadata",
            "ca_taxonomy_governance": f"{CA_TAXONOMY_VERSION} from {CA_TAXONOMY_SOURCE}; reference markdown is informative, not authoritative runtime source.",
            "C1": "Hardcoded key pattern regex (AWS AKIA*, sk-*, ghp_*, api_key=...), excluding obvious placeholder/test values and test/example fixture contexts",
            "C2": "Dependency-manifest-only pin check across requirements/environment/setup.cfg/pyproject dependency sections; ignores non-dependency metadata lines",
            "C3": "Patient metadata patterns in deprecated/legacy/archive directories (regex)",
            "C4": "AST-backed detection of executable fail-open Python exception handlers (except/pass or except/return True)",
            "C6": "Mock-auth, auto-login, or no-auth local/self-host boundary signals surfaced from README/docs/config/code review",
            "CC1": "AST scan for public functions with confidence/threshold parameter defaulting to 0 or 0.0.",
            "CC2": "README code-block import names compared against package __all__; flags names claimed as importable but absent.",
            "CC3": "AST scan for validate_*/check_* functions using len() but no re.match/search structure check.",
            "airi_risk_coverage": "MIT AI Risk Repository V4_03 (airisk.mit.edu, arXiv:2408.12622) local-governed mapping: detectors cross-referenced to AIRI risk IDs through a full local registry, curated runtime bundle, and detector-mapping registry.",
            "BIO-Diagnostics": "Deterministic evidence-only bio diagnostics: conservative SMILES surface checks, SMILES parser-guard checks, silent mock fallback detection, traceability manifest surface checks, and subprocess bio-tool run-trace heuristics.",
            "REG-Scaffolding": "Evidence-only traceability scaffolding signals from manifest/hash/audit-log schema surfaces; intended as structural audit-readiness support rather than compliance proof.",
            "score_cap": "Score ceiling applied when clinical-adjacent signals lack explicit disclaimer",
        },
        "calibration_profile": calibration_profile_metadata(calibration_profile),
    }
    result["audit_freshness"] = _build_audit_freshness(result, audit_date)
    result["regulatory_basis"] = build_regulatory_basis(audit_date)
    result["stage_traceability"] = build_stage_traceability(result)
    result["regulatory_traceability"] = build_regulatory_traceability(result)
    result["reasoning_model"] = build_reasoning_model(result)
    if advisory == "validate":
        result["ai_advisory"] = build_offline_advisory(result)
    elif advisory == "packet":
        result["ai_advisory_input"] = build_provider_advisory_input(result)
        result["ai_advisory_input"]["contract_schemas"] = advisory_contract_schemas()
        result["ai_advisory_input"]["provider_request"] = provider_handoff_metadata(load_provider_config())
        result["ai_advisory_input"]["packet_contract"] = validate_advisory_input_packet(result["ai_advisory_input"])
    elif advisory == "call":
        result["ai_advisory_input"] = build_provider_advisory_input(result)
        result["ai_advisory_input"]["contract_schemas"] = advisory_contract_schemas()
        result["ai_advisory_input"]["provider_request"] = provider_handoff_metadata(load_provider_config())
        result["ai_advisory_input"]["packet_contract"] = validate_advisory_input_packet(result["ai_advisory_input"])
        result["ai_advisory"] = execute_advisory_call(result)
    if advisory_response_path is not None:
        result["ai_advisory"] = validate_advisory_response_file(result, advisory_response_path)
    return result


def _normalize_evidence_finding(finding: dict[str, Any]) -> dict[str, Any]:
    normalized = dict(finding)
    normalized.setdefault("evidence_status", _default_evidence_status(normalized.get("status", "")))
    normalized.setdefault(
        "confidence",
        _default_finding_confidence(normalized.get("status", ""), normalized.get("match_type", "")),
    )
    return normalized


def _default_evidence_status(status: str) -> str:
    mapping = {
        "detected": "confirmed_present",
        "absent": "confirmed_missing",
        "not_detected": "not_found_in_reviewed_sources",
        "not_applicable": "not_applicable",
        "manual_review_required": "manual_review_required",
        "error": "collection_error",
    }
    return mapping.get(str(status), "observed")


def _default_finding_confidence(status: str, match_type: str) -> str:
    status = str(status)
    match_type = str(match_type)
    if status in {"error", "manual_review_required"}:
        return "low"
    if status == "not_detected":
        return "medium"
    if match_type in {"ast", "regex", "file_presence", "dependency"}:
        return "high"
    if match_type in {"aggregate", "metadata", "limit"}:
        return "medium"
    return "medium"


def _build_audit_freshness(result: dict[str, Any], audit_date: date) -> dict[str, Any]:
    classification = result.get("classification", {})
    file_hashes = result.get("file_hashes_sha256", {})
    commit = result.get("target", {}).get("commit")
    ca_severity = classification.get("ca_severity", "none")
    review_after_days = 45 if ca_severity != "none" else 90
    freshness_basis = "clinical_adjacent_short_cycle" if ca_severity != "none" else "standard_cycle"
    expires_on = audit_date + timedelta(days=review_after_days)
    change_triggers = [
        "git_commit_changed",
        "readme_or_docs_claim_surface_changed",
        "dependency_manifest_changed",
        "dataset_or_model_reference_changed",
        "ci_or_reproducibility_surface_changed",
        "changelog_or_release_hygiene_surface_changed",
    ]
    reasons: list[str] = []
    if not commit:
        reasons.append("git_commit_unavailable")
    if not file_hashes:
        reasons.append("key_file_hashes_unavailable")
    return {
        "review_after_days": review_after_days,
        "freshness_basis": freshness_basis,
        "expires_on": expires_on.isoformat(),
        "expired": audit_date > expires_on,
        "anchored_commit": commit,
        "hashes_available_for": sorted(file_hashes.keys()),
        "change_triggered_reaudit_supported": bool(commit) or bool(file_hashes),
        "change_triggered_reaudit_recommended_now": bool(reasons),
        "change_triggered_reaudit_reasons": reasons,
        "change_triggers": change_triggers,
    }


def _score_stage_1(readme: str, docs_text: str, package_text: str, ca_severity: str, has_disclaimer: bool) -> tuple[int, dict[str, Any]]:
    clinical_adjacent = ca_severity != "none"
    responsibility_text = "\n".join([readme, docs_text])
    score = 60
    items: dict[str, Any] = {"baseline": {"score": 60, "evidence": "Non-nascent README evidence baseline."}}
    if readme and BIO_TERMS.search(readme):
        score += _add_stage1_item(items, "S1_domain_readme", 10, "README exposes bio/medical domain vocabulary.")
    if package_text and BIO_TERMS.search(package_text):
        score += _add_stage1_item(items, "S1_domain_package", 5, "Package metadata exposes bio/medical domain vocabulary.")
    hype_penalties = (
        ("H1_clinical_certainty_hype", HYPE_CLINICAL_CERTAINTY, -10, "Clinical certainty or deployment-readiness claim detected."),
        ("H2_regulatory_approval_hype", HYPE_REGULATORY_APPROVAL, -10, "Regulatory approval or clearance claim detected."),
        ("H3_autonomous_replacement_hype", HYPE_AUTONOMOUS_REPLACEMENT, -10, "Autonomous clinician-replacement claim detected."),
        ("H4_breakthrough_marketing_hype", HYPE_BREAKTHROUGH, -5, "Marketing hype language detected."),
        ("H5_universal_generalization_hype", HYPE_UNIVERSAL_GENERALIZATION, -5, "Universal generalization claim detected."),
        ("H6_perfect_accuracy_hype", HYPE_PERFECT_ACCURACY, -10, "Perfect or guaranteed accuracy claim detected."),
    )
    for key, pattern, penalty, evidence in hype_penalties:
        if _has_claim_match(pattern, readme):
            score += _add_stage1_item(items, key, penalty, evidence)
    if LIMITATIONS_SECTION.search(readme):
        score += _add_stage1_item(items, "R1_limitations_section", 15, "README contains an explicit limitations or validation-boundary section.")
    if REGULATORY_FRAMEWORK_TERMS.search(responsibility_text):
        score += _add_stage1_item(items, "R2_regulatory_framework", 15, "Regulatory, IRB, SaMD, or clinical-reporting framework language detected.")
    elif WEAK_REGULATORY_SELF_ASSERTION_TERMS.search(responsibility_text):
        score += _add_stage1_item(items, "R2_regulatory_framework", 5, "Self-asserted privacy/compliance language detected without stronger regulatory-framework evidence.")
    elif ca_severity == "CA-DIRECT":
        score += _add_stage1_item(items, "R2_regulatory_framework", -10, "CA-DIRECT surface lacks regulatory or governance framework language.")
    elif ca_severity == "CA-INDIRECT":
        score += _add_stage1_item(items, "R2_regulatory_framework", -5, "CA-INDIRECT surface lacks regulatory or governance framework language.")
    if has_disclaimer:
        score += _add_stage1_item(items, "R3_clinical_disclaimer", 10, "Explicit non-clinical or non-diagnostic boundary detected.")
    elif ca_severity == "CA-DIRECT":
        score += _add_stage1_item(items, "R3_clinical_disclaimer", -10, "CA-DIRECT surface lacks explicit non-clinical or non-diagnostic boundary.")
    elif ca_severity == "CA-INDIRECT":
        score += _add_stage1_item(items, "R3_clinical_disclaimer", -5, "CA-INDIRECT surface lacks explicit non-clinical or non-diagnostic boundary.")
    if DEMOGRAPHIC_BIAS_TERMS.search(responsibility_text):
        score += _add_stage1_item(items, "R4_demographic_bias_boundary", 10, "Demographic, subgroup, fairness, bias, or validation-cohort language detected.")
    if REPRODUCIBILITY_TERMS.search(responsibility_text):
        score += _add_stage1_item(items, "R5_reproducibility_provisions", 10, "Reproducibility, replication, seed, lockfile, or checksum language detected.")
    if not readme:
        score += _add_stage1_item(items, "S1_missing_readme", -20, "No root README detected.")
    final = _clamp(score)
    items["calculation"] = f"60 plus Stage 1 evidence additions/deductions = {final}"
    items["stage_1_score"] = final
    return final, items


def _add_stage1_item(items: dict[str, Any], key: str, score: int, evidence: str) -> int:
    items[key] = {"score": score, "evidence": evidence}
    return score


def _score_stage_2r(
    readme: str,
    docs_text: str,
    package_text: str,
    workflow_text: str,
    test_text: str,
    changelog_text: str,
    code_text: str,
    clinical_adjacent: bool,
    has_disclaimer: bool,
) -> tuple[int, dict[str, Any]]:
    items: dict[str, Any] = {
        "baseline": {
            "score": 60,
            "evidence": "Non-nascent local repository baseline.",
            "detector_id": "stage2r_baseline",
            "decision_basis": "repository has sufficient local structure to enter repo-local consistency review",
        }
    }
    score = 60
    if readme and package_text and _shared_terms(readme, package_text):
        score += _add_stage2r_item(
            items,
            "R2R_1_readme_package_code_alignment",
            15,
            "README has domain overlap with package metadata or entry points.",
            decision_basis="shared bio-domain terms detected across README and package metadata",
        )
    if readme and docs_text and _shared_terms(readme, docs_text):
        score += _add_stage2r_item(
            items,
            "R2R_2_readme_docs_alignment",
            10,
            "README has domain overlap with docs/tutorial/troubleshooting surfaces.",
            decision_basis="shared bio-domain terms detected across README and docs/tutorial surfaces",
        )
    if (workflow_text or test_text) and re.search(r"(test|pytest|unittest|ci|workflow)", readme + workflow_text + test_text, re.I):
        score += _add_stage2r_item(
            items,
            "R2R_3_readme_test_ci_alignment",
            10,
            "Test/CI surfaces are present and locally consistent.",
            decision_basis="workflow/test support terms found across README and local support surfaces",
        )
    if _limitation_repeated(readme, docs_text, changelog_text):
        score += _add_stage2r_item(
            items,
            "R2R_4_limitation_repetition",
            10,
            "Limitation or validation-boundary language repeats across independent repository surfaces.",
            decision_basis="limitations or validation-boundary language detected in at least two repository surfaces",
        )
    if _has_internal_clinical_contradiction(readme, docs_text, package_text, has_disclaimer):
        score += _add_stage2r_item(
            items,
            "R2R_D1_internal_clinical_boundary_contradiction",
            -20,
            "Explicit clinical-use boundary conflicts with clinical deployment or decision-support claims.",
            decision_basis="explicit clinical boundary and clinical deployment/support claims co-occur in reviewed sources",
        )
    if clinical_adjacent and not has_disclaimer:
        score += _add_stage2r_item(
            items,
            "R2R_D2_missing_clinical_use_boundary",
            -20,
            "Clinical-adjacent surfaces exist without an explicit non-diagnostic/non-clinical boundary.",
            decision_basis="clinical_adjacent=True and explicit non-clinical boundary was not detected",
            tier_impact="can contribute to clinical score ceilings and hard-floor review paths when stronger clinical deployment or certainty claims are also present",
        )
    if _version_mismatch(readme, package_text):
        score += _add_stage2r_item(
            items,
            "R2R_D3_stale_metadata",
            -10,
            "README version metadata conflicts with package metadata version.",
            decision_basis="README version marker and package version marker differ",
        )
    if _unsupported_workflow_claim(readme, docs_text, package_text, workflow_text, test_text, code_text):
        score += _add_stage2r_item(
            items,
            "R2R_D4_unsupported_workflow_claim",
            -15,
            "README/docs claim runnable workflow, CLI, test, or demo support without matching local support surfaces.",
            decision_basis="workflow/demo/CLI claims detected while workflow, tests, or local support entrypoints are absent",
        )
    final = _clamp(score)
    items["calculation"] = f"60 plus local consistency additions/deductions = {final}"
    items["stage_2r_score"] = final
    items["verdict"] = _stage2r_verdict(final)
    return final, items


def _add_stage2r_item(
    items: dict[str, Any],
    key: str,
    score: int,
    evidence: str,
    decision_basis: str | None = None,
    tier_impact: str | None = None,
) -> int:
    items[key] = {
        "score": score,
        "evidence": evidence,
        "detector_id": key,
        "decision_basis": decision_basis or evidence,
        **({"tier_impact": tier_impact} if tier_impact else {}),
    }
    return score


def _stage2r_verdict(score: int) -> str:
    if score >= 80:
        return "Strong Local Consistency"
    if score >= 60:
        return "Mixed Local Consistency"
    return "Local Contradiction / Insufficient Consistency"


def _limitation_repeated(readme: str, docs_text: str, changelog_text: str) -> bool:
    lanes = [
        bool(_has_limitation_language(readme)),
        bool(_has_limitation_language(docs_text)),
        bool(_has_limitation_language(changelog_text)),
    ]
    return sum(lanes) >= 2


def _has_limitation_language(text: str) -> bool:
    return bool(text and (LIMITATIONS_SECTION.search(text) or BIAS_LIMITATION_TERMS.search(text)))


def _has_internal_clinical_contradiction(readme: str, docs_text: str, package_text: str, has_disclaimer: bool) -> bool:
    if not has_disclaimer:
        return False
    return bool(STAGE2R_CLINICAL_DEPLOYMENT_CLAIMS.search("\n".join([readme, docs_text, package_text])))


def _version_mismatch(readme: str, package_text: str) -> bool:
    readme_version = _readme_version(readme)
    package_version = _package_version(package_text)
    return bool(readme_version and package_version and readme_version != package_version)


def _readme_version(text: str) -> str | None:
    match = re.search(r"(?im)\b(?:version|release)\s*[:= ]\s*v?([0-9]+(?:\.[0-9]+){1,3}(?:[-+][A-Za-z0-9_.-]+)?)\b", text)
    return match.group(1).lower() if match else None


def _package_version(text: str) -> str | None:
    match = re.search(r"(?im)^\s*version\s*=\s*['\"]v?([0-9]+(?:\.[0-9]+){1,3}(?:[-+][A-Za-z0-9_.-]+)?)['\"]", text)
    return match.group(1).lower() if match else None


def _unsupported_workflow_claim(readme: str, docs_text: str, package_text: str, workflow_text: str, test_text: str, code_text: str) -> bool:
    claimed = bool(STAGE2R_WORKFLOW_CLAIMS.search("\n".join([readme, docs_text])))
    supported = bool(workflow_text or test_text or code_text or STAGE2R_ENTRYPOINT_TERMS.search(package_text))
    return claimed and not supported


def _score_stage_3(target: Path, readme: str, docs_text: str, workflow_text: str, test_text: str, dep_text: str, changelog_text: str = "") -> tuple[int, dict[str, Any]]:
    ci = 15 if workflow_text else 0
    normalized_test_text = test_text.replace("_", " ")
    tests = 15 if test_text and BIO_TERMS.search(normalized_test_text) else 8 if test_text else 0

    changelog_path = next((p for p in [target / "CHANGELOG.md", target / "CHANGELOG", target / "NEWS.md"] if p.exists()), None)
    changelog, changelog_evidence = _score_changelog(changelog_path, changelog_text)

    provenance, provenance_evidence = _score_provenance(dep_text, readme, docs_text)

    bias_text = "\n".join([readme, docs_text])
    bias, bias_evidence = _score_bias(bias_text, test_text)

    coi_text = "\n".join([readme, docs_text, _read_first(target, ["FUNDING.md", "CITATION.cff", "AUTHORS.md"])])
    coi = 5 if COI_FUNDING_TERMS.search(coi_text) else 0
    raw_score = ci + tests + changelog + provenance + bias + coi
    raw_max = 80
    score = round((raw_score / raw_max) * 100)
    rubric = {
        "T1_CI_CD": {
            "score": ci,
            "max": 15,
            "evidence": "Workflow files detected." if ci else "No workflow files detected.",
            "detector_id": "S3_T1_workflow_files",
            "decision_basis": "workflow files present under .github/workflows/" if ci else "no workflow files present under .github/workflows/",
        },
        "T2_domain_tests": {
            "score": tests,
            "max": 15,
            "evidence": "Domain-specific test text detected." if tests == 15 else "Tests present but domain specificity is limited." if tests else "No tests detected.",
            "detector_id": "S3_T2_domain_tests",
            "decision_basis": "tests surface present with bio-domain vocabulary" if tests == 15 else "tests surface present without clear bio-domain specificity" if tests else "no tests surface detected",
        },
        "T3_changelog_release_hygiene": {
            "score": changelog,
            "max": 15,
            "evidence": changelog_evidence,
            "detector_id": "S3_T3_changelog_release_hygiene",
            "decision_basis": "CHANGELOG/NEWS presence plus bug-fix or patch-entry detection",
        },
        "B1_data_provenance_controls": {
            "score": provenance,
            "max": 15,
            "evidence": provenance_evidence,
            "detector_id": "S3_B1_dependency_manifest",
            "decision_basis": "dependency or lock manifest presence plus data-source, dataset, or IRB language review",
        },
        "B2_bias_limitations": {
            "score": bias,
            "max": 15,
            "evidence": bias_evidence,
            "detector_id": "S3_B2_bias_limitations",
            "decision_basis": "bias/limitations vocabulary with optional measurement-evidence escalation",
        },
        "B3_coi_funding": {
            "score": coi,
            "max": 5,
            "evidence": "COI, funding, sponsor, or acknowledgement language detected." if coi else "No COI/funding disclosure detected by local CLI scan.",
            "detector_id": "S3_B3_coi_funding",
            "decision_basis": "COI/funding/sponsor language review across README, docs, FUNDING, CITATION, and AUTHORS surfaces",
        },
        "stage_3_raw_total": {"score": raw_score, "max": raw_max, "evidence": "Raw rubric total before normalization to 100."},
    }
    return _clamp(score), rubric


def _score_changelog(changelog_path: Path | None, changelog_text: str) -> tuple[int, str]:
    if not changelog_path:
        return 0, "No changelog detected."
    text = changelog_text or _read_text(changelog_path)
    if CHANGELOG_BUG_TERMS.search(text):
        return 15, f"{changelog_path.name} contains bug-fix, patch, or security entries."
    return 5, f"{changelog_path.name} exists but no bug-fix or patch entries detected."


def _score_provenance(dep_text: str, readme: str, docs_text: str) -> tuple[int, str]:
    if not dep_text:
        return 0, "No dependency/provenance manifest detected."
    surface = "\n".join([readme, docs_text])
    if DATA_SOURCE_NEGATIVE_TERMS.search(surface):
        return 10, "Dependency manifest detected; provenance language appears in a negative or non-approval context."
    if DATA_SOURCE_TERMS.search(surface):
        return 15, "Dependency manifest detected with data source, IRB, or dataset citation language."
    return 10, "Dependency manifest detected; no explicit data source or IRB citation found."


def _score_bias(bias_text: str, test_text: str) -> tuple[int, str]:
    if not BIAS_LIMITATION_TERMS.search(bias_text):
        return 0, "No bias/limitations language detected by local CLI scan."
    if BIAS_MEASUREMENT_TERMS.search(bias_text) or (test_text and BIAS_LIMITATION_TERMS.search(test_text)):
        return 15, "Bias/limitations language with measurement evidence or test coverage detected."
    if LIMITATIONS_SECTION.search(bias_text) or len(BIAS_LIMITATION_TERMS.findall(bias_text)) >= 2:
        return 8, "Structured bias/limitations language detected; no quantitative measurement evidence found."
    return 0, "Only a minimal single-term bias/limitations mention was detected; structured boundary language or measurement evidence was not found."


def _code_integrity_from_findings(
    evidence_ledger: list[dict[str, Any]],
    clinical_adjacent: bool,
    has_disclaimer: bool,
) -> dict[str, Any]:
    detector_defaults = {
        "C1_hardcoded_credentials": ("FAIL", "PASS", "No direct credential patterns detected by local CLI scan."),
        "C2_dependency_pinning": ("WARN", "PASS", "Dependency manifest appears pinned or not present."),
        "C3_dead_or_deprecated_patient_adjacent_paths": ("WARN", "PASS", "No deprecated patient-adjacent metadata patterns detected."),
        "C4_exception_handling_clinical_adjacent_paths": ("WARN", "PASS", "No executable fail-open exception handler detected."),
        "C5_compliance_boundary_integrity": ("WARN", "PASS", "No compliance-boundary integrity warning detected in current report layer."),
        "C6_mock_auth_or_fail_open_boundary": ("WARN", "PASS", "No mock-auth or fail-open local-boundary warning detected in reviewed sources."),
    }
    findings_by_detector: dict[str, list[dict[str, Any]]] = {key: [] for key in detector_defaults}
    for finding in evidence_ledger:
        detector = str(finding.get("detector", ""))
        if detector in findings_by_detector:
            findings_by_detector[detector].append(finding)

    result: dict[str, Any] = {}
    for detector, (detected_status, pass_status, pass_message) in detector_defaults.items():
        detected = [f for f in findings_by_detector[detector] if f.get("status") == "detected"]
        if detected:
            evidence = [_finding_summary(f) for f in detected[:5]]
            result[detector] = {"status": detected_status, "evidence": evidence}
        else:
            result[detector] = {"status": pass_status, "evidence": [pass_message]}

    external_dependency_findings = [
        f
        for f in evidence_ledger
        if f.get("detector") == "R2R_D5_single_external_service_dependency" and f.get("status") == "detected"
    ]
    if external_dependency_findings:
        result["C2_dependency_pinning"] = {
            "status": "WARN",
            "evidence": [
                "External operational dependency signal surfaced in code-integrity lane.",
                *[_finding_summary(f) for f in external_dependency_findings[:3]],
            ],
        }

    c5_evidence: list[str] = []
    unsupported_claim_findings = [
        f
        for f in evidence_ledger
        if f.get("detector") == "S1_R2_unsupported_legal_or_compliance_claim" and f.get("status") == "detected"
    ]
    if unsupported_claim_findings:
        c5_evidence.append("Unsupported legal/compliance claim surfaced in boundary-integrity lane.")
        c5_evidence.extend(_finding_summary(f) for f in unsupported_claim_findings[:2])
    if clinical_adjacent and not has_disclaimer:
        c5_evidence.append("Clinical-adjacent surfaces exist without an explicit non-diagnostic/non-clinical boundary.")
    if c5_evidence:
        result["C5_compliance_boundary_integrity"] = {
            "status": "WARN",
            "evidence": c5_evidence[:5],
        }

    c6_findings = [
        f
        for f in evidence_ledger
        if f.get("detector") == "C6_mock_auth_or_fail_open_boundary" and f.get("status") == "detected"
    ]
    if c6_findings:
        result["C6_mock_auth_or_fail_open_boundary"] = {
            "status": "WARN",
            "evidence": [
                "Mock-auth or auto-login boundary surfaced in code-integrity lane.",
                *[_finding_summary(f) for f in c6_findings[:4]],
            ][:5],
        }
    return result


def _finding_summary(finding: dict[str, Any]) -> str:
    file = str(finding.get("file", "."))
    line = int(finding.get("line", 0) or 0)
    snippet = str(finding.get("snippet", "")).strip()
    location = file if file == "." or not line else f"{file}:{line}"
    if snippet:
        return f"{location} {snippet}"
    return location


def _list_files(root: Path) -> list[Path]:
    files: list[Path] = []
    for path in root.rglob("*"):
        if any(part in SKIP_DIRS for part in path.parts):
            continue
        if path.is_file():
            files.append(path)
    return files


def _read_first(root: Path, names: list[str]) -> str:
    for name in names:
        path = root / name
        if path.exists() and path.is_file():
            return _read_text(path)
    return ""


def _read_many(root: Path, max_files: int = 100, suffixes: set[str] | None = None) -> str:
    if not root.exists():
        return ""
    chunks: list[str] = []
    count = 0
    for path in root.rglob("*"):
        if any(part in SKIP_DIRS for part in path.parts):
            continue
        if not path.is_file():
            continue
        if suffixes is not None and path.suffix.lower() not in suffixes:
            continue
        if suffixes is None and path.suffix.lower() not in TEXT_EXTENSIONS:
            continue
        chunks.append(_read_text(path))
        count += 1
        if count >= max_files:
            break
    return "\n".join(chunks)


def _read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8", errors="ignore")
    except OSError:
        return ""


def _read_deprecated_text(root: Path, max_files: int = 120) -> str:
    chunks: list[str] = []
    count = 0
    deprecated_names = {"deprecated", "legacy", "archive", "archives", "old"}
    for path in root.rglob("*"):
        if any(part in SKIP_DIRS for part in path.parts):
            continue
        if not path.is_file() or path.suffix.lower() not in TEXT_EXTENSIONS:
            continue
        lowered_parts = {part.lower() for part in path.parts}
        if not lowered_parts & deprecated_names:
            continue
        chunks.append(_read_text(path))
        count += 1
        if count >= max_files:
            break
    return "\n".join(chunks)


def _repo_name(target: Path) -> str:
    remote = _git(target, "remote", "get-url", "origin")
    if remote:
        cleaned = remote.rstrip("/").removesuffix(".git")
        return "/".join(cleaned.split("/")[-2:])
    return target.name


def _git(target: Path, *args: str) -> str | None:
    try:
        result = subprocess.run(
            ["git", "-c", f"safe.directory={target.as_posix()}", "-C", str(target), *args],
            capture_output=True,
            text=True,
            timeout=5,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    if result.returncode != 0:
        return None
    return result.stdout.strip() or None


def _shared_terms(left: str, right: str) -> bool:
    left_terms = {m.group(0).lower() for m in BIO_TERMS.finditer(left)}
    right_terms = {m.group(0).lower() for m in BIO_TERMS.finditer(right)}
    return bool(left_terms & right_terms)


def _dependency_unpinned(dep_text: str) -> bool:
    if not dep_text:
        return False
    dep_lines = _dependency_entries(dep_text)
    return any(_is_unpinned_dependency(line) for line in dep_lines[:100])


def _dependency_entries(dep_text: str) -> list[str]:
    entries: list[str] = []
    in_pyproject_deps = False
    for raw in dep_text.splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        if re.match(r"^(dependencies|requires|install_requires|pdf|demo)\s*=", line):
            in_pyproject_deps = "[" in line and "]" not in line
            entries.extend(_quoted_dependency_strings(line))
            continue
        if in_pyproject_deps:
            entries.extend(_quoted_dependency_strings(line))
            if "]" in line:
                in_pyproject_deps = False
            continue
        if re.match(r"^[-A-Za-z0-9_.]+(\[[^\]]+\])?(\s*(==|===|>=|<=|~=|!=|>|<|@)\s*|$)", line):
            entries.append(line)
    return entries


def _quoted_dependency_strings(line: str) -> list[str]:
    values = [m.group(1) or m.group(2) for m in re.finditer(r'"([^"]+)"|\'([^\']+)\'', line)]
    return [v.strip() for v in values if re.match(r"^[A-Za-z0-9_.-]+(\[[^\]]+\])?(\s*(==|===|>=|<=|~=|!=|>|<|@)\s*|$)", v.strip())]


def _is_unpinned_dependency(line: str) -> bool:
    normalized = line.split("#", 1)[0].strip().rstrip(",")
    if not normalized or normalized.startswith(("-", "[", "{")):
        return False
    if EXACT_PINNED_DEP.search(normalized):
        return False
    if LOOSE_DEP.search(normalized):
        return True
    return bool(re.match(r"^[A-Za-z0-9_.-]+(\[[^\]]+\])?(\s*;.*)?$", normalized))


def _classify_ca_severity(surface_text: str) -> str:
    if _has_claim_match(CA_DIRECT_TERMS, surface_text):
        return "CA-DIRECT"
    if _has_claim_match(CA_INDIRECT_TERMS, surface_text):
        return "CA-INDIRECT"
    return "none"


def _t0_hard_floor(surface_text: str, ca_severity: str, has_disclaimer: bool) -> bool:
    return ca_severity == "CA-DIRECT" and not has_disclaimer and _has_claim_match(T0_HARD_FLOOR_TERMS, surface_text)


def _has_claim_match(pattern: re.Pattern[str], text: str) -> bool:
    return any(pattern.search(line) and not META_EVIDENCE_LINE.search(line) for line in text.splitlines())


def _score_cap(ca_severity: str, has_disclaimer: bool, t0_hard_floor: bool) -> int | None:
    if t0_hard_floor:
        return 39
    if ca_severity != "none" and not has_disclaimer:
        return 69
    return None


def _positive_evidence(workflow_text: str, test_text: str, docs_text: str, package_text: str) -> list[str]:
    items: list[str] = []
    if package_text:
        items.append("Package metadata was available for repo-local consistency checks.")
    if workflow_text:
        items.append("CI workflow files were detected.")
    if test_text:
        items.append("Test files were detected.")
    if docs_text:
        items.append("Documentation files were detected.")
    return items or ["No strong positive local evidence detected by the CLI scan."]


def _risks(
    clinical_adjacent: bool,
    has_disclaimer: bool,
    code_integrity: dict[str, Any],
    evidence_ledger: list[dict[str, Any]],
    cc_summary: dict[str, Any] | None = None,
    self_asserted_compliance: bool = False,
) -> list[str]:
    risks: list[str] = []
    if clinical_adjacent and not has_disclaimer:
        _append_unique_risk(risks, "Clinical-adjacent surfaces exist without an explicit non-diagnostic/non-clinical boundary.")
    if self_asserted_compliance:
        _append_unique_risk(risks, "Self-asserted compliance or privacy-governance claim requires independent verification.")
    if _detector_detected(evidence_ledger, "S1_R2_unsupported_legal_or_compliance_claim"):
        _append_unique_risk(risks, "Legal, privacy, or compliance claim appears without supporting governance or security-grounding evidence in reviewed repository sources.")
    if _detector_detected(evidence_ledger, "R2R_D5_single_external_service_dependency"):
        _append_unique_risk(risks, "Core workflow appears materially dependent on named external service providers; local or self-host claims may overstate operational independence.")
    for key, item in code_integrity.items():
        if item["status"] in {"WARN", "FAIL"}:
            _append_unique_risk(risks, _humanize_code_integrity_risk(key, item))
    for key, item in (cc_summary or {}).items():
        if isinstance(item, dict) and item.get("status") == "WARN":
            _append_unique_risk(risks, _humanize_contract_risk(key, item))
    return risks or ["No major local risks detected by the CLI scan."]


def _append_unique_risk(risks: list[str], text: str) -> None:
    if text and text not in risks:
        risks.append(text)


def _humanize_code_integrity_risk(key: str, item: dict[str, Any]) -> str:
    evidence = str((item.get("evidence") or [""])[0]).strip()
    if evidence:
        return evidence
    label_map = {
        "C1_hardcoded_credentials": "Hardcoded credential signal surfaced in code-integrity lane.",
        "C2_dependency_pinning": "Dependency pinning or external operational dependency signal surfaced in code-integrity lane.",
        "C3_dead_or_deprecated_patient_adjacent_paths": "Deprecated or legacy patient-adjacent path signal surfaced in code-integrity lane.",
        "C4_exception_handling_clinical_adjacent_paths": "Executable fail-open exception signal surfaced in code-integrity lane.",
        "C5_compliance_boundary_integrity": "Compliance-boundary integrity signal surfaced in code-integrity lane.",
        "C6_mock_auth_or_fail_open_boundary": "Mock-auth or auto-login trust-boundary signal surfaced in code-integrity lane.",
    }
    return label_map.get(key, f"{key} surfaced in code-integrity lane.")


def _humanize_contract_risk(key: str, item: dict[str, Any]) -> str:
    count = item.get("count", "?")
    evidence = str((item.get("evidence") or [""])[0]).strip()
    if evidence:
        return evidence
    return f"Contract check {key} surfaced reviewable issues (count={count})."


def _detector_summary(evidence_ledger: list[dict[str, Any]]) -> dict[str, Any]:
    by_status: dict[str, int] = {}
    by_detector: dict[str, dict[str, int]] = {}
    for finding in evidence_ledger:
        status = str(finding.get("status", "unknown"))
        detector = str(finding.get("detector", "unknown"))
        by_status[status] = by_status.get(status, 0) + 1
        bucket = by_detector.setdefault(detector, {})
        bucket[status] = bucket.get(status, 0) + 1
    return {
        "total_findings": len(evidence_ledger),
        "by_status": by_status,
        "by_detector": by_detector,
    }


def _detector_detected(evidence_ledger: list[dict[str, Any]], detector: str) -> bool:
    return any(
        str(finding.get("detector", "")) == detector and str(finding.get("status", "")) == "detected"
        for finding in evidence_ledger
    )


def _hash_key_files(target: Path) -> dict[str, str]:
    hashes: dict[str, str] = {}
    for name in ["README.md", "pyproject.toml", "environment.yml", "requirements.txt"]:
        path = target / name
        if path.exists() and path.is_file():
            hashes[name] = hashlib.sha256(path.read_bytes()).hexdigest().upper()
    return hashes


def _tier(score: int) -> str:
    if score <= 39:
        return "T0 Rejected"
    if score <= 54:
        return "T1 Quarantine"
    if score <= 69:
        return "T2 Caution"
    if score <= 84:
        return "T3 Supervised"
    return "T4 Candidate"


def _use_scope(score: int) -> str:
    if score <= 39:
        return "Do not rely on this repository without independent expert validation."
    if score <= 54:
        return "Exploratory review only; no patient-adjacent use."
    if score <= 69:
        return "Research reference and supervised non-clinical technical review only."
    return "Supervised institutional review candidate; clinical deployment still requires independent validation."


def _clamp(score: int) -> int:
    return max(0, min(100, int(score)))