#!/usr/bin/env python3
"""CARBON project model, API candidates, regression lifecycle, and confidence dashboard."""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import html
import json
import math
import os
import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

from carbon_brand import REPORT_FAVICON
from carbon_context import safe_execution_context
from carbon_intelligence import refresh_memory
from carbon_quality_confidence import assess_quality_confidence
from carbon_schemas import INTENT_LIBRARY_SCHEMA, ensure_intent_library


SCHEMA = "carbon.framework/v1"
MODEL_SCHEMA = "carbon.product-model/v1"
COVERAGE_SCHEMA = "carbon.coverage/v1"
REGRESSION_SCHEMA = "carbon.regression/v1"
DOCTRINE_PATH = Path(__file__).resolve().parents[1] / "references" / "testing-doctrine.json"
TESTING_DOMAINS_PATH = Path(__file__).resolve().parents[1] / "references" / "testing-domains.json"
INITIAL_TEST_LIMIT = 20
INITIAL_BUDGET_SECONDS = 25 * 60
DOMAIN_BASELINE_COUNT = 5
IGNORED_DIRS = {
    ".git", ".carbon", ".idea", ".vscode", "node_modules", "vendor",
    "dist", "build", "coverage", ".next", ".nuxt", ".venv", "venv",
    "__pycache__", "target", "Pods",
}
SOURCE_EXTENSIONS = {
    ".js", ".jsx", ".ts", ".tsx", ".py", ".rb", ".go", ".java",
    ".kt", ".swift", ".cs", ".php", ".rs", ".html", ".vue", ".svelte",
}
UI_EXTENSIONS = {".html", ".jsx", ".tsx", ".vue", ".svelte", ".swift"}
TEST_MARKERS = ("test", "spec", "e2e", "playwright", "cypress", "selenium")
API_ROUTE_PATTERNS = [
    re.compile(r"(?:app|router|server)\.(get|post|put|patch|delete|head|options)\(\s*['\"]([^'\"]+)", re.I),
    re.compile(r"@(?:app|router)\.(get|post|put|patch|delete|head|options)\(\s*['\"]([^'\"]+)", re.I),
    re.compile(r"\b(GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS)\s+(/(?:api|v\d+)[A-Za-z0-9_{}:./-]*)", re.I),
]
API_USAGE_PATTERN = re.compile(r"['\"](/(?:api|v\d+)(?:/[A-Za-z0-9_{}:.-]+)+)['\"]")
LOCAL_IMPORT_PATTERN = re.compile(
    r"(?:from\s+|require\(\s*)['\"](\.[^'\"]+)['\"]"
)
SOURCE_FLOW_RULES = [
    {
        "key": "create-edit-delete-record",
        "name": "Create, edit, and delete a record",
        "reference": "flows/productivity/create-edit-delete-record.md",
        "sourceThreshold": 1,
        "requiredCooccurrence": 3,
        "signals": [
            r"(?-i:\b(?:[Cc]reate|[Aa]dd|[Nn]ew)(?!(?:Child|Element|Event|EventListener|Item|Listener|Node|Settings|Preferences)\b)(?:[A-Z][A-Za-z0-9]*|[_-][A-Za-z0-9_]+)\b)",
            r"(?-i:\b(?:[Ee]dit|[Uu]pdate|[Ss]ave)(?!(?:Child|Element|Event|EventListener|Item|Listener|Node|Settings|Preferences)\b)(?:[A-Z][A-Za-z0-9]*|[_-][A-Za-z0-9_]+)\b)",
            r"(?-i:\b(?:[Dd]elete|[Rr]emove|[Dd]estroy|[Aa]rchive)(?!(?:Child|Element|Event|EventListener|Item|Listener|Node|Settings|Preferences)\b)(?:[A-Z][A-Za-z0-9]*|[_-][A-Za-z0-9_]+)\b)",
        ],
    },
    {
        "key": "settings-preferences",
        "name": "Change settings and preferences",
        "reference": "flows/engagement/settings-preferences.md",
        "sourceThreshold": 2,
        "signals": [
            r"\b(?:settings?|preferences?)\b",
            r"\b(?:save|update|persist|storage)[A-Z_a-z0-9]*\b",
        ],
    },
    {
        "key": "search-filter-sort",
        "name": "Search, filter, and sort records",
        "reference": "flows/productivity/search-filter-sort.md",
        "sourceThreshold": 2,
        "minimumSignals": 2,
        "signals": [r"\bsearch[A-Z_a-z0-9]*\b", r"\bfilter[A-Z_a-z0-9]*\b", r"\bsort[A-Z_a-z0-9]*\b"],
    },
    {
        "key": "sign-in",
        "name": "Sign in",
        "reference": "flows/identity/sign-in.md",
        "sourceThreshold": 2,
        "signals": [r"\b(?:login|logIn|signIn|sign[-_ ]?in)\b", r"\b(?:password|session|auth)[A-Z_a-z0-9]*\b"],
    },
    {
        "key": "signup-registration",
        "name": "Sign up and register",
        "reference": "flows/identity/signup-registration.md",
        "sourceThreshold": 2,
        "signals": [r"\b(?:signup|signUp|register|registration|sign[-_ ]?up)\b", r"\b(?:account|user|email)[A-Z_a-z0-9]*\b"],
    },
    {
        "key": "checkout",
        "name": "Checkout",
        "reference": "flows/commerce/checkout.md",
        "sourceThreshold": 2,
        "signals": [
            r"(?:\b(?:function|class|def|const|let|var)\s+\w*checkout\w*|\.\w*checkout\w*\s*\(|\b\w*checkout\w*\s*\(|['\"]/checkout)",
            r"(?:\b(?:function|class|def|const|let|var)\s+\w*(?:cart|order|payment)\w*|\.\w*(?:cart|order|payment)\w*\s*\(|\b\w*(?:cart|order|payment)\w*\s*\(|['\"]/(?:cart|orders?|payments?))",
        ],
    },
    {
        "key": "upload-process-download",
        "name": "Upload, process, and download a file",
        "reference": "flows/productivity/upload-process-download.md",
        "sourceThreshold": 2,
        "signals": [r"\bupload[A-Z_a-z0-9]*\b", r"\bdownload[A-Z_a-z0-9]*\b"],
    },
    {
        "key": "support-ticket",
        "name": "Submit and track a support request",
        "reference": "flows/engagement/support-ticket.md",
        "sourceThreshold": 2,
        "signals": [r"\b(?:contact|support|ticket)[A-Z_a-z0-9-]*\b", r"\b(?:submit|message)[A-Z_a-z0-9]*\b"],
    },
]
SECRET_PATTERN = re.compile(
    r"(?i)(authorization\s*[:=]|bearer\s+[a-z0-9._-]{12,}|api[_-]?key\s*[:=]|password\s*[:=])"
)
def utc_now() -> str:
    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def stable_id(prefix: str, *parts: str) -> str:
    raw = "\0".join(str(part) for part in parts)
    return f"{prefix}-{hashlib.sha256(raw.encode()).hexdigest()[:12]}"


def atomic_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
        json.dump(value, handle, indent=2, ensure_ascii=False)
        handle.write("\n")
        temporary = Path(handle.name)
    os.replace(temporary, path)


def append_jsonl(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(value, ensure_ascii=False) + "\n")


def load_json(path: Path, fallback: Any = None) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return fallback
    except json.JSONDecodeError as error:
        raise ValueError(f"Invalid JSON in {path}: {error}") from error


def testing_doctrine() -> dict[str, Any]:
    doctrine = load_json(DOCTRINE_PATH, {})
    if doctrine.get("schema") != "carbon.testing-doctrine/v1":
        raise ValueError("CARBON testing doctrine is missing or has an unsupported schema.")
    return doctrine


def testing_domains() -> list[dict[str, Any]]:
    registry = load_json(TESTING_DOMAINS_PATH, {})
    if registry.get("schema") != "carbon.testing-domains/v1":
        raise ValueError("CARBON testing-domain registry is missing or has an unsupported schema.")
    domains = registry.get("domains")
    if not isinstance(domains, list) or not domains:
        raise ValueError("CARBON testing-domain registry has no domains.")
    return [domain for domain in domains if isinstance(domain, dict) and domain.get("id")]


def framework_dir(root: Path) -> Path:
    return root / ".carbon" / "framework"


def framework_paths(root: Path) -> dict[str, Path]:
    base = framework_dir(root)
    return {
        "base": base,
        "config": base / "config.json",
        "model": base / "product-model.json",
        "coverage": base / "coverage-map.json",
        "regression": base / "regression-suite.json",
        "outcomes": base / "outcomes.jsonl",
        "history": base / "confidence-history.jsonl",
        "orchestration": base / "orchestration.json",
        "plan": base / "orchestration-plan.json",
        "analysis": base / "confidence.json",
        "report": base / "report.html",
        "intentTests": root / ".carbon" / "tests" / "intent-tests.json",
    }


def ensure_framework(root: Path) -> dict[str, Path]:
    paths = framework_paths(root)
    paths["base"].mkdir(parents=True, exist_ok=True)
    now = utc_now()
    defaults = {
        "config": {
            "schema": SCHEMA,
            "projectRoot": str(root),
            "defaultSize": "quick",
            "executionPolicy": "local-first",
            "mutationPolicy": "propose-first; scoped approval required",
            "publicAndPaidPolicy": "explicit approval required",
            "updatedAt": now,
        },
        "model": {
            "schema": MODEL_SCHEMA,
            "project": {"name": root.name, "root": str(root)},
            "surfaces": [], "apis": [], "flows": [], "dependencies": [],
            "changedFiles": [], "warnings": [], "updatedAt": now,
        },
        "coverage": {
            "schema": COVERAGE_SCHEMA,
            "links": [], "gaps": [], "updatedAt": now,
        },
        "regression": {
            "schema": REGRESSION_SCHEMA,
            "tests": [], "updatedAt": now,
        },
    }
    for key, value in defaults.items():
        if not paths[key].exists():
            atomic_json(paths[key], value)
    for key in ("outcomes", "history"):
        paths[key].touch(exist_ok=True)
    return paths


def safe_text(path: Path, limit: int = 500_000) -> str:
    try:
        if path.stat().st_size > limit:
            return ""
        text = path.read_text(encoding="utf-8", errors="replace")
        return "" if SECRET_PATTERN.search(text[:4000]) and path.suffix in {".env", ".pem"} else text
    except OSError:
        return ""


def repository_files(root: Path, max_files: int = 20_000) -> list[Path]:
    found: list[Path] = []
    for current, dirs, files in os.walk(root):
        dirs[:] = sorted(d for d in dirs if d not in IGNORED_DIRS and not d.startswith(".cache"))
        for name in sorted(files):
            path = Path(current) / name
            if path.is_symlink():
                continue
            found.append(path)
            if len(found) >= max_files:
                return found
    return found


def git_changed_files(root: Path) -> list[str]:
    def trackable(name: str) -> bool:
        first = name.strip("/").split("/", 1)[0]
        return bool(name) and first not in IGNORED_DIRS

    def names(command: list[str]) -> list[str]:
        try:
            result = subprocess.run(
                ["git", "-C", str(root), *command],
                check=True, capture_output=True, text=True, timeout=15,
            )
        except (OSError, subprocess.SubprocessError):
            return []
        return [row.strip() for row in result.stdout.splitlines() if trackable(row.strip())]

    try:
        result = subprocess.run(
            ["git", "-C", str(root), "status", "--porcelain=v1", "-z"],
            check=True, capture_output=True, text=True, timeout=15,
        )
    except (OSError, subprocess.SubprocessError):
        return []
    changed: list[str] = []
    for record in result.stdout.split("\0"):
        if len(record) >= 4:
            name = record[3:]
            if " -> " in name:
                name = name.split(" -> ", 1)[1]
            if trackable(name) and name not in changed:
                changed.append(name)
    if changed:
        return changed

    compare_refs: list[str] = []
    explicit_base = os.getenv("CARBON_BASE_REF", "").strip()
    if explicit_base:
        compare_refs.append(explicit_base)
    event_path = os.getenv("GITHUB_EVENT_PATH", "").strip()
    if event_path:
        event = load_json(Path(event_path), {})
        base_sha = (((event or {}).get("pull_request") or {}).get("base") or {}).get("sha")
        if base_sha:
            compare_refs.append(str(base_sha))
    github_base = os.getenv("GITHUB_BASE_REF", "").strip()
    if github_base:
        compare_refs.extend([f"origin/{github_base}", github_base])
    compare_refs.append("HEAD~1")
    for base in dict.fromkeys(compare_refs):
        rows = names(["diff", "--name-only", "--diff-filter=ACMR", f"{base}...HEAD"])
        if rows:
            return rows
    return changed


def parse_document(path: Path) -> tuple[dict[str, Any] | None, str | None]:
    if path.suffix.lower() == ".json":
        value = load_json(path)
        return (value if isinstance(value, dict) else None), None
    try:
        import yaml  # type: ignore
    except ImportError:
        return None, f"Found {path.name}; install PyYAML or provide JSON to expand its operations."
    try:
        value = yaml.safe_load(path.read_text(encoding="utf-8"))
        return (value if isinstance(value, dict) else None), None
    except Exception as error:  # pragma: no cover - optional parser
        return None, f"Could not parse {path.name}: {error}"


def schema_fields(schema: Any, document: dict[str, Any]) -> list[dict[str, Any]]:
    if not isinstance(schema, dict):
        return []
    if "$ref" in schema:
        node: Any = document
        for part in str(schema["$ref"]).removeprefix("#/").split("/"):
            node = node.get(part, {}) if isinstance(node, dict) else {}
        schema = node if isinstance(node, dict) else {}
    required = set(schema.get("required") or [])
    rows = []
    for name, spec in (schema.get("properties") or {}).items():
        spec = spec if isinstance(spec, dict) else {}
        rows.append({
            "name": name,
            "type": spec.get("type") or ("reference" if "$ref" in spec else "unknown"),
            "format": spec.get("format"),
            "required": name in required,
            "minimum": spec.get("minimum"), "maximum": spec.get("maximum"),
            "minLength": spec.get("minLength"), "maxLength": spec.get("maxLength"),
            "enum": spec.get("enum") if isinstance(spec.get("enum"), list) else [],
            "nullable": spec.get("nullable") is True,
        })
    return rows


def openapi_operations(path: Path, document: dict[str, Any]) -> list[dict[str, Any]]:
    operations = []
    global_security = document.get("security") or []
    for route, path_item in (document.get("paths") or {}).items():
        if not isinstance(path_item, dict):
            continue
        for method in ("get", "post", "put", "patch", "delete", "head", "options"):
            operation = path_item.get(method)
            if not isinstance(operation, dict):
                continue
            request_schema = {}
            content = (operation.get("requestBody") or {}).get("content") or {}
            for media in ("application/json", "application/x-www-form-urlencoded", "multipart/form-data"):
                if isinstance(content.get(media), dict):
                    request_schema = content[media].get("schema") or {}
                    break
            parameters = []
            for param in list(path_item.get("parameters") or []) + list(operation.get("parameters") or []):
                if isinstance(param, dict):
                    parameters.append({
                        "name": param.get("name"), "in": param.get("in"),
                        "required": param.get("required") is True,
                        "type": (param.get("schema") or {}).get("type", "unknown"),
                    })
            operation_id = operation.get("operationId") or f"{method}-{route}"
            operations.append({
                "id": stable_id("API", method, route),
                "operationId": operation_id,
                "method": method.upper(), "path": route, "source": str(path),
                "summary": operation.get("summary") or operation.get("description") or operation_id,
                "security": operation.get("security", global_security),
                "parameters": parameters,
                "requestFields": schema_fields(request_schema, document),
                "responses": sorted(str(code) for code in (operation.get("responses") or {}).keys()),
                "idempotencyDeclared": bool(operation.get("x-idempotent") or operation.get("x-idempotency-key")),
                "uiFlows": list(operation.get("x-carbon-ui-flows") or []),
                "evidence": "openapi",
            })
    return operations


def endpoint_from_url(url: str) -> tuple[str, str]:
    parsed = urlparse(url)
    path = re.sub(r"/\d+(?=/|$)", "/{id}", parsed.path)
    path = re.sub(r"/[0-9a-f]{8}-[0-9a-f-]{27,}(?=/|$)", "/{id}", path, flags=re.I)
    return parsed.netloc, path or "/"


def parse_har(path: Path) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    value = load_json(path, {})
    log = value.get("log", {}) if isinstance(value, dict) else {}
    page_titles = {str(row.get("id")): str(row.get("title") or row.get("id")) for row in log.get("pages", []) if isinstance(row, dict)}
    endpoints: dict[tuple[str, str, str], dict[str, Any]] = {}
    flows: dict[str, dict[str, Any]] = {}
    for row in log.get("entries", []):
        if not isinstance(row, dict):
            continue
        request = row.get("request") or {}
        method = str(request.get("method") or "GET").upper()
        host, route = endpoint_from_url(str(request.get("url") or ""))
        if not host:
            continue
        page_ref = str(row.get("pageref") or "traffic")
        flow_id = stable_id("FLOW", page_titles.get(page_ref, page_ref))
        flows.setdefault(flow_id, {"id": flow_id, "name": page_titles.get(page_ref, page_ref), "source": str(path), "apis": []})
        key = (host, method, route)
        endpoint = endpoints.setdefault(key, {
            "id": stable_id("API", host, method, route), "operationId": f"{method}-{route}",
            "method": method, "path": route, "host": host, "source": str(path),
            "summary": f"Observed {method} {route}", "security": "unknown",
            "parameters": [], "requestFields": [], "responses": [],
            "idempotencyDeclared": False, "uiFlows": [], "evidence": "har",
        })
        status = str((row.get("response") or {}).get("status") or "")
        if status and status not in endpoint["responses"]:
            endpoint["responses"].append(status)
        if flow_id not in endpoint["uiFlows"]:
            endpoint["uiFlows"].append(flow_id)
        if endpoint["id"] not in flows[flow_id]["apis"]:
            flows[flow_id]["apis"].append(endpoint["id"])
    return list(endpoints.values()), list(flows.values())


def merge_by_id(existing: list[dict[str, Any]], incoming: list[dict[str, Any]]) -> list[dict[str, Any]]:
    rows = {str(row.get("id")): row for row in existing if isinstance(row, dict) and row.get("id")}
    for row in incoming:
        identifier = str(row.get("id"))
        rows[identifier] = {**rows.get(identifier, {}), **row}
    return sorted(rows.values(), key=lambda row: (str(row.get("kind", "")), str(row.get("name") or row.get("path") or row.get("id"))))


def infer_source_flows(source_cache: dict[str, str]) -> list[dict[str, Any]]:
    """Infer only high-confidence common flows from explicit source signals.

    These are candidate product-model links, not runtime proof that the flow
    works. Requiring every signal group avoids treating one generic verb as a
    discovered journey while still recognizing behavior spread across files.
    """
    flows: list[dict[str, Any]] = []
    for rule in SOURCE_FLOW_RULES:
        signal_counts: dict[str, int] = {}
        matched_signals = 0
        for signal in rule["signals"]:
            matching = {
                relative
                for relative, text in source_cache.items()
                if re.search(signal, text, re.I)
            }
            if not matching:
                continue
            matched_signals += 1
            for relative in matching:
                signal_counts[relative] = signal_counts.get(relative, 0) + 1
        required_signals = int(rule.get("minimumSignals", len(rule["signals"])))
        if matched_signals >= required_signals:
            required_cooccurrence = int(rule.get("requiredCooccurrence", 1))
            if max(signal_counts.values(), default=0) < required_cooccurrence:
                continue
            threshold = int(rule.get("sourceThreshold", 1))
            sources = sorted(
                relative for relative, count in signal_counts.items()
                if count >= threshold
            )
            if not sources:
                continue
            key = str(rule["key"])
            flows.append({
                "id": stable_id("FLOW", "source", key),
                "key": key,
                "name": str(rule["name"]),
                "source": "source-discovery",
                "sources": sources,
                "apis": [],
                "reference": str(rule["reference"]),
                "evidence": "source-inference",
                "confidence": "candidate",
            })
    return flows


def related_source_files(root: Path, source: str, source_cache: dict[str, str]) -> list[str]:
    """Resolve a small local-import closure for source contract inspection."""
    pending = [source]
    related: list[str] = []
    while pending and len(related) < 24:
        current = pending.pop(0)
        if current in related or current not in source_cache:
            continue
        related.append(current)
        current_path = Path(current)
        for raw in LOCAL_IMPORT_PATTERN.findall(source_cache[current]):
            base = (current_path.parent / raw)
            candidates = [base, base.with_suffix(".js"), base / "index.js"]
            match = next(
                (candidate.as_posix() for candidate in candidates if candidate.as_posix() in source_cache),
                None,
            )
            if match and match not in related:
                pending.append(match)
    return related


def source_contracts(root: Path, source: str, source_cache: dict[str, str]) -> list[str]:
    contracts: list[str] = []
    for relative in related_source_files(root, source, source_cache):
        for raw_line in source_cache[relative].splitlines():
            line = " ".join(raw_line.strip().split())
            if not line or SECRET_PATTERN.search(line):
                continue
            if re.search(r"\b(?:SEED|FIXTURE|SAMPLE_DATA)\s*=\s*\[", line):
                continue
            relevant = (
                re.search(r"(?:errors\.push|sendError|new Error)\s*\(", line)
                or re.search(r"\btypeof\s+\w+(?:\.\w+)?\s*!==", line)
                or re.search(r"\b(?:length|size|limit|offset|price)\s*(?:[<>]=?|={2,3}|!={1,2})\s*\d+", line, re.I)
                or re.search(r"\b(?:const|let|var)\s+[A-Z][A-Z0-9_]*\s*=\s*\[", line)
                or re.search(r"\bNumber\.is(?:Integer|Finite)\s*\(", line)
                or "// default" in line.lower()
            )
            if relevant:
                value = f"{relative}: {line[:240]}"
                if value not in contracts:
                    contracts.append(value)
            if len(contracts) >= 40:
                break
    def priority(value: str) -> tuple[int, str]:
        if re.search(r"(?:\.length\b|[<>]=?\s*\d+|Number\.is|\btypeof\b|\b[A-Z][A-Z0-9_]*\s*=\s*\[)", value):
            return (0, value)
        if re.search(r"(?:required|must be|unknown field|default)", value, re.I):
            return (1, value)
        return (2, value)
    return sorted(contracts, key=priority)[:40]


def enrich_source_apis(root: Path, apis: list[dict[str, Any]], source_cache: dict[str, str]) -> list[dict[str, Any]]:
    registrations: dict[str, list[tuple[int, int, str, str]]] = {}
    for source, text in source_cache.items():
        rows: list[tuple[int, int, str, str]] = []
        for pattern in API_ROUTE_PATTERNS:
            for match in pattern.finditer(text):
                method, route = match.groups()
                rows.append((match.start(), match.end(), method.upper(), route))
        registrations[source] = sorted(rows)

    for api in apis:
        if api.get("evidence") != "source":
            continue
        source = str(api.get("source") or "")
        text = source_cache.get(source, "")
        rows = registrations.get(source, [])
        snippet = ""
        for index, (start, _, method, route) in enumerate(rows):
            if method == api.get("method") and route == api.get("path"):
                end = rows[index + 1][0] if index + 1 < len(rows) else len(text)
                snippet = text[start:end]
                break
        statuses = sorted(set(re.findall(r"(?:sendJson|sendError)\s*\(\s*\w+\s*,\s*(\d{3})", snippet)))
        if statuses:
            api["responses"] = statuses
        if re.search(r"\brequireApiKey\s*\(", snippet):
            api["security"] = [{"x-api-key": []}]
        api["codeContracts"] = source_contracts(root, source, source_cache)
    return apis


def discover(root: Path, explicit_specs: list[str] | None = None, har_paths: list[str] | None = None) -> dict[str, Any]:
    paths = ensure_framework(root)
    old_model = load_json(paths["model"], {})
    files = repository_files(root)
    surfaces: list[dict[str, Any]] = []
    apis: list[dict[str, Any]] = []
    flows: list[dict[str, Any]] = []
    tests: list[str] = []
    warnings: list[str] = []
    source_cache: dict[str, str] = {}
    specs: list[Path] = []
    hars: list[Path] = []
    for file in files:
        rel = file.relative_to(root).as_posix()
        lower = rel.lower()
        is_test_source = any(marker in lower for marker in TEST_MARKERS)
        if is_test_source and file.suffix.lower() in SOURCE_EXTENSIONS | {".json", ".yaml", ".yml"}:
            tests.append(rel)
        if file.name.lower() in {"openapi.json", "openapi.yaml", "openapi.yml", "swagger.json", "swagger.yaml", "swagger.yml"}:
            specs.append(file)
        if file.suffix.lower() == ".har":
            hars.append(file)
        if file.suffix.lower() not in SOURCE_EXTENSIONS:
            continue
        text = safe_text(file)
        source_cache[rel] = text
        if file.suffix.lower() in UI_EXTENSIONS or any(part in lower for part in ("/pages/", "/views/", "/screens/", "/components/")):
            surfaces.append({
                "id": stable_id("SURFACE", "ui", rel), "kind": "ui", "name": file.stem,
                "source": rel, "risks": ["functional", "usability", "accessibility", "responsive"],
                "apiIds": [], "flowIds": [], "testFiles": [],
            })
        if any(term in lower for term in ("prompt", "agent", "model", "eval", "rag", "retrieval")):
            surfaces.append({
                "id": stable_id("SURFACE", "ai", rel), "kind": "ai", "name": file.stem,
                "source": rel, "risks": ["nondeterminism", "evaluation", "safety", "monitoring"],
                "apiIds": [], "flowIds": [], "testFiles": [],
            })
        if not is_test_source:
            for pattern in API_ROUTE_PATTERNS:
                for match in pattern.finditer(text):
                    method, route = match.groups()
                    apis.append({
                        "id": stable_id("API", method.lower(), route), "operationId": f"{method.lower()}-{route}",
                        "method": method.upper(), "path": route, "source": rel,
                        "summary": f"Source-declared {method.upper()} {route}", "security": "unknown",
                        "parameters": [], "requestFields": [], "responses": [],
                        "idempotencyDeclared": False, "uiFlows": [], "evidence": "source",
                    })
    for raw in explicit_specs or []:
        specs.append(Path(raw).expanduser().resolve())
    apis = enrich_source_apis(root, apis, source_cache)
    for spec in dict.fromkeys(specs):
        document, warning = parse_document(spec)
        if warning:
            warnings.append(warning)
        if document and ("openapi" in document or "swagger" in document):
            apis.extend(openapi_operations(spec, document))
    for raw in har_paths or []:
        hars.append(Path(raw).expanduser().resolve())
    for har in dict.fromkeys(hars):
        try:
            har_apis, har_flows = parse_har(har)
            apis.extend(har_apis)
            flows.extend(har_flows)
        except (OSError, ValueError) as error:
            warnings.append(f"Could not read HAR {har}: {error}")
    flows.extend(infer_source_flows(source_cache))
    apis = merge_by_id(old_model.get("apis", []), apis)
    retained_flows = [
        flow for flow in old_model.get("flows", [])
        if flow.get("evidence") != "source-inference"
    ]
    flows = merge_by_id(retained_flows, flows)
    surfaces = merge_by_id(old_model.get("surfaces", []), surfaces)
    flows_by_id = {str(flow.get("id")): flow for flow in flows}
    for surface in surfaces:
        source = str(surface.get("source") or "")
        text = source_cache.get(source, "")
        surface["apiIds"] = sorted({
            api["id"] for api in apis
            if api.get("path") and str(api["path"]) in text
        } | set(surface.get("apiIds") or []))
        surface["testFiles"] = sorted({
            test for test in tests
            if surface.get("name") and str(surface["name"]).lower() in source_cache.get(test, "").lower()
        } | set(surface.get("testFiles") or []))
        retained_flow_ids = {
            flow_id for flow_id in (surface.get("flowIds") or [])
            if flow_id in flows_by_id and flows_by_id[flow_id].get("evidence") != "source-inference"
        }
        inferred_flow_ids = {
            flow_id for flow_id, flow in flows_by_id.items()
            if flow.get("evidence") == "source-inference" and source in (flow.get("sources") or [])
        }
        surface["flowIds"] = sorted(retained_flow_ids | inferred_flow_ids)
    model = {
        "schema": MODEL_SCHEMA,
        "project": {"name": root.name, "root": str(root)},
        "surfaces": surfaces, "apis": apis, "flows": flows,
        "dependencies": old_model.get("dependencies", []),
        "changedFiles": git_changed_files(root),
        "testFiles": sorted(set(tests)),
        "warnings": sorted(set(warnings)), "updatedAt": utc_now(),
    }
    atomic_json(paths["model"], model)
    update_coverage(root, model)
    return model


def affected_surfaces(model: dict[str, Any], changed_files: list[str]) -> list[dict[str, Any]]:
    changed = set(changed_files)
    affected = []
    for surface in model.get("surfaces", []):
        source = str(surface.get("source") or "")
        if source in changed or any(source and (source.startswith(item.rsplit("/", 1)[0] + "/") or item.startswith(source.rsplit("/", 1)[0] + "/")) for item in changed):
            affected.append(surface)
            continue
        api_ids = set(surface.get("apiIds") or [])
        if any(api.get("id") in api_ids and str(api.get("source")) in changed for api in model.get("apis", [])):
            affected.append(surface)
    return affected


def candidate_test(identifier: str, title: str, priority: str, kind: str, reason: str,
                   steps: list[str], oracle: str, adapter: str, provenance: list[str],
                   surface_ids: list[str] | None = None, api_operation: str = "",
                   flow_ids: list[str] | None = None, flags: dict[str, bool] | None = None) -> dict[str, Any]:
    return {
        "id": identifier, "title": title, "priority": priority, "type": kind,
        "flow": flow_ids[0] if flow_ids else "", "item": "", "reason": reason,
        "provenance": provenance, "prerequisites": [], "data": ["synthetic authorized test data"],
        "steps": steps, "oracle": oracle,
        "oracleBasis": "risk-derived product expectation; refine from an authoritative requirement when available",
        "predictedOutcome": "unknown",
        "predictionBasis": "No source-backed outcome prediction recorded before execution",
        "estimatedSeconds": 90,
        "dependencies": [], "parallelGroup": adapter, "adapter": adapter,
        "flags": flags or {}, "lifecycle": "candidate",
        "surfaceIds": surface_ids or [], "flowIds": flow_ids or [],
        "apiOperation": api_operation, "evidenceRequirements": [
            "actual result and environment/build",
            "reproducible evidence handle",
            "for every tested browser page/state: safe screenshot, DOM/accessibility snapshot, console/page errors, bounded redacted network/timing, synthetic data, and explicit unavailable or privacy-withheld channels",
            "general issue review across every collected artifact before reporting",
        ],
    }


def intent_candidates(root: Path, model: dict[str, Any]) -> list[dict[str, Any]]:
    path = framework_paths(root)["intentTests"]
    if not path.exists():
        return []
    library = ensure_intent_library(path, root, create=False)
    if not isinstance(library, dict) or library.get("schema") != INTENT_LIBRARY_SCHEMA:
        return []
    candidates: list[dict[str, Any]] = []
    entities = list(model.get("surfaces", [])) + list(model.get("flows", []))
    for intent in library.get("tests", []):
        if not isinstance(intent, dict) or intent.get("status") != "active":
            continue
        scope = str(intent.get("scope") or "").lower()
        matched = [entity.get("id") for entity in entities if scope and any(
            token and token.lower() in scope
            for token in (str(entity.get("id") or ""), str(entity.get("name") or ""), str(entity.get("source") or ""))
        )]
        surface_ids = [identifier for identifier in matched if str(identifier).startswith("SURFACE-")]
        flow_ids = [identifier for identifier in matched if str(identifier).startswith("FLOW-")]
        prompt = str(intent.get("intentPrompt") or "").strip()
        oracle = "; ".join(str(row) for row in intent.get("oracles", []) if str(row).strip())
        test = candidate_test(
            str(intent.get("id")), str(intent.get("title") or "Prompt-defined CARBON test"),
            str(intent.get("priority") or "P2"), "intent",
            "User-authored or reviewed AI test intent from the local CARBON test library.",
            [
                f"Establish the declared scope and setup: {scope or 'resolve from current project context'}",
                f"Fulfill this test intent without weakening its safety boundary: {prompt}",
                "Capture the actual user-visible result, authoritative state, recovery, environment/build, and a reproducible evidence handle.",
            ],
            oracle or "The coding agent must resolve and show an objective oracle before execution; otherwise keep this test blocked or planned.",
            str(intent.get("adapter") or "auto"),
            [str(path), json.dumps(intent.get("source") or {"kind": "prompt"}, sort_keys=True)],
            surface_ids=surface_ids, flow_ids=flow_ids, flags=dict(intent.get("safety") or {}),
        )
        test.update({
            "claim": prompt,
            "userPromise": prompt,
            "intentRevision": intent.get("revision", 1),
            "intentReadiness": intent.get("readiness", {}),
            "evidenceRequirements": list(intent.get("evidenceRequirements") or test["evidenceRequirements"]),
            "data": list(intent.get("inputs") or ["synthetic authorized test data"]),
            "prerequisites": list(intent.get("setup") or []),
            "tags": list(intent.get("tags") or []),
        })
        candidates.append(test)
    return candidates


def definition_hash(test: dict[str, Any]) -> str:
    semantic = {
        key: test.get(key)
        for key in (
            "title", "type", "steps", "oracle", "oracleBasis",
            "predictedOutcome", "predictionBasis", "adapter", "surfaceIds",
            "flowIds", "apiOperation", "flags", "evidenceRequirements",
        )
    }
    encoded = json.dumps(semantic, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return hashlib.sha256(encoded.encode()).hexdigest()


def api_candidates(apis: list[dict[str, Any]], source: str) -> list[dict[str, Any]]:
    tests: list[dict[str, Any]] = []
    for api in apis:
        method, route = str(api.get("method")), str(api.get("path"))
        op = str(api.get("operationId") or f"{method}-{route}")
        base = re.sub(r"[^A-Z0-9]+", "-", op.upper()).strip("-")[:42]
        provenance = [source, str(api.get("source") or ""), f"{method} {route}"]
        code_contracts = [str(value) for value in api.get("codeContracts", []) if str(value).strip()]
        adapter_name = f"direct-http:{api.get('evidence') or 'source'}"
        contract_step = (
            "Apply the source-derived contracts relevant to this operation: "
            + " | ".join(code_contracts[:16])
            if code_contracts
            else "Resolve validation, enum, boundary, default, and payload-size contracts from code before execution"
        )
        provenance.extend(code_contracts[:24])
        flow_ids = list(api.get("uiFlows") or [])
        tests.append(candidate_test(f"API-{base}-POS", f"{method} {route} valid contract", "P1", "api-positive",
            "Prove the documented successful contract and visible/persisted effect.",
            ["Create isolated synthetic preconditions", contract_step, f"Send a valid {method} request to {route}", "Verify status, headers, body shape, state, and linked UI flow"],
            "Documented success status and schema hold; state and linked UI agree without unintended effects.", adapter_name, provenance, api_operation=op, flow_ids=flow_ids))
        required = [field["name"] for field in api.get("requestFields", []) if field.get("required")]
        tests.append(candidate_test(f"API-{base}-NEG", f"{method} {route} missing or malformed input", "P1", "api-negative",
            "Reject invalid input consistently without partial mutation or sensitive disclosure.",
            [contract_step, f"Omit one required field: {', '.join(required) if required else 'a contract-required value'}", "Send wrong type/format and unknown field variants", "Verify error envelope and unchanged state"],
            "Invalid requests fail with stable 4xx semantics, actionable errors, no 5xx, and no partial effect.", adapter_name, provenance, api_operation=op, flow_ids=flow_ids))
        tests.append(candidate_test(f"API-{base}-BND", f"{method} {route} exact boundaries", "P2", "api-boundary",
            "Locate declared and representation boundaries instead of sampling arbitrary values.",
            [contract_step, "Exercise empty/null where distinct", "Test min-1/min/min+1 and max-1/max/max+1 where declared", "Include Unicode and encoded-size variants"],
            "Every boundary is enforced at the declared counting unit and remains consistent through storage and UI.", adapter_name, provenance, api_operation=op, flow_ids=flow_ids))
        tests.append(candidate_test(f"API-{base}-AUTH", f"{method} {route} authentication and authorization", "P0", "api-authorization",
            "Prevent unauthenticated, wrong-role, wrong-owner, and cross-tenant access.",
            ["Use synthetic signed-out and expired identities", "Use wrong-role/wrong-owner identity when authorized", "Compare response, timing, state, logs, and linked UI"],
            "Access is denied consistently with no resource existence, data, side-effect, or cross-tenant leakage.", adapter_name, provenance, api_operation=op, flow_ids=flow_ids, flags={"security": True}))
        if method in {"POST", "PUT", "PATCH", "DELETE"}:
            tests.append(candidate_test(f"API-{base}-IDEM", f"{method} {route} retry and idempotency", "P0", "api-idempotency",
                "Retries and duplicate delivery must not duplicate money, notifications, records, or state transitions.",
                ["Send one authorized synthetic request with a stable idempotency key where supported", "Repeat after success and after a simulated ambiguous timeout", "Inspect final state, audit, events, and linked UI"],
                "At-most-once business effect or the documented idempotent result; conflicts are explicit and recoverable.", adapter_name, provenance, api_operation=op, flow_ids=flow_ids, flags={"destructive": True}))
        tests.append(candidate_test(f"API-{base}-RES", f"{method} {route} timeout, retry, and dependency recovery", "P2", "api-resilience",
            "Prove bounded recovery and truthful terminal states.",
            ["Use an authorized fault-injection or local stub", "Exercise timeout, 429, dependency 5xx, and response-loss variants", "Verify retry budget, cancellation, cleanup, observability, and UI recovery"],
            "No retry storm, false success, orphaned work, duplicate effect, or unrecoverable UI state.", adapter_name, provenance, api_operation=op, flow_ids=flow_ids, flags={"costly": True}))
    return tests


def baseline_domain_candidates(
    model: dict[str, Any],
    affected: list[dict[str, Any]],
    target: str,
) -> list[dict[str, Any]]:
    """Create a broad candidate pool without pretending generic checks already ran."""
    surface_ids = [str(surface.get("id")) for surface in affected if surface.get("id")]
    flow_ids = sorted({
        str(flow_id)
        for surface in affected
        for flow_id in surface.get("flowIds", [])
        if flow_id
    })
    live_target = str(target).startswith(("http://", "https://")) or any(
        str(surface.get("kind") or "").lower() in {"ui", "web", "page"}
        for surface in affected
    )
    candidates: list[dict[str, Any]] = []
    for domain in testing_domains():
        domain_id = str(domain["id"])
        title = str(domain.get("title") or domain_id.replace("-", " ").title())
        purpose = str(domain.get("purpose") or f"Evaluate {title.lower()} risk.")
        families = [str(value) for value in domain.get("families", []) if str(value).strip()]
        if len(families) < DOMAIN_BASELINE_COUNT:
            raise ValueError(f"CARBON domain {domain_id} needs at least {DOMAIN_BASELINE_COUNT} baseline families.")
        if domain_id == "api" and model.get("apis"):
            adapter = "direct-http:discovered-source"
        elif domain_id in {"data"} or not live_target:
            adapter = "code-inspection"
        else:
            adapter = "isolated-browser"
        for index, family in enumerate(families[:DOMAIN_BASELINE_COUNT], start=1):
            identifier = f"DOMAIN-{domain_id.upper().replace('-', '_')}-{index:02d}"
            test = candidate_test(
                identifier,
                f"{title} · {family}",
                "P1" if index == 1 else "P2",
                f"domain-{domain_id}",
                f"Establish a minimum broad-run baseline for {family}; {purpose}",
                [
                    f"Resolve the product promise, code paths, states, and evidence channels relevant to {family}.",
                    "Create safe synthetic setup and representative starting state; keep unavailable setup explicit.",
                    "Exercise the primary behavior plus the most consequential negative, boundary, interruption, or wrong-state variant that applies.",
                    f"Inspect the available evidence for {title.lower()} impact and compare it with source, requirements, and user-visible behavior.",
                    "Record the observed result, evidence handle, blocked channels, and the smallest useful follow-up check.",
                ],
                f"Observed behavior addresses {family} without contradicting the product promise, authoritative state, or applicable {title.lower()} expectation; otherwise record a specific failure or unknown.",
                adapter,
                [str(TESTING_DOMAINS_PATH), f"domain:{domain_id}", f"family:{family}"],
                surface_ids=surface_ids,
                flow_ids=flow_ids,
                flags={"security": domain_id == "security"},
            )
            test.update({
                "domain": domain_id,
                "coverageTrack": "top-level-domain",
                "baselineIndex": index,
                "estimatedSeconds": 60,
            })
            candidates.append(test)
    return candidates


def business_value_candidates(
    affected: list[dict[str, Any]],
    target: str,
) -> list[dict[str, Any]]:
    surface_ids = [str(surface.get("id")) for surface in affected if surface.get("id")]
    flow_ids = sorted({
        str(flow_id)
        for surface in affected
        for flow_id in surface.get("flowIds", [])
        if flow_id
    })
    adapter = "isolated-browser" if str(target).startswith(("http://", "https://")) or affected else "code-inspection"
    definitions = [
        (
            "BV-001", "Primary customer outcome works end to end", "P0",
            "Verify that the product's central promise produces the intended customer outcome, not merely a technically successful screen or response.",
            ["Identify the intended user, job, promised outcome, and observable success state from product context.", "Complete the shortest safe primary journey with representative synthetic data.", "Verify user-visible outcome, authoritative state, downstream effect, and absence of a value-breaking workaround."],
            "The intended customer reaches the promised outcome completely and truthfully, with authoritative state and visible result aligned.",
        ),
        (
            "BV-002", "First use reaches meaningful value", "P1",
            "Measure whether a new user can reach a useful result without hidden setup, insider knowledge, or avoidable delay.",
            ["Start from the realistic first-use state.", "Follow visible onboarding and the primary call to action without undocumented help.", "Record prerequisites, dead ends, time-to-value, and the first meaningful result."],
            "A representative first-time user can discover the path and reach meaningful value within the product's stated expectations.",
        ),
        (
            "BV-003", "Product output is useful and decision-ready", "P1",
            "A response can be technically valid yet fail the user's real decision or work need.",
            ["Identify the decision or next action the output is meant to support.", "Generate or retrieve representative outputs, including one difficult but common case.", "Check accuracy, completeness, clarity, actionability, provenance, and dangerous ambiguity."],
            "The output is sufficiently correct, complete, understandable, and actionable for the intended user's real task.",
        ),
        (
            "BV-004", "Failure and recovery preserve customer value", "P1",
            "The product should preserve work, trust, and the ability to finish when dependencies or users fail.",
            ["Interrupt the primary value path at a safe reversible boundary.", "Exercise retry, resume, correction, and cancellation where supported.", "Verify no silent loss, duplicate business effect, false success, or unrecoverable dead end."],
            "A recoverable problem does not destroy user work or business effect, and the user can understand and complete the intended outcome safely.",
        ),
        (
            "BV-005", "Claims, access, limits, and delivered value agree", "P1",
            "Verify that what the product promises and permits matches what users actually receive across plans, roles, limits, and important states.",
            ["Inventory relevant product claims, plan or role boundaries, quotas, and visible calls to action.", "Inspect one representative allowed state and one restricted or exhausted state without making a payment.", "Compare messaging, access, outcome, limit accounting, upgrade path, and authoritative enforcement."],
            "Claims and entitlements are truthful and consistently enforced without charging, blocking, or misleading users contrary to the stated value proposition.",
        ),
    ]
    candidates: list[dict[str, Any]] = []
    for index, (identifier, title, priority, reason, steps, oracle) in enumerate(definitions, start=1):
        test = candidate_test(
            identifier, title, priority, "business-value", reason, steps, oracle, adapter,
            ["CARBON business-value baseline", "current product context"],
            surface_ids=surface_ids, flow_ids=flow_ids,
        )
        test.update({
            "domain": "business-value",
            "coverageTrack": "business-value",
            "baselineIndex": index,
            "estimatedSeconds": 75 if index == 1 else 60,
        })
        candidates.append(test)
    return candidates


def initial_execution_selection(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Keep the first experience broad, risk-aware, and within a 25-minute plan."""
    ranked = sorted(candidates, key=lambda row: (row.get("priority", "P3"), row.get("estimatedSeconds", 90), row.get("id", "")))
    required = [
        row for row in ranked
        if int(row.get("baselineIndex") or 0) == 1
        and row.get("coverageTrack") == "business-value"
    ]
    required.extend(row for row in ranked if row.get("type") == "api-positive")
    selected: list[dict[str, Any]] = []
    selected_ids: set[str] = set()
    elapsed = 0
    for row in required + ranked:
        identifier = str(row.get("id") or "")
        if not identifier or identifier in selected_ids or len(selected) >= INITIAL_TEST_LIMIT:
            continue
        estimate = max(1, int(row.get("estimatedSeconds") or 90))
        if selected and elapsed + estimate > INITIAL_BUDGET_SECONDS:
            continue
        selected.append(row)
        selected_ids.add(identifier)
        elapsed += estimate
    return selected


def merge_candidates(root: Path, tests: list[dict[str, Any]]) -> dict[str, Any]:
    paths = ensure_framework(root)
    suite = load_json(paths["regression"], {})
    existing = {row["id"]: row for row in suite.get("tests", []) if isinstance(row, dict) and row.get("id")}
    for test in tests:
        prior = existing.get(test["id"], {})
        current_hash = definition_hash(test)
        prior_hash = prior.get("definitionHash")
        lifecycle = prior.get("lifecycle", test.get("lifecycle", "candidate"))
        stale_reason = prior.get("staleReason", "")
        if prior_hash and prior_hash != current_hash and lifecycle in {"repeatable", "regression"}:
            lifecycle = "needs-review"
            stale_reason = "The generated steps, oracle, adapter, mapping, or safety flags changed; review and re-observe this definition."
        existing[test["id"]] = {
            **test,
            "lifecycle": lifecycle,
            "definitionHash": current_hash,
            "staleReason": stale_reason,
            "evidence": prior.get("evidence", []),
            "outcomes": prior.get("outcomes", []),
            "createdAt": prior.get("createdAt", utc_now()),
            "updatedAt": utc_now(),
        }
    suite = {"schema": REGRESSION_SCHEMA, "tests": sorted(existing.values(), key=lambda row: row["id"]), "updatedAt": utc_now()}
    atomic_json(paths["regression"], suite)
    return suite


def update_coverage(root: Path, model: dict[str, Any] | None = None) -> dict[str, Any]:
    paths = ensure_framework(root)
    model = model or load_json(paths["model"], {})
    suite = load_json(paths["regression"], {})
    links = []
    gaps = []
    entities = list(model.get("surfaces", [])) + list(model.get("apis", [])) + list(model.get("flows", []))
    for entity in entities:
        identifier = entity.get("id")
        matching = [
            test for test in suite.get("tests", [])
            if identifier in (test.get("surfaceIds") or [])
            or identifier in (test.get("flowIds") or [])
            or identifier == test.get("apiOperation")
            or entity.get("operationId") == test.get("apiOperation")
        ]
        trusted = [test for test in matching if test.get("lifecycle") == "regression"]
        links.append({
            "entityId": identifier, "kind": entity.get("kind") or ("api" if entity.get("method") else "flow"),
            "candidateTestIds": [test["id"] for test in matching],
            "regressionTestIds": [test["id"] for test in trusted],
            "status": "regression" if trusted else ("candidate" if matching else "uncovered"),
        })
        if not matching:
            gaps.append({"entityId": identifier, "name": entity.get("name") or entity.get("operationId") or entity.get("path"), "priority": "P1", "reason": "No mapped candidate or regression test."})
    coverage = {"schema": COVERAGE_SCHEMA, "links": links, "gaps": gaps, "updatedAt": utc_now()}
    atomic_json(paths["coverage"], coverage)
    return coverage


def choose_test_strategy(requested: str, size: str, changed: list[str], affected: list[dict[str, Any]], apis: list[dict[str, Any]]) -> dict[str, str]:
    requested = requested.lower()
    if requested in {"breadth", "depth", "balanced"}:
        return {
            "requested": requested,
            "selected": requested,
            "rationale": f"The user explicitly requested {requested} coverage.",
        }
    risk_text = " ".join(changed + [str(row.get("name") or row.get("source") or "") for row in affected] + [str(row.get("operationId") or row.get("path") or "") for row in apis]).lower()
    deep_risk = any(token in risk_text for token in ("auth", "permission", "tenant", "payment", "billing", "migration", "schema", "database", "security", "privacy", "prompt", "model", "agent", "rag"))
    if size == "large":
        selected = "balanced"
        rationale = "Large runs map the broad surface first, then drill into high-risk journeys and boundaries."
    elif size == "medium":
        selected = "balanced"
        rationale = "Medium runs combine representative surface coverage with deeper checks on the riskiest changed flow."
    elif deep_risk:
        selected = "depth"
        rationale = "A high-impact state, data, identity, security, or AI boundary was detected in the current change."
    else:
        selected = "breadth"
        rationale = "The change appears localized, so the fastest confidence gain comes from sampling changed and adjacent surfaces."
    return {"requested": "auto", "selected": selected, "rationale": rationale}


def generated_personas(size: str, affected: list[dict[str, Any]], changed: list[str]) -> list[dict[str, Any]]:
    provenance = [str(value) for value in changed[:8]] or [str(row.get("source") or row.get("name") or "discovered product surface") for row in affected[:8]]
    roster = [
        {"id": "PERSONA-PRIMARY", "name": "Primary user", "role": "frequent task owner", "goal": "complete the changed core task efficiently", "risks": ["functional break", "state loss", "unclear recovery"]},
        {"id": "PERSONA-FIRST-TIME", "name": "First-time user", "role": "new or infrequent user", "goal": "understand the page and complete the task without prior knowledge", "risks": ["discoverability", "copy ambiguity", "dead end"]},
        {"id": "PERSONA-POWER", "name": "Power user", "role": "high-frequency advanced user", "goal": "complete the task quickly across realistic state and data volume", "risks": ["workflow friction", "keyboard gaps", "performance"]},
        {"id": "PERSONA-ACCESS", "name": "Access-needs user", "role": "keyboard, zoom, or assistive-technology user", "goal": "complete the same product promise without a mouse-only or visual-only dependency", "risks": ["focus loss", "unlabeled control", "reflow failure"]},
        {"id": "PERSONA-RECOVERY", "name": "Interrupted user", "role": "user recovering from interruption or dependency failure", "goal": "resume safely without duplicate work or silent data loss", "risks": ["retry duplication", "stale state", "false success"]},
        {"id": "PERSONA-BOUNDARY", "name": "Boundary-role user", "role": "wrong-role, wrong-owner, or neighboring-tenant user", "goal": "observe only permitted data and actions", "risks": ["authorization leak", "cross-tenant effect", "unsafe affordance"]},
    ]
    limit = {"quick": 2, "medium": 4, "large": 6}[size]
    personas = []
    for row in roster[:limit]:
        goal = str(row["goal"])
        personas.append({
            **row,
            "intent": goal,
            "plan": [
                "Orient to the product and find the most plausible starting point without test-script hints.",
                f"Attempt to {goal} using realistic user data and behavior.",
                "Verify the visible result, persisted state, and any linked downstream effect.",
                "Explore one adjacent path or recovery action and record delight, confusion, friction, and surprise.",
            ],
            "requiredData": ["realistic non-production account and task data when the journey requires them"],
            "successSignals": ["the intended outcome is understandable, completed, and recoverable without hidden knowledge"],
            "derivedFrom": provenance,
        })
    return personas


def exploration_charters(strategy: str, affected: list[dict[str, Any]]) -> list[dict[str, Any]]:
    labels = [str(row.get("name") or row.get("id") or "surface") for row in affected[:12]]
    target_text = ", ".join(labels) or "the changed and adjacent product surface"
    breadth = {
        "id": "CHARTER-BREADTH",
        "strategy": "breadth",
        "title": "Map the reachable experience",
        "mission": f"Sample routes, widgets, roles, page states, transitions, and linked APIs across {target_text}; record success, delight, joy, confusion, surprising behavior, and coverage gaps.",
        "stopConditions": ["representative changed and adjacent surfaces sampled", "severe failure", "authorization boundary", "new surfaces stop changing the risk picture"],
    }
    depth = {
        "id": "CHARTER-DEPTH",
        "strategy": "depth",
        "title": "Drill into the riskiest journey",
        "mission": "Follow the highest-risk changed behavior through negative and boundary inputs, interruption, retry, recovery, role/state changes, persistence, and linked API effects; record success, delight, confusion, and surprise.",
        "stopConditions": ["risk claim has direct evidence", "failure reproduced", "unsafe action requires approval", "additional depth no longer changes the decision"],
    }
    if strategy == "breadth":
        return [breadth]
    if strategy == "depth":
        return [depth]
    return [breadth, depth]


def orchestration_plan(root: Path, request: str, target: str, size: str, model: dict[str, Any], strategy: str = "auto", coverage_profile: str = "auto") -> dict[str, Any]:
    global_context = safe_execution_context()
    changed = model.get("changedFiles", [])
    affected = affected_surfaces(model, changed)
    if not affected:
        affected = model.get("surfaces", [])[:8 if size == "quick" else 30]
    affected_api_ids = {api_id for surface in affected for api_id in surface.get("apiIds", [])}
    affected_apis = [api for api in model.get("apis", []) if api.get("id") in affected_api_ids or str(api.get("source")) in changed]
    if not affected_apis and model.get("apis"):
        api_limit = {"quick": 8, "medium": 30, "large": 200}[size]
        affected_apis = list(model.get("apis", []))[:api_limit]
    test_strategy = choose_test_strategy(strategy, size, changed, affected, affected_apis)
    personas = generated_personas(size, affected, changed)
    charters = exploration_charters(test_strategy["selected"], affected)
    tests = api_candidates(affected_apis, "automatic change/API mapping")
    managed_intents = intent_candidates(root, model)
    intent_limit = {"quick": 20, "medium": 75, "large": 1000}[size]
    tests.extend(sorted(managed_intents, key=lambda row: (row["priority"], row["id"]))[:intent_limit])
    for surface in affected:
        sid = surface["id"]
        kind = surface.get("kind", "ui")
        adapter = "ai-eval" if kind == "ai" else "isolated-browser"
        tests.append(candidate_test(
            stable_id("AUTO", sid, "primary"), f"{surface.get('name')} primary behavior and adjacent regression", "P1", f"{kind}-change",
            "Validate the changed or highest-risk user promise and its direct consumers.",
            ["Establish isolated synthetic preconditions", "Exercise the primary reversible behavior", "Check positive, negative, boundary, recovery, accessibility, and linked API state as relevant"],
            "The user-visible promise, service state, persistence, and recovery agree without a severe adjacent regression.",
            adapter, [str(surface.get("source") or ""), "automatic affected-surface mapping"], surface_ids=[sid], flow_ids=list(surface.get("flowIds") or []),
        ))
    broad_baseline = coverage_profile == "broad" or (
        coverage_profile == "auto"
        and request.strip().lower() in {"test this", "carbon", "/carbon", "initial carbon run", "clean carbon run"}
    )
    if broad_baseline:
        tests.extend(baseline_domain_candidates(model, affected, target))
        tests.extend(business_value_candidates(affected, target))
    if not tests:
        tests.append(candidate_test(
            "AUTO-REVIEW-001", "Inspect current change and establish executable coverage", "P1", "review",
            "No executable surface was confidently detected; preserve uncertainty instead of inventing a pass.",
            ["Inspect current changed files and repository guidance", "Identify runtime boundary and direct consumers", "Create concrete executable candidates or record the blocker"],
            "Every material change has a named risk, consumer, executable adapter, and evidence requirement—or an explicit unknown.",
            "code-inspection", changed or ["user request"],
        ))
    deduped = {test["id"]: test for test in tests}
    generated = sorted(deduped.values(), key=lambda row: (row["priority"], row["estimatedSeconds"], row["id"]))
    if size == "quick" and broad_baseline:
        selected = initial_execution_selection(generated)
    else:
        max_tests = {"quick": 30, "medium": 150, "large": 1000}[size]
        selected = generated[:max_tests]
    merge_candidates(root, generated)
    selected_adapter_names = {test["adapter"] for test in selected} | {"code-inspection"}
    browser_selected = "isolated-browser" in selected_adapter_names
    adapters = sorted(
        "carbon-browser/v1" if adapter == "isolated-browser" else adapter
        for adapter in selected_adapter_names
    )
    browser_contract = {
        "protocol": "carbon-browser/v1",
        "requiredOperations": [
            "session.start", "session.stop", "page.open", "discover.map",
            "action.click", "capture.screenshot",
        ],
        "requiredEvidence": ["screenshot"],
        "desiredEvidence": [
            "dom", "accessibility", "console", "network", "performance",
            "storage", "trace", "video", "recording",
        ],
        "adapterFamilies": [
            "host-agent", "vibium-webdriver-bidi", "playwright-cdp",
            "selenium-webdriver", "testers-ai-managed-devices",
        ],
        "negotiationRequired": True,
        "degradationPolicy": "Never silently downgrade. Select another adapter or explicitly re-plan with every missing operation and evidence channel visible.",
        "sessionPath": ".carbon/browser/sessions/<session-id>/",
    } if browser_selected else None
    project_memory = refresh_memory(root)
    selected_ids = {test["id"] for test in selected}
    deferred = [test for test in generated if test["id"] not in selected_ids]
    domain_counts: dict[str, int] = {}
    for test in generated:
        domain = str(test.get("domain") or "context-derived")
        domain_counts[domain] = domain_counts.get(domain, 0) + 1
    plan = {
        "title": "CARBON automatic validation",
        "mode": size, "target": target, "environment": "local isolated test environment",
        "revision": "working tree", "scope": [request, "changed and adjacent behavior"],
        "exclusions": ["destructive, paid, public, notification, or high-volume actions without scoped approval"],
        "assumptions": [
            "Generated cases remain candidates until runtime evidence validates their steps and oracles.",
            *( ["The configured global execution prompt and context documents apply without weakening test or safety boundaries."] if global_context["configured"] else [] ),
        ],
        "testingDoctrine": {
            "schema": testing_doctrine()["schema"],
            "version": testing_doctrine()["version"],
            "northStar": testing_doctrine()["northStar"],
            "appliedPrinciples": [principle["id"] for principle in testing_doctrine()["principles"]],
            "practiceBalances": [spice["id"] for spice in testing_doctrine()["spices"]],
            "rule": "Apply the pro, con, and CARBON default for each selected practice; discovery is not trusted regression evidence.",
        },
        "projectMemory": {
            "schema": project_memory["schema"],
            "freshness": project_memory["freshness"],
            "changedDomains": project_memory["changedDomains"],
            "invalidations": project_memory["invalidations"],
            "rule": project_memory["rule"],
        },
        "detectedFlows": [flow.get("name") for flow in model.get("flows", [])[:30]],
        "detectedItems": [surface.get("name") for surface in affected],
        "availableAdapters": adapters,
        "browserContract": browser_contract,
        "setupDependencies": [document["path"] for document in global_context["contextDocuments"]],
        "credentialSources": (["CARBON global execution credential profile"] if global_context["passwordConfigured"] or global_context["username"] else []),
        "globalExecutionContext": global_context,
        "stopConditions": ["severe failure", "shared environment blocker", "authorization boundary", "budget exhausted", "new evidence no longer changes the risk picture"],
        "residualUncertainty": model.get("warnings", []),
        "testStrategy": test_strategy,
        "personas": personas,
        "explorationCharters": charters,
        "coverageProfile": "broad" if broad_baseline else "focused",
        "generatedCandidates": len(generated),
        "deferredCandidates": len(deferred),
        "deferredTestIds": [test["id"] for test in deferred],
        "coverageByDomain": domain_counts,
        "businessValue": {
            "required": broad_baseline,
            "candidateCount": domain_counts.get("business-value", 0),
            "selectedTestIds": [test["id"] for test in selected if test.get("domain") == "business-value"],
        },
        "executionCutLine": {
            "selected": len(selected),
            "generated": len(generated),
            "deferred": len(deferred),
            "maxInitialTests": INITIAL_TEST_LIMIT if broad_baseline and size == "quick" else len(selected),
            "budgetMinutes": 25 if broad_baseline and size == "quick" else max(1, round(sum(test["estimatedSeconds"] for test in selected) / 60)),
            "reason": "The first CARBON pass requires a business-value check, then spends the remaining budget on the highest-risk context-derived and domain checks. Every top-level domain remains represented in the durable candidate pool for /carbon-more.",
        },
        "concurrency": 1, "estimatedTests": len(selected),
        "estimatedMinutes": max(1, round(sum(test["estimatedSeconds"] for test in selected) / 60)),
        "tests": selected,
    }
    paths = ensure_framework(root)
    atomic_json(paths["plan"], plan)
    atomic_json(paths["base"] / "candidate-pool.json", {
        "schema": "carbon.candidate-pool/v1",
        "generatedAt": utc_now(),
        "request": request,
        "target": target,
        "coverageProfile": plan["coverageProfile"],
        "continuationSelectedTestIds": [],
        "tests": generated,
    })
    orchestration = {
        "schema": SCHEMA, "request": request, "target": target, "size": size,
        "coverageProfile": plan["coverageProfile"],
        "generatedCandidates": len(generated), "selectedTests": len(selected), "deferredCandidates": len(deferred),
        "businessValueCandidateIds": [test["id"] for test in generated if test.get("domain") == "business-value"],
        "testStrategy": test_strategy,
        "personaIds": [persona["id"] for persona in personas],
        "explorationCharterIds": [charter["id"] for charter in charters],
        "changedFiles": changed, "affectedSurfaceIds": [surface["id"] for surface in affected],
        "affectedApiIds": [api["id"] for api in affected_apis], "selectedAdapters": adapters,
        "browserContract": browser_contract,
        "activeIntentTestIds": [test["id"] for test in managed_intents],
        "globalExecutionContextConfigured": global_context["configured"],
        "testingDoctrineVersion": testing_doctrine()["version"],
        "projectMemoryFreshness": project_memory["freshness"],
        "plan": str(paths["plan"]), "next": "Initialize a durable CARBON run, execute the cheapest decisive P0/P1 checks, then adapt from evidence.",
        "approvalGates": ["mutation", "destructive or high-volume execution", "live identity", "payment or notification", "public or paid cloud execution", "release"],
        "createdAt": utc_now(),
    }
    atomic_json(paths["orchestration"], orchestration)
    update_coverage(root, model)
    return orchestration


def more_coverage_plan(root: Path, limit: int, budget_minutes: int) -> dict[str, Any]:
    paths = ensure_framework(root)
    pool = load_json(paths["base"] / "candidate-pool.json", {})
    if pool.get("schema") != "carbon.candidate-pool/v1" or not isinstance(pool.get("tests"), list):
        raise ValueError("No broad CARBON candidate pool exists. Run /carbon once before /carbon-more.")
    initial = load_json(paths["plan"], {})
    already_selected = {str(test.get("id")) for test in initial.get("tests", []) if isinstance(test, dict) and test.get("id")}
    already_selected.update(str(test_id) for test_id in pool.get("continuationSelectedTestIds", []) if test_id)
    suite = load_json(paths["regression"], {})
    completed = {
        str(test.get("id"))
        for test in suite.get("tests", [])
        if isinstance(test, dict) and test.get("lifecycle") in {"observed", "repeatable", "regression"}
    }
    remaining = [
        test for test in pool["tests"]
        if isinstance(test, dict) and str(test.get("id") or "") not in already_selected | completed
    ]
    budget_seconds = max(5, min(24 * 60, budget_minutes)) * 60
    selected: list[dict[str, Any]] = []
    elapsed = 0
    for test in sorted(remaining, key=lambda row: (row.get("priority", "P3"), row.get("estimatedSeconds", 90), row.get("id", ""))):
        estimate = max(1, int(test.get("estimatedSeconds") or 90))
        if len(selected) >= limit or (selected and elapsed + estimate > budget_seconds):
            continue
        selected.append(test)
        elapsed += estimate
    if not selected:
        raise ValueError("No deferred CARBON candidates remain. Generate new context-specific coverage before another /carbon-more run.")
    selected_ids = {str(test["id"]) for test in selected}
    still_deferred = [test for test in remaining if str(test.get("id")) not in selected_ids]
    plan = {
        "title": "CARBON more coverage",
        "mode": "medium",
        "target": initial.get("target") or pool.get("target") or "auto-detect local target",
        "environment": initial.get("environment") or "local isolated test environment",
        "revision": initial.get("revision") or "working tree",
        "scope": list(initial.get("scope") or []) + ["deferred coverage from the prior CARBON candidate pool"],
        "exclusions": list(initial.get("exclusions") or []),
        "assumptions": list(initial.get("assumptions") or []),
        "stopConditions": list(initial.get("stopConditions") or []),
        "availableAdapters": sorted({str(test.get("adapter") or "manual") for test in selected}),
        "setupDependencies": list(initial.get("setupDependencies") or []),
        "credentialSources": list(initial.get("credentialSources") or []),
        "generatedCandidates": len(pool["tests"]),
        "selectedFromDeferred": len(selected),
        "deferredCandidates": len(still_deferred),
        "deferredTestIds": [test["id"] for test in still_deferred],
        "sourcePlan": str(paths["plan"]),
        "executionCutLine": {
            "selected": len(selected), "remaining": len(still_deferred),
            "maxTests": limit, "budgetMinutes": budget_minutes,
            "reason": "Continue the durable candidate pool without repeating the initial CARBON pass.",
        },
        "concurrency": 1,
        "estimatedTests": len(selected),
        "estimatedMinutes": max(1, round(elapsed / 60)),
        "tests": selected,
    }
    output = paths["base"] / "more-plan.json"
    atomic_json(output, plan)
    pool["continuationSelectedTestIds"] = sorted(already_selected | selected_ids)
    pool["updatedAt"] = utc_now()
    atomic_json(paths["base"] / "candidate-pool.json", pool)
    return {
        "schema": "carbon.more-coverage/v1",
        "plan": str(output),
        "selectedTests": len(selected),
        "estimatedMinutes": plan["estimatedMinutes"],
        "remainingCandidates": len(still_deferred),
        "generatedCandidates": len(pool["tests"]),
        "next": "Open the live execution plan, initialize a durable medium run from more-plan.json, and execute safe tests until the cut line or a stop condition.",
    }


def find_suite_test(suite: dict[str, Any], test_id: str) -> dict[str, Any]:
    for test in suite.get("tests", []):
        if test.get("id") == test_id:
            return test
    raise ValueError(f"Unknown candidate/regression test: {test_id}")


def observe_candidate(root: Path, test_id: str, run_dir: Path, run_test_id: str) -> dict[str, Any]:
    paths = ensure_framework(root)
    suite = load_json(paths["regression"], {})
    candidate = find_suite_test(suite, test_id)
    run = load_json(run_dir / "run.json")
    observed = next((row for row in run.get("tests", []) if row.get("id") == run_test_id), None)
    if not observed:
        raise ValueError(f"Run test not found: {run_test_id}")
    if observed.get("status") not in {"passed", "failed"}:
        raise ValueError("Candidate evidence must be a terminal passed or failed execution.")
    if not observed.get("actual") or not observed.get("evidence"):
        raise ValueError("Candidate evidence requires an actual result and at least one evidence handle.")
    record = {
        "runId": run.get("runId"), "runDir": str(run_dir), "testId": run_test_id,
        "revision": run.get("revision"), "environment": run.get("environment"),
        "status": observed.get("status"), "actual": observed.get("actual"),
        "evidence": observed.get("evidence"), "observedAt": utc_now(),
        "definitionHash": candidate.get("definitionHash") or definition_hash(candidate),
    }
    evidence = [row for row in candidate.get("evidence", []) if row.get("runId") != record["runId"]]
    evidence.append(record)
    candidate["evidence"] = evidence
    current_hash = candidate.get("definitionHash") or definition_hash(candidate)
    current_evidence = [row for row in evidence if row.get("definitionHash", current_hash) == current_hash]
    last_two = current_evidence[-2:]
    candidate["lifecycle"] = "repeatable" if (
        len(last_two) == 2
        and all(row.get("status") == "passed" for row in last_two)
        and len({row.get("runId") for row in last_two}) == 2
    ) else "observed"
    candidate["staleReason"] = ""
    candidate["lastOutcome"] = observed.get("status")
    candidate["updatedAt"] = utc_now()
    suite["updatedAt"] = utc_now()
    atomic_json(paths["regression"], suite)
    update_coverage(root)
    return candidate


def promote_candidate(root: Path, test_id: str, confirmed: bool) -> dict[str, Any]:
    if not confirmed:
        raise ValueError("Promotion changes the trusted regression suite; pass --confirm after reviewing repeatable evidence.")
    paths = ensure_framework(root)
    suite = load_json(paths["regression"], {})
    candidate = find_suite_test(suite, test_id)
    current_hash = candidate.get("definitionHash") or definition_hash(candidate)
    matching = [row for row in candidate.get("evidence", []) if row.get("definitionHash", current_hash) == current_hash]
    last_two = matching[-2:]
    if candidate.get("lifecycle") != "repeatable" or len(last_two) != 2 or not all(row.get("status") == "passed" for row in last_two) or len({row.get("runId") for row in last_two}) != 2:
        raise ValueError("Promotion requires the two most recent observations of the current definition to pass in distinct runs with actual results and evidence handles.")
    candidate["lifecycle"] = "regression"
    candidate["promotedAt"] = utc_now()
    candidate["updatedAt"] = utc_now()
    suite["updatedAt"] = utc_now()
    atomic_json(paths["regression"], suite)
    update_coverage(root)
    return candidate


def record_outcome(root: Path, test_id: str, disposition: str, note: str, run_id: str) -> dict[str, Any]:
    paths = ensure_framework(root)
    suite = load_json(paths["regression"], {})
    test = find_suite_test(suite, test_id)
    outcome = {"testId": test_id, "disposition": disposition, "note": note[:4000], "runId": run_id, "recordedAt": utc_now()}
    append_jsonl(paths["outcomes"], outcome)
    test.setdefault("outcomes", []).append(outcome)
    if disposition in {"dismissed", "false-positive", "flaky"}:
        test["lifecycle"] = "needs-review"
    elif disposition == "escaped":
        test["priority"] = "P0" if test.get("priority") in {"P0", "P1"} else "P1"
    test["updatedAt"] = utc_now()
    suite["updatedAt"] = utc_now()
    atomic_json(paths["regression"], suite)
    return outcome


def run_documents(root: Path) -> list[dict[str, Any]]:
    rows = []
    runs_dir = root / ".carbon" / "runs"
    if not runs_dir.is_dir():
        return rows
    for path in runs_dir.glob("*/run.json"):
        value = load_json(path, {})
        if isinstance(value, dict):
            value["_runDir"] = str(path.parent)
            rows.append(value)
    return sorted(
        rows,
        key=lambda row: (str(row.get("updatedAt", "")), str(row.get("runId", ""))),
    )


def latest_jank_benchmark(root: Path) -> dict[str, Any]:
    """Return a freshness-stamped, aggregate-only comparative snapshot.

    This deliberately stays outside the quality-confidence calculation: a peer
    average can inform where to look, but cannot establish that this product
    works, is safe, or should ship.
    """
    directory = root / ".carbon" / "integrations" / "jank-cloud"
    snapshots: list[tuple[str, dict[str, Any], Path]] = []
    if directory.is_dir():
        for file in directory.glob("*.json"):
            value = load_json(file, {})
            if isinstance(value, dict) and value.get("provider") == "jank-cloud":
                snapshots.append((str(value.get("observedAt", "")), value, file))
    if not snapshots:
        return {
            "status": "not-assessed",
            "basis": "No Jank Cloud comparative benchmark snapshot is available.",
            "affectsReleaseRecommendation": False,
        }
    _, snapshot, file = max(snapshots, key=lambda row: row[0])
    evidence = snapshot.get("evidence", {}) if isinstance(snapshot.get("evidence"), dict) else {}
    benchmark = evidence.get("benchmark") if isinstance(evidence.get("benchmark"), dict) else None
    if not benchmark:
        return {
            "status": "unavailable",
            "observedAt": snapshot.get("observedAt"),
            "snapshotPath": str(file),
            "basis": str(evidence.get("unavailableReason") or "No comparable public category benchmark is available."),
            "affectsReleaseRecommendation": False,
        }
    return {
        "status": "available",
        "observedAt": snapshot.get("observedAt"),
        "snapshotPath": str(file),
        "targetUrl": evidence.get("targetUrl"),
        "category": benchmark.get("category"),
        "sampleSize": benchmark.get("appCount"),
        "averageJank": benchmark.get("avgJank"),
        "averageIssuesPerPage": benchmark.get("avgIssuesPerPage"),
        "issuesByTypeAverage": benchmark.get("issuesByTypeAvg"),
        "personaAverage": benchmark.get("persona"),
        "source": benchmark.get("source"),
        "basis": "Finished public Jank report aggregates only; use for prioritization and explanation, never as local execution evidence.",
        "affectsReleaseRecommendation": False,
    }


def chi_square_2x2(a_pass: int, a_fail: int, b_pass: int, b_fail: int) -> dict[str, Any] | None:
    cells = [a_pass, a_fail, b_pass, b_fail]
    total = sum(cells)
    if total == 0 or min(a_pass + a_fail, b_pass + b_fail, a_pass + b_pass, a_fail + b_fail) == 0:
        return None
    denominator = (a_pass + a_fail) * (b_pass + b_fail) * (a_pass + b_pass) * (a_fail + b_fail)
    chi2 = total * ((a_pass * b_fail - a_fail * b_pass) ** 2) / denominator
    p_value = math.erfc(math.sqrt(chi2 / 2))
    expected = [
        (a_pass + a_fail) * (a_pass + b_pass) / total,
        (a_pass + a_fail) * (a_fail + b_fail) / total,
        (b_pass + b_fail) * (a_pass + b_pass) / total,
        (b_pass + b_fail) * (a_fail + b_fail) / total,
    ]
    return {
        "method": "Pearson chi-square 2x2, df=1", "chiSquare": round(chi2, 4),
        "pValue": round(p_value, 6), "expectedCounts": [round(value, 2) for value in expected],
        "validity": "exploratory" if min(expected) < 5 else "asymptotic-conditions-met",
        "warning": "Use only for independent categorical runs. Prefer paired analysis when the same cases ran on both builds.",
    }


def confidence_analysis(root: Path, persist: bool = True) -> dict[str, Any]:
    paths = ensure_framework(root)
    model = load_json(paths["model"], {})
    coverage = update_coverage(root, model)
    suite = load_json(paths["regression"], {})
    runs = run_documents(root)
    latest = runs[-1] if runs else None
    previous = runs[-2] if len(runs) > 1 else None
    def counts(run: dict[str, Any] | None) -> dict[str, int]:
        result = {key: 0 for key in ("passed", "failed", "blocked", "planned", "skipped")}
        for test in (run or {}).get("tests", []):
            status = test.get("status")
            if status in result:
                result[status] += 1
        return result
    current_counts, prior_counts = counts(latest), counts(previous)
    severe = [finding for finding in (latest or {}).get("findings", []) if finding.get("severity") in {"critical", "high"}]
    failed_high = [test for test in (latest or {}).get("tests", []) if test.get("status") == "failed" and test.get("priority") in {"P0", "P1"}]
    total_entities = len(coverage.get("links", []))
    trusted_entities = sum(1 for row in coverage.get("links", []) if row.get("status") == "regression")
    candidate_entities = sum(1 for row in coverage.get("links", []) if row.get("status") == "candidate")
    coverage_ratio = trusted_entities / total_entities if total_entities else 0
    latest_manifest = {}
    if latest and latest.get("runId"):
        latest_manifest = load_json(root / ".carbon" / "runs" / str(latest["runId"]) / "evidence" / "manifest.json", {})
    quality = assess_quality_confidence({"targetRoot": str(root), **(latest or {})}, latest_manifest, coverage.get("gaps", []))
    comparative_benchmark = latest_jank_benchmark(root)
    if severe or failed_high:
        recommendation = "hold"
        reason = "Severe findings or failed P0/P1 checks remain."
    elif not latest:
        recommendation = "insufficient-evidence"
        reason = "No executed CARBON run is available."
    elif quality["confidenceInQuality"]["band"] == "low":
        recommendation = "hold"
        reason = quality["confidenceInQuality"]["basis"]
    elif quality["confidenceInQuality"]["band"] in {"not-assessed", "guarded"} or current_counts["blocked"] or coverage.get("gaps"):
        recommendation = "canary"
        reason = "No severe failure is recorded, but confidence remains guarded by incomplete execution, evidence, or coverage."
    else:
        recommendation = "ship"
        reason = "Current executed high-risk checks have no recorded severe failure; human release authority remains required."
    repetitions = []
    current_n = current_counts["passed"] + current_counts["failed"]
    if current_n < 5:
        repetitions.append("Run at least five comparable observations for unstable or nondeterministic claims before interpreting a rate.")
    if previous and str(previous.get("revision")) == str((latest or {}).get("revision")):
        repetitions.append("Separate build identifiers before claiming build-to-build improvement.")
    statistical = None
    if latest and previous:
        statistical = chi_square_2x2(prior_counts["passed"], prior_counts["failed"], current_counts["passed"], current_counts["failed"])
        if statistical and statistical["validity"] == "exploratory":
            repetitions.append("Run more comparable cases on both builds; one or more expected chi-square cells are below five.")
    analysis = {
        "schema": "carbon.confidence/v1", "generatedAt": utc_now(),
        "releaseRecommendation": recommendation, "recommendationReason": reason,
        "latestRun": (latest or {}).get("runId"), "previousRun": (previous or {}).get("runId"),
        "latestRevision": (latest or {}).get("revision"), "previousRevision": (previous or {}).get("revision"),
        "currentCounts": current_counts, "previousCounts": prior_counts,
        "coverage": {"entities": total_entities, "trusted": trusted_entities, "candidate": candidate_entities, "gaps": len(coverage.get("gaps", [])), "trustedRatio": round(coverage_ratio, 4)},
        "regressionLifecycle": {
            stage: sum(1 for test in suite.get("tests", []) if test.get("lifecycle") == stage)
            for stage in ("candidate", "observed", "repeatable", "regression", "needs-review")
        },
        "severeFindings": severe, "failedHighPriority": [test.get("id") for test in failed_high],
        "statisticalComparison": statistical, "recommendedAdditionalEvidence": repetitions,
        "qualityConfidence": quality,
        "comparativeBenchmark": comparative_benchmark,
        "remainingRisk": quality["remainingRisk"],
        "coverageMissing": quality["coverageMissing"],
        "whatWouldChangeConfidence": quality["whatWouldChangeConfidence"],
        "uncertainty": list(model.get("warnings", [])) + (["No executed run."] if not latest else []),
        "authority": "CARBON recommends; a human retains mutation and release authority.",
    }
    if persist:
        atomic_json(paths["analysis"], analysis)
        append_jsonl(paths["history"], analysis)
    return analysis


def adaptive_plan(root: Path, run_dir: Path) -> Path:
    paths = ensure_framework(root)
    run = load_json(run_dir / "run.json")
    suite = load_json(paths["regression"], {})
    selected: dict[str, dict[str, Any]] = {}
    failed = [test for test in run.get("tests", []) if test.get("status") == "failed"]
    for test in failed:
        replay = {key: value for key, value in test.items() if key not in {"status", "attempts", "actual", "evidence", "error", "startedAt", "finishedAt"}}
        replay["id"] = f"REPRO-{test['id']}"
        replay["title"] = f"Reproduce: {test['title']}"
        replay["priority"] = "P0" if test.get("priority") in {"P0", "P1"} else "P1"
        replay["reason"] = "Preserve and rerun the original failure before expanding or repairing."
        selected[replay["id"]] = replay
        related = [candidate for candidate in suite.get("tests", []) if candidate.get("lifecycle") == "regression" and (
            set(candidate.get("surfaceIds") or []) & set(test.get("surfaceIds") or [])
            or candidate.get("apiOperation") == test.get("apiOperation")
            or candidate.get("flow") == test.get("flow")
        )]
        for candidate in related[:12]:
            selected[candidate["id"]] = candidate
    if not selected:
        for candidate in suite.get("tests", []):
            if candidate.get("lifecycle") in {"repeatable", "regression"} and candidate.get("priority") in {"P0", "P1"}:
                selected[candidate["id"]] = candidate
            if len(selected) >= 20:
                break
    plan = {
        "title": "CARBON adaptive evidence pass", "mode": "quick", "target": run.get("target", ""),
        "environment": run.get("environment", "local isolated test environment"), "revision": run.get("revision", ""),
        "scope": ["original reproduction first", "failed/changed surface", "smallest adjacent regression set"],
        "exclusions": ["unapproved mutation or destructive expansion"], "assumptions": [],
        "detectedFlows": run.get("detectedFlows", []), "detectedItems": run.get("detectedItems", []),
        "availableAdapters": sorted({
            "carbon-browser/v1" if test.get("adapter") == "isolated-browser" else test.get("adapter", "manual")
            for test in selected.values()
        }),
        "setupDependencies": [], "credentialSources": run.get("credentialSources", []),
        "stopConditions": ["original failure not reproducible", "severe failure", "shared blocker", "evidence no longer changes decision"],
        "residualUncertainty": run.get("residualUncertainty", []), "concurrency": 1,
        "tests": list(selected.values()) or [candidate_test("ADAPT-REVIEW-001", "Explain why no adaptive case is runnable", "P1", "review", "No failed or trusted regression case was available.", ["Inspect the prior run and coverage gaps"], "A concrete next evidence action or explicit blocker is recorded.", "code-inspection", [str(run_dir)])],
    }
    output = paths["base"] / "adaptive-plan.json"
    atomic_json(output, plan)
    return output


def render_report(root: Path) -> Path:
    paths = ensure_framework(root)
    model = load_json(paths["model"], {})
    coverage = load_json(paths["coverage"], {})
    analysis = confidence_analysis(root, persist=False)
    def esc(value: Any) -> str:
        return html.escape(str(value if value is not None else ""))
    icon_paths = {
        "confidence": '<circle cx="12" cy="12" r="8"/><path d="m8.5 12 2.2 2.2 4.8-5"/><path d="M12 1v2M12 21v2"/>',
        "risk": '<path d="M12 3 2.8 20h18.4L12 3Z"/><path d="M12 9v5M12 17.5v.1"/>',
        "coverage": '<circle cx="12" cy="12" r="8"/><path d="M12 4v8l5.7 5.7M12 12h8"/>',
        "release": '<path d="M5 16c-1.5 1-2 3-2 5 2 0 4-.5 5-2M14 4c3-2 5-2 7-2 0 2 0 4-2 7l-7 7-4-4 6-8Z"/><path d="m9 15-2 2M15 8l2 2"/>',
        "build": '<path d="M4 6h16v12H4zM8 3h8v3M8 18v3M16 18v3"/>',
        "stats": '<path d="M4 20V10M10 20V4M16 20v-7M22 20V7M2 20h22"/>',
        "timeline": '<circle cx="12" cy="12" r="9"/><path d="M12 7v5l3 2"/>',
        "evidence": '<path d="M2 12s3.7-6 10-6 10 6 10 6-3.7 6-10 6S2 12 2 12Z"/><circle cx="12" cy="12" r="2.5"/>',
        "product": '<circle cx="5" cy="6" r="2"/><circle cx="19" cy="6" r="2"/><circle cx="12" cy="18" r="2"/><path d="M7 7.2 10.7 16M17 7.2 13.3 16M7 6h10"/>',
        "artifact": '<path d="M6 3h9l4 4v14H6z"/><path d="M15 3v5h4M9 12h6M9 16h6"/>',
    }
    def icon(name: str) -> str:
        paths = icon_paths.get(name, icon_paths["artifact"])
        return f'<svg class="section-icon" viewBox="0 0 24 24" aria-hidden="true" fill="none" stroke="currentColor" stroke-width="1.65" stroke-linecap="round" stroke-linejoin="round">{paths}</svg>'
    def heading(title: str, name: str, level: int = 2) -> str:
        return f'<h{level} class="icon-heading">{icon(name)}<span>{esc(title)}</span></h{level}>'
    entities = list(model.get("surfaces", [])) + list(model.get("apis", [])) + list(model.get("flows", []))
    entity_rows = "".join(
        f"<tr><td>{esc(row.get('kind') or ('API' if row.get('method') else 'flow'))}</td><td>{esc(row.get('name') or row.get('operationId') or row.get('path'))}</td><td>{esc(row.get('source'))}</td></tr>"
        for row in entities[:250]
    ) or "<tr><td colspan='3'>No surfaces discovered yet.</td></tr>"
    lifecycle = analysis["regressionLifecycle"]
    stats = analysis.get("statisticalComparison")
    stats_html = "<p>No comparable two-run categorical dataset yet.</p>" if not stats else (
        f"<p><strong>{esc(stats['method'])}</strong> · χ² {esc(stats['chiSquare'])} · p {esc(stats['pValue'])} · {esc(stats['validity'])}</p><p>{esc(stats['warning'])}</p>"
    )
    gaps = "".join(f"<li>{esc(row.get('name'))} · {esc(row.get('reason'))}</li>" for row in coverage.get("gaps", [])[:40]) or "<li>No unmapped entities in the current model.</li>"
    quality = analysis["qualityConfidence"]
    benchmark = analysis.get("comparativeBenchmark", {})
    benchmark_html = "<p>No Jank Cloud comparative snapshot has been read for this project. Comparative context remains NOT ASSESSED.</p>"
    if benchmark.get("status") == "available":
        benchmark_html = (
            f"<p><strong>{esc(benchmark.get('category') or 'Comparable category')}</strong> · "
            f"{esc(benchmark.get('sampleSize'))} public apps · average Jank {esc(benchmark.get('averageJank'))} · "
            f"average issues/page {esc(benchmark.get('averageIssuesPerPage'))}</p>"
            f"<p>{esc(benchmark.get('basis'))}</p><p><strong>Observed:</strong> {esc(benchmark.get('observedAt'))}</p>"
        )
    elif benchmark.get("status") == "unavailable":
        benchmark_html = f"<p>Jank Cloud comparison unavailable: {esc(benchmark.get('basis'))}</p>"
    quality_missing = "".join(f"<li>{esc(item)}</li>" for item in quality["coverageMissing"]) or "<li>No material missing coverage is recorded for the selected scope.</li>"
    remaining_risk = "".join(f"<li>{esc(item)}</li>" for item in quality["remainingRisk"]["reasons"]) or "<li>No material remaining risk recorded in the selected scope.</li>"
    confidence_changers = "".join(f"<li>{esc(item)}</li>" for item in quality["whatWouldChangeConfidence"]) or "<li>Broaden only if the decision requires evidence beyond the selected scope.</li>"
    run_candidates: list[tuple[str, Path, dict[str, Any], dict[str, Any]]] = []
    for run_json in (root / ".carbon" / "runs").glob("*/run.json"):
        run_state = load_json(run_json, {})
        evidence = load_json(run_json.parent / "evidence" / "manifest.json", {})
        run_candidates.append((str(run_state.get("updatedAt", "")), run_json.parent, run_state, evidence))
    run_candidates.sort(key=lambda row: row[0], reverse=True)
    latest_run = run_candidates[0] if run_candidates else None
    evidence_html = "<p>No executed run evidence has been registered yet.</p>"
    evidence_metrics_html = ""
    perspective_html = "<p>No executed persona feedback or exploratory observations have been registered yet.</p>"
    if latest_run:
        _, latest_run_dir, latest_run_state, latest_evidence = latest_run
        evidence_summary = latest_evidence.get("summary", {}) if isinstance(latest_evidence, dict) else {}
        screenshots: list[tuple[dict[str, Any], dict[str, Any]]] = []
        for page in latest_evidence.get("pages", []) if isinstance(latest_evidence, dict) else []:
            for artifact in page.get("artifacts", []):
                candidate = Path(str(artifact.get("path", "")))
                if artifact.get("kind") == "screenshot" and not candidate.is_absolute() and ".." not in candidate.parts:
                    screenshots.append((page, artifact))
        gallery = "".join(
            f'<a class="shot" href="../runs/{esc(latest_run_dir.name)}/{esc(artifact.get("path"))}"><img src="../runs/{esc(latest_run_dir.name)}/{esc(artifact.get("path"))}" alt="Screenshot for {esc(page.get("title") or page.get("url") or page.get("id"))}"><span>{esc(page.get("id"))} · {esc(page.get("title") or page.get("url"))}</span></a>'
            for page, artifact in screenshots[:8]
        ) or "<p>No safe screenshots were registered for the latest run.</p>"
        issue_rows = "".join(
            f"<tr><td>{esc(item.get('severity'))}</td><td>{esc(item.get('pageId'))}</td><td>{esc(item.get('category'))}</td><td>{esc(item.get('title'))}</td><td>{esc(item.get('detail'))}</td></tr>"
            for item in latest_evidence.get("issues", [])[:80]
        ) or "<tr><td colspan='5'>No general artifact issue signals were found in the captured evidence.</td></tr>"
        evidence_metrics_html = "".join(
            f'<div class="metric"><div class="metric-top"><span>{esc(label)}</span>{icon("evidence" if key in {"pages", "screenshots"} else "artifact")}</div><strong>{int(evidence_summary.get(key, 0) or 0)}</strong></div>'
            for key, label in (("pages", "Captured pages"), ("artifacts", "Artifacts"), ("screenshots", "Screenshots"), ("issues", "Artifact signals"), ("personaFeedback", "Persona feedback"), ("explorationObservations", "Explore observations"))
        )
        strategy = latest_evidence.get("strategy", {}) if isinstance(latest_evidence, dict) else {}
        feedback_rows = "".join(
            f'<article class="perspective"><span>{esc(item.get("role") or "product user")}</span><strong>{esc(item.get("name") or item.get("personaId"))}</strong><p>{esc("; ".join(item.get("frictions") or item.get("positives") or ["Feedback recorded"]))}</p><small>{esc(item.get("pageId"))} · {esc(item.get("confidence") or "opinion")}</small></article>'
            for item in latest_evidence.get("personaFeedback", [])[:12]
            if isinstance(item, dict)
        ) or '<p>No persona feedback captured in the latest run.</p>'
        explore_rows = "".join(
            f'<article class="perspective explore"><span>{esc(item.get("strategy") or "explore")}</span><strong>{esc(item.get("title") or item.get("charterId"))}</strong><p>{esc(item.get("observation"))}</p><small>{esc(item.get("pageId"))} · {esc(item.get("status") or "observed")}</small></article>'
            for item in latest_evidence.get("explorationObservations", [])[:12]
            if isinstance(item, dict)
        ) or '<p>No exploratory observations captured in the latest run.</p>'
        perspective_html = f'<p><strong>Strategy:</strong> {esc(strategy.get("selected") or latest_run_state.get("testStrategy", {}).get("selected") or "not recorded")} · {esc(strategy.get("rationale") or latest_run_state.get("testStrategy", {}).get("rationale") or "")}</p><div class="perspectives">{feedback_rows}{explore_rows}</div>'
        evidence_html = f"""
        <p><strong>Latest executed run:</strong> <a href="../runs/{esc(latest_run_dir.name)}/report.html">{esc(latest_run_state.get('runId') or latest_run_dir.name)} visual run report</a></p>
        <div class="shots">{gallery}</div>
        <div class="table"><table class="evidence-table"><thead><tr><th>Severity</th><th>Page</th><th>Channel</th><th>Signal</th><th>Evidence</th></tr></thead><tbody>{issue_rows}</tbody></table></div>
        """
    document = f"""<!doctype html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1,viewport-fit=cover"><link rel="icon" href="{REPORT_FAVICON}">
<title>CARBON confidence framework · {esc(root.name)}</title>
<style>
:root{{--bg:#000000;--panel:#070707;--soft:#0a0a0a;--line:#242424;--text:#ffffff;--muted:#9b9b9b;--gold:#ffffff;--good:#72e6b1;--bad:#ff7d7d;--risk:#ff9f43;--risk-soft:rgba(255,159,67,.11);--risk-line:rgba(255,159,67,.48)}}
*{{box-sizing:border-box}}body{{margin:0;background:#000000;color:var(--text);font:15px/1.55 Inter,system-ui,sans-serif}}main{{width:min(1180px,calc(100% - 28px));margin:auto;padding:24px 0 72px}}
.hero,.card{{min-width:0;border:1px solid var(--line);border-radius:4px;background:var(--panel);padding:22px;margin:12px 0}}.hero{{padding:clamp(28px,6vw,58px);background:linear-gradient(135deg,rgba(255,255,255,.06),transparent 52%),var(--panel)}}
.framework-brand{{display:flex;align-items:center;gap:13px;margin-bottom:30px}}.framework-brand svg{{width:58px;height:58px}}.framework-brand-copy{{display:grid;gap:5px}}.framework-brand-copy strong{{font-size:19px;letter-spacing:.24em}}.framework-brand-copy span{{color:var(--muted);font-size:8px;font-weight:800;letter-spacing:.24em}}
h1{{font-size:clamp(38px,7vw,72px);line-height:.98;letter-spacing:-.055em;margin:0}}h2{{margin:0 0 12px}}p{{color:var(--muted)}}.badge{{display:inline-block;border:1px solid rgba(255,255,255,.35);border-radius:2px;padding:5px 10px;color:#ffffff;font-weight:750}}.icon-heading,.metric-top{{display:flex;align-items:center;gap:10px}}.section-icon{{width:22px;height:22px;flex:0 0 auto}}.risk-panel .section-icon,.risk-panel h2,.risk-panel strong{{color:var(--risk)}}
.metrics{{display:grid;grid-template-columns:repeat(auto-fit,minmax(145px,1fr));gap:10px}}.metric{{border:1px solid var(--line);border-radius:5px;padding:16px;background:var(--soft)}}.metric-top{{justify-content:space-between}}.metric-top .section-icon{{width:18px;height:18px;color:#777}}.metric span,.metric strong{{display:block}}.metric span{{color:var(--muted);font-size:11px;text-transform:uppercase}}.metric strong{{font-size:30px}}
.grid{{display:grid;grid-template-columns:repeat(auto-fit,minmax(min(100%,290px),1fr));gap:12px}}.grid>*{{min-width:0}}table{{width:100%;border-collapse:collapse}}th,td{{padding:10px;border-bottom:1px solid var(--line);text-align:left;vertical-align:top;overflow-wrap:anywhere}}th{{color:#ffffff}}.table{{width:100%;max-width:100%;overflow-x:auto}}code{{color:#ffffff;overflow-wrap:anywhere}}ul{{padding-left:20px}}a{{color:#ffffff;text-underline-offset:3px}}.release{{border-color:{'var(--bad)' if analysis['releaseRecommendation']=='hold' else 'var(--gold)'}}}
.decision-spread{{display:grid;grid-template-columns:minmax(0,1.25fr) minmax(290px,.75fr);gap:12px;margin:12px 0}}.decision-main,.risk-panel{{margin:0;min-height:260px}}.decision-main{{border-color:rgba(255,255,255,.3);background:linear-gradient(145deg,rgba(255,255,255,.06),transparent),var(--panel)}}.risk-panel{{border-color:var(--risk-line);background:linear-gradient(145deg,var(--risk-soft),transparent),var(--panel)}}.risk-value{{font-size:clamp(34px,5vw,54px);line-height:1;text-transform:capitalize}}.decision-ledger{{display:grid;grid-template-columns:repeat(3,1fr);border-top:1px solid var(--line);border-bottom:1px solid var(--line);margin-bottom:12px}}.decision-ledger article{{padding:20px;border-right:1px solid var(--line)}}.decision-ledger article:last-child{{border-right:0}}.build-ledger{{display:grid;grid-template-columns:1fr 1fr;border-top:1px solid var(--line);border-bottom:1px solid var(--line)}}.build-ledger .card{{margin:0;border:0;border-right:1px solid var(--line)}}.build-ledger .card:last-child{{border-right:0}}
.shots{{display:grid;grid-template-columns:repeat(auto-fit,minmax(min(100%,280px),1fr));gap:12px;margin:16px 0}}.shot{{display:block;overflow:hidden;border:1px solid var(--line);border-radius:15px;background:#050609;color:var(--muted);text-decoration:none}}.shot img{{display:block;width:100%;max-height:440px;object-fit:contain}}.shot span{{display:block;border-top:1px solid var(--line);padding:8px 10px}}
.perspectives{{display:grid;grid-template-columns:repeat(auto-fit,minmax(min(100%,240px),1fr));gap:10px}}.perspective{{display:grid;gap:5px;border:1px solid var(--line);border-left:3px solid var(--gold);border-radius:4px;padding:14px;background:var(--soft)}}.perspective.explore{{border-left-color:#767676}}.perspective span{{color:#ffffff;font-size:10px;font-weight:800;text-transform:uppercase}}.perspective.explore span{{color:#bdbdbd}}.perspective p{{margin:0}}.perspective small{{color:var(--muted)}}
.footer{{display:flex;justify-content:space-between;gap:20px;margin-top:42px;border-top:1px solid var(--line);padding-top:18px;color:var(--muted);font-size:12px}}.footer a{{color:#fff}}
@media(max-width:640px){{main{{width:min(100% - 18px,1180px);padding-top:9px}}.hero,.card{{padding:18px}}h1{{font-size:42px}}.framework-brand{{margin-bottom:22px}}.decision-spread,.decision-ledger,.build-ledger{{grid-template-columns:1fr}}.decision-ledger article,.build-ledger .card{{border-right:0;border-bottom:1px solid var(--line)}}.decision-ledger article:last-child,.build-ledger .card:last-child{{border-bottom:0}}.evidence-table thead{{display:none}}.evidence-table,.evidence-table tbody,.evidence-table tr,.evidence-table td{{display:block;width:100%}}.evidence-table tr{{border:1px solid var(--line);border-radius:4px;background:var(--soft);padding:10px;margin:0 0 10px}}.evidence-table td{{display:grid;grid-template-columns:72px minmax(0,1fr);gap:8px;border:0;padding:4px 0}}.evidence-table td::before{{color:#ffffff;font-size:10px;font-weight:800;text-transform:uppercase}}.evidence-table td:nth-child(1)::before{{content:'Severity'}}.evidence-table td:nth-child(2)::before{{content:'Page'}}.evidence-table td:nth-child(3)::before{{content:'Channel'}}.evidence-table td:nth-child(4)::before{{content:'Signal'}}.evidence-table td:nth-child(5)::before{{content:'Evidence'}}}}
</style></head><body><main>
<section class="hero"><div class="framework-brand"><svg viewBox="0 0 256 256" aria-hidden="true"><rect width="256" height="256" fill="#000000"/><path fill="#ffffff" d="M56 34h112l48 48-48 48V96l-16-16H94c-15 0-22 7-22 22v52c0 15 7 22 22 22h104v46H56c-21 0-32-11-32-32V66c0-21 11-32 32-32Z"/></svg><span class="framework-brand-copy"><strong>CARBON</strong><span>ENGINEERING CONFIDENCE</span></span></div><p class="badge">Automatic framework</p><h1>{esc(root.name)} confidence</h1><p>Product model, change impact, API and UI coverage, regression lifecycle, build comparison, and release evidence.</p></section>
<section class="metrics"><div class="metric"><div class="metric-top"><span>Recommendation</span>{icon('release')}</div><strong>{esc(analysis['releaseRecommendation'])}</strong></div><div class="metric"><div class="metric-top"><span>Quality confidence</span>{icon('confidence')}</div><strong>{esc(quality['confidenceInQuality']['band'])}</strong></div><div class="metric"><div class="metric-top"><span>Remaining risk</span>{icon('risk')}</div><strong>{esc(quality['remainingRisk']['level'])}</strong></div><div class="metric"><div class="metric-top"><span>Conclusive coverage</span>{icon('coverage')}</div><strong>{quality['coverage']['conclusivePercent']}%</strong></div><div class="metric"><div class="metric-top"><span>Coverage gaps</span>{icon('risk')}</div><strong>{len(quality['coverageMissing'])}</strong></div><div class="metric"><div class="metric-top"><span>Regression tests</span>{icon('timeline')}</div><strong>{lifecycle['regression']}</strong></div></section>
<div class="decision-spread"><section class="card decision-main"><p class="badge">Evidence-qualified quality confidence</p>{heading('Confidence in quality', 'confidence')}<div class="risk-value">{esc(quality['confidenceInQuality']['band'])}</div><p>{esc(quality['confidenceInQuality']['basis'])}</p><p>{esc(quality['interpretation'])}</p></section><section class="card risk-panel">{heading('Remaining risk', 'risk')}<div class="risk-value">{esc(quality['remainingRisk']['level'])}</div><ul>{remaining_risk}</ul></section></div>
<div class="decision-ledger"><article>{heading('Coverage missing','coverage',3)}<ul>{quality_missing}</ul></article><article>{heading('What changes confidence','release',3)}<ul>{confidence_changers}</ul></article><article>{heading('Release evidence','release',3)}<p>{esc(analysis['recommendationReason'])}</p><p><strong>Authority:</strong> {esc(analysis['authority'])}</p></article></div>
<div class="build-ledger"><section class="card">{heading('Current build','build')}<p><strong>Revision:</strong> {esc(analysis.get('latestRevision') or 'not recorded')}</p><p><strong>Run:</strong> {esc(analysis.get('latestRun') or 'none')}</p><p><strong>Results:</strong> {esc(analysis['currentCounts'])}</p></section><section class="card">{heading('Previous build','build')}<p><strong>Revision:</strong> {esc(analysis.get('previousRevision') or 'not recorded')}</p><p><strong>Run:</strong> {esc(analysis.get('previousRun') or 'none')}</p><p><strong>Results:</strong> {esc(analysis['previousCounts'])}</p></section></div>
<section class="card">{heading('Statistical comparison','stats')}{stats_html}<ul>{''.join(f'<li>{esc(item)}</li>' for item in analysis['recommendedAdditionalEvidence']) or '<li>No additional repetition recommendation.</li>'}</ul></section>
<section class="card">{heading('Comparative benchmark context','stats')}{benchmark_html}<p>Peer aggregates can focus investigation, but they do not change CARBON’s release recommendation and do not prove this target was tested.</p></section>
<section class="card">{heading('Regression lifecycle','timeline')}<div class="metrics">{''.join(f'<div class="metric"><div class="metric-top"><span>{esc(stage)}</span>{icon("timeline")}</div><strong>{count}</strong></div>' for stage,count in lifecycle.items())}</div><p>Candidate → observed → repeatable → explicitly promoted regression. Failed or dismissed evidence cannot silently become trusted automation.</p></section>
<section class="card">{heading('Executed page evidence','evidence')}<p>Safe screenshots are shown with cross-artifact signals from DOM/accessibility, console, network, performance, data, traces, and recordings. Open the run report for the complete page matrix, provenance, visual reviews, and evidence gaps.</p><div class="metrics">{evidence_metrics_html}</div>{evidence_html}</section>
<section class="card">{heading('Persona feedback and exploratory coverage','product')}<p>Generated user perspectives stay separate from deterministic test results. Breadth maps the reachable surface; depth follows the riskiest journey through boundaries, interruption, recovery, and state.</p>{perspective_html}</section>
<section class="card">{heading('Modeled coverage gaps','risk')}<ul>{gaps}</ul></section>
<section class="card">{heading('Modeled product','product')}<div class="table"><table><thead><tr><th>Kind</th><th>Surface / operation</th><th>Source</th></tr></thead><tbody>{entity_rows}</tbody></table></div></section>
<section class="card">{heading('Artifacts','artifact')}<p><code>{esc(paths['base'])}</code></p><p>Product model, coverage map, regression suite, outcomes, orchestration plans, adaptive plans, confidence history, and this report remain inspectable JSON/JSONL/HTML.</p></section>
<footer class="footer"><span>CARBON is a <a href="https://testers.ai">testers.ai</a> product.</span><span>Evidence remains inspectable in the project-local CARBON state.</span></footer>
</main></body></html>"""
    paths["report"].write_text(document, encoding="utf-8")
    return paths["report"]


def install_ci(root: Path, apply: bool) -> dict[str, str]:
    script_target = root / "tools" / "carbon" / "carbon_framework.py"
    workflow_target = root / ".github" / "workflows" / "carbon.yml"
    workflow = """name: CARBON confidence\n\non:\n  pull_request:\n  workflow_dispatch:\n  schedule:\n    - cron: '17 3 * * *'\n\npermissions:\n  contents: read\n\njobs:\n  carbon:\n    runs-on: ubuntu-latest\n    steps:\n      - uses: actions/checkout@v4\n        with:\n          fetch-depth: 0\n      - uses: actions/setup-python@v5\n        with:\n          python-version: '3.12'\n      - name: Build change-focused CARBON plan\n        run: python3 tools/carbon/carbon_framework.py prepare --root . --request \"CI validation for $GITHUB_EVENT_NAME\" --size ${{ github.event_name == 'schedule' && 'medium' || 'quick' }}\n      - name: Optional project test command\n        if: ${{ vars.CARBON_TEST_COMMAND != '' }}\n        run: ${{ vars.CARBON_TEST_COMMAND }}\n      - name: Analyze confidence\n        run: python3 tools/carbon/carbon_framework.py analyze --root .\n      - name: Render report\n        run: python3 tools/carbon/carbon_framework.py report --root .\n      - uses: actions/upload-artifact@v4\n        if: always()\n        with:\n          name: carbon-confidence\n          path: .carbon/framework/\n"""
    workflow = workflow.replace(
        "          path: .carbon/framework/",
        "          path: |\n            .carbon/framework/\n            .carbon/runs/",
    )
    if apply:
        script_target.parent.mkdir(parents=True, exist_ok=True)
        workflow_target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(Path(__file__).resolve(), script_target)
        workflow_target.write_text(workflow, encoding="utf-8")
    return {"script": str(script_target), "workflow": str(workflow_target), "applied": str(apply).lower(), "workflowPreview": workflow if not apply else ""}


def command_init(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    print(json.dumps({key: str(value) for key, value in ensure_framework(root).items()}, indent=2))


def command_discover(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    model = discover(root, args.openapi, args.har)
    print(json.dumps({"model": str(framework_paths(root)["model"]), "surfaces": len(model["surfaces"]), "apis": len(model["apis"]), "flows": len(model["flows"]), "changedFiles": len(model["changedFiles"]), "warnings": model["warnings"]}, indent=2))


def command_prepare(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    model = discover(root, args.openapi, args.har)
    print(json.dumps(orchestration_plan(root, args.request, args.target or "auto-detect local target", args.size, model, args.strategy, args.coverage_profile), indent=2))


def command_more(args: argparse.Namespace) -> None:
    print(json.dumps(more_coverage_plan(Path(args.root).expanduser().resolve(), args.limit, args.budget_minutes), indent=2))


def command_api_generate(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    document, warning = parse_document(Path(args.spec).expanduser().resolve())
    if warning or not document:
        raise ValueError(warning or "OpenAPI document is empty")
    operations = openapi_operations(Path(args.spec).expanduser().resolve(), document)
    tests = api_candidates(operations, str(Path(args.spec).expanduser().resolve()))
    suite = merge_candidates(root, tests)
    print(json.dumps({"operations": len(operations), "generatedCandidates": len(tests), "suite": str(framework_paths(root)["regression"]), "totalSuiteTests": len(suite["tests"])}, indent=2))


def command_observe(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    print(json.dumps(observe_candidate(root, args.test_id, Path(args.run_dir).expanduser().resolve(), args.run_test_id or args.test_id), indent=2))


def command_promote(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    print(json.dumps(promote_candidate(root, args.test_id, args.confirm), indent=2))


def command_outcome(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    print(json.dumps(record_outcome(root, args.test_id, args.disposition, args.note, args.run_id), indent=2))


def command_adapt(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    print(adaptive_plan(root, Path(args.run_dir).expanduser().resolve()))


def command_analyze(args: argparse.Namespace) -> None:
    root = Path(args.root).expanduser().resolve()
    print(json.dumps(confidence_analysis(root, persist=True), indent=2))


def command_report(args: argparse.Namespace) -> None:
    print(render_report(Path(args.root).expanduser().resolve()))


def command_ci(args: argparse.Namespace) -> None:
    print(json.dumps(install_ci(Path(args.root).expanduser().resolve(), args.apply), indent=2))


def parser() -> argparse.ArgumentParser:
    root = argparse.ArgumentParser(description=__doc__)
    sub = root.add_subparsers(dest="command", required=True)
    init = sub.add_parser("init")
    init.add_argument("--root", required=True)
    init.set_defaults(func=command_init)

    discover_parser = sub.add_parser("discover")
    discover_parser.add_argument("--root", required=True)
    discover_parser.add_argument("--openapi", action="append", default=[])
    discover_parser.add_argument("--har", action="append", default=[])
    discover_parser.set_defaults(func=command_discover)

    prepare = sub.add_parser("prepare")
    prepare.add_argument("--root", required=True)
    prepare.add_argument("--request", default="test this")
    prepare.add_argument("--target", default="")
    prepare.add_argument("--size", choices=("quick", "medium", "large"), default="quick")
    prepare.add_argument("--strategy", choices=("auto", "breadth", "depth", "balanced"), default="auto")
    prepare.add_argument("--coverage-profile", choices=("auto", "broad", "focused"), default="auto")
    prepare.add_argument("--openapi", action="append", default=[])
    prepare.add_argument("--har", action="append", default=[])
    prepare.set_defaults(func=command_prepare)

    more = sub.add_parser("more")
    more.add_argument("--root", required=True)
    more.add_argument("--limit", type=int, choices=range(1, 101), default=40)
    more.add_argument("--budget-minutes", type=int, choices=range(5, 1441), default=60)
    more.set_defaults(func=command_more)

    api = sub.add_parser("api-generate")
    api.add_argument("--root", required=True)
    api.add_argument("--spec", required=True)
    api.set_defaults(func=command_api_generate)

    observe = sub.add_parser("observe")
    observe.add_argument("--root", required=True)
    observe.add_argument("--test-id", required=True)
    observe.add_argument("--run-dir", required=True)
    observe.add_argument("--run-test-id")
    observe.set_defaults(func=command_observe)

    promote = sub.add_parser("promote")
    promote.add_argument("--root", required=True)
    promote.add_argument("--test-id", required=True)
    promote.add_argument("--confirm", action="store_true")
    promote.set_defaults(func=command_promote)

    outcome = sub.add_parser("outcome")
    outcome.add_argument("--root", required=True)
    outcome.add_argument("--test-id", required=True)
    outcome.add_argument(
        "--disposition",
        required=True,
        choices=("confirmed", "dismissed", "false-positive", "escaped", "fixed", "flaky"),
    )
    outcome.add_argument("--note", required=True)
    outcome.add_argument("--run-id", default="")
    outcome.set_defaults(func=command_outcome)

    adapt = sub.add_parser("adapt")
    adapt.add_argument("--root", required=True)
    adapt.add_argument("--run-dir", required=True)
    adapt.set_defaults(func=command_adapt)

    analyze = sub.add_parser("analyze")
    analyze.add_argument("--root", required=True)
    analyze.set_defaults(func=command_analyze)

    report = sub.add_parser("report")
    report.add_argument("--root", required=True)
    report.set_defaults(func=command_report)

    ci = sub.add_parser("ci")
    ci.add_argument("--root", required=True)
    ci.add_argument("--apply", action="store_true")
    ci.set_defaults(func=command_ci)
    return root


def main() -> int:
    try:
        args = parser().parse_args()
        args.func(args)
        return 0
    except (OSError, ValueError, subprocess.SubprocessError) as error:
        print(f"carbon_framework: {error}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())
