feat(agentx): fail incomplete profiling metric coverage / 检测 AgentX 分析指标覆盖不完整

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

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`),

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",

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."""

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):

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,
 )

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: (
@@ -1088,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
@@ -1117,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()

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
 
 

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(

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)