feat(algorithms)!: unify PyTorch runtime contracts

docs/STABILITY.md

@@ -97,6 +97,7 @@ from the legacy setup-only propagation rule.
 | `tributo.algorithms.api.context` — `UserExecutionContext` | `alpha` | Restricted context for trusted module-qualified Worker functions |
 | `tributo.algorithms.api.errors` | `alpha` | Portable execution error taxonomy |
 | `tributo.algorithms.api.support` | `alpha` | Trusted Wheel support evidence, execution semantics, expiry, and revocation |
+| `tributo.algorithms.api.torch_runtime` | `alpha` | Versioned Torch Runtime helpers, Stage identity, checkpoint and reducer contracts |
 | `tributo.algorithms.conformance` | `alpha` | Descriptor-only and installed algorithm Wheel Conformance Testkit |
 | `tributo.algorithms.builtin.*` | `deprecated` | Production algorithms moved to the official `tributo-algorithms` Wheels; Core retains only public SPI and Ray runtimes |
 | `tributo.algorithms.core.builder` — `AlgorithmBuilder` | `alpha` | Provisional sklearn and Custom Ray Function registration builders |
@@ -105,7 +106,7 @@ from the legacy setup-only propagation rule.
 | `tributo.algorithms.spi.execution` | `alpha` | Provisional operation and Runtime execution protocols |
 | `tributo.algorithms.spi.contracts` | `alpha` | Executable algorithm contract validator protocol |
 | `tributo.algorithms.spi.input` | `alpha` | Two-stage input resolution and Driver/Worker ownership contracts |
-| `tributo.algorithms.spi.torch` | `alpha` | Narrow model/loss/optimizer/metric recipe contract lowered to Ray Train |
+| `tributo.algorithms.spi.torch` | `alpha` | Versioned TorchRecipe and RayTorchAdapter contracts |
 
 ### Data (tributo.data.*)
 
@@ -202,7 +203,7 @@ from the legacy setup-only propagation rule.
 | `tributo.integrations.validators.*` | `beta` | Built-in validator implementations |
 | `tributo.integrations.sources` | `beta` | Built-in source provider package |
 | `tributo.integrations.sources.*` | `beta` | Built-in source providers |
-| `tributo.integrations.sources.ray_torch_recipe` | `alpha` | Generic trusted Torch recipe checkpoint provider |
+| `tributo.integrations.sources.ray_torch` | `alpha` | Generic trusted Torch checkpoint provider |
 | `tributo.integrations.storage` | `beta` | Built-in storage adapter package |
 | `tributo.integrations.storage.*` | `beta` | Built-in storage backends |
 | `tributo.integrations.hooks` | `beta` | Built-in Hook package |

docs/architecture/ray-first-torch-recipes.md

@@ -1,11 +1,10 @@
 # Ray-first torch recipes
 
-`TorchTrainingRecipe` is the low-code path for scalar-column dense tabular
-PyTorch models.
-An algorithm package defines model, loss, optimizer, and metric factories.
-Tributo lowers that recipe to the existing formal collective runtime and keeps
-deployment, data movement, distributed coordination, checkpoint upload, and
-Bundle publication out of user code.
+Tributo exposes one versioned `TorchRecipe` for Core-owned training loops and
+one `RayTorchAdapter` for framework-owned loops. Both are selected by the
+`RAY_TRAIN_TORCH` strategy and executed by `tributo.ray_train_torch`.
+`TorchPolicy.execution_plan` is the single source of truth for routing, Stage
+order, checkpoint dependencies, state layout and final export Stage.
 
 ## Ray reuse audit
 
@@ -21,8 +20,8 @@ implementation and its public annotations, not unversioned latest docs.
 | Torch batches | Public `DataIterator.iter_torch_batches()` | Validate feature/label roles and pass bounded batch options |
 | Device and DDP | Stable `ray.train.torch.prepare_model()` | Reject unvalidated BatchNorm and GPU claims |
 | Metrics | PyTorch tensors and `torch.distributed` collectives | Apply the descriptor's existing `MetricReduction`; do not infer reducers from names |
-| Checkpoint transfer | Ray `Checkpoint`, `train.report()`, `RunConfig`, and retention | Write bounded model/optimizer/RNG metadata and validate epoch-boundary resume identity |
-| Run identity | V2 `RunConfig.name` and storage context | Derive a unique Ray run name from the existing Tributo `run_id` so independent runs cannot inherit each other's checkpoints |
+| Checkpoint transfer | Ray `Checkpoint`, `train.report()`, `RunConfig`, and retention | Write bounded model, optimizer, scheduler, gradient scaling, and RNG metadata at complete optimizer-window boundaries and validate Stage identity |
+| Run identity | V2 `RunConfig.name` and storage context | Derive a unique Ray run name from `run_id`, `invocation_id`, `stage_id`, and the Policy/execution-plan identity digest so retries reuse only the matching Stage directory |
 | Export and serving | Existing Torch ONNX exporter, ONNX Runtime validator, Bundle publisher, reader, batch inference, and Serve flavor | Reconstruct the trusted recipe model and create one `ExportSource` |
 | Local execution | `ray.init(address="local")` | Own and close only the runtime created by Tributo |
 | Existing or provisioned cluster | `ray.init(address="auto")`, Ray Jobs, KubeRay RayJob, or Cluster Launcher | Attach or expose the same entrypoint; workload code never provisions or deletes the cluster |
@@ -34,12 +33,18 @@ changes only the `equal` argument while preserving Ray execution options,
 training-resource exclusion, and locality hints. A future Ray public option
 should replace this adapter.
 
-V2 rejects a non-null `resume_from_checkpoint` argument as deprecated. Worker
-recovery within one Ray run still uses `train.get_checkpoint()`. For an
-explicit new-run resume, Tributo passes the validated, worker-visible epoch
-checkpoint path to the recipe and restores model, optimizer, and RNG state in
-the worker. This does not replace Ray's checkpoint upload, retention, or
-failure-recovery controller.
+Ray Train V2 rejects a non-null `resume_from_checkpoint` argument as deprecated.
+The Core Runtime never supplies it: failure retry uses `train.get_checkpoint()`;
+cross-Stage and cross-Run recovery use a credential-free Core Worker control
+envelope. A full `TorchRecoveryEnvelope` records completed Stage locators and an
+optional active Stage; completed Stage evidence is persisted in the validated
+checkpoint sidecar so a recovery-only invocation can produce the same Receipt.
+A typed progress manifest preserves per-rank coverage/metric prefixes, reducer
+observation, Dataset cursor and whether the current epoch's Scheduler step has
+already run.  Remote Stage payloads remain unavailable under a deterministic
+staging prefix until a matching commit marker is written.
+A Torch-only preflight and invocation-local one-shot lease complete before Ray
+or input resources are opened.
 
 ## Uneven input decision
 
@@ -54,11 +59,19 @@ The first implementation therefore uses a narrow dynamic alignment protocol:
 - all ranks exchange only the active row count before each step;
 - exhausted ranks run a zero-row DDP forward/backward and contribute zero
   gradient while active ranks retain every observed row;
-- loss scaling uses the global active row count, so uneven final batches are
-  sample weighted;
+- loss scaling uses the algorithm-declared global normalizer for the complete
+  accumulation window, so uneven final batches remain mathematically weighted
+  by the declared unit (rows, sample weights, or valid tokens);
 - no row is dropped or replayed;
-- all ranks report once per epoch and only rank zero attaches the replicated
-  checkpoint.
+- all ranks report at the declared `checkpoint_interval_windows` cadence (one
+  complete optimizer window by default), and only the checkpoint-owner rank
+  attaches the replicated checkpoint.
+
+Optional validation and test roles use the same rank-alignment rule.  When a
+present role has fewer rows than Workers, exhausted ranks execute a typed
+zero-row validation step and contribute zero metrics; all ranks still perform
+the same metric collectives.  A role that is absent remains explicitly absent,
+while an explicitly empty role is rejected by the route contract.
 
 This protocol supplements domain correctness. It does not replace Ray Dataset
 sharding, PyTorch DDP, process-group setup, or Ray checkpoint transport. Empty
@@ -77,10 +90,17 @@ original column names as separate ONNX inputs so Bundle inference does not lose
 the declared `InputBinding` contract. Vector, sequence, jagged, and graph inputs
 remain later profiles rather than implicit shape inference.
 
-Advanced models may override the bounded `forward()` and `compute_loss()`
-methods. Models that need custom training steps, multiple optimizers, framework
-callbacks, graph sampling, sharded embeddings, or framework-owned checkpoints
-use a framework adapter or the existing full worker-loop SPI.
+Advanced models implement the typed `TorchRecipe` hooks. Models that need
+custom training steps, framework callbacks, graph sampling, sharded embeddings,
+or framework-owned checkpoints use `RayTorchAdapter`; the Adapter still cannot
+create a nested Trainer or declare a second execution plan.
+
+An Adapter declares its `TorchArtifactPlan` through the Core Provider.  The
+Provider attaches that plan to the Adapter's `ExportSource` before invoking the
+single Core Bundle exporter, and rejects conflicting plan metadata.  Adapter
+worker configuration receives algorithm-owned values plus credential-free input
+binding metadata; Core Ray paths, recovery locators, and output publication
+URIs are never accepted in that JSON payload.
 
 Capabilities not described by this contract are not implied by the recipe,
 dependency set, or framework installation.

docs/how-to/custom-distributed-algorithms.md

@@ -217,63 +217,74 @@ not restrict the algorithm family.
 
 ## Use a low-code PyTorch recipe
 
-For scalar-column dense tabular models, subclass `TorchTrainingRecipe` and implement only
-the four factories. The runtime stacks the feature columns declared by
-`InputBinding` into one float32 tensor and supplies the declared label and
-optional sample-weight roles.
-
-For multi-worker exact coverage, an exhausted rank can invoke `forward()` with
-a zero-row tensor while another rank consumes its final batch. The model must
-accept an empty leading batch dimension and return the corresponding empty
-output shape; no observed row is replayed as padding.
+For Core-owned training, subclass `TorchRecipe` and implement the typed
+`build_modules`, `adapt_batch`, `training_step`, `validation_step`,
+`configure_optimizers`, `metric_plan`, and `artifact_plan` hooks. Framework-
+owned training instead subclasses `RayTorchAdapter`; the Adapter validates its
+environment, binds role datasets, receives a Core-selected checkpoint context,
+and cannot create a nested Trainer or declare a second execution plan.
 
 ```python
-from collections.abc import Mapping
-from typing import Any
-
-from tributo.algorithms import TorchTrainingRecipe
+from tributo.algorithms import (
+    TorchBatch,
+    TorchLossContribution,
+    TorchMetricPlan,
+    TorchOptimizationPlan,
+    TorchRecipe,
+    TorchStepResult,
+)
 
 
-class BinaryLinearRecipe(TorchTrainingRecipe):
-    def model_factory(self, config: Mapping[str, Any]) -> object:
+class BinaryLinearRecipe(TorchRecipe):
+    def build_modules(self, context):
         import torch
 
-        return torch.nn.Linear(int(config["input_features"]), 1)
+        return {"model": torch.nn.Linear(2, 1), "loss": torch.nn.BCEWithLogitsLoss()}
 
-    def loss_factory(self, config: Mapping[str, Any]) -> object:
+    def adapt_batch(self, batch, context):
         import torch
 
-        return torch.nn.BCEWithLogitsLoss()
+        features = torch.as_tensor(batch["features"])
+        targets = torch.as_tensor(batch["label"])
+        return TorchBatch(positional=(features,), targets=targets, local_rows=len(targets))
 
-    def optimizer_factory(self, model: object, config: Mapping[str, Any]) -> object:
+    def training_step(self, modules, batch, context):
         import torch
 
-        return torch.optim.Adam(model.parameters(), lr=float(config["lr"]))
+        predictions = modules["model"](batch.positional[0])
+        numerator = torch.nn.functional.binary_cross_entropy_with_logits(
+            predictions, batch.targets.float(), reduction="sum"
+        )
+        return TorchStepResult(
+            outputs={"prediction": predictions},
+            loss=TorchLossContribution(numerator, batch.local_rows),
+        )
 
-    def metric_factories(self, config: Mapping[str, Any]):
+    def validation_step(self, modules, batch, context):
+        return self.training_step(modules, batch, context)
+
+    def configure_optimizers(self, modules, context):
         import torch
 
-        return {
-            "accuracy": lambda prediction, target: (
-                (torch.sigmoid(prediction) >= 0.5) == target.bool()
-            )
-        }
+        return TorchOptimizationPlan(torch.optim.Adam(modules["model"].parameters(), lr=1e-3))
+
+    def metric_plan(self, context):
+        return TorchMetricPlan({"train_loss": "sum_count"})
+
+    def artifact_plan(self, context):
+        return {"source_kind": "torch_module", "roles": {"inference": "onnx-model"}}
 ```
 
-Declare the reducer and use `AlgorithmBuilder.from_torch_recipe()` to lower the
-class to the existing Ray Train collective runtime. `train_loss` is added with
-the fixed `sum_count` reducer. The supported reducer set is `sum_count`,
-`weighted_mean`, `min`, and `max`; a weighted metric requires
-`InputBinding.sample_weight_name`.
+Use `AlgorithmBuilder.from_torch()` to lower a Recipe, or
+`AlgorithmBuilder.from_torch_adapter()` for a framework Adapter. Losses always
+submit explicit numerator/normalizer pairs; model-specific composite reducers
+remain owned by the algorithm Wheel while Core owns collectives and scaling.
 
 The complete independent package used by conformance lives under
-`tests/fixtures/torch_recipe_algorithm_plugin`. It contains no Tributo builtin,
+the installed algorithm Wheel. It contains no Tributo private Runtime import,
 Ray worker loop, checkpoint upload, Bundle publisher, or deployment code.
-Default recipes receive a pre-bound training Dataset and do not own train/test
-splitting. Use the advanced recipe hooks only for a custom model invocation or
-loss call. Choose a framework adapter or `CollectiveAlgorithm` when the model
-needs a custom training step, multiple optimizers, framework callbacks, or
-special collectives.
+Default Recipes receive role-routed datasets; the Core Runtime owns sharding,
+checkpoint handling, evidence and Bundle publication.
 
 ## Choose the state coordination strategy
 

docs/how-to/training.md

@@ -1,9 +1,9 @@
 # Training on Ray
 
 Run XGBoost, X-Learner, DNN, and PU through real distributed state coordination.
-XGBoost and X-Learner use framework-native coordination; formal DNN lowers a first-party
-`TorchTrainingRecipe` to the common Ray-owned worker loop, while PU retains its
-specialized Ray Train/PyTorch DDP kernel. A one-worker local run is supported
+XGBoost and X-Learner use framework-native coordination; PyTorch algorithms use
+the unified `TorchRecipe` or `RayTorchAdapter` contract and one Core-owned Ray
+Train Torch Runtime. A one-worker local run is supported
 but is not reported as distributed training.
 
 Formal algorithms choose an explicit `local` or `cluster` execution profile.
@@ -18,7 +18,8 @@ the Ray Jobs API on an isolated Docker cluster with two worker nodes. It proves
 deployment-independent sharding, state coordination, receipt, and Bundle
 semantics without making Docker, Kubernetes, or VM providers execution profiles.
 
-Ordinary third-party PyTorch models should implement `TorchTrainingRecipe`.
+Ordinary third-party PyTorch models should implement `TorchRecipe`; framework-
+owned loops should implement `RayTorchAdapter`.
 Advanced packages may implement `CollectiveAlgorithm`, `MapReduceAlgorithm`,
 or `FrameworkNativeAlgorithm` and publish a
 `DistributedAlgorithmDescriptor` through the `tributo.algorithms` entry-point