feat: add single-point prediction Python API

README.md

@@ -127,6 +127,12 @@ aisimulate predict \
   --output-dir ./aisimulate-prediction
 ```
 
+### Python prediction API
+
+Use the same `prediction.yaml` with the
+[Python prediction API](docs/core-api.md#python-prediction-api).
+See the API reference for the runnable example, arguments and results.
+
 ### Dynamo-integrated prediction
 
 <!-- readme-check: dynamo-predict -->
@@ -203,6 +209,12 @@ aisimulate recommend \
   --output-dir ./aisimulate-recommendation
 ```
 
+### Python recommendation API
+
+Use the same `recommendation.yaml` with the
+[Python recommendation API](docs/core-api.md#python-recommendation-api).
+See the API reference for the runnable example, arguments and results.
+
 ### Dynamo-integrated recommendation
 
 <!-- readme-check: dynamo-recommend -->
@@ -310,6 +322,7 @@ Use the focused SDK documentation instead of treating CLI internals as public
 APIs:
 
 - [Estimator/FPE Python and Rust SDK](docs/perf-model/api/python.md)
+- [Prediction and recommendation Python APIs](docs/core-api.md#python-prediction-api)
 - [FPM self-service: onboard a model on target hardware](docs/perf-model/fpm-self-service/README.md)
 - [FPM self-service implementation and CLI reference](docs/perf-model/fpm-self-service/implementation.md)
 - [FPM self-service examples](docs/perf-model/fpm-self-service/examples.md)

docs/core-api.md

@@ -0,0 +1,146 @@
+<!--
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# Prediction and recommendation Python APIs
+
+These application APIs compile public configuration into Replay and Sweeper runs.
+For per-operation and forward-pass estimates, see the
+[performance-model Python API](perf-model/api/python.md). For direct runner
+control, see the [Replay Python API](replay/api/python.md).
+
+## Table of contents
+
+- [Python prediction API](#python-prediction-api)
+- [Python recommendation API](#python-recommendation-api)
+
+## Python prediction API
+
+Use `aisimulate.predict.run_prediction` to compile and execute one public
+`CorePredictionConfig`. It checks runner capabilities, creates and closes the
+runner, and returns a `PredictionResult` with the same summary the CLI renders.
+
+Use `prediction.yaml` from the [README prediction example](../README.md#predict-one-deployment).
+In the activated AISimulate environment, save the following as
+`predict_example.py` and run `python predict_example.py` from the directory
+containing the YAML file.
+
+```python
+from aisimulate.config.cli import CorePredictionConfig
+from aisimulate.predict import run_prediction
+from aisimulate.runner import EngineReplayRunnerFactory
+
+if __name__ == "__main__":
+    result = run_prediction(
+        CorePredictionConfig.from_yaml("prediction.yaml"),
+        stack="engine",
+        runner_factory=EngineReplayRunnerFactory(),
+    )
+    print(result.summary)
+```
+
+See the [prediction configuration guide](reference/cli.md) for timing and workload options.
+
+| Argument | Contract |
+| --- | --- |
+| `config` | Required validated `CorePredictionConfig`. |
+| `stack` | Required keyword naming the execution stack; use `"engine"` with `EngineReplayRunnerFactory`. |
+| `runner_factory` | Required keyword supplying a factory compatible with the compiled replay. |
+| `adapter_configs` | Optional raw adapter blocks keyed by section name. |
+| `providers` | Resolved config adapters keyed by `"<stack>.<section>"`; required for every supplied adapter block. Core config does not accept Router/Planner top-level sections. |
+| `execution_mode` | Defaults to `"offline"`; the selected runner must support the requested mode. The built-in engine stack is offline. |
+| `output_requirements` | Optional `ReplayOutputRequirements` from `aisimulate.sweeper.replay`. Omission enables raw-report capture except for analytical EPD. An explicit value replaces that default. |
+
+`PredictionResult` contains:
+
+- `summary`: merged prediction metrics, including normalized power fields.
+- `native`: the native report with the summary merged in, or a summary fallback
+  when the runner supplies no native report. Analytical EPD retains its metadata
+  and approximation semantics instead of a token-replay report.
+- `replay_spec`: the compiled specification that was executed.
+- `report`: the runner's original `ReplayReport`, including metrics and metadata.
+
+After creating a runner, the call closes it even when execution fails. Runner
+execution exceptions are wrapped in `PredictionExecutionError`, preserving the
+cause and any `fpm_query_coverage` attribute. `KeyboardInterrupt` and
+`ResourceLimitError` propagate unchanged; configuration, compilation, capability
+and runner-creation failures also propagate directly.
+
+The call does not create output directories, save report files or print results.
+The caller owns those actions and resource budgets. Passing the plain engine
+factory does not enable the CLI's host-memory admission or subprocess supervision;
+use the CLI for the automatic [local resource controls](reference/local-resources.md).
+
+Source: [prediction entry point](../python/aisimulate/src/aisimulate/predict.py).
+
+## Python recommendation API
+
+Use `aisimulate.recommend.run_recommendation` to search a public
+`CoreRecommendationConfig` and return a `SweepResult`. This entry point lowers
+the public configuration into the [Sweeper SDK](sweeper/sdk.md); it does not
+accept a `SmartSearchConfig` in place of `CoreRecommendationConfig`.
+
+Use `recommendation.yaml` from the [README recommendation example](../README.md#recommend-a-deployment).
+In the activated AISimulate environment, save the following as
+`recommend_example.py` and run `python recommend_example.py` from the directory
+containing the YAML file.
+
+Keep the main guard because recommendation runs in supervised subprocesses.
+
+```python
+from aisimulate.config.cli import CoreRecommendationConfig
+from aisimulate.recommend import run_recommendation
+from aisimulate.runner import EngineReplayRunnerFactory
+
+if __name__ == "__main__":
+    result = run_recommendation(
+        CoreRecommendationConfig.from_yaml("recommendation.yaml"),
+        stack="engine",
+        runner_factory=EngineReplayRunnerFactory(),
+        show_progress=False,
+    )
+    print(result.counts)
+    for candidate in result.selected_candidates:
+        print(candidate.score, candidate.used_gpus, candidate.prediction_config)
+```
+
+| Argument | Contract |
+| --- | --- |
+| `config` | Required validated `CoreRecommendationConfig`. |
+| `stack` | Required keyword naming the execution stack; use `"engine"` with `EngineReplayRunnerFactory`. |
+| `runner_factory` | Required keyword supplying the candidate replay factory. |
+| `adapter_configs` | Optional raw adapter blocks keyed by section name. |
+| `providers` | Resolved config adapters keyed by `"<stack>.<section>"`; required for every supplied adapter block. |
+| `output_configs` | Optional output-adapter configuration blocks for candidate and round callbacks. |
+| `afd_performance_model` | Optional performance-model provider for analytical AFD search. |
+| `show_progress` | Defaults to `True`; set `False` to disable progress display. |
+| `output_requirements` | Optional `ReplayOutputRequirements` forwarded to each candidate replay. |
+
+`SweepResult` contains:
+
+- `candidates`: candidate records, including status, metrics, score, provenance,
+  concrete prediction configuration and any rejection or failure reason.
+- `counts`: totals for evaluated, feasible, failed, resource-limited and other
+  candidate outcomes. A returned result does not imply that every candidate succeeded.
+- `selected_candidate_ids` and `selected_candidates`: selected ledger IDs and
+  corresponding `Candidate` objects in the same order. Each selected candidate's
+  `prediction_config` can be used for a prediction.
+- `execution_resources`: host-resource supervision evidence for this run.
+
+Call `result.to_json()` to serialize the sweep result. Without output adapters,
+this API does not publish the CLI's recommendation directory or YAML files.
+
+Recommendation applies host-resource admission and subprocess supervision,
+limits worker concurrency and cleans up its workers. Put calls behind
+`if __name__ == "__main__":` in scripts; factories and providers must be
+pickleable. See [local resource controls](reference/local-resources.md) for budgets,
+resource-limited candidate results and bounded partial evidence.
+
+Invalid core input fails schema validation. Supervised execution failures raise
+`RuntimeError`; host-resource refusal or interruption raises `ResourceLimitError`,
+and cancellation raises `KeyboardInterrupt`. Individual candidate failures are
+recorded in the result rather than necessarily stopping the entire search.
+
+Source: [recommendation entry point](../python/aisimulate/src/aisimulate/recommend.py)
+and [result contract](../python/aisimulate/src/aisimulate/sweeper/result.py).

python/aisimulate/src/aisimulate/main.py

@@ -10,13 +10,13 @@
 import sys
 from collections.abc import Sequence
 from copy import deepcopy
+from pathlib import Path
 from typing import Any
 
 from pydantic import ValidationError
 
 from .afd_artifacts import write_afd_qualification_artifacts
 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,
     CoreRecommendationConfig,
@@ -29,7 +29,7 @@
     SimulationConfigAdapter,
     resolve_config_adapters,
 )
-from .detail import build_prediction_details, energy_diagnostics, prediction_summary
+from .detail import build_prediction_details, energy_diagnostics
 from .output import (
     format_prediction_stdout,
     format_recommendation_stdout,
@@ -48,6 +48,7 @@
     write_output_adapters,
 )
 from .power import normalize_power_summary
+from .predict import PredictionExecutionError, run_prediction
 from .resources import (
     GuardedRunnerFactory,
     ResourceLimitError,
@@ -65,6 +66,60 @@ class _CliExecutionError(RuntimeError):
     pass
 
 
+def _write_incomplete_fpm_coverage(root: Path, error: BaseException) -> None:
+    coverage = getattr(error, "fpm_query_coverage", None)
+    try:
+        if isinstance(coverage, str):
+            coverage = json.loads(coverage)
+        if isinstance(coverage, dict):
+            write_fpm_coverage(root, {**coverage, "status": "incomplete", "error": str(error)})
+    except (ValueError, OSError) as write_error:
+        sys.stderr.write(f"could not save FPM coverage: {write_error}\n")
+
+
+class _PredictionCliRunner:
+    def __init__(self, runner, output_root: Path, mark_execution_ready, mark_shutdown) -> None:
+        self._runner = runner
+        self._output_root = output_root
+        self._mark_execution_ready = mark_execution_ready
+        self._mark_shutdown = mark_shutdown
+
+    def run(self, spec, *, output_requirements=None):
+        try:
+            return self._runner.run(spec, output_requirements=output_requirements)
+        except BaseException as exc:
+            # Persist native query evidence before close() can replace the
+            # execution error.
+            _write_incomplete_fpm_coverage(self._output_root, exc)
+            raise
+
+    def close(self) -> None:
+        self._mark_shutdown()
+        try:
+            self._runner.close()
+        finally:
+            self._mark_execution_ready()
+
+
+class _PredictionCliRunnerFactory:
+    def __init__(self, factory, output_dir, overwrite, mark_execution_ready, mark_shutdown) -> None:
+        self._factory = factory
+        self._output_dir = output_dir
+        self._overwrite = overwrite
+        self._mark_execution_ready = mark_execution_ready
+        self._mark_shutdown = mark_shutdown
+        self.output_root: Path | None = None
+
+    def capabilities(self):
+        return self._factory.capabilities()
+
+    def create(self, worker_id):
+        self.output_root = prepare_output_directory(self._output_dir, overwrite=self._overwrite)
+        runner = self._factory.create(worker_id)
+        self._mark_execution_ready()
+        return _PredictionCliRunner(runner, self.output_root, self._mark_execution_ready, self._mark_shutdown)
+
+
 def _resolve_section_adapters(sections: dict[str, dict[str, Any]], stack: str) -> dict[str, SimulationConfigAdapter]:
     adapters = resolve_config_adapters(f"{stack}.{section}" for section in sections)
     for name, adapter in adapters.items():
@@ -113,63 +168,35 @@ def _predict(args: argparse.Namespace, raw: dict[str, Any], factory) -> int:
     if epd and (args.stack != "engine" or args.online or args.capture_per_request or adapter_raw):
         raise ValueError("analytical EPD requires offline --stack engine without adapters or per-request capture")
     adapters = _resolve_section_adapters(adapter_raw, args.stack)
-    adapter_specs = _compile_prediction_adapters(
-        adapter_raw,
-        adapters,
-        stack=args.stack,
-        context=_prediction_adapter_context(config),
+    factory = _PredictionCliRunnerFactory(
+        factory,
+        args.output_dir,
+        args.overwrite,
+        mark_execution_ready,
+        mark_shutdown,
     )
-    spec = prediction_to_replay_spec(
+    result = run_prediction(
         config,
-        adapter_specs=adapter_specs,
+        adapter_configs=adapter_raw,
+        stack=args.stack,
+        runner_factory=factory,
+        providers=adapters,
         execution_mode="online" if args.online else "offline",
+        output_requirements=ReplayOutputRequirements(
+            include_raw_report=not epd,
+            capture_per_request=args.capture_per_request,
+            capture_memory_diagnostics="memory" in args.detail,
+            capture_performance_diagnostics=bool({"time", "source"}.intersection(args.detail)),
+        ),
     )
-    factory.capabilities().require_compatible(spec)
-    root = prepare_output_directory(args.output_dir, overwrite=args.overwrite)
-    runner = factory.create(0)
-    mark_execution_ready()
-    try:
-        try:
-            report = runner.run(
-                spec,
-                output_requirements=ReplayOutputRequirements(
-                    include_raw_report=not epd,
-                    capture_per_request=args.capture_per_request,
-                    capture_memory_diagnostics="memory" in args.detail,
-                    capture_performance_diagnostics=bool({"time", "source"}.intersection(args.detail)),
-                ),
-            )
-        except BaseException as exc:
-            coverage = getattr(exc, "fpm_query_coverage", None)
-            if isinstance(coverage, str):
-                coverage = json.loads(coverage)
-            if isinstance(coverage, dict):
-                write_fpm_coverage(root, {**coverage, "status": "incomplete", "error": str(exc)})
-            if isinstance(exc, (KeyboardInterrupt, ResourceLimitError)) or not isinstance(exc, Exception):
-                raise
-            raise _CliExecutionError(f"{type(exc).__name__}: {exc}") from exc
-    finally:
-        mark_shutdown()
-        runner.close()
-        mark_execution_ready()
-    native = report.metadata.get("native_report")
-    if not isinstance(native, dict):
-        native = {"summary": dict(report.metrics)}
-    if epd:
-        native = {"summary": dict(report.metrics), "metadata": dict(report.metadata)}
-        if "memory_diagnostics" in native["metadata"]:
-            native["memory_diagnostics"] = native["metadata"].pop("memory_diagnostics")
-        # JSON stdout, like prediction.json, must identify the approximation.
-        native["summary"]["metric_semantics"] = report.metadata["metric_semantics"]
-        native["summary"]["total_gpus"] = report.metadata["total_gpus"]
+    root = factory.output_root
+    if root is None:
+        raise RuntimeError("prediction runner was not created")
+    spec = result.replay_spec
+    summary = result.summary
+    native = result.native
     if isinstance(native.get("fpm_query_coverage"), dict):
         write_fpm_coverage(root, native["fpm_query_coverage"])
-    summary = prediction_summary(native)
-    summary.update(normalize_power_summary(report.metrics))
-    if "summary" in native:
-        native = {**native, "summary": summary}
-    else:
-        native = {**native, **summary}
     power_diagnostics = None
     if args.diagnostics == "power":
         power_diagnostics = energy_diagnostics(native)
@@ -200,7 +227,7 @@ def _predict(args: argparse.Namespace, raw: dict[str, Any], factory) -> int:
         if any(not isinstance(record, dict) for record in records):
             raise RuntimeError("per-request records must be JSON mappings")
         write_requests(root, records)
-    phases = report.metadata.get("agentic_phases")
+    phases = result.report.metadata.get("agentic_phases")
     if isinstance(phases, dict) and phases.get("phase") == "aborted":
         sys.stderr.write(
             f"ERROR: agentic preparation aborted: {phases.get('failure_reason') or 'preparation did not complete'}; "
@@ -396,7 +423,7 @@ def main(argv: Sequence[str] | None = None) -> int:
         except (OSError, ValueError) as output_error:
             sys.stderr.write(f"could not save resource plan: {output_error}\n")
         return 3
-    except _CliExecutionError as exc:
+    except PredictionExecutionError as exc:
         sys.stderr.write(f"aisimulate {args.command} failed: {exc}\n")
         return 1
     except Exception as exc: