stem_ai/detector_bio.py

from __future__ import annotations

import ast
from collections import deque
from functools import lru_cache
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Iterable

from .detector_utils import add_finding, is_fixture_like_path, iter_code_files, iter_python_files, read_text, source_line
from .evidence import EvidenceFinding


_RDKIT_UNAVAILABLE = object()

SMILES_ALLOWED = re.compile(r"^[A-Za-z0-9@+\-\[\]\(\)=#$%\\/\.]+$")
HEX_COLOR = re.compile(r"^#[0-9A-Fa-f]{6,8}$")
VERSIONISH = re.compile(r"^[A-Za-z0-9_.-]*[._-]\d+[A-Za-z0-9_.-]*$")
GENERATED_PATH_PARTS = {"build", "dist", "audits", "stem_output", "tmp", ".manual_verify"}
SMILES_PARSER_CALLS = {"MolFromSmiles", "Chem.MolFromSmiles", "AllChem.MolFromSmiles", "dm.to_mol", "to_mol"}
SMILES_CONTEXT_TOKENS = ("smiles", "smarts", "molecule", "mol", "ligand", "compound", "substrate", "chem")
SMILES_MULTI_CHAR_ATOMS = (
    "Cl", "Br", "Si", "Na", "Li", "Ca", "Mg", "Zn", "Fe", "Cu",
    "Mn", "Hg", "Pb", "Sn", "Ag", "Au", "Pt", "Pd", "Co", "Ni", "Se", "As",
)
SMILES_ONE_CHAR_ATOMS = set("BCNOFPSIHKbcnops")
MOCK_FLAG_NAMES = {"USE_MOCK", "DEMO_MODE", "SIMULATE_DATA", "SIMULATED_DATA", "MOCK_MODE"}
MOCK_NAME_TOKENS = ("mock", "fake", "dummy", "simulat", "synthetic", "demo")
BIO_IMPORT_MODULES = ("rdkit", "Bio", "biopython", "scanpy", "anndata", "FlowCytometryTools", "pysam")
TRACE_MANIFEST_NAMES = {
    "manifest.json",
    "checksums.txt",
    "model_hashes.yaml",
    "model_hashes.yml",
    "audit_log_schema.json",
    "event_log_schema.yaml",
    "event_log_schema.yml",
}
TRACE_KEYWORDS = re.compile(
    r"\b(decision_event|override_event|model_version|dataset_hash|operator_id|timestamp|input_hash|output_hash)\b",
    re.I,
)
TRACE_SCAN_SUFFIXES = {".json", ".yaml", ".yml", ".toml", ".md", ".txt"}
TRACE_FILE_HINT = re.compile(r"(trace|audit|event|log|manifest|checksum|hash|schema|override|decision)", re.I)
BIO_TOOL_NAMES = {"blast", "blastn", "blastp", "samtools", "bwa", "bcftools", "bedtools", "minimap2"}
TAINT_NAME_TOKENS = ("query", "input", "sample", "path", "request", "user", "fastq", "bam", "vcf")


@dataclass
class AstCodeContext:
    path: Path
    tree: ast.AST
    lines: list[str]
    constants: list[ast.Constant]
    constant_assign_parents: dict[int, ast.Assign]
    assigns: list[ast.Assign]
    assign_body_context: dict[int, tuple[list[ast.stmt], int]]
    calls: list[ast.Call]
    try_nodes: list[ast.Try]
    if_nodes: list[ast.If]


def collect_bio_findings(
    target: Path,
    findings: list[EvidenceFinding],
    counters: dict[tuple[str, str], int],
) -> None:
    code_paths = list(iter_code_files(target, max_files=240))
    ast_contexts = list(_iter_ast_code_contexts(code_paths))
    _collect_smiles_surface_findings(target, findings, counters, ast_contexts)
    _collect_smiles_rdkit_validation_findings(target, findings, counters, ast_contexts)
    _collect_smiles_parser_guard_findings(target, findings, counters, ast_contexts)
    _collect_silent_mock_findings(target, findings, counters, ast_contexts)
    _collect_trace_manifest_findings(target, findings, counters)
    _collect_run_trace_findings(target, findings, counters, ast_contexts)


def _iter_ast_code_contexts(paths: list[Path]) -> Iterable[AstCodeContext]:
    for path in paths:
        if path.suffix.lower() != ".py" or is_fixture_like_path(path) or _is_generated_path(path):
            continue
        try:
            stat = path.stat()
        except OSError:
            continue
        ctx = _build_ast_context_cached(str(path), stat.st_mtime_ns, stat.st_size)
        if ctx is not None:
            yield ctx


@lru_cache(maxsize=512)
def _build_ast_context_cached(path_str: str, mtime_ns: int, size: int) -> AstCodeContext | None:
    del mtime_ns, size
    path = Path(path_str)
    text = read_text(path)
    if not text:
        return None
    try:
        tree = ast.parse(text)
    except SyntaxError:
        return None
    constants: list[ast.Constant] = []
    constant_assign_parents: dict[int, ast.Assign] = {}
    assigns: list[ast.Assign] = []
    assign_body_context: dict[int, tuple[list[ast.stmt], int]] = {}
    calls: list[ast.Call] = []
    try_nodes: list[ast.Try] = []
    if_nodes: list[ast.If] = []
    for owner in ast.walk(tree):
        if isinstance(owner, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Module)):
            for index, stmt in enumerate(owner.body):
                if isinstance(stmt, ast.Assign):
                    assign_body_context[id(stmt)] = (owner.body, index)
    queue: deque[ast.AST] = deque([tree])
    while queue:
        node = queue.popleft()
        if isinstance(node, ast.Constant):
            constants.append(node)
        elif isinstance(node, ast.Assign):
            assigns.append(node)
            if isinstance(node.value, ast.Constant):
                constant_assign_parents[id(node.value)] = node
        elif isinstance(node, ast.Call):
            calls.append(node)
        elif isinstance(node, ast.Try):
            try_nodes.append(node)
        elif isinstance(node, ast.If):
            if_nodes.append(node)
        queue.extend(ast.iter_child_nodes(node))
    return AstCodeContext(
        path=path,
        tree=tree,
        lines=text.splitlines(),
        constants=constants,
        constant_assign_parents=constant_assign_parents,
        assigns=assigns,
        assign_body_context=assign_body_context,
        calls=calls,
        try_nodes=try_nodes,
        if_nodes=if_nodes,
    )


def _collect_smiles_surface_findings(
    target: Path,
    findings: list[EvidenceFinding],
    counters: dict[tuple[str, str], int],
    ast_contexts: Iterable[AstCodeContext],
) -> None:
    detected = False
    for ctx in ast_contexts:
        for node in ctx.constants:
            if not isinstance(node, ast.Constant) or not isinstance(node.value, str):
                continue
            value = node.value.strip()
            if not _looks_like_smiles(value):
                continue
            if not _smiles_context_permits(node, value, ctx.constant_assign_parents.get(id(node))):
                continue
            issues = _smiles_issues(value)
            if not issues:
                continue
            detected = True
            add_finding(
                target,
                findings,
                counters,
                "BIO_smiles_surface_integrity",
                "bio_smiles_surface_v1",
                "detected",
                "warn",
                ctx.path,
                getattr(node, "lineno", 0),
                source_line(ctx.lines, getattr(node, "lineno", 0)),
                "ast",
                "Suspicious or malformed SMILES-like string detected by conservative surface checks.",
                {"smiles_text": value, "issues": issues},
            )
    if not detected:
        add_finding(
            target,
            findings,
            counters,
            "BIO_smiles_surface_integrity",
            "bio_smiles_surface_v1",
            "not_detected",
            "info",
            Path("."),
            0,
            "",
            "ast",
            "No malformed or suspicious SMILES-like strings detected by conservative surface checks.",
        )


def _collect_smiles_parser_guard_findings(
    target: Path,
    findings: list[EvidenceFinding],
    counters: dict[tuple[str, str], int],
    ast_contexts: Iterable[AstCodeContext],
) -> None:
    detected = False
    for ctx in ast_contexts:
        for node in ctx.assigns:
            if not isinstance(node, ast.Assign) or len(node.targets) != 1:
                continue
            target_node = node.targets[0]
            if not isinstance(target_node, ast.Name):
                continue
            parser_call = _call_name(node.value)
            if parser_call not in SMILES_PARSER_CALLS:
                continue
            if _guarded_against_none(node, target_node.id, ctx.assign_body_context.get(id(node))):
                continue
            detected = True
            add_finding(
                target,
                findings,
                counters,
                "BIO_smiles_parser_guard",
                "bio_smiles_parser_guard_v1",
                "detected",
                "warn",
                ctx.path,
                getattr(node, "lineno", 0),
                source_line(ctx.lines, getattr(node, "lineno", 0)),
                "ast",
                "SMILES parser result is used without an explicit None/invalid guard.",
                {"variable": target_node.id, "parser_call": parser_call},
            )
    if not detected:
        add_finding(
            target,
            findings,
            counters,
            "BIO_smiles_parser_guard",
            "bio_smiles_parser_guard_v1",
            "not_detected",
            "info",
            Path("."),
            0,
            "",
            "ast",
            "No missing None/invalid guards detected after SMILES parser calls.",
        )


def _collect_smiles_rdkit_validation_findings(
    target: Path,
    findings: list[EvidenceFinding],
    counters: dict[tuple[str, str], int],
    ast_contexts: Iterable[AstCodeContext],
) -> None:
    detected = False
    candidate_seen = False
    for ctx in ast_contexts:
        for node in ctx.constants:
            if not isinstance(node, ast.Constant) or not isinstance(node.value, str):
                continue
            value = node.value.strip()
            if not _looks_like_smiles(value):
                continue
            if not _smiles_context_permits(node, value, ctx.constant_assign_parents.get(id(node))):
                continue
            candidate_seen = True
            parsed = _rdkit_mol_from_smiles(value)
            if parsed is _RDKIT_UNAVAILABLE:
                add_finding(
                    target,
                    findings,
                    counters,
                    "BIO_smiles_rdkit_validation",
                    "bio_smiles_rdkit_invalid_v1",
                    "not_applicable",
                    "info",
                    Path("."),
                    0,
                    "",
                    "ast",
                    "RDKit optional validation lane unavailable in current environment.",
                    {"lane": "A1_optional_rdkit"},
                )
                return
            if parsed is not None:
                continue
            detected = True
            add_finding(
                target,
                findings,
                counters,
                "BIO_smiles_rdkit_validation",
                "bio_smiles_rdkit_invalid_v1",
                "detected",
                "warn",
                ctx.path,
                getattr(node, "lineno", 0),
                source_line(ctx.lines, getattr(node, "lineno", 0)),
                "ast",
                "RDKit optional validation lane rejected a SMILES-like string candidate.",
                {"smiles_text": value, "lane": "A1_optional_rdkit"},
            )
    if detected:
        return
    add_finding(
        target,
        findings,
        counters,
        "BIO_smiles_rdkit_validation",
        "bio_smiles_rdkit_invalid_v1",
        "not_detected",
        "info",
        Path("."),
        0,
        "",
        "ast",
        (
            "RDKit optional validation lane found no invalid SMILES-like candidates."
            if candidate_seen
            else "RDKit optional validation lane not exercised because no SMILES-like candidates were detected."
        ),
        {"lane": "A1_optional_rdkit"},
    )


def _process_mock_node_finding(
    target: Path,
    findings: list[EvidenceFinding],
    counters: dict[tuple[str, str], int],
    path: Path,
    node: ast.AST,
    lines: list[str],
) -> str | None:
    if isinstance(node, ast.Try):
        reason = _import_failure_sets_mock_flag(node)
        if reason:
            add_finding(
                target, findings, counters,
                "BIO_silent_mock_fallback", "bio_silent_mock_v1", "detected", "warn",
                path, getattr(node, "lineno", 0), source_line(lines, getattr(node, "lineno", 0)),
                "ast", "Silent mock or simulated-data fallback detected in a biological dependency path.",
                {"reason": reason},
            )
            return "detected"
    if isinstance(node, ast.If) and _is_explicit_demo_mode_branch(node):
        add_finding(
            target, findings, counters,
            "BIO_silent_mock_fallback", "bio_silent_mock_v1", "not_applicable", "info",
            path, getattr(node, "lineno", 0), source_line(lines, getattr(node, "lineno", 0)),
            "ast", "Explicit demo or simulated-data branch detected; treated as disclosed non-production behavior.",
            {"reason": "explicit_demo_mode_branch"},
        )
        return "not_applicable"
    return None


def _collect_silent_mock_findings(
    target: Path,
    findings: list[EvidenceFinding],
    counters: dict[tuple[str, str], int],
    ast_contexts: Iterable[AstCodeContext],
) -> None:
    detected = False
    not_applicable = False
    for ctx in ast_contexts:
        for node in ctx.try_nodes:
            result = _process_mock_node_finding(target, findings, counters, ctx.path, node, ctx.lines)
            if result == "detected":
                detected = True
            elif result == "not_applicable":
                not_applicable = True
        for node in ctx.if_nodes:
            result = _process_mock_node_finding(target, findings, counters, ctx.path, node, ctx.lines)
            if result == "detected":
                detected = True
            elif result == "not_applicable":
                not_applicable = True
    if not detected and not not_applicable:
        add_finding(
            target, findings, counters,
            "BIO_silent_mock_fallback", "bio_silent_mock_v1", "not_detected", "info",
            Path("."), 0, "", "ast",
            "No silent mock or simulated-data fallback patterns detected in production code paths.",
        )


def _collect_trace_manifest_findings(
    target: Path,
    findings: list[EvidenceFinding],
    counters: dict[tuple[str, str], int],
) -> None:
    detected = False
    for name in sorted(TRACE_MANIFEST_NAMES):
        path = target / name
        if not path.is_file():
            continue
        detected = True
        add_finding(
            target,
            findings,
            counters,
            "BIO_trace_manifest",
            "bio_trace_manifest_file_v1",
            "detected",
            "info",
            path,
            0,
            "",
            "file_presence",
            "Traceability manifest or audit-log schema file detected.",
            {"file_name": name},
        )
        text = read_text(path)
        lines = text.splitlines()
        for line_number, line in enumerate(lines, start=1):
            match = TRACE_KEYWORDS.search(line)
            if not match:
                continue
            add_finding(
                target,
                findings,
                counters,
                "BIO_trace_manifest",
                "bio_trace_manifest_keyword_v1",
                "detected",
                "info",
                path,
                line_number,
                source_line(lines, line_number),
                "regex",
                "Traceability-related runtime event keyword detected.",
                {"matched_term": match.group(1).lower()},
            )
    for path in _iter_trace_candidate_files(target):
        text = read_text(path)
        if not text:
            continue
        lines = text.splitlines()
        for line_number, line in enumerate(lines, start=1):
            match = TRACE_KEYWORDS.search(line)
            if not match:
                continue
            detected = True
            add_finding(
                target,
                findings,
                counters,
                "BIO_trace_manifest",
                "bio_trace_manifest_keyword_v1",
                "detected",
                "info",
                path,
                line_number,
                source_line(lines, line_number),
                "regex",
                "Traceability-related runtime event keyword detected.",
                {"matched_term": match.group(1).lower()},
            )
    if not detected:
        add_finding(
            target,
            findings,
            counters,
            "BIO_trace_manifest",
            "bio_trace_manifest_file_v1",
            "not_detected",
            "info",
            Path("."),
            0,
            "",
            "file_presence",
            "No traceability manifest or runtime audit-log schema surface detected.",
        )


def _iter_trace_candidate_files(target: Path) -> Iterable[Path]:
    root = target.resolve()
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [
            name for name in dirnames
            if name.lower() not in GENERATED_PATH_PARTS
            and name.lower() not in {"__pycache__", ".git", ".hg", ".svn", ".venv", "node_modules"}
        ]
        current = Path(dirpath)
        for filename in filenames:
            path = current / filename
            if path.suffix.lower() not in TRACE_SCAN_SUFFIXES:
                continue
            if path.name in TRACE_MANIFEST_NAMES:
                continue
            if is_fixture_like_path(path):
                continue
            rel = path.relative_to(root).as_posix()
            if not TRACE_FILE_HINT.search(rel):
                continue
            yield path


def _collect_run_trace_findings(
    target: Path,
    findings: list[EvidenceFinding],
    counters: dict[tuple[str, str], int],
    ast_contexts: Iterable[AstCodeContext],
) -> None:
    detected = False
    for ctx in ast_contexts:
        for node in ctx.calls:
            if not isinstance(node, ast.Call):
                continue
            finding = _run_trace_metadata(node)
            if not finding:
                continue
            detected = True
            add_finding(
                target,
                findings,
                counters,
                "BIO_run_trace",
                finding["pattern_id"],
                "detected",
                finding["severity"],
                ctx.path,
                getattr(node, "lineno", 0),
                source_line(ctx.lines, getattr(node, "lineno", 0)),
                "ast",
                finding["explanation"],
                finding["metadata"],
            )
    if not detected:
        add_finding(
            target,
            findings,
            counters,
            "BIO_run_trace",
            "bio_run_trace_v1",
            "not_detected",
            "info",
            Path("."),
            0,
            "",
            "ast",
            "No risky subprocess or os.system bio-tool execution patterns detected.",
        )


def _smiles_has_structure_or_entropy(value: str) -> bool:
    if any(ch in value for ch in "()[]=#123456789"):
        return True
    condensed = value.replace(".", "")
    return len(condensed) >= 8 and len(set(condensed)) <= 2 and bool(re.search(r"[BCNOFPSIbcnops]", condensed))


_SMILES_RULES: list[tuple[str, Callable[[str], bool]]] = [
    ("len_range",    lambda v: 3 <= len(v) <= 256),
    ("charset",      lambda v: bool(SMILES_ALLOWED.fullmatch(v))),
    ("not_excluded", lambda v: not _is_obviously_not_smiles(v)),
    ("token_stream", lambda v: _token_stream_is_smiles_like(v)),
    ("min_atoms",    lambda v: _smiles_atom_count(v) >= 2),
    ("structure",    lambda v: _smiles_has_structure_or_entropy(v)),
    ("bio_atom",     lambda v: bool(re.search(r"[BCNOFPSIbcnops]", v))),
]


def _apply_smiles_rule(value: str) -> bool:
    return all(check(value) for _, check in _SMILES_RULES)


def _looks_like_smiles(value: str) -> bool:
    return _apply_smiles_rule(value)


def _smiles_assign_target_permits(target: ast.Name, value: str) -> bool:
    if any(tok in target.id.lower() for tok in SMILES_CONTEXT_TOKENS):
        return True
    condensed = value.replace("(", "").replace(")", "").replace("[", "").replace("]", "")
    if len(condensed) >= 8 and len(set(condensed)) <= 2 and re.search(r"[BCNOFPSIbcnops]", condensed):
        return True
    return _is_strong_smiles_variable(target.id, value)


def _smiles_context_permits(node: ast.Constant, value: str, parent_assign: ast.Assign | None) -> bool:
    if isinstance(parent_assign, ast.Assign):
        for target in parent_assign.targets:
            if isinstance(target, ast.Name) and _smiles_assign_target_permits(target, value):
                return True
    if any(tok in value.lower() for tok in ("cl", "br", "@", "#", "=")):
        return True
    if re.search(r"\d", value) and any(ch in value for ch in "()[]=#"):
        return True
    return False


def _smiles_issues(value: str) -> list[str]:
    issues: list[str] = []
    if value.count("(") != value.count(")"):
        issues.append("unbalanced_parentheses")
    if value.count("[") != value.count("]"):
        issues.append("unbalanced_brackets")
    ring_counts = {digit: value.count(digit) for digit in set(re.findall(r"\d", value))}
    for digit, count in ring_counts.items():
        if count % 2 != 0:
            issues.append("unclosed_ring_label")
            break
    condensed = value.replace("(", "").replace(")", "").replace("[", "").replace("]", "")
    if len(condensed) >= 8 and len(set(condensed)) <= 2:
        issues.append("low_entropy_placeholder_pattern")
    return issues


def _call_name(node: ast.AST) -> str:
    if not isinstance(node, ast.Call):
        return ""
    if isinstance(node.func, ast.Name):
        return node.func.id
    if isinstance(node.func, ast.Attribute):
        parts: list[str] = []
        cur: ast.AST | None = node.func
        while isinstance(cur, ast.Attribute):
            parts.append(cur.attr)
            cur = cur.value
        if isinstance(cur, ast.Name):
            parts.append(cur.id)
        return ".".join(reversed(parts))
    return ""


def _guarded_against_none(
    assign_node: ast.Assign,
    variable_name: str,
    body_context: tuple[list[ast.stmt], int] | None,
) -> bool:
    if body_context is None:
        return False
    body, index = body_context
    for stmt in body[index + 1 : index + 4]:
        if _stmt_checks_none(stmt, variable_name):
            return True
    return False


def _stmt_checks_none(stmt: ast.stmt, variable_name: str) -> bool:
    if not isinstance(stmt, ast.If):
        return False
    test = stmt.test
    if isinstance(test, ast.Compare) and isinstance(test.left, ast.Name) and test.left.id == variable_name:
        for comparator in test.comparators:
            if isinstance(comparator, ast.Constant) and comparator.value is None:
                return True
    if isinstance(test, ast.Call) and _call_name(test) in {"assert", "raise"}:
        return variable_name in ast.unparse(test)
    return False


def _import_failure_sets_mock_flag(node: ast.Try) -> str | None:
    imports_bio = any(_stmt_imports_bio_module(stmt) for stmt in node.body)
    if not imports_bio:
        return None
    for handler in node.handlers:
        if handler.type is not None and not _handler_is_import_error(handler.type):
            continue
        if any(_stmt_sets_mock_behavior(stmt) for stmt in handler.body):
            return "import_failure_sets_mock_flag"
    return None


def _stmt_imports_bio_module(stmt: ast.stmt) -> bool:
    if isinstance(stmt, ast.Import):
        return any(alias.name.startswith(BIO_IMPORT_MODULES) for alias in stmt.names)
    if isinstance(stmt, ast.ImportFrom):
        return bool(stmt.module) and stmt.module.startswith(BIO_IMPORT_MODULES)
    return False


def _handler_is_import_error(node: ast.AST) -> bool:
    if isinstance(node, ast.Name):
        return node.id in {"ImportError", "ModuleNotFoundError", "Exception"}
    if isinstance(node, ast.Tuple):
        return any(_handler_is_import_error(elt) for elt in node.elts)
    return False


def _assign_sets_mock_behavior(node: ast.Assign) -> bool:
    for target in node.targets:
        if not isinstance(target, ast.Name):
            continue
        if target.id in MOCK_FLAG_NAMES and isinstance(node.value, ast.Constant) and node.value.value is True:
            return True
        if any(tok in target.id.lower() for tok in MOCK_NAME_TOKENS):
            return True
    return False


def _stmt_sets_mock_behavior(stmt: ast.stmt) -> bool:
    for node in ast.walk(stmt):
        if isinstance(node, ast.Assign) and _assign_sets_mock_behavior(node):
            return True
        if isinstance(node, ast.Constant) and isinstance(node.value, str):
            if any(tok in node.value.lower() for tok in MOCK_NAME_TOKENS):
                return True
    return False


def _is_explicit_demo_mode_branch(node: ast.If) -> bool:
    names = {name.id for name in ast.walk(node.test) if isinstance(name, ast.Name)}
    if not names & MOCK_FLAG_NAMES:
        return False
    branch_text = " ".join(ast.unparse(stmt).lower() for stmt in node.body)
    return any(tok in branch_text for tok in MOCK_NAME_TOKENS)


def _subprocess_trace_result(call_name: str, node: ast.Call) -> dict[str, object] | None:
    tool = _matched_bio_tool(_command_text(node))
    if not tool:
        return None
    shell_true = _keyword_bool(node, "shell")
    timeout_present = _has_keyword(node, "timeout")
    tainted = _command_uses_tainted_name(node)
    meta = {"call_name": call_name, "bio_tool": tool, "shell_true": shell_true,
            "external_input_taint": tainted, "timeout_present": timeout_present}
    if shell_true and tainted:
        pid, sev = "bio_run_trace_shell_tainted_v1", "warn"
        msg = "Known bio-tool subprocess call uses shell=True with probable external-input taint."
    elif (call_name == "subprocess.run" and node.args
          and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str) and tainted):
        pid, sev = "bio_run_trace_string_tainted_v1", "warn"
        msg = "Known bio-tool subprocess uses a string command with probable external-input taint."
    elif shell_true:
        pid, sev = "bio_run_trace_shell_v1", "info"
        msg = "Known bio-tool subprocess uses shell=True."
    elif not timeout_present:
        pid, sev = "bio_run_trace_timeout_v1", "info"
        msg = "Known bio-tool subprocess call has no explicit timeout."
    else:
        return None
    return {"pattern_id": pid, "severity": sev, "explanation": msg, "metadata": meta}


def _run_trace_metadata(node: ast.Call) -> dict[str, object] | None:
    call_name = _call_name(node)
    if call_name in {"subprocess.run", "subprocess.Popen"}:
        return _subprocess_trace_result(call_name, node)
    if call_name == "os.system":
        tool = _matched_bio_tool(_command_text(node))
        if not tool:
            return None
        return {
            "pattern_id": "bio_run_trace_os_system_v1",
            "severity": "warn",
            "explanation": "Known bio-tool invocation uses os.system, which is hard to constrain safely.",
            "metadata": {
                "call_name": call_name, "bio_tool": tool,
                "shell_true": True, "external_input_taint": _command_uses_tainted_name(node),
                "timeout_present": False,
            },
        }
    return None


def _command_text(node: ast.Call) -> str:
    if not node.args:
        return ""
    arg = node.args[0]
    try:
        return ast.unparse(arg)
    except Exception:
        return ""


def _matched_bio_tool(command_text: str) -> str:
    lowered = command_text.lower()
    for tool in BIO_TOOL_NAMES:
        if tool in lowered:
            return tool
    return ""


def _command_uses_tainted_name(node: ast.Call) -> bool:
    for child in ast.walk(node):
        if isinstance(child, ast.Name) and any(tok in child.id.lower() for tok in TAINT_NAME_TOKENS):
            return True
    return False


def _keyword_bool(node: ast.Call, name: str) -> bool:
    for kw in node.keywords:
        if kw.arg == name and isinstance(kw.value, ast.Constant) and isinstance(kw.value.value, bool):
            return kw.value.value
    return False


def _has_keyword(node: ast.Call, name: str) -> bool:
    return any(kw.arg == name for kw in node.keywords)


def _is_generated_path(path: Path) -> bool:
    return any(part.lower() in GENERATED_PATH_PARTS for part in path.parts)


def _is_obviously_not_smiles(value: str) -> bool:
    if HEX_COLOR.fullmatch(value):
        return True
    lowered = value.lower()
    if any(token in lowered for token in ("sha256", "sha512", "utf-8", "oauth", "http://", "https://")):
        return True
    if "/" in value and not any(ch in value for ch in "()[]=@#\\"):
        return True
    if "-" in value and "=" not in value and not any(ch in value for ch in "()[]@#"):
        return True
    if VERSIONISH.fullmatch(value) and not any(ch in value for ch in "()[]=@#"):
        return True
    return False


def _token_stream_is_smiles_like(value: str) -> bool:
    i = 0
    while i < len(value):
        if value.startswith("%", i):
            if i + 2 >= len(value) or not value[i + 1 : i + 3].isdigit():
                return False
            i += 3
            continue
        for token in SMILES_MULTI_CHAR_ATOMS:
            if value.startswith(token, i):
                i += len(token)
                break
        else:
            ch = value[i]
            if ch.isdigit() or ch in "@+-[]=#$()/\\.":
                i += 1
                continue
            if ch in SMILES_ONE_CHAR_ATOMS:
                i += 1
                continue
            return False
    return True


def _is_strong_smiles_variable(name: str, value: str) -> bool:
    lowered = name.lower()
    if any(tok in lowered for tok in SMILES_CONTEXT_TOKENS):
        return True
    condensed = value.replace("(", "").replace(")", "").replace("[", "").replace("]", "").replace(".", "")
    return len(condensed) >= 8 and len(set(condensed)) <= 2 and all(ch in "BCNOFPSIHKbcnops" for ch in condensed)


def _smiles_atom_count(value: str) -> int:
    count = 0
    i = 0
    while i < len(value):
        if value.startswith("%", i):
            i += 3
            continue
        matched = False
        for token in SMILES_MULTI_CHAR_ATOMS:
            if value.startswith(token, i):
                count += 1
                i += len(token)
                matched = True
                break
        if matched:
            continue
        ch = value[i]
        if ch in SMILES_ONE_CHAR_ATOMS:
            count += 1
        i += 1
    return count


@lru_cache(maxsize=1)
def _load_rdkit_chem():
    try:
        from rdkit import Chem  # type: ignore
        return Chem
    except ImportError:
        return _RDKIT_UNAVAILABLE


def _rdkit_mol_from_smiles(smiles: str):
    chem = _load_rdkit_chem()
    if chem is _RDKIT_UNAVAILABLE:
        return _RDKIT_UNAVAILABLE
    try:
        return chem.MolFromSmiles(smiles)
    except Exception:
        return None


def _rdkit_is_available() -> bool:
    return _load_rdkit_chem() is not _RDKIT_UNAVAILABLE