Support multi-node shared Target training and per-Draft packing

README.md

@@ -25,8 +25,8 @@ independent experiments, and take a checkpoint through evaluation and export.
 - **One Target, multiple concurrent Draft experiments.** Compare algorithms, structures or
   hyperparameters while generating compatible Target features once. Each Draft keeps its own
   weights, optimizer, progress and checkpoints, without a shared optimizer-step barrier.
-  See [shared Target training](docs/user-guide/sharing.md) and the
-  [comparison and recovery walkthrough](examples/qwen3-8b-shared-target/README.md).
+  See [shared Target training](docs/user-guide/sharing.md) for single-node and static multi-node workflows,
+  and the [comparison and recovery walkthrough](examples/qwen3-8b-shared-target/README.md).
 - **Make long-context training practical.** Combine sequence packing, context parallelism and
   optimized attention execution to reduce padding waste and distribute long sequences across GPUs.
 - **Explore without rebuilding the workflow.** Switch between algorithm and model recipes with

README_zh-CN.md

@@ -22,7 +22,7 @@ ASpec 基于 [TorchSpec](https://github.com/lightseekorg/TorchSpec) 源码开发
   按有效工作量组织训练，减少两端等待。
 - **一个 Target，多个 Draft 并行实验。** 比较算法、结构或超参数时，兼容的 Target 特征只生成
   一次。各 Draft 独立维护权重、优化器、进度和 checkpoint，无需每次更新都相互等待。
-  当前单机用法见[共享训练指南](docs/user-guide/sharing.md)。
+  单机及静态多机的用法见[共享训练指南](docs/user-guide/sharing.md)。
 - **高效处理长上下文。** 结合序列打包、上下文并行与优化的注意力执行，减少 padding 浪费，
   将长序列训练分布到多张 GPU。
 - **换算法，不重搭工作流。** 通过 recipe 选择算法与结构，沿用一致的配置、报告、恢复、评估和

ROADMAP.md

@@ -16,7 +16,8 @@ The authoritative details and limitations are in [supported paths](docs/support.
 - [x] Sequence packing, context parallelism and independent Drafts sharing compatible Target data.
 - [x] Primary SGLang + Qwen3/Ling path on single-node multi-GPU systems.
 - [x] Bounded static multi-node, single-Draft execution.
-- [ ] Multi-node multi-Draft training: share compatible Target features across independent Draft
+- [x] Static multi-node multi-Draft training with independent packing:
+  share compatible Target features across independent Draft
   experiments, preserving per-Draft progress and checkpoint recovery without requiring a shared filesystem.
 - [ ] Elastic membership and broader failure recovery.
 

aspec/commands/recipes.py

@@ -391,7 +391,7 @@ def run_acceptance(recipe, entry, output, steps, *, inputs, prepared_config=None
                 run = stage / label
                 overrides = [*base, f"training.num_train_steps={limit}"]
                 if label == "resumed":
-                    if recipe == "shared-target":
+                    if entry.get("draft_experiments", 1) > 1:
                         from omegaconf import OmegaConf
 
                         experiments = OmegaConf.to_container(
@@ -425,7 +425,7 @@ def run_acceptance(recipe, entry, output, steps, *, inputs, prepared_config=None
                     limit,
                     label == "resumed",
                     require_eval=recipe != "quickstart",
-                    expected_ranks=entry["draft_gpus"] // (2 if recipe == "shared-target" else 1),
+                    expected_ranks=entry["draft_gpus"] // entry.get("draft_experiments", 1),
                 )
                 if recipe == "qwen27b-long":
                     from aspec.commands._recipe_capacity import audit_long_run

aspec/controller/inference_manager.py

@@ -56,7 +56,7 @@
 )
 from aspec.utils.logging import logger
 from aspec.utils.types import InferenceInput, InferenceOutput
-from aspec.work_lifecycle import WORK_IDENTITY_KEYS, require_work_identity
+from aspec.work_lifecycle import EXECUTION_ID_KEY, WORK_IDENTITY_KEYS, require_work_identity
 
 MOONCAKE_BACKPRESSURE_POLL_INTERVAL = 0.5  # seconds
 MOONCAKE_BACKPRESSURE_LOG_INTERVAL = 5.0  # seconds
@@ -774,6 +774,11 @@ async def get_status(self) -> dict:
         if self._node_ready_watermarks:
             status.update(
                 {
+                    "ready_credit_scope": (
+                        "shared_admission_audience"
+                        if getattr(getattr(self, "args", None), "aspec_shared_source", False)
+                        else "controller"
+                    ),
                     "ready_watermark_by_node": dict(self._node_ready_watermarks),
                     "sample_budget_by_node": dict(self._sample_budget_by_node),
                     "scheduled_samples_by_node": dict(
@@ -868,6 +873,15 @@ async def _start_dispatch(self, dispatch_size: int) -> bool:
             if inflight + dispatch_size
             > self._batch_size * self._engines.capacity_by_node[node]
         } if self._node_ready_watermarks else set()
+        pinned_source = bool(getattr(getattr(self, "args", None), "aspec_shared_source", False)) and bool(self._node_ready_watermarks)
+        if pinned_source:
+            for node in self._node_ready_watermarks:
+                available = sum(
+                    entry.metadata.get("shared_source_node") == node
+                    for entry in self._prompt_buffer
+                )
+                if available < dispatch_size:
+                    excluded_nodes.add(node)
         while True:
             lease = self._engines.acquire_nowait(excluded_nodes=excluded_nodes)
             if lease is None:
@@ -883,9 +897,13 @@ async def _start_dispatch(self, dispatch_size: int) -> bool:
                     excluded_nodes.add(node)
                     continue
             try:
-                admitted = await self._reserve_pool_capacity(
-                    dispatch_size, node=node
-                )
+                source_execution_ids = [
+                    entry.metadata[EXECUTION_ID_KEY]
+                    for entry in self._prompt_buffer
+                    if entry.metadata.get("shared_source_node") == node
+                ][:dispatch_size] if pinned_source else None
+                admission_options = {"source_execution_ids": source_execution_ids} if pinned_source else {}
+                admitted = await self._reserve_pool_capacity(dispatch_size, node=node, **admission_options)
             except BaseException:
                 lease.release()
                 raise
@@ -894,7 +912,17 @@ async def _start_dispatch(self, dispatch_size: int) -> bool:
             lease.release()
             excluded_nodes.add(lease.node)
 
-        entries = self._take_batch(dispatch_size)
+        if pinned_source:
+            entries = []
+            remaining = deque()
+            for entry in self._prompt_buffer:
+                if len(entries) < dispatch_size and entry.metadata.get("shared_source_node") == node:
+                    entries.append(entry)
+                else:
+                    remaining.append(entry)
+            self._prompt_buffer = remaining
+        else:
+            entries = self._take_batch(dispatch_size)
         self._scheduled_samples += len(entries)
         if node is not None and self._sample_budget_by_node:
             self._scheduled_samples_by_node[node] += len(entries)
@@ -972,7 +1000,11 @@ async def _refill_buffer(self) -> None:
         if len(self._prompt_buffer) >= self._buffer_low_watermark:
             return
 
-        if self._max_pool_size > 0:
+        # Shared readers can have opposite node pressures. The global maximum
+        # can be full everywhere while a particular admission still has credit.
+        # Fetch bounded prompts here and gate their actual audiences at dispatch.
+        pinned_source = bool(getattr(getattr(self, "args", None), "aspec_shared_source", False)) and bool(self._node_ready_watermarks)
+        if self._max_pool_size > 0 and not pinned_source:
             if self._node_ready_watermarks:
                 pool_sizes = await self.controller.get_ready_pool_size_by_node.remote()
                 pool_full = all(
@@ -1018,6 +1050,7 @@ async def _reserve_pool_capacity(
         incoming_samples: int,
         *,
         node: str | None = None,
+        source_execution_ids: list[str] | None = None,
     ) -> bool:
         """Gate on ready outputs while accounting for independently bounded requests.
 
@@ -1041,7 +1074,10 @@ async def _reserve_pool_capacity(
                     raise RuntimeError(
                         f"engine lease has no planned locality credit: node={node!r}"
                     )
-                pool_sizes = await self.controller.get_ready_pool_size_by_node.remote()
+                if source_execution_ids is None:
+                    pool_sizes = await self.controller.get_ready_pool_size_by_node.remote()
+                else:
+                    pool_sizes = await self.controller.get_ready_pool_size_by_node.remote(source_execution_ids)
                 unknown_nodes = set(pool_sizes) - set(self._node_ready_watermarks)
                 if unknown_nodes:
                     raise RuntimeError(