# -*- coding: utf-8 -*- """tests/callback_authority_4source/test_4source_validator.py task-2680 — CALLBACK_SELF_KEY_REGISTRATION_HARDENING_FIX_IMPLEMENTATION chair_authorization_id: CHAIR-AUTH-CALLBACK-SELF-KEY-HARDENING-FIX-20260526-JJONGS-IMPLEMENT-001 Regression suite — 6 수정 목표 1:1: R1 task md ANU key 명시인데 실제 cron self-key 등록 방지 (전 layer) R2 helper --key 인자 cron layer 강제 R3 actual owner key 검증 callback collector gate R4 self-key callback NON_AUTHORITATIVE_SELF_COLLECTOR 자동 분류 R5 ANU independent reverify flow 강제 R6 4-source 교차 검증 doctrine (regression 재발 방지) Track A (91DDBCDA → 78F385CF, dev4 self-key 7943afbe12c12f7d) 및 Track J (A6200C2F → 33E60E8B, dev5 self-key 109fa85250c6d46b) 사고 모사. """ from __future__ import annotations import json import sys from pathlib import Path import pytest # Make repo root (worktree) importable. Insert at index 0 to override any # sibling workspace path the global tests/conftest.py may have prepended, # AND clear cached 'utils.*' / 'dispatch.*' modules so that the worktree # variant of the helper (with our task-2680 additions) loads rather than # any version cached by an earlier test collection from a different path. REPO_ROOT = Path(__file__).resolve().parents[2] _repo_str = str(REPO_ROOT) sys.path[:] = [_repo_str] + [p for p in sys.path if p != _repo_str] for _cached in [m for m in list(sys.modules) if m.startswith(("utils.", "dispatch."))]: sys.modules.pop(_cached, None) from utils.callback_authority_4source_validator import ( # noqa: E402 ANU_AUTHORITATIVE, ANU_KEY, AuthorityClassification, NON_AUTHORITATIVE_KEY_DRIFT, NON_AUTHORITATIVE_SELF_COLLECTOR, PROMPT_DRIFT, REVERIFY_TRIGGER_CLASSIFICATIONS, UNDETERMINED_HISTORY_GAP, VALIDATOR_SCHEMA, build_anu_independent_reverify_request, classify_collector_authority, classify_from_observed, is_self_key_callback, ) # ── fixtures ──────────────────────────────────────────────────────────────── DEV4_SELF_KEY = "7943afbe12c12f7d" # Track A executor key (verbatim from task-2677) DEV5_SELF_KEY = "109fa85250c6d46b" # Track J executor key (verbatim from task-2677) TRACK_A_SCHEDULE_ID = "78F385CF" TRACK_J_SCHEDULE_ID = "33E60E8B" DEV2_TASK_ID = "task-2680" def _make_runner(*, anu_count: int, self_count: int, anu_error: bool = False, self_error: bool = False): """Build a fake cokacdir subprocess runner that returns scripted counts for cron-history queries based on the --key value. Returns: callable(argv, timeout) → CompletedProcess-like object. """ class _Proc: def __init__(self, rc: int, stdout: str, stderr: str = ""): self.returncode = rc self.stdout = stdout self.stderr = stderr def runner(argv, _timeout): argv = list(argv) # find --key value try: key_idx = argv.index("--key") key_val = argv[key_idx + 1] except (ValueError, IndexError): return _Proc(1, "", "missing --key") if key_val == ANU_KEY: if anu_error: return _Proc(1, "", "simulated ANU query error") payload = {"status": "ok", "id": argv[2], "count": anu_count, "history": []} return _Proc(0, json.dumps(payload), "") # treat any other key as the "self" branch if self_error: return _Proc(1, "", "simulated self query error") payload = {"status": "ok", "id": argv[2], "count": self_count, "history": []} return _Proc(0, json.dumps(payload), "") return runner @pytest.fixture def isolated_dirs(tmp_path): """Provide isolated schedule_history + result artifact dirs.""" hist = tmp_path / "schedule_history" artifact = tmp_path / "events" hist.mkdir() artifact.mkdir() return hist, artifact def _write_schedule_history(hist_dir: Path, sid: str, prompt: str, bot_key_verifier: str = "fakehash", chat_id: int = 6937032012, workspace: str = "/home/jay/.cokacdir/workspace/X"): log_path = hist_dir / f"{sid}.log" record = { "ts": "2026-05-26T01:00:00+09:00", "schedule_id": sid, "chat_id": chat_id, "prompt": prompt, "status": "ok", "bot_key_verifier": bot_key_verifier, "workspace_path": workspace, "response": "", } log_path.write_text(json.dumps(record), encoding="utf-8") return log_path def _envelope_text(owner_key: str = ANU_KEY, self_key: str = "0", task_id: str = DEV2_TASK_ID) -> str: return json.dumps({ "schema": "anu.normal.callback.envelope.v1", "task_id": task_id, "owner_key": owner_key, "self_key": self_key, "callback_id": "test-callback", }, ensure_ascii=False, sort_keys=True) # ───────────────────────────────────────────────────────────────────────────── # R1: task md ANU key 명시인데 실제 cron self-key 등록 방지 # ───────────────────────────────────────────────────────────────────────────── def test_r1_track_a_self_key_detected_classification_self_collector(isolated_dirs): """Track A (91DDBCDA → 78F385CF, dev4 self-key) 사고 모사 — envelope text 가 ANU key 명기지만 actual cron --key=dev4 self-key 인 경우 즉시 NON_AUTHORITATIVE_SELF_COLLECTOR 로 분류한다.""" hist, artifact = isolated_dirs _write_schedule_history(hist, TRACK_A_SCHEDULE_ID, prompt=_envelope_text(owner_key=ANU_KEY)) runner = _make_runner(anu_count=0, self_count=1) # ANU 등록 0, self 등록 1 result = classify_collector_authority( schedule_id=TRACK_A_SCHEDULE_ID, executor_key=DEV4_SELF_KEY, task_id="task-2672", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == NON_AUTHORITATIVE_SELF_COLLECTOR assert is_self_key_callback(result) is True assert result.is_authoritative is False assert result.requires_anu_reverify is True def test_r1_track_j_self_key_detected(isolated_dirs): """Track J (A6200C2F → 33E60E8B, dev5 self-key) 동일 패턴.""" hist, artifact = isolated_dirs _write_schedule_history(hist, TRACK_J_SCHEDULE_ID, prompt=_envelope_text(owner_key=ANU_KEY)) runner = _make_runner(anu_count=0, self_count=1) result = classify_collector_authority( schedule_id=TRACK_J_SCHEDULE_ID, executor_key=DEV5_SELF_KEY, task_id="task-2676", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == NON_AUTHORITATIVE_SELF_COLLECTOR def test_r1_anu_authoritative_when_only_anu_registered(isolated_dirs): """Happy path: ANU key 등록 only → ANU_AUTHORITATIVE.""" hist, artifact = isolated_dirs _write_schedule_history(hist, "GOODCAFE", prompt=_envelope_text(owner_key=ANU_KEY)) runner = _make_runner(anu_count=1, self_count=0) result = classify_collector_authority( schedule_id="GOODCAFE", executor_key=DEV4_SELF_KEY, task_id="task-2680", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == ANU_AUTHORITATIVE assert result.is_authoritative is True assert result.requires_anu_reverify is False # ───────────────────────────────────────────────────────────────────────────── # R2: helper --key 인자 cron layer 강제 # ───────────────────────────────────────────────────────────────────────────── def test_r2_registrar_argv_builder_rejects_self_key(): """utils.anu_callback_registrar._build_cokacdir_cron_argv 가 anu_key 가 INDEPENDENT_ANU_KEY 가 아니면 SelfKeyForbidden 예외를 던진다.""" from utils.anu_callback_registrar import ( SelfKeyForbidden, _build_cokacdir_cron_argv, ) with pytest.raises(SelfKeyForbidden): _build_cokacdir_cron_argv( envelope={"task_id": "task-2680"}, delay_seconds=30, chat_id="6937032012", anu_key=DEV4_SELF_KEY, # ★ self-key — must raise cokacdir_path="/usr/local/bin/cokacdir", at_value=None, ) def test_r2_registrar_argv_builder_accepts_anu_key(): """ANU key 일 때만 정상 argv 생성.""" from utils.anu_callback_registrar import ( INDEPENDENT_ANU_KEY, _build_cokacdir_cron_argv, ) argv = _build_cokacdir_cron_argv( envelope={"task_id": "task-2680", "owner_key": INDEPENDENT_ANU_KEY}, delay_seconds=30, chat_id="6937032012", anu_key=INDEPENDENT_ANU_KEY, cokacdir_path="/usr/local/bin/cokacdir", at_value="30s", ) assert "--key" in argv idx = argv.index("--key") assert argv[idx + 1] == INDEPENDENT_ANU_KEY def test_r2_assert_cron_argv_uses_anu_key_strict(): """assert_cron_argv_uses_anu_key 가 --key 누락 / 잘못된 key 모두 차단.""" from utils.anu_callback_registrar import ( INDEPENDENT_ANU_KEY, SelfKeyForbidden, assert_cron_argv_uses_anu_key, ) # missing --key with pytest.raises(SelfKeyForbidden): assert_cron_argv_uses_anu_key(["cokacdir", "--cron", "x"]) # --key with self-key value with pytest.raises(SelfKeyForbidden): assert_cron_argv_uses_anu_key( ["cokacdir", "--cron", "x", "--key", DEV4_SELF_KEY] ) # PASS assert_cron_argv_uses_anu_key( ["cokacdir", "--cron", "x", "--key", INDEPENDENT_ANU_KEY] ) def _load_helper_module(): """Worktree 의 dispatch.normal_fallback_callback_helper 를 fresh import. pytest 의 다른 test 가 main 의 dispatch package 를 sys.modules 에 캐시 했을 수 있으므로, 우선 cache 를 비운 뒤 worktree path 를 강제 우선해 재import 한다. (importlib.util.spec_from_file_location 만으로는 sub- module ``from dispatch.callback_owner_enforcer import ...`` 가 동작하지 않으므로 dispatch package 자체를 worktree 로 재로딩해야 한다.) """ import importlib # Clear all dispatch / utils caches so that worktree variants load. for cached in [m for m in list(sys.modules) if m == "dispatch" or m.startswith("dispatch.")]: sys.modules.pop(cached, None) # Force worktree path to the very front. _repo_str = str(REPO_ROOT) sys.path[:] = [_repo_str] + [p for p in sys.path if p != _repo_str] return importlib.import_module("dispatch.normal_fallback_callback_helper") def test_r2_helper_assert_argv_uses_anu_key_returns_dict(): """dispatch.normal_fallback_callback_helper.assert_argv_uses_anu_key 가 dict (verdict + reasons) 형태 반환.""" helper = _load_helper_module() assert_argv_uses_anu_key = helper.assert_argv_uses_anu_key # ok path good_argv = ["cokacdir", "--cron", "x", "--key", ANU_KEY, "--once"] res = assert_argv_uses_anu_key(good_argv, owner_key=ANU_KEY) assert res["ok"] is True # fail path: missing --key res = assert_argv_uses_anu_key(["cokacdir", "--cron", "x"], owner_key=ANU_KEY) assert res["ok"] is False # fail path: argv --key value differs from owner_key res = assert_argv_uses_anu_key( ["cokacdir", "--cron", "x", "--key", DEV4_SELF_KEY], owner_key=ANU_KEY, ) assert res["ok"] is False # fail path: owner_key not an ANU key res = assert_argv_uses_anu_key( ["cokacdir", "--cron", "x", "--key", DEV4_SELF_KEY], owner_key=DEV4_SELF_KEY, ) assert res["ok"] is False def test_r2_build_anu_owned_callback_request_self_key_fails(): """build_anu_owned_callback_request 가 self-key owner_key 인 경우 FAIL + argv=None.""" helper = _load_helper_module() req = helper.build_anu_owned_callback_request( kind="normal", task_id="task-2680", executor_key=DEV4_SELF_KEY, owner_key=DEV4_SELF_KEY, # ★ self-key chat_id="6937032012", prompt="x", at="30s", cron_id="DEADBEEF", dispatch_cron_id="0123ABCD", no_fallback=True, ) assert req.verdict == helper.FAIL assert req.argv is None def test_r2_build_anu_owned_callback_request_anu_key_passes(): """ANU key owner_key 인 경우 PASS + argv 박제.""" helper = _load_helper_module() req = helper.build_anu_owned_callback_request( kind="normal", task_id="task-2680", executor_key=DEV4_SELF_KEY, owner_key=ANU_KEY, chat_id="6937032012", prompt="x", at="30s", cron_id="DEADBEEF", dispatch_cron_id="0123ABCD", no_fallback=True, ) assert req.verdict == helper.PASS, f"unexpected FAIL reasons: {req.reasons}" assert req.argv is not None assert "--key" in req.argv idx = req.argv.index("--key") assert req.argv[idx + 1] == ANU_KEY # ───────────────────────────────────────────────────────────────────────────── # R3: actual owner key 검증 callback collector gate # ───────────────────────────────────────────────────────────────────────────── def test_r3_collector_gate_calls_4source_validator(isolated_dirs): """utils.callback_collector_helper_integration.classify_collector_authority_4source 가 4-source validator 를 호출하고 결과를 SCHEMA wrapping 한다.""" from utils.callback_collector_helper_integration import ( classify_collector_authority_4source, NON_AUTHORITATIVE_SELF_COLLECTOR as INTEG_SELF_COLLECTOR, ) hist, artifact = isolated_dirs # self-key 사고 모사 — runner 만 inject (file-system path 는 validator # default 를 쓰므로 history 미설정 case 도 cls=SELF_COLLECTOR 가 나오는지) runner = _make_runner(anu_count=0, self_count=1) payload = classify_collector_authority_4source( schedule_id="78F385CF", executor_key=DEV4_SELF_KEY, task_id="task-2672", cokacdir_runner=runner, ) assert payload["classification"] == INTEG_SELF_COLLECTOR assert payload["verdict"] == "FAIL" assert payload["is_authoritative"] is False assert payload["requires_anu_reverify"] is True def test_r3_actual_owner_key_overrides_envelope_text(isolated_dirs): """envelope text 가 ANU 라고 명기되어도 actual cron owner 가 self-key 면 NON_AUTHORITATIVE_SELF_COLLECTOR — 텍스트 신뢰 0, actual 검증 우선.""" hist, artifact = isolated_dirs # envelope owner_key = ANU 텍스트 (correct intent) _write_schedule_history(hist, "TRACKAS1", prompt=_envelope_text(owner_key=ANU_KEY, self_key="0")) # 그러나 actual cron owner = self-key runner = _make_runner(anu_count=0, self_count=1) result = classify_collector_authority( schedule_id="TRACKAS1", executor_key=DEV4_SELF_KEY, task_id="task-2672", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == NON_AUTHORITATIVE_SELF_COLLECTOR # envelope text 정확 채록도 동시 evidence assert result.evidence is not None assert result.evidence.s3_envelope["owner_key"] == ANU_KEY # ───────────────────────────────────────────────────────────────────────────── # R4: self-key callback NON_AUTHORITATIVE_SELF_COLLECTOR 자동 분류 # ───────────────────────────────────────────────────────────────────────────── def test_r4_classify_from_observed_self_key(): """test-friendly fast-path: observed_owner_key == executor_key → NON_AUTHORITATIVE_SELF_COLLECTOR.""" cls = classify_from_observed( schedule_id="78F385CF", executor_key=DEV4_SELF_KEY, task_id="task-2672", observed_owner_key=DEV4_SELF_KEY, ) assert cls.classification == NON_AUTHORITATIVE_SELF_COLLECTOR def test_r4_classify_from_observed_anu(): cls = classify_from_observed( schedule_id="GOODCAFE", executor_key=DEV4_SELF_KEY, task_id="task-2680", observed_owner_key=ANU_KEY, ) assert cls.classification == ANU_AUTHORITATIVE def test_r4_classify_from_observed_unknown_3rd_key_drift(): cls = classify_from_observed( schedule_id="GOODCAFE", executor_key=DEV4_SELF_KEY, task_id="task-2680", observed_owner_key="deadbeefdeadbeef", # neither ANU nor self ) assert cls.classification == NON_AUTHORITATIVE_KEY_DRIFT def test_r4_classify_from_observed_none_undetermined(): cls = classify_from_observed( schedule_id="GOODCAFE", executor_key=DEV4_SELF_KEY, task_id="task-2680", observed_owner_key=None, ) assert cls.classification == UNDETERMINED_HISTORY_GAP def test_r4_classify_from_observed_envelope_prompt_drift(): cls = classify_from_observed( schedule_id="GOODCAFE", executor_key=DEV4_SELF_KEY, task_id="task-2680", observed_owner_key=ANU_KEY, envelope_owner_key="abc123def456", # envelope text drifted ) assert cls.classification == PROMPT_DRIFT def test_r4_reverify_trigger_membership(): """NON_AUTHORITATIVE_* / UNDETERMINED_* / PROMPT_DRIFT 모두 reverify trigger set 에 속한다. ANU_AUTHORITATIVE 만 제외.""" assert NON_AUTHORITATIVE_SELF_COLLECTOR in REVERIFY_TRIGGER_CLASSIFICATIONS assert NON_AUTHORITATIVE_KEY_DRIFT in REVERIFY_TRIGGER_CLASSIFICATIONS assert UNDETERMINED_HISTORY_GAP in REVERIFY_TRIGGER_CLASSIFICATIONS assert PROMPT_DRIFT in REVERIFY_TRIGGER_CLASSIFICATIONS assert ANU_AUTHORITATIVE not in REVERIFY_TRIGGER_CLASSIFICATIONS # ───────────────────────────────────────────────────────────────────────────── # R5: ANU independent reverify flow 강제 # ───────────────────────────────────────────────────────────────────────────── def test_r5_build_reverify_request_for_self_collector(): cls = AuthorityClassification( schema=VALIDATOR_SCHEMA, classification=NON_AUTHORITATIVE_SELF_COLLECTOR, schedule_id="78F385CF", task_id="task-2672", executor_key=DEV4_SELF_KEY, anu_key=ANU_KEY, is_authoritative=False, requires_anu_reverify=True, reasons=["self-key detected"], ) req = build_anu_independent_reverify_request( classification=cls, chair_authorization_id="CHAIR-AUTH-TEST-001", ) assert req is not None assert req.task_id == "task-2672" assert req.original_schedule_id == "78F385CF" assert req.classification == NON_AUTHORITATIVE_SELF_COLLECTOR assert req.chair_authorization_id == "CHAIR-AUTH-TEST-001" assert "independent_anu_reverify" in req.artifact_path_hint def test_r5_no_reverify_request_for_authoritative(): cls = AuthorityClassification( schema=VALIDATOR_SCHEMA, classification=ANU_AUTHORITATIVE, schedule_id="GOODCAFE", task_id="task-2680", executor_key=DEV4_SELF_KEY, anu_key=ANU_KEY, is_authoritative=True, requires_anu_reverify=False, reasons=["actual owner is ANU"], ) req = build_anu_independent_reverify_request( classification=cls, chair_authorization_id="CHAIR-AUTH-TEST-001", ) assert req is None def test_r5_emit_anu_independent_reverify_request_self_collector(isolated_dirs): """integration layer 의 emit_anu_independent_reverify_request 가 self-key classification 시 REVERIFY_REQUESTED action 을 반환.""" from utils.callback_collector_helper_integration import ( emit_anu_independent_reverify_request, ) runner = _make_runner(anu_count=0, self_count=1) payload = emit_anu_independent_reverify_request( schedule_id="78F385CF", executor_key=DEV4_SELF_KEY, task_id="task-2672", chair_authorization_id="CHAIR-AUTH-TEST-001", cokacdir_runner=runner, ) assert payload["action"] == "REVERIFY_REQUESTED" assert payload["reverify_request"] is not None assert payload["reverify_request"]["classification"] == NON_AUTHORITATIVE_SELF_COLLECTOR def test_r5_emit_no_reverify_for_authoritative(isolated_dirs): from utils.callback_collector_helper_integration import ( emit_anu_independent_reverify_request, ) runner = _make_runner(anu_count=1, self_count=0) payload = emit_anu_independent_reverify_request( schedule_id="GOODCAFE", executor_key=DEV4_SELF_KEY, task_id="task-2680", chair_authorization_id="CHAIR-AUTH-TEST-001", cokacdir_runner=runner, ) assert payload["action"] == "NO_REVERIFY_NEEDED" assert payload["reverify_request"] is None # ───────────────────────────────────────────────────────────────────────────── # R6: 4-source cross-check doctrine (regression — 사고 재발 방지) # ───────────────────────────────────────────────────────────────────────────── def test_r6_history_gap_undetermined(isolated_dirs): """schedule_history + cron-history 양쪽 다 빈 경우 UNDETERMINED_HISTORY_GAP.""" hist, artifact = isolated_dirs runner = _make_runner(anu_count=0, self_count=0) result = classify_collector_authority( schedule_id="MISSINGSCHED", executor_key=DEV4_SELF_KEY, task_id="task-2680", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == UNDETERMINED_HISTORY_GAP def test_r6_dual_owner_key_drift(isolated_dirs): """ANU + self 양쪽 다 등록된 경우 NON_AUTHORITATIVE_KEY_DRIFT (collision).""" hist, artifact = isolated_dirs _write_schedule_history(hist, "COLLIDED", prompt=_envelope_text(owner_key=ANU_KEY)) runner = _make_runner(anu_count=1, self_count=1) result = classify_collector_authority( schedule_id="COLLIDED", executor_key=DEV4_SELF_KEY, task_id="task-2680", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == NON_AUTHORITATIVE_KEY_DRIFT def test_r6_prompt_drift_envelope_text_not_anu(isolated_dirs): """ANU 등록 OK 인데 envelope text 의 owner_key 가 ANU 가 아니면 PROMPT_DRIFT.""" hist, artifact = isolated_dirs bad_envelope = _envelope_text(owner_key="abcd1234") # text drift _write_schedule_history(hist, "PROMPTDR", prompt=bad_envelope) runner = _make_runner(anu_count=1, self_count=0) result = classify_collector_authority( schedule_id="PROMPTDR", executor_key=DEV4_SELF_KEY, task_id="task-2680", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == PROMPT_DRIFT def test_r6_subprocess_query_error_undetermined(isolated_dirs): """cokacdir CLI 호출 실패 시 UNDETERMINED_HISTORY_GAP.""" hist, artifact = isolated_dirs runner = _make_runner(anu_count=0, self_count=0, anu_error=True, self_error=True) result = classify_collector_authority( schedule_id="ERR", executor_key=DEV4_SELF_KEY, task_id="task-2680", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == UNDETERMINED_HISTORY_GAP def test_r6_evidence_has_4_sources(isolated_dirs): """classification.evidence 가 4 source 채록을 모두 노출한다 (audit trail).""" hist, artifact = isolated_dirs sid = "AUDIT001" _write_schedule_history(hist, sid, prompt=_envelope_text(owner_key=ANU_KEY)) runner = _make_runner(anu_count=1, self_count=0) result = classify_collector_authority( schedule_id=sid, executor_key=DEV4_SELF_KEY, task_id="task-2680", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) ev = result.evidence assert ev is not None # S1 assert ev.s1_schedule_history is not None assert ev.s1_schedule_history.get("schedule_id") == sid # S2 assert ev.s2_anu_cron_history is not None assert ev.s2_anu_cron_history.get("count") == 1 assert ev.s2_self_cron_history is not None assert ev.s2_self_cron_history.get("count") == 0 # S3 assert ev.s3_envelope is not None assert ev.s3_envelope.get("owner_key") == ANU_KEY # S4 may be None (no artifact in fixture dir) — acceptable; field exists assert hasattr(ev, "s4_result_artifact") def test_r6_to_json_roundtrip(isolated_dirs): """AuthorityClassification.to_json() 결과가 JSON serializable + 핵심 필드 포함.""" hist, artifact = isolated_dirs sid = "JSON001" _write_schedule_history(hist, sid, prompt=_envelope_text(owner_key=ANU_KEY)) runner = _make_runner(anu_count=0, self_count=1) result = classify_collector_authority( schedule_id=sid, executor_key=DEV4_SELF_KEY, task_id="task-2680", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) js = result.to_json() # roundtrip assert json.loads(json.dumps(js, ensure_ascii=False)) == js assert js["classification"] == NON_AUTHORITATIVE_SELF_COLLECTOR assert js["evidence"] is not None assert js["escalation_hint"] is not None # ───────────────────────────────────────────────────────────────────────────── # Track A / Track J 사고 정확 재현 (regression anchor) # ───────────────────────────────────────────────────────────────────────────── def test_track_a_91DDBCDA_78F385CF_dev4_self_key_full_flow(isolated_dirs): """Track A: 91DDBCDA executor cron + 78F385CF callback (dev4 self-key 7943afbe12c12f7d) 사고 1:1 재현.""" hist, artifact = isolated_dirs # callback cron 78F385CF 의 schedule_history record (executor 가 등록) envelope = _envelope_text(owner_key=ANU_KEY) # 정확 명기 _write_schedule_history(hist, TRACK_A_SCHEDULE_ID, prompt=envelope, bot_key_verifier="2e90efbf4bd100fe2077ae6e383facfafc4b30d0e6e279ec376ff92d9604a038") # actual cron-history: ANU key count=0, dev4 self-key count=1 runner = _make_runner(anu_count=0, self_count=1) result = classify_collector_authority( schedule_id=TRACK_A_SCHEDULE_ID, executor_key=DEV4_SELF_KEY, task_id="task-2672", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == NON_AUTHORITATIVE_SELF_COLLECTOR assert result.requires_anu_reverify is True # reverify request build req = build_anu_independent_reverify_request( classification=result, chair_authorization_id=( "CHAIR-AUTH-CALLBACK-SELF-KEY-HARDENING-FIX-20260526-" "JJONGS-IMPLEMENT-001" ), ) assert req is not None assert req.task_id == "task-2672" def test_track_j_A6200C2F_33E60E8B_dev5_self_key_full_flow(isolated_dirs): """Track J: A6200C2F executor cron + 33E60E8B callback (dev5 self-key 109fa85250c6d46b) 사고 1:1 재현.""" hist, artifact = isolated_dirs envelope = _envelope_text(owner_key=ANU_KEY) _write_schedule_history(hist, TRACK_J_SCHEDULE_ID, prompt=envelope, bot_key_verifier="bdd7d2cbecf77adf195ea4b16a41d53f1d81660834353f27fd306c1a407f8f69") runner = _make_runner(anu_count=0, self_count=1) result = classify_collector_authority( schedule_id=TRACK_J_SCHEDULE_ID, executor_key=DEV5_SELF_KEY, task_id="task-2676", cokacdir_runner=runner, history_dir=hist, artifact_dir=artifact, ) assert result.classification == NON_AUTHORITATIVE_SELF_COLLECTOR if __name__ == "__main__": sys.exit(pytest.main([__file__, "-v"]))