feat(engine): per-tool tool-schema tiering + tier-aware guard (slab Phases 1-2) [A/B-gated]

servers/engine/server.py

@@ -13591,5 +13591,92 @@ def mark_climax(campaign_id: str) -> dict:
         return {"climax_landed": True, "climax_day": arc.climax_day}
 
 
+# --- Tool-schema tiering (slab decision, Phases 1-2) -----------------------------------------
+# The engine tool-schema slab is injected into EVERY DM beat. Pinning all 153 tools (whole-server
+# `alwaysLoad`) taxes every request AND dilutes tool-selection accuracy (a 285-transcript census
+# found 92/153 tools never called in real play). We PIN a census-backed core — the hot beat loop,
+# the full active-combat verb set (die-triggered, no payload hint, so a mid-fight ToolSearch hop is
+# the worst failure), the cold-open path (no payload names them; inside the 22-turn give-up band),
+# and the reach-for surface — and let the harness DEFER the cold authoring/catalog/campaign/arc/
+# economy tail behind ToolSearch. New tools default DEFERRED (absent from the allowlist) so the slab
+# can't silently re-bloat; promoting one to the core is a reviewed edit here + a ratchet-test bump.
+#
+# Per-tool pinning rides `_meta["anthropic/alwaysLoad"]`, which claude resolves PER-TOOL and FastMCP
+# propagates from a tool's `.meta` (verified on the installed mcp 1.27.1 / claude 2.1.160). Cold
+# tools stay FINDABLE: the engine names them in the obligations/director payloads the DM already
+# holds, or they are explicit-intent-gated, so the DM ToolSearches them on demand (Step-1.7 reach-
+# for validation found no selection regression from deferring them).
+#
+# A/B control = the EXISTING whole-server `WORLDOS_ENGINE_ALWAYSLOAD` flag (qa launchers / .mcp.json).
+# When '1' (default = today's baseline) the harness ORs the server-level pin over EVERY tool, so the
+# per-tool meta is inert — we skip applying it to keep the baseline slab byte-identical. When '0'
+# (the tiered A/B arm) we annotate the core and the harness defers the rest. Flipping the production
+# default to tiered is the post-A/B cutover, gated on the duo A/B (cache + cold-open + selection).
+PINNED_ALLOWLIST: frozenset[str] = frozenset({
+    # beat loop / narration / state / recall
+    "log_event", "persist_beat", "scene_context", "present_events", "remember",
+    "record_decision", "recall", "recall_npc",
+    # perception / movement / world reads
+    "look_around", "travel_to", "get_scene", "get_state", "find_npcs", "active_campaign",
+    # checks / rolls
+    "skill_check", "social_check", "saving_throw", "roll", "spell_save_dc",
+    # combat — the full active-verb set (die-triggered, no payload hint; pin generously)
+    "start_combat", "end_combat", "attack", "cast_spell", "next_turn", "use_action", "use_resource",
+    "apply_damage", "apply_healing", "add_condition", "remove_condition",
+    "concentration_save", "drop_concentration", "roll_death_save", "stabilize",
+    "set_hp", "set_temp_hp", "grapple", "shove", "escape_grapple",
+    "add_combatant", "remove_combatant", "place_combatant", "combatants_in_zone", "move_to_zone",
+    "spawn_monster",
+    # combat — grid twins of the zone verbs (rebased over #1246/#1248/#1250; same hot verb set)
+    "set_grid", "place_combatant_at_coords", "move_to_coords",
+    # companions (hot interaction + combat suggestion)
+    "companion_advise", "companion_suggest_action", "recruit_companion",
+    # cold-open path / session / character setup (no payload names them; the give-up band)
+    "start_world", "start_session", "start_adventure", "get_prelude", "get_quest_hooks",
+    "create_character", "load_canon_character", "list_canon_characters", "start_character",
+    "generate_image", "list_worlds",
+    # rest / time
+    "long_rest", "short_rest", "advance_time",
+    # quest / progression hot
+    "add_quest", "complete_objective", "level_up", "award_xp",
+    # frequent edits
+    "update_character", "add_location", "set_attitude",
+})
+
+_ALWAYS_LOAD_META_KEY = "anthropic/alwaysLoad"
+
+
+def _apply_tool_tiering(server: "FastMCP" = mcp, *, force: bool = False) -> dict:
+    """Pin only the PINNED_ALLOWLIST tools so the harness keeps the core in-context every beat and
+    DEFERS the rest behind ToolSearch. Runs at import (after every @mcp.tool is registered).
+
+    Inert under the whole-server baseline (`WORLDOS_ENGINE_ALWAYSLOAD` == '1', default): the harness
+    ORs the server pin over every tool, so we skip the per-tool meta to keep the baseline byte-clean.
+    Applies it for the tiered arm (`WORLDOS_ENGINE_ALWAYSLOAD` == '0') or when `force=True` (tests).
+    Returns a small summary. Always validates the allowlist names exist (catches a typo/rename loud).
+    """
+    tiered = force or os.environ.get("WORLDOS_ENGINE_ALWAYSLOAD", "1") != "1"
+    tools = server._tool_manager._tools  # FastMCP private registry (pinned to mcp 1.27.1)
+    unknown = PINNED_ALLOWLIST - set(tools)
+    if unknown:
+        raise RuntimeError(
+            f"PINNED_ALLOWLIST names tools that don't exist (typo/rename?): {sorted(unknown)}"
+        )
+    if not tiered:
+        return {"tiered": False, "pinned": len(tools), "deferred": 0, "total": len(tools)}
+    for name, tool in tools.items():
+        meta = dict(tool.meta or {})
+        if name in PINNED_ALLOWLIST:
+            meta[_ALWAYS_LOAD_META_KEY] = True
+        else:
+            meta.pop(_ALWAYS_LOAD_META_KEY, None)
+        tool.meta = meta or None
+    return {"tiered": True, "pinned": len(PINNED_ALLOWLIST),
+            "deferred": len(tools) - len(PINNED_ALLOWLIST), "total": len(tools)}
+
+
+_apply_tool_tiering()
+
+
 if __name__ == "__main__":
     mcp.run()

servers/engine/tests/test_tool_schema_budget.py

@@ -1,35 +1,39 @@
-"""SYN-02 (F13-1 + F14-6): the engine tool-schema slab is pinned via ``alwaysLoad``
-into EVERY DM request — it is ~54% of the lean first-request floor and the #1
-latency line item against #753. This test is the CI byte-budget guard that keeps the
-docstring diet from regressing: a careless verbose-docstring re-add re-inflates the
-pinned mass and silently re-taxes every beat.
-
-Source: docs/audits/ENGINE-AUDIT-2026-06-11.md (SYN-02 / F13-1 / F14-6).
-
-Two assertions, both load-bearing:
-  1. The serialized ``list_tools()`` JSON (what the wrapper pins) stays under budget.
-  2. The "reach-for" first sentence of a sample of CORE tools survives the diet — the
-     diet must compress prose/examples, NEVER amputate the one sentence the DM needs to
-     pick the right tool (the reach-for lesson: a wrongly-blanked tool the DM needs is
-     worse than the token cost).
+"""SYN-02 — tool-schema slab budget guard, now TIER-AWARE (slab decision, Phase 2).
+
+The engine tool-schema slab is injected via the MCP tool list into EVERY DM request. Under the
+tiered config (slab decision) only the census-backed ``server.PINNED_ALLOWLIST`` core carries
+``_meta['anthropic/alwaysLoad']`` and stays in-context every beat; the cold tail is deferred behind
+the harness ToolSearch. This guard pins that win and keeps it from regressing:
+
+  1. The PINNED-core slab (what ships every beat in the tiered arm) stays under a RATCHET ceiling.
+  2. The pinned set is EXACTLY ``PINNED_ALLOWLIST`` — so a NEW @mcp.tool defaults to DEFERRED and a
+     careless promotion into the core is a visible, reviewed diff (the forcing function).
+  3. Per-tool ``_meta['anthropic/alwaysLoad']`` actually propagates to ``list_tools()`` — a
+     FastMCP/claude upgrade that broke the path would silently un-pin the hot set; fail loud instead.
+  4. The full ``list_tools()`` JSON (the baseline arm still ships all tools) stays under a secondary
+     cap, and the 18 core "reach-for" tools keep their disambiguating first sentence.
+
+Baselines (measure with qa/schema_mass.py): full slab (all 156) ~124,835 B; pinning the
+census-backed core keeps the per-beat tiered slab to ~66,955 B. Lower a ratchet when
+tiering/trimming reclaims more; raise it ONLY with a justification (a tool promoted into the core).
+Rebase over #1246/#1248/#1250 added 3 grid-combat verbs (set_grid, place_combatant_at_coords,
+move_to_coords) — grid twins of the already-pinned zone verbs — promoted into the core, which lifts
+both the pinned ratchet (63,862 -> 66,955 B) and the full-slab cap (118,739 -> 124,835 B).
 """
 
 import asyncio
+import contextlib
 import json
 
 import server
 
 
-# Authoritative metric: the exact wire shape the harness pins (name + description +
-# inputSchema), serialized compactly. Baseline was 167,989 B BEFORE the docstring diet
-# and 105,483 B AFTER (−37%). The budget sits with headroom above the post-diet size so
-# normal tool additions don't trip it, but a wholesale verbose-prose re-add (the
-# regression we're guarding — it silently re-taxes every pinned DM request) does.
-# Re-measure with qa/schema_mass.py and only raise this number with a justification.
-SCHEMA_JSON_BUDGET_BYTES = 120_000
+# Ratchet ceiling for the PINNED core (the slab injected every beat in the tiered arm).
+PINNED_SLAB_BUDGET_BYTES = 67_000
+# Secondary cap on the whole list_tools() JSON (the baseline arm still ships all 156 tools).
+FULL_SLAB_BUDGET_BYTES = 125_000
 
 # Core "reach-for" tools whose first descriptive sentence the DM relies on to disambiguate.
-# Each must keep a non-trivial leading sentence after the diet.
 REACH_FOR_TOOLS = {
     "start_world",
     "look_around",
@@ -53,30 +57,106 @@
 
 
 def _wire_tools():
-    tools = asyncio.run(server.mcp.list_tools())
-    return [t.model_dump(exclude_none=True) if hasattr(t, "model_dump") else dict(t) for t in tools]
-
-
-def test_list_tools_json_under_budget():
-    wire = _wire_tools()
-    blob = json.dumps(wire, ensure_ascii=False, separators=(",", ":"))
-    size = len(blob.encode("utf-8"))
-    assert size <= SCHEMA_JSON_BUDGET_BYTES, (
-        f"engine list_tools JSON is {size} B, over the {SCHEMA_JSON_BUDGET_BYTES} B "
-        f"pinned-schema budget (SYN-02). The docstring diet regressed — compress the "
-        f"verbose docstrings (keep the first reach-for sentence + args; cut prose/examples)."
+    return list(asyncio.run(server.mcp.list_tools()))
+
+
+def _bytes(tools) -> int:
+    blob = json.dumps(
+        [t.model_dump(by_alias=True, exclude_none=True) for t in tools],
+        ensure_ascii=False,
+        separators=(",", ":"),
+    )
+    return len(blob.encode("utf-8"))
+
+
+def _alwaysload(tool) -> bool:
+    meta = tool.model_dump(by_alias=True, exclude_none=True).get("_meta") or {}
+    return meta.get("anthropic/alwaysLoad") is True
+
+
+@contextlib.contextmanager
+def _tiered():
+    """Apply per-tool tiering, yield the wire tools, then restore the byte-clean baseline so sibling
+    tests (and the full-slab assertion) see a registry with no leaked ``_meta``."""
+    server._apply_tool_tiering(force=True)
+    try:
+        yield _wire_tools()
+    finally:
+        for t in server.mcp._tool_manager._tools.values():
+            if t.meta:
+                t.meta.pop(server._ALWAYS_LOAD_META_KEY, None)
+                if not t.meta:
+                    t.meta = None
+
+
+def test_pinned_slab_under_ratchet():
+    """The PINNED core — the slab actually injected into every DM beat under tiering — stays under
+    the ratchet ceiling (and per-tool _meta propagates so the pin is real)."""
+    with _tiered() as tools:
+        pinned = [t for t in tools if t.name in server.PINNED_ALLOWLIST]
+        size = _bytes(pinned)
+        assert size <= PINNED_SLAB_BUDGET_BYTES, (
+            f"pinned tool slab is {size} B, over the {PINNED_SLAB_BUDGET_BYTES} B ratchet "
+            f"({len(pinned)} pinned tools). Either a pinned description grew (trim its prose, keep "
+            f"the reach-for first sentence + args) or a tool was promoted into PINNED_ALLOWLIST — "
+            f"in which case lower/raise this ratchet WITH a justification."
+        )
+        # Per-tool alwaysLoad must actually reach the wire, else the harness silently un-pins the
+        # hot set (a FastMCP/claude upgrade could change the path).
+        pb = next(t for t in tools if t.name == "persist_beat")
+        assert _alwaysload(pb), (
+            "per-tool _meta['anthropic/alwaysLoad'] did NOT propagate to list_tools() — the "
+            "FastMCP/claude per-tool pinning path changed; refusing to silently ship an un-pinned slab"
+        )
+
+
+def test_pinned_set_is_exactly_the_allowlist():
+    """The pinned set == the reviewed PINNED_ALLOWLIST, so a NEW @mcp.tool defaults to DEFERRED and a
+    promotion into the always-loaded core is a visible, reviewed change (the growth forcing-function)."""
+    with _tiered() as tools:
+        pinned = {t.name for t in tools if _alwaysload(t)}
+        allow = set(server.PINNED_ALLOWLIST)
+        assert pinned == allow, (
+            "pinned set != PINNED_ALLOWLIST. "
+            f"unexpectedly pinned: {sorted(pinned - allow)}; "
+            f"in allowlist but not pinned: {sorted(allow - pinned)}. "
+            "New tools must default DEFERRED; update PINNED_ALLOWLIST deliberately if a tool is "
+            "genuinely hot/cold-open/combat-path."
+        )
+
+
+def test_baseline_is_byte_clean():
+    """The whole-server baseline arm (WORLDOS_ENGINE_ALWAYSLOAD default) carries NO per-tool _meta —
+    so adopting tiering is byte-identical until the production default is flipped (post-A/B cutover)."""
+    tools = _wire_tools()
+    assert not any(_alwaysload(t) for t in tools), (
+        "baseline list_tools() carries per-tool alwaysLoad _meta — the import-time apply should be "
+        "inert under the whole-server baseline (it would bloat the baseline slab)."
+    )
+
+
+def test_full_list_tools_under_secondary_budget():
+    """The full list_tools() JSON (baseline arm ships all 153) stays under a secondary cap — a
+    backstop against unbounded total schema growth independent of which tools are pinned."""
+    size = _bytes(_wire_tools())
+    assert size <= FULL_SLAB_BUDGET_BYTES, (
+        f"full list_tools JSON is {size} B, over the {FULL_SLAB_BUDGET_BYTES} B secondary cap. "
+        f"Trim verbose descriptions (keep the reach-for first sentence + args; cut prose/examples)."
     )
 
 
 def test_reach_for_first_sentence_intact():
-    by_name = {t["name"]: t for t in _wire_tools()}
+    by_name = {t.name: t for t in _wire_tools()}
     missing = sorted(REACH_FOR_TOOLS - set(by_name))
     assert not missing, f"reach-for tools vanished from list_tools: {missing}"
+    # Every reach-for tool must also be PINNED (the DM reaches for it directly — never defer it).
+    not_pinned = sorted(REACH_FOR_TOOLS - set(server.PINNED_ALLOWLIST))
+    assert not not_pinned, (
+        f"reach-for tools are not in PINNED_ALLOWLIST (would be deferred): {not_pinned}"
+    )
     thin = []
     for name in sorted(REACH_FOR_TOOLS):
-        desc = (by_name[name].get("description") or "").strip()
-        # The reach-for sentence must survive — guard against a blanked / amputated
-        # description (a couple of words is not enough to disambiguate the tool).
+        desc = (by_name[name].description or "").strip()
         if len(desc) < 25 or len(desc.split()) < 4:
             thin.append((name, desc[:60]))
     assert not thin, (