Host overhead 5/N: cheaper resource planning, admit and commit

docs/environment_variables.rst

@@ -646,9 +646,11 @@ Worker scheduling
        master write. ``0`` keeps every node on the per-request tensor path.
    * - ``MSTAR_INGRAPH_DECODE_ROPE``
      - config's ``knobs: ingraph_decode_rope``, else ``0``
-     - A captured decode step builds its rotary tables inside the graph from
-       the position resource's planned positions instead of staging them.
-       Experiment knob, measured and gated on its own.
+     - A captured step of a text walk (Qwen3.5 decode, text prefill and
+       mixed) builds its rotary tables inside the graph from the position
+       resource's planned positions instead of building them before the
+       replay (``LLMSubmodule._rope_in_graph``). Experiment knob, measured
+       and gated on its own.
    * - ``MSTAR_SAMPLER_INGRAPH_SCATTER``
      - config's ``knobs: sampler_ingraph_scatter``, else ``0``
      - The captured sampler gathers its RNG offsets from the slot masters

docs/installation.rst

@@ -244,6 +244,9 @@ The GPU model families depend on:
 
 - **FlashInfer** (``flashinfer-python``) — paged attention and continuous batching for the
   autoregressive backbones (every model with a ``KV_CACHE`` node runs attention through it).
+  Pinned to the version the attention wrappers' direct planner calls were checked against
+  (``DIRECT_PLAN_FLASHINFER_VERSIONS`` in ``mstar/engine/resources/attn/wrappers.py``); on any
+  other version they fall back to FlashInfer's own ``plan()``.
 - **flash-attn** — used by LingBot-World's ``reference_numerics`` mode. **Not installed by any extra**; install it separately
   (see `flash-attn (LingBot-World)`_).
 - **flash-attn-4** — Waypoint's flex-attention ``FLASH`` backend. **Not installed by any

mstar/engine/engine.py

@@ -1696,7 +1696,10 @@ def _drive_step(
                 )
             t0 = phase_start()
             try:
-                self._runner.plan(step)
+                # promoted and inline plans per resource, apart
+                self._runner.plan(
+                    step, "res.plan" if batch.preplan_event is not None else "res.plan_fresh",
+                )
             finally:
                 phase_end(
                     "engine.plan.promoted" if batch.preplan_event is not None

mstar/engine/resources/attn/base.py

@@ -246,8 +246,9 @@ class PlanCacheKey(NamedTuple):
     """Fingerprint of a wrapper ``plan`` call's inputs. When it is unchanged
     between steps the re-plan is skippable."""
     q_seq_lens: tuple
-    page_indices: tuple
-    last_page_lens: tuple
+    # the int32 arrays' bytes
+    page_indices: bytes
+    last_page_lens: bytes
 
 
 class EagerSlotKey(NamedTuple):

mstar/engine/resources/attn/cross.py

@@ -2,6 +2,7 @@
 
 from typing import Any, NamedTuple
 
+import numpy as np
 import torch
 
 from mstar.engine.resources.attn.base import (
@@ -29,6 +30,7 @@
 from mstar.engine.resources.kv.config import PagedKVConfig
 from mstar.engine.resources.kv.plan import (
     SINK_PAGE,
+    HostLens,
     KVPlanOutputs,
     PagedIndptrs,
     SequenceView,
@@ -203,11 +205,11 @@ def plan(self, step: AttentionStep, ctx: StepContext):
 
             # The context pages are immutable once written, so only the query
             # side moves between steps; when neither moved, FlashInfer's plan
-            # is skippable. Every tensor here is on CPU: `.tolist()` costs no sync.
+            # is skippable. Every tensor here is on CPU.
             plan_key = PlanCacheKey(
                 q_seq_lens=qo_indptr,
-                page_indices=tuple(indptrs.paged_kv_indices.tolist()),
-                last_page_lens=tuple(indptrs.paged_kv_last_page_len.tolist()),
+                page_indices=indptrs.paged_kv_indices.numpy().tobytes(),
+                last_page_lens=indptrs.paged_kv_last_page_len.numpy().tobytes(),
             )
             if last is None or last.plan_key != plan_key:
                 wrapper.plan(
@@ -307,6 +309,7 @@ def _build_indptrs(
         kv_indptr = [0]
         all_pages: list[int] = []
         last_page_lens: list[int] = []
+        kv_lens: list[int] = []
 
         for request_id in packing.request_ids:
             context = context_views.get(request_id)
@@ -323,13 +326,16 @@ def _build_indptrs(
                 all_pages.append(SINK_PAGE)
                 kv_indptr.append(kv_indptr[-1] + 1)
                 last_page_lens.append(1)
+                kv_lens.append(1)
                 continue
             # stream can have more pages than context needs takes only the prefix
             # written length covers
             n_pages = (context.length + page_size - 1) // page_size
             all_pages.extend(context.page_idxs[:n_pages])
             kv_indptr.append(kv_indptr[-1] + n_pages)
-            last_page_lens.append(context.last_page_len(page_size) or page_size)
+            last = context.last_page_len(page_size) or page_size
+            last_page_lens.append(last)
+            kv_lens.append((n_pages - 1) * page_size + last)
 
         # fork repeats pages, so batch can index more than held in cache.
         assert len(all_pages) <= self._kv_config.max_num_pages, (
@@ -338,11 +344,15 @@ def _build_indptrs(
             "raise max_num_pages on the context cache's PagedKVConfig"
         )
 
+        # numpy builds an int32 array from a list in about half the time
+        # `torch.tensor(list)` takes; host `kv_lens` let a captured decode
+        # wrapper call FlashInfer's planner directly (see `_plan_direct`)
         return PagedIndptrs(
             qo_indptr=packing.qo_indptr,
-            paged_kv_indptr=torch.tensor(kv_indptr, dtype=torch.int32),
-            paged_kv_indices=torch.tensor(all_pages, dtype=torch.int32),
-            paged_kv_last_page_len=torch.tensor(last_page_lens, dtype=torch.int32),
+            paged_kv_indptr=torch.from_numpy(np.array(kv_indptr, np.int32)),
+            paged_kv_indices=torch.from_numpy(np.array(all_pages, np.int32)),
+            paged_kv_last_page_len=torch.from_numpy(np.array(last_page_lens, np.int32)),
+            kv_lens=torch.from_numpy(np.array(kv_lens, np.int32)).as_subclass(HostLens),
         )
 
     def _wrapper_for(

mstar/engine/resources/attn/flashinfer.py

@@ -233,7 +233,7 @@ def plan(self, step: AttentionStep, ctx: StepContext):
         plan_states = self._preplan_states if ctx.is_preplan else self._current_plan_states
 
         plan_states.clear()
-        laps = PhaseLaps("res.pre_plan.attn." if ctx.is_preplan else "res.plan.attn.")
+        laps = PhaseLaps(f"{ctx.plan_phase}.attn.")
         for label, kv_out in plan_outputs.items():
             indptrs = kv_out.cpu_indptrs
             if lease is not None: