[Core] Refactor omni prefix cache into Manager/Controller

docs/design/feature/omni_async_output_materialization.md

@@ -318,18 +318,18 @@ The same behavior applies to the Base and VoiceDesign Qwen3-TTS checkpoints
 because they use the same Talker implementation.
 
 !!! warning
-    Do not pass `--no-async-chunk` or enable prefix caching when you want this
-    optimization. Either change causes the runner to fall back to synchronous
-    output construction. Enabling async chunk alone cannot activate the feature
-    for a model that has not opted into async Omni output.
+    Do not pass `--no-async-chunk` when you want this optimization. Disabling
+    async chunk causes the runner to fall back to synchronous output
+    construction. Enabling async chunk alone cannot activate the feature for a
+    model that has not opted into async Omni output.
 
 !!! note "Prefix cache compatibility"
-    Async Omni output materialization and Omni prefix caching cannot currently
-    run together. `_should_use_async_omni_output()` returns `False` whenever
-    `self.omni_prefix_cache` is present, so prefix caching selects synchronous
-    output materialization. Supporting both features requires snapshotting or
-    otherwise synchronizing prefix-cache merge and update state before the
-    background output path can consume it safely.
+    Prefix cache and async Omni output can run together. `save_outputs` returns
+    a `step_id`; the output builder later calls `materialize(sid, req_ids)` with
+    the save-time request list, which may be a step late relative to the live
+    `input_batch`. `_should_use_async_omni_output()` does not disable itself
+    when the cache is present. See
+    [Automatic Prefix Caching](prefix_caching.md#implementation).
 
 ## Compatibility and Fallbacks
 
@@ -341,10 +341,9 @@ runtime conditions hold:
 | AR async scheduling is enabled | The optimization relies on the scheduler advancing while the prior output is materialized |
 | `async_chunk` is enabled | The feature targets incremental downstream Omni payloads |
 | The model stage opts in with `use_async_omni_output` | Models must declare that their output lifecycle is safe to defer |
-| Omni prefix cache is disabled | Prefix-cache merge and update ordering currently requires synchronous materialization |
 | Speculative decoding is disabled | Speculative output state is not included in this deferred path |
 | Routed-expert output is disabled | Routed-expert extraction currently requires the synchronous path |
-| Postprocess is absent or explicitly runs eagerly | State needed by the next decode step must be updated before the runner returns |
+| Postprocess is absent or explicitly runs eagerly | State needed by the next decode step must be updated before the runner returns. Eager postprocess sees the scheduled slice (`None, None` combined tensors), not the prefix-cache-merged full prompt; it must only consume the tail |
 
 These checks are evaluated per stage on runners based on `GPUARModelRunner`.
 An unsupported combination does not prevent serving; it only falls back to
@@ -380,6 +379,8 @@ so Omni payload materialization remains synchronous.
 - `tests/worker/test_gpu_ar_model_runner.py`: Snapshot, guard, connector
   ordering, and background error-propagation tests.
 - [Async Chunk](async_chunk.md): Inter-stage chunking and scheduling design.
+- [Automatic Prefix Caching](prefix_caching.md#implementation): Prefix-cache
+  `step_id` / `materialize` contract used by the deferred output builder.
 - [Qwen3-Omni optimization blog](https://vllm.ai/blog/2026-07-01-qwen3-omni-optimization):
   Optimization context and controlled performance results.
 - [PR #4476](https://github.com/vllm-project/vllm-omni/pull/4476): Feature

docs/design/feature/prefix_caching.md

@@ -9,6 +9,8 @@
 - [High-Level Approach](#high-level-approach)
 - [Example](#example)
 - [What About Multimodal Inputs?](#what-about-multimodal-inputs)
+- [Implementation](#implementation)
+- [Related Files](#related-files)
 
 ---
 
@@ -162,3 +164,53 @@ Because we have multimodal data in a scheduled span that isn't fully precomputed
 When we pass our multimodal tensors to the language model component in the same stage, we'll then expect the same outputs, because the prefix caching behaviors in vLLM-Omni / vLLM match, so the LLM will use vLLM's KV cache manager's prefix caching to correctly handle the attention information for `Block 1` while calculating the outputs for `Block 2`, giving us the correct results for processing `Block 2` with the context of `Block 1`.
 
 Finally, we look up the output hidden states/multimodal tensors corresponding to the prefix cache hit `Block 1` and concatenate it with the forward pass result to get the final result, which is expected to be identical to the full hidden states when prefix caching is disabled.
+
+### Implementation
+
+The block/slot model is `vllm_omni/core/prefix_cache/`.
+`OmniPrefixCacheManager` owns `(slot, key)` occupancy, hit spans, and merge.
+`OmniPrefixCacheController` moves data: a reusable `StagingBufferPool` for
+this step's D2H, and scatter into the durable `PrefixBlockPool`. The state
+lock covers those tables only.
+
+Miss is not an error (this step's forward slice only). A hit span that
+resolves to absent slots is fatal. Abort still writes: once a hash entered
+this step's batch it must land in the cache.
+
+Schedule is a key split: hidden and non-deferred mm use `JOIN_NEXT_STEP`
+(D2H at save, join at the next save); `deferred_keys` use `JOIN_ON_FINISH`
+(committer copies the GPU freeze on finish/abort, or earlier under cap
+pressure).
+
+Hit reads follow the same split. A `JOIN_NEXT_STEP` in-transit span waits
+the owner's `done` (scatter, not just D2H), drains so occupancy flips to
+committed, then reads the pool. A `JOIN_ON_FINISH` in-transit span still
+uses `fetch_host` on the GPU freeze so a concurrent same-prefix hit does
+not force the whole deferred payload to disk. Staging-slot reader holders
+are gone: once scattered, pool rows persist across slot reuse.
+
+```python
+cache.register_policy(ModelCachePolicy.from_model(model))   # load_model
+cache.new_step_starts(scheduler_output)   # before _update_states
+sid = cache.save_outputs(hidden, mm_flat, num_tokens_unpadded=n,
+                         num_tokens_padded=n_pad)
+outs = cache.materialize(sid, req_ids)    # or discard_step(sid)
+```
+
+Each `sid` is consumed exactly once. `req_ids` must be a subset of the save
+snapshot. `materialize` may run on the async output builder after the engine
+has entered the next step; leftover mm (deferred tails and uncached
+passthrough) is copied to CPU at `save_outputs` so the builder never reads
+live CUDA-graph buffers. See
+[Async Omni Output Materialization](omni_async_output_materialization.md).
+
+The cache is constructed only on the last pipeline-parallel rank
+(`_ensure_omni_prefix_cache`). Other ranks skip it: they never call
+`save_outputs`, so a hit table there would resolve to absent slots.
+
+### Related Files
+
+- `vllm_omni/core/prefix_cache/`
+- `vllm_omni/worker/gpu_model_runner.py` (`_ensure_omni_prefix_cache`)
+- `tests/core/test_prefix_cache.py`
+- [Async Omni Output Materialization](omni_async_output_materialization.md)