"""task-2738 — rule_7_block_executor_collector_callback dedicated tests.

Goal (instruction §2): an executor/dev session firing
`cokacdir --cron "[ANU 수집 콜백]..." --key <self>` from Bash must be DENY/HOLD,
while normal reminders/schedules and the legitimate ANU-owned pickup runner path
stay ALLOW.  Prompt/command TEXT is never an authorization basis — a textual
`key=ANU` / `collector_role=ANU` / `self_key_used=false` claim is a BLOCK signal.

Covers the chair-10 matrix (instruction §4):
  ① Bash cokacdir --cron "[ANU 수집 콜백]" --key <dev/self>            → DENY/HOLD
  ② collector_role=ANU / self_key_used=false / key=ANU direct schedule → DENY/HOLD
  ③ normal non-ANU reminder/schedule                                  → ALLOW
  ④ ANU-owned pickup runner legit path (runtime owner-proof)          → ALLOW
  ⑤ existing high-risk DENY / benign ALLOW regression (6-rule intact)
  ⑥ import_guard crash increase 0 (import PASS, rule_7 wired)
  ⑦ session-bound polling 0 (no sleep/poll/watcher introduced by rule_7)
  ⑧ ANU key literal 0 (no raw secret in rule_7 source)
  ⑨ systemd/P0B/driver/ACTIVE=true 0 (rule_7 introduces none)
  ⑩ expected-files-outside diff 0 is asserted in the closeout report, not here.

Isolation: ANU_V36_HARNESS_TEST_MODE=1 → JSONL writes go to /tmp. No real tool
is executed; only mock payloads are evaluated.
"""
from __future__ import annotations

import re
from pathlib import Path

import pytest

# Repo root = .worktrees/task-2738-dev2 (three parents up from this file).
_REPO_ROOT = Path(__file__).resolve().parents[2]
_RULES_SRC = _REPO_ROOT / "scripts/harness/v36/rules.py"


@pytest.fixture(autouse=True)
def _test_mode_env(monkeypatch):
    monkeypatch.setenv("ANU_V36_HARNESS_TEST_MODE", "1")
    monkeypatch.delenv("V36_HARNESS_LOG_ALLOW", raising=False)
    yield


# Owner-proof context for the legitimate ANU-owned pickup runner path. This is a
# RUNTIME label assembled by the runner harness — never derivable from command
# text. owner proof is a boolean/label only (no raw key).
_RUNNER_CTX = {
    "runner_identity": "anu_owned_pickup_runner",
    "owner_key_proof_present": True,
    "schedule_owner_proof": "ANU",
}


# ---------------------------------------------------------------------------
# ① executor/dev self-key "[ANU 수집 콜백]" creation → DENY/HOLD
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        '/usr/local/bin/cokacdir --cron "[ANU 수집 콜백] task-2738 완료 수집" '
        "--at 30m --chat 6937032012 --key SELFKEY123",
        'cokacdir --cron "[ANU 수집 콜백] 결과 pickup" --at "0 9 * * 1" --key devkey',
        'cokacdir --cron "ANU 수집 콜백 발사 — dev2 self" --at 1h --key abc',
    ],
)
def test_t1_executor_anu_collector_envelope_blocked(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})  # no runner identity (dev session)
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), r
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# ①b task-2738-r1 [HIGH] separator-bypass regression — `_` / `-` / repeated
#     spaces / mixed separators / case must ALL be DENY (previously ALLOW).
#     This is the core defect fixed in r1: the Korean label markers matched only
#     `\s*`, so swapping spaces for `_`/`-` slipped past as an unauthorized
#     self-collector.
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        # underscore separators
        'cokacdir --cron "[ANU_수집_콜백] task-2738 완료 수집" --at 30m --key SELFKEY',
        # hyphen separators
        'cokacdir --cron "[ANU-수집-콜백] 결과 pickup" --at "0 9 * * 1" --key devkey',
        # repeated/multiple spaces
        'cokacdir --cron "[ANU   수집   콜백] 발사" --at 1h --key abc',
        'cokacdir --cron "ANU   수집   콜백 — dev2 self" --at 1h --key abc',
        # mixed separators (underscore + hyphen)
        'cokacdir --cron "ANU_수집-콜백 dispatch" --at 10m --key k',
        'cokacdir --cron "[ANU_수집-콜백] pickup" --at 10m --key k',
        # no separator at all (glued tokens)
        'cokacdir --cron "[ANU수집콜백] fire" --at 10m --key k',
        # case-insensitive ANU token
        'cokacdir --cron "[anu_수집_콜백] pickup" --at 10m --key k',
        # English collector-callback with _/- separators
        'cokacdir --cron "ANU_collector_callback dispatch" --at 10m --key k',
        'cokacdir --cron "ANU-collector-callback dispatch" --at 10m --key k',
    ],
)
def test_t1b_separator_bypass_variants_blocked(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})  # dev session, no runner identity
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), r
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


def test_t1c_separator_variants_still_blocked_at_rule_level():
    """Assert directly on rule_7 so the fix is pinned at the marker layer."""
    from scripts.harness.v36.rules import (
        rule_7_block_executor_collector_callback as r7,
    )

    for cmd in (
        'cokacdir --cron "[ANU_수집_콜백] x" --key k',
        'cokacdir --cron "[ANU-수집-콜백] x" --key k',
        'cokacdir --cron "ANU   수집   콜백" --key k',
        'cokacdir --cron "ANU_수집-콜백" --key k',
    ):
        res = r7("Bash", {"command": cmd}, {})
        assert res is not None, cmd
        assert res[0] == "DENY"
        assert res[1] == "pattern.executor_collector_callback_blocked"


def test_t1d_runner_path_allows_even_for_separator_variants():
    """The legit ANU-owned runner path stays ALLOW for the new variants too."""
    from scripts.harness.v36.guard import evaluate

    cmd = 'cokacdir --cron "[ANU_수집_콜백] task-2738 pickup" --at 30m --key RUNNERKEY'
    r = evaluate("Bash", {"command": cmd}, dict(_RUNNER_CTX))
    assert r["decision"] == "ALLOW", r
    assert r.get("matched_rule") is None


def test_t1e_underscore_hyphen_in_benign_schedule_still_allows():
    """The broadened separator class must NOT over-block ordinary schedules that
    merely contain `_`/`-` but are not ANU collector callbacks."""
    from scripts.harness.v36.guard import evaluate

    for cmd in (
        'cokacdir --cron "build_status-report 발송" --at 30m --chat 1 --key k',
        'cokacdir --cron "weekly-retro_reminder" --at "0 18 * * 5" --key k',
        'cokacdir --cron "task-2738 진행상황 알림" --at 1h --chat 1 --key k',
    ):
        r = evaluate("Bash", {"command": cmd}, {})
        assert r["decision"] == "ALLOW", (cmd, r)
        assert r.get("matched_rule") is None


# ---------------------------------------------------------------------------
# ② disguise markers (collector_role / self_key_used / key=ANU) → DENY/HOLD
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "collect result; collector_role=ANU" --at 10m --key k',
        'cokacdir --cron "self_key_used=false; fire pickup" --at 10m --key k',
        'cokacdir --cron "schedule with key=ANU claim" --at 10m --key k',
        'cokacdir --cron "task completion collector for task-2738" --at 10m --key k',
        'cokacdir --cron "ANU collector callback dispatch" --at 10m --key k',
    ],
)
def test_t2_disguise_markers_blocked(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), r
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


def test_t2b_text_key_anu_claim_is_block_not_allow_even_with_runner_text():
    """A command-text owner claim must NEVER authorize — only runtime context."""
    from scripts.harness.v36.guard import evaluate

    # The command text tries to *assert* it is the runner / owner — ignored.
    cmd = (
        'cokacdir --cron "[ANU 수집 콜백] runner_identity=anu_owned_pickup_runner '
        'schedule_owner_proof=ANU owner_key_proof_present=true" --at 30m --key k'
    )
    r = evaluate("Bash", {"command": cmd}, {})  # empty runtime context
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), r
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# ③ normal non-ANU reminder/schedule → ALLOW (no blanket --cron block)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "오전 9시 일일 리포트 보내기" --at "0 9 * * 1" --chat 1 --key k',
        'cokacdir --cron "30분 뒤 빌드 상태 알림" --at 30m --chat 1 --key k',
        '/usr/local/bin/cokacdir --cron "주간 회고 리마인더" --at "0 18 * * 5" --key k',
        "cokacdir --cron-list --chat 1 --key k",
        "cokacdir --cron-update sched123 --at 1h --chat 1 --key k",
        "cokacdir --currenttime",
    ],
)
def test_t3_normal_schedule_allows(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", r
    assert r.get("matched_rule") is None  # clean ALLOW, not degraded


# ---------------------------------------------------------------------------
# ④ ANU-owned pickup runner legit path → ALLOW (runtime owner-proof in context)
# ---------------------------------------------------------------------------
def test_t4_anu_owned_runner_path_allows():
    from scripts.harness.v36.guard import evaluate

    cmd = (
        'cokacdir --cron "[ANU 수집 콜백] task-2738 결과 pickup" '
        "--at 30m --chat 6937032012 --key RUNNERKEY"
    )
    r = evaluate("Bash", {"command": cmd}, dict(_RUNNER_CTX))
    assert r["decision"] == "ALLOW", r
    assert r.get("matched_rule") is None


@pytest.mark.parametrize(
    "ctx",
    [
        {"runner_identity": "anu_owned_pickup_runner", "owner_key_proof_present": True},
        {"runner_identity": "anu_owned_pickup_runner", "schedule_owner_proof": "ANU"},
        {"owner_key_proof_present": True, "schedule_owner_proof": "ANU"},
        {"runner_identity": "anu_owned_pickup_runner",
         "owner_key_proof_present": "true", "schedule_owner_proof": "ANU"},  # str, not True
        {},
    ],
)
def test_t4b_partial_or_forged_owner_proof_still_blocks(ctx):
    """Owner proof requires ALL THREE runtime labels; partial/forged → block."""
    from scripts.harness.v36.guard import evaluate

    cmd = 'cokacdir --cron "[ANU 수집 콜백] pickup" --at 30m --key k'
    r = evaluate("Bash", {"command": cmd}, dict(ctx))
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (ctx, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# ⑤ regression: existing high-risk DENY / benign ALLOW — 6-rule untouched
# ---------------------------------------------------------------------------
def test_t5_six_rule_contract_intact():
    from scripts.harness.v36.rules import ALL_RULES, RULE_7_EXTENSION

    # ALL_RULES stays the canonical 6 (length/order contract intact).
    assert len(ALL_RULES) == 6
    assert len(RULE_7_EXTENSION) == 1
    assert (
        RULE_7_EXTENSION[0].__name__ == "rule_7_block_executor_collector_callback"
    )


@pytest.mark.parametrize(
    "cmd,expect",
    [
        ("git push origin main", "DENY"),
        ("gh run watch 123", "DENY"),
        ("git branch -D main", "DENY"),
        ("ls -la", "ALLOW"),
        ("git status", "ALLOW"),
        ("cat README.md", "ALLOW"),
    ],
)
def test_t5b_existing_rules_regression(cmd, expect):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == expect, (cmd, r)


def test_t5c_rule7_does_not_shadow_cron_remove_high_risk_degraded(monkeypatch):
    """--cron-remove must not be swallowed by rule_7; degraded path fails closed."""
    import scripts.harness.v36.guard as g
    from scripts.harness.v36.guard import evaluate

    # rule_7 must return None for a non-create cron sub-command.
    from scripts.harness.v36.rules import rule_7_block_executor_collector_callback as r7
    assert r7("Bash", {"command": "cokacdir --cron-remove abc --key k"}, {}) is None

    # And on the degraded path the high-risk cron-remove still fails closed.
    def _boom(*_a, **_k):
        raise RuntimeError("forced")

    monkeypatch.setattr(g, "_evaluate_core", _boom)
    r = evaluate("Bash", {"command": "cokacdir --cron-remove abc --key k"}, {})
    assert r["decision"] == "DENY"
    assert r.get("degraded") is True


# ---------------------------------------------------------------------------
# ⑥ import_guard crash increase 0 — import PASS + rule_7 wired into evaluate
# ---------------------------------------------------------------------------
def test_t6_import_and_wiring_pass():
    from scripts.harness.v36.guard import _EVAL_RULES, evaluate  # noqa: F401
    from scripts.harness.v36.rules import (  # noqa: F401
        ALL_RULES,
        RULE_7_EXTENSION,
        rule_7_block_executor_collector_callback,
    )

    names = [fn.__name__ for fn in _EVAL_RULES]
    assert "rule_7_block_executor_collector_callback" in names
    assert len(_EVAL_RULES) == 7  # 6 canonical + rule_7


# ---------------------------------------------------------------------------
# ⑦ session-bound polling 0 — rule_7 source has no sleep/poll/watch/background
# ---------------------------------------------------------------------------
def test_t7_no_polling_or_watcher_introduced():
    src = _RULES_SRC.read_text(encoding="utf-8")
    # Isolate the rule_7 region for the scan.
    start = src.index("def rule_7_block_executor_collector_callback")
    region = src[start:]
    for forbidden in (
        "time.sleep",
        "while True",
        "run_in_background",
        "inotifywait",
        "subprocess",
        "threading",
    ):
        assert forbidden not in region, f"rule_7 must not introduce {forbidden!r}"


# ---------------------------------------------------------------------------
# ⑧ ANU key literal 0 — no raw secret in rule_7 source
# ---------------------------------------------------------------------------
def test_t8_no_raw_anu_key_literal():
    src = _RULES_SRC.read_text(encoding="utf-8")
    start = src.index("def rule_7_block_executor_collector_callback")
    region = src[start:]
    # The literal word "ANU" and the detection marker "key=ANU" are fine; an
    # actual secret is not. Heuristic: no long token assigned to a key-like name,
    # and no PEM/sealed-key constant smuggled into the rule_7 region.
    assert not re.search(
        r"(ANU_KEY|anu_key|SELF_KEY)\s*=\s*['\"][A-Za-z0-9+/=]{16,}['\"]", src
    )
    assert "-----BEGIN" not in region
    # No long opaque token literal inside the rule_7 region.
    assert not re.search(r"['\"][A-Za-z0-9+/]{24,}={0,2}['\"]", region)


# ---------------------------------------------------------------------------
# ⑨ systemd/P0B/driver/ACTIVE=true 0 — rule_7 introduces no activation
# ---------------------------------------------------------------------------
def test_t9_no_activation_artifacts_in_rule7():
    src = _RULES_SRC.read_text(encoding="utf-8")
    start = src.index("def rule_7_block_executor_collector_callback")
    region = src[start:]
    for forbidden in ("systemctl", "ACTIVE=true", "ACTIVE = true", "p0b_driver", "driver_loop"):
        assert forbidden not in region, f"rule_7 must not reference {forbidden!r}"


# ===========================================================================
# task-2738-r2 — structured segment classifier (fresh NEW actual 3 + chair-10)
#
# r1 fixed the separator-bypass at the MARKER layer but still evaluated the
# command as a FLAT string: a `--cron-list/-history` management subcommand
# ANYWHERE in the command early-returned ALLOW, masking a direct collector
# cron-create in a later chaining segment.  r2 replaces that with a quote-aware
# per-segment classifier.  These tests pin the three fresh NEW actuals and the
# chair-mandated 10 regressions (instruction §4).
# ===========================================================================


# ---------------------------------------------------------------------------
# r2-① [CRITICAL] subcommand-chaining bypass: a management subcommand in an
#      earlier segment must NOT mask a direct collector cron-create later.
#      (instruction §0-1 / §4.1 / §4.2)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        # §4.1 — cron-list ; collector create
        'cokacdir --cron-list ; cokacdir --cron "[ANU 수집 콜백] task-2738 결과 pickup" --key self',
        # §4.2 — cron-history && collector create
        'cokacdir --cron-history && cokacdir --cron "[ANU 수집 콜백] pickup" --key self',
        # other management subcommands first, then collector create
        'cokacdir --cron-remove old123 ; cokacdir --cron "[ANU 수집 콜백] fire" --key self',
        'cokacdir --cron-update s1 --at 1h || cokacdir --cron "ANU 수집 콜백 발사" --key self',
        # pipe boundary
        'cokacdir --cron-list | cokacdir --cron "[ANU_수집_콜백] pickup" --key self',
        # three segments, collector buried last
        'echo start ; cokacdir --cron-list ; cokacdir --cron "[ANU 수집 콜백]" --key self',
    ],
)
def test_r2_t1_subcommand_chaining_does_not_mask_collector_create(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})  # dev session, no runner identity
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


def test_r2_t1b_chaining_block_at_rule_level():
    """Pin the structural fix directly at rule_7 (not only via guard)."""
    from scripts.harness.v36.rules import (
        rule_7_block_executor_collector_callback as r7,
    )

    cmd = 'cokacdir --cron-list ; cokacdir --cron "[ANU 수집 콜백]" --key self'
    res = r7("Bash", {"command": cmd}, {})
    assert res is not None, cmd
    assert res[0] == "DENY"
    assert res[1] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# r2-② [HIGH] key=ANU_PROD / key=ANU_DEV / key=ANU_TEST / key=ANU-* disguise
#      must DENY (the r1 `\bkey\s*=\s*ANU\b` boundary let the `_`/`-` suffixed
#      variants through).  (instruction §0-1 / §2-1 / §4.3)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "schedule key=ANU_PROD" --at 10m --key k',
        'cokacdir --cron "schedule key=ANU_DEV" --at 10m --key k',
        'cokacdir --cron "schedule key=ANU_TEST" --at 10m --key k',
        'cokacdir --cron "schedule key=ANU-stg" --at 10m --key k',
        'cokacdir --cron "fire key=ANU" --at 10m --key k',  # plain key=ANU still blocked
        # chained form: management subcommand first, key=ANU_PROD create later
        'cokacdir --cron-list ; cokacdir --cron "pickup key=ANU_PROD" --at 10m --key k',
    ],
)
def test_r2_t2_key_anu_suffix_disguise_blocked(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


def test_r2_t2b_key_anu_marker_regex_matches_suffix_variants():
    """Pin the marker-regex fix: every key=ANU* / key=ANU-* text matches."""
    from scripts.harness.v36.rules import _R7_COLLECTOR_MARKERS

    def _any_marker(text):
        return any(p.search(text) for p in _R7_COLLECTOR_MARKERS)

    for text in ("key=ANU", "key=ANU_PROD", "key=ANU_DEV", "key=ANU_TEST",
                 "key=ANU-stg", "key = ANU_PROD", "key='ANU_PROD'"):
        assert _any_marker(text), text
    # The real --key hex *argument* form (space-separated, no =ANU) is NOT a marker.
    assert not _any_marker("--key a1b2c3d4e5f60718")
    assert not _any_marker("--key deadbeefcafe0001")


# ---------------------------------------------------------------------------
# r2-③ [MED] quote-aware segment splitting: separators INSIDE quotes are NOT
#      chaining boundaries → benign multi-clause schedule ALLOW.
#      (instruction §4.7)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "회의 ; 정리 && 보고" --at 9am',
        "cokacdir --cron '회의 ; 정리 && 보고 | 마무리' --at 9am --key k",
        'cokacdir --cron "build; test && deploy | notify" --at 1h --chat 1 --key k',
    ],
)
def test_r2_t3_quoted_separators_are_not_boundaries_allow(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


def test_r2_t3b_split_segments_is_quote_aware():
    from scripts.harness.v36.rules import _split_segments

    # quoted separators stay inside one segment
    segs = _split_segments('cokacdir --cron "회의 ; 정리 && 보고" --at 9am')
    assert len(segs) == 1, segs
    # unquoted separators split
    segs2 = _split_segments("a ; b && c || d | e")
    assert len(segs2) == 5, segs2


# ---------------------------------------------------------------------------
# r2-④ real --key <hex> argument on a non-collector schedule → ALLOW
#      (instruction §2-1 / §4.8 — no over-block)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "일정" --key a1b2c3d4e5f60718',
        'cokacdir --cron "주간 회고 리마인더" --at "0 18 * * 5" --chat 1 --key deadbeefcafe1234',
        'cokacdir --cron-list ; cokacdir --cron "30분 뒤 빌드 알림" --at 30m --key a1b2c3d4e5f60718',
    ],
)
def test_r2_t4_real_hex_key_non_collector_allows(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


# ---------------------------------------------------------------------------
# r2-⑤ standalone management subcommands stay ALLOW (instruction §4.4)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        "cokacdir --cron-list --chat 1 --key k",
        "cokacdir --cron-history sched123 --chat 1 --key k",
        "cokacdir --cron-update sched123 --at 1h --chat 1 --key k",
        "cokacdir --cron-remove sched123 --chat 1 --key k",
    ],
)
def test_r2_t5_standalone_management_subcommand_allows(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    # cron-remove is high-risk on the *degraded* path only; on the normal path
    # rule_7 must NOT fire (no collector create) → not blocked by rule_7.
    assert r.get("matched_rule") != "pattern.executor_collector_callback_blocked", (cmd, r)
    assert r["decision"] == "ALLOW", (cmd, r)


# ---------------------------------------------------------------------------
# r2-⑥ ANU-owned runner safe path stays ALLOW even with chaining (instruction §4.6)
# ---------------------------------------------------------------------------
def test_r2_t6_runner_path_allows_with_chaining():
    from scripts.harness.v36.guard import evaluate

    cmd = 'cokacdir --cron-list ; cokacdir --cron "[ANU 수집 콜백] pickup" --at 30m --key RUNNERKEY'
    r = evaluate("Bash", {"command": cmd}, dict(_RUNNER_CTX))
    assert r["decision"] == "ALLOW", r
    assert r.get("matched_rule") is None


# ---------------------------------------------------------------------------
# r2-⑦ r1 separator-variant regression still DENY (instruction §4.9)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "[ANU 수집 콜백]" --key self',
        'cokacdir --cron "[ANU_수집_콜백]" --key self',
        'cokacdir --cron "[ANU-수집-콜백]" --key self',
        'cokacdir --cron "[ANU   수집   콜백]" --key self',
        'cokacdir --cron "ANU_수집-콜백 dispatch" --key self',
    ],
)
def test_r2_t7_r1_separator_regression_still_denied(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# r2-⑧ parser is pure (no shell exec/eval) + crash 0 on malformed/abnormal quotes
#      (instruction §2.5 / §2.6 / §4.10)
# ---------------------------------------------------------------------------
def test_r2_t8_parser_has_no_shell_exec_or_eval():
    src = _RULES_SRC.read_text(encoding="utf-8")
    # _split_segments must be a pure char scan — no exec/eval/shell-out.
    start = src.index("def _split_segments")
    end = src.index("def _segment_is_direct_collector_cron_create")
    region = src[start:end]
    for forbidden in (
        "subprocess", "os.system", "os.popen", "eval(", "exec(",
        "shell=True", "Popen", "shlex.split", "run(", "check_output",
    ):
        assert forbidden not in region, f"_split_segments must not use {forbidden!r}"


@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "[ANU 수집 콜백] --key self',   # unbalanced double quote
        "cokacdir --cron '[ANU 수집 콜백] --key self",   # unbalanced single quote
        'cokacdir --cron "a; b" && cokacdir --cron "[ANU 수집 콜백]" --key self"',  # trailing quote
        '',                                               # empty
        ';;;|&&||',                                       # only separators
        'cokacdir --cron "\\"escaped\\" [ANU 수집 콜백]" --key self',  # escaped quotes
    ],
)
def test_r2_t8b_malformed_quotes_no_crash_explicit_decision(cmd):
    """Abnormal/unbalanced quotes must NOT crash the guard — an explicit
    ALLOW/DENY/HOLD decision is always returned (fail-safe, crash increase 0)."""
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("ALLOW", "DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    # never a degraded/crash decision from rule_7 parsing
    assert r.get("matched_rule") != "degraded.fail_closed_unknown", (cmd, r)


def test_r2_t8c_split_segments_never_raises():
    from scripts.harness.v36.rules import _split_segments

    for cmd in ('"', "'", '"unterminated', "a;b", "", "|||", '\\', 'x"y\'z'):
        segs = _split_segments(cmd)
        assert isinstance(segs, list) and len(segs) >= 1, cmd


# ---------------------------------------------------------------------------
# r2-⑨ benign chaining (no ANU collector marker anywhere) → ALLOW
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron-list ; cokacdir --cron "주간 회고 리마인더" --at "0 18 * * 5" --key k',
        'git status && cokacdir --cron "빌드 알림" --at 30m --chat 1 --key k',
        'cokacdir --cron "task-2738 진행상황 알림" --at 1h --chat 1 --key k | cat',
    ],
)
def test_r2_t9_benign_chaining_allows(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


# ===========================================================================
# task-2738-r3 — key=ANU over-block correction (fresh MEDIUM 1)
#
# r2's `key=ANU(?:[_\-][\w-]*)?` suffix run was FULLY OPTIONAL, so the marker
# also matched benign keys that merely START with "ANU" and then CONTINUE with
# alphanumerics (`key=anurag`/`key=anup`/`key=ANUX`/`key=ANU123`/`key=ANUPROD`),
# over-blocking ordinary non-ANU schedules.  r3 inserts a negative lookahead
# `(?![a-zA-Z0-9])` after `ANU`: a directly-following letter/digit → benign
# (no match → ALLOW); `_`/`-`/end/quote/space → disguise marker stays DENY.
# These tests pin the chair-mandated 8 regressions (instruction §4).
# ===========================================================================


# ---------------------------------------------------------------------------
# r3-① §4.1 — key=ANU_PROD / key=ANU_DEV / key=ANU_TEST disguise → DENY (kept)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "schedule key=ANU_PROD" --at 10m --key k',
        'cokacdir --cron "schedule key=ANU_DEV" --at 10m --key k',
        'cokacdir --cron "schedule key=ANU_TEST" --at 10m --key k',
        'cokacdir --cron "schedule key=ANU_" --at 10m --key k',   # trailing underscore
        'cokacdir --cron "schedule key=ANU-" --at 10m --key k',   # trailing hyphen
        'cokacdir --cron "fire key=ANU" --at 10m --key k',        # bare key=ANU
    ],
)
def test_r3_t1_key_anu_marker_disguise_still_denied(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# r3-② §4.2 — benign lowercase keys starting with "anu" → ALLOW (over-block gone)
# r3-③ §4.3 — benign keys "ANU" + alphanumeric continuation → ALLOW
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        # §4.2 — anurag / anup / anubis / anuria
        'cokacdir --cron "schedule key=anurag" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=anup" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=anubis" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=anuria" --at 10m --chat 1 --key k',
        # §4.3 — ANUX / ANU123 / ANUPROD
        'cokacdir --cron "schedule key=ANUX" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=ANU123" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=ANUPROD" --at 10m --chat 1 --key k',
    ],
)
def test_r3_t2_benign_anu_prefixed_keys_allow(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


def test_r3_t2b_marker_regex_excludes_benign_keeps_disguise():
    """Pin the lookahead fix directly at the marker layer: benign ANU-prefixed
    keys do NOT match any marker; the `_`/`-`/bare disguise forms still do."""
    from scripts.harness.v36.rules import _R7_COLLECTOR_MARKERS

    def _any_marker(text):
        return any(p.search(text) for p in _R7_COLLECTOR_MARKERS)

    # benign — must NOT match (was over-blocked under r2)
    for text in ("key=anurag", "key=anup", "key=anubis", "key=anuria",
                 "key=ANUX", "key=ANU123", "key=ANUPROD", "key='ANUPROD'"):
        assert not _any_marker(text), text
    # disguise — must STILL match (DENY kept)
    for text in ("key=ANU", "key=ANU_PROD", "key=ANU_DEV", "key=ANU_TEST",
                 "key=ANU_", "key=ANU-", "key=ANU-stg", "key='ANU_PROD'"):
        assert _any_marker(text), text


# ---------------------------------------------------------------------------
# r3-④ §4.4 — chaining collector create still DENY (over-block fix unrelated)
# ---------------------------------------------------------------------------
def test_r3_t4_chaining_collector_still_denied():
    from scripts.harness.v36.guard import evaluate

    cmd = (
        'cokacdir --cron-list ; cokacdir --cron "[ANU 수집 콜백] 결과 pickup" '
        "--key self"
    )
    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# r3-⑤ §4.5 — quote-aware general schedule ALLOW (unaffected by the fix)
# r3-⑥ §4.6 — real --key <hex> general schedule ALLOW (unaffected)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "회의 ; 정리 && 보고" --at 9am',                       # §4.5
        'cokacdir --cron "일정 정리" --at 1h --chat 1 --key deadbeefcafe0042',  # §4.6 real --key hex (fake placeholder)
    ],
)
def test_r3_t5_general_schedule_allows(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


# ---------------------------------------------------------------------------
# r3-⑦ §4.7 — r1/r2 separator/marker regression still DENY
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "[ANU_수집_콜백] x" --key self',
        'cokacdir --cron "[ANU-수집-콜백] x" --key self',
        'cokacdir --cron "collect; collector_role=ANU" --at 10m --key k',
    ],
)
def test_r3_t7_r1_r2_regression_still_denied(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# r3-⑧ §4.8 — no crash; benign ANU-prefixed key under chaining also ALLOW
# ---------------------------------------------------------------------------
def test_r3_t8_benign_anu_key_under_chaining_allows_no_crash():
    from scripts.harness.v36.guard import evaluate

    cmd = (
        'cokacdir --cron-list ; cokacdir --cron "pickup key=ANUPROD" '
        "--at 10m --chat 1 --key k"
    )
    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


# ===========================================================================
# task-2738-r4 — §2 segment-cokacdir predicate over-block fix
#
# Defect: `_segment_is_direct_collector_cron_create(seg)` judged a segment a
# collector cron-create from `--cron`(create) + marker ALONE, without checking
# that THIS segment actually invokes cokacdir.  The global `_R7_COKACDIR`
# pre-filter on the whole command therefore let a sibling cokacdir segment make
# a benign non-cokacdir text segment (e.g. `echo "--cron [ANU 수집 콜백]"`) look
# like a collector create → over-block.  Fix: require the cokacdir invocation in
# THE SAME SEGMENT (one predicate condition; no parser rewrite).
# ===========================================================================


# ---------------------------------------------------------------------------
# r4-① §4.1 / r4-② §4.2 — non-cokacdir text segment carrying the marker → ALLOW
# (the over-block being corrected: a sibling cokacdir management call no longer
#  contaminates a benign echo/text segment).
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        # §4.1 — separator chaining; second segment is plain echo text, not cokacdir
        'cokacdir --cron-list ; echo "--cron [ANU 수집 콜백]"',
        # §4.2 — && chaining; second segment is echo text
        'cokacdir --cron-list && echo "--cron collector_role=ANU"',
    ],
)
def test_r4_t1_non_cokacdir_marker_text_segment_allows(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


def test_r4_t1b_predicate_requires_cokacdir_in_same_segment():
    """Pin the fix at the predicate layer: a segment without a cokacdir
    invocation is never a direct collector cron-create, even though it carries
    both a `--cron` create form and a collector marker."""
    from scripts.harness.v36.rules import _segment_is_direct_collector_cron_create

    # benign text segments (no cokacdir in-segment) → False (was True under r3)
    assert not _segment_is_direct_collector_cron_create(
        'echo "--cron [ANU 수집 콜백]"'
    )
    assert not _segment_is_direct_collector_cron_create(
        'echo "--cron collector_role=ANU"'
    )
    # real in-segment cokacdir collector create → still True (DENY kept)
    assert _segment_is_direct_collector_cron_create(
        'cokacdir --cron "[ANU 수집 콜백]" --key self'
    )


# ---------------------------------------------------------------------------
# r4-③ §4.3 — a REAL standalone cokacdir collector create → DENY (kept)
# ---------------------------------------------------------------------------
def test_r4_t3_real_cokacdir_collector_create_still_denied():
    from scripts.harness.v36.guard import evaluate

    cmd = 'cokacdir --cron "[ANU 수집 콜백]" --at 30m --chat 1 --key x'
    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# r4-④ §4.4 — cokacdir management + REAL cokacdir collector create chaining → DENY
# (the in-segment predicate must not weaken genuine chaining detection)
# ---------------------------------------------------------------------------
def test_r4_t4_chaining_real_collector_create_still_denied():
    from scripts.harness.v36.guard import evaluate

    cmd = (
        'cokacdir --cron-list ; cokacdir --cron "[ANU 수집 콜백]" '
        "--at 30m --chat 1 --key x"
    )
    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# r4-⑤ §4.5 — key=ANU_PROD / _DEV / _TEST disguise → DENY (kept)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "schedule key=ANU_PROD" --at 10m --key k',
        'cokacdir --cron "schedule key=ANU_DEV" --at 10m --key k',
        'cokacdir --cron "schedule key=ANU_TEST" --at 10m --key k',
    ],
)
def test_r4_t5_key_anu_suffix_disguise_still_denied(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


# ---------------------------------------------------------------------------
# r4-⑥ §4.6 — benign ANU-prefixed keys → ALLOW (kept)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "schedule key=anurag" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=anup" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=anubis" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=ANUX" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=ANU123" --at 10m --chat 1 --key k',
        'cokacdir --cron "schedule key=ANUPROD" --at 10m --chat 1 --key k',
    ],
)
def test_r4_t6_benign_anu_prefixed_keys_still_allow(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


# ---------------------------------------------------------------------------
# r4-⑦ §4.7 — quote-aware general schedule + real --key hex → ALLOW (kept)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "회의 ; 정리 && 보고" --at 9am',
        'cokacdir --cron "일정 정리" --at 1h --chat 1 --key deadbeefcafe0042',
    ],
)
def test_r4_t7_general_schedule_and_real_key_still_allow(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] == "ALLOW", (cmd, r)
    assert r.get("matched_rule") is None


# ---------------------------------------------------------------------------
# r4-⑧ §4.8 — r1 separator regression DENY kept · no crash on the over-block
#               corrected forms (explicit ALLOW decision, never an exception)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
    "cmd",
    [
        'cokacdir --cron "[ANU_수집_콜백] x" --key self',
        'cokacdir --cron "[ANU-수집-콜백] x" --key self',
    ],
)
def test_r4_t8_r1_separator_regression_still_denied(cmd):
    from scripts.harness.v36.guard import evaluate

    r = evaluate("Bash", {"command": cmd}, {})
    assert r["decision"] in ("DENY", "HOLD_FOR_CHAIR"), (cmd, r)
    assert r["matched_rule"] == "pattern.executor_collector_callback_blocked"


def test_r4_t8b_corrected_forms_no_crash_explicit_allow():
    from scripts.harness.v36.guard import evaluate

    for cmd in (
        'cokacdir --cron-list ; echo "--cron [ANU 수집 콜백]"',
        'cokacdir --cron-list && echo "--cron collector_role=ANU"',
    ):
        r = evaluate("Bash", {"command": cmd}, {})
        assert r["decision"] == "ALLOW", (cmd, r)