Implementation Plan: T5-P2-A6-WP2 Ensure Every Frozenset/Registry Constant Has a Production Consumer

src/autoskillit/server/tools/tools_agents.py

@@ -6,7 +6,7 @@
 from fastmcp.dependencies import CurrentContext
 from fastmcp.exceptions import ResourceError
 
-from autoskillit.core import AGENT_PACK_REGISTRY, get_logger, pkg_root
+from autoskillit.core import AGENT_PACK_REGISTRY, RETIRED_AGENT_NAMES, get_logger, pkg_root
 from autoskillit.server import mcp
 from autoskillit.server.tools._cancellation_shield import _cancellation_shield
 
@@ -53,6 +53,10 @@ async def unlock_agent_pack(pack_name: str, ctx: Context = CurrentContext()) ->
     Never raises.
     """
     try:
+        if pack_name in RETIRED_AGENT_NAMES:
+            return json.dumps(
+                {"success": False, "error": f"Agent pack {pack_name!r} has been retired."}
+            )
         if pack_name not in AGENT_PACK_TAGS:
             return json.dumps({"success": False, "error": f"Unknown agent pack: {pack_name}"})
         tags = AGENT_PACK_TAGS[pack_name]

src/autoskillit/server/tools/tools_recipe.py

@@ -13,7 +13,7 @@
     build_config_authoritative_layer,
     resolve_ingredient_defaults,
 )
-from autoskillit.core import get_logger, temp_dir_display_str
+from autoskillit.core import FLEET_DISPATCH_TOOLS, get_logger, temp_dir_display_str  # noqa: F401
 from autoskillit.pipeline import GATED_TOOLS, UNGATED_TOOLS  # noqa: F401
 from autoskillit.server import mcp
 from autoskillit.server._guards import _require_enabled

tests/arch/test_canonical_constant_consumption.py

@@ -69,3 +69,104 @@ def test_env_forward_constants_have_production_consumer() -> None:
         f"consumers: {unconsumed}. Each env-var-set constant must be imported and consumed "
         f"by production code to prevent dead-canonical-constant drift."
     )
+
+
+_REGISTRY_CANONICAL_PATTERN = re.compile(r"^[A-Z][A-Z0-9_]*(?:_REGISTRY|_TOOLS|_TAGS|_NAMES)$")
+
+_REGISTRY_EXEMPTIONS: dict[str, str] = {
+    "FREE_RANGE_TOOLS": (
+        "alias-derived: backing constant for UNGATED_TOOLS which is imported "
+        "in pipeline/gate.py and server/tools/tools_recipe.py; "
+        "direct import would be redundant"
+    ),
+    "FLEET_TOOLS": (
+        "test-consumed: architectural enforcement constant imported by "
+        "test_layer_enforcement.py, test_transforms_hygiene.py, and "
+        "test_lifespan_fleet_boot.py for fleet tool-tag parity guards; "
+        "no runtime production consumer needed"
+    ),
+}
+
+
+def _find_registry_constants(constants_file: Path) -> list[str]:
+    """Find all module-level names matching registry/tools/tags/names patterns."""
+    tree = ast.parse(constants_file.read_text())
+    names: list[str] = []
+    for node in ast.iter_child_nodes(tree):
+        if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
+            if _REGISTRY_CANONICAL_PATTERN.match(node.target.id):
+                names.append(node.target.id)
+        elif isinstance(node, ast.Assign):
+            for target in node.targets:
+                if isinstance(target, ast.Name) and _REGISTRY_CANONICAL_PATTERN.match(target.id):
+                    names.append(target.id)
+    return names
+
+
+def test_registry_constants_have_production_consumer() -> None:
+    """Every registry/tools/tags/names constant must be imported by production code or exempted."""
+    from autoskillit.core import paths
+
+    src_root = paths.pkg_root()
+    constants_files = [
+        src_root / "core" / "types" / "_type_constants_registries.py",
+        src_root / "core" / "types" / "_type_constants.py",
+    ]
+
+    all_constants: list[tuple[str, Path]] = []
+    for cf in constants_files:
+        for name in _find_registry_constants(cf):
+            all_constants.append((name, cf))
+
+    assert all_constants, f"No registry constants found in {constants_files} — test premise broken"
+
+    unconsumed = []
+    for name, def_file in all_constants:
+        if name in _REGISTRY_EXEMPTIONS:
+            continue
+        if not _has_production_import(src_root, name, def_file):
+            unconsumed.append(name)
+
+    assert not unconsumed, (
+        f"Registry constants (*_REGISTRY / *_TOOLS / *_TAGS / *_NAMES) with zero "
+        f"production consumers and no exemption: {unconsumed}. Each constant must be "
+        f"imported by production src/ code or documented in _REGISTRY_EXEMPTIONS "
+        f"with a rationale."
+    )
+
+
+def test_exemption_rationales_are_nonempty() -> None:
+    """Every exemption must have a non-empty rationale string."""
+    empty = [k for k, v in _REGISTRY_EXEMPTIONS.items() if not v.strip()]
+    assert not empty, f"Exemptions with empty rationales: {empty}"
+
+
+def test_exemptions_reference_real_constants() -> None:
+    """Every exempted name must exist as a constant in the scanned files."""
+    from autoskillit.core import paths
+
+    src_root = paths.pkg_root()
+    constants_files = [
+        src_root / "core" / "types" / "_type_constants_registries.py",
+        src_root / "core" / "types" / "_type_constants.py",
+    ]
+    all_names: set[str] = set()
+    for cf in constants_files:
+        all_names.update(_find_registry_constants(cf))
+
+    stale = set(_REGISTRY_EXEMPTIONS.keys()) - all_names
+    assert not stale, (
+        f"Exemption dict contains names not found in constants files: {sorted(stale)}. "
+        f"Remove stale entries."
+    )
+
+
+def test_fleet_dispatch_tools_subset_of_gated_tools() -> None:
+    """FLEET_DISPATCH_TOOLS must be a subset of GATED_TOOLS."""
+    from autoskillit.core import FLEET_DISPATCH_TOOLS
+    from autoskillit.pipeline import GATED_TOOLS
+
+    extra = FLEET_DISPATCH_TOOLS - GATED_TOOLS
+    assert not extra, (
+        f"FLEET_DISPATCH_TOOLS must be a subset of GATED_TOOLS — extra: {sorted(extra)}"
+    )

tests/server/test_tools_agents.py

@@ -122,6 +122,22 @@ async def test_unlock_agent_pack_unknown_name():
     assert "nonexistent" in data["error"]
 
 
+# T7b: unlock_agent_pack with retired name returns error
+@pytest.mark.anyio
+async def test_unlock_agent_pack_rejects_retired_names():
+    """Retired agent names must be rejected with a structured error."""
+    from autoskillit.core import RETIRED_AGENT_NAMES
+    from autoskillit.server.tools.tools_agents import unlock_agent_pack
+
+    assert RETIRED_AGENT_NAMES, "RETIRED_AGENT_NAMES must not be empty — test premise broken"
+    name = next(iter(sorted(RETIRED_AGENT_NAMES)))
+    mock_ctx = _make_mock_ctx()
+    result = json.loads(await unlock_agent_pack(name, ctx=mock_ctx))
+    assert result["success"] is False
+    assert "retired" in result["error"].lower()
+    assert name in result["error"]
+
+
 # T8: unlock_agent_pack enables agent resources in same session
 @pytest.mark.anyio
 async def test_unlock_agent_pack_enables_resources():