[PERF] Add CUDA IPC Ulysses overlap and lossless LingBot streaming

benchmarks/telefuser_aiperf/configs/stream_lingbot_world_v2_1min.json

@@ -19,6 +19,7 @@
     "sample_shift": 10.0,
     "control_mode": "cam",
     "show_control_hud": false,
+    "delivery_mode": "lossless",
     "benchmark_metrics": true
   },
   "transport": {

benchmarks/telefuser_aiperf/telefuser_aiperf/adapter.py

@@ -70,12 +70,16 @@ def __init__(
         self.connected = False
         self.target_ready = False
         self.active_event = asyncio.Event()
+        self.target_done_event = asyncio.Event()
+        self.delivery_complete_event = asyncio.Event()
         self.first_frame_event = asyncio.Event()
         self.done_event = asyncio.Event()
         self.control_task: asyncio.Task[None] | None = None
+        self.delivery_ack_task: asyncio.Task[None] | None = None
         self.pending_control_acks: deque[int] = deque()
         self.pending_control_frames: deque[int] = deque()
         self.control_sent_at: dict[int, float] = {}
+        self.expected_published_frames: int | None = None
 
     def _active_start(self) -> float:
         if self.active_started_at is None:
@@ -104,8 +108,32 @@ def _handle_video_frame(self) -> None:
             self.events.record("first_frame")
         self.last_frame_at = now
         self._mark_control_frame(now)
+        self._mark_delivery_complete()
         self._try_start_active_window()
 
+    def _mark_delivery_complete(self) -> None:
+        if self.expected_published_frames is None or self.result.frames_received < self.expected_published_frames:
+            return
+        if not self.result.done_received:
+            return
+        if self.delivery_ack_task is None:
+            self.delivery_ack_task = asyncio.create_task(self._send_delivery_ack())
+
+    async def _send_delivery_ack(self) -> None:
+        if self.expected_published_frames is None:
+            return
+        try:
+            await self.room.publish_data(
+                {"type": "delivery_ack", "published_frames": self.expected_published_frames},
+                topic=_CONTROL_TOPIC,
+                reliable=True,
+            )
+            self.events.record("delivery_ack_sent", published_frames=self.expected_published_frames)
+        except Exception as exc:  # noqa: BLE001 - delivery transport errors are benchmark data
+            self.result.error = redact_string(f"delivery acknowledgement failed: {exc}")
+        finally:
+            self.delivery_complete_event.set()
+
     def _mark_control_frame(self, now: float) -> None:
         if not self.pending_control_frames:
             return
@@ -121,6 +149,7 @@ def _handle_room_event(self, event: str, payload: Mapping[str, Any]) -> None:
         self.events.record(event, **dict(payload))
         if event == "disconnected":
             self.done_event.set()
+            self.target_done_event.set()
 
     def _handle_data_message(
         self,
@@ -146,15 +175,24 @@ def _handle_data_message(
             return
         if payload.get("type") == "done":
             self.result.done_received = True
+            published_frames = payload.get("published_frames")
+            if isinstance(published_frames, int) and published_frames >= 0:
+                self.expected_published_frames = published_frames
+            self._mark_delivery_complete()
+            if self.expected_published_frames is None:
+                self.delivery_complete_event.set()
+            self.target_done_event.set()
             self.done_event.set()
-            self.events.record("done_message", topic=topic)
+            self.events.record("done_message", topic=topic, published_frames=self.expected_published_frames)
             return
         data = payload.get("data") if isinstance(payload.get("data"), dict) else payload
         self._record_transport_profile(data)
         if data.get("type") == "error" or payload.get("error"):
             error = data.get("error") or payload.get("error")
             self.result.error = redact_string(str(error))
             self.done_event.set()
+            self.target_done_event.set()
+            self.delivery_complete_event.set()
         stage = data.get("stage")
         if stage is not None:
             self._handle_status_stage(str(stage), data, now)
@@ -314,9 +352,21 @@ async def _wait_for_media(self) -> None:
             0.0,
         )
         try:
-            await asyncio.wait_for(self.done_event.wait(), timeout=remaining)
+            await asyncio.wait_for(self.target_done_event.wait(), timeout=remaining)
         except asyncio.TimeoutError:
             self.events.record("session_duration_elapsed")
+            return
+        if self.expected_published_frames is not None:
+            try:
+                await asyncio.wait_for(
+                    self.delivery_complete_event.wait(),
+                    timeout=float(self.adapter.options.shutdown_timeout_s),
+                )
+            except asyncio.TimeoutError as exc:
+                raise TimeoutError(
+                    "Target completed publishing but the client did not receive all frames "
+                    f"({self.result.frames_received}/{self.expected_published_frames})"
+                ) from exc
 
     async def run(self) -> SessionResult:
         self.started_at = time.perf_counter()

benchmarks/telefuser_aiperf/tests/test_livekit_adapter.py

@@ -373,3 +373,49 @@ async def test_adapter_maps_control_ack_and_next_frame(tmp_path: Path) -> None:
     assert result.control_events[0].next_frame_latency_ms is not None
     room = _ControlLiveKitRoom.instances[0]
     assert room.published[0][1:] == ("tf.control", True)
+
+
+class _IncompleteDeliveryRoom(_FakeLiveKitRoom):
+    async def connect(
+        self,
+        url: str,
+        token: str,
+        *,
+        timeout_s: float,
+        on_data: Callable[[bytes | str, str, str], None],
+        on_video_frame: Callable[[], None],
+        on_event: Callable[[str, Mapping[str, Any]], None],
+    ) -> None:
+        self.connected = (url, token)
+        on_data(
+            orjson.dumps({"type": "chunk", "data": {"stage": "worker_running"}}),
+            "tf.status",
+            "telefuser-worker-0",
+        )
+        on_video_frame()
+        on_video_frame()
+        on_data(
+            orjson.dumps({"type": "done", "session_id": "livekit-session", "published_frames": 3}),
+            "tf.status",
+            "telefuser-worker-0",
+        )
+
+
+@pytest.mark.asyncio
+async def test_adapter_fails_when_client_does_not_receive_target_published_frames(tmp_path: Path) -> None:
+    _IncompleteDeliveryRoom.instances.clear()
+    adapter = TeleFuserLiveKitAdapter(
+        contract=_contract(),
+        config=_config(tmp_path),
+        artifacts_dir=tmp_path,
+        room_client_factory=_IncompleteDeliveryRoom,
+        http_client=_FakeHttpClient(),
+    )
+
+    result = await adapter.run_session(_plan())
+
+    assert result.success is False
+    assert result.frames_received == 2
+    assert result.done_received is True
+    assert result.error is not None
+    assert "2/3" in result.error

docs/en/benchmark_aiperf.md

@@ -132,29 +132,34 @@ and uses `tools/validation/benchmark_lingbot_world_v2_direct.py`.
 
 ### Current one-minute streaming replay
 
-The one-minute workload was rerun on 2026-08-03 at TeleFuser commit
-`284996dd616cfd44a55523687b7f2a63a281abb9`. It validates sustained target generation, bounded KV-cache capacity,
-and the paced LiveKit delivery path on the current communication-optimized revision.
+The one-minute workload was rerun on 2026-08-06 with the current source tree, four H100 80 GB GPUs, Python
+3.11.13, PyTorch 2.11.0+cu128, CUDA 12.8, BF16 DiT, FP32 VAE, FlashAttention-4, disabled FSDP, and disabled
+`torch.compile`. The source tree includes the tagged Q/K/V Copy Engine Ulysses path.
 
 The run used the `stream_lingbot_world_v2_1min.json` workload and AIPerf 0.11.0 at commit
-`e977ffbb1648510acec431b2a3fbd1a0f7bb8a35`. The 60-second request was truncated to 60 complete latent chunks:
-957 generated frames representing 59.75 seconds of media. With `local_attn_size=18` and `sink_size=6`, the
-240-latent-frame session reported a fixed 28,080-token KV capacity.
+`e977ffbb1648510acec431b2a3fbd1a0f7bb8a35`. Its `delivery_mode=lossless` request uses FIFO backpressure and
+keeps the LiveKit video track open until the AIPerf client confirms the sender declared frame count. The 60-second
+request generated 957 frames across 60 complete latent chunks, representing 59.75 seconds of media. The steady
+summary excludes the first 13-frame chunk; the 240-latent-frame session reported a fixed 28,080-token KV capacity
+with `local_attn_size=18` and `sink_size=6`.
 
 | Metric | Result |
 |---|---:|
+| Successful sessions | 1 / 1 |
 | Generated target frames / chunks | 957 / 60 |
 | Steady frames / chunks after excluding chunk 0 | 944 / 59 |
-| Steady target compute time / FPS | 58.2791 s / **16.1979** |
-| Chunk compute mean / p50 / p90 / p99 / max | 0.9878 / 0.9593 / 1.0624 / 1.0932 / 1.1149 s |
-| LiveKit stream FPS / client frames | 13.1967 / 803 |
-| First client frame / session runtime | 6.0682 / 66.8948 s |
-| Runtime creation | 1.4176 s |
-| Artifact | `20260803_095518_62ec043c` |
-
-The target completed all 60 chunks and cleared the average 16 FPS compute gate. It did not keep every chunk below one
-second: p99 was 1.0932 seconds and the maximum was 1.1149 seconds. The lower client frame count belongs to the paced
-delivery measurement and must not be conflated with target generation completeness.
+| Steady target compute time / FPS | 53.6901 s / **17.5824** |
+| Chunk compute mean / p50 / p90 / p99 / max | 0.9100 / 0.9088 / 0.9177 / 0.9751 / 1.0549 s |
+| LiveKit declared / decoded client frames | 958 / 961 |
+| Client callback FPS after first frame | 15.2313 |
+| First client frame / session runtime | 5.9822 / 79.2523 s |
+| Runtime creation | 1.4279 s |
+| Artifact | `20260806_132517_46419f8f` |
+
+The target completed all 60 chunks and clears the average 16 FPS compute gate. Lossless delivery completed only
+after the client confirmed all 958 declared frames. The three additional client decoder callbacks come from the
+LiveKit startup track and must not be interpreted as generated model frames.
+
 
 ## Reproducibility
 

docs/en/parallel.md

@@ -88,6 +88,15 @@ Input: (B, S_LOCAL, H_GLOBAL, D)
 - Suitable for medium-length sequences
 - Requires number of heads to be divisible by GPU count
 
+When the installed `tf-kernel` wheel contains the Ulysses CUDA IPC operators and every rank in the Ulysses process
+group is on the same host, TeleFuser uses the source-built Copy Engine backend for grouped Q/K/V scatter. It writes
+directly into each peer final-layout target buffer, keeps up to 12 target allocations in a tag/shape/dtype LRU cache,
+and fans out over one high-priority copy stream. Eviction synchronizes participating devices before closing peer
+mappings. Q, K, and V stay as separate submissions so projection compute can overlap
+with communication, while the three transfers share one CUDA stream-memory handshake that does not occupy an SM.
+Single collectives and output gather stay on the faster PyTorch/NCCL path. Multi-host groups, missing kernels,
+and unsupported CUDA IPC configurations also use the PyTorch/NCCL fallback.
+
 ### Ring Attention
 
 Sequence parallelism based on P2P communication:
@@ -145,9 +154,9 @@ Asynchronous All-to-All implementation, overlapping computation and communicatio
 
 ```python
 # Initiate async All-to-All
-q_wait = ulysses_scatter_heads(q, group)
-k_wait = ulysses_scatter_heads(k, group)
-v_wait = ulysses_scatter_heads(v, group)
+q_wait = ulysses_scatter_heads(q, group, tag="q", barrier=False)
+k_wait = ulysses_scatter_heads(k, group, tag="k", barrier=False)
+v_wait = ulysses_scatter_heads(v, group, tag="v")
 
 # Wait for completion
 q = q_wait()

docs/zh/parallel.md

@@ -87,6 +87,15 @@ telefuser/distributed/
 - 适合中等长度序列
 - 需要头数能被 GPU 数整除
 
+当已安装的 `tf-kernel` wheel 包含 Ulysses CUDA IPC 算子，且 Ulysses 进程组内所有 rank 位于同一主机时，
+TeleFuser 会对成组的 Q/K/V scatter 使用源码编译的 Copy Engine 后端。该后端直接写入对端最终布局的
+target buffer，在按 tag/shape/dtype 组织的 LRU 中最多缓存 12 个 target allocation，并使用一条高优先级
+copy stream。淘汰前会同步参与设备，再关闭 peer mapping。Q、K、V 保持独立提交，因此 projection 计算仍
+可与通信重叠；三次传输只共享一次不占用 SM 的 CUDA
+stream-memory 握手。
+单次 collective 和输出 gather 继续使用实测更快的 PyTorch/NCCL 路径。跨主机进程组、缺少算子或 CUDA IPC
+不受支持时也会回退到 PyTorch/NCCL。
+
 ### Ring Attention
 
 基于 P2P 通信的序列并行：
@@ -144,9 +153,9 @@ config = ParallelConfig(
 
 ```python
 # 发起异步 All-to-All
-q_wait = ulysses_scatter_heads(q, group)
-k_wait = ulysses_scatter_heads(k, group)
-v_wait = ulysses_scatter_heads(v, group)
+q_wait = ulysses_scatter_heads(q, group, tag="q", barrier=False)
+k_wait = ulysses_scatter_heads(k, group, tag="k", barrier=False)
+v_wait = ulysses_scatter_heads(v, group, tag="v")
 
 # 等待完成
 q = q_wait()

examples/lingbot/README.md

@@ -33,31 +33,30 @@ these versions before attributing a difference to code changes.
 | GPU | 4 x NVIDIA H100 80 GB HBM3 (SM90) |
 | NVIDIA driver | `590.48.01` |
 | Python | `3.11.13` |
-| PyTorch | `2.11.0+cu130` |
-| PyTorch CUDA runtime | `13.0` |
+| PyTorch | `2.11.0+cu128` |
+| PyTorch CUDA runtime | `12.8` |
 | FlashAttention 4 | `flash-attn-4==4.0.0b19` |
 | CUTLASS DSL | `nvidia-cutlass-dsl==4.6.0` |
 | CUDA Python | `cuda-python==13.3.1` |
 
-Create an isolated Python 3.11 environment and install the CUDA 13.0 PyTorch build from the wheel index used by
+Create an isolated Python 3.11 environment and install the CUDA 12.8 PyTorch build from the wheel index used by
 your deployment. Install PyTorch before TeleFuser so optional CUDA packages resolve against the intended ABI:
 
 ```bash
 python3.11 -m venv .venv-lingbot
 source .venv-lingbot/bin/activate
 python -m pip install --upgrade pip setuptools wheel
 
-# Install torch==2.11.0+cu130 from your CUDA 13.0 PyTorch wheel index first.
+# Install torch==2.11.0+cu128 from your CUDA 12.8 PyTorch wheel index first.
 python -m pip install -e ".[dev]"
 python -m pip install \
-  "flash-attn-4[cu13]==4.0.0b19" \
+  "flash-attn-4==4.0.0b19" \
   "nvidia-cutlass-dsl==4.6.0" \
   "cuda-python==13.3.1"
 ```
 
-The `cu13` extra installs FA4's CUDA 13 dependency variant. For a CUDA 12.8 PyTorch environment, install
-`flash-attn-4==4.0.0b19` without that extra and use matching CUDA 12.x dependencies; do not mix cu128 and cu130
-interpreters in one distributed run.
+The current AIPerf validation uses the CUDA 12.8 PyTorch build above. For a CUDA 13.0 PyTorch environment, install
+`flash-attn-4[cu13]==4.0.0b19` and matching CUDA 13 dependencies; do not mix cu128 and cu130 interpreters in one distributed run.
 
 Verify both the package versions and TeleFuser's runtime backend selection before benchmarking:
 
@@ -127,23 +126,22 @@ python examples/lingbot/lingbot_world_v2_image_to_video_h100.py \
 
 ### Validated Four-H100 Real-Time Gate
 
-Commit `540b579` was validated on 2026-08-03 with four H100 80 GB GPUs, PyTorch 2.11.0+cu128,
+The direct pipeline-service path was validated with four H100 80 GB GPUs, PyTorch 2.11.0+cu128,
 FlashAttention-4, BF16 DiT, FP32 VAE, disabled FSDP, and disabled `torch.compile`. The default 832x480
 request generated all 77 frames in five chunks at a 16 FPS playback target.
 
-| Metric | Result |
-| --- | ---: |
-| Steady compute FPS | **17.14** |
-| Steady chunk mean / p50 / p90 | 0.9335 / 0.9409 / 0.9410 s |
-| Slowest steady chunk | 1.0058 s |
-| Generated frames / chunks | 77 / 5 |
+| Revision / communication path | Steady compute FPS | Steady chunk mean / p50 / p90 | Slowest steady chunk | Generated frames / chunks |
+| --- | ---: | ---: | ---: | ---: |
+| `540b579` (2026-08-03) | 17.14 | 0.9335 / 0.9409 / 0.9410 s | 1.0058 s | 77 / 5 |
+| Current source, tagged Q/K/V Copy Engine Ulysses (2026-08-06) | **19.08** | **0.8385 / 0.8415 / 0.8706 s** | **0.9040 s** | 77 / 5 |
 
 The steady summary excludes chunk 0 and covers four 16-frame chunks. `compute_seconds` synchronizes all target CUDA
 devices and includes condition handling, DiT, clean-KV update, spatial VAE decode, GPU-to-CPU transfer, and frame
 conversion. It excludes model loading, runtime creation, LiveKit pacing/encoding, network delivery, and client
-rendering. The average therefore clears the 16 FPS target-side real-time gate, while the slowest chunk exceeds its
-one-second budget by 5.8 ms; treat this as a validated configuration, not a guarantee for other hardware, resolutions,
-durations, concurrent sessions, or transport conditions.
+rendering. The current source measurement used the local SM90 `tf-kernel` build with the CUDA IPC Copy Engine backend;
+its V transfer is issued before Q projection, Q before K projection, and K is the completion barrier. The current
+result clears the 16 FPS target-side real-time gate, but remains a point measurement rather than a guarantee for other
+hardware, resolutions, durations, concurrent sessions, or transport conditions.
 
 Reproduce the measured direct pipeline-service path without LiveKit or codec time:
 
@@ -162,6 +160,33 @@ The offline CLI was also validated to produce an H.264 832x480 video containing
 [AIPerf benchmark guide](../../docs/en/benchmark_aiperf.md) for the one-minute workload, client delivery metrics,
 and comparisons that require identical environments.
 
+### Validated One-Minute AIPerf Replay
+
+The standard `stream_lingbot_world_v2_1min.json` workload was run on 2026-08-06 with the current source tree,
+AIPerf 0.11.0 at `e977ffbb1648510acec431b2a3fbd1a0f7bb8a35`, four H100 80 GB GPUs, and the tagged Q/K/V
+Copy Engine Ulysses path. The workload requests `delivery_mode=lossless`: target output uses FIFO backpressure,
+the sender declares its published-frame count, and the client confirms receipt before the LiveKit track closes. It
+requests 957 generated frames across 60 chunks (59.75 seconds of media), excludes chunk 0, and retains the
+28,080-token KV capacity from `local_attn_size=18` and `sink_size=6`.
+
+| Metric | AIPerf result |
+| --- | ---: |
+| Successful sessions | 1 / 1 |
+| Generated target frames / chunks | 957 / 60 |
+| Steady target frames / chunks after warmup | 944 / 59 |
+| Steady target compute time / FPS | 53.6901 s / **17.5824** |
+| Chunk compute mean / p50 / p90 / p99 / max | 0.9100 / 0.9088 / 0.9177 / 0.9751 / 1.0549 s |
+| LiveKit declared / decoded client frames | 958 / 961 |
+| Client callback FPS after first frame | 15.2313 |
+| First client frame / session runtime | 5.9822 / 79.2523 s |
+| Runtime creation | 1.4279 s |
+| Artifact | `20260806_132517_46419f8f` |
+
+The target-side compute rate clears the average 16 FPS gate. Lossless delivery completed after the client confirmed
+all 958 declared frames; the three additional decoder callbacks are transport-level callbacks and are separate
+from the 957 generated target frames.
+
+
 ## Usage
 
 ### Four H100 GPUs