feat(broker): add Core-only Broker Alpha

docs/STABILITY.md

@@ -27,6 +27,7 @@ This page provides module-level guidance and deprecation notes.
 | `tributo.config` — `AlgorithmExecutionConfig` and nested algorithm execution models | `alpha` | Strict JSON envelope shared by local Ray and Kubernetes-hosted Ray execution |
 | `tributo.job` — `TributoClient` | `stable` | Primary Ray Jobs client |
 | `tributo.job` — `RayJob` | `stable` annotation with runtime deprecation warning | Use `TributoClient`; the annotation and warning conflict is documented without changing the public contract in this documentation update |
+| `tributo.ray_jobs` | `alpha` | Workload-neutral submission identity, ambiguous-submit reconciliation, status, logs, and stop helpers |
 | `tributo.exceptions` — core exceptions | `stable` | ``TributoError`` and 16 common subtypes |
 | `tributo.exceptions` — `ResultMaterializationError` | `alpha` | Credential-safe lazy inference action failure |
 | `tributo.exceptions` — Bundle/Plugin exceptions | `beta` | ``BundleExportError``, ``BundleCommitBusyError``, ``AliasConflict``, ``UnsupportedArtifactFormat``, ``PostPublishCallbackError``, ``PluginLoadIssue`` |
@@ -204,6 +205,8 @@ from the legacy setup-only propagation rule.
 | `tributo.integrations.model_importers.*` | `alpha` | Canonical ModelImporter protocol/registry plus explicit MLflow and typed artifact-to-Bundle implementations |
 | `tributo.integrations.sinks.parquet` | `alpha` | Parquet inference ResultSink adapter |
 | `tributo.integrations.sinks.lance` | `alpha` | Generic Lance inference ResultSink adapter |
+| `tributo.integrations.broker` | `alpha` | Minimal transport-neutral Broker API v1; transport implementations and consume loops are external |
+| `tributo.integrations.broker_registry` | `alpha` | Lazy broker discovery and explicit provider resolution |
 
 ### Inference (tributo.inference.*)
 

docs/adr/002-broker-plugin-boundary.md

@@ -0,0 +1,66 @@
+# Broker provider boundary
+
+## Status
+
+Accepted for the Alpha API.
+
+## Context
+
+Tributo needs optional message-broker integrations without becoming a message
+queue platform. A broker task is a control-plane admission request; bounded
+data ingestion, unbounded inference streams, training, inference, Bundle
+publication, and result sinks retain their existing Tributo contracts.
+
+Transport clients and external wire protocols must remain independently
+installable. Core must be usable and testable without Redis, Kafka, RabbitMQ,
+or another provider dependency.
+
+## Decision
+
+Core owns a deliberately small Broker API v1 with Alpha stability:
+
+- `BrokerPlugin` discovery, structural version checks, capability metadata,
+  stability metadata, explicit config validation, and runtime construction;
+- opaque `Message` payloads with a delivery token and restricted string
+  metadata;
+- `TaskConsumer`, `BrokerRuntime`, `TaskDisposition`, and a minimal
+  `TaskOutcome` with an optional credential-safe `BrokerError`;
+- workload-neutral `RayJobSubmission` identity and deterministic submission
+  IDs derived from an operation namespace, `run_id`, and `attempt_id`;
+- ambiguous Ray submission reconciliation plus status and stop operations
+  keyed by `submission_id`.
+
+`BROKER_API_VERSION = 1` checks structural compatibility; it does not imply a
+Beta or long-term compatibility promise. Discovery is lazy and fail-open with
+diagnostics. Explicit resolution and configuration validation fail closed.
+Discovery never instantiates a provider or performs connectivity checks.
+
+The following concerns belong to provider packages:
+
+- broker connections, polling, acknowledgments, re-delivery, recovery, dead
+  letters, cancellation watchers, and the production consume CLI/runtime;
+- external request and event schemas, operation mapping, capability profiles,
+  credential references, error mapping, redaction, and event durability;
+- structured terminal-event publication from existing `TrainingResult`,
+  `InferenceResult`, Bundle, and result-sink receipts.
+
+Core does not define a workload registry or an external operation schema.
+Providers submit one thin execution-driver Ray Job through the generic helper;
+that driver calls existing in-process training or batch-inference APIs. Worker
+side broker cancellation, arbitrary execution context, a generic Core consume
+loop, and durable workflow semantics are outside the Alpha contract.
+
+`submission_id` is the primary Ray Jobs identity for admission, status, logs,
+and stop. `ray_job_id` is optional execution metadata and is populated only
+from a real Ray `JobDetails.job_id`; Core never substitutes `submission_id` for
+it. An optional credential-free `request_digest` may be recorded as Ray
+metadata, but Core does not persist it or promise cross-restart conflict
+detection.
+
+## Consequences
+
+Normal Tributo installations remain free of broker dependencies, and a
+provider can evolve transport and protocol behavior independently. Providers
+must own their infrastructure and healthy-path tests. The first release does
+not promise exactly-once execution, durable terminal events, high availability,
+or complete pending-message recovery.

docs/architecture/index.md

@@ -18,4 +18,5 @@ benchmark-protocol
 version-policy
 decision-log
 ../adr/001-data-and-bundle-contracts
+../adr/002-broker-plugin-boundary
 ```

docs/reference/api/algorithms-training.md

@@ -589,6 +589,9 @@ documentation for every public stability tier.
 ```{autofunction} tributo.training.job_submitter.submit_training_job
 ```
 
+```{autofunction} tributo.training.job_submitter.submit_training_job_with_identity
+```
+
 ```{autofunction} tributo.training.job_submitter.submit_training_job_with_retry
 ```
 
@@ -708,3 +711,6 @@ documentation for every public stability tier.
 
 ```{autofunction} tributo.training.xgboost_trainer.run_training_from_json
 ```
+
+```{autofunction} tributo.training.xgboost_trainer.run_training_result_with_config
+```

docs/reference/api/core.md

@@ -26,8 +26,7 @@ documentation for every public stability tier.
 ```{autoexception} tributo._common.dependencies.DependencyUnavailableError
 ```
 
-```{autoclass} tributo._common.dependencies.MissingOptionalDependency
-:no-members:
+```{autoexception} tributo._common.dependencies.MissingOptionalDependency
 ```
 
 ```{autofunction} tributo._common.dependencies.probe_dependency
@@ -73,8 +72,7 @@ documentation for every public stability tier.
 
 ## `tributo.exceptions`
 
-```{autoclass} tributo.exceptions.AliasConflict
-:no-members:
+```{autoexception} tributo.exceptions.AliasConflict
 ```
 
 ```{autoexception} tributo.exceptions.ArtifactCorruptedError
@@ -131,8 +129,7 @@ documentation for every public stability tier.
 ```{autoexception} tributo.exceptions.ModelSchemaMismatchError
 ```
 
-```{autoclass} tributo.exceptions.PluginLoadIssue
-:no-members:
+```{autoexception} tributo.exceptions.PluginLoadIssue
 ```
 
 ```{autoexception} tributo.exceptions.PostPublishCallbackError
@@ -159,8 +156,7 @@ documentation for every public stability tier.
 ```{autoexception} tributo.TributoError
 ```
 
-```{autoclass} tributo.exceptions.UnsupportedArtifactFormat
-:no-members:
+```{autoexception} tributo.exceptions.UnsupportedArtifactFormat
 ```
 
 
@@ -173,3 +169,22 @@ documentation for every public stability tier.
 ```{autoclass} tributo.TributoClient
 :no-members:
 ```
+
+
+## `tributo.ray_jobs`
+
+```{autoclass} tributo.ray_jobs.RayJobSubmission
+:no-members:
+```
+
+```{autofunction} tributo.ray_jobs.get_ray_job_logs
+```
+
+```{autofunction} tributo.ray_jobs.get_ray_job_status
+```
+
+```{autofunction} tributo.ray_jobs.stop_ray_job
+```
+
+```{autofunction} tributo.ray_jobs.submit_ray_job
+```

docs/reference/api/extensions.md

@@ -9,6 +9,48 @@ a public annotation or moving a public object.
 Stable, Beta, and Alpha objects appear because Ray-style API policy requires
 documentation for every public stability tier.
 
+## `tributo.integrations.broker`
+
+```{autoclass} tributo.integrations.broker.BrokerError
+:no-members:
+```
+
+```{autoclass} tributo.integrations.broker.BrokerPlugin
+:no-members:
+```
+
+```{autoclass} tributo.integrations.broker.BrokerRuntime
+:no-members:
+```
+
+```{autoclass} tributo.integrations.broker.Message
+:no-members:
+```
+
+```{autoclass} tributo.integrations.broker.TaskConsumer
+:no-members:
+```
+
+```{autoclass} tributo.integrations.broker.TaskDisposition
+:no-members:
+```
+
+```{autoclass} tributo.integrations.broker.TaskOutcome
+:no-members:
+```
+
+
+## `tributo.integrations.broker_registry`
+
+```{autoclass} tributo.integrations.broker_registry.BrokerDescriptor
+:no-members:
+```
+
+```{autoclass} tributo.integrations.broker_registry.BrokerRegistry
+:no-members:
+```
+
+
 ## `tributo.pipeline.core`
 
 ```{autoclass} tributo.pipeline.core.ArtifactRef

docs/reference/api/inference-serving.md

@@ -299,12 +299,18 @@ documentation for every public stability tier.
 ```{autofunction} tributo.inference.job_runner.submit_inference_request
 ```
 
+```{autofunction} tributo.inference.job_runner.submit_inference_request_with_identity
+```
+
 ```{autofunction} tributo.inference.job_runner.submit_inference_request_with_retry
 ```
 
 ```{autofunction} tributo.inference.job_runner.submit_resolved_inference
 ```
 
+```{autofunction} tributo.inference.job_runner.submit_resolved_inference_with_identity
+```
+
 ```{autofunction} tributo.inference.job_runner.wait_for_job
 ```
 

src/tributo/cli.py

@@ -26,7 +26,49 @@
 logger = logging.getLogger(__name__)
 
 
-@click.group()
+class _LazyTributoGroup(click.Group):
+    """Load the broker command module only when that command is selected."""
+
+    def get_command(self, ctx: click.Context, cmd_name: str) -> click.Command | None:
+        if cmd_name == "broker":
+            from tributo.cli_broker import broker
+
+            return broker
+        return super().get_command(ctx, cmd_name)
+
+    def list_commands(self, ctx: click.Context) -> list[str]:
+        commands = super().list_commands(ctx)
+        if "broker" not in commands:
+            commands.append("broker")
+        return sorted(commands)
+
+    def format_commands(
+        self,
+        ctx: click.Context,
+        formatter: click.HelpFormatter,
+    ) -> None:
+        """Render root help without importing the broker command module."""
+        commands: list[tuple[str, click.Command]] = []
+        for command_name in self.list_commands(ctx):
+            command = (
+                click.Command(
+                    "broker",
+                    help="Discover and validate explicitly selected broker providers.",
+                )
+                if command_name == "broker"
+                else self.get_command(ctx, command_name)
+            )
+            if command is not None and not command.hidden:
+                commands.append((command_name, command))
+        if not commands:
+            return
+        limit = formatter.width - 6 - max(len(name) for name, _ in commands)
+        rows = [(name, command.get_short_help_str(limit)) for name, command in commands]
+        with formatter.section("Commands"):
+            formatter.write_dl(rows)
+
+
+@click.group(cls=_LazyTributoGroup)
 @click.version_option(package_name="tributo")
 def main():
     """Tributo: Unified framework for submitting Ray Jobs."""

src/tributo/cli_broker.py

@@ -0,0 +1,72 @@
+"""Broker-specific CLI commands, mounted by :mod:`tributo.cli`."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import click
+
+from tributo.exceptions import JobConfigurationError
+from tributo.integrations.broker_registry import BrokerRegistry
+
+
+def _load_config(path: str) -> dict[str, Any]:
+    config_path = Path(path)
+    if config_path.suffix.lower() in {".yaml", ".yml"}:
+        raise click.ClickException("YAML broker config is not supported; use JSON.")
+    try:
+        value = json.loads(config_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise click.ClickException(f"Unable to read broker config: {exc}") from exc
+    if not isinstance(value, dict):
+        raise click.ClickException("Broker config root must be a JSON object.")
+    return value
+
+
+@click.group()
+def broker() -> None:
+    """Discover and validate explicitly selected broker providers."""
+
+
+@broker.command("list")
+def broker_list() -> None:
+    """List installed broker plugins without connecting to a broker."""
+    registry = BrokerRegistry()
+    descriptors = registry.list()
+    for descriptor in descriptors:
+        capabilities = ",".join(descriptor.capabilities) or "-"
+        click.echo(
+            f"{descriptor.broker_id}\tapi={descriptor.api_version}"
+            f"\tstability={descriptor.stability}\tcapabilities={capabilities}"
+        )
+    for diagnostic in registry.diagnostics():
+        click.echo(
+            f"diagnostic\t{diagnostic.entry_point_name}\t{diagnostic.reason}",
+            err=True,
+        )
+
+
+@broker.command("validate")
+@click.option("--broker", "broker_id", required=True)
+@click.option("--config", "config_path", required=True, type=click.Path(exists=True))
+@click.option(
+    "--check-connectivity",
+    is_flag=True,
+    help="Ask the provider to perform an explicit connectivity probe.",
+)
+def broker_validate(broker_id: str, config_path: str, check_connectivity: bool) -> None:
+    """Validate provider-owned JSON config."""
+    try:
+        config = _load_config(config_path)
+        BrokerRegistry().validate(
+            broker_id,
+            config,
+            check_connectivity=check_connectivity,
+        )
+    except (JobConfigurationError, click.ClickException) as exc:
+        if isinstance(exc, click.ClickException):
+            raise
+        raise click.ClickException(str(exc)) from exc
+    click.echo(f"Broker configuration is valid: {broker_id}")

src/tributo/inference/__init__.py

@@ -40,7 +40,9 @@ def __call__(self, batch): ...
 from tributo.inference.job_runner import (
     submit_inference_job,
     submit_inference_request,
+    submit_inference_request_with_identity,
     submit_resolved_inference,
+    submit_resolved_inference_with_identity,
 )
 from tributo.inference.pipeline import (
     InferenceConfig,
@@ -72,5 +74,7 @@ def __call__(self, batch): ...
     "XGBoostONNXPredictor",
     "submit_inference_job",
     "submit_inference_request",
+    "submit_inference_request_with_identity",
     "submit_resolved_inference",
+    "submit_resolved_inference_with_identity",
 ]