"""Tracked-file privacy gate. Diagnostics never include matching source text.

The upstream snapshot exception expires when its owner supplies a cleaned import.
All rectification version files remain in scope, including hash-bound packages.
Local rules are optional, one literal per nonblank line; # starts a comment.
"""
from __future__ import annotations

import ast
import codecs
import json
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path, PurePosixPath
import re
import stat
import subprocess
import sys

import pytest

ROOT = Path(__file__).resolve().parents[1]
SELF_PATH = "tests/test_repo_privacy_markers.py"
GOLDEN_PATH = "tests/test_consultation_contract_golden.py"
LOCAL_PATH = "tests/privacy_markers.local.txt"
UPSTREAM_EXCEPTION = "references/upstream/yinduzhanxing/SKILL.md"
PROTECTED_PREFIX = "skills/jyotish-birth-time-rectification/versions/"
USER_HOME_PREFIX = "/Users/"

# Reviewed synthetic security fixtures, not a username or file-wide allowlist.
# Pin the location AND full source line; moving or extending one fails closed.
# Only the matching prefix span is allowed, never the rest of the line.
SAFE_LINES = {
    "frontend/tests/agent-evals.test.ts": {
        603: 'path: "/Users/example/private.txt",',
    },
    "frontend/tests/personal-report-contract.test.ts": {
        131: '"file:///Users/example/private/chart.json",',
    },
    "tests/test_personal_report_contract.py": {
        190: '"file:///Users/example/private/chart.json",',
    },
    "frontend/tests/agent-observability.test.ts": {
        122: 'modelVersion: "/Users/example/private/model.json",',
        219: 'const rawFailure = "provider rejected user@example.com at /Users/alice/private.json";',
    },
    "frontend/tests/personal-report-api.test.ts": {
        832: 'workflow.internal_path = "/Users/private/project/engine.py";',
        860: 'summary: "Bearer opaque-conflict at /Users/private/trace.py",',
        890: '"/Users/private",',
    },
    "frontend/tests/personal-report-generation.test.ts": {
        1097: 'summary: "Authorization: Bearer secret-token at /Users/private/trace.py",',
        1271: 'wealthWorkflow.internal_path = "/Users/private/project/engine.py";',
        1288: 'summary: "Authorization: Bearer secret-token at /Users/private/trace.py",',
        1291: 'wealthSections["Authorization: Bearer section-secret at /Users/private/tool.py"] = {',
        1373: '"/Users/private",',
    },
    "tests/test_calculation_profile_contract.py": {
        243: 'first = build_with_path("/Users/alice/repo/references/open_source_sources/vedic-astro-skills/ephe")',
    },
    "tests/test_commercial_skill_truth_contract.py": {
        24: 'assert "/Users/" not in str(overlay)',
    },
    "tests/test_professional_report_reference_api.py": {
        113: '"source=scripts/private_engine.py::compute path=/Users/example/report.json "',
        124: 'assert "/Users/" not in sanitized',
        134: '"fixture_path": "/Users/example/references/oracle.json",',
        153: 'assert "/Users/" not in rendered',
    },
    "docs/tasks/UPSTREAM-INSTRUCTION-owner-case-purge-20260919.md": {
        68: '验收：`git grep -c "/Users/" -- . \':!references/open_source_sources\' | wc -l` 为 0。',
        72: '新增 `tests/test_repo_privacy_markers.py`：读 `tests/privacy_markers.local.txt`（gitignore 的）里的模式，扫 `git ls-files`（排除 pdf / wasm / `open_source_sources`），命中即失败；文件不存在时 skip 并打印提示。`/Users/` 前缀作为内置模式始终检查。',
        99: 'git grep -l "/Users/" -- . \':!references/open_source_sources\' | wc -l                                                    # 0',
    },
}


class ScanError(Exception):
    """Only stable codes, never exception strings containing source or paths."""


@dataclass(frozen=True)
class Rule:
    code: str
    value: str = field(repr=False)


@dataclass(frozen=True)
class Finding:
    rule: str
    path: str  # already redacted
    line: int  # zero means filename / file-level failure


@dataclass
class ScanResult:
    findings: list[Finding] = field(default_factory=list)
    scanned: int = 0
    excluded: int = 0
    upstream_exceptions: int = 0
    deleted: int = 0
    protected: int = 0
    permitted: int = 0


def _assignment(tree: ast.Module, name: str) -> ast.Assign | None:
    nodes = [n for n in tree.body if isinstance(n, ast.Assign)
             and len(n.targets) == 1 and isinstance(n.targets[0], ast.Name)
             and n.targets[0].id == name]
    return nodes[0] if len(nodes) == 1 else None


def extract_markers(source: str) -> tuple[str, ...]:
    """Parse, never import or execute the API golden test module."""
    try:
        node = _assignment(ast.parse(source), "FORBIDDEN_PRIVACY_MARKERS")
        if node is None or not isinstance(node.value, (ast.Tuple, ast.List)):
            raise ValueError
        if not all(isinstance(n, ast.Constant) and isinstance(n.value, str)
                   and n.value for n in node.value.elts):
            raise ValueError
        values = tuple(n.value for n in node.value.elts)
        if not values or len(set(values)) != len(values):
            raise ValueError
        return values
    except (SyntaxError, ValueError, TypeError):
        raise ScanError("CONFIG_GOLDEN") from None


def build_rules(golden_source: str, local_source: str = "") -> tuple[Rule, ...]:
    rules = [Rule(f"R{i:03}", value)
             for i, value in enumerate(extract_markers(golden_source), 1)]
    rules.append(Rule("HOME_PREFIX", USER_HOME_PREFIX))
    for i, line in enumerate(local_source.removeprefix(codecs.BOM_UTF8.decode("utf8")).splitlines(), 1):
        value = line.strip()
        if value and not value.startswith("#"):
            rules.append(Rule(f"LOCAL_{i:03}", value))
    return tuple(rules)


def _node_span(source: str, node: ast.AST) -> tuple[int, int]:
    # AST columns count UTF-8 bytes, not Unicode code points.
    lines = source.splitlines(keepends=True)
    start = sum(map(len, lines[:node.lineno - 1]))
    end = sum(map(len, lines[:node.end_lineno - 1]))
    start += len(lines[node.lineno - 1].encode("utf8")[:node.col_offset].decode("utf8"))
    end += len(lines[node.end_lineno - 1].encode("utf8")[:node.end_col_offset].decode("utf8"))
    return start, end


def _declaration_spans(path: str, source: str, rules: tuple[Rule, ...]) -> dict[str, list[tuple[int, int]]]:
    spans: dict[str, list[tuple[int, int]]] = {}
    if path not in {GOLDEN_PATH, SELF_PATH}:
        return spans
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return spans  # malformed files still get scanned, with no exemptions
    if path == GOLDEN_PATH:
        node = _assignment(tree, "FORBIDDEN_PRIVACY_MARKERS")
        if node is not None and isinstance(node.value, (ast.Tuple, ast.List)):
            for child in node.value.elts:
                for rule in rules:
                    if rule.code.startswith("R") and isinstance(child, ast.Constant) and child.value == rule.value:
                        spans.setdefault(rule.code, []).append(_node_span(source, child))
    else:
        prefix = _assignment(tree, "USER_HOME_PREFIX")
        if prefix is not None and isinstance(prefix.value, ast.Constant) and prefix.value.value == USER_HOME_PREFIX:
            spans.setdefault("HOME_PREFIX", []).append(_node_span(source, prefix.value))
        table = _assignment(tree, "SAFE_LINES")
        if table is not None:
            try:
                matches = ast.literal_eval(table.value) == SAFE_LINES
            except (ValueError, TypeError):
                matches = False
            if matches:
                for child in ast.walk(table.value):
                    if isinstance(child, ast.Constant) and isinstance(child.value, str):
                        spans.setdefault("HOME_PREFIX", []).append(_node_span(source, child))
    return spans


def _collision_spans(path: str, source: str, rules: tuple[Rule, ...]) -> list[tuple[int, int]]:
    """Reviewed non-personal R003 collisions; preserve existing numerical evidence.

    The replay generator declares this case public/synthetic, not an owner case.
    Permit only its computed endpoint, not any birth input or another JSON node.
    """
    marker = next((r.value for r in rules if r.code == "R003"), None)
    if marker is None:
        return []
    service_path = "tests/test_rectification_v5_services.py"
    replay_path = "references/oracle/pyjhora_multi_case_panchanga_gulika_replay_2026_07_23.json"
    if path not in {service_path, replay_path}:
        return []
    try:
        tree = ast.parse(source)
    except SyntaxError:
        return []
    if path == service_path:
        name = "test_hour_window_keeps_all_seventeen_signature_clusters_including_late_tail"
        functions = [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == name]
        if len(functions) != 1:
            return []
        spans = []
        for node in ast.walk(functions[0]):
            if not isinstance(node, ast.Set) or node.lineno not in {1182, 1190}:
                continue
            try:
                values = ast.literal_eval(node)
            except (ValueError, TypeError):
                continue
            if values == {f"14:{minute:02}" for minute in range(46, 52)} and len(node.elts) == 6:
                for child in node.elts:
                    if isinstance(child, ast.Constant) and child.value == marker:
                        spans.append(_node_span(source, child))
        return spans
    try:
        data = json.loads(source)
        case = data["cases"][1]
        expected_input = {
            "date": "1996-12-07", "time": "10:34:00", "place": "Chennai",
            "latitude": 13.0878, "longitude": 80.2785, "timezone": "+5.5", "ayanamsa": "Lahiri",
        }
        if case["case_id"] != "public_smoke_chennai_1996" or case["input"] != expected_input:
            return []
        if case["pyjhora_raw"]["yamaganda_kaalam"] != ["13:24:50", marker + ":18"]:
            return []
        # Follow the JSON pointer in its syntax tree, retaining exact value span.
        node = tree.body[0].value
        for key in ("cases", 1, "pyjhora_raw", "yamaganda_kaalam", 1):
            if isinstance(key, int):
                node = node.elts[key]
            else:
                matches = [v for k, v in zip(node.keys, node.values)
                           if isinstance(k, ast.Constant) and k.value == key]
                if len(matches) != 1:
                    return []
                node = matches[0]
        if isinstance(node, ast.Constant) and node.lineno == 149 and node.value == marker + ":18":
            return [_node_span(source, node)]
    except (ValueError, KeyError, IndexError, AttributeError, TypeError):
        pass
    return []


def safe_path(path: str, rules: tuple[Rule, ...]) -> str:
    # Mask whole user-path tails, not just the prefix. Literal markers in
    # filenames must also never reach pytest assertion output or log summaries.
    result = path
    if USER_HOME_PREFIX in result:
        result = result[:result.index(USER_HOME_PREFIX)] + "/[redacted-path]"
    for rule in sorted(rules, key=lambda r: len(r.value), reverse=True):
        result = result.replace(rule.value, "[redacted]")
    return "".join(c if c.isprintable() else "?" for c in result)


def _excluded(path: str) -> bool:
    # Root-relative exclusions only: no broad 'references' or version exemption.
    return (PurePosixPath(path).suffix.lower() in {".pdf", ".wasm"}
            or path.startswith(("vendor/", "references/open_source_sources/")))


def scan_text(path: str, source: str, rules: tuple[Rule, ...]) -> tuple[list[Finding], int]:
    allowed = _declaration_spans(path, source, rules)
    allowed.setdefault("R003", []).extend(_collision_spans(path, source, rules))
    offset = 0
    for line_number, line in enumerate(source.splitlines(keepends=True), 1):
        approved = SAFE_LINES.get(path, {}).get(line_number)
        if approved is not None and line.strip() == approved:
            for match in re.finditer(re.escape(USER_HOME_PREFIX), line):
                allowed.setdefault("HOME_PREFIX", []).append((offset + match.start(), offset + match.end()))
        offset += len(line)
    findings: list[Finding] = []
    permitted = 0
    for rule in rules:
        for match in re.finditer(re.escape(rule.value), source):
            if any(start <= match.start() and match.end() <= end
                   for start, end in allowed.get(rule.code, ())):
                permitted += 1
                continue
            findings.append(Finding(rule.code, safe_path(path, rules), source.count("\n", 0, match.start()) + 1))
    return findings, permitted


def _git(root: Path, *args: str) -> bytes:
    try:
        # Required even for read-only git operations in this shared worktree.
        subprocess.run(["git", "status", "-sb"], cwd=root, check=True,
                       stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30)
        return subprocess.run(["git", *args], cwd=root, check=True,
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30).stdout
    except (OSError, subprocess.SubprocessError):
        raise ScanError("GIT_READ") from None


def tracked_paths(root: Path) -> tuple[str, ...]:
    try:
        return tuple(sorted(set(p for p in _git(root, "ls-files", "-z").decode("utf8").split("\0") if p)))
    except UnicodeError:
        raise ScanError("PATH_ENCODING") from None


def _read(root: Path, path: str) -> str:
    relative = PurePosixPath(path)
    if relative.is_absolute() or ".." in relative.parts or "\\" in path or ":" in path:
        raise OSError("READ_ERROR")
    target = root / path
    # Check from the trusted repository root down before touching a child.
    # On Windows junctions are reparse points even when S_ISLNK is false.
    for ancestor in (root, *(root.joinpath(*relative.parts[:i])
                             for i in range(1, len(relative.parts)))):
        info = ancestor.lstat()
        if (stat.S_ISLNK(info.st_mode)
                or getattr(info, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT
                or not stat.S_ISDIR(info.st_mode)):
            raise OSError("READ_ERROR")
    info = target.lstat()
    if stat.S_ISLNK(info.st_mode):
        try:
            return str(target.readlink())
        except OSError:
            # A failed readlink is not an ordinary deleted tracked file.
            raise OSError("READ_ERROR") from None
    if (getattr(info, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT
            or not stat.S_ISREG(info.st_mode)):
        raise OSError("READ_ERROR")
    data = target.read_bytes()
    # Binary extensions other than the explicit exclusions are not silently
    # skipped. Replacement decoding preserves embedded ASCII privacy markers.
    if data.startswith((b"\xff\xfe", b"\xfe\xff")):
        return data.decode("utf16", errors="replace")
    return data.decode("utf8", errors="replace")


def scan_repository(root: Path) -> ScanResult:
    try:
        golden = _read(root, GOLDEN_PATH)
        try:
            local = _read(root, LOCAL_PATH)
        except FileNotFoundError:
            local = ""
    except OSError:
        raise ScanError("CONFIG_READ") from None
    rules = build_rules(golden, local)
    paths = tracked_paths(root)
    result = ScanResult()
    if LOCAL_PATH in paths:
        result.findings.append(Finding("LOCAL_TRACKED", LOCAL_PATH, 0))
    if _git(root, "check-ignore", "--no-index", LOCAL_PATH).decode("utf8").strip() != LOCAL_PATH:
        raise ScanError("LOCAL_NOT_IGNORED")
    for path in paths:
        if _excluded(path):
            result.excluded += 1
            continue
        if path == UPSTREAM_EXCEPTION:
            result.upstream_exceptions += 1
            continue
        for rule in rules:
            if rule.value in path:
                result.findings.append(Finding(rule.code, safe_path(path, rules), 0))
        try:
            source = _read(root, path)
        except FileNotFoundError:
            # Unstaged tracked deletions remain in ls-files. Report their count;
            # committed CI checkouts naturally have none. Never delete anything.
            result.deleted += 1
            continue
        except OSError:
            result.findings.append(Finding("READ_ERROR", safe_path(path, rules), 0))
            continue
        result.scanned += 1
        if path.startswith(PROTECTED_PREFIX):
            result.protected += 1
        findings, permitted = scan_text(path, source, rules)
        result.findings.extend(findings)
        result.permitted += permitted
    return result


def format_report(result: ScanResult) -> str:
    counts = Counter((f.path, f.rule) for f in result.findings)
    rows = [f"scanned={result.scanned} excluded={result.excluded} deleted={result.deleted} "
            f"upstream_exceptions={result.upstream_exceptions} protected={result.protected} "
            f"permitted={result.permitted} files={len({f.path for f in result.findings})} hits={len(result.findings)}"]
    for (path, rule), count in sorted(counts.items()):
        lines = sorted({f.line for f in result.findings if f.path == path and f.rule == rule})
        rows.append(f"{rule} {path} count={count} lines={','.join(map(str, lines))}")
    return "\n".join(rows)


def test_tracked_repository_has_no_privacy_markers() -> None:
    __tracebackhide__ = True
    try:
        result = scan_repository(ROOT)
    except ScanError as error:
        pytest.fail(str(error), pytrace=False)
    if result.findings:
        pytest.fail(format_report(result), pytrace=False)


# All self-test inputs below are synthetic. Private marker values are never
# copied out of the existing golden declaration, hashed, or encoded here.
def _synthetic_rules() -> tuple[Rule, ...]:
    return (Rule("R001", "fictional-forbidden-token"), Rule("HOME_PREFIX", USER_HOME_PREFIX))


def test_extract_uses_ast_without_executing_module() -> None:
    source = 'raise RuntimeError("must not execute")\nFORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
    assert extract_markers(source) == ("synthetic",)


@pytest.mark.parametrize("source", ["", "FORBIDDEN_PRIVACY_MARKERS = load()",
    'FORBIDDEN_PRIVACY_MARKERS = ("",)', 'FORBIDDEN_PRIVACY_MARKERS = (1,)',
    'FORBIDDEN_PRIVACY_MARKERS = ("x", "x")',
    'FORBIDDEN_PRIVACY_MARKERS = ("x",)\nFORBIDDEN_PRIVACY_MARKERS = ("y",)',
    'FORBIDDEN_PRIVACY_MARKERS = ("unterminated',
])
def test_invalid_marker_declaration_fails_closed(source: str) -> None:
    with pytest.raises(ScanError, match="^CONFIG_GOLDEN$"):
        extract_markers(source)


def test_golden_node_not_comments_other_nodes_or_new_files() -> None:
    rules = _synthetic_rules()
    source = f'FORBIDDEN_PRIVACY_MARKERS = ({rules[0].value!r},)'
    assert scan_text(GOLDEN_PATH, source, rules)[0] == []
    dirty = source + f' # {rules[0].value}\nother = {rules[0].value!r}'
    findings, _ = scan_text(GOLDEN_PATH, dirty, rules)
    assert [f.line for f in findings] == [1, 2]
    assert len(scan_text("tests/new_test.py", source, rules)[0]) == 1


def test_prefix_declaration_is_not_a_private_path_permission() -> None:
    rules = _synthetic_rules()
    source = f'USER_HOME_PREFIX = {USER_HOME_PREFIX!r}'
    assert scan_text(SELF_PATH, source, rules)[0] == []
    assert scan_text(SELF_PATH, f'USER_HOME_PREFIX = {USER_HOME_PREFIX + "new-person/private.txt"!r}', rules)[0]
    dirty = source + f' # {USER_HOME_PREFIX}new-person/private.txt\nother = {USER_HOME_PREFIX!r}'
    assert len(scan_text(SELF_PATH, dirty, rules)[0]) == 2
    assert scan_text("tests/new_test.py", source, rules)[0]


@pytest.mark.parametrize("path,line_number,source_line", [
    (path, number, line) for path, lines in SAFE_LINES.items() for number, line in lines.items()
], ids=[f"safe-{i:02}" for i in range(sum(map(len, SAFE_LINES.values())))])
def test_exact_synthetic_line_permission_rejects_mutations(path: str, line_number: int, source_line: str) -> None:
    rules = _synthetic_rules()
    source = "\n" * (line_number - 1) + source_line
    assert scan_text(path, source, rules)[0] == []
    assert scan_text("tests/new_test.py", source, rules)[0]
    assert scan_text(path, "\n" + source, rules)[0]
    assert scan_text(path, source + "\n" + source_line, rules)[0]
    assert scan_text(path, source.replace(USER_HOME_PREFIX, USER_HOME_PREFIX + "new-person/"), rules)[0]
    assert scan_text(path, source + " # " + rules[0].value, rules)[0]


def test_self_file_and_allowlist_nodes_are_not_whole_file_exemptions() -> None:
    source = (ROOT / SELF_PATH).read_text(encoding="utf8")
    rules = (Rule("HOME_PREFIX", USER_HOME_PREFIX),)
    assert scan_text(SELF_PATH, source, rules)[0] == []
    dirty = source + f'\nextra = {USER_HOME_PREFIX + "new-person/private.txt"!r}'
    assert len(scan_text(SELF_PATH, dirty, rules)[0]) == 1
    mutated = source.replace(USER_HOME_PREFIX + "example/private.txt", USER_HOME_PREFIX + "new-person/private.txt")
    assert scan_text(SELF_PATH, mutated, rules)[0]


@pytest.mark.parametrize("path,excluded", [
    ("assets/file.PDF", True), ("assets/file.wasm", True), ("vendor/lib.txt", True),
    ("references/open_source_sources/source.txt", True),
    ("references/oracle/evidence.json", False),
    ("nested/vendor/local.txt", False),
    (PROTECTED_PREFIX + "sample/references/evidence.json", False),
])
def test_explicit_exclusions_only(path: str, excluded: bool) -> None:
    assert _excluded(path) is excluded


def test_optional_local_markers_are_additive_and_not_values_in_logs() -> None:
    rules = build_rules('FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)', "# note\n\nfictional-local-secret\n")
    assert [r.code for r in rules] == ["R001", "HOME_PREFIX", "LOCAL_003"]
    path = USER_HOME_PREFIX + "new-person/fictional-local-secret.txt"
    findings, _ = scan_text(path, "fictional-local-secret\nsynthetic", rules)
    report = format_report(ScanResult(findings=findings))
    assert "fictional-local-secret" not in report
    assert "new-person" not in report
    assert "synthetic" not in report
    assert "fictional-local-secret" not in repr(rules)
    assert "lines=1" in report and "lines=2" in report


def test_local_python_venv_is_not_tracked() -> None:
    paths = tracked_paths(ROOT)
    assert ".venv" not in paths
    assert not any(path.startswith(".venv/") for path in paths)


def test_gitignore_covers_venv_symlink_name() -> None:
    assert _git(ROOT, "check-ignore", "--no-index", ".venv").decode("utf8").strip() == ".venv"


def test_git_enumeration_is_nul_delimited_and_status_first(monkeypatch: pytest.MonkeyPatch) -> None:
    calls = []
    def run(argv, **kwargs):
        calls.append(argv)
        return subprocess.CompletedProcess(argv, 0, b"tests/a b.py\0tests/\xe4\xb8\xad.py\0" if argv[1] == "ls-files" else b"", b"")
    monkeypatch.setattr(subprocess, "run", run)
    assert tracked_paths(ROOT) == ("tests/a b.py", "tests/中.py")
    assert calls == [["git", "status", "-sb"], ["git", "ls-files", "-z"]]


def test_repository_scan_includes_new_tracked_and_all_protected_files(monkeypatch: pytest.MonkeyPatch) -> None:
    rules = _synthetic_rules()
    paths = {f"{PROTECTED_PREFIX}sample/{i}.md": rules[0].value for i in range(186)}
    paths.update({"tests/new_test.py": rules[0].value,
                  UPSTREAM_EXCEPTION: rules[0].value,
                  "nested/" + UPSTREAM_EXCEPTION: rules[0].value,
                  "vendor/source.py": rules[0].value})
    paths[GOLDEN_PATH] = f'FORBIDDEN_PRIVACY_MARKERS = ({rules[0].value!r},)'
    monkeypatch.setattr(Path, "exists", lambda _: False)
    monkeypatch.setattr(sys.modules[__name__], "tracked_paths", lambda _: tuple(paths))
    monkeypatch.setattr(sys.modules[__name__], "_read", lambda root, path: paths.get(path, ""))
    monkeypatch.setattr(sys.modules[__name__], "_git", lambda *args: LOCAL_PATH.encode())
    result = scan_repository(ROOT)
    assert result.protected == 186
    assert len(result.findings) == 188
    assert result.upstream_exceptions == 1
    assert result.excluded == 1


@pytest.mark.parametrize("path,count", [
    ("tests/test_rectification_v5_services.py", 2),
    ("references/oracle/pyjhora_multi_case_panchanga_gulika_replay_2026_07_23.json", 1),
])
def test_reviewed_numerical_collisions_are_exact_nodes_only(path: str, count: int) -> None:
    __tracebackhide__ = True
    rules = build_rules((ROOT / GOLDEN_PATH).read_text(encoding="utf8"))
    source = (ROOT / path).read_text(encoding="utf8")
    marker = next(r.value for r in rules if r.code == "R003")
    findings, permitted = scan_text(path, source, rules)
    assert len(findings) == 0 and permitted == count
    assert len(scan_text("new/" + path, source, rules)[0]) == count
    # A same-file appended field cannot borrow a reviewed value's permission.
    if path.endswith(".py"):
        dirty = source + f'\nextra = {marker!r}\n'
        assert len(scan_text(path, dirty, rules)[0]) == 1
        dirty = source.replace("test_hour_window_keeps_all_seventeen_signature_clusters_including_late_tail", "test_unreviewed")
        assert len(scan_text(path, dirty, rules)[0]) == count
        dirty = source.replace('"14:46", "14:47"', '"14:45", "14:47"')
        assert len(scan_text(path, dirty, rules)[0]) == count
    else:
        dirty = source.replace('"scope":', f'"extra": {json.dumps(marker)}, "scope":', 1)
        assert len(scan_text(path, dirty, rules)[0]) == 1
        for old, new in [("public_smoke_chennai_1996", "unreviewed_case"),
                         ("yamaganda_kaalam", "birth_time"),
                         ("1996-12-07", "2000-01-01")]:
            assert len(scan_text(path, source.replace(old, new), rules)[0]) == count
        assert len(scan_text(path, source.replace(marker + ":18", marker + ":19"), rules)[0]) == count


def test_local_rule_is_not_exempted_by_golden_declaration() -> None:
    source = 'FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
    rules = build_rules(source, "synthetic")
    findings, permitted = scan_text(GOLDEN_PATH, source, rules)
    assert permitted == 1
    assert [f.rule for f in findings] == ["LOCAL_001"]


def test_unicode_ast_columns_and_same_line_extra_marker() -> None:
    rules = _synthetic_rules()
    source = f'FORBIDDEN_PRIVACY_MARKERS = ("虚构", {rules[0].value!r}); extra = {rules[0].value!r}'
    findings, permitted = scan_text(GOLDEN_PATH, source, rules)
    assert permitted == 1 and len(findings) == 1
    assert findings[0].line == 1


def test_repository_scan_reports_local_tracked_deleted_and_unreadable(monkeypatch: pytest.MonkeyPatch) -> None:
    source = 'FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
    paths = (GOLDEN_PATH, LOCAL_PATH, "deleted.md", "unreadable.md", "binary.dat")
    def read(root, path):
        if path == "deleted.md":
            raise FileNotFoundError
        if path == "unreadable.md":
            raise PermissionError("must never be rendered")
        return {GOLDEN_PATH: source, LOCAL_PATH: "fictional-local", "binary.dat": "\0synthetic\0"}[path]
    monkeypatch.setattr(Path, "exists", lambda _: True)
    monkeypatch.setattr(sys.modules[__name__], "tracked_paths", lambda _: paths)
    monkeypatch.setattr(sys.modules[__name__], "_read", read)
    monkeypatch.setattr(sys.modules[__name__], "_git", lambda *args: LOCAL_PATH.encode())
    result = scan_repository(ROOT)
    assert result.deleted == 1
    assert {f.rule for f in result.findings} == {"LOCAL_TRACKED", "LOCAL_001", "READ_ERROR", "R001"}
    assert "must never be rendered" not in format_report(result)


def test_git_failures_do_not_leak_stderr_or_command_values(monkeypatch: pytest.MonkeyPatch) -> None:
    def run(*args, **kwargs):
        raise subprocess.CalledProcessError(1, ["git", "fictional-sensitive-argv"], stderr="fictional-sensitive-stderr")
    monkeypatch.setattr(subprocess, "run", run)
    with pytest.raises(ScanError, match="^GIT_READ$"):
        tracked_paths(ROOT)


def test_optional_file_missing_does_not_disable_repository_scan(monkeypatch: pytest.MonkeyPatch) -> None:
    source = 'FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
    monkeypatch.setattr(Path, "exists", lambda _: False)
    monkeypatch.setattr(sys.modules[__name__], "tracked_paths", lambda _: ("new.md",))
    def read(root, path):
        if path == LOCAL_PATH:
            raise FileNotFoundError
        return source if path == GOLDEN_PATH else "synthetic"
    monkeypatch.setattr(sys.modules[__name__], "_read", read)
    monkeypatch.setattr(sys.modules[__name__], "_git", lambda *args: LOCAL_PATH.encode())
    result = scan_repository(ROOT)
    assert len(result.findings) == 1 and result.findings[0].rule == "R001"


@pytest.mark.parametrize("target_text", [
    "../fictional-outside/fictional-forbidden-token.txt",
    "../fictional-missing/fictional-forbidden-token.txt",
], ids=["external-target", "dangling-target"])
def test_symlink_scans_target_text_without_opening_target(monkeypatch: pytest.MonkeyPatch, target_text: str) -> None:
    from types import SimpleNamespace
    target = ROOT / "tracked-link"
    calls = []
    def lstat(path):
        calls.append(path)
        return SimpleNamespace(st_mode=stat.S_IFLNK if path == target else stat.S_IFDIR)
    def forbidden_read(path):
        pytest.fail("symlink content must not be opened")
    monkeypatch.setattr(Path, "lstat", lstat)
    monkeypatch.setattr(Path, "readlink", lambda path: Path(target_text))
    monkeypatch.setattr(Path, "read_bytes", forbidden_read)
    text = _read(ROOT, "tracked-link")
    findings, _ = scan_text("tracked-link", text, _synthetic_rules())
    assert len(findings) == 1 and findings[0].rule == "R001"
    assert calls == [ROOT, target]


@pytest.mark.parametrize("junction", [False, True], ids=["symlink", "junction"])
def test_linked_ancestor_fails_without_accessing_descendants(monkeypatch: pytest.MonkeyPatch, junction: bool) -> None:
    from types import SimpleNamespace
    ancestor = ROOT / "linked-directory"
    calls = []
    def lstat(path):
        calls.append(path)
        if path == ROOT:
            return SimpleNamespace(st_mode=stat.S_IFDIR)
        assert path == ancestor
        return SimpleNamespace(st_mode=stat.S_IFDIR if junction else stat.S_IFLNK,
                               st_file_attributes=stat.FILE_ATTRIBUTE_REPARSE_POINT if junction else 0)
    def forbidden_access(path):
        pytest.fail("ancestor link must not be followed")
    monkeypatch.setattr(Path, "lstat", lstat)
    monkeypatch.setattr(Path, "readlink", forbidden_access)
    monkeypatch.setattr(Path, "read_bytes", forbidden_access)
    with pytest.raises(OSError, match="^READ_ERROR$"):
        _read(ROOT, "linked-directory/child.txt")
    assert calls == [ROOT, ancestor]


def test_readlink_failure_is_not_reported_as_deleted(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace
    target = ROOT / "tracked-link"
    monkeypatch.setattr(Path, "lstat", lambda path: SimpleNamespace(st_mode=stat.S_IFLNK if path == target else stat.S_IFDIR))
    def readlink(path):
        raise FileNotFoundError("fictional-sensitive-error")
    monkeypatch.setattr(Path, "readlink", readlink)
    with pytest.raises(OSError, match="^READ_ERROR$") as error:
        _read(ROOT, "tracked-link")
    assert not isinstance(error.value, FileNotFoundError)


def test_missing_regular_tracked_file_remains_deleted(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace
    def lstat(path):
        if path == ROOT:
            return SimpleNamespace(st_mode=stat.S_IFDIR)
        raise FileNotFoundError
    monkeypatch.setattr(Path, "lstat", lstat)
    with pytest.raises(FileNotFoundError):
        _read(ROOT, "deleted.txt")


def test_regular_file_still_scans_bytes(monkeypatch: pytest.MonkeyPatch) -> None:
    from types import SimpleNamespace
    target = ROOT / "regular.txt"
    calls = []
    monkeypatch.setattr(Path, "lstat", lambda path: SimpleNamespace(st_mode=stat.S_IFREG if path == target else stat.S_IFDIR))
    def read_bytes(path):
        calls.append(path)
        return b"fictional-forbidden-token"
    monkeypatch.setattr(Path, "read_bytes", read_bytes)
    assert _read(ROOT, "regular.txt") == "fictional-forbidden-token"
    assert calls == [target]


def test_optional_local_utf8_bom_does_not_disable_first_rule() -> None:
    source = 'FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
    local = "fictional-local-secret"
    plain = build_rules(source, local)
    with_bom = build_rules(source, codecs.BOM_UTF8.decode("utf8") + local)
    assert plain == with_bom
    findings, _ = scan_text("tests/synthetic.txt", local, with_bom)
    assert len(findings) == 1 and findings[0].rule == "LOCAL_001"


def test_quick_profile_explicitly_includes_repository_guard() -> None:
    source = (ROOT / "scripts/run_quality_gate.py").read_text(encoding="utf8")
    node = _assignment(ast.parse(source), "CORE_PYTEST_TARGETS")
    assert node is not None and SELF_PATH in ast.literal_eval(node.value)