From 2e146d305376970d96e3c81000b17367ffd937f0 Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Mon, 3 Aug 2026 13:56:02 -0500 Subject: [PATCH 1/2] feat(agentx): fail incomplete profiling metric coverage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Require TTFT and inter-token-latency observations through 98% of valid AgentX profiling phases. Preserve diagnostic artifacts, mark invalid submissions, and exit non-zero with an actionable server error.\n\n中文：要求有效 AgentX 分析阶段的 TTFT 和 token 间延迟观测覆盖至少 98% 的时长。保留诊断产物，将提交标记为无效，并通过明确的服务端错误以非零状态退出。 Signed-off-by: Cam Quilici --- .../semianalysis-agentx-faq.md | 12 +- docs/tutorials/agentx-mvp.md | 7 + src/aiperf/common/models/__init__.py | 2 + src/aiperf/common/models/record_models.py | 42 ++++++ src/aiperf/common/scenario/base.py | 11 ++ .../common/scenario/inferencex_agentx_mvp.py | 1 + src/aiperf/controller/system_controller.py | 12 ++ .../aggregate/aggregate_base_exporter.py | 13 +- src/aiperf/exporters/metrics_json_exporter.py | 11 ++ src/aiperf/metrics/accumulator.py | 56 +++++++- src/aiperf/records/records_manager.py | 130 ++++++++++++++++++ .../common/scenario/test_scenario_registry.py | 1 + .../unit/controller/test_system_controller.py | 40 +++++- .../exporters/test_submission_valid_field.py | 11 ++ .../test_metrics_accumulator.py | 109 +++++++++++++++ .../test_records_manager_process_results.py | 87 ++++++++++++ 16 files changed, 537 insertions(+), 8 deletions(-) diff --git a/docs/benchmark-modes/semianalysis-agentx-faq.md b/docs/benchmark-modes/semianalysis-agentx-faq.md index aae226b1fb..8ae2a5d781 100644 --- a/docs/benchmark-modes/semianalysis-agentx-faq.md +++ b/docs/benchmark-modes/semianalysis-agentx-faq.md @@ -819,9 +819,8 @@ are treated as a 0% rate. So a handful of overflows on a large run won't invalid that's systematically too small will. ### Q: Do generic request errors (HTTP 500s, timeouts) invalidate the run? -No — and this surprises people. The only three things that flip `submission_valid` to `false` are an -unsafe override, a >1% context-overflow rate, and a cancelled run; a generic-error rate is not among -them. A run with substantial 500s still reports `submission_valid: true` — error responses even land +Not by themselves. A generic-error rate is not a direct submission-validity input. A run with +substantial 500s can still report `submission_valid: true` — error responses even land in the *denominator* of the overflow-rate computation, so heavy generic errors make overflow invalidation *less* likely, not more. Errors are counted in the `error_request_count` metric and are excluded from the latency distributions (an errored request contributes no TTFT/ITL samples). So @@ -837,6 +836,13 @@ it, the run cancels, which then invalidates it with reason `run_cancelled`. The the failed and total request counts, observed failure percentage, configured limit, and directs the operator to the inference-server logs. +For profiling phases that meet the AgentX scenario's minimum valid duration, the scenario also +requires TTFT and inter-token-latency observations to extend through at least 98% of the phase. This +catches a server that stops returning responses while AIPerf still has requests in flight: the run +exits non-zero, the JSON artifact is retained with `submission_valid: false` and reason +`insufficient_profile_metric_coverage`, and the error directs the operator to the server logs. +Warmup observations and intentionally short `--unsafe-override` smoke runs do not count. + ### Q: My server has a ~256k context window and the run keeps overflowing — what's the right fix? Switch to a `_256k` corpus (see [§3](#3-how-realistic-are-the-prompts-and-token-counts)) — sizing the dataset to your server is the correct fix, not `--unsafe-override`, which would just leave you with a diff --git a/docs/tutorials/agentx-mvp.md b/docs/tutorials/agentx-mvp.md index eb67d780cd..0172d077f4 100644 --- a/docs/tutorials/agentx-mvp.md +++ b/docs/tutorials/agentx-mvp.md @@ -653,6 +653,13 @@ the aggregate file — divide it by (the same denominator the aggregate exporter uses for the 1% threshold) to see how close you were to the limit. +**Run exits non-zero with `ProfileMetricCoverageError`** +The server stopped producing TTFT or inter-token-latency observations before 98% of the configured +profiling duration elapsed. AIPerf retains the result artifact, marks it invalid with +`insufficient_profile_metric_coverage`, and reports the observed coverage for both signals. Check +the inference-server logs for a crash or stalled request processing. Warmup metrics and profiling +phases shorter than the scenario's minimum valid duration are excluded. + **"scenario `'inferencex-agentx-mvp'` requires loader=any of …" / cannot verify corpus identity for a local weka_trace directory** The AgentX MVP scenario stamps `submission_valid: true` only for a pinned public SemiAnalysis Weka corpus (`semianalysisai/cc-traces-weka-062126`), diff --git a/src/aiperf/common/models/__init__.py b/src/aiperf/common/models/__init__.py index 2ffa47bf2a..a1c557be43 100644 --- a/src/aiperf/common/models/__init__.py +++ b/src/aiperf/common/models/__init__.py @@ -80,6 +80,7 @@ ParsedResponseRecord, PhaseProfileResults, ProcessRecordsResult, + ProfileMetricDurationCoverage, ProfileResults, RAGSources, RankingsResponseData, @@ -233,6 +234,7 @@ "ProcessServerMetricsResult", "ProcessTelemetryResult", "ProcessingStats", + "ProfileMetricDurationCoverage", "ProfileResults", "RAGSources", "RankingsResponseData", diff --git a/src/aiperf/common/models/record_models.py b/src/aiperf/common/models/record_models.py index 8ccce975b1..2a9e06adc7 100644 --- a/src/aiperf/common/models/record_models.py +++ b/src/aiperf/common/models/record_models.py @@ -416,6 +416,36 @@ class PhaseProfileResults(AIPerfBaseModel): ) +class ProfileMetricDurationCoverage(AIPerfBaseModel): + """Observed profiling-metric coverage relative to a configured phase duration.""" + + phase_name: str = Field(description="Name of the profiling phase checked.") + expected_duration_seconds: float = Field( + gt=0.0, description="Configured profiling duration in seconds." + ) + required_ratio: float = Field( + gt=0.0, le=1.0, description="Minimum required metric coverage ratio." + ) + ttft_ratio: float = Field( + ge=0.0, + le=1.0, + description="Fraction of the phase covered by TTFT observations.", + ) + inter_token_latency_ratio: float = Field( + ge=0.0, + le=1.0, + description="Fraction of the phase covered by inter-token-latency observations.", + ) + + @property + def passed(self) -> bool: + """Return whether every required latency signal met the threshold.""" + return ( + self.ttft_ratio >= self.required_ratio + and self.inter_token_latency_ratio >= self.required_ratio + ) + + class ProfileResults(AIPerfBaseModel): """The results of a profile run.""" @@ -479,6 +509,14 @@ class ProfileResults(AIPerfBaseModel): "normal metric accumulation and stream export, retained only for " "aggregate runtime submission validation.", ) + metric_duration_coverage: list[ProfileMetricDurationCoverage] = Field( + default_factory=list, + description="Post-run TTFT and ITL duration-coverage checks, when enabled.", + ) + runtime_submission_invalid_reasons: list[str] = Field( + default_factory=list, + description="Runtime reason tags that invalidate a scenario submission.", + ) phase_records: list[PhaseProfileResults] | None = Field( default=None, description="Internal per-phase metric summaries used for phase artifacts.", @@ -511,6 +549,10 @@ class ProcessRecordsResult(AIPerfBaseModel): default_factory=list, description="Any error that occurred while processing the profile results", ) + fatal_errors: list[ErrorDetails] = Field( + default_factory=list, + description="Post-processing validation errors that require a non-zero exit.", + ) def get(self, tag: MetricTagT) -> MetricResult | None: """Get a metric result by tag, if it exists.""" diff --git a/src/aiperf/common/scenario/base.py b/src/aiperf/common/scenario/base.py index f5c1104d63..964be1d88d 100644 --- a/src/aiperf/common/scenario/base.py +++ b/src/aiperf/common/scenario/base.py @@ -138,6 +138,17 @@ class ScenarioSpec(AIPerfBaseModel): "(which stamps submission_valid=false)." ), ) + minimum_profile_metric_coverage_ratio: float | None = Field( + default=None, + gt=0.0, + le=1.0, + description=( + "Minimum fraction of each duration-based profiling phase that must " + "contain both TTFT and inter-token-latency observations. A run that " + "falls below the threshold exits non-zero and is not a valid scenario " + "submission. None disables the post-run check." + ), + ) class ScenarioViolation(AIPerfBaseModel): diff --git a/src/aiperf/common/scenario/inferencex_agentx_mvp.py b/src/aiperf/common/scenario/inferencex_agentx_mvp.py index 601fb834cf..0a34079f10 100644 --- a/src/aiperf/common/scenario/inferencex_agentx_mvp.py +++ b/src/aiperf/common/scenario/inferencex_agentx_mvp.py @@ -36,4 +36,5 @@ system_idle_gap_cap_seconds=10.0, forbid_inter_turn_delay_cap=True, require_cache_bust=CacheBustTarget.FIRST_TURN_PREFIX, + minimum_profile_metric_coverage_ratio=0.98, ) diff --git a/src/aiperf/controller/system_controller.py b/src/aiperf/controller/system_controller.py index 7bd77c7736..7883ad1d0d 100644 --- a/src/aiperf/controller/system_controller.py +++ b/src/aiperf/controller/system_controller.py @@ -677,6 +677,18 @@ async def _on_process_records_result_message( self.error( f"Received process records result message with errors: {message.results.errors}" ) + for fatal_error in message.results.fatal_errors: + self.error( + "Received fatal profile-results validation error: " + f"{fatal_error.message}" + ) + self._exit_errors.append( + ExitErrorInfo( + error_details=fatal_error, + operation="profile_results_validation", + service_id=message.service_id, + ) + ) self.debug( lambda: ( diff --git a/src/aiperf/exporters/aggregate/aggregate_base_exporter.py b/src/aiperf/exporters/aggregate/aggregate_base_exporter.py index 3d44934901..7722041229 100644 --- a/src/aiperf/exporters/aggregate/aggregate_base_exporter.py +++ b/src/aiperf/exporters/aggregate/aggregate_base_exporter.py @@ -74,14 +74,16 @@ def compute_submission_outcome( total_responses: int = 0, context_overflow_count: int = 0, was_cancelled: bool = False, + runtime_invalid_reasons: list[str] | None = None, ) -> tuple[bool | None, list[str]]: """Combine validator outcome with runtime threshold checks into a final verdict. The validator-side outcome covers static config violations (handled at UserConfig.model_post_init by ``validate_scenario``). This helper folds in runtime-only signals that are only knowable post-run -- the >1% - context-overflow rate per spec §7, and early cancellation (Ctrl+C): a - cancelled run produces partial metrics and is never a valid submission. + context-overflow rate per spec §7, post-run validation failures, and early + cancellation (Ctrl+C): a cancelled run produces partial metrics and is never + a valid submission. Rate semantics: strictly greater than ``Environment.AGENTX.CONTEXT_OVERFLOW_RATE_LIMIT`` (default 0.01 per @@ -110,6 +112,8 @@ def compute_submission_outcome( was_cancelled: Whether the run was cancelled early (graceful Ctrl+C). True flips ``submission_valid`` to False with reason ``"run_cancelled"``. + runtime_invalid_reasons: Runtime validation reason tags to merge into + the final verdict. Any value makes the submission invalid. Returns: A ``(submission_valid, reasons)`` tuple suitable for feeding into @@ -138,6 +142,11 @@ def compute_submission_outcome( if RUN_CANCELLED_REASON not in reasons: reasons.append(RUN_CANCELLED_REASON) + for reason in runtime_invalid_reasons or []: + valid = False + if reason not in reasons: + reasons.append(reason) + return valid, reasons diff --git a/src/aiperf/exporters/metrics_json_exporter.py b/src/aiperf/exporters/metrics_json_exporter.py index e9bff71387..88addd237c 100644 --- a/src/aiperf/exporters/metrics_json_exporter.py +++ b/src/aiperf/exporters/metrics_json_exporter.py @@ -91,6 +91,14 @@ def _generate_content(self) -> str: dataset = public_dataset_provenance(self._cfg) if dataset is not None: run_metadata["dataset"] = dataset + metric_duration_coverage = getattr( + self._results, "metric_duration_coverage", [] + ) + if metric_duration_coverage: + run_metadata["metric_duration_coverage"] = [ + coverage.model_dump(mode="json") + for coverage in metric_duration_coverage + ] # ProfileResults.context_overflow_count is the AGENTIC_REPLAY skip-path # side channel only (not in error_request_count / ContextOverflowCountMetric). @@ -176,6 +184,9 @@ def _metric_avg(tag: str) -> int: total_responses=total_responses, context_overflow_count=context_overflow_count, was_cancelled=bool(self._results.was_cancelled), + runtime_invalid_reasons=list( + getattr(self._results, "runtime_submission_invalid_reasons", []) + ), ) run_metadata.update( _build_run_metadata_dict( diff --git a/src/aiperf/metrics/accumulator.py b/src/aiperf/metrics/accumulator.py index 2531af25f1..7c65859095 100644 --- a/src/aiperf/metrics/accumulator.py +++ b/src/aiperf/metrics/accumulator.py @@ -20,8 +20,13 @@ ) from aiperf.common.environment import Environment from aiperf.common.exceptions import NoMetricValue +from aiperf.common.finite import is_finite_value from aiperf.common.messages import MetricRecordsData -from aiperf.common.models import MetricResult, TimesliceResult +from aiperf.common.models import ( + MetricResult, + ProfileMetricDurationCoverage, + TimesliceResult, +) from aiperf.common.types import MetricTagT from aiperf.metrics.accumulator_models import AccumulatorMetricsSummary from aiperf.metrics.accumulator_sweeps import compute_sweep_curves @@ -190,6 +195,8 @@ async def process_record(self, record: MetricRecordsData) -> None: gen_start = ( float(meta.request_start_ns + int(ttft_ns)) if ttft_ns is not None else None ) + inter_token_latency = record.metrics.get("inter_token_latency") + has_inter_token_latency = is_finite_value(inter_token_latency) self._column_store.ingest( idx=idx, @@ -216,6 +223,7 @@ async def process_record(self, record: MetricRecordsData) -> None: metadata_bool={ "was_cancelled": meta.was_cancelled, "has_error": record.error is not None, + "has_inter_token_latency": has_inter_token_latency, }, metadata_categorical={ "worker_id": meta.worker_id, @@ -342,6 +350,52 @@ def _mask_for_export_context(self, ctx: ExportContext | None) -> BoolArray | Non mask &= self._column_store.start_ns[:n] < ctx.end_ns return mask + def profile_metric_duration_coverage( + self, + ctx: ExportContext, + *, + phase_name: str, + expected_duration_seconds: float, + required_ratio: float, + ) -> ProfileMetricDurationCoverage: + """Measure how far required latency signals extend into a profile phase. + + TTFT is observed when the first token arrives. ITL is observed only on a + successful response carrying at least one inter-token interval, so its + last observation is the response end timestamp. The phase mask excludes + warmup even when warmup and profiling records share the same accumulator. + """ + if ctx.start_ns is None: + raise ValueError("profiling phase start time is unavailable") + + duration_ns = expected_duration_seconds * NANOS_PER_SECOND + if duration_ns <= 0: + raise ValueError("profiling phase duration must be positive") + + mask = self._mask_for_export_context(ctx) + if mask is None: + mask = np.ones(self._column_store.count, dtype=bool) + + def _coverage_ratio(timestamps: FloatArray, signal_mask: BoolArray) -> float: + selected = timestamps[signal_mask & np.isfinite(timestamps)] + if selected.size == 0: + return 0.0 + elapsed_ns = float(np.max(selected)) - ctx.start_ns + return min(max(elapsed_ns / duration_ns, 0.0), 1.0) + + n = self._column_store.count + generation_start_ns = self._column_store.generation_start_ns[:n] + end_ns = self._column_store.end_ns[:n] + has_itl = self._column_store.metadata_bool("has_inter_token_latency") == 1 + + return ProfileMetricDurationCoverage( + phase_name=phase_name, + expected_duration_seconds=expected_duration_seconds, + required_ratio=required_ratio, + ttft_ratio=_coverage_ratio(generation_start_ns, mask), + inter_token_latency_ratio=_coverage_ratio(end_ns, mask & has_itl), + ) + def _aggregate_values(self, tag: MetricTagT, values: np.ndarray) -> float: """Apply the tag's aggregation function to an array of values.""" kind = self._aggregation_kinds.get(tag, AggregationKind.SUM) diff --git a/src/aiperf/records/records_manager.py b/src/aiperf/records/records_manager.py index 5280772228..9e5cb94669 100644 --- a/src/aiperf/records/records_manager.py +++ b/src/aiperf/records/records_manager.py @@ -60,10 +60,12 @@ ProcessRecordsResult, ProcessServerMetricsResult, ProcessTelemetryResult, + ProfileMetricDurationCoverage, ProfileResults, TimesliceResult, WorkerProcessingStats, ) +from aiperf.common.scenario.registry import get_scenario from aiperf.common.types import MetricTagT from aiperf.common.utils import yield_to_event_loop from aiperf.config.comm import ZMQDualBindConfig @@ -1720,6 +1722,126 @@ async def _phase_metric_results( ) return records_results, error_results + def _validate_profile_metric_duration_coverage( + self, + phase: CreditPhase, + cancelled: bool, + ) -> tuple[list[ProfileMetricDurationCoverage], list[ErrorDetails]]: + """Apply a scenario's post-run latency-signal coverage requirement.""" + if phase != CreditPhase.PROFILING or cancelled: + return [], [] + + scenario_name = self.run.cfg.scenario + if scenario_name is None: + return [], [] + # Scenario resolution validates the name before services start. + scenario_spec = get_scenario(scenario_name) + required_ratio = scenario_spec.minimum_profile_metric_coverage_ratio + if required_ratio is None: + return [], [] + + accumulator = self._accumulators.get(AccumulatorType.METRIC_RESULTS) + calculate = getattr(accumulator, "profile_metric_duration_coverage", None) + if not callable(calculate): + error = ErrorDetails( + type="ProfileMetricCoverageError", + message=( + "Profiling metric coverage could not be validated because the " + "metrics accumulator does not expose coverage timestamps." + ), + ) + return [], [error] + + phase_configs = self.run.cfg.get_profiling_phases() + concrete_stats = self._iter_concrete_phase_stats(CreditPhase.PROFILING) + aggregate_stats = RecordsManager._create_result_stats_for_phase( + self, CreditPhase.PROFILING + ) + coverage_results: list[ProfileMetricDurationCoverage] = [] + fatal_errors: list[ErrorDetails] = [] + + for profiling_index, phase_config in enumerate(phase_configs): + if phase_config.duration is None: + continue + if phase_config.duration < scenario_spec.min_benchmark_duration_seconds: + self.info( + "Skipping profiling metric coverage validation for " + f"{phase_config.name!r}: configured duration " + f"{float(phase_config.duration):.1f}s is below scenario minimum " + f"{scenario_spec.min_benchmark_duration_seconds}s." + ) + continue + stats = next( + ( + item + for item in concrete_stats + if item.phase_name == phase_config.name + or item.profiling_index == profiling_index + ), + aggregate_stats if len(phase_configs) == 1 else None, + ) + if stats is None or stats.start_ns is None: + fatal_errors.append( + ErrorDetails( + type="ProfileMetricCoverageError", + message=( + "Profiling metric coverage could not be validated for " + f"phase {phase_config.name!r} because its start time is " + "unavailable." + ), + ) + ) + continue + + ctx = ExportContext( + start_ns=stats.start_ns, + end_ns=stats.requests_end_ns, + phase=CreditPhase.PROFILING, + phase_index=stats.phase_index, + phase_name=phase_config.name, + phase_kind="profiling", + is_phase_scoped=True, + cancelled=False, + ) + coverage = calculate( + ctx, + phase_name=phase_config.name, + expected_duration_seconds=float(phase_config.duration), + required_ratio=required_ratio, + ) + coverage_results.append(coverage) + if coverage.passed: + self.info( + "Profiling metric coverage passed for " + f"{phase_config.name!r}: TTFT={coverage.ttft_ratio:.1%}, " + "inter-token latency=" + f"{coverage.inter_token_latency_ratio:.1%} " + f"(required={required_ratio:.1%})." + ) + continue + + allowed_tail_seconds = float(phase_config.duration) * (1.0 - required_ratio) + message = ( + f"Profiling metric coverage below the required {required_ratio:.1%} " + f"for phase {phase_config.name!r}: TTFT={coverage.ttft_ratio:.1%}, " + "inter-token latency=" + f"{coverage.inter_token_latency_ratio:.1%} over the configured " + f"{float(phase_config.duration):.1f}s duration. At least one required " + f"metric stopped more than {allowed_tail_seconds:.1f}s before the " + "nominal profiling end; check inference server logs for a stalled " + "or unavailable server." + ) + self.error(message) + fatal_errors.append( + ErrorDetails( + type="ProfileMetricCoverageError", + message=message, + details=coverage.model_dump(mode="json"), + ) + ) + + return coverage_results, fatal_errors + async def _phase_telemetry_results( self, stats: PhaseRecordsStats, @@ -1979,6 +2101,9 @@ async def _process_results_impl( phase_records = await RecordsManager._build_phase_profile_results( self, phase, cancelled ) + metric_duration_coverage, fatal_errors = ( + self._validate_profile_metric_duration_coverage(phase, cancelled) + ) # Snapshot count BEFORE extending with derived aggregates (efficiency, # analyzers) — `completed` reports request-derived records only. records_completed = len(records_results) @@ -2005,12 +2130,17 @@ async def _process_results_impl( context_overflow_count=self._skipped_context_overflow_counts_by_phase.get( phase, 0 ), + metric_duration_coverage=metric_duration_coverage, + runtime_submission_invalid_reasons=( + ["insufficient_profile_metric_coverage"] if fatal_errors else [] + ), phase_records=phase_records, pooled_spec_decode_acceptance_histogram=_pooled_spec_decode_histogram( summary_ctx ), ), errors=error_results, + fatal_errors=fatal_errors, ) self.debug(lambda: f"Process records result: {result}") self.debug("Publishing ProcessRecordsResultMessage...") diff --git a/tests/unit/common/scenario/test_scenario_registry.py b/tests/unit/common/scenario/test_scenario_registry.py index 56df54c1b5..3164d73450 100644 --- a/tests/unit/common/scenario/test_scenario_registry.py +++ b/tests/unit/common/scenario/test_scenario_registry.py @@ -36,6 +36,7 @@ def test_inferencex_agentx_mvp_registered(): assert spec.inter_turn_delay_cap_seconds is None assert spec.trace_idle_gap_cap_seconds is None assert spec.system_idle_gap_cap_seconds == 10.0 + assert spec.minimum_profile_metric_coverage_ratio == 0.98 assert spec.forbid_trace_idle_gap_cap is False assert spec.forbid_inter_turn_delay_cap is True diff --git a/tests/unit/controller/test_system_controller.py b/tests/unit/controller/test_system_controller.py index 3998865acc..1463b559ae 100644 --- a/tests/unit/controller/test_system_controller.py +++ b/tests/unit/controller/test_system_controller.py @@ -13,9 +13,14 @@ ) from aiperf.common.environment import Environment from aiperf.common.exceptions import LifecycleOperationError -from aiperf.common.messages import ProfileCancelCommand +from aiperf.common.messages import ProcessRecordsResultMessage, ProfileCancelCommand from aiperf.common.messages.command_messages import CommandErrorResponse -from aiperf.common.models import ErrorDetails, ExitErrorInfo +from aiperf.common.models import ( + ErrorDetails, + ExitErrorInfo, + ProcessRecordsResult, + ProfileResults, +) from aiperf.controller.system_controller import SystemController from aiperf.plugin.enums import AccuracyBenchmarkType from tests.unit.controller.conftest import MockTestException @@ -81,6 +86,37 @@ async def test_system_controller_no_error_on_start_success( class TestSystemControllerExitScenarios: """Test exit scenarios for the SystemController.""" + @pytest.mark.asyncio + async def test_fatal_profile_result_validation_records_exit_error( + self, system_controller: SystemController + ) -> None: + """A post-run coverage failure makes the final process exit non-zero.""" + fatal_error = ErrorDetails( + type="ProfileMetricCoverageError", + message="Profiling metric coverage below the required 98.0%.", + ) + message = ProcessRecordsResultMessage( + service_id="records_manager", + results=ProcessRecordsResult( + results=ProfileResults( + records=[], + completed=0, + start_ns=1, + end_ns=2, + ), + fatal_errors=[fatal_error], + ), + ) + system_controller._check_and_trigger_shutdown = AsyncMock() + + await system_controller._on_process_records_result_message(message) + + assert any( + item.error_details == fatal_error + and item.operation == "profile_results_validation" + for item in system_controller._exit_errors + ) + @pytest.mark.asyncio async def test_system_controller_exits_on_profile_configure_error_response( self, diff --git a/tests/unit/exporters/test_submission_valid_field.py b/tests/unit/exporters/test_submission_valid_field.py index 37c8ac02b0..9a0ce43c65 100644 --- a/tests/unit/exporters/test_submission_valid_field.py +++ b/tests/unit/exporters/test_submission_valid_field.py @@ -95,6 +95,17 @@ def test_cancelled_run_appends_reason_to_existing_reasons() -> None: assert reasons == ["unsafe_override", CONTEXT_OVERFLOW_REASON, RUN_CANCELLED_REASON] +def test_runtime_validation_failure_invalidates_submission() -> None: + valid, reasons = compute_submission_outcome( + scenario_name="inferencex-agentx-mvp", + validator_submission_valid=True, + runtime_invalid_reasons=["insufficient_profile_metric_coverage"], + ) + + assert valid is False + assert reasons == ["insufficient_profile_metric_coverage"] + + def test_overflow_rate_boundary_without_double_count() -> None: """101 overflows / 10_000 responses must flip submission_valid (regression against double-counting overflows into the denominator).""" valid, reasons = compute_submission_outcome( diff --git a/tests/unit/post_processors/test_metrics_accumulator.py b/tests/unit/post_processors/test_metrics_accumulator.py index 3dfd05cdd9..4a382794d7 100644 --- a/tests/unit/post_processors/test_metrics_accumulator.py +++ b/tests/unit/post_processors/test_metrics_accumulator.py @@ -30,6 +30,7 @@ from aiperf.metrics.types.request_throughput_metric import RequestThroughputMetric from tests.unit.post_processors.conftest import ( create_accumulator_with_metrics, + create_metric_metadata, create_metric_records_data, ) @@ -76,6 +77,114 @@ async def test_process_record_record_metric( values = processor._column_store.numeric("test_record") assert list(values[~np.isnan(values)]) == [42.0, 84.0] + @pytest.mark.asyncio + async def test_profile_metric_duration_coverage_is_phase_scoped( + self, mock_metric_registry: Mock, mock_run + ) -> None: + """Warmup observations cannot hide profiling metrics that stopped early.""" + processor = MetricsAccumulator(mock_run) + phase_start_ns = 100 * NANOS_PER_SECOND + + profiling = create_metric_records_data( + session_num=0, + request_start_ns=147 * NANOS_PER_SECOND, + request_end_ns=149 * NANOS_PER_SECOND, + results=[ + {"time_to_first_token": NANOS_PER_SECOND}, + {"inter_token_latency": 100_000_000}, + ], + ) + warmup_metadata = create_metric_metadata( + session_num=0, + benchmark_phase=CreditPhase.WARMUP, + request_start_ns=197 * NANOS_PER_SECOND, + request_end_ns=199 * NANOS_PER_SECOND, + ) + warmup = create_metric_records_data( + metadata=warmup_metadata, + results=[ + {"time_to_first_token": NANOS_PER_SECOND}, + {"inter_token_latency": 100_000_000}, + ], + ) + await processor.process_record(profiling) + await processor.process_record(warmup) + + coverage = processor.profile_metric_duration_coverage( + ExportContext( + start_ns=phase_start_ns, + phase=CreditPhase.PROFILING, + ), + phase_name="profiling", + expected_duration_seconds=100.0, + required_ratio=0.98, + ) + + assert coverage.ttft_ratio == pytest.approx(0.48) + assert coverage.inter_token_latency_ratio == pytest.approx(0.49) + assert coverage.passed is False + + @pytest.mark.asyncio + async def test_profile_metric_duration_coverage_accepts_threshold_boundary( + self, mock_metric_registry: Mock, mock_run + ) -> None: + """A last TTFT exactly at the threshold and a later ITL record pass.""" + processor = MetricsAccumulator(mock_run) + phase_start_ns = 100 * NANOS_PER_SECOND + record = create_metric_records_data( + session_num=0, + request_start_ns=197 * NANOS_PER_SECOND, + request_end_ns=199 * NANOS_PER_SECOND, + results=[ + {"time_to_first_token": NANOS_PER_SECOND}, + {"inter_token_latency": 100_000_000}, + ], + ) + await processor.process_record(record) + + coverage = processor.profile_metric_duration_coverage( + ExportContext( + start_ns=phase_start_ns, + phase=CreditPhase.PROFILING, + ), + phase_name="profiling", + expected_duration_seconds=100.0, + required_ratio=0.98, + ) + + assert coverage.ttft_ratio == pytest.approx(0.98) + assert coverage.inter_token_latency_ratio == pytest.approx(0.99) + assert coverage.passed is True + + @pytest.mark.asyncio + async def test_profile_metric_duration_coverage_requires_itl( + self, mock_metric_registry: Mock, mock_run + ) -> None: + """TTFT alone cannot make an interactive streaming run valid.""" + processor = MetricsAccumulator(mock_run) + phase_start_ns = 100 * NANOS_PER_SECOND + record = create_metric_records_data( + session_num=0, + request_start_ns=198 * NANOS_PER_SECOND, + request_end_ns=199 * NANOS_PER_SECOND, + results=[{"time_to_first_token": 0}], + ) + await processor.process_record(record) + + coverage = processor.profile_metric_duration_coverage( + ExportContext( + start_ns=phase_start_ns, + phase=CreditPhase.PROFILING, + ), + phase_name="profiling", + expected_duration_seconds=100.0, + required_ratio=0.98, + ) + + assert coverage.ttft_ratio == pytest.approx(0.98) + assert coverage.inter_token_latency_ratio == 0.0 + assert coverage.passed is False + @pytest.mark.asyncio async def test_process_record_record_metric_list_values( self, mock_metric_registry: Mock, mock_run diff --git a/tests/unit/records/test_records_manager_process_results.py b/tests/unit/records/test_records_manager_process_results.py index 249b99eafd..f58e98bb32 100644 --- a/tests/unit/records/test_records_manager_process_results.py +++ b/tests/unit/records/test_records_manager_process_results.py @@ -34,6 +34,7 @@ MetricResult, PhaseRecordsStats, ProcessRecordsResult, + ProfileMetricDurationCoverage, TimesliceResult, ) from aiperf.metrics.accumulator_models import AccumulatorMetricsSummary @@ -143,6 +144,8 @@ def _make_manager_mock( mgr.run = MagicMock() mgr.run.cfg.gpu_telemetry_disabled = user_config_telemetry_disabled mgr.run.cfg.server_metrics_disabled = user_config_server_metrics_disabled + mgr.run.cfg.scenario = None + mgr.run.cfg.get_profiling_phases.return_value = [] # Logging mgr.debug = MagicMock() @@ -169,6 +172,12 @@ def _make_manager_mock( mgr._summarize_warmup_metric_records = ( RecordsManager._summarize_warmup_metric_records.__get__(mgr) ) + mgr._validate_profile_metric_duration_coverage = ( + RecordsManager._validate_profile_metric_duration_coverage.__get__(mgr) + ) + mgr._iter_concrete_phase_stats = RecordsManager._iter_concrete_phase_stats.__get__( + mgr + ) mgr._summarize_one_accumulator = RecordsManager._summarize_one_accumulator.__get__( mgr ) @@ -271,6 +280,84 @@ async def test_accumulator_summarize_failure_does_not_abort(self) -> None: mgr.error.assert_called() assert any("summarize boom" in str(err.message or err) for err in result.errors) + @pytest.mark.asyncio + async def test_agentx_metric_coverage_failure_is_fatal(self) -> None: + """A duration-based AgentX phase below 98% remains exportable but fatal.""" + acc = _make_summary_accumulator([_STUB_METRIC_RESULT]) + acc.profile_metric_duration_coverage.return_value = ( + ProfileMetricDurationCoverage( + phase_name="profiling", + expected_duration_seconds=3600.0, + required_ratio=0.98, + ttft_ratio=0.861194, + inter_token_latency_ratio=0.861569, + ) + ) + mgr = _make_manager_mock(accumulators={AccumulatorType.METRIC_RESULTS: acc}) + mgr.run.cfg.scenario = "inferencex-agentx-mvp" + phase_config = MagicMock() + phase_config.name = "profiling" + phase_config.duration = 3600.0 + mgr.run.cfg.get_profiling_phases.return_value = [phase_config] + + result = await mgr._process_results( + phase=CreditPhase.PROFILING, cancelled=False + ) + + assert len(result.fatal_errors) == 1 + assert result.fatal_errors[0].type == "ProfileMetricCoverageError" + assert "required 98.0%" in result.fatal_errors[0].message + assert "check inference server logs" in result.fatal_errors[0].message + assert result.results.runtime_submission_invalid_reasons == [ + "insufficient_profile_metric_coverage" + ] + assert result.results.metric_duration_coverage[0].ttft_ratio == pytest.approx( + 0.861194 + ) + + @pytest.mark.asyncio + async def test_agentx_metric_coverage_passes_at_threshold(self) -> None: + acc = _make_summary_accumulator([_STUB_METRIC_RESULT]) + acc.profile_metric_duration_coverage.return_value = ( + ProfileMetricDurationCoverage( + phase_name="profiling", + expected_duration_seconds=3600.0, + required_ratio=0.98, + ttft_ratio=0.98, + inter_token_latency_ratio=0.999, + ) + ) + mgr = _make_manager_mock(accumulators={AccumulatorType.METRIC_RESULTS: acc}) + mgr.run.cfg.scenario = "inferencex-agentx-mvp" + phase_config = MagicMock() + phase_config.name = "profiling" + phase_config.duration = 3600.0 + mgr.run.cfg.get_profiling_phases.return_value = [phase_config] + + result = await mgr._process_results( + phase=CreditPhase.PROFILING, cancelled=False + ) + + assert result.fatal_errors == [] + assert result.results.runtime_submission_invalid_reasons == [] + + @pytest.mark.asyncio + async def test_agentx_short_unsafe_smoke_skips_metric_coverage(self) -> None: + acc = _make_summary_accumulator([_STUB_METRIC_RESULT]) + mgr = _make_manager_mock(accumulators={AccumulatorType.METRIC_RESULTS: acc}) + mgr.run.cfg.scenario = "inferencex-agentx-mvp" + phase_config = MagicMock() + phase_config.name = "profiling" + phase_config.duration = 30.0 + mgr.run.cfg.get_profiling_phases.return_value = [phase_config] + + result = await mgr._process_results( + phase=CreditPhase.PROFILING, cancelled=False + ) + + acc.profile_metric_duration_coverage.assert_not_called() + assert result.fatal_errors == [] + @pytest.mark.asyncio async def test_empty_accumulators_produces_empty_records(self) -> None: mgr = _make_manager_mock(accumulators={}) From f77c31a715850ce8f0eb9dd37ddc90ce0bffcdbe Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Mon, 3 Aug 2026 14:00:47 -0500 Subject: [PATCH 2/2] fix(agentx): retain artifacts on coverage failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Export usable partial profiling results before reporting fatal post-run validation errors and exiting non-zero.\n\n中文：在报告致命的运行后验证错误并以非零状态退出前，先导出可用的部分分析结果。 Signed-off-by: Cam Quilici --- src/aiperf/controller/system_controller.py | 19 +++++++-- .../unit/controller/test_system_controller.py | 39 +++++++++++++++++++ 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/src/aiperf/controller/system_controller.py b/src/aiperf/controller/system_controller.py index 7883ad1d0d..244277694b 100644 --- a/src/aiperf/controller/system_controller.py +++ b/src/aiperf/controller/system_controller.py @@ -1100,10 +1100,7 @@ async def _stop_system_controller(self) -> None: # on Windows PIPE'd stdout — but any rendering bug has the same blast # radius, so we catch broadly. try: - if not self._exit_errors: - await self._print_post_benchmark_info_and_metrics() - else: - self._print_exit_errors_and_log_file() + await self._report_post_shutdown_results_and_errors() if Environment.DEV.MODE: # Print a warning message to the console if developer mode is enabled, on exit after results @@ -1129,6 +1126,20 @@ async def _stop_system_controller(self) -> None: # Exit the process in a more explicit way, to ensure that it stops os._exit(1 if self._exit_errors else 0) + async def _report_post_shutdown_results_and_errors(self) -> None: + """Export any usable results before reporting a fatal run error.""" + has_exportable_results = bool( + self._profile_results and self._profile_results.results.records + ) + if has_exportable_results: + await self._print_post_benchmark_info_and_metrics() + if self._exit_errors: + self._print_exit_errors_and_log_file() + elif self._exit_errors: + self._print_exit_errors_and_log_file() + else: + await self._print_post_benchmark_info_and_metrics() + def _print_exit_errors_and_log_file(self) -> None: """Print post exit errors and log file info to the console.""" console = Console() diff --git a/tests/unit/controller/test_system_controller.py b/tests/unit/controller/test_system_controller.py index 1463b559ae..20924bb7e3 100644 --- a/tests/unit/controller/test_system_controller.py +++ b/tests/unit/controller/test_system_controller.py @@ -18,6 +18,7 @@ from aiperf.common.models import ( ErrorDetails, ExitErrorInfo, + MetricResult, ProcessRecordsResult, ProfileResults, ) @@ -117,6 +118,44 @@ async def test_fatal_profile_result_validation_records_exit_error( for item in system_controller._exit_errors ) + @pytest.mark.asyncio + async def test_fatal_validation_exports_partial_results_before_error_report( + self, system_controller: SystemController + ) -> None: + """A fatal post-run verdict retains aggregate artifacts for diagnosis.""" + system_controller._profile_results = ProcessRecordsResult( + results=ProfileResults( + records=[ + MetricResult( + tag="request_count", + header="Request Count", + unit="requests", + avg=1.0, + ) + ], + completed=1, + start_ns=1, + end_ns=2, + ) + ) + system_controller._exit_errors = [ + ExitErrorInfo( + error_details=ErrorDetails( + type="ProfileMetricCoverageError", + message="coverage failed", + ), + operation="profile_results_validation", + service_id="records_manager", + ) + ] + system_controller._print_post_benchmark_info_and_metrics = AsyncMock() + system_controller._print_exit_errors_and_log_file = MagicMock() + + await system_controller._report_post_shutdown_results_and_errors() + + system_controller._print_post_benchmark_info_and_metrics.assert_awaited_once() + system_controller._print_exit_errors_and_log_file.assert_called_once() + @pytest.mark.asyncio async def test_system_controller_exits_on_profile_configure_error_response( self,