# -*- coding: utf-8 -*-
"""task-2729+24 ENVELOPE_PROMPT_OWNER_KEY_HYGIENE — 필수 regression (7).

발번: canary preflight 에서 ``_build_envelope_prompt`` 가 wake prompt 에
``owner_key={anu_key}`` raw ANU key literal 줄을 박아 cokacdir cron body →
schedule_history/system_prompt 로 적재(content-driven)됨을 발견. 회장 2026-06-09
봇 위임 승인 하에 해당 1줄만 surgical 제거한다(production activation 아님).

owner proof 는 envelope 텍스트가 아니라 ``cokacdir --cron-history <sid> --key
<ANU_KEY>`` 실조회(``resolve_authoritative_owner`` / ``OwnerProbe``)로 판정하므로
owner_key 줄은 redundant — 제거해도 owner proof 무손상이다.

regression
  1. ``_build_envelope_prompt`` 출력에 raw ANU key / ``owner_key=`` /
     fingerprint / key-derived id 전부 부재.
  2. 나머지 envelope 필드(task_id/result_path/report_path/sha256/collector_role/
     chat_id/canonical_root) 유지.
  3. owner proof 경로(``resolve_authoritative_owner``, fake cron-history probe
     주입) 무영향 PASS.
  4. ``anu_key`` 시그니처/호출부 호환(호출 시 에러 0).
  5. ``build_anu_owned_callback_request`` argv **prompt body** raw key 0.
  6. 소스/diff raw key 0 (envelope prompt builder).
  7. ACTIVE=false / systemd enable 0 / activation_epoch absent / real spawn 0.
"""
from __future__ import annotations

import importlib.util
import inspect
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path

_ROOT = Path(__file__).resolve().parent.parent.parent
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))


def _load(modname: str, relpath: str):
    if modname in sys.modules:
        return sys.modules[modname]
    spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath)
    assert spec is not None and spec.loader is not None
    mod = importlib.util.module_from_spec(spec)
    sys.modules[modname] = mod
    spec.loader.exec_module(mod)
    return mod


# 실 runtime entrypoint 직접 호출 (재사용 의존 enforcer/helper 도 실모듈 로드).
_load("dispatch.callback_owner_enforcer",
      "dispatch/callback_owner_enforcer.py")
_load("dispatch.normal_fallback_callback_helper",
      "dispatch/normal_fallback_callback_helper.py")
M = _load("dispatch.anu_owned_callback_enforcement",
          "dispatch/anu_owned_callback_enforcement.py")

ANU_KEY = M.ANU_KEY                       # 독립 ANU key (import 상수 — literal 비노출)
DEV4_SELF_KEY = "7943afbe12c12f7d"        # incident: dev4 self-collector key
ANU_SCHEDULE = "ANUOWNED1"                # ANU 가 소유한 schedule (조회 성공)

# envelope 필드 검증용 result (raw key 값과 무관한 합성 데이터).
_RESULT = {
    "task_id": "task-2729+24",
    "result_json_path": "/home/jay/workspace/memory/results/task-2729+24.json",
    "report_path": "/home/jay/workspace/memory/reports/task-2729+24.md",
    "sha256": "deadbeef" * 8,
}


def make_probe(owned_schedules):
    """fake cron-history probe — 실 cokacdir json 형태 재현(subprocess 0)."""
    owned = set(owned_schedules)

    def _probe(schedule_id: str) -> dict:
        if schedule_id in owned:
            return {
                "status": "ok",
                "id": schedule_id,
                "count": 1,
                "history": [{"status": "ok", "ts": "2026-06-09 01:10:00"}],
            }
        return {
            "status": "error",
            "message": f"schedule not found or access denied: {schedule_id}",
        }

    return _probe


class Regression2729p24(unittest.TestCase):

    # ── regression 1 — envelope prompt 에 raw key / owner_key= / fp 부재 ───────
    def test_1_envelope_prompt_no_owner_key_no_raw_key(self):
        prompt = M._build_envelope_prompt(
            _RESULT, anu_key=ANU_KEY, chat_id=M.ANU_CHAT_ID
        )
        # raw ANU key 값이 출력에 전혀 없어야 한다.
        self.assertNotIn(ANU_KEY, prompt)
        # owner_key= 줄 자체 부재.
        self.assertNotIn("owner_key=", prompt)
        self.assertNotIn("owner_key", prompt)
        # fingerprint / key-derived id 류 필드 추가 금지 — 부재 확인.
        for forbidden in ("owner_key_fp", "fingerprint", "key_fp", "key_sha",
                          "owner_id", "self_key_used"):
            self.assertNotIn(forbidden, prompt)

    # ── regression 2 — 나머지 envelope 필드 유지 ─────────────────────────────
    def test_2_remaining_envelope_fields_preserved(self):
        prompt = M._build_envelope_prompt(
            _RESULT, anu_key=ANU_KEY, chat_id=M.ANU_CHAT_ID
        )
        self.assertIn(f"task_id={_RESULT['task_id']}", prompt)
        self.assertIn(f"result_path={_RESULT['result_json_path']}", prompt)
        self.assertIn(f"report_path={_RESULT['report_path']}", prompt)
        self.assertIn(f"sha256={_RESULT['sha256']}", prompt)
        self.assertIn("collector_role=ANU", prompt)
        self.assertIn(f"chat_id={M.ANU_CHAT_ID}", prompt)
        self.assertIn(f"canonical_root={M.CANONICAL_ROOT}", prompt)
        # owner_key 줄이 빠지면서 정확히 7개 envelope 라인만 남는다.
        lines = [ln for ln in prompt.splitlines() if ln.strip()]
        self.assertEqual(len(lines), 7)
        keys = {ln.split("=", 1)[0] for ln in lines}
        self.assertEqual(
            keys,
            {"task_id", "result_path", "report_path", "sha256",
             "collector_role", "chat_id", "canonical_root"},
        )

    # ── regression 3 — owner proof 경로 무영향 PASS ──────────────────────────
    def test_3_owner_proof_path_unaffected(self):
        # owner proof 는 envelope 텍스트가 아니라 cron-history 실조회로 판정.
        probe = make_probe({ANU_SCHEDULE})
        res_owned = M.resolve_authoritative_owner(ANU_SCHEDULE, probe=probe)
        self.assertTrue(res_owned.owner_is_anu)
        self.assertEqual(res_owned.outcome, M.OWNER_ANU)
        # 비-ANU schedule 은 access denied → 비-ANU 판정.
        res_denied = M.resolve_authoritative_owner("A372DD6B", probe=probe)
        self.assertFalse(res_denied.owner_is_anu)
        self.assertEqual(res_denied.outcome, M.OWNER_NOT_ANU)
        # resolve_authoritative_owner 시그니처는 envelope/owner_key 텍스트를
        # 입력으로 받지 않는다 — owner proof 가 prompt 와 무관함을 명시 증명.
        params = set(inspect.signature(M.resolve_authoritative_owner).parameters)
        self.assertNotIn("envelope", params)
        self.assertNotIn("owner_key", params)
        self.assertNotIn("prompt", params)

    # ── regression 4 — anu_key 시그니처/호출부 호환 ──────────────────────────
    def test_4_anu_key_signature_callsite_compat(self):
        sig = inspect.signature(M._build_envelope_prompt)
        self.assertIn("anu_key", sig.parameters)
        self.assertIn("chat_id", sig.parameters)
        # keyword-only 호출(호출부와 동일 형태)로 에러 0 확인.
        out = M._build_envelope_prompt(
            _RESULT, anu_key=ANU_KEY, chat_id=M.ANU_CHAT_ID
        )
        self.assertIsInstance(out, str)
        self.assertTrue(out)

    # ── regression 5 — build request argv prompt body raw key 0 ──────────────
    def test_5_callback_request_argv_prompt_body_no_raw_key(self):
        with tempfile.TemporaryDirectory(prefix="task2729p24-") as tmp:
            rj = os.path.join(tmp, "result.json")
            with open(rj, "w", encoding="utf-8") as fh:
                json.dump(_RESULT, fh)
            runner = M.anu_runner_pickup_and_fire(
                result_json_path=rj,
                executor_key=DEV4_SELF_KEY,   # self key (ANU 와 다름)
                anu_key=ANU_KEY,
            )
        self.assertEqual(runner.verdict, "PASS")
        self.assertIsNotNone(runner.argv)
        argv = list(runner.argv)
        # argv 의 prompt body(--cron 다음 인자)에는 raw key 가 없어야 한다.
        prompt_body = argv[argv.index("--cron") + 1]
        self.assertNotIn(ANU_KEY, prompt_body)
        self.assertNotIn("owner_key", prompt_body)

    # ── regression 6 — 소스(envelope builder) raw key literal 0 ──────────────
    def test_6_source_envelope_builder_no_raw_key_literal(self):
        src = inspect.getsource(M._build_envelope_prompt)
        # f"owner_key={anu_key}" 줄이 소스에서 제거됐는지(=raw key 노출 패턴 0).
        self.assertNotIn('f"owner_key=', src)
        self.assertNotIn("owner_key=", src.replace("# ", ""))
        # 실 raw key literal 도 소스에 박혀 있지 않아야 한다.
        self.assertNotIn(ANU_KEY, src)

    # ── regression 7 — ACTIVE=false / activation_epoch absent / spawn 0 ──────
    def test_7_no_activation_no_real_spawn(self):
        with tempfile.TemporaryDirectory(prefix="task2729p24-fire-") as tmp:
            rj = os.path.join(tmp, "result.json")
            with open(rj, "w", encoding="utf-8") as fh:
                json.dump(_RESULT, fh)
            runner = M.anu_runner_pickup_and_fire(
                result_json_path=rj,
                executor_key=DEV4_SELF_KEY,
                anu_key=ANU_KEY,
            )
            fired = M.fire_callback_request(runner, activate=False)
        # Phase 1: 실 fire 0 — 항상 dry-run, fired False.
        self.assertFalse(fired["fired"])
        self.assertTrue(fired["dry_run"])
        self.assertEqual(fired["status"], M.FIRE_NOT_ACTIVATED)
        # activation_epoch / ACTIVE=true / systemd 류 신호가 패치에 의해
        # 도입되지 않았음을 envelope builder 소스로 확인.
        src = inspect.getsource(M._build_envelope_prompt)
        for forbidden in ("activation_epoch", "ACTIVE=true", "systemctl",
                          "subprocess", "Popen", "os.system"):
            self.assertNotIn(forbidden, src)


if __name__ == "__main__":
    unittest.main()
