diff --git a/.env.sample b/.env.sample index 167b6da..adf7862 100644 --- a/.env.sample +++ b/.env.sample @@ -6,7 +6,7 @@ DATAROBOT_API_TOKEN= # JointFM compatibility metadata for the selected deployment. JOINTFM_SCHEMA_VERSION=v5 # 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 +# JOINTFM_MODEL_VERSION=jointfm-inference:+ckpt. # Choose exactly one deployment selector. # 1. Deployment ID: SDK builds the hosted predictionsUnstructured URL. @@ -14,7 +14,7 @@ JOINTFM_SCHEMA_VERSION=v5 # 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 +# JOINTFM_DEPLOYMENT_IDS=, # 2. Deployment URL: SDK appends predictionsUnstructured. # JOINTFM_DEPLOYMENT_URL=https://app.datarobot.com/api/v2/deployments/ @@ -24,10 +24,14 @@ JOINTFM_SCHEMA_VERSION=v5 # 4. Named Pulumi target resolved from saved outputs JSON. # The JSON file must map target names to Pulumi stack outputs, for example: -# {"fin-studentt": {"deployment_id": ""}} -# JOINTFM_DEPLOYMENT_TARGET=fin-studentt +# {"my-target": {"deployment_id": ""}} +# JOINTFM_DEPLOYMENT_TARGET=my-target # JOINTFM_PULUMI_OUTPUTS_PATH=./jointfm-pulumi-outputs.json -# 5. Direct local REST service started with `task service:start` in joint. +# 5. Direct local REST service: a JointFM inference service you run yourself. # Local service calls use GET /healthz and POST /predict without DataRobot auth. -# JOINTFM_LOCAL_BASE_URL=http://127.0.0.1:8080 \ No newline at end of file +# JOINTFM_LOCAL_BASE_URL=http://127.0.0.1:8080 +# 5b. Comma-separated local service base URLs for load-balanced local calls (same checkpoint), +# for example one instance per GPU on distinct ports. Requires at least two unique URLs. +# Mutually exclusive with JOINTFM_LOCAL_BASE_URL and other selectors. +# JOINTFM_LOCAL_BASE_URLS=http://127.0.0.1:8080,http://127.0.0.1:8081 diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 100644 index caeb538..0000000 --- a/CLAUDE.md +++ /dev/null @@ -1,7 +0,0 @@ -# CLAUDE.md - -This project keeps its agent instructions in AGENTS.md, which is imported here: - -@AGENTS.md - -Do not duplicate guidance in this file — edit AGENTS.md instead. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 120000 index 0000000..47dc3e3 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file diff --git a/README.md b/README.md index 19d21f5..7fda5fb 100644 --- a/README.md +++ b/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. 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`. +`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. Local options are `JOINTFM_LOCAL_BASE_URL` or load-balanced `JOINTFM_LOCAL_BASE_URLS` (comma-separated same-checkpoint local services; mutually exclusive with other selectors). Hosted selectors also require `DATAROBOT_ENDPOINT` and `DATAROBOT_API_TOKEN`; the local selectors do 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. @@ -53,7 +53,7 @@ deployment: schema_version: v5 deployment_id: # Optional model-version pin; the SDK discovers it from /healthz when unset: - # model_version: jointfm-inference:0.3.0+ckpt.fin-2026-05-22 + # model_version: jointfm-inference:+ckpt. transport: timeout: connect_seconds: 10.0 @@ -71,16 +71,16 @@ DATAROBOT_API_TOKEN= JOINTFM_SCHEMA_VERSION=v5 JOINTFM_DEPLOYMENT_ID= # Optional drift-detection pin; the SDK discovers the model version from /healthz when unset: -# JOINTFM_MODEL_VERSION=jointfm-inference:0.3.0+ckpt.fin-2026-05-22 +# JOINTFM_MODEL_VERSION=jointfm-inference:+ckpt. ``` -Equivalent local REST configuration for a service started from the `joint` repository with `task service:start CONFIG=nvidia-studentt-m4cr2`: +Equivalent configuration for a JointFM inference service you run yourself: ```dotenv JOINTFM_LOCAL_BASE_URL=http://127.0.0.1:8080 JOINTFM_SCHEMA_VERSION=v5 # Optional drift-detection pin; the SDK discovers the model version from /healthz when unset: -# JOINTFM_MODEL_VERSION=jointfm-inference:0.3.0+ckpt.fin_i504_o63_f0_t10_h16l16_mam7_af_t3r1_cnn_k3l4_hpst_h16l2_studentt_m4cr2df8skew +# JOINTFM_MODEL_VERSION=jointfm-inference:+ckpt. ``` Choose exactly one service selector: @@ -91,12 +91,13 @@ Choose exactly one service selector: - `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` - `JOINTFM_LOCAL_BASE_URL`: builds direct local `GET /healthz` and `POST /predict` URLs without DataRobot authentication +- `JOINTFM_LOCAL_BASE_URLS`: comma-separated local service base URLs (≥2 unique after normalization, same checkpoint), for example one instance per GPU on distinct ports, load-balanced exactly like `JOINTFM_DEPLOYMENT_IDS` and without DataRobot authentication; mutually exclusive with other selectors Pulumi deployment discovery is explicit and file-backed. Export stack outputs to a JSON object keyed by target name, then set `JOINTFM_DEPLOYMENT_TARGET` to the key and `JOINTFM_PULUMI_OUTPUTS_PATH` to that JSON file: ```json { - "fin-studentt": { + "my-target": { "deployment_id": "" } } @@ -118,9 +119,9 @@ The SDK still decodes hosted prediction responses as JSON; the broad `Accept` va Direct local URL helpers are used by the local service selector: `build_local_health_url("http://localhost:8080")` returns `/healthz`, and `build_local_predict_url("http://localhost:8080")` returns `/predict`. -Hosted settings also derive `health_url` from the resolved deployment URL as `deployments/{deployment_id}/healthz`. `JointFMClient.health(cache=True)` stores typed `HealthMetadata` only when the caller asks for caching, and `JointFMClient.refresh_health()` fetches a fresh copy. +Hosted deployments are probed through the prediction route: `health()` POSTs `{"request_type": "health"}` to the resolved `predict_url`, because the DataRobot deployment gateway proxies only the unstructured prediction route. `JointFMClient.health(cache=True)` stores typed `HealthMetadata` only when the caller asks for caching, and `JointFMClient.refresh_health()` fetches a fresh copy. -Each endpoint's health payload describes only that endpoint. With `JOINTFM_DEPLOYMENT_IDS`, the client probes every configured peer and aggregates locally: `health()` returns consensus metadata whose `max_sample_count` is the **minimum** reachable cap (the sample-batch size), while `health_instances()` returns one `InstanceHealth` per configured ID plus the **sum** of reachable caps (one request per instance; each instance's `max_concurrent_requests` says how many such requests it serves at once), a compact `topology` / `topology_label` (for example `2x5000` or `1x7000, 1x3000`), and errors for unavailable peers. +Each endpoint's health payload describes only that endpoint. With a pool (`JOINTFM_DEPLOYMENT_IDS` or `JOINTFM_LOCAL_BASE_URLS`), the client probes every configured peer by POSTing `{"request_type": "health"}` to its prediction route, which local services answer as well, and aggregates locally: `health()` returns consensus metadata whose `max_sample_count` is the **minimum** reachable cap (the sample-batch size), while `health_instances()` returns one `InstanceHealth` per configured instance, named by its `instance_id` (the deployment ID, or the local base URL), plus the **sum** of reachable caps (one request per instance; each instance's `max_concurrent_requests` says how many such requests it serves at once), a compact `topology` / `topology_label` (for example `2x5000` or `1x7000, 1x3000`), and errors for unavailable peers. ## CLI Workflows @@ -157,7 +158,7 @@ Check that the deployment serves the largest request its health envelope adverti uv run jointfm-client max-utilization ``` -`max-utilization` fills every request slot at once with a seeded synthetic sample forecast of `max_series` target columns, an `n_input`-row history, an `n_output`-step horizon, and `max_sample_count` samples: each reachable endpoint (every pool instance with `JOINTFM_DEPLOYMENT_IDS`) receives as many simultaneous maximal requests as its `max_concurrent_requests` admits. It prints the burst's wall-clock time and a final `VERDICT: PASS` or `VERDICT: FAIL` line. It exits with status 1 on FAIL: when a request raises a service or transport error or returns the wrong sample count, or when an endpoint fails the health probe. Retries are disabled for this command, so a request that fails under load is reported instead of retried. Each run uses real GPU time on the deployment. `task max-utilization` runs the same command, and `notebooks/max_utilization.ipynb` shows the burst timing as a table. +`max-utilization` fills every request slot at once with a seeded synthetic sample forecast of `max_series` target columns, an `n_input`-row history, an `n_output`-step horizon, and `max_sample_count` samples: each reachable endpoint (every pool instance with `JOINTFM_DEPLOYMENT_IDS` or `JOINTFM_LOCAL_BASE_URLS`) receives as many simultaneous maximal requests as its `max_concurrent_requests` admits. It prints the burst's wall-clock time and a final `VERDICT: PASS` or `VERDICT: FAIL` line. It exits with status 1 on FAIL: when a request raises a service or transport error or returns the wrong sample count, or when an endpoint fails the health probe. Retries are disabled for this command, so a request that fails under load is reported instead of retried. Each run uses real GPU time on the deployment. `task max-utilization` runs the same command, and `notebooks/max_utilization.ipynb` shows the burst timing as a table. ## Notebook Workflows @@ -183,7 +184,7 @@ The current forecast request contract is: - `time_column`: required for `"absolute_datetime"`, and used for ordered ordinal or continuous histories when supplied - `query_times`: non-empty future forecast times only - `requested_columns`: optional column names or integer column indices, with duplicates rejected -- `n_samples`: positive sample count for sampled forecasts and quantile estimation. When `return_mode="samples"` exceeds the `max_sample_count` advertised by the deployment's health metadata, `forecast_samples(...)` splits the request into capped prediction batches up front and returns one merged `SampleForecastResult`. With `JOINTFM_DEPLOYMENT_IDS`, every sample request is split evenly across all reachable endpoints, whatever its size, and each endpoint serves at most one of its batches at a time. Batch `i` uses seed `seed + i`, so a seeded result depends on how many endpoints were reachable. +- `n_samples`: positive sample count for sampled forecasts and quantile estimation. When `return_mode="samples"` exceeds the `max_sample_count` advertised by the deployment's health metadata, `forecast_samples(...)` splits the request into capped prediction batches up front and returns one merged `SampleForecastResult`. With a pool (`JOINTFM_DEPLOYMENT_IDS` or `JOINTFM_LOCAL_BASE_URLS`), every sample request is split evenly across all reachable endpoints, whatever its size, and each endpoint serves at most one of its batches at a time. Batch `i` uses seed `seed + i`, so a seeded result depends on how many endpoints were reachable. - `condition`: required with `query_mode="condition"` and forbidden otherwise. One `EqualityCondition` or `IntervalCondition`, or a list of them, each covering the future positions its `query_time_indices` names; sent on the wire as the `conditions` list, see below. ### Conditional Queries @@ -290,9 +291,9 @@ DATAROBOT_ENDPOINT=https://app.datarobot.com/api/v2 DATAROBOT_API_TOKEN= JOINTFM_SCHEMA_VERSION=v5 JOINTFM_DEPLOYMENT_ID= -# Or: JOINTFM_DEPLOYMENT_IDS=chevron-id,research-id +# Or: JOINTFM_DEPLOYMENT_IDS=, # Optional drift-detection pin; the SDK discovers the model version from /healthz when unset: -# JOINTFM_MODEL_VERSION=jointfm-inference:0.3.0+ckpt.fin-2026-05-22 +# JOINTFM_MODEL_VERSION=jointfm-inference:+ckpt. ``` Check that the SDK can resolve the deployment and that the service metadata matches the configured schema and model pins: diff --git a/config.sample.yaml b/config.sample.yaml index dd45a56..9c6c822 100644 --- a/config.sample.yaml +++ b/config.sample.yaml @@ -11,6 +11,7 @@ environment: predict_url: JOINTFM_PREDICT_URL deployment_target: JOINTFM_DEPLOYMENT_TARGET local_base_url: JOINTFM_LOCAL_BASE_URL + local_base_urls: JOINTFM_LOCAL_BASE_URLS pulumi_outputs_path: JOINTFM_PULUMI_OUTPUTS_PATH schema_version: JOINTFM_SCHEMA_VERSION model_version: JOINTFM_MODEL_VERSION @@ -23,6 +24,7 @@ deployment: predict_url: null deployment_target: null local_base_url: null + local_base_urls: null pulumi_outputs_path: null schema_version: null model_version: null diff --git a/docs/api-reference.md b/docs/api-reference.md index 734942d..01b30f6 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -6,9 +6,9 @@ This reference covers the supported public Python surface exported by `jointfm_c | Name | Purpose | | --- | --- | -| `JointFMClient` | Synchronous client for hosted or local JointFM endpoints. Use `from_env()` for `.env` and `config.yaml` backed hosted settings, `health()` for consensus typed service metadata, `health_instances()` for per-deployment probe results and pooled sample topology, `endpoint_client(deployment_id)` for an independent single-endpoint client bound to one configured deployment, `predict(payload)` for low-level JSON prediction, `forecast(...)` for validated tabular forecasts, and the `forecast_mean(...)`, `forecast_samples(...)`, `forecast_quantiles(...)`, and `forecast_log_prob(...)` convenience methods for typed forecast results. `forecast_log_prob(...)` is the one that asks about values the caller already holds: it takes `query_rows`, one observed row per entry of `query_times` carrying every declared column, and returns their log density under the model's joint. Each forecast method accepts `condition=`, one `EqualityCondition` or `IntervalCondition` or a list of them, which switches the request to the `condition` query mode and, before anything is sent, checks that the deployment's health metadata advertises the mode and every condition kind the conditions use. `health()` probes `GET /healthz` for local deployments and POSTs `{"request_type": "health"}` to `predict_url` for hosted DataRobot deployments because the DataRobot deployment gateway only proxies the unstructured prediction route. `feature_importance(...)` runs permutation feature importance: one baseline `forecast_samples` call plus one per shuffled feature column, returning a list of `{"feature", "mean", "distance"}` dicts, each holding that feature's absolute forecast-mean shift and centered squared 2-Wasserstein distance indexed by target and horizon. | +| `JointFMClient` | Synchronous client for hosted or local JointFM endpoints. Use `from_env()` for `.env` and `config.yaml` backed hosted settings, `health()` for consensus typed service metadata, `health_instances()` for per-instance probe results and pooled sample topology, `endpoint_client(instance_id)` for an independent single-endpoint client bound to one configured instance, `predict(payload)` for low-level JSON prediction, `forecast(...)` for validated tabular forecasts, and the `forecast_mean(...)`, `forecast_samples(...)`, `forecast_quantiles(...)`, and `forecast_log_prob(...)` convenience methods for typed forecast results. `forecast_log_prob(...)` is the one that asks about values the caller already holds: it takes `query_rows`, one observed row per entry of `query_times` carrying every declared column, and returns their log density under the model's joint. Each forecast method accepts `condition=`, one `EqualityCondition` or `IntervalCondition` or a list of them, which switches the request to the `condition` query mode and, before anything is sent, checks that the deployment's health metadata advertises the mode and every condition kind the conditions use. `health()` probes `GET /healthz` for a single local service and POSTs `{"request_type": "health"}` to `predict_url` for hosted DataRobot deployments, because the DataRobot deployment gateway only proxies the unstructured prediction route, and for every peer of a pool. `feature_importance(...)` runs permutation feature importance: one baseline `forecast_samples` call plus one per shuffled feature column, returning a list of `{"feature", "mean", "distance"}` dicts, each holding that feature's absolute forecast-mean shift and centered squared 2-Wasserstein distance indexed by target and horizon. | -`JointFMClient.from_env()` loads `config.yaml`, optional `.env` values, and process environment variables. The read-only `timeout` and `retry_config` properties return the `JointFMTimeoutConfig` and `JointFMRetryConfig` the client was built with. `JointFMClient.health(cache=True)` caches health metadata only when requested. `JointFMClient.health_instances()` returns the same probe as a `HealthInstances` object: one `InstanceHealth` per configured deployment (including failures), `max_sample_count` as the sum of reachable caps (one request per instance; each instance's `max_concurrent_requests` says how many such requests it serves at once), and `topology` / `topology_label` grouping those caps (unavailable peers are listed but excluded from the sum and topology). Each endpoint's health payload describes only that endpoint; the client aggregates by calling each configured peer. `health()` still exposes the minimum reachable `max_sample_count`, which is the sample-batch cap used by forecast helpers, and the minimum reachable `max_concurrent_requests`. `JointFMClient.endpoint_client(deployment_id)` returns a new client bound to one configured deployment (a pool instance, or the client's own `settings.deployment_id`) with this client's timeout and retry policy but its own HTTP session and health cache, so several such clients can send requests to the same deployment at once; it raises `JointFMConfigurationError` for any other deployment ID. `JointFMClient.predict(payload)` requires `payload["model_version"]`; high-level forecast helpers resolve the configured model version when the caller does not pass one explicitly. When `forecast_samples(...)` requests an explicit `n_samples`, the client learns the deployment's `max_sample_count` from health metadata before the first prediction, splits oversized requests into capped prediction batches, and returns one merged `SampleForecastResult`. A `JOINTFM_DEPLOYMENT_IDS` pool splits every sample request evenly across all routable endpoints, at least one sample per batch, because downloading the response dominates a sample request and downloads from separate endpoints overlap; each endpoint serves at most one batch at a time. Batch `i` uses seed `seed + i`, so a seeded result depends on the number of routable endpoints. Clients configured without a reachable health route fall back to discovering the cap from the structured service error. +`JointFMClient.from_env()` loads `config.yaml`, optional `.env` values, and process environment variables. The read-only `timeout` and `retry_config` properties return the `JointFMTimeoutConfig` and `JointFMRetryConfig` the client was built with. `JointFMClient.health(cache=True)` caches health metadata only when requested. `JointFMClient.health_instances()` returns the same probe as a `HealthInstances` object: one `InstanceHealth` per configured instance (including failures), `max_sample_count` as the sum of reachable caps (one request per instance; each instance's `max_concurrent_requests` says how many such requests it serves at once), and `topology` / `topology_label` grouping those caps (unavailable peers are listed but excluded from the sum and topology). Each endpoint's health payload describes only that endpoint; the client aggregates by calling each configured peer. `health()` still exposes the minimum reachable `max_sample_count`, which is the sample-batch cap used by forecast helpers, and the minimum reachable `max_concurrent_requests`. `JointFMClient.endpoint_client(instance_id)` returns a new client bound to one configured instance (a pool instance, or the client's own `settings.instance_id`) with this client's timeout and retry policy but its own HTTP session and health cache, so several such clients can send requests to the same instance at once; a local pool instance becomes a `JOINTFM_LOCAL_BASE_URL` client and a hosted one a `JOINTFM_DEPLOYMENT_ID` client. It raises `JointFMConfigurationError` for any other instance ID. `JointFMClient.predict(payload)` requires `payload["model_version"]`; high-level forecast helpers resolve the configured model version when the caller does not pass one explicitly. When `forecast_samples(...)` requests an explicit `n_samples`, the client learns the deployment's `max_sample_count` from health metadata before the first prediction, splits oversized requests into capped prediction batches, and returns one merged `SampleForecastResult`. A pool (`JOINTFM_DEPLOYMENT_IDS` or `JOINTFM_LOCAL_BASE_URLS`) splits every sample request evenly across all routable endpoints, at least one sample per batch, because downloading the response dominates a sample request and downloads from separate endpoints overlap; each endpoint serves at most one batch at a time. Batch `i` uses seed `seed + i`, so a seeded result depends on the number of routable endpoints. Clients configured without a reachable health route fall back to discovering the cap from the structured service error. ## Contract Classes @@ -24,8 +24,8 @@ This reference covers the supported public Python surface exported by `jointfm_c | `ConditionPlausibility` | What the model thinks of the conditions it was given: `equality_log_density` of the pinned values and `region_log_probability` of the interval region given the pinned values at the same position, each `None` when the request carried no condition of that kind. Reported by the service and never refused on. | | `IntervalEstimator` | Numerical accounting (`points`, `effective_sample_size`) behind a region probability that had to be estimated, which happens when some position bounds more than one column. Across several estimated positions it describes the one with the smallest effective sample size. | | `HealthMetadata` | Typed service-health payload with service status, schema and model versions, checkpoint metadata, device, head, `decoding_strategy`, advertised query modes, `supported_condition_kinds` (empty when the deployment cannot condition), return modes, time-index modes, time-index encoding, `max_sample_count`, `max_concurrent_requests` (how many requests the container serves at once; a further request waits for a free slot), an optional `data_generation` block carrying advertised capacity limits, and `supported_metadata_traits` (the expert-hint traits the mounted model reads; empty when the field is absent). The container exposes it on `GET /healthz` for direct local access and as the response to `POST {"request_type": "health"}` on the unstructured prediction route for DataRobot-hosted deployments. Each endpoint reports only its own capabilities. | -| `InstanceHealth` | One configured deployment's probe outcome: `deployment_id`, optional `metadata` (`HealthMetadata` when reachable), and optional `error` when the peer was skipped. | -| `HealthInstances` | Client aggregation of `health_instances()`: `instances` (one `InstanceHealth` per configured ID), `max_sample_count` (sum of reachable caps, one request per instance), `topology` as `(count, cap)` pairs sorted by descending cap, and `topology_label` such as `2x5000` or `1x7000, 1x3000`. Unavailable peers stay in `instances` but are omitted from the sum and topology. | +| `InstanceHealth` | One configured instance's probe outcome: `instance_id` (the deployment ID of a hosted instance, the normalized base URL of a local one), optional `metadata` (`HealthMetadata` when reachable), and optional `error` when the peer was skipped. | +| `HealthInstances` | Client aggregation of `health_instances()`: `instances` (one `InstanceHealth` per configured instance), `max_sample_count` (sum of reachable caps, one request per instance), `topology` as `(count, cap)` pairs sorted by descending cap, and `topology_label` such as `2x5000` or `1x7000, 1x3000`. Unavailable peers stay in `instances` but are omitted from the sum and topology. | | `DataGenerationCapabilities` | Optional service-health block describing the deployed checkpoint's data-generation capacity. Fields are `sampler_type`, `min_series`, `max_series`, `t_input`, `t_output`, `n_input`, and `n_output`. `min_series` and `max_series` bound the total number of series in one request, features and targets counted together — there is no separate per-role cap. | | `UtilizationStage` | The burst of `run_max_utilization_check`: `name`, the total requested `n_samples`, the `request_count` of maximal requests sent at once, `wallclock_seconds` (an upper bound on each request's duration), and `errors`, one entry per request that ended in a JointFM error or a wrong sample count. `passed` is true without errors. | | `MaxUtilizationReport` | Result of `run_max_utilization_check`: the `envelope` the requests were sized from, `decoding_strategy`, `topology_label`, `request_slots` (the sum of `max_concurrent_requests` over reachable endpoints), `unavailable_instances`, `read_timeout_seconds`, `health_seconds`, and the `concurrent_saturation` stage. `passed` requires every endpoint to be reachable and the stage to pass; `format_lines()` renders the report ending in a `VERDICT:` line. | @@ -45,7 +45,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, optional load-balanced `instances` pool, 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. `is_local` tells local selectors from hosted ones; `instance_id` names the single target as a pool instance would (local base URL or deployment ID). 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. | @@ -132,7 +132,7 @@ All SDK-specific exceptions inherit from `JointFMError`. | `DATAROBOT_ENDPOINT` | Hosted calls | HTTPS DataRobot API v2 endpoint, normalized without a trailing slash and required to end in `/api/v2`. | | `DATAROBOT_API_TOKEN` | Hosted calls | Non-empty, whitespace-free API token used in the hosted bearer authorization header. | | `JOINTFM_SCHEMA_VERSION` | Hosted calls | Request schema pin. The SDK supports only `v5`. | -| `JOINTFM_MODEL_VERSION` | Hosted calls | Exact JointFM deployment model version expected from the service-health payload and prediction responses. | +| `JOINTFM_MODEL_VERSION` | Optional | Exact JointFM deployment model version expected from the service-health payload and prediction responses. When unset, the SDK discovers the version from the service-health payload on first use; when set, a mismatch is a hard compatibility error. | | `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`. `health()` uses the minimum reachable `max_sample_count` as the sample-batch cap; `health_instances()` sums reachable caps, one request per instance, and reports topology. Sample forecasts are split evenly across all reachable instances. | | `JOINTFM_DEPLOYMENT_URL` | One selector | Hosted deployment URL; the SDK derives the `/predictionsUnstructured` route from it and reuses that route for health probes. | @@ -140,9 +140,10 @@ All SDK-specific exceptions inherit from `JointFMError`. | `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. | +| `JOINTFM_LOCAL_BASE_URLS` | One selector | Comma-separated local service base URLs for round-robin load balancing (at least two unique URLs after normalization), each validated like `JOINTFM_LOCAL_BASE_URL` and reached without DataRobot authorization. Mutually exclusive with other selectors. Routing, failover, the health gate, and the sample split behave as for `JOINTFM_DEPLOYMENT_IDS`; each instance is named by its base URL. | | `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_IDS`, `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`, `JOINTFM_LOCAL_BASE_URL`, and `JOINTFM_LOCAL_BASE_URLS`. ## Payload Fields @@ -173,7 +174,7 @@ The string literals are exposed as `PREDICT_REQUEST_TYPE`, `HEALTH_REQUEST_TYPE` | `time_column` | For absolute datetime, optional otherwise | Name of the history time column. It must not duplicate a modeled column name. | | `requested_columns` | Optional | Output column names or integer indices, answered in the order given. Duplicates are rejected. Any declared column may be named, whatever its `role` and whatever a condition says about it. Defaults to every declared column in declared order, so a condition response and a forecast response over the same schema line up column for column. Under `return_mode="log_prob"` an explicit list must name every declared column in declared order, because the score is joint over them. | | `query_rows` | `log_prob` mode | Observed rows to score, one per entry of `query_times`, each carrying a value for every declared column. Required by `return_mode="log_prob"` and rejected for every other mode. A row that contradicts a condition — a pinned column given another value, or a bounded column outside its range — is refused by the service instead of scored. | -| `n_samples` | Samples and quantiles controls | Positive sample count when sampling controls are needed. Oversized sample forecasts are batched automatically against the cap advertised in health metadata; a deployment-ID pool splits every sample forecast evenly across its reachable instances. | +| `n_samples` | Samples and quantiles controls | Positive sample count when sampling controls are needed. Oversized sample forecasts are batched automatically against the cap advertised in health metadata; a pool splits every sample forecast evenly across its reachable instances. | | `quantiles` | Quantiles mode | Quantile levels in `(0, 1)`, required for `return_mode="quantiles"`. | | `seed` | Optional | Integer random seed for reproducible stochastic outputs. | | `conditions` | With `query_mode="condition"` | Non-empty list of `{"column", "kind": "equality", "value", "query_time_indices"}` or `{"column", "kind": "interval", "lower", "upper", "query_time_indices"}` entries, `null` marking an open bound. `query_time_indices` is a list of distinct indices into `query_times`, or `null` for every position. Two conditions on one column must cover disjoint positions, and every covered position must leave at least one column unconditioned. Forbidden with any other query mode. | @@ -238,6 +239,7 @@ The string literals are exposed as `PREDICT_REQUEST_TYPE`, `HEALTH_REQUEST_TYPE` | `checkpoint_version` | Loaded checkpoint version. | | `checkpoint_path` | Loaded checkpoint path reported by the service. | | `device` | Device used by inference. | +| `generation` | JointFM generation of the mounted checkpoint: `jointfm-1` or `jointfm-2`. Required; a deployment that does not report it fails to parse. | | `head` | Active forecast head. | | `decoding_strategy` | Horizon decoding mode advertised by the mounted model. Must be one of `SUPPORTED_DECODING_STRATEGIES`: `parallel_dense`, `parallel_scalable`, or `autoregressive`. Parallel strategies decode every horizon in one pass; `autoregressive` rolls horizons sequentially. | | `supported_query_modes` | Non-empty subset of the SDK's query modes (`forecast`, `condition`); the service derives it from the mounted head. | diff --git a/notebooks/forecast_condition.ipynb b/notebooks/forecast_condition.ipynb index 25a199a..e1078a3 100644 --- a/notebooks/forecast_condition.ipynb +++ b/notebooks/forecast_condition.ipynb @@ -51,7 +51,7 @@ "\n", "The `condition` query mode answers this in closed form from the model's joint distribution at each future position. Every condition names the positions it covers by index into `query_times` (`query_time_indices`), and covers every position when it names none. `condition=` takes one condition or a list of them; conditions on different columns may cover the same position, while two conditions on the same column must cover disjoint positions:\n", "\n", - "- An **equality condition** pins a column to a value (`EqualityCondition`). The pinned column leaves the read-out set, because reading it back would only repeat the request.\n", + "- An **equality condition** pins a column to a value (`EqualityCondition`). The pinned column stays nameable in `requested_columns` and reads back the value you pinned, so a scenario answer lines up column for column with an unconditioned one.\n", "- An **interval condition** confines a column to a range whose bounds may be open on either side (`IntervalCondition`). The column stays readable: what comes back is its distribution inside the range.\n", "\n", "The cell below asks for the same forecast twice, once without the block and once with it, because a conditional mean is only readable next to the unconditional one: what the what-if is worth is the shift between them, not the level of either.\n", @@ -105,10 +105,10 @@ "\n", "client = JointFMClient.from_env()\n", "health = client.health(cache=True)\n", - "deployment_ids = [\n", - " entry.deployment_id for entry in client.health_instances(cache=True).instances\n", + "instance_ids = [\n", + " entry.instance_id for entry in client.health_instances(cache=True).instances\n", "]\n", - "print(f\"JointFM deployment ID(s): {deployment_ids}\")\n", + "print(f\"JointFM instance(s): {instance_ids}\")\n", "if \"condition\" not in health.supported_query_modes:\n", " raise RuntimeError(\n", " f\"This deployment serves {list(health.supported_query_modes)} only; \"\n", @@ -552,9 +552,9 @@ ], "metadata": { "kernelspec": { - "display_name": "Python (joint-client-python)", + "display_name": "jointfm-client (3.13.x)", "language": "python", - "name": "joint-client-python" + "name": "python3" }, "language_info": { "codemirror_mode": { @@ -566,7 +566,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/notebooks/forecast_csv.ipynb b/notebooks/forecast_csv.ipynb index c241bda..17d0a56 100644 --- a/notebooks/forecast_csv.ipynb +++ b/notebooks/forecast_csv.ipynb @@ -102,10 +102,10 @@ " )\n", "\n", "client = JointFMClient.from_env()\n", - "deployment_ids = [\n", - " entry.deployment_id for entry in client.health_instances(cache=True).instances\n", + "instance_ids = [\n", + " entry.instance_id for entry in client.health_instances(cache=True).instances\n", "]\n", - "print(f\"JointFM deployment ID(s): {deployment_ids}\")\n", + "print(f\"JointFM instance(s): {instance_ids}\")\n", "plan = plan_forecast_columns(\n", " health=client.health(cache=True),\n", " feature_columns=FEATURE_COLUMNS,\n", @@ -152,7 +152,7 @@ ], "metadata": { "kernelspec": { - "display_name": "joint-client-python (3.13.3)", + "display_name": "jointfm-client (3.13.x)", "language": "python", "name": "python3" }, @@ -166,7 +166,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/notebooks/forecast_mean.ipynb b/notebooks/forecast_mean.ipynb index 5167366..3572706 100644 --- a/notebooks/forecast_mean.ipynb +++ b/notebooks/forecast_mean.ipynb @@ -83,10 +83,10 @@ " raise ValueError(f\"Expected {INPUT_STEPS} history rows, got {len(history)}\")\n", "\n", "client = JointFMClient.from_env()\n", - "deployment_ids = [\n", - " entry.deployment_id for entry in client.health_instances(cache=True).instances\n", + "instance_ids = [\n", + " entry.instance_id for entry in client.health_instances(cache=True).instances\n", "]\n", - "print(f\"JointFM deployment ID(s): {deployment_ids}\")\n", + "print(f\"JointFM instance(s): {instance_ids}\")\n", "plan = plan_forecast_columns(\n", " health=client.health(cache=True),\n", " feature_columns=FEATURE_COLUMNS,\n", @@ -113,9 +113,9 @@ ], "metadata": { "kernelspec": { - "display_name": "Python (joint-client-python)", + "display_name": "jointfm-client (3.13.x)", "language": "python", - "name": "joint-client-python" + "name": "python3" }, "language_info": { "codemirror_mode": { @@ -127,7 +127,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/notebooks/forecast_quantiles.ipynb b/notebooks/forecast_quantiles.ipynb index 7c7ec98..16b83c0 100644 --- a/notebooks/forecast_quantiles.ipynb +++ b/notebooks/forecast_quantiles.ipynb @@ -84,10 +84,10 @@ " raise ValueError(f\"Expected {INPUT_STEPS} history rows, got {len(history)}\")\n", "\n", "client = JointFMClient.from_env()\n", - "deployment_ids = [\n", - " entry.deployment_id for entry in client.health_instances(cache=True).instances\n", + "instance_ids = [\n", + " entry.instance_id for entry in client.health_instances(cache=True).instances\n", "]\n", - "print(f\"JointFM deployment ID(s): {deployment_ids}\")\n", + "print(f\"JointFM instance(s): {instance_ids}\")\n", "plan = plan_forecast_columns(\n", " health=client.health(cache=True),\n", " feature_columns=FEATURE_COLUMNS,\n", @@ -115,7 +115,7 @@ ], "metadata": { "kernelspec": { - "display_name": "joint-client-python (3.13.3)", + "display_name": "jointfm-client (3.13.x)", "language": "python", "name": "python3" }, @@ -129,7 +129,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/notebooks/forecast_samples.ipynb b/notebooks/forecast_samples.ipynb index 94475d9..9e325f3 100644 --- a/notebooks/forecast_samples.ipynb +++ b/notebooks/forecast_samples.ipynb @@ -84,10 +84,10 @@ " raise ValueError(f\"Expected {INPUT_STEPS} history rows, got {len(history)}\")\n", "\n", "client = JointFMClient.from_env()\n", - "deployment_ids = [\n", - " entry.deployment_id for entry in client.health_instances(cache=True).instances\n", + "instance_ids = [\n", + " entry.instance_id for entry in client.health_instances(cache=True).instances\n", "]\n", - "print(f\"JointFM deployment ID(s): {deployment_ids}\")\n", + "print(f\"JointFM instance(s): {instance_ids}\")\n", "plan = plan_forecast_columns(\n", " health=client.health(cache=True),\n", " feature_columns=FEATURE_COLUMNS,\n", @@ -115,7 +115,7 @@ ], "metadata": { "kernelspec": { - "display_name": "joint-client-python (3.13.3)", + "display_name": "jointfm-client (3.13.x)", "language": "python", "name": "python3" }, @@ -129,7 +129,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/notebooks/max_utilization.ipynb b/notebooks/max_utilization.ipynb index 4e3b590..41b4ef9 100644 --- a/notebooks/max_utilization.ipynb +++ b/notebooks/max_utilization.ipynb @@ -49,7 +49,7 @@ "# Maximum Utilization\n", "Fill every request slot of the configured JointFM deployment at once with the largest request its `/healthz` capacity envelope admits, and report a PASS/FAIL verdict with the wall-clock time. The request is sized from the `data_generation` block: `max_series` columns, an `n_input`-row history, an `n_output`-step horizon, and `max_sample_count` sample paths. Every column is a target, so the response is also the largest the deployment can be asked for. The history is a seeded synthetic random walk; only the request's size matters here, not its values.\n", "\n", - "Each reachable endpoint (every pool instance when `JOINTFM_DEPLOYMENT_IDS` configures a pool) advertises `max_concurrent_requests`, the number of requests it serves at the same time, and receives that many maximal requests simultaneously, each for its own `max_sample_count`. This is the one place the client occupies more than one slot per endpoint; ordinary forecasts send at most one request per endpoint at a time.\n", + "Each reachable endpoint (every pool instance when `JOINTFM_DEPLOYMENT_IDS` or `JOINTFM_LOCAL_BASE_URLS` configures a pool) advertises `max_concurrent_requests`, the number of requests it serves at the same time, and receives that many maximal requests simultaneously, each for its own `max_sample_count`. This is the one place the client occupies more than one slot per endpoint; ordinary forecasts send at most one request per endpoint at a time.\n", "\n", "Each run spends real GPU time on the hosted deployment. The terminal equivalent is `uv run jointfm-client max-utilization`, which exits with status 1 on FAIL." ] @@ -134,9 +134,9 @@ ], "metadata": { "kernelspec": { - "display_name": "Python (joint-client-python)", + "display_name": "jointfm-client (3.13.x)", "language": "python", - "name": "joint-client-python" + "name": "python3" }, "language_info": { "codemirror_mode": { @@ -148,7 +148,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/notebooks/pandas_result_conversion.ipynb b/notebooks/pandas_result_conversion.ipynb index d7950c9..1bd7417 100644 --- a/notebooks/pandas_result_conversion.ipynb +++ b/notebooks/pandas_result_conversion.ipynb @@ -78,7 +78,7 @@ ], "metadata": { "kernelspec": { - "display_name": "joint-client-python (3.13.3)", + "display_name": "jointfm-client (3.13.x)", "language": "python", "name": "python3" }, @@ -92,7 +92,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/notebooks/predict_json.ipynb b/notebooks/predict_json.ipynb index 349b8a1..f80c666 100644 --- a/notebooks/predict_json.ipynb +++ b/notebooks/predict_json.ipynb @@ -66,10 +66,10 @@ "from jointfm_client import JointFMClient\n", "\n", "client = JointFMClient.from_env()\n", - "deployment_ids = [\n", - " entry.deployment_id for entry in client.health_instances(cache=True).instances\n", + "instance_ids = [\n", + " entry.instance_id for entry in client.health_instances(cache=True).instances\n", "]\n", - "print(f\"JointFM deployment ID(s): {deployment_ids}\")\n", + "print(f\"JointFM instance(s): {instance_ids}\")\n", "payload = json.loads(Path(\"tests/fixtures/forecast_mean_request.json\").read_text())\n", "payload[\"model_version\"] = client.health(cache=True).model_version\n", "response = client.predict(payload)\n", @@ -83,7 +83,7 @@ ], "metadata": { "kernelspec": { - "display_name": "joint-client-python (3.13.3)", + "display_name": "jointfm-client (3.13.x)", "language": "python", "name": "python3" }, @@ -97,7 +97,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/notebooks/service_health.ipynb b/notebooks/service_health.ipynb index f382847..a104b1d 100644 --- a/notebooks/service_health.ipynb +++ b/notebooks/service_health.ipynb @@ -49,7 +49,7 @@ "# Service Health\n", "Fetch the typed `/healthz` payload from the configured JointFM deployment and display every field. Use this notebook to diagnose health-endpoint issues: version pins, advertised modes, `decoding_strategy` (parallel vs autoregressive horizon decoding), sample-count budgets, and the optional `data_generation` capability block.\n", "\n", - "`JOINTFM_DEPLOYMENT_IDS` configures a round-robin pool of hosted deployments. Each endpoint's health payload describes only that endpoint, so the client probes every configured peer and aggregates locally: `health()` returns consensus metadata whose `max_sample_count` is the **minimum** reachable cap, which is the sample-batch size used to split oversized sample requests. `cache=True` stores this probe so the topology section below reuses it instead of issuing a second round of requests." + "`JOINTFM_DEPLOYMENT_IDS` configures a round-robin pool of hosted deployments, and `JOINTFM_LOCAL_BASE_URLS` one of local services. Each endpoint's health payload describes only that endpoint, so the client probes every configured peer and aggregates locally: `health()` returns consensus metadata whose `max_sample_count` is the **minimum** reachable cap, which is the sample-batch size used to split oversized sample requests. `cache=True` stores this probe so the topology section below reuses it instead of issuing a second round of requests." ] }, { @@ -80,8 +80,8 @@ "language": "markdown" }, "source": [ - "## Deployment Topology\n", - "`health_instances()` returns the per-endpoint results behind that consensus: one entry per configured deployment ID, including peers skipped as unreachable or contract-incompatible. Its `max_sample_count` is the **sum** of reachable caps, the overall parallel capacity of the pool, and `topology_label` groups those caps as `x`. Reachable peers must agree on `model_version` and `checkpoint_version`; a mismatch fails the probe rather than silently mixing models across requests." + "## Pool Topology\n", + "`health_instances()` returns the per-endpoint results behind that consensus: one entry per configured instance, named by its deployment ID or local base URL, including peers skipped as unreachable or contract-incompatible. Its `max_sample_count` is the **sum** of reachable caps, the overall parallel capacity of the pool, and `topology_label` groups those caps as `x`. Reachable peers must agree on `model_version` and `checkpoint_version`; a mismatch fails the probe rather than silently mixing models across requests." ] }, { @@ -97,11 +97,9 @@ "settings = client.settings\n", "instances = client.health_instances(cache=True)\n", "# A pool repeats its primary endpoint in `instances`; the seed covers single-endpoint configs.\n", - "predict_url_by_id: dict[str | None, str] = {\n", - " settings.deployment_id: settings.predict_url\n", - "}\n", + "predict_url_by_id: dict[str | None, str] = {settings.instance_id: settings.predict_url}\n", "for instance in settings.instances:\n", - " predict_url_by_id[instance.deployment_id] = instance.predict_url\n", + " predict_url_by_id[instance.instance_id] = instance.predict_url\n", "\n", "print(f\"selector: {settings.deployment_selector}\")\n", "print(f\"configured endpoints: {len(instances.instances)}\")\n", @@ -112,11 +110,11 @@ "for instance in instances.instances:\n", " print()\n", " if instance.metadata is None:\n", - " print(f\"{instance.deployment_id}: unavailable\")\n", + " print(f\"{instance.instance_id}: unavailable\")\n", " print(f\" error: {instance.error}\")\n", " continue\n", - " print(f\"{instance.deployment_id}: available\")\n", - " print(f\" url: {predict_url_by_id[instance.deployment_id]}\")\n", + " print(f\"{instance.instance_id}: available\")\n", + " print(f\" url: {predict_url_by_id[instance.instance_id]}\")\n", " print(f\" device: {instance.metadata.device}\")\n", " print(f\" image: {instance.metadata.image_version}\")\n", " print(f\" checkpoint: {instance.metadata.checkpoint_version}\")\n", @@ -126,9 +124,9 @@ ], "metadata": { "kernelspec": { - "display_name": "Python (joint-client-python)", + "display_name": "jointfm-client (3.13.x)", "language": "python", - "name": "joint-client-python" + "name": "python3" }, "language_info": { "codemirror_mode": { @@ -140,7 +138,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.3" + "version": "3.13.-1" } }, "nbformat": 4, diff --git a/src/jointfm_client/__init__.py b/src/jointfm_client/__init__.py index bf7c12e..74a45c6 100644 --- a/src/jointfm_client/__init__.py +++ b/src/jointfm_client/__init__.py @@ -126,6 +126,7 @@ JOINTFM_DEPLOYMENT_TARGET_ENV, JOINTFM_DEPLOYMENT_URL_ENV, JOINTFM_LOCAL_BASE_URL_ENV, + JOINTFM_LOCAL_BASE_URLS_ENV, JOINTFM_MODEL_VERSION_ENV, JOINTFM_PREDICT_URL_ENV, JOINTFM_PULUMI_OUTPUTS_PATH_ENV, @@ -200,6 +201,7 @@ "JOINTFM_DEPLOYMENT_TARGET_ENV", "JOINTFM_DEPLOYMENT_URL_ENV", "JOINTFM_LOCAL_BASE_URL_ENV", + "JOINTFM_LOCAL_BASE_URLS_ENV", "JOINTFM_MODEL_VERSION_ENV", "JOINTFM_PREDICT_URL_ENV", "JOINTFM_PULUMI_OUTPUTS_PATH_ENV", diff --git a/src/jointfm_client/cli.py b/src/jointfm_client/cli.py index bd60894..8cbf49c 100644 --- a/src/jointfm_client/cli.py +++ b/src/jointfm_client/cli.py @@ -174,7 +174,7 @@ def _health_command(args: argparse.Namespace, stdout: TextIO) -> int: "service": asdict(health), "instances": [ { - "deployment_id": entry.deployment_id, + "instance_id": entry.instance_id, "available": entry.metadata is not None, "max_sample_count": ( None if entry.metadata is None else entry.metadata.max_sample_count diff --git a/src/jointfm_client/client.py b/src/jointfm_client/client.py index 788183d..d7d1fd9 100644 --- a/src/jointfm_client/client.py +++ b/src/jointfm_client/client.py @@ -76,6 +76,7 @@ from jointfm_client.settings import ( JointFMSettings, build_hosted_deployment_url, + build_local_health_url, load_settings, validate_jointfm_model_version, ) @@ -183,28 +184,30 @@ def retry_config(self) -> JointFMRetryConfig: def health(self, *, cache: bool = False, refresh: bool = False) -> HealthMetadata: """Return service metadata from the configured JointFM endpoint. - Local deployments (``deployment_selector == "local_service"``) probe the + A single local service (``JOINTFM_LOCAL_BASE_URL``) is probed on the container's ``GET /healthz`` route directly. Hosted DataRobot deployments POST ``{"request_type": "health"}`` to ``predict_url`` because DataRobot's 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 + When a pool is configured (``JOINTFM_DEPLOYMENT_IDS`` or + ``JOINTFM_LOCAL_BASE_URLS``), every peer is probed through that + predict-route health request, and reachable peers must share + ``model_version`` and ``checkpoint_version``; the sample-batch cap is the minimum ``max_sample_count`` across those peers, and the advertised ``max_concurrent_requests`` is likewise the minimum. Use - ``health_instances()`` for the per-deployment results of this probe. + ``health_instances()`` for the per-instance results of this probe. """ return self._probe_health(cache=cache, refresh=refresh).metadata def health_instances( self, *, cache: bool = False, refresh: bool = False ) -> HealthInstances: - """Return per-deployment health from the same probe as ``health()``. + """Return per-instance health from the same probe as ``health()``. - A single-endpoint client yields one entry. A deployment-ID pool yields - one entry per configured ID, including peers skipped as unreachable or + A single-endpoint client yields one entry. A pool yields one entry per + configured instance, including peers skipped as unreachable or incompatible. ``max_sample_count`` is the sum of each reachable instance's ``max_sample_count``, one request per instance; each instance's own ``max_concurrent_requests`` says how many such requests it serves at @@ -217,26 +220,27 @@ def health_instances( self._probe_health(cache=cache, refresh=refresh).instances ) - def endpoint_client(self, deployment_id: str | None) -> JointFMClient: - """Return a new single-endpoint client for one configured deployment. - - ``deployment_id`` names one instance of a ``JOINTFM_DEPLOYMENT_IDS`` - pool; for a single-endpoint client it must equal this client's own - ``settings.deployment_id``, which is ``None`` for a local service. The - new client keeps this client's timeout and retry policy but owns its - own HTTP session and health cache, so several of them can send - requests to the same deployment at once. A transport injected into - this client is shared instead, as the pool shares it across peers. - Raises ``JointFMConfigurationError`` without settings or for a - deployment ID this client does not configure. + def endpoint_client(self, instance_id: str | None) -> JointFMClient: + """Return a new single-endpoint client for one configured instance. + + ``instance_id`` is an ``InstanceHealth.instance_id``: one pool + instance's deployment ID (``JOINTFM_DEPLOYMENT_IDS``) or local base + URL (``JOINTFM_LOCAL_BASE_URLS``), or, for a single-endpoint client, + this client's own ``settings.instance_id``. The new client keeps this + client's timeout and retry policy but owns its own HTTP session and + health cache, so several of them can send requests to the same + instance at once. A transport injected into this client is shared + instead, as the pool shares it across peers. Raises + ``JointFMConfigurationError`` without settings or for an instance this + client does not configure. """ settings = self._require_settings("endpoint_client") if self._uses_pool(): - settings = _pool_instance_settings(settings, deployment_id) - elif deployment_id != settings.deployment_id: + settings = _pool_instance_settings(settings, instance_id) + elif instance_id != settings.instance_id: raise JointFMConfigurationError( - f"deployment_id {deployment_id!r} is not configured; this client " - f"targets {settings.deployment_id!r}" + f"instance_id {instance_id!r} is not configured; this client " + f"targets {settings.instance_id!r}" ) return JointFMClient( settings=settings, @@ -270,7 +274,7 @@ def _health_metadata(self) -> HealthMetadata | None: return None if probed is None else probed.metadata def _probe_pool_health(self) -> _ProbedHealth: - """Probe a ``JOINTFM_DEPLOYMENT_IDS`` pool via the shared pool client.""" + """Probe a configured pool via the shared pool client.""" result = self._require_pool().probe_health() return _ProbedHealth(result.metadata, result.instances) @@ -283,20 +287,22 @@ def _probe_single_health(self) -> _ProbedHealth: validate_service_metadata( payload, - expected_model_version=getattr(self.settings, "model_version", None), + expected_model_version=( + None if self.settings is None else self.settings.model_version + ), ) metadata = HealthMetadata.from_payload(payload) - deployment_id = getattr(self.settings, "deployment_id", None) + instance_id = None if self.settings is None else self.settings.instance_id return _ProbedHealth( metadata, - (InstanceHealth(deployment_id=deployment_id, metadata=metadata),), + (InstanceHealth(instance_id=instance_id, metadata=metadata),), ) def _uses_predict_route_for_health(self) -> bool: """Return whether hosted health probes must POST to the predict route.""" if self.settings is None: return False - return self.settings.deployment_selector != "local_service" + return not self.settings.is_local def _health_route_configured(self) -> bool: """Return whether a health probe can be issued with the current settings.""" @@ -936,7 +942,7 @@ def _uses_pool(self) -> bool: def _require_pool(self) -> JointFMInstancePool: if not self._uses_pool(): raise JointFMConfigurationError( - "JointFMClient pool routing requires multiple deployment instances" + "JointFMClient pool routing requires multiple instances" ) if self._pool is None: assert self.settings is not None @@ -1137,21 +1143,35 @@ def _is_history_row_sequence(value: Any) -> bool: def _pool_instance_settings( - settings: JointFMSettings, deployment_id: str | None + settings: JointFMSettings, instance_id: str | None ) -> JointFMSettings: - """Return single-deployment settings for one instance of a pool.""" + """Return the settings one instance of a pool would have on its own. + + A local pool instance becomes a ``JOINTFM_LOCAL_BASE_URL`` target probed + on ``GET /healthz``; a hosted one becomes a ``JOINTFM_DEPLOYMENT_ID`` + target. + """ instance = next( ( instance for instance in settings.instances - if instance.deployment_id == deployment_id + if instance.instance_id == instance_id ), None, ) if instance is None: raise JointFMConfigurationError( - f"deployment_id {deployment_id!r} is not one of the configured pool " - f"instances {[entry.deployment_id for entry in settings.instances]}" + f"instance_id {instance_id!r} is not one of the configured pool " + f"instances {[entry.instance_id for entry in settings.instances]}" + ) + if settings.is_local: + return replace( + settings, + health_url=build_local_health_url(instance.instance_id), + predict_url=instance.predict_url, + deployment_selector="local_service", + instances=(), + local_base_url=instance.instance_id, ) assert settings.datarobot_endpoint is not None return replace( @@ -1160,9 +1180,9 @@ def _pool_instance_settings( predict_url=instance.predict_url, deployment_selector="deployment_id", instances=(), - deployment_id=instance.deployment_id, + deployment_id=instance.instance_id, deployment_url=build_hosted_deployment_url( - settings.datarobot_endpoint, instance.deployment_id + settings.datarobot_endpoint, instance.instance_id ), ) diff --git a/src/jointfm_client/configuration.py b/src/jointfm_client/configuration.py index a82b7bd..0b9e91b 100644 --- a/src/jointfm_client/configuration.py +++ b/src/jointfm_client/configuration.py @@ -59,6 +59,7 @@ class EnvironmentVariableConfig(_ConfigModel): predict_url: str = "JOINTFM_PREDICT_URL" deployment_target: str = "JOINTFM_DEPLOYMENT_TARGET" local_base_url: str = "JOINTFM_LOCAL_BASE_URL" + local_base_urls: str = "JOINTFM_LOCAL_BASE_URLS" pulumi_outputs_path: str = "JOINTFM_PULUMI_OUTPUTS_PATH" schema_version: str = "JOINTFM_SCHEMA_VERSION" model_version: str = "JOINTFM_MODEL_VERSION" @@ -76,14 +77,18 @@ def _validate_environment_name(cls, value: str) -> str: ) return value - def deployment_selector_names(self) -> tuple[str, str, str, str, str]: - """Return the environment names that select one service target.""" + def deployment_selector_names( + self, + ) -> tuple[str, str, str, str, str, str, str]: + """Return the environment names that select a service target or pool.""" return ( self.deployment_id, + self.deployment_ids, self.deployment_url, self.predict_url, self.deployment_target, self.local_base_url, + self.local_base_urls, ) @@ -98,6 +103,7 @@ class HostedDeploymentConfig(_ConfigModel): predict_url: str | None = None deployment_target: str | None = None local_base_url: str | None = None + local_base_urls: str | None = None pulumi_outputs_path: str | None = None schema_version: str | None = None model_version: str | None = None @@ -122,6 +128,7 @@ def to_environment_values( values, environment.deployment_target, self.deployment_target ) _set_if_configured(values, environment.local_base_url, self.local_base_url) + _set_if_configured(values, environment.local_base_urls, self.local_base_urls) _set_if_configured( values, environment.pulumi_outputs_path, self.pulumi_outputs_path ) @@ -363,6 +370,7 @@ def _set_if_configured(values: dict[str, str], name: str, value: str | None) -> JOINTFM_PREDICT_URL_ENV: Final = DEFAULT_ENVIRONMENT_CONFIG.predict_url JOINTFM_DEPLOYMENT_TARGET_ENV: Final = DEFAULT_ENVIRONMENT_CONFIG.deployment_target JOINTFM_LOCAL_BASE_URL_ENV: Final = DEFAULT_ENVIRONMENT_CONFIG.local_base_url +JOINTFM_LOCAL_BASE_URLS_ENV: Final = DEFAULT_ENVIRONMENT_CONFIG.local_base_urls JOINTFM_PULUMI_OUTPUTS_PATH_ENV: Final = DEFAULT_ENVIRONMENT_CONFIG.pulumi_outputs_path JOINTFM_SCHEMA_VERSION_ENV: Final = DEFAULT_ENVIRONMENT_CONFIG.schema_version JOINTFM_MODEL_VERSION_ENV: Final = DEFAULT_ENVIRONMENT_CONFIG.model_version @@ -441,6 +449,7 @@ def _set_if_configured(values: dict[str, str], name: str, value: str | None) -> "JOINTFM_DEPLOYMENT_TARGET_ENV", "JOINTFM_DEPLOYMENT_URL_ENV", "JOINTFM_LOCAL_BASE_URL_ENV", + "JOINTFM_LOCAL_BASE_URLS_ENV", "JOINTFM_MODEL_VERSION_ENV", "JOINTFM_PREDICT_URL_ENV", "JOINTFM_PULUMI_OUTPUTS_PATH_ENV", diff --git a/src/jointfm_client/contract.py b/src/jointfm_client/contract.py index 0b54363..21a2b73 100644 --- a/src/jointfm_client/contract.py +++ b/src/jointfm_client/contract.py @@ -819,9 +819,11 @@ def from_payload(cls, payload: Mapping[str, Any]) -> Self: class HealthMetadata: """Typed representation of `/healthz` service metadata. - ``max_sample_count`` caps the samples one request may ask for, and - ``max_concurrent_requests`` is how many requests the container serves at - once; a further request waits for a free slot. + ``generation`` names the JointFM generation of the mounted checkpoint + (``jointfm-1`` or ``jointfm-2``). ``max_sample_count`` caps the samples + one request may ask for, and ``max_concurrent_requests`` is how many + requests the container serves at once; a further request waits for a + free slot. """ status: str @@ -831,6 +833,7 @@ class HealthMetadata: checkpoint_version: str checkpoint_path: str device: str + generation: str head: str decoding_strategy: DecodingStrategy supported_query_modes: tuple[str, ...] @@ -882,6 +885,10 @@ def from_payload(cls, payload: Mapping[str, Any]) -> Self: field="checkpoint_path", ), device=_require_string(payload.get("device"), field="device"), + generation=_require_string( + payload.get("generation"), + field="generation", + ), head=_require_string(payload.get("head"), field="head"), decoding_strategy=_require_decoding_strategy( payload.get("decoding_strategy") diff --git a/src/jointfm_client/pool.py b/src/jointfm_client/pool.py index 8859a30..44e2f72 100644 --- a/src/jointfm_client/pool.py +++ b/src/jointfm_client/pool.py @@ -12,7 +12,11 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Round-robin load balancing across multiple hosted JointFM deployments.""" +"""Round-robin load balancing across several JointFM instances. + +An instance is a hosted DataRobot deployment or a local service; the pool +treats both alike because each answers the health request on its predict route. +""" from __future__ import annotations @@ -46,7 +50,7 @@ class PoolPeer: - """One hosted deployment endpoint with a Session-safe POST lock.""" + """One JointFM instance endpoint with a Session-safe POST lock.""" def __init__( self, @@ -59,9 +63,9 @@ def __init__( self._post_lock = post_lock @property - def deployment_id(self) -> str: - """Deployment identifier used for routing and health accounting.""" - return self.settings.deployment_id + def instance_id(self) -> str: + """Instance identifier used for routing and health accounting.""" + return self.settings.instance_id def post_json(self, payload: Mapping[str, Any]) -> Mapping[str, Any]: """POST ``payload`` to this peer's predict URL under the peer lock.""" @@ -71,16 +75,16 @@ def post_json(self, payload: Mapping[str, Any]) -> Mapping[str, Any]: @dataclass(frozen=True, slots=True) class InstanceHealth: - """Health probe outcome for one configured JointFM deployment.""" + """Health probe outcome for one configured JointFM instance.""" - deployment_id: str | None + instance_id: str | None metadata: HealthMetadata | None = None error: str | None = None @dataclass(frozen=True, slots=True) class HealthInstances: - """Per-deployment health results and the summed sample budget. + """Per-instance health results and the summed sample budget. ``max_sample_count`` is the sum of reachable peers' caps, one request per peer; a peer advertising ``max_concurrent_requests`` above one serves that @@ -96,7 +100,7 @@ class HealthInstances: @classmethod def from_instances(cls, instances: tuple[InstanceHealth, ...]) -> Self: - """Build the response from per-deployment probe outcomes.""" + """Build the response from per-instance probe outcomes.""" caps = tuple( instance.metadata.max_sample_count for instance in instances @@ -114,7 +118,7 @@ def from_instances(cls, instances: tuple[InstanceHealth, ...]) -> Self: @dataclass(frozen=True, slots=True) class HealthProbeResult: - """Merged health metadata, healthy peer IDs, and per-deployment outcomes.""" + """Merged health metadata, healthy peer IDs, and per-instance outcomes.""" metadata: HealthMetadata healthy_ids: frozenset[str] @@ -127,13 +131,13 @@ class PeerRoutingState: def __init__( self, *, - deployment_ids: Sequence[str], + instance_ids: Sequence[str], peer_cooldown_seconds: float, ) -> None: self._peer_cooldown_seconds = peer_cooldown_seconds self._lock = threading.Lock() self._index = 0 - self._active_ids = set(deployment_ids) + self._active_ids = set(instance_ids) self._cooldown_until: dict[str, float] = {} def eligible(self, peers: Sequence[PoolPeer]) -> tuple[PoolPeer, ...]: @@ -159,29 +163,29 @@ def set_healthy(self, healthy_ids: Sequence[str]) -> None: with self._lock: self._active_ids = set(healthy_ids) self._index = 0 - for deployment_id in healthy_ids: - self._cooldown_until.pop(deployment_id, None) + for instance_id in healthy_ids: + self._cooldown_until.pop(instance_id, None) - def cool_down(self, deployment_id: str) -> None: - """Temporarily exclude ``deployment_id`` from preferred routing.""" + def cool_down(self, instance_id: str) -> None: + """Temporarily exclude ``instance_id`` from preferred routing.""" with self._lock: - self._cooldown_until[deployment_id] = ( + self._cooldown_until[instance_id] = ( time.monotonic() + self._peer_cooldown_seconds ) - def reactivate(self, deployment_id: str) -> None: - """Mark ``deployment_id`` active and clear any cooldown.""" + def reactivate(self, instance_id: str) -> None: + """Mark ``instance_id`` active and clear any cooldown.""" with self._lock: - self._active_ids.add(deployment_id) - self._cooldown_until.pop(deployment_id, None) + self._active_ids.add(instance_id) + self._cooldown_until.pop(instance_id, None) def _eligible_unlocked(self, peers: Sequence[PoolPeer]) -> tuple[PoolPeer, ...]: now = time.monotonic() - active = tuple(peer for peer in peers if peer.deployment_id in self._active_ids) + active = tuple(peer for peer in peers if peer.instance_id in self._active_ids) not_cooling = tuple( peer for peer in active - if self._cooldown_until.get(peer.deployment_id, 0.0) <= now + if self._cooldown_until.get(peer.instance_id, 0.0) <= now ) # All cooling: still try health-active peers rather than stall. return not_cooling or active or tuple(peers) @@ -208,7 +212,7 @@ def probe(self, peers: Sequence[PoolPeer]) -> HealthProbeResult: healthy.append((peer, metadata)) instances.append( InstanceHealth( - deployment_id=peer.deployment_id, + instance_id=peer.instance_id, metadata=metadata, ) ) @@ -216,10 +220,10 @@ def probe(self, peers: Sequence[PoolPeer]) -> HealthProbeResult: # Skip unreachable or incompatible peers; a bad backup must not # take down a healthy primary. last_error = error - _log_unavailable(peer.deployment_id, error) + _log_unavailable(peer.instance_id, error) instances.append( InstanceHealth( - deployment_id=peer.deployment_id, + instance_id=peer.instance_id, error=str(error), ) ) @@ -230,7 +234,7 @@ def probe(self, peers: Sequence[PoolPeer]) -> HealthProbeResult: reference = healthy[0][1] for peer, metadata in healthy[1:]: - _reject_metadata_mismatch(peer.deployment_id, metadata, reference) + _reject_metadata_mismatch(peer.instance_id, metadata, reference) metadata = replace( reference, max_sample_count=min( @@ -242,13 +246,13 @@ def probe(self, peers: Sequence[PoolPeer]) -> HealthProbeResult: ) return HealthProbeResult( metadata=metadata, - healthy_ids=frozenset(peer.deployment_id for peer, _ in healthy), + healthy_ids=frozenset(peer.instance_id for peer, _ in healthy), instances=tuple(instances), ) class JointFMInstancePool: - """Round-robin JointFM requests across hosted deployments. + """Round-robin JointFM requests across hosted deployments or local services. Use one fail-fast transport per peer (``max_attempts=1``); this pool retries peers. Each transport is locked independently so parallel batches stay @@ -280,9 +284,9 @@ def __init__( post_lock = locks_by_transport.setdefault(id(transport), threading.Lock()) peers.append(PoolPeer(instance, transport, post_lock)) self._peers = tuple(peers) - self._peers_by_id = {peer.deployment_id: peer for peer in self._peers} + self._peers_by_id = {peer.instance_id: peer for peer in self._peers} self._routing = PeerRoutingState( - deployment_ids=tuple(peer.deployment_id for peer in self._peers), + instance_ids=tuple(peer.instance_id for peer in self._peers), peer_cooldown_seconds=peer_cooldown_seconds, ) self._health_gate = PoolHealthGate( @@ -325,7 +329,7 @@ def post_json_to( self, instance: JointFMInstanceSettings, payload: Mapping[str, Any] ) -> Mapping[str, Any]: """POST to ``instance`` first; on retryable failure try remaining peers.""" - preferred = self._peers_by_id[instance.deployment_id] + preferred = self._peers_by_id[instance.instance_id] candidates = self._failover_candidates(preferred) last_error: BaseException | None = None for candidate in candidates: @@ -335,10 +339,10 @@ def post_json_to( if not _is_pool_retryable(error): raise last_error = error - self._routing.cool_down(candidate.deployment_id) - _log_unavailable(candidate.deployment_id, error) + self._routing.cool_down(candidate.instance_id) + _log_unavailable(candidate.instance_id, error) continue - self._routing.reactivate(candidate.deployment_id) + self._routing.reactivate(candidate.instance_id) return result assert last_error is not None raise last_error @@ -346,43 +350,41 @@ def post_json_to( def _failover_candidates(self, preferred: PoolPeer) -> tuple[PoolPeer, ...]: """Prefer health-eligible peers; then try health-excluded peers last.""" eligible = self._routing.eligible(self._peers) - if preferred.deployment_id in {peer.deployment_id for peer in eligible}: + if preferred.instance_id in {peer.instance_id for peer in eligible}: preferred_first = (preferred,) + tuple( - peer - for peer in eligible - if peer.deployment_id != preferred.deployment_id + peer for peer in eligible if peer.instance_id != preferred.instance_id ) else: preferred_first = eligible - seen = {peer.deployment_id for peer in preferred_first} + seen = {peer.instance_id for peer in preferred_first} last_resort = tuple( - peer for peer in self._peers if peer.deployment_id not in seen + peer for peer in self._peers if peer.instance_id not in seen ) return preferred_first + last_resort def _reject_metadata_mismatch( - deployment_id: str, + instance_id: str, metadata: HealthMetadata, reference: HealthMetadata, ) -> None: if metadata.model_version != reference.model_version: raise UnsupportedModelVersionError( - "JointFM deployment pool model_version mismatch: " - f"{deployment_id!r} advertises {metadata.model_version!r}, " + "JointFM instance pool model_version mismatch: " + f"{instance_id!r} advertises {metadata.model_version!r}, " f"expected {reference.model_version!r}" ) if metadata.checkpoint_version != reference.checkpoint_version: raise UnsupportedServiceContractError( - "JointFM deployment pool checkpoint_version mismatch: " - f"{deployment_id!r} advertises {metadata.checkpoint_version!r}, " + "JointFM instance pool checkpoint_version mismatch: " + f"{instance_id!r} advertises {metadata.checkpoint_version!r}, " f"expected {reference.checkpoint_version!r}" ) -def _log_unavailable(deployment_id: str, error: BaseException) -> None: +def _log_unavailable(instance_id: str, error: BaseException) -> None: logger.warning( - "JointFM instance unavailable: deployment_id=%s error=%s", deployment_id, error + "JointFM instance unavailable: instance_id=%s error=%s", instance_id, error ) diff --git a/src/jointfm_client/settings.py b/src/jointfm_client/settings.py index 78b5dd3..89f998f 100644 --- a/src/jointfm_client/settings.py +++ b/src/jointfm_client/settings.py @@ -36,6 +36,7 @@ JOINTFM_DEPLOYMENT_TARGET_ENV, JOINTFM_DEPLOYMENT_URL_ENV, JOINTFM_LOCAL_BASE_URL_ENV, + JOINTFM_LOCAL_BASE_URLS_ENV, JOINTFM_MODEL_VERSION_ENV, JOINTFM_PREDICT_URL_ENV, JOINTFM_PULUMI_OUTPUTS_PATH_ENV, @@ -58,14 +59,24 @@ "predict_url", "pulumi_target", "local_service", + "local_services", ] +_LOCAL_DEPLOYMENT_SELECTORS: frozenset[DeploymentSelector] = frozenset( + {"local_service", "local_services"} +) + @dataclass(frozen=True, slots=True) class JointFMInstanceSettings: - """One hosted JointFM deployment target in a load-balanced pool.""" + """One JointFM service target in a load-balanced pool. - deployment_id: str + ``instance_id`` names the peer for routing, health accounting, and + :meth:`JointFMClient.endpoint_client`: the deployment ID of a hosted + deployment, or the normalized base URL of a local service. + """ + + instance_id: str predict_url: str @@ -86,6 +97,20 @@ class JointFMSettings: deployment_target: str | None = None local_base_url: str | None = None + @property + def is_local(self) -> bool: + """Return whether the target is a local service reached without DataRobot.""" + return self.deployment_selector in _LOCAL_DEPLOYMENT_SELECTORS + + @property + def instance_id(self) -> str | None: + """Return the identifier of the single target, as a pool instance names it. + + That is the local base URL for a local service and the deployment ID + for a hosted deployment; a pool reports its first instance. + """ + return self.local_base_url if self.is_local else self.deployment_id + def load_settings( *, @@ -107,9 +132,9 @@ def load_settings( _required_env(env_values, environment.schema_version) ) model_version = _optional_model_version(env_values, environment.model_version) - deployment_ids_value = env_values.get(environment.deployment_ids) - if deployment_ids_value is not None and deployment_ids_value != "": - _reject_conflicting_selectors_with_deployment_ids(env_values, environment) + selector_name = _resolve_deployment_selector(env_values, environment) + + if selector_name == environment.deployment_ids: return _load_hosted_deployment_pool_settings( env_values, environment, @@ -117,7 +142,13 @@ def load_settings( model_version=model_version, ) - selector_name = _resolve_single_deployment_selector(env_values, environment) + if selector_name == environment.local_base_urls: + return _load_local_service_pool_settings( + env_values, + environment, + schema_version=schema_version, + model_version=model_version, + ) if selector_name == environment.local_base_url: local_base_url = normalize_local_service_base_url( @@ -436,24 +467,7 @@ def build_local_predict_url(service_base_url: str) -> str: def normalize_local_service_base_url(value: str) -> str: """Validate and normalize a direct local JointFM service base URL.""" - normalized_url = _normalize_non_whitespace_string( - value, - JOINTFM_LOCAL_BASE_URL_ENV, - ).rstrip("/") - parsed_url = urlparse(normalized_url) - if parsed_url.scheme not in {"http", "https"}: - raise JointFMConfigurationError( - f"{JOINTFM_LOCAL_BASE_URL_ENV} must be an http or https URL" - ) - if not parsed_url.netloc: - raise JointFMConfigurationError( - f"{JOINTFM_LOCAL_BASE_URL_ENV} must include a hostname" - ) - if parsed_url.params or parsed_url.query or parsed_url.fragment: - raise JointFMConfigurationError( - f"{JOINTFM_LOCAL_BASE_URL_ENV} must not include params, query, or fragment" - ) - return normalized_url + return _normalize_local_service_base_url(value, JOINTFM_LOCAL_BASE_URL_ENV) def build_datarobot_prediction_headers(api_token: str) -> dict[str, str]: @@ -497,7 +511,7 @@ def _required_env(env: Mapping[str, str], name: str) -> str: return value -def _resolve_single_deployment_selector( +def _resolve_deployment_selector( env: Mapping[str, str], environment: EnvironmentVariableConfig, ) -> str: @@ -507,12 +521,15 @@ def _resolve_single_deployment_selector( for selector_name in deployment_selector_envs if selector_name in env and env[selector_name] != "" ] - if len(selector_names) != 1: - formatted_selectors = ", ".join( - (*deployment_selector_envs, environment.deployment_ids) + if not selector_names: + raise JointFMConfigurationError( + "Exactly one deployment selector is required: " + f"{', '.join(deployment_selector_envs)}" ) + if len(selector_names) > 1: raise JointFMConfigurationError( - f"Exactly one deployment selector is required: {formatted_selectors}" + "Exactly one deployment selector is required; " + f"{selector_names[0]} cannot be combined with {', '.join(selector_names[1:])}" ) return selector_names[0] @@ -535,7 +552,7 @@ def _load_hosted_deployment_pool_settings( ) instances = tuple( JointFMInstanceSettings( - deployment_id=deployment_id, + instance_id=deployment_id, predict_url=build_hosted_predict_url(datarobot_endpoint, deployment_id), ) for deployment_id in deployment_ids @@ -550,14 +567,73 @@ def _load_hosted_deployment_pool_settings( schema_version=schema_version, instances=instances, model_version=model_version, - deployment_id=primary.deployment_id, + deployment_id=primary.instance_id, deployment_url=build_hosted_deployment_url( datarobot_endpoint, - primary.deployment_id, + primary.instance_id, ), ) +def _load_local_service_pool_settings( + env: Mapping[str, str], + environment: EnvironmentVariableConfig, + *, + schema_version: str, + model_version: str | None, +) -> JointFMSettings: + """Build pool settings for several local services; the first one is primary. + + Every instance is named by its normalized base URL and reached without + DataRobot credentials, exactly like a single ``JOINTFM_LOCAL_BASE_URL``. + """ + base_urls = _parse_local_base_urls(_required_env(env, environment.local_base_urls)) + instances = tuple( + JointFMInstanceSettings( + instance_id=base_url, + predict_url=build_local_predict_url(base_url), + ) + for base_url in base_urls + ) + primary = instances[0] + return JointFMSettings( + datarobot_endpoint=None, + datarobot_api_token=None, + health_url=build_local_health_url(primary.instance_id), + predict_url=primary.predict_url, + deployment_selector="local_services", + schema_version=schema_version, + instances=instances, + model_version=model_version, + local_base_url=primary.instance_id, + ) + + +def _parse_local_base_urls(value: str) -> tuple[str, ...]: + """Return the unique normalized base URLs of a ``JOINTFM_LOCAL_BASE_URLS`` list. + + URLs that normalize to the same base (for example with and without a + trailing slash) count once, so at least two distinct services remain. + """ + if "," not in value: + raise JointFMConfigurationError( + f"{JOINTFM_LOCAL_BASE_URLS_ENV} must be a comma-separated list of " + "at least two unique local service base URLs" + ) + base_urls = tuple( + dict.fromkeys( + _normalize_local_service_base_url(part, JOINTFM_LOCAL_BASE_URLS_ENV) + for part in value.split(",") + ) + ) + if len(base_urls) < 2: + raise JointFMConfigurationError( + f"{JOINTFM_LOCAL_BASE_URLS_ENV} must contain at least two unique " + "local service base URLs" + ) + return base_urls + + def _parse_deployment_ids(value: str) -> tuple[str, ...]: if value.strip() != value or "," not in value: raise JointFMConfigurationError( @@ -583,21 +659,6 @@ def _parse_deployment_ids(value: str) -> tuple[str, ...]: return tuple(deployment_ids) -def _reject_conflicting_selectors_with_deployment_ids( - env: Mapping[str, str], - environment: EnvironmentVariableConfig, -) -> None: - conflicting = [ - name - for name in environment.deployment_selector_names() - if name in env and env[name] != "" - ] - if conflicting: - raise JointFMConfigurationError( - f"{JOINTFM_DEPLOYMENT_IDS_ENV} cannot be combined with {', '.join(conflicting)}" - ) - - def _load_pulumi_target_outputs( outputs_path: str, deployment_target: str, @@ -664,6 +725,20 @@ def _normalize_absolute_https_url(value: str, env_name: str) -> str: return normalized_url +def _normalize_local_service_base_url(value: str, env_name: str) -> str: + normalized_url = _normalize_non_whitespace_string(value, env_name).rstrip("/") + parsed_url = urlparse(normalized_url) + if parsed_url.scheme not in {"http", "https"}: + raise JointFMConfigurationError(f"{env_name} must be an http or https URL") + if not parsed_url.netloc: + raise JointFMConfigurationError(f"{env_name} must include a hostname") + if parsed_url.params or parsed_url.query or parsed_url.fragment: + raise JointFMConfigurationError( + f"{env_name} must not include params, query, or fragment" + ) + return normalized_url + + def _build_local_service_url(service_base_url: str, route: str) -> str: return f"{normalize_local_service_base_url(service_base_url)}{route}" diff --git a/src/jointfm_client/transport.py b/src/jointfm_client/transport.py index 4945ec1..dddea1e 100644 --- a/src/jointfm_client/transport.py +++ b/src/jointfm_client/transport.py @@ -170,7 +170,7 @@ def from_settings( ) -> "JointFMHTTPTransport": """Create an HTTP transport from hosted or local SDK settings.""" headers = None - if settings.deployment_selector != "local_service": + if not settings.is_local: if settings.datarobot_api_token is None: raise JointFMConfigurationError( "hosted JointFM settings require a DataRobot API token" diff --git a/src/jointfm_client/utilization.py b/src/jointfm_client/utilization.py index 8b409c8..3f10c24 100644 --- a/src/jointfm_client/utilization.py +++ b/src/jointfm_client/utilization.py @@ -78,7 +78,7 @@ class MaxUtilizationReport: """Verdict and timings of one maximum-utilization check. ``envelope`` is the advertised capacity the requests were sized from. - ``unavailable_instances`` lists configured deployment IDs that failed the + ``unavailable_instances`` lists configured instance IDs that failed the health probe; any entry fails the check, because the pool then runs below its configured capacity. ``request_slots`` is the sum of ``max_concurrent_requests`` over the reachable endpoints. @@ -111,8 +111,8 @@ def format_lines(self) -> list[str]: f"health probe: {self.health_seconds:.2f}s", ] lines.extend( - f"unavailable: {deployment_id}" - for deployment_id in self.unavailable_instances + f"unavailable: {instance_id}" + for instance_id in self.unavailable_instances ) stage = self.concurrent_saturation status = "PASS" if stage.passed else "FAIL" @@ -156,7 +156,7 @@ def send(self, client: JointFMClient, *, n_samples: int, seed: int) -> str | Non class _RequestSlot: """One request the saturation stage sends at the same time as the others.""" - deployment_id: str | None + instance_id: str | None n_samples: int seed: int @@ -218,18 +218,18 @@ def run_max_utilization_check( query_times=list(range(envelope.n_input, envelope.n_input + envelope.n_output)), ) endpoint_slots = [ - (instance.deployment_id, instance.metadata) + (instance.instance_id, instance.metadata) for instance in instances.instances if instance.metadata is not None for _ in range(instance.metadata.max_concurrent_requests) ] slots = tuple( _RequestSlot( - deployment_id=deployment_id, + instance_id=instance_id, n_samples=metadata.max_sample_count, seed=seed + index, ) - for index, (deployment_id, metadata) in enumerate(endpoint_slots) + for index, (instance_id, metadata) in enumerate(endpoint_slots) ) concurrent_saturation = _run_concurrent_saturation( @@ -242,7 +242,7 @@ def run_max_utilization_check( topology_label=instances.topology_label, request_slots=len(slots), unavailable_instances=tuple( - str(instance.deployment_id) + str(instance.instance_id) for instance in instances.instances if instance.metadata is None ), @@ -265,7 +265,7 @@ def _run_concurrent_saturation( forecast is sent. """ total_samples = sum(slot.n_samples for slot in slots) - slot_clients = tuple(client.endpoint_client(slot.deployment_id) for slot in slots) + slot_clients = tuple(client.endpoint_client(slot.instance_id) for slot in slots) try: for slot_client in slot_clients: slot_client.health(cache=True) @@ -291,9 +291,7 @@ def _send_slot(index: int) -> str | None: # Recorded per slot for the same reason as a failed health probe. except JointFMError as caught: error = f"{type(caught).__name__}: {caught}" - return ( - None if error is None else f"slot {index} ({slot.deployment_id}): {error}" - ) + return None if error is None else f"slot {index} ({slot.instance_id}): {error}" started = time.perf_counter() with ThreadPoolExecutor(max_workers=len(slots)) as executor: diff --git a/tests/fixtures/health_metadata.json b/tests/fixtures/health_metadata.json index 8e50014..5ede7e2 100644 --- a/tests/fixtures/health_metadata.json +++ b/tests/fixtures/health_metadata.json @@ -6,6 +6,7 @@ "checkpoint_version": "sdk-test", "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "studentt", "decoding_strategy": "parallel_dense", "supported_query_modes": ["forecast"], @@ -16,7 +17,7 @@ "continuous_float", "absolute_datetime" ], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": 4096, "max_concurrent_requests": 1, "data_generation": { diff --git a/tests/test_cli.py b/tests/test_cli.py index 4b48ca7..5af4aef 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -64,6 +64,7 @@ def health(self, *, cache: bool = False, refresh: bool = False) -> HealthMetadat checkpoint_version="sdk-test", checkpoint_path="/models/jointfm.pt", device="cpu", + generation="jointfm-1", head="studentt", decoding_strategy="parallel_dense", supported_query_modes=("forecast",), @@ -74,7 +75,7 @@ def health(self, *, cache: bool = False, refresh: bool = False) -> HealthMetadat "continuous_float", "absolute_datetime", ), - time_index_encoding="legacy_discrete_grid", + time_index_encoding="discrete_grid", max_sample_count=4096, max_concurrent_requests=1, ) @@ -87,7 +88,7 @@ def health_instances( return HealthInstances.from_instances( ( InstanceHealth( - deployment_id="deployment-id", + instance_id="deployment-id", metadata=self.health(), ), ) @@ -153,7 +154,7 @@ def test_health_command_prints_non_secret_metadata(monkeypatch, capsys) -> None: assert payload["deployment"]["deployment_id"] == "deployment-id" assert payload["instances"] == [ { - "deployment_id": "deployment-id", + "instance_id": "deployment-id", "available": True, "max_sample_count": 4096, "max_concurrent_requests": 1, diff --git a/tests/test_condition_mode.py b/tests/test_condition_mode.py index 6c08268..d66af7f 100644 --- a/tests/test_condition_mode.py +++ b/tests/test_condition_mode.py @@ -106,6 +106,7 @@ def _health( checkpoint_version="sdk-test", checkpoint_path="/models/jointfm.pt", device="cpu", + generation="jointfm-1", head="gmm", decoding_strategy="parallel_dense", supported_query_modes=query_modes, @@ -561,6 +562,7 @@ def _health_payload( "checkpoint_version": "sdk-test", "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "gmm", "decoding_strategy": "parallel_dense", "supported_query_modes": query_modes, @@ -571,7 +573,7 @@ def _health_payload( "continuous_float", "absolute_datetime", ], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": max_sample_count, "max_concurrent_requests": 1, } diff --git a/tests/test_configuration.py b/tests/test_configuration.py index 35f3c42..22882ef 100644 --- a/tests/test_configuration.py +++ b/tests/test_configuration.py @@ -53,6 +53,7 @@ class _HealthTransport: "checkpoint_version": "yaml", "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "studentt", "decoding_strategy": "parallel_dense", "supported_query_modes": ["forecast"], @@ -63,7 +64,7 @@ class _HealthTransport: "continuous_float", "absolute_datetime", ], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": 4096, "max_concurrent_requests": 1, } diff --git a/tests/test_contract.py b/tests/test_contract.py index 0f694b9..d50ebc5 100644 --- a/tests/test_contract.py +++ b/tests/test_contract.py @@ -82,6 +82,7 @@ def _health_metadata() -> dict[str, object]: "checkpoint_version": "smoke-1", "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "dummy", "decoding_strategy": "parallel_dense", "supported_query_modes": ["forecast"], @@ -92,7 +93,7 @@ def _health_metadata() -> dict[str, object]: "continuous_float", "ordinal", ], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": 4096, "max_concurrent_requests": 1, "data_generation": { @@ -816,6 +817,27 @@ def test_health_metadata_parses_the_series_envelope() -> None: assert health.data_generation.max_series == 16 +def test_health_metadata_parses_the_generation() -> None: + """The served generation is read from its own field, not inferred from another.""" + metadata = _health_metadata() + metadata["generation"] = "jointfm-2" + + assert HealthMetadata.from_payload(metadata).generation == "jointfm-2" + + +@pytest.mark.parametrize("value", [None, "", 2]) +def test_health_metadata_requires_the_generation(value: object) -> None: + """A deployment that does not name its generation fails instead of parsing.""" + metadata = _health_metadata() + if value is None: + del metadata["generation"] + else: + metadata["generation"] = value + + with pytest.raises(ValueError, match="generation"): + HealthMetadata.from_payload(metadata) + + @pytest.mark.parametrize("value", [None, 0, 1.5]) def test_health_metadata_requires_a_positive_concurrency(value: object) -> None: """Request concurrency is advertised capacity, so a missing or invalid value fails.""" diff --git a/tests/test_metadata_traits.py b/tests/test_metadata_traits.py index 9acb5ee..ac00682 100644 --- a/tests/test_metadata_traits.py +++ b/tests/test_metadata_traits.py @@ -42,6 +42,7 @@ def _health(traits: list[str] | None) -> dict[str, Any]: "checkpoint_version": "smoke-1", "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "gmm", "decoding_strategy": "parallel_dense", "supported_query_modes": ["forecast"], @@ -52,7 +53,7 @@ def _health(traits: list[str] | None) -> dict[str, Any]: "continuous_float", "absolute_datetime", ], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": 64, "max_concurrent_requests": 1, } diff --git a/tests/test_pool.py b/tests/test_pool.py index 9e74f82..a3be42a 100644 --- a/tests/test_pool.py +++ b/tests/test_pool.py @@ -42,7 +42,7 @@ def _instance(deployment_id: str) -> JointFMInstanceSettings: """Instance.""" return JointFMInstanceSettings( - deployment_id=deployment_id, + instance_id=deployment_id, predict_url=( "https://app.datarobot.com/api/v2/deployments/" f"{deployment_id}/predictionsUnstructured" @@ -65,6 +65,7 @@ def _health( "checkpoint_version": checkpoint_version, "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "studentt", "decoding_strategy": "parallel_dense", "supported_query_modes": ["forecast"], @@ -75,7 +76,7 @@ def _health( "continuous_float", "absolute_datetime", ], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": max_sample_count, "max_concurrent_requests": 1, } @@ -136,8 +137,8 @@ def _pool( def test_pool_retries_next_instance_on_470() -> None: """Pool retries next instance on 470.""" pool = _pool(transport=_Transport(fail_ids=frozenset({"a"}))) - assert pool.next_instance().deployment_id == "a" - assert pool.next_instance().deployment_id == "b" + assert pool.next_instance().instance_id == "a" + assert pool.next_instance().instance_id == "b" assert pool.post_json({"schema_version": "v5"}) == { "ok": True, "deployment_id": "b", @@ -163,7 +164,7 @@ def test_pool_health_rejects_mismatch_and_aligns_sample_cap() -> None: ) with pytest.raises(UnsupportedModelVersionError, match="model_version"): mismatched.probe_all_health() - assert {mismatched.instance_at(i).deployment_id for i in range(2)} == {"a", "b"} + assert {mismatched.instance_at(i).instance_id for i in range(2)} == {"a", "b"} with pytest.raises(UnsupportedServiceContractError, match="checkpoint_version"): _pool( @@ -229,7 +230,7 @@ def test_health_instances_excludes_unavailable_peer_from_sum_and_topology() -> N ).probe_health() instances = HealthInstances.from_instances(result.instances) - assert [entry.deployment_id for entry in instances.instances] == ["a", "b"] + assert [entry.instance_id for entry in instances.instances] == ["a", "b"] assert instances.instances[0].metadata is None assert instances.instances[0].error is not None assert instances.instances[1].metadata is not None @@ -246,7 +247,7 @@ def _typed_health(*, max_sample_count: int = 4096) -> HealthMetadata: def test_health_instances_from_single_entry() -> None: """A single reachable entry yields one instance and a 1x topology.""" entry = InstanceHealth( - deployment_id="deployment-id", + instance_id="deployment-id", metadata=_typed_health(max_sample_count=4096), ) instances = HealthInstances.from_instances((entry,)) @@ -340,9 +341,9 @@ def test_pool_health_routes_only_reachable_peers() -> None: ) ) pool.probe_all_health() - assert pool.instance_at(0).deployment_id == "b" - assert pool.instance_at(1).deployment_id == "b" - assert pool.next_instance().deployment_id == "b" + assert pool.instance_at(0).instance_id == "b" + assert pool.instance_at(1).instance_id == "b" + assert pool.next_instance().instance_id == "b" assert pool.post_json({"schema_version": "v5"})["deployment_id"] == "b" @@ -354,11 +355,11 @@ def test_pool_failover_retries_health_excluded_peer() -> None: ) pool = _pool(transport=transport) pool.probe_all_health() - assert pool.instance_at(0).deployment_id == "b" + assert pool.instance_at(0).instance_id == "b" transport.fail_ids = frozenset({"b"}) assert pool.post_json({"schema_version": "v5"})["deployment_id"] == "a" - assert pool.instance_at(0).deployment_id == "a" + assert pool.instance_at(0).instance_id == "a" def test_pool_health_skips_incompatible_peer_when_another_matches_pin() -> None: @@ -375,8 +376,8 @@ def test_pool_health_skips_incompatible_peer_when_another_matches_pin() -> None: ) metadata = pool.probe_all_health() assert metadata.model_version == pinned - assert pool.instance_at(0).deployment_id == "a" - assert pool.instance_at(1).deployment_id == "a" + assert pool.instance_at(0).instance_id == "a" + assert pool.instance_at(1).instance_id == "a" assert pool.post_json({"schema_version": "v5"})["deployment_id"] == "a" @@ -390,13 +391,13 @@ def test_pool_cooldown_restores_peer_after_transient_failure( pool = _pool(transport=transport, peer_cooldown_seconds=10.0) assert pool.post_json({"schema_version": "v5"})["deployment_id"] == "b" - assert pool.instance_at(0).deployment_id == "b" - assert pool.instance_at(1).deployment_id == "b" + assert pool.instance_at(0).instance_id == "b" + assert pool.instance_at(1).instance_id == "b" transport.fail_ids = frozenset() clock["now"] = 109.0 - assert pool.instance_at(0).deployment_id == "b" + assert pool.instance_at(0).instance_id == "b" clock["now"] = 110.0 - assert {pool.instance_at(i).deployment_id for i in range(2)} == {"a", "b"} + assert {pool.instance_at(i).instance_id for i in range(2)} == {"a", "b"} assert pool.post_json({"schema_version": "v5"})["deployment_id"] in {"a", "b"} diff --git a/tests/test_result_arrays.py b/tests/test_result_arrays.py index 80a47b8..4f9c856 100644 --- a/tests/test_result_arrays.py +++ b/tests/test_result_arrays.py @@ -307,13 +307,14 @@ def post_json(self, url: str, payload: Mapping[str, Any]) -> Mapping[str, Any]: "checkpoint_version": "smoke-1", "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "studentt", "decoding_strategy": "parallel_dense", "supported_query_modes": ["forecast"], "supported_condition_kinds": [], "supported_return_modes": ["mean", "samples", "quantiles", "log_prob"], "supported_time_index_modes": ["ordinal", "continuous_float", "absolute_datetime"], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": 64, "max_concurrent_requests": 1, } diff --git a/tests/test_settings.py b/tests/test_settings.py index 2769036..160b75b 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -26,6 +26,7 @@ JOINTFM_DEPLOYMENT_TARGET_ENV, JOINTFM_DEPLOYMENT_URL_ENV, JOINTFM_LOCAL_BASE_URL_ENV, + JOINTFM_LOCAL_BASE_URLS_ENV, JOINTFM_MODEL_VERSION_ENV, JOINTFM_PREDICT_URL_ENV, JOINTFM_PULUMI_OUTPUTS_PATH_ENV, @@ -283,12 +284,12 @@ def test_load_settings_resolves_named_pulumi_target_from_saved_outputs( """Load settings resolves named pulumi target from saved outputs.""" outputs_path = tmp_path / "jointfm-pulumi-outputs.json" outputs_path.write_text( - json.dumps({"fin-studentt": {"deployment_id": "pulumi-deployment-id"}}), + json.dumps({"my-target": {"deployment_id": "pulumi-deployment-id"}}), encoding="utf-8", ) env = _hosted_env( **{ - JOINTFM_DEPLOYMENT_TARGET_ENV: "fin-studentt", + JOINTFM_DEPLOYMENT_TARGET_ENV: "my-target", JOINTFM_PULUMI_OUTPUTS_PATH_ENV: str(outputs_path), } ) @@ -296,7 +297,7 @@ def test_load_settings_resolves_named_pulumi_target_from_saved_outputs( settings = load_settings(env=env, dotenv_path=None) - assert settings.deployment_target == "fin-studentt" + assert settings.deployment_target == "my-target" assert settings.deployment_id == "pulumi-deployment-id" assert settings.predict_url == ( "https://app.datarobot.com/api/v2/deployments/" @@ -335,12 +336,12 @@ def test_load_settings_resolves_named_pulumi_target_url_outputs( """Load settings resolves named pulumi target url outputs.""" outputs_path = tmp_path / "jointfm-pulumi-outputs.json" outputs_path.write_text( - json.dumps({"fin-studentt": target_outputs}), + json.dumps({"my-target": target_outputs}), encoding="utf-8", ) env = _hosted_env( **{ - JOINTFM_DEPLOYMENT_TARGET_ENV: "fin-studentt", + JOINTFM_DEPLOYMENT_TARGET_ENV: "my-target", JOINTFM_PULUMI_OUTPUTS_PATH_ENV: str(outputs_path), } ) @@ -349,7 +350,7 @@ def test_load_settings_resolves_named_pulumi_target_url_outputs( settings = load_settings(env=env, dotenv_path=None) assert settings.deployment_selector == "pulumi_target" - assert settings.deployment_target == "fin-studentt" + assert settings.deployment_target == "my-target" assert settings.deployment_id == expected_deployment_id assert settings.predict_url == ( "https://app.datarobot.com/api/v2/deployments/" @@ -360,10 +361,10 @@ def test_load_settings_resolves_named_pulumi_target_url_outputs( def test_load_settings_rejects_pulumi_target_without_url_outputs(tmp_path) -> None: """Load settings rejects pulumi target without url outputs.""" outputs_path = tmp_path / "jointfm-pulumi-outputs.json" - outputs_path.write_text(json.dumps({"fin-studentt": {}}), encoding="utf-8") + outputs_path.write_text(json.dumps({"my-target": {}}), encoding="utf-8") env = _hosted_env( **{ - JOINTFM_DEPLOYMENT_TARGET_ENV: "fin-studentt", + JOINTFM_DEPLOYMENT_TARGET_ENV: "my-target", JOINTFM_PULUMI_OUTPUTS_PATH_ENV: str(outputs_path), } ) @@ -383,12 +384,10 @@ def test_load_settings_rejects_invalid_pulumi_outputs(tmp_path) -> None: missing_target_path = tmp_path / "missing-target.json" missing_target_path.write_text(json.dumps({"other": {}}), encoding="utf-8") non_object_target_path = tmp_path / "non-object-target.json" - non_object_target_path.write_text( - json.dumps({"fin-studentt": []}), encoding="utf-8" - ) + non_object_target_path.write_text(json.dumps({"my-target": []}), encoding="utf-8") invalid_output_path = tmp_path / "invalid-output.json" invalid_output_path.write_text( - json.dumps({"fin-studentt": {"deployment_id": ""}}), + json.dumps({"my-target": {"deployment_id": ""}}), encoding="utf-8", ) @@ -404,7 +403,7 @@ def test_load_settings_rejects_invalid_pulumi_outputs(tmp_path) -> None: for outputs_path, expected_message in invalid_cases: env = _hosted_env( **{ - JOINTFM_DEPLOYMENT_TARGET_ENV: "fin-studentt", + JOINTFM_DEPLOYMENT_TARGET_ENV: "my-target", JOINTFM_PULUMI_OUTPUTS_PATH_ENV: str(outputs_path), } ) @@ -513,7 +512,7 @@ def test_load_settings_with_deployment_ids_builds_instance_pool() -> None: settings = load_settings(env=env, dotenv_path=None) assert settings.deployment_selector == "deployment_ids" - assert [instance.deployment_id for instance in settings.instances] == [ + assert [instance.instance_id for instance in settings.instances] == [ "primary-id", "backup-id", ] @@ -545,3 +544,82 @@ def test_load_settings_rejects_deployment_ids_with_fewer_than_two_unique_ids() - del env[JOINTFM_DEPLOYMENT_ID_ENV] with pytest.raises(JointFMConfigurationError, match="at least two unique"): load_settings(env=env, dotenv_path=None) + + +def _local_pool_env(base_urls: str) -> dict[str, str]: + """Credential-free env selecting a local service pool.""" + return { + JOINTFM_LOCAL_BASE_URLS_ENV: base_urls, + JOINTFM_SCHEMA_VERSION_ENV: "v5", + } + + +def test_load_settings_with_local_base_urls_builds_local_instance_pool() -> None: + """Each local base URL becomes a pool instance named by its normalized URL.""" + settings = load_settings( + env=_local_pool_env("http://127.0.0.1:8080/,http://127.0.0.1:8081"), + dotenv_path=None, + ) + + assert settings.deployment_selector == "local_services" + assert settings.is_local + assert settings.datarobot_endpoint is None + assert settings.datarobot_api_token is None + assert settings.deployment_id is None + assert [instance.instance_id for instance in settings.instances] == [ + "http://127.0.0.1:8080", + "http://127.0.0.1:8081", + ] + assert [instance.predict_url for instance in settings.instances] == [ + "http://127.0.0.1:8080/predict", + "http://127.0.0.1:8081/predict", + ] + assert settings.local_base_url == "http://127.0.0.1:8080" + assert settings.health_url == "http://127.0.0.1:8080/healthz" + assert settings.predict_url == "http://127.0.0.1:8080/predict" + + +@pytest.mark.parametrize( + "base_urls", + ["http://127.0.0.1:8080", "http://127.0.0.1:8080/,http://127.0.0.1:8080"], +) +def test_load_settings_rejects_local_base_urls_with_fewer_than_two_unique_urls( + base_urls: str, +) -> None: + """URLs that normalize to the same service count once.""" + with pytest.raises(JointFMConfigurationError, match="at least two unique"): + load_settings(env=_local_pool_env(base_urls), dotenv_path=None) + + +@pytest.mark.parametrize( + ("base_urls", "message"), + [ + ("http://127.0.0.1:8080,,http://127.0.0.1:8081", "must be non-empty"), + ("http://127.0.0.1:8080, http://127.0.0.1:8081", "must not contain whitespace"), + ("ftp://127.0.0.1:8080,http://127.0.0.1:8081", "must be an http or https URL"), + ("http://127.0.0.1:8080?x=1,http://127.0.0.1:8081", "must not include params"), + ], +) +def test_load_settings_rejects_malformed_local_base_urls( + base_urls: str, message: str +) -> None: + """Every entry is validated like JOINTFM_LOCAL_BASE_URL and named by the list.""" + with pytest.raises( + JointFMConfigurationError, match=f"{JOINTFM_LOCAL_BASE_URLS_ENV} {message}" + ): + load_settings(env=_local_pool_env(base_urls), dotenv_path=None) + + +@pytest.mark.parametrize( + "other_selector", + [JOINTFM_LOCAL_BASE_URL_ENV, JOINTFM_DEPLOYMENT_IDS_ENV, JOINTFM_DEPLOYMENT_ID_ENV], +) +def test_load_settings_rejects_local_base_urls_combined_with_another_selector( + other_selector: str, +) -> None: + """The local pool selector is mutually exclusive with every other selector.""" + env = _local_pool_env("http://127.0.0.1:8080,http://127.0.0.1:8081") + env[other_selector] = "other-value" + + with pytest.raises(JointFMConfigurationError, match="cannot be combined"): + load_settings(env=env, dotenv_path=None) diff --git a/tests/test_transport.py b/tests/test_transport.py index ecd0505..ffe92f7 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -41,6 +41,8 @@ JointFMRequestError, JointFMHTTPStatusError, JointFMHTTPTransport, + JOINTFM_LOCAL_BASE_URLS_ENV, + JOINTFM_SCHEMA_VERSION_ENV, JointFMInstanceSettings, JointFMResponseDecodeError, JointFMServiceError, @@ -51,6 +53,7 @@ SampleForecastResult, UnsupportedModelVersionError, UnsupportedServiceContractError, + load_settings, ) from jointfm_client.contract import PACKAGE_VERSION @@ -161,6 +164,7 @@ def _health_payload( "checkpoint_version": checkpoint_version, "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "studentt", "decoding_strategy": "parallel_dense", "supported_query_modes": ["forecast"], @@ -171,7 +175,7 @@ def _health_payload( "continuous_float", "absolute_datetime", ], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": 4096, "max_concurrent_requests": 1, } @@ -687,7 +691,7 @@ def test_client_health_instances_returns_one_entry_for_single_endpoint() -> None instances = client.health_instances() assert len(instances.instances) == 1 - assert instances.instances[0].deployment_id == "deployment-id" + assert instances.instances[0].instance_id == "deployment-id" assert instances.instances[0].metadata is not None assert ( instances.max_sample_count == instances.instances[0].metadata.max_sample_count @@ -1284,8 +1288,8 @@ def _pool_settings(primary: str, backup: str) -> JointFMSettings: deployment_selector="deployment_ids", schema_version="v5", instances=( - JointFMInstanceSettings(deployment_id="primary-id", predict_url=primary), - JointFMInstanceSettings(deployment_id="backup-id", predict_url=backup), + JointFMInstanceSettings(instance_id="primary-id", predict_url=primary), + JointFMInstanceSettings(instance_id="backup-id", predict_url=backup), ), model_version="jointfm-inference:0.3.0+ckpt.sdk-test", deployment_id="primary-id", @@ -1481,3 +1485,167 @@ def post_json(self, url: str, payload: Mapping[str, Any]) -> Mapping[str, Any]: assert sorted(seed for _, _, seed in predicts) == expected_seeds assert len({url for url, _, _ in predicts}) == len(expected_sizes) assert result.diagnostics.seed == 7 + + +_LOCAL_PRIMARY = "http://127.0.0.1:8080" +_LOCAL_BACKUP = "http://127.0.0.1:8081" + + +def _local_pool_settings() -> JointFMSettings: + """Local pool settings resolved through the JOINTFM_LOCAL_BASE_URLS selector.""" + return load_settings( + env={ + JOINTFM_LOCAL_BASE_URLS_ENV: f"{_LOCAL_PRIMARY},{_LOCAL_BACKUP}", + JOINTFM_SCHEMA_VERSION_ENV: "v5", + }, + dotenv_path=None, + ) + + +class _LocalPoolTransport: + """Fake transport for local services: health on POST /predict, samples echo seeds.""" + + def __init__( + self, + *, + down: frozenset[str] = frozenset(), + checkpoint_by_url: Mapping[str, str] | None = None, + ) -> None: + """Init.""" + self.down = down + self.checkpoint_by_url = checkpoint_by_url or {} + self.requests: list[tuple[str, str, Mapping[str, Any] | None]] = [] + + def get_json(self, url: str) -> Mapping[str, Any]: + """Get json.""" + self.requests.append(("GET", url, None)) + return _health_payload() + + def post_json(self, url: str, payload: Mapping[str, Any]) -> Mapping[str, Any]: + """Post json.""" + self.requests.append(("POST", url, payload)) + if url in self.down: + raise JointFMRequestError(f"connection refused: {url}") + if payload.get("request_type") == "health": + health = _health_payload( + checkpoint_version=self.checkpoint_by_url.get(url, "sdk-test") + ) + health["max_sample_count"] = 2 + return health + if payload.get("return_mode") != "samples": + return _forecast_response_payload() + sample_count = cast(int, payload["n_samples"]) + seed = cast(int, payload["seed"]) + response_payload = _forecast_response_payload(return_mode="samples") + outputs = cast(dict[str, object], response_payload["outputs"]) + outputs["samples"] = [[[float(seed)]] for _ in range(sample_count)] + diagnostics = cast(dict[str, object], response_payload["diagnostics"]) + diagnostics["seed"] = seed + return response_payload + + +def _forecast_local_pool_samples(client: JointFMClient) -> SampleForecastResult: + """Request four seeded samples of one target column.""" + schema = DataFrameSchema( + columns=(ColumnSpec(name="target", modality="numeric", role="target"),), + time_index_mode="ordinal", + ) + return client.forecast_samples( + [{"target": 10.0}, {"target": 11.0}], + schema=schema, + query_times=[2], + requested_columns=["target"], + model_version="jointfm-inference:0.3.0+ckpt.sdk-test", + n_samples=4, + seed=7, + ) + + +def test_local_pool_gates_health_and_splits_samples_across_local_services() -> None: + """A local pool probes every service and spreads sample batches over them.""" + transport = _LocalPoolTransport() + client = JointFMClient(settings=_local_pool_settings(), transport=transport) + + instances = client.health_instances(cache=True) + result = _forecast_local_pool_samples(client) + + assert [entry.instance_id for entry in instances.instances] == [ + _LOCAL_PRIMARY, + _LOCAL_BACKUP, + ] + assert instances.topology_label == "2x2" + assert all(method == "POST" for method, _, _ in transport.requests) + predict_urls = [ + url + for _, url, payload in transport.requests + if payload is not None and payload.get("request_type") != "health" + ] + assert sorted(predict_urls) == [ + f"{_LOCAL_PRIMARY}/predict", + f"{_LOCAL_BACKUP}/predict", + ] + np.testing.assert_array_equal( + result.samples, (((7.0,),), ((7.0,),), ((8.0,),), ((8.0,),)) + ) + + +def test_local_pool_fails_over_when_a_local_service_is_unreachable() -> None: + """An unreachable local service is skipped by the health gate and by routing.""" + transport = _LocalPoolTransport(down=frozenset({f"{_LOCAL_PRIMARY}/predict"})) + client = JointFMClient(settings=_local_pool_settings(), transport=transport) + + instances = client.health_instances(cache=True) + result = _forecast_local_pool_samples(client) + + assert instances.instances[0].metadata is None + assert instances.instances[1].metadata is not None + assert len(result.samples) == 4 + predict_urls = { + url + for _, url, payload in transport.requests + if payload is not None and payload.get("request_type") != "health" + } + assert predict_urls == {f"{_LOCAL_BACKUP}/predict"} + + +def test_local_pool_rejects_local_services_on_different_checkpoints() -> None: + """Local peers must serve the same checkpoint, as hosted peers must.""" + transport = _LocalPoolTransport( + checkpoint_by_url={f"{_LOCAL_BACKUP}/predict": "other-checkpoint"} + ) + client = JointFMClient(settings=_local_pool_settings(), transport=transport) + + with pytest.raises(UnsupportedServiceContractError, match=_LOCAL_BACKUP): + client.health() + + +def test_local_pool_endpoint_client_targets_one_local_service() -> None: + """A local pool instance's own client probes that service on GET /healthz.""" + transport = _LocalPoolTransport() + client = JointFMClient(settings=_local_pool_settings(), transport=transport) + + endpoint = client.endpoint_client(_LOCAL_BACKUP) + endpoint.health() + + assert endpoint.settings is not None + assert endpoint.settings.deployment_selector == "local_service" + assert endpoint.settings.instances == () + assert endpoint.settings.instance_id == _LOCAL_BACKUP + assert endpoint.predict_url == f"{_LOCAL_BACKUP}/predict" + assert transport.requests == [("GET", f"{_LOCAL_BACKUP}/healthz", None)] + with pytest.raises(JointFMConfigurationError, match="http://127.0.0.1:9999"): + client.endpoint_client("http://127.0.0.1:9999") + + +def test_transport_from_local_pool_settings_omits_hosted_auth_headers() -> None: + """Local pool peers are reached without DataRobot credentials.""" + session = requests.Session() + transport = JointFMHTTPTransport.from_settings( + session=session, settings=_local_pool_settings() + ) + adapter = StaticResponseAdapter(_response(body=b'{"ok": true}')) + session.mount("http://", adapter) + + transport.post_json(f"{_LOCAL_BACKUP}/predict", {"request_type": "health"}) + + assert "Authorization" not in dict(adapter.requests[0].headers or {}) diff --git a/tests/test_utilization.py b/tests/test_utilization.py index 08816cc..ccb749e 100644 --- a/tests/test_utilization.py +++ b/tests/test_utilization.py @@ -72,7 +72,7 @@ def _settings(*deployment_ids: str) -> JointFMSettings: schema_version="v5", instances=tuple( JointFMInstanceSettings( - deployment_id=deployment_id, predict_url=_url(deployment_id) + instance_id=deployment_id, predict_url=_url(deployment_id) ) for deployment_id in deployment_ids ) @@ -98,6 +98,7 @@ def _health_payload( "checkpoint_version": "sdk-test", "checkpoint_path": "/models/jointfm.pt", "device": "cpu", + "generation": "jointfm-1", "head": "studentt", "decoding_strategy": "parallel_dense", "supported_query_modes": ["forecast"], @@ -108,7 +109,7 @@ def _health_payload( "continuous_float", "absolute_datetime", ], - "time_index_encoding": "legacy_discrete_grid", + "time_index_encoding": "discrete_grid", "max_sample_count": max_sample_count, "max_concurrent_requests": max_concurrent_requests, }