[TRTLLM-12648][test] implement disagg cancellation injector thread

[chienchunhung]

Summary by CodeRabbit

• New Features

  • Stress testing now supports YAML-configured injection schedules for triggering signal actions (pause/resume/terminate) on tracked workers, with automatic respawning and health verification for terminated workers.

• Documentation

  • Updated test execution guidance and configuration workflows.

• Tests

  • Added comprehensive unit and integration test coverage for injection scheduling and worker state transitions.

Summary

Third thread body in the disaggregated cancellation stress-test suite — the marathon-style (hours-running) harness that gates regressions of the bug class fixed by PR #13713. The skeleton + log_scanner_thread landed in PR #14375, and the metrics_thread in PR #14807; this PR adds the injector_thread.

This PR ships:

• injector_thread body — reads the stress_config.injections schedule, waits until each entry's at_min offset from marathon start, and fires the failure injection against the tracked workers. Each fired event (with at_min, elapsed_s, target, role, index, signal outcomes) is appended to _injection_events for the end-of-marathon injection-count / recovery assertions. Only the log scanner is fail-fast; the injector trips fail-fast solely on a respawn-health timeout.
• _parse_injection_schedule — validates and normalizes the YAML injections: list into sorted _InjectionSpec entries; malformed entries are logged at ERROR and skipped (a typo in one slot can't abort the marathon). Enforces that sigstop requires duration_s, and only sigstop / sigkill types are accepted.
• _resolve_injection_target — maps gen_worker_random, ctx_worker_random, and indexed {ctx,gen}_worker_<N> targets to a tracked worker; unknown / unmatched targets raise and are skipped without fail-fast.
• _execute_sigstop_pause — SIGSTOP → bounded pause → SIGCONT. The pause is interruptible by stop_event / failed_event, and SIGCONT is always sent in a finally so a worker is never left stopped during teardown.
• _execute_sigkill + _respawn_tracked_worker — SIGKILL the worker, then optionally relaunch it on a freshly allocated port via disagg_test_utils._run_worker, with a bounded /health poll (_wait_for_worker_health). Respawn launch errors and health timeouts return False → fail-fast.
• WorkerLaunchSpec + bind_tracked_workers — shadow-tracked launch context recorded at cluster setup so the injector can relaunch a SIGKILLed worker without modifying the shared ProcessWrapper infrastructure.
• injector_poll_interval_s — new ctor parameter on DisaggCancellationStressHarness, matching the existing log_scanner_poll_interval_s / metrics_scrape_interval_s pattern. Tests pass small values for fast turnaround; the marathon uses defaults.
  The thread sits inside the lifecycle wired up by PR [TRTLLM-12648][test] add disagg cancellation stress-test harness skeleton #14375 — start() already spawns it, stop() joins it, collect_results() already returns injection_events. Previous PRs left it as a no-op stub;
  this PR makes it actually inject.

PR chain plan

PR	Adds	Marathon assertion it enables
PR #14375	skeleton + log_scanner	hard-zero log patterns
PR #14807	metrics_thread	KV-cache utilization growth bound
this PR	injector_thread	SIGSTOP / SIGCONT / SIGKILL+respawn; injection_events count
next	canary_thread	canary error rate + token-equivalence + recovery time
next + 1	load_thread + real setup() with save_log=True	sustained load over duration_min; marathon entry point becomes a real marathon

The final PR flips setup_disagg_cluster to launch workers with explicit per-worker ports + save_log=True and calls bind_tracked_workers() — at which point the injector acts on real worker subprocesses. Today it correctly no-ops when its inputs are absent: with the cluster launch still stubbed, _tracked_workers is empty and the thread logs a warning and exits.

Tests added

All deterministic, GPU-free, second-scale. Under tests/integration/defs/stress_test/disagg_cancel/.

• test_injector.py — unit tests across three layers:
  • Parser / resolver: schedule sort + validation (skips malformed / unsupported entries), fixed-index target, random target drawn from the role pool, unknown target raises.
  • Signal helpers (real subprocess): SIGSTOP pause stops-then-resumes a live child, SIGKILL terminates, SIGSTOP on an already-dead process is skipped gracefully.
  • Thread-body integration: immediate SIGSTOP fires and records an event, two scheduled events run in order, SIGKILL respawn-timeout trips fail-fast, SIGKILL respawn-success records respawned=True, stop_event exits before a distant injection, no-tracked-workers / empty-schedule exit cleanly, unknown target is skipped without fail-fast, and respawn uses the allocated port with a bounded health wait (isolated via a fake disagg_test_utils module so no GPU / venv is required).

Run locally (no GPU, no cluster):

```bash
PYTHONPATH=tests/integration/defs:tests/integration/defs/disaggregated
python3 -m pytest -c /dev/null -o addopts=
--confcutdir=tests/integration/defs/stress_test
tests/integration/defs/stress_test/disagg_cancel/test_injector.py -v
```

Out of scope (deferred to later PRs in this chain)

• YAML schema wiring of injector_poll_interval_s (constructor-only for now; the marathon uses defaults).
• The remaining two thread bodies: canary, load.
• Real setup_disagg_cluster invocation + bind_tracked_workers() (cluster launch is still a stub, so _tracked_workers is empty in the lifecycle smoke; the injector logs a warning and exits in that case).
• Disagg-server re-registration of a respawned worker — the respawn path confirms the worker's own /health on a fresh port, but does not yet assert the disagg server re-registers it into the routable pool (design open question Kaiyu/update main #5). This is validated when setup() + load / canary land.
• Registration in qa/llm_function_stress.txt for nightly / weekly CI.

PR Checklist

Please review the following before submitting your PR:

• PR description clearly explains what and why. If using CodeRabbit's summary, please make sure it makes sense.

• PR Follows TRT-LLM CODING GUIDELINES to the best of your knowledge.

• Test cases are provided for new code paths (see test instructions)

• If PR introduces API changes, an appropriate PR label is added - either api-compatible or api-breaking. For api-breaking, include BREAKING in the PR title.

• Any new dependencies have been scanned for license and vulnerabilities

• CODEOWNERS updated if ownership changes

• Documentation updated as needed

• Update tava architecture diagram if there is a significant design change in PR.

• The reviewers assigned automatically/manually are appropriate for the PR.

• Please check this after reviewing the above items as appropriate for this PR.

GitHub Bot Help

To see a list of available CI bot commands, please comment /bot help.

[chienchunhung]

/bot run --disable-fail-fast

[tensorrt-cicd]

PR_Github #51922 [ run ] triggered by Bot. Commit: 7c1a702 Link to invocation

[coderabbitai]

📝 Walkthrough

Walkthrough

This PR implements a YAML-driven stress injection system for the disaggregated cancellation harness, enabling scheduled signal injection (SIGSTOP, SIGKILL) into worker processes with optional respawn and health verification. Updates include harness API extensions, comprehensive test coverage, and documentation reflecting the incremental implementation progress.

Changes

Injector Stress System

Layer / File(s)	Summary
Documentation and Runbook tests/integration/defs/stress_test/disagg_cancel/README.md	Status section rewritten to reflect partially implemented harness state with log scanner and injector threads. File layout expanded to include new unit test modules. New "Unit tests (no GPU, no cluster)" section added with concrete pytest commands for component testing and marathon YAML validation. Local marathon instructions updated to reflect current setup() behavior.
Injection Infrastructure: Dataclasses and Imports tests/integration/defs/stress_test/disagg_cancel/harness.py	Added imports for signal handling (os, signal), subprocess coordination, and filesystem utilities. Introduced _TrackedWorker dataclass linking a worker launch spec to its live ProcessWrapper, and _InjectionSpec dataclass capturing timing, signal type, target, pause duration, and optional respawn parameters.
Injection Helpers: Schedule Parsing and Signal Execution tests/integration/defs/stress_test/disagg_cancel/harness.py	Implemented _parse_injection_schedule for YAML normalization and validation with sorting by at_min. Added _resolve_injection_target for deterministic index-based or random role-pool target selection. Implemented _worker_process_pid for PID lookup and _signal_process for resilient signal sending. Added _sigstop_worker, _sigcont_worker, and _sigkill_worker executors with graceful "no process" skipping and optional post-kill wait.
Harness API: Constructor, Initialization, and Worker Registration tests/integration/defs/stress_test/disagg_cancel/harness.py	Extended __init__ with injector_poll_interval_s parameter (default 1.0s). Added public bind_tracked_workers(ctx_workers, gen_workers, ctx_specs, gen_specs) method enforcing strict 1:1 alignment via zip(..., strict=True). Initialized tracking fields _cluster, _worker_specs, _tracked_workers, and _marathon_start_monotonic. Updated start() to record harness start time and adjust thread startup logging.
Injector Thread: Scheduler Loop and Respawn Support tests/integration/defs/stress_test/disagg_cancel/harness.py	Replaced injector stub with full scheduler: parses stress_config.injections, polls until each injection's scheduled at_min, resolves targets, executes signal actions, optionally triggers respawn for sigkill entries with respawn_within_s, records structured events, and exits on stop/fail-fast. Implemented _respawn_tracked_worker for respawn via shadow WorkerLaunchSpec with free port allocation and health timeout. Implemented _wait_for_worker_health for HTTP /health polling via requests.get with early abort on stop/fail conditions. Added _record_injection_event for thread-local event tracking.
Injector Tests: Unit and Integration Coverage tests/integration/defs/stress_test/disagg_cancel/test_injector.py	New test module with fixtures for creating worker specs, subprocess wrappers, and YAML schedules. Unit tests validate _parse_injection_schedule sorting/validation and _resolve_injection_target with fixed indices and random pools. Signal helper tests verify SIGSTOP pause/resume, SIGKILL termination, and graceful handling of already-dead processes. Integration tests wire harness to temporary YAML schedules, validating: immediate SIGSTOP firing, two scheduled events, SIGKILL fail-fast with respawn failures, successful SIGKILL respawn (monkeypatched), respawn port allocation and health polling, early injector exit on stop event, exit without events when no workers tracked, exit on empty schedule, and graceful skipping of unknown targets without fail-fast.
Log Scanner Tests: Keyword Argument Migration tests/integration/defs/stress_test/disagg_cancel/test_log_scanner.py	Updated all make_spec(...) invocations to use log_path= keyword argument instead of positional third argument across the shared harness_with_two_workers fixture and multiple test functions (test_specs_with_none_log_path_skip_gracefully, test_no_hard_zero_patterns_exits_immediately, test_invalid_regex_is_skipped_with_warning, test_log_source_poll_returns_false_when_file_absent, test_log_source_poll_returns_false_on_empty_read, test_log_source_close_is_idempotent). No test logic or assertions changed.

Sequence Diagram

sequenceDiagram
  participant Harness
  participant InjectorThread as Injector Thread
  participant Worker as Worker Process
  participant Health as Health Check
  
  Harness->>Harness: parse_injection_schedule()
  Harness->>InjectorThread: start thread
  
  loop for each injection
    InjectorThread->>InjectorThread: wait until scheduled at_min
    InjectorThread->>InjectorThread: resolve_injection_target()
    InjectorThread->>Worker: signal_process(SIGSTOP or SIGKILL)
    Worker-->>InjectorThread: signal delivered
    
    alt SIGKILL with respawn_within_s
      Worker-->>InjectorThread: process exits
      InjectorThread->>InjectorThread: _respawn_tracked_worker()
      InjectorThread->>Health: _wait_for_worker_health()
      alt health check succeeds
        Health-->>InjectorThread: HTTP 200
        InjectorThread->>Harness: record_injection_event(respawned=True)
      else health check timeout
        Health-->>InjectorThread: timeout
        InjectorThread->>Harness: fail_event set, respawned=False
      end
    else SIGSTOP only
      InjectorThread->>Harness: record_injection_event(skipped=False)
    end
  end
  
  InjectorThread-->>Harness: thread exits on stop/fail

Loading

🎯 3 (Moderate) | ⏱️ ~25 minutes

🚥 Pre-merge checks | ✅ 3 | ❌ 2

❌ Failed checks (1 warning, 1 inconclusive)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 43.75% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.
Description check	❓ Inconclusive	The PR description is comprehensive and well-structured, covering the rationale, changes shipped, test coverage, and out-of-scope items. However, it does not strictly follow the provided template structure (missing explicit 'Description' and 'Test Coverage' sections with the template's exact formatting).	Consider restructuring to explicitly map content to template sections: move 'Summary' content into 'Description', consolidate test details into the 'Test Coverage' section, and ensure 'PR Checklist' items are verified before submission.

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly identifies the main change: implementation of the injector thread for the disaggregated cancellation stress-test harness, with appropriate ticket reference and type classification.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 3

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@tests/integration/defs/stress_test/disagg_cancel/harness.py`:
- Around line 1055-1058: The bug: updating spec.log_path after respawning a
worker leaves the previously snapshotted _LogSource.path objects (used by
_log_scanner_thread_body()) pointing at the old file, so the scanner misses the
new worker's logs. Fix by updating the live log-source objects when you replace
tracked.wrapper: after setting tracked.wrapper = new_wrapper and spec.log_path =
new_wrapper.log_path, locate the worker's existing _LogSource instances (the
collection created at startup that _log_scanner_thread_body() uses) and update
their .path attributes to spec.log_path (or recreate/synchronize those
_LogSource objects) so the scanner tails the new file; ensure this update logic
is invoked in _run_worker() or the respawn code path that handles
tracked.wrapper changes.
- Around line 1055-1068: The respawn health check is always probing localhost
instead of the actual worker host; update calls to _wait_for_worker_health to
use the respawned worker's host (tracked.host or spec.host from the
WorkerLaunchSpec) along with new_wrapper.port when building the health probe
endpoint (replace any hardcoded 'localhost' usage). Make the same change for the
second occurrence around the 1070-1085 block so both respawn paths call
_wait_for_worker_health(new_wrapper.port, host=tracked.host or spec.host,
timeout_s=timeout_s) or the equivalent parameters expected by
_wait_for_worker_health.
- Around line 357-389: In _parse_injection_schedule(), after converting
at_min/duration_s/respawn_within_s to floats, validate their signs and skip
malformed entries: reject entries with at_min < 0 (log and continue), reject
sigstop entries with duration_s < 0 (log and continue), and reject sigkill
respawn_within_s values that are <= 0 (log and continue); update the existing
logger.error calls to reflect these specific validation failures and ensure the
function continues past the bad entry when these checks fail.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Enterprise

Run ID: 5630068d-0f67-4f56-89c5-88558191bb2f

📥 Commits

Reviewing files that changed from the base of the PR and between 5f64e7d and 7c1a702.

📒 Files selected for processing (4)

• tests/integration/defs/stress_test/disagg_cancel/README.md
• tests/integration/defs/stress_test/disagg_cancel/harness.py
• tests/integration/defs/stress_test/disagg_cancel/test_injector.py
• tests/integration/defs/stress_test/disagg_cancel/test_log_scanner.py

[tensorrt-cicd]

PR_Github #51922 [ run ] completed with state FAILURE. Commit: 7c1a702
/LLM/main/L0_MergeRequest_PR pipeline #41276 completed with status: 'FAILURE'

CI Report

⚠️ Action Required:

• Please check the failed tests and fix your PR
• If you cannot view the failures, ask the CI triggerer to share details
• Once fixed, request an NVIDIA team member to trigger CI again

CI Agent Failure Analysis

Link to invocation

[chienchunhung]

/bot run --disable-fail-fast

[tensorrt-cicd]

PR_Github #52134 [ run ] triggered by Bot. Commit: 7c1a702 Link to invocation

[tensorrt-cicd]

PR_Github #52134 [ run ] completed with state FAILURE. Commit: 7c1a702
/LLM/main/L0_MergeRequest_PR pipeline #41459 completed with status: 'FAILURE'

CI Report

⚠️ Action Required:

• Please check the failed tests and fix your PR
• If you cannot view the failures, ask the CI triggerer to share details
• Once fixed, request an NVIDIA team member to trigger CI again

CI Agent Failure Analysis

Link to invocation

[chienchunhung]

/bot run --disable-fail-fast

[tensorrt-cicd]

PR_Github #52163 [ run ] triggered by Bot. Commit: 7c1a702 Link to invocation

[tensorrt-cicd]

PR_Github #52163 [ run ] completed with state SUCCESS. Commit: 7c1a702
/LLM/main/L0_MergeRequest_PR pipeline #41484 completed with status: 'SUCCESS'

CI Report

Link to invocation