nirs4all.pipeline.dagml.in_process_runner module

In-process dag-ml execution (Mechanism B) — the perf-parity twin of cli_runner.

cli_runner.run_cv_refit_bundle drives dag-ml-cli as a SUBPROCESS: it writes the DSL / envelope / graph to disk, launches the process adapter, and reads back bundle.json. This module runs the SAME campaign IN-PROCESS through the PyO3 bridge (dag_ml._dag_ml.run_cv_refit_in_process) — no subprocess, no per-call import tax. The bridge owns the data path (the SAME envelope-based Rust InMemoryDataProvider the CLI uses, so the per-fold data_views/sample_ids are produced IDENTICALLY); only operator execution crosses back to Python, through an op_callback that closes over the SAME node_runner.run_node() the subprocess adapter calls.

The returned {node_results, scores} carries the native ScoreSet (cross-fold OOF average + the REFIT final/test reports) — byte-identical to the subprocess bundle.scores — so the caller maps it into a RunResult with the SAME score-extraction (result._scores_to_run_result()).

Parity contract: in-process scores == subprocess scores == legacy, for any pipeline the static subprocess path supports. The dataset / fold-children / sample-metadata are loaded with the EXACT same logic the subprocess adapter uses (process_adapter._build_handler()), so identity (and therefore the wire ids) match — the only divergence is that the materialization happens in THIS process, not a child. Generated-view campaigns can also run in an isolated Python worker through generated_subprocess; the static CLI has no generated-view callback.

nirs4all.pipeline.dagml.in_process_runner.run_cv_refit_bundle(*, dsl: dict[str, Any], envelope: dict[str, Any], graph: dict[str, Any], dataset_path: str, selection_metric: str = 'rmse', sample_metadata: dict[str, dict[str, Any]] | None = None, dataset_pickle: str | None = None, dataset: Any | None = None, fold_children: dict[str, dict[int, list[int]]] | None = None, fold_feature_views: dict[str, tuple[Any, dict[int, int], set[int]]] | None = None, random_state: int | None = None, refit: bool = True, refit_top_k: int = 1) → dict[str, Any]

Run a CV+refit bundle IN-PROCESS; return {returncode, stdout, results, scores}.

Drop-in for cli_runner.run_cv_refit_bundle() (same keyword signature minus the subprocess-only workdir / dagml_cli / venv_python): the router can dispatch to either by the env flag.

• scores — the native ScoreSet ({reports: [...]}), byte-identical to the subprocess bundle.scores the CLI emits; mapped by result._scores_to_run_result().

• results — the per-node NodeResult frames (FIT_CV validation/OOF + REFIT predictions), matching the subprocess results.jsonl capture.

• returncode / stdout — always 0 / "" on success; a bridge refusal raises (DagMlError) instead of returning a non-zero code, so the caller’s returncode != 0 guard is a no-op here (success-path parity).

dataset is the host’s ALREADY-MATERIALIZED SpectroDataset (with fold_children and optional fold_feature_views for a fold-local augmentation run): when given, the resolver is built from it directly — no disk reload. run_via_dagml already materialized this exact dataset (its identity fingerprint equals the reloadable path’s, verified in dataset._dataset_inputs()) and, for augmentation / rep-fusion, it IS the object that was pickled for the subprocess — so the resolver is byte-identical to the reload/pickle path. When dataset is None the dataset is loaded per _load_dataset().

nirs4all.pipeline.dagml.in_process_runner.run_cv_refit_bundle_router(*, dsl: dict[str, Any], envelope: dict[str, Any], graph: dict[str, Any], dataset_path: str, workdir: Any, dagml_cli: str, venv_python: str, selection_metric: str = 'rmse', sample_metadata: dict[str, dict[str, Any]] | None = None, dataset_pickle: str | None = None, dataset: Any | None = None, fold_children: dict[str, dict[int, list[int]]] | None = None, fold_feature_views: dict[str, tuple[Any, dict[int, int], set[int]]] | None = None, random_state: int | None = None, refit: bool = True, refit_top_k: int = 1) → dict[str, Any]

Route CV/refit to the direct binding, static CLI, or generated Python worker.

All branches return the SAME outcome shape — {returncode, stdout, results, scores} — so the _run_* call sites read outcome["scores"] uniformly (no bundle.json re-read). The in-process branch (the DEFAULT since the ADR-17 cutover — in_process_enabled()) drives run_cv_refit_bundle() here and gets scores directly from the bridge. The subprocess branch (Mechanism A) drives cli_runner.run_cv_refit_bundle() and lifts scores from the bundle.json the CLI wrote.

MECHANISM CASCADE (ADR-17): try in-process when it is selected AND the dag_ml._dag_ml extension loads (_dagml_extension_loads()); if the extension does NOT load (a wheel install missing / with a broken .so), fall through to the subprocess CLI; the subprocess branch then requires the dag-ml-cli binary and raises DagMlUnavailable when it is missing too (NEITHER mechanism available → run() falls back to legacy with a warning). The dag-ml-cli existence check lives HERE, in the subprocess branch ONLY — so an in-process-capable wheel never needs the sibling CLI.

Crucially the extension probe gates ONLY the import/LOAD, never the campaign: a real in-process runtime/operator error from run_cv_refit_bundle() propagates untouched (never silently rerouted to subprocess or swallowed into legacy).

Keeping workdir / dagml_cli / venv_python in the signature lets the call sites pass the SAME kwargs to either path; the in-process branch ignores those subprocess-only inputs.

dataset (the host’s already-materialized SpectroDataset, with fold_children for a fold-local augmentation run) lets the direct binding build its resolver without reloading it. The static CLI ignores it and re-materializes from dataset_path / dataset_pickle; the generated worker uses its view store but reloads the fixed PLAN dataset separately.

random_state seeds the static CLI adapter as before. The direct host is seeded by run_via_dagml; the generated worker seeds its own Python RNG and passes the same seed to DAG-ML’s control RNG. Generated views never use the static CLI because it has no callback or receipt transport.