#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""scripts/diag_team_lead_subagent_contract_2553plus56.py

task-2553+56 TRACK A — TEAM_LEAD_SUBAGENT_USAGE_CONTRACT_DIAGNOSIS.

READ-ONLY static diagnostic. Imports the REAL runtime modules and
introspects whether the team-lead / subagent external boundary
(result schema · evidence-vs-authoritative · proposed verdict ·
independent-ANU-only authoritative · subagent direct ANU guard ·
team-lead normal callback owner · expected/forbidden/regression/
unresolved recording · concealment WARN/HOLD) is enforced by
code/schema or only by free-text prompt guidance.

Zero side effects: no write, no cron, no dispatch, no subprocess,
no cokacdir. The function ``run_diagnosis()`` returns a structured
dict; ``main()`` prints it as JSON to stdout. Artifact files are
written by the orchestrating executor, never by this module — so the
script stays a pure read-only probe (회장 §3 정적 점검).

Verdict vocabulary per check:
  PASS    — boundary is enforced by code/schema (fail-closed).
  PARTIAL — enforced transitively (via self-chain / self-key guards)
            but no explicit subagent-scoped construct.
  GAP     — only free-text prompt guidance; no code/schema enforcement.
"""
from __future__ import annotations

import inspect
import json
import os
import sys
from typing import Any, Dict, List

# Resolve anu_v3 / dispatch / utils from the repo root regardless of cwd
# (read-only path bootstrap; no write).
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _REPO_ROOT not in sys.path:
    sys.path.insert(0, _REPO_ROOT)


def _hermetic_load(modname: str, relpath: str):
    """Hermetic file-path import — collision-proof vs the ``dispatch.py``
    shim that otherwise shadows the real ``dispatch/`` package under
    pytest (identical strategy to the +49 regression suites). Read-only:
    loads the genuine workspace module, no write."""
    import importlib.util

    if modname in sys.modules:
        return sys.modules[modname]
    spec = importlib.util.spec_from_file_location(
        modname, os.path.join(_REPO_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


# Pre-seed the real (non-mock) dispatch guard so the anu_v3 runtime
# validators import the genuine entrypoint even when the dispatch.py
# shim is already registered as ``sys.modules['dispatch']``.
_hermetic_load(
    "dispatch.callback_owner_enforcer",
    "dispatch/callback_owner_enforcer.py",
)

SCHEMA = "diag.team_lead_subagent_contract.v1"
TASK_ID = "task-2553+56"
TRACK = "A"

PASS = "PASS"
PARTIAL = "PARTIAL"
GAP = "GAP"

# Fields the team-result packet would need to satisfy check 1 / check 7.
SUBAGENT_RESULT_FIELDS = (
    "subagents_used",
    "subagent_roles",
    "subagent_outputs_summary",
)
BOUNDARY_RECORD_FIELDS = (
    "expected_files",
    "forbidden_files",
    "regression",
    "unresolved_findings",
)


def _packet_keys() -> List[str]:
    """Build a representative team-result packet from the REAL builder and
    return its top-level keys (read-only introspection)."""
    from anu_v3.goal_result_planner import build_final_packet

    pkt = build_final_packet(
        ts="1970-01-01T00:00:00Z",
        task_id=TASK_ID,
        goal_contract={"goal": "diag"},
        derived_contract=None,
        gate_decision=None,
        activation_decision={},
        activation_result={},
        callback_contract={},
        adjudication={},
        codex_audit_result={},
        hold_packet=None,
    )
    return sorted(pkt.keys())


def _check1_result_schema(keys: List[str]) -> Dict[str, Any]:
    present = [f for f in SUBAGENT_RESULT_FIELDS if f in keys]
    missing = [f for f in SUBAGENT_RESULT_FIELDS if f not in keys]
    verdict = PASS if not missing else GAP
    return {
        "id": 1,
        "question": (
            "team result에 subagents_used / subagent_roles / "
            "subagent_outputs_summary 필드가 있는가"
        ),
        "verdict": verdict,
        "present_fields": present,
        "missing_fields": missing,
        "entrypoint": "anu_v3.goal_result_planner.build_final_packet",
        "evidence": (
            "16-field final packet has NO subagent-result fields; team "
            "prompt only carries a free-text '서브에이전트(Task tool) 결과 "
            "규칙' (≤500자 요약) — not a code/schema-enforced structured "
            "field set."
        ),
    }


def _check2_evidence_not_authoritative() -> Dict[str, Any]:
    from anu_v3.authoritative_verdict_selector import (
        AUTHORITATIVE_VERDICT_PENDING,
        FAIL,
        VerdictRecord,
        select_authoritative_verdict,
    )

    # A subagent's output reaches ANU through the team-lead (executor)
    # self-session -> session_is_executor_self=True -> self_chain.
    rec = VerdictRecord(
        kind="collector_result",
        verdict="PASS",
        task_id=TASK_ID,
        executor_key="EXEC",
        collector_key="EXEC",
        collector_role="ANU",
        session_is_executor_self=True,
        claimed_origin="independent_anu",
    )
    res = select_authoritative_verdict(
        [rec], task_id=TASK_ID, anu_keys=["ANUKEY"]
    )
    quarantined = (
        res.quarantined_count == 1
        and res.independent_anu_count == 0
        and res.classification == AUTHORITATIVE_VERDICT_PENDING
        and res.verdict == FAIL
    )
    verdict = PARTIAL if quarantined else GAP
    return {
        "id": 2,
        "question": "subagent 산출은 evidence이고 authoritative verdict가 아닌가",
        "verdict": verdict,
        "self_chain_quarantined": quarantined,
        "entrypoint": (
            "anu_v3.authoritative_verdict_selector.select_authoritative_verdict"
        ),
        "evidence": (
            "subagent output flows through the team-lead executor "
            "self-session, so derived_origin -> self_chain -> QUARANTINED "
            "and never authoritative. Enforced transitively via the "
            "self-chain rule; there is NO explicit subagent evidence kind "
            "or 'evidence-only' tag."
        ),
    }


def _check3_team_lead_verdict_proposed(keys: List[str]) -> Dict[str, Any]:
    from anu_v3.authoritative_verdict_selector import (
        FAIL,
        VerdictRecord,
        select_authoritative_verdict,
    )

    rec = VerdictRecord(
        kind="collector_result",
        verdict="PASS",
        task_id=TASK_ID,
        executor_key="LEAD",
        collector_key="LEAD",
        collector_role="ANU",
        session_is_executor_self=True,
    )
    res = select_authoritative_verdict(
        [rec], task_id=TASK_ID, anu_keys=["ANUKEY"]
    )
    functionally_non_auth = res.verdict == FAIL and res.authoritative_verdict is None
    has_proposed_field = "proposed_verdict" in keys
    verdict = PASS if has_proposed_field else PARTIAL
    return {
        "id": 3,
        "question": "team lead verdict는 proposed verdict인가",
        "verdict": verdict,
        "functionally_non_authoritative": functionally_non_auth,
        "explicit_proposed_verdict_field": has_proposed_field,
        "entrypoint": (
            "anu_v3.authoritative_verdict_selector.select_authoritative_verdict"
        ),
        "evidence": (
            "team-lead (executor self-session) verdict is structurally "
            "non-authoritative (self_chain). But there is NO explicit "
            "'proposed_verdict' field in the team-result schema and no "
            "code that labels/records the team-lead verdict as 'proposed' "
            "— it is only implied by quarantine."
        ),
    }


def _check4_anu_only_authoritative() -> Dict[str, Any]:
    from anu_v3.authoritative_verdict_selector import (
        AUTHORITATIVE_PASS,
        PASS as A_PASS,
        VerdictRecord,
        select_authoritative_verdict,
    )

    self_rec = VerdictRecord(
        kind="collector_result",
        verdict="PASS",
        task_id=TASK_ID,
        executor_key="EXEC",
        collector_key="EXEC",
        collector_role="ANU",
        session_is_executor_self=True,
    )
    anu_rec = VerdictRecord(
        kind="collector_result",
        verdict="PASS",
        task_id=TASK_ID,
        executor_key="EXEC",
        collector_key="ANUKEY",
        collector_role="ANU",
        session_is_executor_self=False,
    )
    only_self = select_authoritative_verdict(
        [self_rec], task_id=TASK_ID, anu_keys=["ANUKEY"]
    )
    with_anu = select_authoritative_verdict(
        [self_rec, anu_rec], task_id=TASK_ID, anu_keys=["ANUKEY"]
    )
    enforced = (
        only_self.authoritative_verdict is None
        and with_anu.classification == AUTHORITATIVE_PASS
        and with_anu.verdict == A_PASS
        and with_anu.authoritative_source_kind == "collector_result"
    )
    return {
        "id": 4,
        "question": "ANU independent collector만 authoritative인가",
        "verdict": PASS if enforced else GAP,
        "code_enforced": enforced,
        "entrypoint": (
            "anu_v3.authoritative_verdict_selector.select_authoritative_verdict"
        ),
        "evidence": (
            "self-chain-only -> authoritative_verdict=None (FAIL/PENDING); "
            "adding an independent-ANU record promotes it to "
            "AUTHORITATIVE_PASS. derived_origin recomputes independence "
            "from owner identity (claimed_origin ignored) — fail-closed."
        ),
    }


def _check5_subagent_direct_anu_guard() -> Dict[str, Any]:
    from anu_v3.callback_owner_validator import (
        CallbackRegistrationBlocked,
        assert_registration_permitted,
        validate_callback_owner_runtime,
    )
    from anu_v3.self_collector_guard import (
        guard_self_collector_session,
        guard_self_dispatch,
    )
    from utils.delegate_controller import BLOCKED_TOOLS

    # A subagent shares the team-lead/executor key channel. A subagent
    # that tries to own the completion collector / self-dispatch is caught
    # by the executor-self-key guards.
    coll = guard_self_collector_session(
        executor_key="LEAD", collector_key="LEAD", collector_role="ANU"
    )
    disp = guard_self_dispatch(
        executor_key="LEAD", actor_key="LEAD", is_followup_dispatch=True
    )
    val = validate_callback_owner_runtime(
        task_id=TASK_ID,
        executor_key="LEAD",
        collector_key="LEAD",
        collector_role="ANU",
        normal_collector_cron_id="c1",
        fallback_callback_cron_id="c2",
        dispatch_cron_id="c0",
    )
    blocked = False
    try:
        assert_registration_permitted(val)
    except CallbackRegistrationBlocked:
        blocked = True

    self_key_guarded = (
        not coll.ok and not disp.ok and blocked
    )
    # delegate_controller blocks a narrow toolset on the SDK delegate path
    # only — NOT cron/cokacdir/ANU on the Task-tool subagent path.
    delegate_blocked = sorted(BLOCKED_TOOLS)
    explicit_subagent_write_guard = False  # none found in code
    verdict = PARTIAL if self_key_guarded else GAP
    return {
        "id": 5,
        "question": (
            "subagent가 ANU로 직접 callback/dispatch/write하지 못하도록 "
            "guard가 있는가"
        ),
        "verdict": verdict,
        "executor_self_key_guard_blocks_subagent_channel": self_key_guarded,
        "delegate_controller_blocked_tools": delegate_blocked,
        "explicit_subagent_scoped_write_guard": explicit_subagent_write_guard,
        "entrypoint": (
            "anu_v3.callback_owner_validator.validate_callback_owner_runtime "
            "+ anu_v3.self_collector_guard.guard_self_dispatch / "
            "utils.delegate_controller.BLOCKED_TOOLS"
        ),
        "evidence": (
            "A subagent uses the team-lead/executor key channel, so a "
            "direct ANU callback/self-dispatch is fail-closed via the "
            "executor-self-key guards (transitive). delegate_controller "
            "blocks {delegate,clarify,memory,send_message} on the SDK "
            "delegate path only; there is NO explicit subagent-scoped "
            "guard for direct cokacdir/ANU write on the Task-tool path."
        ),
    }


def _check6_team_lead_callback_owner_anu() -> Dict[str, Any]:
    from anu_v3.callback_owner_validator import (
        CallbackRegistrationBlocked,
        assert_registration_permitted,
        validate_callback_owner_runtime,
    )

    self_val = validate_callback_owner_runtime(
        task_id=TASK_ID,
        executor_key="c38fb9955616e24d",
        collector_key="c38fb9955616e24d",
        collector_role="ANU",
        normal_collector_cron_id="c1",
        fallback_callback_cron_id="c2",
        dispatch_cron_id="c0",
    )
    self_blocked = False
    try:
        assert_registration_permitted(self_val)
    except CallbackRegistrationBlocked:
        self_blocked = True

    anu_val = validate_callback_owner_runtime(
        task_id=TASK_ID,
        executor_key="c38fb9955616e24d",
        collector_key="c119085addb0f8b7",
        collector_role="ANU",
        normal_collector_cron_id="c1",
        fallback_callback_cron_id="c2",
        dispatch_cron_id="c0",
    )
    anu_ok = anu_val.ok and anu_val.registration_allowed
    enforced = self_blocked and anu_ok
    return {
        "id": 6,
        "question": "팀장 normal callback은 ANU key로 가는가",
        "verdict": PASS if enforced else GAP,
        "self_key_registration_blocked": self_blocked,
        "anu_key_registration_allowed": anu_ok,
        "entrypoint": (
            "anu_v3.callback_owner_validator.assert_registration_permitted"
        ),
        "evidence": (
            "executor self-key collector -> CallbackRegistrationBlocked "
            "(fail-closed, +49 코드 정본). Independent ANU-key collector "
            "-> registration allowed. Team-lead normal callback owner = "
            "ANU key is code-enforced."
        ),
    }


def _check7_boundary_record_fields(keys: List[str]) -> Dict[str, Any]:
    present = [f for f in BOUNDARY_RECORD_FIELDS if f in keys]
    # regression is represented as 'regression_result' in the packet.
    has_regression = "regression_result" in keys or "regression" in keys
    missing = [
        f
        for f in BOUNDARY_RECORD_FIELDS
        if f not in keys and not (f == "regression" and has_regression)
    ]
    verdict = PASS if not missing else GAP
    return {
        "id": 7,
        "question": (
            "expected_files / forbidden_files / regression / unresolved "
            "findings가 team result에 기록되는가"
        ),
        "verdict": verdict,
        "present": present + (["regression_result"] if has_regression else []),
        "missing": missing,
        "entrypoint": "anu_v3.goal_result_planner.build_final_packet",
        "evidence": (
            "Only 'regression_result' exists in the 16-field packet. "
            "expected_files / forbidden_files / unresolved_findings are "
            "NOT recorded in the team-result schema by code."
        ),
    }


def _check8_concealment_warn_hold() -> Dict[str, Any]:
    from anu_v3 import authoritative_verdict_selector as avs
    from dispatch import callback_owner_enforcer as coe

    src = inspect.getsource(avs) + inspect.getsource(coe)
    tokens = (
        "SUBAGENT_RESULT_CONCEALED",
        "subagent_concealed",
        "subagent_omitted",
        "hidden_subagent",
    )
    has_classifier = any(t in src for t in tokens)
    verdict = PASS if has_classifier else GAP
    return {
        "id": 8,
        "question": (
            "팀장이 subagent 결과를 숨기거나 누락했을 때 WARN/HOLD로 "
            "분류 가능한가"
        ),
        "verdict": verdict,
        "concealment_classifier_present": has_classifier,
        "entrypoint": (
            "anu_v3.authoritative_verdict_selector / "
            "dispatch.callback_owner_enforcer (source scan)"
        ),
        "evidence": (
            "No SUBAGENT_RESULT_CONCEALED / concealment classification "
            "exists in any guard or verdict module. A team-lead omitting "
            "or hiding subagent results is NOT detectable -> cannot be "
            "auto-classified WARN/HOLD by code."
        ),
    }


def run_diagnosis() -> Dict[str, Any]:
    """Execute all 8 read-only checks against REAL entrypoints."""
    keys = _packet_keys()
    checks = [
        _check1_result_schema(keys),
        _check2_evidence_not_authoritative(),
        _check3_team_lead_verdict_proposed(keys),
        _check4_anu_only_authoritative(),
        _check5_subagent_direct_anu_guard(),
        _check6_team_lead_callback_owner_anu(),
        _check7_boundary_record_fields(keys),
        _check8_concealment_warn_hold(),
    ]
    counts = {PASS: 0, PARTIAL: 0, GAP: 0}
    for c in checks:
        counts[c["verdict"]] = counts.get(c["verdict"], 0) + 1

    # External authoritative/callback boundary (checks 4 & 6) MUST be PASS;
    # subagent-specific structured contract (1,7,8) are the GAPs.
    boundary_ok = checks[3]["verdict"] == PASS and checks[5]["verdict"] == PASS
    overall = (
        "BOUNDARY_ENFORCED_SUBAGENT_CONTRACT_GAPS"
        if boundary_ok
        else "BOUNDARY_NOT_ENFORCED"
    )
    return {
        "schema": SCHEMA,
        "task_id": TASK_ID,
        "track": TRACK,
        "mode": "read_only_diagnosis",
        "overall": overall,
        "authoritative_callback_boundary_code_enforced": boundary_ok,
        "counts": counts,
        "checks": checks,
        "note": (
            "Read-only. External boundary (ANU-only authoritative + "
            "team-lead callback owner=ANU) is code-enforced fail-closed. "
            "Subagent-specific structured contract (result schema fields, "
            "expected/forbidden/unresolved recording, concealment "
            "WARN/HOLD) is prompt-only — GAP."
        ),
    }


def main(argv: List[str]) -> int:
    _ = argv
    print(json.dumps(run_diagnosis(), ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))