feat: add output adapters

.github/codeowners/areas.yaml

@@ -57,6 +57,7 @@ areas:
       - tests/test_epd_cli.py
       - examples/cli/
       - tests/test_stack_discovery.py
+      - tests/test_output_adapter_discovery.py
       - tests/test_power_diagnostics.py
       - tests/test_unified_cli.py
       - tests/test_ngram_cli.py
@@ -197,6 +198,8 @@ shared:
     owners: [replay, sweeper, forward-pass-engine, maintainers]
   - glob: tests/test_stack_discovery.py
     owners: [replay, sweeper, maintainers]
+  - glob: tests/test_output_adapter_discovery.py
+    owners: [replay, sweeper, forward-pass-engine, maintainers]
   - glob: python/aisimulate/src/aisimulate/resources.py
     owners: [replay, sweeper, maintainers]
   - glob: python/aisimulate/src/aisimulate/resource_scheduler.py

CODEOWNERS

@@ -78,6 +78,7 @@
 /tests/test_state_cache_config.py                        @ai-dynamo/aisimulate-replay-codeowners
 /tests/fixtures/power-contract-v1.json                   @ai-dynamo/aisimulate-replay-codeowners
 /tests/test_unified_traffic_runtime.py                   @ai-dynamo/aisimulate-replay-codeowners
+/tests/test_output_adapter_discovery.py                  @ai-dynamo/aisimulate-replay-codeowners
 
 # === mocker  (@ai-dynamo/aisimulate-mocker-codeowners) ===
 /crates/core/src/engine/                                 @ai-dynamo/aisimulate-mocker-codeowners
@@ -204,6 +205,7 @@
 /tests/fixtures/power-contract-v1.json                   @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_spline_estimator_config.py                   @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /tests/test_unified_traffic_runtime.py                   @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/access-aisimulate-maintain
+/tests/test_output_adapter_discovery.py                  @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /python/aisimulate/src/aisimulate/replay/                @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/access-aisimulate-maintain
 /.claude/skills/aisimulate-fpe-performance               @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain
 /docs/schemas/power-metrics-v1.schema.json               @ai-dynamo/aisimulate-replay-codeowners @ai-dynamo/aisimulate-sweeper-codeowners @ai-dynamo/aisimulate-forward-pass-engine-codeowners @ai-dynamo/access-aisimulate-maintain

docs/cli/user-guide.md

@@ -84,6 +84,10 @@ warmup fields and retains both worker pools across the barrier. See
 | `predict` | Evaluate one concrete deployment under a workload. | `aisimulate predict -c prediction.yaml --output-dir ./prediction-output` | A metrics summary and `prediction-output/prediction.json`. |
 | `recommend` | Search deployment and load choices for an optimization goal. | `aisimulate recommend -c recommendation.yaml --output-dir ./recommendation-output` | Ranked configurations, `recommendation.json`, and concrete YAML files under `recommendations/`. |
 
+`recommend` also accepts repeatable `--output NAME` options. Each selected output adapter requires
+a same-named top-level configuration section, may observe live candidate and round notifications,
+and writes additional artifacts after recommendation.
+
 Unless labeled as captured output, metric values in example results are hypothetical and
 illustrate the output format. Captured detail examples are simulation results, not hardware measurements.
 
@@ -1883,7 +1887,10 @@ after adapter canonicalization and deduplication so it maps one-to-one to the nu
 Read [Understand your prediction](understand-your-prediction.md) for an annotated
 report, ITL/TPOT definitions, incomplete-request handling, and SLA interpretation.
 
-Output controls are CLI-only. They never appear in an input or recommended YAML file.
+Core output controls are CLI-only. A selected output adapter owns its same-named top-level input
+section; that section configures artifact generation and is excluded from recommended prediction
+YAML files. Output adapters receive the prepared output directory directly and must preserve
+unrelated files.
 
 Recommendation output uses the schema-versioned `SweepResult` contract documented in
 [`docs/sweeper/results.md`](../sweeper/results.md). It preserves run metadata, a candidate-attempt

docs/sweeper/architecture.md

@@ -52,6 +52,32 @@ The config-adapter ABI has three operations: compile a concrete prediction secti
 recommendation section into a search plan, and materialize one candidate. The legacy Sweeper
 provider ABI remains a separate SDK compatibility surface.
 
+Recommendation outputs are discovered independently through
+`aisimulate.output_adapters`. An output adapter name is an accepted `--output` value and identifies
+the same-named top-level configuration section. AISimulate removes explicitly selected output
+sections before validating simulation configuration. A selected adapter may subscribe to live
+candidate and round notifications. After the recommendation completes, AISimulate passes the
+adapter's section, the final `SweepResult`, and the prepared `--output-dir` to its writer. The
+adapter returns the relative paths it wrote. Output adapters do not affect simulation, ranking, or
+recommended prediction configurations.
+
+The output-adapter ABI has one required operation and one optional operation:
+
+```python
+adapter.subscribe(config)  # optional: RecommendationOutputCallbacks or None
+adapter.write(config, result=result, output_dir=output_dir)
+```
+
+Adapters are loaded only when selected. Names must be unique, implementations must declare the
+supported output-adapter API version, and reported paths must be relative to the supplied output
+directory and exist after the call. AISimulate writes canonical recommendation output before calling
+`write()`, and a writer failure makes the command fail. Adapters share that output directory and must
+not modify canonical files. `subscribe()` runs in the supervised recommendation worker and may
+return `on_candidate` and `on_round` callbacks. These callbacks execute synchronously on the search
+path, must return promptly, and abort the recommendation if they fail. The worker and final writer
+may use separate adapter instances, so plugins must not depend on shared in-memory state between
+callbacks and `write()`.
+
 See [Sweep Configuration Providers](sweep-config-provider.md) for the SDK provider contract.
 
 ## Sweep Flow

python/aisimulate/src/aisimulate/cli_args.py

@@ -5,6 +5,8 @@
 from __future__ import annotations
 
 import argparse
+import importlib.metadata
+from collections.abc import Sequence
 from pathlib import Path
 from typing import Any
 
@@ -76,6 +78,14 @@ def build_parser() -> argparse.ArgumentParser:
         action="store_true",
         help="pace prediction against the real wall clock instead of virtual time",
     )
+    subparsers.choices["recommend"].add_argument(
+        "--output",
+        dest="outputs",
+        action="append",
+        default=[],
+        metavar="NAME",
+        help="write additional artifacts using an installed output adapter",
+    )
     add_support_parser(subparsers)
     return parser
 
@@ -93,6 +103,35 @@ def _load_mapping(path: str) -> dict[str, Any]:
     return value
 
 
+def _extract_output_configs(
+    raw: dict[str, Any], outputs: Sequence[str], *, stack: str
+) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]:
+    """Validate selected output sections before supervision can clear old outputs."""
+
+    remaining = dict(raw)
+    configs: dict[str, dict[str, Any]] = {}
+    if not outputs:
+        return remaining, configs
+
+    from .config.common import CONFIG_ADAPTER_ENTRY_POINT_GROUP, RECOMMENDATION_CORE_SECTIONS
+
+    config_adapter_names = {
+        entry.name for entry in importlib.metadata.entry_points().select(group=CONFIG_ADAPTER_ENTRY_POINT_GROUP)
+    }
+    for name in dict.fromkeys(outputs):
+        if not name or "." in name:
+            raise _CliConfigError(f"invalid --output name {name!r}")
+        if name in RECOMMENDATION_CORE_SECTIONS or f"{stack}.{name}" in config_adapter_names:
+            raise _CliConfigError(f"output adapter name {name!r} collides with a recommendation input section")
+        value = remaining.pop(name, None)
+        if value is None:
+            raise _CliConfigError(f"--output {name!r} requires a top-level {name!r} configuration section")
+        if not isinstance(value, dict):
+            raise _CliConfigError(f"output section {name!r} must be a mapping")
+        configs[name] = value
+    return remaining, configs
+
+
 def _apply_overrides(data: dict[str, Any], overrides: list[str], *, command: str) -> None:
     for assignment in overrides:
         if "=" not in assignment:

python/aisimulate/src/aisimulate/config/common.py

@@ -215,6 +215,7 @@ def load_yaml(path: str | Path) -> dict[str, Any]:
     return data
 
 
+CONFIG_ADAPTER_ENTRY_POINT_GROUP = "aisimulate.config_adapters"
 PREDICTION_CORE_SECTIONS = frozenset({"traffic", "engine", "evaluation", "execution"})
 RECOMMENDATION_CORE_SECTIONS = frozenset({*PREDICTION_CORE_SECTIONS, "optimization", "optimizer"})
 

python/aisimulate/src/aisimulate/config_adapter.py

@@ -11,6 +11,7 @@
 from dataclasses import dataclass
 from typing import Any, Protocol, runtime_checkable
 
+from .config.common import CONFIG_ADAPTER_ENTRY_POINT_GROUP
 from .sweeper.provider import API_VERSION as SWEEP_PROVIDER_API_VERSION
 from .sweeper.provider import (
     AdapterReplaySpec,
@@ -21,7 +22,6 @@
 )
 
 CONFIG_ADAPTER_API_VERSION = 3
-CONFIG_ADAPTER_ENTRY_POINT_GROUP = "aisimulate.config_adapters"
 
 
 @dataclass(frozen=True)

python/aisimulate/src/aisimulate/main.py

@@ -14,7 +14,7 @@
 from pydantic import ValidationError
 
 from .afd_artifacts import write_afd_qualification_artifacts
-from .cli_args import _apply_overrides, _CliConfigError, _load_mapping, build_parser
+from .cli_args import _apply_overrides, _CliConfigError, _extract_output_configs, _load_mapping, build_parser
 from .compiler import prediction_to_replay_spec
 from .config.cli import (
     CorePredictionConfig,
@@ -40,6 +40,12 @@
     write_recommendations,
     write_requests,
 )
+from .output_adapter import (
+    OutputAdapterExecutionError,
+    OutputAdapterResolutionError,
+    resolve_output_adapters,
+    write_output_adapters,
+)
 from .power import normalize_power_summary
 from .resources import (
     GuardedRunnerFactory,
@@ -218,6 +224,8 @@ def _predict(args: argparse.Namespace, raw: dict[str, Any], factory) -> int:
 def _recommend(args: argparse.Namespace, raw: dict[str, Any], factory) -> int:
     from .recommend import run_recommendation
 
+    raw, output_configs = _extract_output_configs(raw, getattr(args, "outputs", []), stack=args.stack)
+    output_adapters = resolve_output_adapters(output_configs)
     core_raw, adapter_raw = split_config_sections(raw, command="recommend")
     config = CoreRecommendationConfig.model_validate(core_raw)
     adapters = _resolve_section_adapters(adapter_raw, args.stack)
@@ -227,6 +235,7 @@ def _recommend(args: argparse.Namespace, raw: dict[str, Any], factory) -> int:
         stack=args.stack,
         runner_factory=factory,
         providers=adapters,
+        output_configs=output_configs,
         show_progress=args.format == "table",
     )
     selected: list[tuple[str, Any, dict[str, Any]]] = []
@@ -268,6 +277,15 @@ def _recommend(args: argparse.Namespace, raw: dict[str, Any], factory) -> int:
         sys.stderr.write(f"no feasible candidate found; saved full result to: {result_path}\n")
         return 3 if getattr(result.counts, "resource_limited", 0) else 1
     paths = write_recommendations(root, [config for _, _, config in selected])
+    try:
+        write_output_adapters(
+            output_adapters,
+            output_configs,
+            result=result,
+            output_dir=root,
+        )
+    except OutputAdapterExecutionError as exc:
+        raise _CliExecutionError(str(exc)) from exc
     rows = []
     for index, ((_, candidate, _), path) in enumerate(zip(selected, paths, strict=True), start=1):
         row = {
@@ -329,6 +347,7 @@ def main(argv: Sequence[str] | None = None) -> int:
     except (
         _CliConfigError,
         ConfigAdapterResolutionError,
+        OutputAdapterResolutionError,
         ValidationError,
         ValueError,
     ) as exc: