refactor(accuracy): re-home results processor as accuracy accumulator

docs/accuracy/accuracy-benchmarking.md

@@ -252,7 +252,7 @@ CSV file: `<artifact_dir>/accuracy_results.csv`
 ```text
 AccuracyDatasetLoader          → Conversation/Turn objects (dataset pipeline)
 AccuracyRecordProcessor        → grades each response (record pipeline)
-AccuracyResultsProcessor       → aggregates per-task accuracy (results pipeline)
+AccuracyAccumulator            → aggregates per-task accuracy (accumulator pipeline)
 AccuracyConsoleExporter         → Rich table output
 AccuracyDataExporter            → CSV export
 ```

docs/accuracy/accuracy_stubs.md

@@ -32,7 +32,7 @@ graph TD
     A --> C[AccuracyGrader<br/>4 graders<br/>grade + extract]
     B --> D[AccuracyRecordProcessor<br/>process_record]
     C --> D
-    D --> E[AccuracyResultsProcessor<br/>process_result<br/>summarize]
+    D --> E[AccuracyAccumulator<br/>process_record<br/>summarize]
     E --> F[AccuracyConsoleExporter<br/>AccuracyDataExporter]
 ```
 
@@ -210,22 +210,22 @@ async def process_record(
 
 **Reference implementation:** `MetricRecordProcessor` in `src/aiperf/post_processors/metric_record_processor.py`
 
-### AccuracyResultsProcessor — IMPLEMENTED in PR #815
+### AccuracyAccumulator — IMPLEMENTED in PR #815 (re-homed from AccuracyResultsProcessor)
 
-**File:** `src/aiperf/accuracy/accuracy_results_processor.py`
+**File:** `src/aiperf/accuracy/accuracy_accumulator.py`
 **Parent:** `AIPerfLifecycleMixin`
-**Implements:** `ResultsProcessorProtocol`
-**Plugin key:** `accuracy_results` (under `results_processor`)
+**Implements:** accumulator `process_record` / `summarize` contract (see `AccumulatorProtocol`)
+**Plugin key:** `accuracy_results` (under `accumulator`, `record_types: [metric_records]`)
 **Disables via:** `PostProcessorDisabled` when `not cfg.accuracy.enabled`
 
 This class is fully implemented and serves as the canonical reference for aggregating per-task accuracy metrics.
 
 ```python
-async def process_result(self, record_data: MetricRecordsData) -> None         # IMPLEMENTED in PR #815
-async def summarize(self) -> list[MetricResult]                                # IMPLEMENTED in PR #815
+async def process_record(self, record_data: MetricRecordsData) -> None                # IMPLEMENTED in PR #815
+async def summarize(self, ctx: SummaryContext | None = None) -> list[MetricResult]    # IMPLEMENTED in PR #815
 ```
 
-**Reference implementation:** `MetricResultsProcessor` in `src/aiperf/post_processors/metric_results_processor.py`
+**Reference implementation:** `MetricsAccumulator` in `src/aiperf/metrics/accumulator.py`
 
 ---
 
@@ -287,7 +287,7 @@ All stubs are registered in `src/aiperf/plugin/plugins.yaml` and `src/aiperf/plu
 | Category | Plugin Key | Class |
 |----------|-----------|-------|
 | `record_processor` | `accuracy_record` | `AccuracyRecordProcessor` |
-| `results_processor` | `accuracy_results` | `AccuracyResultsProcessor` |
+| `accumulator` | `accuracy_results` | `AccuracyAccumulator` |
 | `console_exporter` | `accuracy` | `AccuracyConsoleExporter` |
 | `data_exporter` | `accuracy_csv` | `AccuracyDataExporter` |
 
@@ -302,7 +302,7 @@ All stubs are registered in `src/aiperf/plugin/plugins.yaml` and `src/aiperf/plu
 | Graders | 7 (all) | 0 | — | 0 |
 | Benchmarks | 9 (all) | 0 | — | 0 |
 | Record Processor | 1 (`AccuracyRecordProcessor`) | 0 | — | 0 |
-| Results Processor | 1 (`AccuracyResultsProcessor`) | 0 | — | 0 |
+| Accumulator | 1 (`AccuracyAccumulator`) | 0 | — | 0 |
 | Console Exporter | 1 (`AccuracyConsoleExporter`) | 0 | — | 0 |
 | Data Exporter | 1 (`AccuracyDataExporter`) | 0 | — | 0 |
 | Stub-plugin Validator | 1 (`AccuracyConfig._reject_stub_plugins`, idle until next stub) | 0 | — | 0 |
@@ -323,7 +323,7 @@ The processors, exporters, all seven graders, and all nine benchmarks are wired
 | **Canonical grader** | `src/aiperf/accuracy/graders/multiple_choice.py` |
 | **Canonical benchmark** | `src/aiperf/accuracy/benchmarks/mmlu.py` |
 | **Canonical record processor** | `src/aiperf/accuracy/accuracy_record_processor.py` |
-| **Canonical results processor** | `src/aiperf/accuracy/accuracy_results_processor.py` |
+| **Canonical accumulator** | `src/aiperf/accuracy/accuracy_accumulator.py` |
 | **Canonical console exporter** | `src/aiperf/accuracy/accuracy_console_exporter.py` |
 | **Canonical data exporter** | `src/aiperf/accuracy/accuracy_data_exporter.py` |
 | Disabled exception pattern | `src/aiperf/post_processors/raw_record_writer_processor.py:47` |

src/aiperf/accuracy/accuracy_results_processor.py → src/aiperf/accuracy/accuracy_accumulator.py

@@ -17,25 +17,30 @@
 from aiperf.common.models import MetricResult
 
 if TYPE_CHECKING:
+    from aiperf.common.accumulator_protocols import SummaryContext
     from aiperf.common.messages.inference_messages import MetricRecordsData
     from aiperf.common.models.dataset_models import DatasetMetadata
     from aiperf.config.resolution.plan import BenchmarkRun
 
 
-class AccuracyResultsProcessor(AIPerfLifecycleMixin):
-    """Results processor for accuracy benchmarking.
+class AccuracyAccumulator(AIPerfLifecycleMixin):
+    """Metric-records accumulator for accuracy benchmarking.
 
-    Receives task names via on_dataset_configured (called by RecordsManager
-    when DatasetConfiguredNotification arrives). Accumulates per-record grading
-    results from AccuracyRecordProcessor, then summarizes into per-task and
-    overall accuracy MetricResult objects.
+    Registered as ``accumulator:accuracy_results`` with ``record_types:
+    [metric_records]``, so RecordsManager dispatches every inference metric
+    record to :meth:`process_record` and folds :meth:`summarize`'s output into
+    ``ProfileResults.records``. Receives task names via
+    :meth:`on_dataset_configured` (called by RecordsManager when
+    DatasetConfiguredNotification arrives, before any records are processed),
+    accumulates the per-record grading results stamped by AccuracyRecordProcessor,
+    then summarizes into per-task and overall accuracy MetricResult objects.
     """
 
     def __init__(self, run: BenchmarkRun, **kwargs: Any) -> None:
         acc_cfg = run.cfg.accuracy
         if acc_cfg is None or not acc_cfg.enabled:
             raise PostProcessorDisabled(
-                "Accuracy results processor is disabled: accuracy mode is not enabled"
+                "Accuracy accumulator is disabled: accuracy mode is not enabled"
             )
 
         super().__init__(**kwargs)
@@ -54,15 +59,15 @@ def on_dataset_configured(self, metadata: DatasetMetadata) -> None:
 
         Called by RecordsManager before any records are processed. Builds the
         ordered list of task names from ConversationMetadata so that
-        process_result can bucket results without re-loading the benchmark.
+        process_record can bucket results without re-loading the benchmark.
         """
         self._tasks = [
             c.accuracy_task
             for c in metadata.conversations
             if c.accuracy_task is not None
         ]
 
-    async def process_result(self, record_data: MetricRecordsData) -> None:
+    async def process_record(self, record_data: MetricRecordsData) -> None:
         """Accumulate per-task accuracy counts from a single record's metrics.
 
         Reads ``accuracy.correct`` from ``record_data.metrics`` (produced by
@@ -74,8 +79,8 @@ async def process_result(self, record_data: MetricRecordsData) -> None:
         """
         if self._tasks is None:
             raise RuntimeError(
-                "AccuracyResultsProcessor: dataset not configured; "
-                "on_dataset_configured must be called before process_result"
+                "AccuracyAccumulator: dataset not configured; "
+                "on_dataset_configured must be called before process_record"
             )
         metrics = record_data.metrics
         correct = metrics.get("accuracy.correct")
@@ -98,9 +103,12 @@ async def process_result(self, record_data: MetricRecordsData) -> None:
         if is_unparsed:
             self._task_unparsed[task] += 1
 
-    async def summarize(self) -> list[MetricResult]:
+    async def summarize(self, ctx: SummaryContext | None = None) -> list[MetricResult]:
         """Return overall and per-task accuracy and unparsed counts as MetricResult list.
 
+        Conforms to the accumulator summarize contract (``ctx`` is accepted for
+        protocol compatibility and unused; this accumulator owns its own state).
+
         Emits:
         - ``accuracy.overall``: overall correct/total ratio
         - ``accuracy.task.<name>``: per-task correct/total ratio (sorted alphabetically)

src/aiperf/dataset/loader/accuracy_dataset_loader.py

@@ -12,7 +12,7 @@
 BenchmarkProblem i. Each Conversation carries accuracy_ground_truth and
 accuracy_task so that DatasetManager can propagate them through
 ConversationMetadata inside DatasetConfiguredNotification. Processors
-(AccuracyRecordProcessor, AccuracyResultsProcessor) receive these values
+(AccuracyRecordProcessor, AccuracyAccumulator) receive these values
 from the notification instead of independently re-loading the benchmark.
 The session_num % len(conversations) mapping handles both single-pass and
 multi-pass (num_requests > dataset size) runs and is only valid when the

src/aiperf/plugin/enums.py

@@ -79,7 +79,7 @@
 
 ResultsProcessorTypeStr: TypeAlias = str
 ResultsProcessorType = plugins.create_enum(PluginType.RESULTS_PROCESSOR, "ResultsProcessorType", module=__name__)
-"""Dynamic enum for results processor. Example: ResultsProcessorType.ACCURACY_RESULTS, ResultsProcessorType.NETWORK_LATENCY_JSONL_WRITER, ResultsProcessorType.TIMESLICE"""
+"""Dynamic enum for results processor. Example: ResultsProcessorType.GPU_TELEMETRY_ACCUMULATOR, ResultsProcessorType.OTEL_METRICS_STREAMER, ResultsProcessorType.TIMESLICE"""
 
 GPUTelemetryProcessorTypeStr: TypeAlias = str
 GPUTelemetryProcessorType = plugins.create_enum(PluginType.GPU_TELEMETRY_PROCESSOR, "GPUTelemetryProcessorType", module=__name__)
@@ -91,7 +91,7 @@
 
 AccumulatorTypeStr: TypeAlias = str
 AccumulatorType = plugins.create_enum(PluginType.ACCUMULATOR, "AccumulatorType", module=__name__)
-"""Dynamic enum for accumulator. Example: AccumulatorType.GPU_TELEMETRY, AccumulatorType.METRIC_RESULTS, AccumulatorType.SERVER_METRICS"""
+"""Dynamic enum for accumulator. Example: AccumulatorType.ACCURACY_RESULTS, AccumulatorType.METRIC_RESULTS, AccumulatorType.SERVER_METRICS"""
 
 StreamExporterTypeStr: TypeAlias = str
 StreamExporterType = plugins.create_enum(PluginType.STREAM_EXPORTER, "StreamExporterType", module=__name__)

src/aiperf/plugin/plugins.yaml

@@ -1023,12 +1023,6 @@ results_processor:
       time slices, enabling time-series analysis of benchmark performance.
       Enabled when timeslice config is set.
 
-  accuracy_results:
-    class: aiperf.accuracy.accuracy_results_processor:AccuracyResultsProcessor
-    description: |
-      Accuracy results processor that aggregates grading results and computes
-      summary accuracy metrics. Self-disables when accuracy mode is off.
-
 # =============================================================================
 # Accumulators
 # =============================================================================
@@ -1065,6 +1059,16 @@ accumulator:
     metadata:
       record_types: [server_metrics]
 
+  accuracy_results:
+    class: aiperf.accuracy.accuracy_accumulator:AccuracyAccumulator
+    description: |
+      Accuracy accumulator that ingests per-record grading results
+      (accuracy.correct / accuracy.unparsed stamped by AccuracyRecordProcessor)
+      and summarizes per-task and overall accuracy MetricResults into
+      ProfileResults.records. Self-disables when accuracy mode is off.
+    metadata:
+      record_types: [metric_records]
+
 # =============================================================================
 # Stream Exporters
 # =============================================================================

tests/unit/accuracy/test_accuracy_accumulator_protocol_path.py

@@ -0,0 +1,115 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Drive AccuracyAccumulator through the registered plugin protocol path.
+
+Mirrors how RecordsManager actually uses the accumulator: class resolution
+via ``accumulator:accuracy_results``, construction with the exact kwargs
+``load_accumulators`` passes, record-type routing via plugin metadata, and
+summarize invocation through ``generate_realtime_metrics`` (the engine's
+realtime path calls ``acc.summarize()`` with no arguments). Direct-method
+unit tests stay green when any of these seams drift; these tests do not.
+"""
+
+import pytest
+
+from aiperf.accuracy.accuracy_accumulator import AccuracyAccumulator
+from aiperf.common.exceptions import PostProcessorDisabled
+from aiperf.common.messages.inference_messages import MetricRecordsData
+from aiperf.common.models.dataset_models import ConversationMetadata, DatasetMetadata
+from aiperf.plugin import plugins
+from aiperf.plugin.enums import (
+    AccumulatorType,
+    AccuracyBenchmarkType,
+    DatasetSamplingStrategy,
+    EndpointType,
+    PluginType,
+)
+from aiperf.records.records_manager_processing import (
+    accumulators_for_record_type,
+    generate_realtime_metrics,
+)
+from tests.unit.conftest import make_benchmark_run
+from tests.unit.post_processors.conftest import create_metric_metadata
+
+
+def _make_registered_accumulator() -> AccuracyAccumulator:
+    AccumulatorClass = plugins.get_class(PluginType.ACCUMULATOR, "accuracy_results")
+    run = make_benchmark_run(
+        model_names=["test-model"],
+        endpoint_type=EndpointType.COMPLETIONS,
+        streaming=False,
+        accuracy={"benchmark": AccuracyBenchmarkType.MMLU},
+    )
+    # Exact kwarg shape used by records_manager_processing.load_accumulators.
+    return AccumulatorClass(service_id="records_manager", run=run, pub_client=None)
+
+
+class TestAccuracyAccumulatorProtocolPath:
+    def test_registry_resolves_accuracy_accumulator_class(self) -> None:
+        AccumulatorClass = plugins.get_class(PluginType.ACCUMULATOR, "accuracy_results")
+        assert AccumulatorClass is AccuracyAccumulator
+
+    def test_metadata_routes_to_metric_records_dispatch(self) -> None:
+        accumulator = _make_registered_accumulator()
+        accumulators = {AccumulatorType("accuracy_results"): accumulator}
+
+        assert accumulators_for_record_type(accumulators, "metric_records") == [
+            accumulator
+        ]
+        assert accumulators_for_record_type(accumulators, "gpu_telemetry") == []
+
+    def test_disabled_accuracy_raises_post_processor_disabled(self) -> None:
+        """load_accumulators skips PostProcessorDisabled - the self-disable contract."""
+        AccumulatorClass = plugins.get_class(PluginType.ACCUMULATOR, "accuracy_results")
+        run = make_benchmark_run(
+            model_names=["test-model"],
+            endpoint_type=EndpointType.COMPLETIONS,
+            streaming=False,
+        )
+
+        with pytest.raises(PostProcessorDisabled):
+            AccumulatorClass(service_id="records_manager", run=run, pub_client=None)
+
+    @pytest.mark.asyncio
+    async def test_process_record_and_summarize_through_engine_helper(self) -> None:
+        accumulator = _make_registered_accumulator()
+        accumulator.on_dataset_configured(
+            DatasetMetadata(
+                conversations=[
+                    ConversationMetadata(
+                        conversation_id="conv-0",
+                        accuracy_ground_truth="A",
+                        accuracy_task="algebra",
+                    ),
+                    ConversationMetadata(
+                        conversation_id="conv-1",
+                        accuracy_ground_truth="B",
+                        accuracy_task="history",
+                    ),
+                ],
+                sampling_strategy=DatasetSamplingStrategy.SEQUENTIAL,
+            )
+        )
+
+        await accumulator.process_record(
+            MetricRecordsData(
+                metadata=create_metric_metadata(session_num=0),
+                metrics={"accuracy.correct": 1.0, "accuracy.unparsed": 0.0},
+            )
+        )
+        await accumulator.process_record(
+            MetricRecordsData(
+                metadata=create_metric_metadata(session_num=1),
+                metrics={"accuracy.correct": 0.0, "accuracy.unparsed": 1.0},
+            )
+        )
+
+        # generate_realtime_metrics calls acc.summarize() exactly like the
+        # RecordsManager realtime path and flattens list-shaped results.
+        results = await generate_realtime_metrics([accumulator])
+
+        by_tag = {r.tag: r for r in results}
+        assert by_tag["accuracy.overall"].current == pytest.approx(0.5)
+        assert by_tag["accuracy.task.algebra"].current == pytest.approx(1.0)
+        assert by_tag["accuracy.task.history"].current == pytest.approx(0.0)
+        assert by_tag["accuracy.unparsed"].sum == 1

tests/unit/accuracy/test_accuracy_results_processor_summarize.py → tests/unit/accuracy/test_accuracy_accumulator_summarize.py

@@ -3,13 +3,13 @@
 
 import pytest
 
-from aiperf.accuracy.accuracy_results_processor import AccuracyResultsProcessor
+from aiperf.accuracy.accuracy_accumulator import AccuracyAccumulator
 from aiperf.plugin.enums import AccuracyBenchmarkType, EndpointType
 from tests.unit.conftest import make_benchmark_run
 
 
-def _make_processor() -> AccuracyResultsProcessor:
-    return AccuracyResultsProcessor(
+def _make_processor() -> AccuracyAccumulator:
+    return AccuracyAccumulator(
         run=make_benchmark_run(
             model_names=["test-model"],
             endpoint_type=EndpointType.COMPLETIONS,
@@ -20,7 +20,7 @@ def _make_processor() -> AccuracyResultsProcessor:
 
 
 @pytest.mark.asyncio
-class TestAccuracyResultsProcessorSummarize:
+class TestAccuracyAccumulatorSummarize:
     async def test_empty_returns_no_results(self) -> None:
         processor = _make_processor()
         results = await processor.summarize()