#!/usr/bin/env python3
"""Production failure → sanitised regression fixture (#157, plan 2026-09-07-157).

One command, zero ceremony: a payload file (raw error text, JSON blob, or a
subsystem_health record excerpt) plus ``--class`` and ``--source`` become a
redacted fixture under ``tests/fixtures/production/<class>/<id>/payload.json``
and a manifest entry in ``tests/fixtures/production/manifest.json``. The
manifest carries the payload's sha256 so a fixture edited without re-running
capture fails the generated regression tests (tamper-evidence).

Redaction reuses the credential patterns discharged for the failure auto-filer
(#122, commit 9105ca8 — query-string credential params, URL userinfo/DSNs,
Authorization/bearer headers, assignment forms) so that a fixture sees exactly
what the filer would have persisted. The patterns are copied here because the
filer module ships on the failure-auto-filing branch; when that branch merges,
import ``rune.failure_auto_filer.redact`` directly and delete the copy.
``tests/test_capture_failure.py::test_redact_matches_filer_on_shared_cases``
pins parity on the filer's own test literals so the copies cannot drift.

Sanitisation is conservative (plan design decision 1): redact first, then run
an INDEPENDENT residual scan for credential-shaped spans. If anything
credential-shaped survives, capture aborts loudly (ResidualSecretError /
exit code 3) and persists nothing — a leaked fixture is an incident.

Usage:
    python3 scripts/capture_failure.py payload.txt --class ci-crash \
        --source ci [--id my-slug] [--fixtures-root tests/fixtures/production]
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
from datetime import date, datetime, timezone
from pathlib import Path

EXIT_OK = 0
EXIT_USAGE = 2
EXIT_RESIDUAL_SECRET = 3

DEFAULT_FIXTURES_ROOT = "tests/fixtures/production"
VALID_SOURCES = ("issue", "ci", "manual")
SAFE_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$", re.IGNORECASE)

# ---------------------------------------------------------------------------
# Redaction — copied verbatim from src/rune/failure_auto_filer.py (9105ca8, #122)
# ---------------------------------------------------------------------------
# Credentials must never persist in fixtures. Patterns (order matters
# little; all are applied): query-string credential params, URL userinfo
# (scheme://user:pass@host — covers postgres/redis/mysql DSNs and basic
# auth), Authorization headers / bearer tokens, and assignment forms
# (password=..., secret=...). Only the credential SPAN is replaced — the
# surrounding error text (class, host, port, message shape) survives.
_QUERY_CREDS = re.compile(
    r"([?&](?:token|password|passwd|secret|api_key|apikey|access_token|key|sig|signature|credential)=)[^&\s'\"]+",
    re.IGNORECASE,
)
_URL_USERINFO = re.compile(r"\b([a-zA-Z][a-zA-Z0-9+.-]*://)([^\s:/@'\"]+)?:([^\s/@'\"]+)@")
_BEARER = re.compile(r"(?i)\b((?:authorization\s*:?\s*(?:bearer\s+)?|bearer\s+))[^\s,'\";]+")
_ASSIGNMENT = re.compile(
    r"(?i)\b((?:password|passwd|secret|api_key|apikey|access_token|auth)\s*[=:]\s*)['\"]?([^\s'\"]{4,})"
)

# Scan passes get progressively less specific; report the FIRST matching pass
# per hit so the abort message names the tightest classification.
REDACTION_PASSES = (
    ("query-creds", _QUERY_CREDS),
    ("dsn-userinfo", _URL_USERINFO),
    ("bearer", _BEARER),
    ("assignment", _ASSIGNMENT),
)


def redact(text: str) -> str:
    """Strip credentials from an error string (no-op on clean text)."""
    out = _QUERY_CREDS.sub(r"\1<redacted>", text)
    out = _URL_USERINFO.sub(r"\1<redacted>:<redacted>@", out)
    out = _BEARER.sub(r"\1<redacted>", out)
    out = _ASSIGNMENT.sub(lambda m: f"{m.group(1)}<redacted>", out)
    return out


# ---------------------------------------------------------------------------
# Residual-secret scan — deliberately independent of the redaction spans
# ---------------------------------------------------------------------------
# The scan must catch what redaction MISSES (e.g. a quote inside the value
# defeats the assignment span). It therefore re-derives credential shapes from
# scratch and only forgives values that are already the redaction marker or
# are too short to be a credential (1-char leftovers of truncation).

_SCAN_ASSIGN = re.compile(
    r"(?i)\b((?:password|passwd|secret|api_key|apikey|access_token|auth)\s*[=:]\s*)['\"]?([^\s'\"]{2,})"
)
_SCAN_QUERY = re.compile(
    r"(?i)((?:[?&])(?:token|password|passwd|secret|api_key|apikey|access_token|key|sig|signature|credential)=)"
    r"([^\s&'\"]+)"
)
_SCAN_BEARER = re.compile(r"(?i)\b((?:authorization\s*:?\s*(?:bearer\s+)?|bearer\s+))([^\s,'\"]+)")
_SCAN_DSN = re.compile(r"\b([a-zA-Z][a-zA-Z0-9+.-]*://)([^\s/@'\"]*)@")
_REDACTED = re.compile(r"redacted", re.IGNORECASE)


def residual_secrets(text: str) -> list[str]:
    """Credential-shaped spans that survived redaction (empty list = clean)."""
    hits: list[str] = []
    for pattern in (_SCAN_ASSIGN, _SCAN_QUERY, _SCAN_BEARER, _SCAN_DSN):
        for match in pattern.finditer(text):
            groups = match.groups()
            value = groups[-1] if groups else match.group(0)
            if _REDACTED.search(value) or _REDACTED.search(match.group(0)):
                continue
            # truncation slivers (a single stray char) are noise, not secrets
            if len(value) < 2:
                continue
            hits.append(match.group(0))
    return hits


# ---------------------------------------------------------------------------
# Capture
# ---------------------------------------------------------------------------


def _assert_safe_name(name: str, what: str) -> None:
    if not name or not SAFE_NAME_RE.match(name) or ".." in name:
        raise ValueError(f"unsafe {what}: {name!r} (must be a flat slug, no path separators)")


def _original_shape(raw: str) -> str:
    stripped = raw.strip()
    if stripped.startswith(("{", "[")):
        try:
            json.loads(stripped)
            return "json"
        except json.JSONDecodeError:
            return "text"
    return "text"


def capture(
    payload_path: str | Path,
    failure_class: str,
    source: str,
    fixture_id: str | None = None,
    fixtures_root: str | Path = DEFAULT_FIXTURES_ROOT,
    now: datetime | None = None,
) -> dict:
    """Redact a payload, store the fixture, upsert the manifest entry.

    Returns the manifest entry dict. Raises ResidualSecretError (persisting
    nothing) when credential-shaped text survives redaction; ValueError for
    bad failure_class/source/fixture_id.
    """
    if source not in VALID_SOURCES:
        raise ValueError(f"unknown --source {source!r} (expected one of {', '.join(VALID_SOURCES)})")
    _assert_safe_name(failure_class, "failure class")

    payload_file = Path(payload_path)
    raw = payload_file.read_text(encoding="utf-8")

    shape = _original_shape(raw)
    redacted = redact(raw)
    residuals = residual_secrets(redacted)
    if residuals:
        raise ResidualSecretError(
            "residual credential-shaped text survived redaction — fixture NOT written. "
            f"Redact manually (or broaden the redaction patterns) and retry: {residuals[:5]}"
        )

    kinds = [kind for kind, pattern in REDACTION_PASSES if pattern.search(raw)]
    redactions_applied = max(redacted.count("<redacted>") - raw.count("<redacted>"), 0)

    slug = fixture_id or _slugify(f"{failure_class}-{_headline(redacted)}")
    _assert_safe_name(slug, "fixture id")

    class_dir = Path(fixtures_root) / failure_class
    fixture_dir = class_dir / slug
    fixture_dir.mkdir(parents=True, exist_ok=True)
    fixture_file = fixture_dir / "payload.json"

    if fixture_file.exists():
        stored = json.loads(fixture_file.read_text(encoding="utf-8"))
    else:
        stored = {}
    stored.update(
        {
            "id": slug,
            "failure_class": failure_class,
            "source": source,
            "date": (now or datetime.now(timezone.utc)).date().isoformat(),
            "original_shape": shape,
            "redactions_applied": redactions_applied,
            "redaction_kinds": kinds,
            "payload": redacted,
        }
    )
    fixture_file.write_text(json.dumps(stored, indent=2, sort_keys=True) + "\n", encoding="utf-8")

    sha256 = hashlib.sha256(redacted.encode("utf-8")).hexdigest()
    # Does the class's checker FLAG this payload, or must it stay quiet?
    # Defect-class payloads (the recorded payload exhibits the bug) are
    # "flags": the generated regression test asserts the checker still
    # catches them, so a recurrence fails CI. Everything else is "passes".
    try:
        check_payload(failure_class, redacted)
        expectation = "passes"
    except AssertionError:
        expectation = "flags"
    entry = {
        "id": slug,
        "class": failure_class,
        "source": source,
        "date": stored["date"],
        "original_shape": shape,
        "redactions_applied": redactions_applied,
        "redaction_kinds": kinds,
        "checker_expectation": expectation,
        "sha256": sha256,
    }
    _upsert_manifest(Path(fixtures_root), entry)
    return entry


class ResidualSecretError(RuntimeError):
    """A credential-shaped span survived redaction — capture aborted."""


def _headline(text: str) -> str:
    """First line, trimmed, for slug derivation."""
    line = text.strip().splitlines()[0] if text.strip() else "failure"
    return line[:60]


def _slugify(text: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")
    slug = re.sub(r"-{2,}", "-", slug)
    if not slug or not SAFE_NAME_RE.match(slug):
        slug = f"fixture-{hashlib.sha256(text.encode()).hexdigest()[:8]}"
    return slug[:80]


def _upsert_manifest(fixtures_root: Path, entry: dict) -> None:
    manifest_file = fixtures_root / "manifest.json"
    manifest = {"version": 1, "fixtures": []}
    if manifest_file.exists():
        manifest = json.loads(manifest_file.read_text(encoding="utf-8"))
    manifest.setdefault("version", 1)
    fixtures = [f for f in manifest.get("fixtures", []) if f.get("id") != entry["id"]]
    fixtures.append(entry)
    manifest["fixtures"] = sorted(fixtures, key=lambda f: (f.get("class", ""), f["id"]))
    manifest_file.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    _ensure_generated_test(fixtures_root)


# ---------------------------------------------------------------------------
# Per-class payload checkers — the generated regression tests dispatch here
# ---------------------------------------------------------------------------


def check_payload(failure_class: str, payload: str) -> None:
    """Assert the system's CURRENT handling CATCHES the recorded payload.

    One checker per failure class; a payload that its class's checker fails
    to flag means the class's regression coverage has rotted. Unknown classes
    are a no-op (generic fixture assertions still apply).
    """
    checker = CLASS_CHECKERS.get(failure_class)
    if checker is not None:
        checker(payload)


def _check_env_case_mismatch(payload: str) -> None:
    """env-case-mismatch (#192): a workflow boolean env export emitting
    Python-cased True/False while shell steps compare lowercase 'true'/
    'false' — conditionals unreachable. The recorded payload must still be
    flagged by the boolean-export guard, so a recurrence fails CI.

    Mirrors tests/test_workflow_script_contract.py::test_boolean_env_exports_are_lowercase
    (the #192 discharge, commit b4ddd9d on the release-safety-net branch).
    When that branch merges, import its guard instead of this copy.
    """
    boolean_keys = ("CONVERGED", "AUTO_BUMPED", "DRIFT")
    lowercase_literal = re.compile(r"""['"](true|false)['"]""")
    json_bool_access = re.compile(r"""json\.load\(|\[\s*['"](?:converged|merged)['"]\s*\]""")
    for key in boolean_keys:
        for line in payload.splitlines():
            capture = re.search(rf"^\s*{key}=\$\((.*)\)\s*$", line)
            if not capture:
                continue
            expr = capture.group(1)
            if ".lower()" in expr or lowercase_literal.search(expr):
                continue
            if re.search(r"\b(True|False)\b", expr) or json_bool_access.search(expr):
                raise AssertionError(
                    f"env-case-mismatch recurrence: boolean env export {key} does not force lowercase "
                    f"(str(...).lower()) — shell conditionals compare env.{key} == 'false'-style strings: {line.strip()}"
                )


CLASS_CHECKERS: dict[str, object] = {
    "env-case-mismatch": _check_env_case_mismatch,
}


# ---------------------------------------------------------------------------
# Generated regression test file
# ---------------------------------------------------------------------------

GENERATED_TEST_NAME = "test_production_fixtures.py"

GENERATED_TEST_TEMPLATE = '''"""Generated regression tests over production failure fixtures (#157).

WRITTEN BY scripts/capture_failure.py (template embedded there) — do not
hand-edit the assertions; change the template so generator and generated stay
in lockstep. Parametrizes over tests/fixtures/production/manifest.json: one
regression test per fixture asserting the system's CURRENT handling of the
historical payload (plan 2026-09-07-157):

* sha256 tamper-evidence — a payload edited without re-running capture FAILS;
* sanitiser idempotence — the stored payload is already sanitised and the
  current redactor must leave it unchanged;
* metadata shape — class/source/date/redactions recorded;
* per-class checker — the payload must be CAUGHT by the checker for its
  failure class (see scripts/capture_failure.py::CLASS_CHECKERS), so a
  recurrence of the recorded failure fails CI.

Payloads are snapshots, not live replays: we assert current-code handling of
the historical payload, never that old behaviour is preserved.
"""

from __future__ import annotations

import hashlib
import importlib.util
import json
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[3]
SCRIPTS_DIR = Path("__GENERATED_SCRIPTS_DIR__")
FIXTURES_DIR = Path(__file__).resolve().parent
MANIFEST = FIXTURES_DIR / "manifest.json"


def _load_capture_module():
    # __GENERATED_SCRIPTS_DIR__ is substituted at generation time with the
    # capturing repo's scripts/ directory (absolute). The walk-up below is a
    # secondary fallback for relocated checkouts.
    candidates = [SCRIPTS_DIR]
    candidates.extend(parent / "scripts" for parent in FIXTURES_DIR.parents)
    scripts_dir = next((c for c in candidates if (c / "capture_failure.py").is_file()), SCRIPTS_DIR)
    spec = importlib.util.spec_from_file_location("capture_failure", scripts_dir / "capture_failure.py")
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _manifest_entries() -> list[dict]:
    if not MANIFEST.exists():
        return []
    manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
    return list(manifest.get("fixtures", []))


def assert_fixture_ok(entry: dict, fixtures_dir: Path = FIXTURES_DIR) -> None:
    """The full per-fixture assertion set (shared by the parametrized tests
    and by the pipeline's own tamper-evidence tests)."""
    fixture_file = fixtures_dir / entry["class"] / entry["id"] / "payload.json"
    assert fixture_file.exists(), f"manifest entry {entry['id']} has no payload fixture at {fixture_file}"
    stored = json.loads(fixture_file.read_text(encoding="utf-8"))
    payload = stored.get("payload")
    assert isinstance(payload, str) and payload.strip(), f"{entry['id']}: payload must be a non-empty string"

    # 1. Tamper-evidence: the payload's sha256 must match the manifest.
    actual_sha = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    assert actual_sha == entry["sha256"], (
        f"{entry['id']}: payload sha256 mismatch — the fixture was edited without re-running "
        f"scripts/capture_failure.py (expected {entry['sha256']}, found {actual_sha})"
    )

    # 2. Metadata shape: the manifest entry and the stored metadata agree.
    assert stored.get("failure_class") == entry["class"], f"{entry['id']}: class disagrees with manifest"
    assert stored.get("source") in ("issue", "ci", "manual"), f"{entry['id']}: source must be issue|ci|manual"
    assert stored.get("date"), f"{entry['id']}: capture date missing"
    assert isinstance(stored.get("redactions_applied"), int), f"{entry['id']}: redactions_applied missing"

    # 3. Sanitiser idempotence: the current redactor must be a no-op on the
    #    stored payload (it was sanitised at capture time; re-redacting must
    #    not change it — otherwise the redaction surface has drifted).
    cf = _load_capture_module()
    assert cf.redact(payload) == payload, f"{entry['id']}: payload is not idempotent under the current redactor"
    assert not cf.residual_secrets(payload), f"{entry['id']}: residual credential-shaped text in stored payload"

    # 4. Per-class checker, per the recorded expectation: a "flags" fixture
    #    exhibits the defect — the current checker MUST still raise (the
    #    raise IS the recurrence alarm); a "passes" fixture must stay clean.
    cf = _load_capture_module()
    expectation = entry.get("checker_expectation", "passes")
    if expectation == "flags":
        try:
            cf.check_payload(entry["class"], payload)
        except AssertionError as exc:
            pass  # expected: the checker still catches the recorded defect
        else:
            raise AssertionError(
                f"{entry['id']}: recorded defect is NO LONGER caught by the {entry['class']} checker "
                f"— the class's regression coverage has rotted (checker_expectation=flags)"
            )
    else:
        cf.check_payload(entry["class"], payload)


@pytest.mark.parametrize("entry", _manifest_entries(), ids=lambda e: e["id"])
def test_production_fixture_handling(entry: dict) -> None:
    assert_fixture_ok(entry)


def test_manifest_registry_is_present_and_nonempty() -> None:
    """Guards against vacuous parametrization: once the pipeline is seeded,
    the manifest must exist and carry at least one fixture."""
    assert MANIFEST.exists(), (
        "tests/fixtures/production/manifest.json missing — run scripts/capture_failure.py to seed the registry"
    )
    assert _manifest_entries(), "manifest exists but carries no fixtures — run scripts/capture_failure.py"
'''


def _ensure_generated_test(fixtures_root: Path) -> None:
    """Write the generated regression test file if absent (idempotent, and
    never clobbers: the file is manifest-driven, so regeneration is churn).

    The capturing repo's scripts/ directory (this file's location) is baked
    in as an absolute path so the generated assertions can always import the
    checkers, wherever the fixtures root lives."""
    generated = fixtures_root / GENERATED_TEST_NAME
    if generated.exists():
        return
    body = GENERATED_TEST_TEMPLATE.replace("__GENERATED_SCRIPTS_DIR__", str(Path(__file__).resolve().parent))
    generated.write_text(body, encoding="utf-8")


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="capture_failure.py",
        description="Capture a production failure as a sanitised regression fixture (#157).",
    )
    parser.add_argument("payload", help="path to the raw payload file (error text or JSON blob)")
    parser.add_argument(
        "--class", dest="failure_class", required=True, help="failure class slug, e.g. env-case-mismatch"
    )
    parser.add_argument("--source", required=True, choices=VALID_SOURCES, help="where the failure came from")
    parser.add_argument(
        "--id", dest="fixture_id", default=None, help="fixture slug (default: derived from the payload)"
    )
    parser.add_argument("--fixtures-root", default=DEFAULT_FIXTURES_ROOT, help="fixture root (default: %(default)s)")
    args = parser.parse_args(argv)

    payload_file = Path(args.payload)
    if not payload_file.is_file():
        print(f"capture_failure: payload file not found: {payload_file}", file=sys.stderr)
        return EXIT_USAGE
    try:
        entry = capture(
            payload_file,
            failure_class=args.failure_class,
            source=args.source,
            fixture_id=args.fixture_id,
            fixtures_root=args.fixtures_root,
        )
    except ResidualSecretError as exc:
        print(f"capture_failure: REFUSED — {exc}", file=sys.stderr)
        return EXIT_RESIDUAL_SECRET
    except ValueError as exc:
        print(f"capture_failure: {exc}", file=sys.stderr)
        return EXIT_USAGE

    print(f"captured {entry['id']} -> {Path(args.fixtures_root) / entry['class'] / entry['id'] / 'payload.json'}")
    return EXIT_OK


if __name__ == "__main__":
    sys.exit(main())