Account for hybrid GGUF recurrent caches in served memory admission

CLAUDE.md

@@ -108,6 +108,16 @@ If `nix fmt` changes any files, stage them before committing. The CI runs `nix f
 
 ### Node Composition
 
+GGUF hybrid cache admission uses `ModelCard.gguf_cache_geometry` from the exact
+artifact header. Qwen3.5 scalar metadata separates attention, recurrent and NextN
+layers; recurrent FP32 buffers scale with slots and speculative rollback rows.
+NodeResources advertises `llama_server_settings`, placement stamps them on shards,
+and the runner refuses changed local settings before spawning. Registry geometry
+is explicitly derived from the separately signed `v1/gguf-metadata.json` target,
+bound to the catalog snapshot, signed role version and exact artifact. Canonical
+cards remain unchanged for older clients. Runtime cards retain the header evidence
+in `registry_gguf_metadata`; it participates in the full-card approval digest.
+
 Exact non-RPC placements resolve omitted backends from advertised compatible
 engines before memory admission; restored unstamped GPU-host shards reserve
 conservatively. Exact and quick-launch master refusals retain correlated instance

src/skulk/api/tests/test_data_plane_dispatch.py

@@ -107,23 +107,25 @@ async def test_state_surfaces_split_data_transport_health() -> None:
     payload = await api.get_cluster_state()
 
     assert payload["nodeResources"] == {
-            "remote-management-node": {
-                "backends": [],
-                "engineBuilds": {},
-                "hardwareClasses": [],
-                "participation": "management",
-                "apiAvailable": True,
-                "dataTransport": "zenoh",
+        "remote-management-node": {
+            "backends": [],
+            "engineBuilds": {},
+            "llamaServerSettings": None,
+            "hardwareClasses": [],
+            "participation": "management",
+            "apiAvailable": True,
+            "dataTransport": "zenoh",
             "zenohConnectedPeers": None,
             "capabilityConflicts": [],
         },
-            "worker-node": {
-                "backends": ["mlx"],
-                "engineBuilds": {},
-                "hardwareClasses": [],
-                "participation": "full",
-                "apiAvailable": True,
-                "dataTransport": "gossipsub",
+        "worker-node": {
+            "backends": ["mlx"],
+            "engineBuilds": {},
+            "llamaServerSettings": None,
+            "hardwareClasses": [],
+            "participation": "full",
+            "apiAvailable": True,
+            "dataTransport": "gossipsub",
             "zenohConnectedPeers": None,
             "capabilityConflicts": [],
         },

src/skulk/master/main.py

@@ -1127,7 +1127,12 @@ def _record_freed_instance(self, instance: Instance) -> None:
             fraction = shard_fraction_of_model(shard)
             if fraction is None or fraction <= 0.0:
                 continue
-            footprint = estimate_shard_footprint(shard.model_card, fraction)
+            footprint = estimate_shard_footprint(
+                shard.model_card,
+                fraction,
+                resolved_backend=shard.resolved_backend,
+                llama_server_settings=shard.llama_server_settings,
+            )
             self._recently_freed_bytes.setdefault(node_id, []).append(
                 (footprint.in_bytes, deadline)
             )
@@ -3468,7 +3473,12 @@ def _steward_replacement_memory_inputs(
             fraction = shard_fraction_of_model(shard)
             if fraction is None or fraction <= 0.0:
                 continue
-            footprint = estimate_shard_footprint(shard.model_card, fraction)
+            footprint = estimate_shard_footprint(
+                shard.model_card,
+                fraction,
+                resolved_backend=shard.resolved_backend,
+                llama_server_settings=shard.llama_server_settings,
+            )
             credit[node_id] = credit.get(node_id, 0) + footprint.in_bytes
         memory = {
             node_id: (

src/skulk/master/placement.py

@@ -85,7 +85,11 @@
     instance_meta_of,
 )
 from skulk.shared.types.worker.runners import ShardAssignments
-from skulk.shared.types.worker.shards import Sharding, TensorShardMetadata
+from skulk.shared.types.worker.shards import (
+    RpcDonorShardMetadata,
+    Sharding,
+    TensorShardMetadata,
+)
 
 # Ring/coordinator/donor listener ports are drawn from a band BELOW every
 # OS's default ephemeral range: macOS assigns outgoing-connection local ports
@@ -233,6 +237,7 @@ def add_instance_to_placements(
         assignments = assignments.model_copy(
             update={"runner_to_shard": resolved_shards}
         )
+    assignments = _stamp_llama_server_settings(assignments, node_resources or {})
     fixed_memory_by_node: dict[NodeId, Memory] = {}
     first_shard = next(iter(assignments.runner_to_shard.values()), None)
     if first_shard is not None:
@@ -290,12 +295,55 @@ def add_instance_to_placements(
                 context_budget=ceiling
                 if ceiling is not None
                 else KV_CONTEXT_BUDGET_TOKENS,
+                resolved_backend=shard.resolved_backend,
+                llama_server_settings=shard.llama_server_settings,
             )
             if ceiling == 0 or footprint > available:
                 raise PlacementError("Insufficient GPU memory for the exact placement")
     return {**current_instances, instance.instance_id: instance}
 
 
+def _stamp_llama_server_settings(
+    assignments: ShardAssignments, resources_by_node: Mapping[NodeId, NodeResources]
+) -> ShardAssignments:
+    shards = dict(assignments.runner_to_shard)
+    has_rpc_donors = any(
+        isinstance(shard, RpcDonorShardMetadata) for shard in shards.values()
+    )
+    for node_id, runner_id in assignments.node_to_runner.items():
+        shard = shards[runner_id]
+        if isinstance(shard, RpcDonorShardMetadata):
+            # Donors provide memory to the driver's one context; they do not
+            # launch a served instance with their own parallel-slot controls.
+            continue
+        if (
+            shard.resolved_backend is None
+            and shard.model_card.gguf_cache_geometry is not None
+            and not has_rpc_donors
+        ):
+            # A worker may resolve this shard to a served engine after telemetry
+            # catches up. Default or caller-supplied slots cannot authorize that
+            # later process's recurrent allocation.
+            raise PlacementError(
+                "Backend telemetry is required for recurrent memory admission"
+            )
+        if not has_rpc_donors and (
+            shard.resolved_backend is None
+            or not shard.resolved_backend.startswith("llama_server")
+        ):
+            continue
+        resources = resources_by_node.get(node_id)
+        settings = resources.llama_server_settings if resources is not None else None
+        if settings is None and shard.model_card.gguf_cache_geometry is not None:
+            raise PlacementError(
+                "Serving settings are required for recurrent memory admission"
+            )
+        # Exact requests cannot substitute client-supplied slot settings for the
+        # node's observation. The worker verifies this stamp before spawning.
+        shards[runner_id] = shard.model_copy(update={"llama_server_settings": settings})
+    return assignments.model_copy(update={"runner_to_shard": shards})
+
+
 def _get_node_download_fraction(
     node_id: NodeId,
     model_id: ModelId,
@@ -920,6 +968,20 @@ def place_instance(
             command.sharding,
             node_vram=node_vram,
             fixed_memory_by_node=standard_fixed,
+            resolved_backends={
+                node_id: resolve_node_backend(
+                    _card_platform_backends(command.model_card, resources),
+                    command.model_card.placement.backend_preference,
+                    resources.backends,
+                )
+                for node_id in standard_cycle.node_ids
+                if (resources := resolved_node_resources.get(node_id)) is not None
+            },
+            llama_server_settings={
+                node_id: resources.llama_server_settings
+                for node_id in standard_cycle.node_ids
+                if (resources := resolved_node_resources.get(node_id)) is not None
+            },
         )
         cycles_with_sufficient_memory.extend(standard_fit)
         memory_diagnostics.pending_info_node_ids.extend(
@@ -978,6 +1040,18 @@ def place_instance(
                 node_vram=rpc_vram_map,
                 exact_pipeline_layers=False,
                 fixed_memory_by_node=rpc_fixed,
+                resolved_backends={
+                    node_id: "llama_server" for node_id in rpc_cycle.node_ids
+                },
+                llama_server_settings={
+                    node_id: (
+                        resources.llama_server_settings
+                        if (resources := resolved_node_resources.get(driver))
+                        is not None
+                        else None
+                    )
+                    for node_id in rpc_cycle.node_ids
+                },
             )
             cycles_with_sufficient_memory.extend(rpc_fit)
             if rpc_fit:
@@ -1179,6 +1253,10 @@ def place_instance(
         node_to_runner=shard_assignments.node_to_runner,
     )
 
+    shard_assignments = _stamp_llama_server_settings(
+        shard_assignments, resolved_node_resources
+    )
+
     # Stamp the context-admission ceiling into the placement decision (#279
     # slice 2). Computed once here from the hosting nodes' static ram_total, so
     # every rank reads the identical value off replicated state rather than

src/skulk/master/placement_utils.py

@@ -5,13 +5,16 @@
 from loguru import logger
 from pydantic import Field
 
+from skulk.shared.models.llama_server_settings import LlamaServerSettings
 from skulk.shared.models.memory_estimate import (
     GPU_VRAM_WORKING_SET_FRACTION,
     GPU_WORKING_SET_FRACTION,
     UMA_GPU_OS_HEADROOM,
     backend_offloads_to_vram,
+    estimate_recurrent_cache_bytes,
     estimate_shard_footprint,
     memory_overhead_factor,
+    per_token_kv_bytes,
     shard_fraction_of_model,
 )
 from skulk.shared.models.memory_estimate import (
@@ -247,6 +250,8 @@ def reserve_instance_vram(
             footprint = estimate_shard_footprint(
                 shard.model_card,
                 fraction,
+                resolved_backend=shard.resolved_backend,
+                llama_server_settings=shard.llama_server_settings,
                 context_budget=(
                     instance.context_token_limit
                     if instance.context_token_limit is not None
@@ -458,6 +463,8 @@ def filter_cycles_by_memory(
     *,
     exact_pipeline_layers: bool = True,
     fixed_memory_by_node: Mapping[NodeId, Memory] | None = None,
+    resolved_backends: Mapping[NodeId, str | None] | None = None,
+    llama_server_settings: Mapping[NodeId, LlamaServerSettings | None] | None = None,
 ) -> tuple[list[Cycle], CycleMemoryDiagnostics]:
     """Keep cycles whose every node can hold its shard with runtime headroom.
 
@@ -559,10 +566,37 @@ def filter_cycles_by_memory(
         overhead_factor = memory_overhead_factor(model_card)
         overloaded: list[str] = []
         for node_id, share in node_shares.items():
-            kv_share = share * kv_ratio
+            backend = (resolved_backends or {}).get(node_id)
+            settings = (llama_server_settings or {}).get(node_id)
+            node_kv_ratio = kv_ratio
+            recurrent_share = Memory()
+            if model_card.gguf_cache_geometry is not None:
+                if (
+                    backend is not None
+                    and backend.startswith("llama_server")
+                    and settings is None
+                ):
+                    overloaded.append(
+                        f"node {node_id} has no observed llama-server serving settings"
+                    )
+                    continue
+                node_kv_ratio = (
+                    per_token_kv_bytes(
+                        model_card,
+                        resolved_backend=backend,
+                        llama_server_settings=settings,
+                    )
+                    * context_budget
+                    / required_memory.in_bytes
+                )
+                recurrent_share = estimate_recurrent_cache_bytes(
+                    model_card, resolved_backend=backend, llama_server_settings=settings
+                ) * (share.in_bytes / required_memory.in_bytes)
+            kv_share = share * node_kv_ratio
             required_with_overhead = (
                 share * overhead_factor
                 + kv_share
+                + recurrent_share
                 + PLACEMENT_MEMORY_OVERHEAD_FLOOR
                 + fixed_memory_by_node.get(node_id, Memory())
             )