# -*- coding: utf-8 -*- """tests/regression/test_team_lead_subagent_contract_2553plus56.py task-2553+56 TRACK A regression — 8 confirmation items verified against the REAL runtime entrypoints (mock-only = FAIL). Hard mock-only gate: this suite imports and exercises the genuine anu_v3 / dispatch / utils modules through scripts.diag_team_lead_subagent_contract_2553plus56.run_diagnosis(). If those real modules are unavailable / mocked, collection fails and the suite FAILS (no mock fallback path exists). """ from __future__ import annotations import importlib import importlib.util import os import sys import pytest _REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) if _REPO_ROOT not in sys.path: sys.path.insert(0, _REPO_ROOT) # Load the diagnostic FIRST by file path (scripts/ is not a package): its # module-level hermetic pre-seed loads the genuine # dispatch.callback_owner_enforcer, defeating the dispatch.py shim that # otherwise shadows the real package under pytest. def _load_by_path(modname: str, relpath: str): 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 _DIAG = _load_by_path( "diag_team_lead_subagent_contract_2553plus56", "scripts/diag_team_lead_subagent_contract_2553plus56.py", ) def _require_real(modname: str): """Import a real module; FAIL (not skip) if it cannot be resolved.""" try: return importlib.import_module(modname) except Exception as exc: # noqa: BLE001 pytest.fail( f"real entrypoint {modname!r} unavailable ({exc!r}); " "mock-only execution is FAIL per task-2553+56 §3." ) @pytest.fixture(scope="module") def diag(): # Real modules MUST resolve — these are the genuine runtime guards. _require_real("anu_v3.goal_result_planner") _require_real("anu_v3.authoritative_verdict_selector") _require_real("anu_v3.callback_owner_validator") _require_real("anu_v3.self_collector_guard") _require_real("dispatch.callback_owner_enforcer") _require_real("utils.delegate_controller") result = _DIAG.run_diagnosis() assert result["mode"] == "read_only_diagnosis" return result def _check(diag, cid: int): for c in diag["checks"]: if c["id"] == cid: return c raise AssertionError(f"check {cid} missing") def test_diag_runs_against_real_modules(diag): assert diag["schema"] == "diag.team_lead_subagent_contract.v1" assert diag["task_id"] == "task-2553+56" assert len(diag["checks"]) == 8 assert {c["id"] for c in diag["checks"]} == set(range(1, 9)) def test_check1_subagent_result_fields_gap(diag): c = _check(diag, 1) # Real 16-field packet has none of the subagent-result fields. assert c["verdict"] == "GAP" assert c["present_fields"] == [] assert set(c["missing_fields"]) == { "subagents_used", "subagent_roles", "subagent_outputs_summary", } def test_check2_subagent_output_is_evidence_not_authoritative(diag): c = _check(diag, 2) assert c["verdict"] == "PARTIAL" assert c["self_chain_quarantined"] is True def test_check3_team_lead_verdict_proposed_only_implicit(diag): c = _check(diag, 3) assert c["verdict"] == "PARTIAL" assert c["functionally_non_authoritative"] is True assert c["explicit_proposed_verdict_field"] is False def test_check4_anu_independent_collector_only_authoritative(diag): c = _check(diag, 4) # External authoritative boundary MUST be code-enforced. assert c["verdict"] == "PASS" assert c["code_enforced"] is True def test_check5_subagent_direct_anu_guard_partial(diag): c = _check(diag, 5) assert c["verdict"] == "PARTIAL" assert c["executor_self_key_guard_blocks_subagent_channel"] is True assert c["explicit_subagent_scoped_write_guard"] is False assert "send_message" in c["delegate_controller_blocked_tools"] def test_check6_team_lead_callback_owner_is_anu_key(diag): c = _check(diag, 6) # Fail-closed: executor self-key blocked, ANU key allowed. assert c["verdict"] == "PASS" assert c["self_key_registration_blocked"] is True assert c["anu_key_registration_allowed"] is True def test_check7_expected_forbidden_unresolved_recording_gap(diag): c = _check(diag, 7) assert c["verdict"] == "GAP" assert set(c["missing"]) == { "expected_files", "forbidden_files", "unresolved_findings", } def test_check8_concealment_warn_hold_gap(diag): c = _check(diag, 8) assert c["verdict"] == "GAP" assert c["concealment_classifier_present"] is False def test_overall_boundary_enforced_with_subagent_gaps(diag): assert diag["authoritative_callback_boundary_code_enforced"] is True assert diag["overall"] == "BOUNDARY_ENFORCED_SUBAGENT_CONTRACT_GAPS" assert diag["counts"] == {"PASS": 2, "PARTIAL": 3, "GAP": 3} def test_real_authoritative_selector_quarantines_self_chain(): """Direct real-entrypoint assertion (independent of the diag wrapper).""" avs = importlib.import_module("anu_v3.authoritative_verdict_selector") rec = avs.VerdictRecord( kind="collector_result", verdict="PASS", task_id="task-2553+56", executor_key="EXEC", collector_key="EXEC", collector_role="ANU", session_is_executor_self=True, claimed_origin="independent_anu", ) res = avs.select_authoritative_verdict( [rec], task_id="task-2553+56", anu_keys=["ANUKEY"] ) assert res.authoritative_verdict is None assert res.classification == avs.AUTHORITATIVE_VERDICT_PENDING def test_real_callback_owner_blocks_executor_self_key(): """Direct real-entrypoint assertion: self-key callback is blocked.""" cov = importlib.import_module("anu_v3.callback_owner_validator") val = cov.validate_callback_owner_runtime( task_id="task-2553+56", executor_key="c38fb9955616e24d", collector_key="c38fb9955616e24d", collector_role="ANU", normal_collector_cron_id="c1", fallback_callback_cron_id="c2", dispatch_cron_id="c0", ) with pytest.raises(cov.CallbackRegistrationBlocked): cov.assert_registration_permitted(val)