[JFMGC-8] Build failover solution for model serving infrastructure

.env.sample

@@ -12,6 +12,10 @@ JOINTFM_SCHEMA_VERSION=v1
 # 1. Deployment ID: SDK builds the hosted predictionsUnstructured URL.
 # JOINTFM_DEPLOYMENT_ID=
 
+# 1b. Comma-separated deployment IDs for load-balanced hosted calls (same checkpoint).
+# Requires at least two unique IDs. Mutually exclusive with JOINTFM_DEPLOYMENT_ID and other selectors.
+# JOINTFM_DEPLOYMENT_IDS=chevron-id,research-id
+
 # 2. Deployment URL: SDK appends predictionsUnstructured.
 # JOINTFM_DEPLOYMENT_URL=https://app.datarobot.com/api/v2/deployments/<deployment-id>
 

README.md

@@ -38,7 +38,7 @@ The direct local service exposes `GET /healthz` and `POST /predict`.
 
 Structured SDK defaults live in `jointfm_client.configuration.JointFMConfig` and are mirrored in the checked-in `config.sample.yaml`. Copy `config.sample.yaml` to `config.yaml` and change only the fields needed for your deployment or transport defaults. `JointFMClient.from_env()` and `load_settings()` read `config.yaml` by default, then layer `.env` values over it, then layer process environment variables or the supplied `env` mapping over both. Explicit Python arguments such as `timeout=` and `retry_config=` still override YAML transport defaults.
 
-`JointFMClient.from_env()` and `load_settings()` resolve `JOINTFM_SCHEMA_VERSION` and exactly one service selector from that layered configuration. `JOINTFM_MODEL_VERSION` is optional: when unset the SDK discovers the model version from `/healthz` on first use, and when set the SDK validates it against `/healthz` as a drift-detection guard. Hosted selectors also require `DATAROBOT_ENDPOINT` and `DATAROBOT_API_TOKEN`; the direct local selector does not use DataRobot credentials. Missing credentials, missing schema version, malformed credentials, unsupported schema versions, missing selectors, and multiple selectors raise `JointFMConfigurationError`.
+`JointFMClient.from_env()` and `load_settings()` resolve `JOINTFM_SCHEMA_VERSION` and exactly one service selector from that layered configuration. Hosted options include `JOINTFM_DEPLOYMENT_ID` or load-balanced `JOINTFM_DEPLOYMENT_IDS` (comma-separated same-checkpoint peers; mutually exclusive with other selectors). `JOINTFM_MODEL_VERSION` is optional: when unset the SDK discovers the model version from `/healthz` on first use, and when set the SDK validates it against `/healthz` as a drift-detection guard. Hosted selectors also require `DATAROBOT_ENDPOINT` and `DATAROBOT_API_TOKEN`; the direct local selector does not use DataRobot credentials. Missing credentials, missing schema version, malformed credentials, unsupported schema versions, missing selectors, and multiple selectors raise `JointFMConfigurationError`.
 
 `DATAROBOT_ENDPOINT` must be a normalized HTTPS DataRobot API v2 URL ending in `/api/v2`; the SDK stores it without a trailing slash. `DATAROBOT_API_TOKEN` must be non-empty and whitespace-free. The token is excluded from `JointFMSettings` repr output.
 
@@ -86,6 +86,7 @@ JOINTFM_SCHEMA_VERSION=v1
 Choose exactly one service selector:
 
 - `JOINTFM_DEPLOYMENT_ID`: builds `DATAROBOT_ENDPOINT.rstrip("/") + "/"` plus `deployments/{deployment_id}/predictionsUnstructured`
+- `JOINTFM_DEPLOYMENT_IDS`: comma-separated hosted deployment IDs (≥2 unique, same checkpoint) for round-robin load balancing; mutually exclusive with other selectors
 - `JOINTFM_DEPLOYMENT_URL`: appends `/predictionsUnstructured` to a hosted deployment URL
 - `JOINTFM_PREDICT_URL`: uses a full hosted prediction URL ending in `/predictionsUnstructured`
 - `JOINTFM_DEPLOYMENT_TARGET` with `JOINTFM_PULUMI_OUTPUTS_PATH`: resolves a named target from saved Pulumi outputs JSON, preferring `deployment_id`, then `deployment_url`, then `predict_url`
@@ -245,6 +246,7 @@ DATAROBOT_ENDPOINT=https://app.datarobot.com/api/v2
 DATAROBOT_API_TOKEN=<token>
 JOINTFM_SCHEMA_VERSION=v1
 JOINTFM_DEPLOYMENT_ID=<deployment-id>
+# Or: JOINTFM_DEPLOYMENT_IDS=chevron-id,research-id
 # Optional drift-detection pin; the SDK discovers the model version from /healthz when unset:
 # JOINTFM_MODEL_VERSION=jointfm-inference:0.2.0+ckpt.fin-2026-05-22
 ```

config.sample.yaml

@@ -6,6 +6,7 @@ environment:
   datarobot_endpoint: DATAROBOT_ENDPOINT
   datarobot_api_token: DATAROBOT_API_TOKEN
   deployment_id: JOINTFM_DEPLOYMENT_ID
+  deployment_ids: JOINTFM_DEPLOYMENT_IDS
   deployment_url: JOINTFM_DEPLOYMENT_URL
   predict_url: JOINTFM_PREDICT_URL
   deployment_target: JOINTFM_DEPLOYMENT_TARGET
@@ -17,6 +18,7 @@ deployment:
   datarobot_endpoint: null
   datarobot_api_token: null
   deployment_id: null
+  deployment_ids: null
   deployment_url: null
   predict_url: null
   deployment_target: null

docs/api-reference.md

@@ -34,7 +34,7 @@ This reference covers the supported public Python surface exported by `jointfm_c
 
 | Name | Purpose |
 | --- | --- |
-| `JointFMSettings` | Validated hosted or local service settings: optional normalized DataRobot endpoint, optional secret token, health and prediction URLs, service selector, schema pin, model pin, and optional selector details. The API token is excluded from `repr`. |
+| `JointFMSettings` | Validated hosted or local service settings: optional normalized DataRobot endpoint, optional secret token, health and prediction URLs, service selector, schema pin, optional load-balanced `instances` pool, model pin, and optional selector details. The API token is excluded from `repr`. |
 | `JointFMConfig` | Top-level structured configuration loaded from defaults, YAML, and explicit overrides. |
 | `PathConfig` | Default local file names for `config.yaml`, `config.sample.yaml`, and `.env`. |
 | `EnvironmentVariableConfig` | Environment variable names consumed by settings loading. |
@@ -119,14 +119,15 @@ All SDK-specific exceptions inherit from `JointFMError`.
 | `JOINTFM_SCHEMA_VERSION` | Hosted calls | Request schema pin. The SDK supports only `v1`. |
 | `JOINTFM_MODEL_VERSION` | Hosted calls | Exact JointFM deployment model version expected from the service-health payload and prediction responses. |
 | `JOINTFM_DEPLOYMENT_ID` | One selector | Deployment ID used to build hosted health and prediction URLs. |
+| `JOINTFM_DEPLOYMENT_IDS` | One selector | Comma-separated hosted deployment IDs for round-robin load balancing (at least two unique IDs). Mutually exclusive with other selectors. Peers must share `model_version` and `checkpoint_version`; the SDK uses the minimum `max_sample_count`. |
 | `JOINTFM_DEPLOYMENT_URL` | One selector | Hosted deployment URL; the SDK derives the `/predictionsUnstructured` route from it and reuses that route for health probes. |
 | `JOINTFM_PREDICT_URL` | One selector | Full hosted prediction URL ending in `/predictionsUnstructured`; the SDK derives the owning deployment URL. |
 | `JOINTFM_DEPLOYMENT_TARGET` | One selector with outputs path | Key in a saved Pulumi outputs JSON file. |
 | `JOINTFM_PULUMI_OUTPUTS_PATH` | With target selector | JSON file containing target outputs with exactly one of `deployment_id`, `deployment_url`, or `predict_url`. |
 | `JOINTFM_LOCAL_BASE_URL` | One selector | Direct local JointFM REST service base URL. The SDK calls `GET /healthz` and `POST /predict` without DataRobot authorization. |
 | `DATAROBOT_DEPLOYMENT_ID` | Optional live tests | Hosted deployment ID used only by the optional live smoke test so normal CI does not call DataRobot accidentally. |
 
-Set exactly one selector among `JOINTFM_DEPLOYMENT_ID`, `JOINTFM_DEPLOYMENT_URL`, `JOINTFM_PREDICT_URL`, `JOINTFM_DEPLOYMENT_TARGET`, and `JOINTFM_LOCAL_BASE_URL`.
+Set exactly one selector among `JOINTFM_DEPLOYMENT_ID`, `JOINTFM_DEPLOYMENT_IDS`, `JOINTFM_DEPLOYMENT_URL`, `JOINTFM_PREDICT_URL`, `JOINTFM_DEPLOYMENT_TARGET`, and `JOINTFM_LOCAL_BASE_URL`.
 
 ## V1 Payload Fields
 

src/jointfm_client/__init__.py

@@ -99,6 +99,7 @@
     UnsupportedSchemaVersionError,
     UnsupportedServiceContractError,
 )
+from jointfm_client.pool import JointFMInstancePool
 from jointfm_client.notebooks import (
     WORKSPACE_ROOT_MARKERS,
     bootstrap_notebook,
@@ -108,13 +109,15 @@
     DATAROBOT_API_TOKEN_ENV,
     DATAROBOT_ENDPOINT_ENV,
     JOINTFM_DEPLOYMENT_ID_ENV,
+    JOINTFM_DEPLOYMENT_IDS_ENV,
     JOINTFM_DEPLOYMENT_TARGET_ENV,
     JOINTFM_DEPLOYMENT_URL_ENV,
     JOINTFM_LOCAL_BASE_URL_ENV,
     JOINTFM_MODEL_VERSION_ENV,
     JOINTFM_PREDICT_URL_ENV,
     JOINTFM_PULUMI_OUTPUTS_PATH_ENV,
     JOINTFM_SCHEMA_VERSION_ENV,
+    JointFMInstanceSettings,
     JointFMSettings,
     build_datarobot_prediction_headers,
     build_hosted_deployment_url,
@@ -170,6 +173,7 @@
     "IMPORT_NAMESPACE",
     "MeanForecastResult",
     "JOINTFM_DEPLOYMENT_ID_ENV",
+    "JOINTFM_DEPLOYMENT_IDS_ENV",
     "JOINTFM_DEPLOYMENT_TARGET_ENV",
     "JOINTFM_DEPLOYMENT_URL_ENV",
     "JOINTFM_LOCAL_BASE_URL_ENV",
@@ -189,6 +193,8 @@
     "JointFMResponseDecodeError",
     "JointFMResponseError",
     "JointFMServiceError",
+    "JointFMInstancePool",
+    "JointFMInstanceSettings",
     "JointFMRetryConfig",
     "JointFMSettings",
     "JointFMTimeoutConfig",

src/jointfm_client/client.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 import re
 from typing import Any, Self, cast
@@ -57,6 +58,7 @@
     QuantileForecastResult,
     SampleForecastResult,
 )
+from jointfm_client.pool import JointFMInstancePool
 from jointfm_client.settings import (
     JointFMSettings,
     load_settings,
@@ -108,6 +110,7 @@ def __init__(
         self._datarobot_request_id_headers = datarobot_request_id_headers
         self._health_metadata: HealthMetadata | None = None
         self._sample_batch_cap: int | None = None
+        self._pool: JointFMInstancePool | None = None
 
     @classmethod
     def from_env(
@@ -152,10 +155,21 @@ def health(self, *, cache: bool = False, refresh: bool = False) -> HealthMetadat
         deployment gateway only proxies the unstructured prediction route; the
         container short-circuits that body before any schema or model version
         validation and returns the same typed health payload.
+
+        When ``JOINTFM_DEPLOYMENT_IDS`` is set, reachable peers are probed and must
+        share ``model_version`` and ``checkpoint_version``; the sample-batch cap is
+        the minimum ``max_sample_count`` across those peers.
         """
         if cache and not refresh and self._health_metadata is not None:
             return self._health_metadata
 
+        if self._uses_pool():
+            metadata = self._require_pool().probe_all_health()
+            self._sample_batch_cap = metadata.max_sample_count
+            if cache:
+                self._health_metadata = metadata
+            return metadata
+
         if self._uses_predict_route_for_health():
             payload = self._fetch_hosted_health_payload()
         else:
@@ -195,14 +209,14 @@ def _fetch_hosted_health_payload(self) -> Mapping[str, Any]:
 
     def predict(self, payload: Mapping[str, Any]) -> Mapping[str, Any]:
         """Submit one V1 JSON prediction payload to the configured endpoint."""
-        predict_url = self._require_predict_url("predict")
+        self._require_predict_url("predict")
         model_version = payload.get("model_version")
         if not isinstance(model_version, str):
             raise JointFMConfigurationError(
                 "JointFMClient.predict() requires payload['model_version']"
             )
         self._resolve_model_version(model_version=model_version)
-        response_payload = self._transport_for_request().post_json(predict_url, payload)
+        response_payload = self._post_predict_json(payload)
         ForecastResponse.raise_for_errors(response_payload)
         return response_payload
 
@@ -243,7 +257,7 @@ def forecast(
         | None = None,
     ) -> ForecastResponse:
         """Build and submit a forecast request from tabular history inputs."""
-        predict_url = self._require_predict_url("forecast")
+        self._require_predict_url("forecast")
         resolved_model_version = self._resolve_model_version(
             model_version=model_version,
         )
@@ -303,18 +317,16 @@ def forecast(
             )
         sample_cap = self._resolve_sample_batch_cap(payload)
         if sample_cap is not None:
-            return self._forecast_sample_batches(predict_url, payload, sample_cap)
+            return self._forecast_sample_batches(payload, sample_cap)
 
         try:
-            response_payload = self._transport_for_request().post_json(
-                predict_url, payload
-            )
+            response_payload = self._post_predict_json(payload)
         except JointFMHTTPStatusError as error:
             sample_cap = _sample_batch_cap_from_error(error, payload)
             if sample_cap is None:
                 raise
             self._sample_batch_cap = sample_cap
-            return self._forecast_sample_batches(predict_url, payload, sample_cap)
+            return self._forecast_sample_batches(payload, sample_cap)
 
         return _forecast_response_from_payload(response_payload, payload)
 
@@ -484,44 +496,119 @@ def _resolve_sample_batch_cap(self, payload: Mapping[str, Any]) -> int | None:
 
     def _forecast_sample_batches(
         self,
-        predict_url: str,
         payload: Mapping[str, Any],
         sample_cap: int,
     ) -> SampleForecastResult:
         requested_samples = cast(int, payload["n_samples"])
         remaining_samples = requested_samples
+        batch_payloads: list[dict[str, Any]] = []
         batch_index = 0
-        batch_results: list[SampleForecastResult] = []
 
         while remaining_samples > 0:
             batch_samples = min(sample_cap, remaining_samples)
             batch_payload = dict(payload)
             batch_payload["n_samples"] = batch_samples
             _set_batch_seed(batch_payload, batch_index)
-            response_payload = self._transport_for_request().post_json(
-                predict_url,
-                batch_payload,
-            )
-            batch_result = _forecast_response_from_payload(
-                response_payload,
-                batch_payload,
-            )
-            if not isinstance(batch_result, SampleForecastResult):
-                raise JointFMServiceError(
-                    "JointFM forecast response violated the V1 contract: "
-                    "sample batching requires sample forecast responses"
-                )
-            batch_results.append(batch_result)
+            batch_payloads.append(batch_payload)
             remaining_samples -= batch_samples
             batch_index += 1
 
+        if self._uses_pool() and len(batch_payloads) > 1:
+            batch_results = self._forecast_sample_batches_parallel(batch_payloads)
+        else:
+            batch_results = [
+                self._sample_forecast_from_batch_payload(batch_payload)
+                for batch_payload in batch_payloads
+            ]
+
         try:
             return _merge_sample_forecast_results(batch_results, payload)
         except ValueError as error:
             raise JointFMServiceError(
                 f"JointFM forecast response violated the V1 contract: {error}"
             ) from error
 
+    def _forecast_sample_batches_parallel(
+        self, batch_payloads: Sequence[Mapping[str, Any]]
+    ) -> list[SampleForecastResult]:
+        pool = self._require_pool()
+        max_workers = min(len(batch_payloads), pool.instance_count)
+
+        def _run_batch(item: tuple[int, Mapping[str, Any]]) -> SampleForecastResult:
+            batch_index, batch_payload = item
+            response_payload = pool.post_json_to(
+                pool.instance_at(batch_index), batch_payload
+            )
+            return self._sample_forecast_from_response(response_payload, batch_payload)
+
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            return list(
+                executor.map(
+                    _run_batch,
+                    enumerate(batch_payloads),
+                )
+            )
+
+    def _sample_forecast_from_batch_payload(
+        self, batch_payload: Mapping[str, Any]
+    ) -> SampleForecastResult:
+        return self._sample_forecast_from_response(
+            self._post_predict_json(batch_payload), batch_payload
+        )
+
+    def _sample_forecast_from_response(
+        self, response_payload: Mapping[str, Any], batch_payload: Mapping[str, Any]
+    ) -> SampleForecastResult:
+        batch_result = _forecast_response_from_payload(response_payload, batch_payload)
+        if not isinstance(batch_result, SampleForecastResult):
+            raise JointFMServiceError(
+                "JointFM forecast response violated the V1 contract: "
+                "sample batching requires sample forecast responses"
+            )
+        return batch_result
+
+    def _post_predict_json(self, payload: Mapping[str, Any]) -> Mapping[str, Any]:
+        if self._uses_pool():
+            return self._require_pool().post_json(payload)
+        predict_url = self._require_predict_url("predict")
+        return self._transport_for_request().post_json(predict_url, payload)
+
+    def _uses_pool(self) -> bool:
+        return self.settings is not None and len(self.settings.instances) > 1
+
+    def _require_pool(self) -> JointFMInstancePool:
+        if not self._uses_pool():
+            raise JointFMConfigurationError(
+                "JointFMClient pool routing requires multiple deployment instances"
+            )
+        if self._pool is None:
+            assert self.settings is not None
+            # One Session per peer for thread-safe parallel sample batches.
+            # An injected transport is reused across peers (tests/mocks only);
+            # production from_env builds a distinct fail-fast transport each.
+            if self._transport is not None:
+                transports = tuple(self._transport for _ in self.settings.instances)
+            else:
+                transports = tuple(
+                    self._new_pool_peer_transport() for _ in self.settings.instances
+                )
+            self._pool = JointFMInstancePool(
+                instances=self.settings.instances,
+                transports=transports,
+                expected_model_version=self.settings.model_version,
+            )
+        return self._pool
+
+    def _new_pool_peer_transport(self) -> JSONTransport:
+        assert self.settings is not None
+        return JointFMHTTPTransport.from_settings(
+            self.settings,
+            timeout=self._timeout,
+            retry_config=JointFMRetryConfig(max_attempts=1),
+            response_body_excerpt_characters=(self._response_body_excerpt_characters),
+            datarobot_request_id_headers=self._datarobot_request_id_headers,
+        )
+
     def _require_settings(self, method_name: str) -> JointFMSettings:
         if self.settings is None:
             raise JointFMConfigurationError(
@@ -576,6 +663,10 @@ def _resolve_model_version(
             assert self._health_metadata is not None
             return self._health_metadata.model_version
 
+        # Pool peers must share checkpoint identity before any traffic.
+        if self._uses_pool() and self._health_metadata is None:
+            self.health(cache=True)
+
         normalized_model_version = validate_jointfm_model_version(
             configured_model_version
         )