Distributed tests for flux and wan with pytorch_xdit image

[amd-droy]

### Files Updated

New test suites:

cvs/tests/inference/pytorch_xdit/pytorch_xdit_flux1_dev_distributed.py — Multi-node FLUX.1-dev distributed benchmark: container cleanup, model verification, parallelism validation, unified torchrun launch, rank-0 result parsing, and threshold checks.
cvs/tests/inference/pytorch_xdit/pytorch_xdit_wan22_14b_distributed.py — Same pattern for WAN 2.2 I2V-A14B distributed inference.

New library modules:

cvs/lib/inference/pytorch_xdit/pytorch_xdit_flux_job.py — Shared FLUX benchmark launcher: NCCL env construction, torchrun command building, distributed log verification, fatal-output scanning, benchmark failure excerpt logging, and launch_flux_benchmark() entry point.
cvs/lib/inference/pytorch_xdit/pytorch_xdit_wan_job.py — WAN equivalent launcher with parallelism validation, shared output directory handling for rank-0, and launch_wan_benchmark() entry point.
cvs/lib/inference/pytorch_xdit/pytorch_xdit_flux.py — Moved FLUX output parser from cvs/parsers/; adds FluxOutputParser, log_results_summary() for multi-node result logging.
cvs/lib/inference/pytorch_xdit/pytorch_xdit_wan.py — Moved WAN output parser; adds WanOutputParser with multi-run aggregation under a shared output base directory.

Sample configs:

cvs/input/config_file/inference/pytorch_xdit/mi300x_pytorch_xdit_flux1_dev_distributed.json — Distributed FLUX template with nnodes, rendezvous, NCCL IB/socket settings, and ulysses_degree / ring_degree.
cvs/input/config_file/inference/pytorch_xdit/mi300x_pytorch_xdit_wan22_14b_distributed.json — Distributed WAN template with matching parallelism and NCCL fields.

Schema and validation:

cvs/parsers/schemas.py — Extended PytorchXditFluxConfig and PytorchXditWanConfig with distributed fields (nnodes, master_addr, master_port, NCCL IB/socket/GLOO settings); added PytorchXditDistributedNcclExamples base for example* NCCL hint fields in sample JSONs; added WAN ulysses_size / ring_size and FLUX ulysses_degree / ring_degree benchmark params.

Existing test touch-ups:

cvs/tests/inference/pytorch_xdit/pytorch_xdit_flux1_dev_single.py — Import parser from lib; use shared log_results_summary() for multi-node scale-out summary.
cvs/tests/inference/pytorch_xdit/pytorch_xdit_wan22_14b_single.py — Import WanOutputParser from lib instead of cvs/parsers/.

Unit tests:

cvs/lib/inference/pytorch_xdit/unittests/test_pytorch_xdit_flux.py — Tests for log_results_summary().
cvs/lib/inference/pytorch_xdit/unittests/test_pytorch_xdit_wan_job.py — Tests for WAN parallelism validation, torchrun command construction, and benchmark failure excerpt logging.
cvs/parsers/unittests/test_pytorch_xdit_schemas.py — Tests for distributed xDiT config schema validation including example* NCCL fields.

[speriaswamy-amd]

Thank you for working on this @amd-droy , overall, the PR looks good, but couple of changes could be done to improve resilience

[speriaswamy-amd]

We do

• RCCL env construction
• torchrun/docker command building
• launch planning
• parallelism validation
• run()

We need to add corresponding unit tests for these functions

[amd-droy]

Added the tests.

[speriaswamy-amd]

We need to add unit tests for wan output parsing logic

[amd-droy]

Added the tests.

[speriaswamy-amd]

exit codes are only returned when detailed=True is passed in

[amd-droy]

Addressed. The exec_cmd_list fast path is only used when detailed=False; benchmark run() uses detailed=True and gates on exit_code != 0.

[speriaswamy-amd]

scan_fatal_output/scan_wan_fatal_output only flags a run as failed if the captured text contains one of ~6 known substrings (Traceback, ModuleNotFoundError, ChildFailedError, OSError:, plus two WAN-specific GPU-driver strings) we would silently pass hardware, HIP, ROCm failures, segfaults, etc. we should probably gate on exit code instead of regex scan

[amd-droy]

Benchmark pass/fail is now gated on docker exit_code from exec(..., detailed=True), not regex. Removed FLUX scan_fatal_output; WAN only uses scan_wan_fatal_output for the timeout/hang override path, plus post-failure hint regex. Added tests for non-zero exit without traceback

[speriaswamy-amd]

This module is deprecated, we should be using :

    # Recommended (multi-process with auto-sharding):
    from cvs.lib.parallel.multiprocess_pssh import MultiProcessPssh

    # For single-process SSH only:
    from cvs.lib.parallel.pssh import Pssh

    # Configuration:
    from cvs.lib.parallel.config import ParallelConfig

[amd-droy]

Done.

[speriaswamy-amd]

FluxBenchmarkJob and WanBenchmarkJob (pytorch_xdit_flux_job.py/pytorch_xdit_wan_job.py) duplicate check_kfd, _fetch_hostnames, _build_env_args, _build_volume_args, _build_docker_cmd, build_launch_plan, run(), and store_output_dir_hint we probably need to write a baseclass and extend

[amd-droy]

new file cvs\lib\inference\pytorch_xdit\pytorch_xdit_benchmark_job.py is added.

[speriaswamy-amd]

pytorch_xdit_flux1_dev_distributed.py:282 verifies actual required files (model_index.json, transformer/vae weights) the WAN distributed test's equivalent branch both single and distributed tests only do do test -d && echo EXISTS no file level check.

[amd-droy]

Fixed — WAN single and distributed tests now use the shared pytorch_xdit_model_verify.py helper to verify required files (model_index.json, transformer/vae weights, or native HF snapshot artifacts) via verify_required_checks_on_nodes, matching FLUX parity.

[speriaswamy-amd]

FluxBenchmarkJob.validate_parallelism() and WanBenchmarkJob.validate_parallelism() each call the module-level validate_parallelism() twice with identical arguments (once for the error check, again just to re-fetch world_size/product for logging)

[amd-droy]

Fixed — validate_parallelism() is now called once per job; the returned (world_size, product, err) tuple is used for both the error check and the success log line.

[speriaswamy-amd]

FluxBenchmarkJob.validate_parallelism() and WanBenchmarkJob.validate_parallelism() each call the module-level validate_parallelism() twice with identical arguments (once for the error check, again just to re-fetch world_size/product for logging)

[amd-droy]

Fixed — validate_parallelism() is now called once per job; the returned (world_size, product, err) tuple is used for both the error check and the success log line.