InstrumentedScheduler builds two SchedulerOutputs from scratch during
DYN_BENCHMARK_MODE=decode: the synthetic decode batch from
_bench_inject_fake_decode and the empty drain frame in schedule().
Neither populated kv_connector_metadata.

When a KV connector is configured (e.g. NixlConnector for any disagg
worker), vLLM's worker-side _get_kv_connector_output asserts
"scheduler_output.kv_connector_metadata is not None" before calling
bind_connector_metadata. The synthetic outputs failed this assertion
on the first decode-sweep iteration, killing EngineCore before the
worker could register and the planner could ever read get_perf_metrics.

Mirror the parent scheduler (vllm/v1/core/sched/scheduler.py) and call
connector.build_connector_meta(output) on both benchmark-built outputs.
Our fake decode reqs have KV pre-allocated locally via allocate_slots,
so build_connector_meta produces a no-op metadata: no transfers
planned, just a non-None object the worker contract requires.

End-to-end repro (Qwen3-0.6B, RTX 6000 Ada, vllm 0.20):

  python -m dynamo.vllm \
    --model Qwen/Qwen3-0.6B \
    --disaggregation-mode decode \
    --kv-transfer-config '{"kv_connector":"NixlConnector","kv_role":"kv_both"}' \
    --benchmark-mode decode

Before: EngineCore fatal at gpu_model_runner execute_model on the
first benchmark decode batch. After: sweep advances past the connector
contract; the previous failure mode is gone.

Add three unit tests covering both connector-set and connector-absent
paths via the schedule() empty-frame branch (the _bench_inject_fake_decode
path uses the same pattern and is covered by symmetry). Tests fail on
the unpatched scheduler.

Signed-off-by: Hongkuan Zhou <hongkuanz@nvidia.com>
Signed-off-by: hongkuanz <hongkuanz@nvidia.com>

…nchmark

Pure unit tests on _bench_inject_fake_decode and the empty-frame
schedule branch can't catch the two recently-fixed regressions in
isolation -- both only manifest when the model actually executes on a
GPU:

* kv_connector_metadata must be attached to benchmark-built
  SchedulerOutputs, otherwise the worker-side
  _get_kv_connector_output assertion fires (any disagg config / any
  worker with a KV connector).
* The synthetic decode prompt must be padded to ctx_len + 1 so the
  async-scheduler's -1 placeholder write doesn't pollute the read
  slot the next batch reuses (CUDA OOB on the second decode point,
  first batch>1 sweep).

Add a single-GPU end-to-end test that spawns python -m dynamo.vllm
with --benchmark-mode {agg,decode}, waits for the benchmark JSON
to be written, validates structure (right mode, every point produces
at least one FPM with positive wall_time), and tears down. Two
scenarios:

  * test_self_benchmark_agg_mode -- aggregated worker, no
    kv-transfer-config; covers the prompt-padding regression on the
    common-case path.
  * test_self_benchmark_disagg_decode_with_nixl_connector -- disagg
    decode + NixlConnector kv_both; this is the original
    user-reported configuration and exercises both fixes together.

Both runs are tight (granularity=2, max-model-len=1024,
gpu-memory-utilization=0.4 on Qwen3-0.6B) so each completes in ~30s
of GPU time inside the 45-minute pre-merge slot.

Markers (vllm, gpu_1, e2e, pre_merge, model(QWEN)) match the
selector pr.yaml uses for the vllm-test GPU stage:

    gpu_test_markers: pre_merge and vllm and gpu_1

so the tests gate every PR. Verified the selector picks them up via
``pytest --collect-only -m "pre_merge and vllm and gpu_1"``.

Signed-off-by: Hongkuan Zhou <hongkuanz@nvidia.com>
Signed-off-by: hongkuanz <hongkuanz@nvidia.com>

Extend the real-GPU pre-merge gate so it also verifies the worker can
serve normal traffic after self-benchmarking. Replaces the earlier
"start worker just for the benchmark" approach with the production
flow: worker startup runs the benchmark, registers with the runtime,
then serves a real chat completion.

Setup mirrors tests/fault_tolerance/cancellation/test_vllm.py for CI
parity:

* Same model: FAULT_TOLERANCE_MODEL_NAME (Qwen/Qwen3-0.6B).
* Same DynamoFrontendProcess + ManagedProcess worker layout, same
  health-check URLs, same env (DYN_HEALTH_CHECK_ENABLED=false,
  DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS, DYN_SYSTEM_PORT,
  DYN_HTTP_PORT).
* Same NixlConnector kv_both wiring and same prefill-side
  --kv-events-config + VLLM_NIXL_SIDE_CHANNEL_PORT for the disagg
  case.
* Same fixtures: runtime_services_dynamic_ports, predownload_models,
  set_ucx_tls_no_mm (for the disagg pair).
* Same per-worker allocate_port(9100) / deallocate_port pattern.

Two scenarios:

* test_self_benchmark_agg_serves_after_bench (gpu_1):
    aggregated worker, --benchmark-mode agg. Worker startup runs the
    full agg sweep (prefill + decode points, decode includes
    batch>1), then serves a chat completion.

* test_self_benchmark_disagg_serves_after_bench (gpu_2):
    prefill worker --benchmark-mode prefill plus decode worker
    --benchmark-mode decode kv_both. Both workers run their sweeps,
    register, and together serve a chat completion that traverses
    both engines.

Each test, after both workers' health checks pass:

1. Validates the per-worker benchmark JSON: right mode, every
   benchmark point produced at least one FPM with positive
   wall_time -- catches a worker that registered but ran a degenerate
   sweep.
2. Asserts the decode sweep covers batch>1 (the prompt-padding
   regression case) so a future regression that only affects batch>1
   doesn't slip through.
3. Sends a real chat completion via tests/utils/client.send_request
   and asserts a non-empty assistant content.

Marker selectors verified locally:

* pre_merge and vllm and gpu_1 -> agg test
  (matches pr.yaml's vllm-test gpu_test_markers)
* pre_merge and vllm and (gpu_2 or gpu_4) -> disagg test
  (matches pr.yaml's vllm-multi-gpu-test gpu_test_markers)

so both gate every PR.

Signed-off-by: Hongkuan Zhou <hongkuanz@nvidia.com>
Signed-off-by: hongkuanz <hongkuanz@nvidia.com>

Three changes addressing PR review:

1. (dynamo-ops) ``_bench_generate_decode_grid`` previously sized
   ``max_batch`` from raw ``ctx_len`` token count
   (``total_kv_tokens // ctx_len``). After the prompt-padding fix
   ``_bench_inject_fake_decode`` actually requests ``ctx_len + 1``
   tokens, which the KV cache manager rounds UP to the next block
   boundary. So at boundary ctx_lens (notably ctx_len that is a
   multiple of block_size) the advertised ``max_batch`` was
   2x what the allocator could fit, and the allocator silently
   truncated the batch (logged as "KV exhausted at ctx_len=...,
   truncating batch") -- the benchmark recorded the partial result
   under the wrong batch size.

   Fix: size ``max_batch`` from blocks-per-request derived from the
   padded length:

       blocks_per_req = ceil((ctx_len + 1) / block_size)
       max_batch = num_gpu_blocks // blocks_per_req

   Add two unit tests that fail on the unpatched grid sizer (e.g. at
   block_size=16 / num_gpu_blocks=100 / ctx_len=16 the old code
   advertised batch_size=100, which needs 200 blocks).

2. (CodeRabbit) Replace ``getattr(self, "connector", None)`` and
   ``getattr(self, "ec_connector", None)`` guards with direct
   attribute access (``self.connector is not None``,
   ``self.ec_connector is not None``). Both attributes are part of
   the parent vLLM ``Scheduler``'s public contract; using
   ``getattr`` with a default would silently mask a future vLLM bump
   that drops them. Direct access makes such a contract violation
   fail loudly. Same fix at both call sites (the empty-frame
   schedule branch and ``_bench_inject_fake_decode``).

3. (CodeRabbit) Hoist ``from unittest.mock import MagicMock`` and
   ``from dynamo.vllm.instrumented_scheduler import _BenchPhase`` to
   the module-level imports in
   ``test_vllm_instrumented_scheduler.py``, matching the project's
   python-guidelines (all imports at module top).

Local pytest: 21/21 unit tests pass; the two new grid-sizing tests
verified to fail on the unpatched scheduler.

Signed-off-by: Hongkuan Zhou <hongkuanz@nvidia.com>
Signed-off-by: hongkuanz <hongkuanz@nvidia.com>

…rop shebang

Three CI failures on PR #9360 addressed:

1. ``check-shebang-scripts-are-executable`` (pre-commit):
   ``tests/vllm/test_self_benchmark_gpu.py`` carried a ``#!/usr/bin/env
   python3`` shebang but isn't intended to be directly executed (pytest
   collects it). Drop the shebang to satisfy the hook without granting
   exec bits.

2. ``pytest-marker-report`` (pre-commit) -- 9 collection errors:
   ``tests/vllm/__init__.py`` makes ``tests/vllm`` a Python package
   that shadows the real ``vllm`` distribution when ``tests`` is on
   sys.path during marker discovery. Every other vLLM-suite test that
   does ``from vllm import LLM`` (or similar) broke with
   ``ImportError: cannot import name 'LLM' from 'vllm'`` pointing at
   our ``tests/vllm/__init__.py``. Renamed to
   ``tests/vllm_self_benchmark/`` to remove the name collision; all 9
   shadowed imports resolve to the real ``vllm`` again.

3. ``vllm-runtime / Multi-GPU Test`` -- disagg test failed with:
       (EngineCore pid=...) zmq.error.ZMQError: Address already in use
                                  (addr='tcp://*:20380')
   ``InstrumentedScheduler`` (auto-injected by ``--benchmark-mode``)
   binds the FPM publisher to ``tcp://*:DYN_FORWARDPASS_METRIC_PORT +
   dp_rank``, defaulting to 20380. Two workers on the same host (the
   disagg prefill + decode pair) both try to bind 20380 and the second
   loses. The operator works around this by injecting
   ``DYN_FORWARDPASS_METRIC_PORT`` per engine
   (deploy/operator/internal/dynamo/component_worker.go:111); the
   single-GPU agg test happens to avoid the collision because only
   one worker runs. Apply the same fix in the test: allocate a
   per-worker port via ``allocate_port(20380)`` and inject it as the
   env var, with matching ``deallocate_port`` on teardown.

   (The cancellation/migration tests don't hit this because they
   don't pass ``--benchmark-mode`` and so don't get
   InstrumentedScheduler.)

Verified locally:
* ``import vllm`` resolves to the real site-packages vllm again.
* pytest collection of the renamed test passes; marker selectors
  ``pre_merge and vllm and gpu_1`` (agg test) and
  ``pre_merge and vllm and (gpu_2 or gpu_4)`` (disagg test) still
  match -- so both stay wired to ``pr.yaml``'s vllm-test and
  vllm-multi-gpu-test GPU stages and continue to gate the PR.

The single-GPU agg test passed in the prior CI run (only the disagg
test failed) so issue #3 is specifically what was needed for disagg
to pass.

Signed-off-by: Hongkuan Zhou <hongkuanz@nvidia.com>
Signed-off-by: hongkuanz <hongkuanz@nvidia.com>