diff --git a/charts/grid-operator/crds/gridnetwork.yaml b/charts/grid-operator/crds/gridnetwork.yaml index 7822f0d..4007ed5 100644 --- a/charts/grid-operator/crds/gridnetwork.yaml +++ b/charts/grid-operator/crds/gridnetwork.yaml @@ -381,6 +381,24 @@ spec: items: type: string type: array + selectionPolicy: + description: |- + Local request distribution policy for the active selection group. + + This is independent of scoring. When absent, the overlay carries no + selection override and Praxis uses deterministic selection. + nullable: true + properties: + mode: + description: Local selection mode used by the data-plane gateway. + enum: + - deterministic + - roundRobin + - random + type: string + required: + - mode + type: object staleCandidateTtlSeconds: description: |- Maximum age in seconds before a stale (`fresh=false`) remote routing diff --git a/charts/grid-site/templates/gridnetwork.yaml b/charts/grid-site/templates/gridnetwork.yaml index 379f297..bd5cf48 100644 --- a/charts/grid-site/templates/gridnetwork.yaml +++ b/charts/grid-site/templates/gridnetwork.yaml @@ -1,4 +1,15 @@ {{- if .Values.gridNetwork.name }} +{{- $existing := lookup "grid.praxis-proxy.io/v1alpha1" "GridNetwork" .Release.Namespace .Values.gridNetwork.name }} +{{- $configuredSelectionPolicy := .Values.gridNetwork.selectionPolicy }} +{{- $existingSelectionPolicy := dig "spec" "selectionPolicy" nil $existing }} +{{- $selectionPolicy := $configuredSelectionPolicy }} +{{- if not $selectionPolicy }} + {{- if $existingSelectionPolicy }} + {{- $selectionPolicy = $existingSelectionPolicy }} + {{- else if not $existing }} + {{- $selectionPolicy = dict "mode" "roundRobin" }} + {{- end }} +{{- end }} apiVersion: grid.praxis-proxy.io/v1alpha1 kind: GridNetwork metadata: @@ -33,6 +44,12 @@ spec: admissionPolicy: {{- toYaml . | nindent 4 }} {{- end }} + {{- with $selectionPolicy }} + {{- if .mode }} + selectionPolicy: + mode: {{ .mode | quote }} + {{- end }} + {{- end }} {{- with .Values.gridNetwork.metricsRefreshInterval }} {{- if . }} metricsRefreshInterval: {{ . | quote }} diff --git a/charts/grid-site/tests/gridnetwork_test.yaml b/charts/grid-site/tests/gridnetwork_test.yaml index fbe5df9..7ad37ae 100644 --- a/charts/grid-site/tests/gridnetwork_test.yaml +++ b/charts/grid-site/tests/gridnetwork_test.yaml @@ -32,6 +32,15 @@ tests: path: spec.gridId value: "" + - it: defaults a new GridNetwork to roundRobin selection + set: + gridNetwork.name: prod-grid + gridSite.name: prod-site + asserts: + - equal: + path: spec.selectionPolicy.mode + value: roundRobin + - it: renders budgetPolicy.tenants verbatim when set set: gridNetwork.name: prod-grid diff --git a/charts/grid-site/values.schema.json b/charts/grid-site/values.schema.json index 19b825b..ac1b416 100644 --- a/charts/grid-site/values.schema.json +++ b/charts/grid-site/values.schema.json @@ -71,6 +71,17 @@ } } }, + "selectionPolicy": { + "type": "object", + "additionalProperties": false, + "required": ["mode"], + "properties": { + "mode": { + "type": "string", + "enum": ["deterministic", "roundRobin", "random"] + } + } + }, "metricsRefreshInterval": { "type": "string", "pattern": "^([1-9][0-9]*s|[1-9][0-9]{3,}ms)$" diff --git a/deploy/crds/gridnetwork.yaml b/deploy/crds/gridnetwork.yaml index 7822f0d..4007ed5 100644 --- a/deploy/crds/gridnetwork.yaml +++ b/deploy/crds/gridnetwork.yaml @@ -381,6 +381,24 @@ spec: items: type: string type: array + selectionPolicy: + description: |- + Local request distribution policy for the active selection group. + + This is independent of scoring. When absent, the overlay carries no + selection override and Praxis uses deterministic selection. + nullable: true + properties: + mode: + description: Local selection mode used by the data-plane gateway. + enum: + - deterministic + - roundRobin + - random + type: string + required: + - mode + type: object staleCandidateTtlSeconds: description: |- Maximum age in seconds before a stale (`fresh=false`) remote routing diff --git a/docs/README.md b/docs/README.md index 79fe397..b528df1 100644 --- a/docs/README.md +++ b/docs/README.md @@ -9,6 +9,9 @@ - [Routing](architecture/routing.md) — versioned overlay contract, revision lifecycle, candidate ordering, `intelligent_route`, `peer_identity_trust`, and provider-side request forwarding. +- [Provider Selection and Load Balancing](architecture/provider-selection-and-load-balancing.md) — + eligibility, routing groups, scoring, selection modes, affinity, and + overlay lifecycle. - [Scoring](architecture/scoring.md) — operator-side candidate scoring, metrics input, and request-time scoring boundaries. - [Auth and Policy](architecture/auth.md) — provider authentication strategies, diff --git a/docs/architecture/provider-selection-and-load-balancing.md b/docs/architecture/provider-selection-and-load-balancing.md new file mode 100644 index 0000000..7576617 --- /dev/null +++ b/docs/architecture/provider-selection-and-load-balancing.md @@ -0,0 +1,442 @@ +# Provider Selection and Load Balancing + +Grid and Praxis divide provider routing into two parts: + +- Grid observes provider state asynchronously, decides which providers are + eligible, orders them, and publishes a versioned routing overlay. +- Praxis consumes that overlay and makes the final request-time choice locally + in the `intelligent_route` filter. + +This separation keeps Kubernetes, Grid reconciliation, EPP metrics, and remote +coordination out of the request hot path. + +```text +Client + | + v +Consumer gateway + | + v +intelligent_route + | + v +First viable selection group + +--> Provider gateway A + +--> Provider gateway B + `--> Provider gateway C + +Lower-priority group + `--> Provider gateway D +``` + +A, B, and C can share active traffic when the selected policy permits it. D is +fallback capacity and is not selected while the earlier group remains viable. +A selection group is a priority and resilience boundary; it is not a score +bucket and does not represent a traffic percentage. + +This layer balances requests across provider gateways. After a provider gateway +is selected, its local serving stack can make a separate backend-level decision. +For example, an llm-d provider gateway can delegate endpoint selection to EPP. +Grid does not use round-robin to choose individual inference replicas hidden +behind one provider gateway. + +## Choose a configuration + +Use these starting points, then adjust them to match the deployment's routing +goal: + +| Goal | Routing policy | Scoring strategy | Selection mode | +|---|---|---|---| +| Share traffic across nearby generic providers | `geographyFirst` | `noMetrics` | `roundRobin` | +| Keep nearby providers active and remote providers as fallback | `geographyFirst` | Any | `roundRobin` | +| Send new traffic to the highest-ranked provider | `geographyFirst` or `scoreFirst` | Any | `deterministic` | +| Share traffic across sites without inference metrics | `scoreFirst` | `noMetrics` | `roundRobin` | +| Randomize selection inside the preferred provider group | Either | Any | `random` | + +The three policy fields answer different questions: + +- `routingPolicy`: Which providers belong in the same active or fallback group? +- `scoringPolicy`: How should providers be ranked using available metrics? +- `selectionPolicy`: How should Praxis choose within the active group? + +## The decision sequence + +```text +Request for a capability + | + v +Eligibility and admission + | + v +Routing policy orders candidates and creates priority groups + | + v +Session-affinity lookup + +-- permitted existing binding -> reuse its provider + `-- no usable binding -> find the first viable group + | + v + apply the configured selection mode + +-- deterministic mode + +-- roundRobin mode + `-- random mode +``` + +Scores contribute to candidate ordering. They do not create groups. The +selection mode operates only inside the first group that can serve the request. + +## Eligibility and admission + +Before ordering, Grid builds candidates for the requested capability. The +overlay already reflects capability matching, authorization and trust, +provider health, freshness, and provider availability. Admission is a hard +boundary: + +| Admission state | New requests | Existing sessions | +|---|---:|---:| +| `newAndExisting` | Allowed | Allowed | +| `existingOnly` | Not selected | Allowed when the binding is permitted | +| `excluded` | Not selected | Not selected | + +An `existingOnly` provider can finish work for a session that is already bound +to it, but it does not receive new bindings. An excluded provider cannot be +selected. Neither scoring nor a selection mode can override these states. + +## Routing policy and groups + +`spec.routingPolicy` controls candidate ordering and hard priority boundaries. +The supported values are `geographyFirst` and `scoreFirst`. + +### `geographyFirst` + +Candidates are ordered by admission, locality tier, freshness, score, and +deterministic identity tie-breakers. Groups are separated by admission, +locality tier, and freshness. Scores order candidates within a group but do +not split that group. + +```text +Closest healthy and fresh tier: A, B, C <- active selection +More distant healthy tier: D, E <- fallback +``` + +In plain language: balance within the closest healthy provider tier and use +more distant capacity as fallback. A remote provider does not join local +active traffic merely because its score is higher. + +### `scoreFirst` + +Candidates are ordered by admission, freshness, score, locality, and +deterministic identity tie-breakers. Groups are separated by admission and +freshness only. Fresh admitted providers from different sites can therefore +share one active group. Score differences affect order, not group membership. + +In plain language: allow fresh admitted providers across sites to participate +in the same active traffic group. + +## Scoring policy + +`spec.scoringPolicy.strategy` selects the provider-level signal used for score +calculation. It is independent of request-time selection: + +- `noMetrics` requires no EPP, Prometheus, or inference-specific metrics. + Dynamic score contributions are zero, while health, admission, freshness, + authorization, locality, affinity, and selection policy still apply. This is + the normal choice for generic or heterogeneous provider gateways. Use it when + llm-d EPP metrics are unavailable or are not comparable across providers. +- `queueDepth` uses asynchronously observed, normalized provider-pool queue + pressure. Lower pressure produces a higher preference score. It requires + comparable queue metrics and a meaningful queue capacity. For an llm-d + provider, Grid retrieves this signal from the configured EPP metrics endpoint. +- `kvCachePressure` uses provider-level KV-cache utilization as a capacity + pressure signal. Lower utilization produces a higher score. It is not + request-specific prefix-cache affinity; that decision belongs inside the + inference scheduler. For an llm-d provider, Grid retrieves this signal from + the configured EPP metrics endpoint. + +Grid currently uses one explicitly selected strategy rather than blending +unrelated signals into an opaque total. Missing local samples can use the +implementation's neutral fallback values, and a recent local sample can be +reused while it remains within `staleMetricsSeconds`. Deployments using a +metric strategy should provide fresh, comparable telemetry for every competing +provider. + +The important rule is: + +```text +score != traffic weight +``` + +Scores are preference and observability signals. They do not turn a score of +`0.8` versus `0.4` into a 2:1 traffic split. With `roundRobin`, candidates in +the active group receive equal turns regardless of their scores. + +For detailed metric input and normalization, see [Provider Scoring](scoring.md). + +## Selection policy + +`spec.selectionPolicy.mode` controls request-time selection inside the first +viable group. The selection mode is applied from an accepted in-memory +snapshot by Praxis; Grid is not called for each request. + +### `deterministic` + +Selects the first viable candidate in the active group. This is strict +preference behavior: Grid's ordering determines which provider receives new +unbound traffic. It is useful when locality, score, primary/standby order, or +predictability should dominate. When `selectionPolicy` is absent from an +overlay, Praxis uses `deterministic`. + +### `roundRobin` + +Takes equal turns across viable candidates in the active group. It does not +require inference metrics and does not distribute across lower-priority groups +while the active group is viable. It balances selections, not necessarily +tokens, latency, request cost, or concurrent work. Session affinity is checked +before this mode runs. + +### `random` + +Selects uniformly from viable candidates in the active group. It follows the +same admission, group, and affinity rules as round-robin. Random state is local +to the gateway process and is not a global coordinator. + +## Policy matrix + +| Routing policy | Selection policy | Effective behavior | +|---|---|---| +| `geographyFirst` | `deterministic` | Strict preference for the highest-ranked provider in the closest viable tier | +| `geographyFirst` | `roundRobin` | Equal selection in the closest viable tier; remote tiers are fallback | +| `geographyFirst` | `random` | Uniform selection in the closest viable tier | +| `scoreFirst` | `deterministic` | Strict preference for the highest-ranked fresh admitted provider across sites | +| `scoreFirst` | `roundRobin` | Equal selection across fresh admitted providers in the active group | +| `scoreFirst` | `random` | Uniform selection across fresh admitted providers in the active group | + +The scoring strategy changes ordering, not the selection mode: + +| Scoring strategy | Metrics required | Deterministic | Round-robin | +|---|---|---|---| +| `noMetrics` | No | Ordering and deterministic tie-breaks decide | Equal selection inside the active group | +| `queueDepth` | Compatible queue metrics | Highest queue-based preference is first | Scores remain visible; selection remains equal | +| `kvCachePressure` | Compatible KV metrics | Highest available-capacity preference is first | Scores remain visible; selection remains equal | + +## Configuration examples + +### Generic provider-gateway balancing + +```yaml +apiVersion: grid.praxis-proxy.io/v1alpha1 +kind: GridNetwork +metadata: + name: provider-grid +spec: + routingPolicy: geographyFirst + scoringPolicy: + strategy: noMetrics + selectionPolicy: + mode: roundRobin +``` + +Providers in the nearest viable group share selections equally. No inference +metrics are required, and remote groups remain available for fallback. + +### Strict metric preference + +```yaml +apiVersion: grid.praxis-proxy.io/v1alpha1 +kind: GridNetwork +metadata: + name: inference-grid +spec: + routingPolicy: scoreFirst + scoringPolicy: + strategy: queueDepth + selectionPolicy: + mode: deterministic + metricsRefreshInterval: "10s" +``` + +Grid refreshes the selected signal asynchronously. Deterministic selection +uses the resulting ordering; it does not query EPP during a request. + +### Cross-site active/active selection + +```yaml +apiVersion: grid.praxis-proxy.io/v1alpha1 +kind: GridNetwork +metadata: + name: active-active-grid +spec: + routingPolicy: scoreFirst + scoringPolicy: + strategy: noMetrics + selectionPolicy: + mode: roundRobin +``` + +Fresh, admitted providers from multiple sites can share the active group. + +### Random selection + +```yaml +apiVersion: grid.praxis-proxy.io/v1alpha1 +kind: GridNetwork +metadata: + name: random-provider-grid +spec: + routingPolicy: geographyFirst + scoringPolicy: + strategy: noMetrics + selectionPolicy: + mode: random +``` + +Random selection is uniform within the active group. It is useful when an +equal probabilistic distribution is sufficient and a repeating sequence is not +required. + +## Request-time behavior and affinity + +For a new request, Praxis reads the accepted snapshot, resolves the requested +capability, checks affinity, finds the first viable group, applies the selection mode, +and records a binding when the selection succeeds. + +```text +Client + | + | request + v +Consumer gateway / intelligent_route + | 1. Read accepted in-memory overlay + | 2. Resolve capability and eligibility + | 3. Check session affinity + | 4. Find the first viable group + | 5. Apply the configured selection mode + | 6. Record the successful binding + v +Selected provider gateway + | + v +Provider backend +``` + +For an existing permitted binding, no new selection mode is applied: + +```text +Client with an existing session + | + v +intelligent_route + | + | permitted affinity binding found + v +Previously selected provider +``` + +Round-robin does not move an established session just to improve aggregate +balance. The observed request distribution can therefore differ from an exact +split when sessions generate different amounts of traffic. + +## Multiple consumer gateways + +The design supports multiple consumer gateways. Each gateway receives an +accepted overlay snapshot and keeps its own local selection state: + +```text + Grid overlay + / \ + v v + Consumer gateway 1 Consumer gateway 2 + local counter local counter + A -> B -> C A -> B -> C +``` + +Counters are not coordinated globally. Each gateway can produce a balanced +local sequence, while aggregate traffic depends on request rates, affinity, +restarts, and snapshot replacement. A globally synchronized quota would need +a different coordination design and would add hot-path trade-offs. + +## Overlay lifecycle and re-ranking + +```text +Provider health and optional EPP metrics + | + v +Grid operator reconciliation + | eligibility, admission, ordering, scores, groups, selection policy + v +Content-addressed routing overlay + | + v +overlay-sync validation and publication + | + v +Praxis validates and atomically loads a snapshot + | precomputed group index and local selection state + v +Request-time selection from memory +``` + +Reconciliation is triggered by watched provider, site, and network changes, +remote Grid state, and the periodic `metricsRefreshInterval`. The default +periodic cadence is 300 seconds for plaintext metrics; TLS-protected metrics +use a 60-second safety cap. A configured interval must be at least one second. +The interval controls observation and overlay publication, not request-path +latency. A demo or operator can cause an earlier reconcile through a real +watched resource change. + +After an overlay is accepted, requests do not call Grid, Kubernetes, +ConfigMaps, EPP, Prometheus, or a remote scoring service. An unchanged semantic +revision should not continually rebuild selection state. An accepted semantic +change may create fresh snapshot-scoped state. + +## Failure and fallback + +Grid observes health, admission, freshness, and optional metrics +asynchronously. Praxis serves from the last accepted snapshot until a valid +new overlay is delivered. An unavailable or excluded provider cannot receive +new selections, and the first viable group is preferred. Later groups are +fallback capacity, not part of normal active distribution. + +Failover is therefore bounded by observation, reconciliation, overlay +distribution, and snapshot acceptance. It is not an immediate request-time +call to Grid. A request already sent upstream can fail before a newer snapshot +is accepted; do not assume automatic retry unless the gateway configuration +explicitly provides it. + +## Overlay contract and future weighting + +`selectionPolicy` is optional in both the Grid API and the overlay. An omitted +field remains omitted, and Praxis interprets it as deterministic selection. +The Helm chart explicitly renders `roundRobin` by default. Users applying a +`GridNetwork` directly can either set the selection mode explicitly or omit the policy +to select `deterministic`. + +`selection_group` and `selection_policy` are part of the semantic digest when +present. Group numbers are zero-based and contiguous per capability. Unknown +mode values and malformed policy structures are rejected. Overlays without +the optional selection fields remain valid and use deterministic selection. + +The provider ordering used to construct groups is also part of the routing +contract. In particular, the updated `geographyFirst` ordering evaluates +freshness before score after locality. Existing resources that omit +`selectionPolicy` still use deterministic selection, but an upgrade can change +which provider is first when candidates differ in freshness. Deployments that +require a fixed primary provider should set their routing and selection policy +explicitly and validate the resulting overlay during upgrade. + +Weighted selection is a future extension, not part of the current API. A +future mode such as `weightedRandom` would need an explicit overlay weight, +normalization, capacity semantics, missing-metric behavior, bounds, and +stability controls. It must not be inferred from score, rank, metric presence, +or candidate count, and it must not change admission, locality, +authorization, freshness, or group boundaries. + +## Demonstration reference + +The [Grid provider-selection research spike](https://github.com/praxis-proxy/grid/issues/31) +describes the focused provider-traffic demonstration: one consumer gateway, +three provider gateways, one active group, `noMetrics`, `roundRobin`, 60 +successful requests, exact 20/20/20 attribution, and a stable overlay during +the measured window. That proof demonstrates equal selection, not weighted +routing, coordinated round-robin across multiple consumers, retry behavior, +or fallback groups unless separate evidence is provided. diff --git a/docs/architecture/routing.md b/docs/architecture/routing.md index 5be5d35..438ca3b 100644 --- a/docs/architecture/routing.md +++ b/docs/architecture/routing.md @@ -32,6 +32,12 @@ llm-d / EPP / inference backend Grid does not proxy traffic. It writes the overlay used by Praxis filters. +For the complete provider-selection model, including selection groups and the +`deterministic`, `roundRobin`, and `random` modes, see [Provider Selection +and Load Balancing](provider-selection-and-load-balancing.md). Routing policy +defines candidate ordering and hard group boundaries; selection policy controls +request distribution within the first viable group. + ## Control-plane rendering path For each `GridNetwork` and gateway reference, the operator: @@ -363,8 +369,8 @@ The operator orders candidates before writing the overlay. Ordering proceeds in two phases: 1. **Scoring.** Each provider is scored by `scoring::score_backends` using - the weights selected by `GridNetwork.spec.scoringPolicy.strategy`, - optional live metrics, and optional CRDT-propagated provider metrics. + the signal selected by `GridNetwork.spec.scoringPolicy.strategy`, optional + live metrics, and optional CRDT-propagated provider metrics. Providers with no live metrics use neutral metric scores. See [Scoring](scoring.md) for the available strategies. @@ -375,7 +381,7 @@ in two phases: | Policy | Order | |---|---| - | `geographyFirst` (default) | admission, locality, score descending, freshness, deterministic tie-break | + | `geographyFirst` (default) | admission, locality, freshness, score descending, deterministic tie-break | | `scoreFirst` | admission, freshness, score descending, locality, deterministic tie-break | The deterministic tie-break is `(site, name, cluster)`. After diff --git a/docs/architecture/scoring.md b/docs/architecture/scoring.md index c9796a0..3f25bb4 100644 --- a/docs/architecture/scoring.md +++ b/docs/architecture/scoring.md @@ -1,5 +1,10 @@ # Provider Scoring +For the complete relationship between scoring, routing groups, and request-time +selection, see [Provider Selection and Load Balancing](provider-selection-and-load-balancing.md). +Scores influence candidate ordering; they are not traffic weights and do not +split selection groups. + Grid scores provider pools when the operator renders a Praxis routing overlay. Praxis reads that overlay from memory at request time; it does not call Grid, Kubernetes, the operator, or an EPP metrics endpoint on the request path. @@ -53,7 +58,7 @@ spec: Omitting `scoringPolicy` has the same effect. Every admitted candidate receives zero dynamic score. "No metrics" does not mean "no policy": health, admission, freshness, model compatibility, geography, selection tiers, session affinity, -and Praxis picker policy continue to apply. +and Praxis selection policy continue to apply. Use this strategy for heterogeneous grids, external APIs such as OpenAI, Anthropic, or Bedrock, and providers that do not expose comparable pool diff --git a/operator/src/controller/grid_network.rs b/operator/src/controller/grid_network.rs index ffd97c8..21dd57d 100644 --- a/operator/src/controller/grid_network.rs +++ b/operator/src/controller/grid_network.rs @@ -4378,6 +4378,7 @@ mod tests { network: "net".to_owned(), local_site: "site".to_owned(), candidates: Vec::new(), + selection_policy: None, generated_at: Some("2026-07-29T01:00:00Z".to_owned()), }, }, diff --git a/operator/src/crd/grid_network.rs b/operator/src/crd/grid_network.rs index 2dcc342..7684aba 100644 --- a/operator/src/crd/grid_network.rs +++ b/operator/src/crd/grid_network.rs @@ -60,7 +60,7 @@ pub enum ScoringStrategy { /// Do not prefer providers using dynamic metrics. /// /// All score contributions are zero. Health, admission, freshness, - /// geography, selection tiers, session affinity, and request-time picker + /// geography, selection tiers, session affinity, and request-time selection /// policy still apply. This is the generic default. #[default] NoMetrics, @@ -143,6 +143,28 @@ pub struct ScoringPolicyConfig { pub strategy: ScoringStrategy, } +/// Local request-selection mode applied by Praxis inside the active group. +#[derive(Clone, Copy, Debug, Default, Deserialize, JsonSchema, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum SelectionMode { + /// Select the first admitted candidate in the active group. + #[default] + Deterministic, + /// Distribute new requests equally in the active group. + RoundRobin, + /// Distribute new requests randomly in the active group. + Random, +} + +/// Request selection policy published in the routing overlay. +#[derive(Clone, Debug, Default, Deserialize, JsonSchema, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +#[schemars(deny_unknown_fields)] +pub struct SelectionPolicyConfig { + /// Local selection mode used by the data-plane gateway. + pub mode: SelectionMode, +} + /// Resolve the effective [`scoring::ScoringWeights`] from a scoring policy. /// /// The public API selects one scorer. The weight adapter is internal and @@ -517,7 +539,8 @@ pub struct GridNetworkSpec { /// Admission state (`newAndExisting` before `existingOnly`) always /// outranks both geography and score in either mode. In `scoreFirst` /// mode, freshness also outranks both; in `geographyFirst` mode, - /// freshness is a tiebreaker below geography and score. + /// freshness is below locality but above score so it cannot interleave + /// candidates across freshness-based selection groups. #[serde(default, skip_serializing_if = "Option::is_none")] pub routing_policy: Option, @@ -540,6 +563,13 @@ pub struct GridNetworkSpec { #[serde(default, skip_serializing_if = "Option::is_none")] pub admission_policy: Option, + /// Local request distribution policy for the active selection group. + /// + /// This is independent of scoring. When absent, the overlay carries no + /// selection override and Praxis uses deterministic selection. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub selection_policy: Option, + /// Maximum time between metric refreshes and score/ranking recalculation. /// /// This controls the `GridNetwork` reconcile cadence for provider metrics; @@ -1932,6 +1962,37 @@ mod tests { assert!(result.is_err(), "unknown strategy must be rejected"); } + #[test] + fn selection_policy_round_trips_and_rejects_unknown_mode() { + let spec: GridNetworkSpec = serde_json::from_value(serde_json::json!({ + "seeds": [], + "selectionPolicy": { "mode": "roundRobin" } + })) + .unwrap_or_else(|_| std::process::abort()); + assert_eq!( + spec.selection_policy.map(|policy| policy.mode), + Some(SelectionMode::RoundRobin) + ); + + let result = serde_json::from_value::(serde_json::json!({ + "seeds": [], + "selectionPolicy": { "mode": "weightedRandom" } + })); + let Err(error) = result else { + std::process::abort(); + }; + assert!(error.to_string().contains("unknown variant")); + } + + #[test] + fn round_robin_mode_serializes_for_explicit_policy() { + let serialized = serde_json::to_value(&SelectionPolicyConfig { + mode: SelectionMode::RoundRobin, + }) + .unwrap_or_else(|_| std::process::abort()); + assert_eq!(serialized, serde_json::json!({"mode": "roundRobin"})); + } + #[test] fn scoring_policy_rejects_removed_weights_field() { let json = serde_json::json!({ diff --git a/operator/src/resources/consumer_config.rs b/operator/src/resources/consumer_config.rs index 1c1c9ae..e96e8d2 100644 --- a/operator/src/resources/consumer_config.rs +++ b/operator/src/resources/consumer_config.rs @@ -23,7 +23,7 @@ use std::collections::{BTreeMap, BTreeSet}; use k8s_openapi::api::core::v1::ConfigMap; use crate::{ - crd::grid_network::{ClusterEndpointConfig, TransportMode}, + crd::grid_network::{ClusterEndpointConfig, SelectionMode, TransportMode}, resources::routing_overlay::{RoutingCandidate, RoutingOverlay}, }; @@ -149,6 +149,7 @@ pub(crate) fn generate_consumer_praxis_config( } let candidates_yaml = render_candidates(&overlay.candidates); + let selection_policy_yaml = render_selection_policy(overlay.selection_policy.as_ref()); let local_site = yaml_scalar(&overlay.local_site)?; let credential_inject_section = render_credential_inject(&overlay.candidates, credential_mount_base); @@ -170,6 +171,7 @@ pub(crate) fn generate_consumer_praxis_config( \x20 - filter: intelligent_route\n\ \x20 local_site: {local_site}\n\ \x20 model_header: \"X-Model\"\n\ + {selection_policy_yaml}\ \x20 candidates:\n\ {candidates_yaml}" ); @@ -229,7 +231,24 @@ fn render_candidates(candidates: &[RoutingCandidate]) -> String { candidates.iter().map(render_candidate).collect::>().join("\n") } +/// Render the explicit Grid-owned request-selection policy. +fn render_selection_policy(policy: Option<&crate::crd::grid_network::SelectionPolicyConfig>) -> String { + let Some(policy) = policy else { + return String::new(); + }; + let mode = match policy.mode { + SelectionMode::Deterministic => "deterministic", + SelectionMode::RoundRobin => "roundRobin", + SelectionMode::Random => "random", + }; + format!(" selection_policy:\n mode: {mode}\n") +} + /// Render one `intelligent_route` candidate. +#[expect( + clippy::too_many_lines, + reason = "Candidate YAML fields are kept together to mirror the wire contract." +)] fn render_candidate(c: &RoutingCandidate) -> String { let mut lines = vec![ format!( @@ -250,6 +269,15 @@ fn render_candidate(c: &RoutingCandidate) -> String { ), format!(" fresh: {}", c.fresh), ]; + if let Some(admission) = c.admission_state { + lines.push(format!( + " admission_state: {}", + serde_json::to_string(&admission).unwrap_or_default() + )); + } + if let Some(group) = c.selection_group { + lines.push(format!(" selection_group: {group}")); + } if let Some(cred) = &c.credential { lines.extend(render_credential_reference(cred)); } @@ -524,6 +552,7 @@ mod tests { score: None, score_breakdown: None, rank: None, + selection_group: None, } } @@ -556,6 +585,7 @@ mod tests { score: None, score_breakdown: None, rank: None, + selection_group: None, } } @@ -564,6 +594,7 @@ mod tests { network: "test-net".to_owned(), local_site: "site-a".to_owned(), candidates, + selection_policy: None, generated_at: None, } } @@ -625,6 +656,25 @@ mod tests { assert!(yaml.contains("gateway-site-a"), "cluster must appear in load_balancer"); } + #[test] + fn explicit_selection_policy_reaches_intelligent_route_config() { + let mut overlay = simple_overlay(vec![plain_candidate( + "inference_model", + "model", + "site-a", + "cluster-a", + true, + )]); + overlay.selection_policy = Some(crate::crd::grid_network::SelectionPolicyConfig { + mode: SelectionMode::RoundRobin, + }); + let endpoints = endpoint_coverage(&overlay); + let config = generate_consumer_praxis_config(&overlay, MOUNT_BASE, &endpoints, "/run/tls", 8080) + .unwrap_or_else(|_| std::process::abort()); + assert!(config.contains("selection_policy:")); + assert!(config.contains("mode: roundRobin")); + } + #[test] fn generated_praxis_yaml_omits_operator_overlay_metadata() { let mut candidate = plain_candidate("inference_model", "model-a", "site-a", "gateway-site-a", true); @@ -632,6 +682,7 @@ mod tests { candidate.admission_state = Some(AdmissionState::NewAndExisting); candidate.selection_tier = Some(LocalityTier::SameSite); candidate.rank = Some(0); + candidate.selection_group = Some(0); let mut overlay = simple_overlay(vec![candidate]); overlay.generated_at = Some("2026-07-24T12:00:00Z".to_owned()); @@ -645,12 +696,14 @@ mod tests { ) .unwrap(); - for forbidden in ["stable_id", "admission_state", "selection_tier", "rank", "generated_at"] { + for forbidden in ["stable_id", "selection_tier", "rank", "generated_at"] { assert!( !yaml.contains(forbidden), "operator-only metadata field {forbidden} must not enter generated Praxis YAML" ); } + assert!(yaml.contains("admission_state: \"new_and_existing\"")); + assert!(yaml.contains("selection_group: 0")); } #[test] @@ -931,6 +984,7 @@ mod tests { network: "n".to_owned(), local_site: String::new(), candidates: vec![], + selection_policy: None, generated_at: None, }; assert!( diff --git a/operator/src/resources/overlay_bridge.rs b/operator/src/resources/overlay_bridge.rs index 588950f..ca0dd71 100644 --- a/operator/src/resources/overlay_bridge.rs +++ b/operator/src/resources/overlay_bridge.rs @@ -98,8 +98,10 @@ mod tests { score: None, score_breakdown: None, rank: None, + selection_group: None, }) .collect(), + selection_policy: None, generated_at: None, } } diff --git a/operator/src/resources/overlay_envelope.rs b/operator/src/resources/overlay_envelope.rs index f1d8939..9461dbe 100644 --- a/operator/src/resources/overlay_envelope.rs +++ b/operator/src/resources/overlay_envelope.rs @@ -4,9 +4,9 @@ //! //! Wraps a routing overlay in a versioned envelope with a SHA-256 digest //! computed over the [RFC 8785][] canonical form of the semantic payload. -//! The semantic payload includes only routing-relevant fields (`network`, -//! `local_site`, `candidates`), so the revision changes only when routing -//! behavior changes — timestamp and provenance changes are no-ops. +//! The semantic payload includes `network`, `local_site`, and `candidates`, +//! plus `selection_policy` when present, so the revision changes only when +//! routing behavior changes — timestamp and provenance changes are no-ops. //! //! The envelope is published as the `routing-overlay.json` key in the overlay //! `ConfigMap`, alongside the `routing-config.json` legacy key. @@ -182,11 +182,17 @@ pub struct EnvelopeBuildResult { /// [`ProjectedCredentialRef`]: super::routing_overlay::ProjectedCredentialRef pub fn compute_semantic_digest(overlay: &RoutingOverlay) -> Result { let candidates_value = serde_json::to_value(&overlay.candidates)?; - let semantic_payload = serde_json::json!({ + let mut semantic_payload = serde_json::json!({ "candidates": candidates_value, "local_site": overlay.local_site, "network": overlay.network, }); + if let Some(policy) = &overlay.selection_policy { + let Some(object) = semantic_payload.as_object_mut() else { + return Err(serde_json::Error::custom("semantic payload must be an object")); + }; + object.insert("selection_policy".to_owned(), serde_json::to_value(policy)?); + } let canonical = serde_json_canonicalizer::to_vec(&semantic_payload).map_err(serde_json::Error::custom)?; @@ -307,7 +313,9 @@ mod tests { score: None, score_breakdown: None, rank: Some(0), + selection_group: None, }], + selection_policy: None, generated_at: Some("2026-07-29T00:00:00Z".to_owned()), } } @@ -341,6 +349,7 @@ mod tests { score: None, score_breakdown: None, rank: Some(0), + selection_group: None, }, RoutingCandidate { kind: "inference_model".to_owned(), @@ -355,8 +364,10 @@ mod tests { score: None, score_breakdown: None, rank: Some(1), + selection_group: None, }, ], + selection_policy: None, generated_at: Some("2026-07-29T01:00:00Z".to_owned()), } } diff --git a/operator/src/resources/routing_overlay.rs b/operator/src/resources/routing_overlay.rs index 9d5991b..4fbc00a 100644 --- a/operator/src/resources/routing_overlay.rs +++ b/operator/src/resources/routing_overlay.rs @@ -270,6 +270,7 @@ pub(crate) fn remote_crdt_provider_to_candidates(provider: &crdt::ProviderState) score: None, score_breakdown: None, rank: None, + selection_group: None, }) .collect() } @@ -877,6 +878,13 @@ pub struct RoutingCandidate { /// Zero-based position in the final sorted overlay. #[serde(skip_serializing_if = "Option::is_none")] pub rank: Option, + + /// Zero-based active selection group for request distribution. + /// + /// Group metadata is additive and does not change the default ordered + /// routing behavior until the data plane explicitly enables it. + #[serde(skip_serializing_if = "Option::is_none")] + pub selection_group: Option, } /// The full routing overlay for a single [`GridNetwork`]. @@ -905,10 +913,15 @@ pub struct RoutingOverlay { /// higher than remote candidates. pub local_site: String, - /// Routing candidates, ordered by admission state, locality tier, score, - /// freshness, then alphabetical tiebreak. + /// Routing candidates, ordered by admission state, locality tier, + /// freshness, score, then alphabetical tiebreak. pub candidates: Vec, + /// Optional explicit local selection policy for Praxis. An absent field is + /// intentionally backward-compatible and means deterministic selection. + #[serde(skip_serializing_if = "Option::is_none")] + pub selection_policy: Option, + /// RFC 3339 timestamp of when this overlay was rendered. #[serde(skip_serializing_if = "Option::is_none")] pub generated_at: Option, @@ -1017,6 +1030,71 @@ fn locality_sort_key(tier: Option) -> u8 { } } +/// Assign deterministic priority groups after the existing candidate sort. +/// +/// Groups are hard eligibility/priority boundaries, not score buckets. The +/// current producer contract deliberately leaves metric differences inside a +/// group so a later traffic-weight field can distribute among eligible +/// candidates without changing the active priority boundary. +/// +/// Geography-first keeps admission, locality, and freshness as boundaries; +/// Score-first keeps admission and freshness as boundaries. Scores are +/// therefore never used to split a group, and floating-point equality is not +/// part of the grouping contract. +#[derive(Clone, Copy)] +struct GroupState { + /// Admission boundary for the previous candidate. + admission: AdmissionState, + /// Freshness boundary for the previous candidate. + fresh: bool, + /// Locality boundary for the previous candidate. + tier: LocalityTier, + /// Group assigned to the previous candidate. + group: u32, +} + +/// Assign zero-based contiguous groups for each capability. +#[expect( + clippy::too_many_lines, + reason = "The two routing policies are explicit at this contract boundary." +)] +fn assign_selection_groups(candidates: &mut [RoutingCandidate], policy: crate::crd::grid_network::RoutingPolicy) { + let mut last_by_capability: HashMap<(String, String), GroupState> = HashMap::new(); + for candidate in candidates { + let key = (candidate.kind.clone(), candidate.name.clone()); + let admission = candidate.admission_state.unwrap_or(AdmissionState::NewAndExisting); + let tier = candidate.selection_tier.unwrap_or(LocalityTier::Unknown); + let next = match last_by_capability.get(&key) { + Some(last) => { + let boundary = match policy { + crate::crd::grid_network::RoutingPolicy::GeographyFirst => { + admission != last.admission || tier != last.tier || candidate.fresh != last.fresh + }, + crate::crd::grid_network::RoutingPolicy::ScoreFirst => { + admission != last.admission || candidate.fresh != last.fresh + }, + }; + if boundary { + last.group.saturating_add(1) + } else { + last.group + } + }, + None => 0, + }; + candidate.selection_group = Some(next); + last_by_capability.insert( + key, + GroupState { + admission, + fresh: candidate.fresh, + tier, + group: next, + }, + ); + } +} + // --------------------------------------------------------------------------- // Renderer // --------------------------------------------------------------------------- @@ -1049,10 +1127,14 @@ fn locality_sort_key(tier: Option) -> u8 { /// **`GeographyFirst`** (default): /// 1. admission state: `new_and_existing` before `existing_only`; /// 2. geography tier: same site, same zone, same region, cross region; -/// 3. scoring engine score, descending; -/// 4. `fresh=true` before `fresh=false`; +/// 3. freshness: `fresh=true` before `fresh=false`; +/// 4. scoring engine score, descending; /// 5. deterministic `(site, name, cluster)` tiebreak. /// +/// Freshness intentionally precedes score in this policy. Group assignment +/// treats freshness as a hard boundary, so this ordering prevents a stale +/// candidate from interleaving with fresh candidates of the same capability. +/// /// **`ScoreFirst`**: /// 1. admission state: `new_and_existing` before `existing_only`; /// 2. `fresh=true` before `fresh=false`; @@ -1221,8 +1303,8 @@ pub fn render_routing_overlay_with_admission( admission_sort_key(a.admission_state) .cmp(&admission_sort_key(b.admission_state)) .then(locality_sort_key(a.selection_tier).cmp(&locality_sort_key(b.selection_tier))) - .then_with(|| score_of(&b.cluster).total_cmp(&score_of(&a.cluster))) .then(b.fresh.cmp(&a.fresh)) + .then_with(|| score_of(&b.cluster).total_cmp(&score_of(&a.cluster))) .then(a.site.cmp(&b.site)) .then(a.name.cmp(&b.name)) .then(a.cluster.cmp(&b.cluster)) @@ -1256,10 +1338,19 @@ pub fn render_routing_overlay_with_admission( } } + assign_selection_groups(&mut candidates, policy); + + // Preserve omission for existing GridNetwork resources. New Helm + // installations resolve their round-robin default in chart values, while + // an upgraded resource without this field must retain deterministic + // Praxis compatibility behavior. + let selection_policy = network.spec.selection_policy.clone(); + Ok(RoutingOverlay { network: network_name.to_owned(), local_site: local_site.to_owned(), candidates, + selection_policy, generated_at: generated_at.map(str::to_owned), }) } @@ -1455,6 +1546,7 @@ fn candidates_from_provider( score: None, score_breakdown: None, rank: None, + selection_group: None, }); } } @@ -1670,6 +1762,156 @@ mod tests { .unwrap_or_else(|_| std::process::abort()) } + #[expect( + clippy::too_many_arguments, + reason = "Test fixture mirrors all grouping dimensions explicitly." + )] + fn group_candidate( + name: &str, + site: &str, + cluster: &str, + admission_state: AdmissionState, + tier: LocalityTier, + fresh: bool, + score: f64, + ) -> RoutingCandidate { + RoutingCandidate { + kind: CANDIDATE_KIND.to_owned(), + name: name.to_owned(), + site: site.to_owned(), + cluster: cluster.to_owned(), + fresh, + credential: None, + stable_id: None, + admission_state: Some(admission_state), + selection_tier: Some(tier), + score: Some(score), + score_breakdown: None, + rank: None, + selection_group: None, + } + } + + #[test] + fn equivalent_candidates_share_group_without_identity_comparison() { + let mut candidates = vec![ + group_candidate( + "model-a", + "site-a", + "cluster-a", + AdmissionState::NewAndExisting, + LocalityTier::SameSite, + true, + 0.8, + ), + group_candidate( + "model-a", + "site-b", + "cluster-b", + AdmissionState::NewAndExisting, + LocalityTier::SameSite, + true, + 0.8, + ), + ]; + assign_selection_groups(&mut candidates, crate::crd::grid_network::RoutingPolicy::GeographyFirst); + assert_eq!(candidates[0].selection_group, Some(0)); + assert_eq!(candidates[1].selection_group, Some(0)); + } + + #[test] + fn changed_routing_properties_split_groups_deterministically() { + let mut candidates = vec![ + group_candidate( + "model-a", + "site-a", + "cluster-a", + AdmissionState::NewAndExisting, + LocalityTier::SameSite, + true, + 0.8, + ), + group_candidate( + "model-a", + "site-b", + "cluster-b", + AdmissionState::NewAndExisting, + LocalityTier::CrossRegion, + true, + 0.8, + ), + group_candidate( + "model-a", + "site-c", + "cluster-c", + AdmissionState::ExistingOnly, + LocalityTier::CrossRegion, + false, + 0.2, + ), + ]; + assign_selection_groups(&mut candidates, crate::crd::grid_network::RoutingPolicy::GeographyFirst); + assert_eq!( + candidates.iter().map(|c| c.selection_group).collect::>(), + vec![Some(0), Some(1), Some(2)] + ); + } + + #[test] + fn metric_differences_stay_in_same_group_for_both_policies() { + let mut geography = vec![ + group_candidate( + "model-a", + "site-a", + "cluster-a", + AdmissionState::NewAndExisting, + LocalityTier::SameSite, + true, + 0.9, + ), + group_candidate( + "model-a", + "site-b", + "cluster-b", + AdmissionState::NewAndExisting, + LocalityTier::SameSite, + true, + 0.1, + ), + ]; + assign_selection_groups(&mut geography, crate::crd::grid_network::RoutingPolicy::GeographyFirst); + assert_eq!( + geography.iter().map(|c| c.selection_group).collect::>(), + vec![Some(0), Some(0)] + ); + + let mut score_first = vec![ + group_candidate( + "model-a", + "site-a", + "cluster-a", + AdmissionState::NewAndExisting, + LocalityTier::SameSite, + true, + 0.9, + ), + group_candidate( + "model-a", + "site-b", + "cluster-b", + AdmissionState::NewAndExisting, + LocalityTier::SameSite, + true, + 0.1, + ), + ]; + assign_selection_groups(&mut score_first, crate::crd::grid_network::RoutingPolicy::ScoreFirst); + assert_eq!( + score_first.iter().map(|c| c.selection_group).collect::>(), + vec![Some(0), Some(0)] + ); + } + fn test_site(name: &str, network: &str) -> GridSite { serde_json::from_value(serde_json::json!({ "apiVersion": "grid.praxis-proxy.io/v1alpha1", @@ -1714,6 +1956,27 @@ mod tests { .unwrap_or_else(|_| std::process::abort()) } + #[test] + fn omitted_selection_policy_remains_omitted_from_overlay() { + let network = test_network_score_first("net"); + let provider = test_provider("provider", "net", &["model"]); + let overlay = render_routing_overlay( + &network, + &[], + &[provider], + &[], + "site-a", + None, + None, + &scoring::ScoringWeights::default(), + ) + .unwrap_or_else(|_| std::process::abort()); + assert!( + overlay.selection_policy.is_none(), + "an upgraded GridNetwork that omits selectionPolicy must not be migrated implicitly" + ); + } + fn test_provider_with_selector( name: &str, network: &str, @@ -2308,9 +2571,9 @@ mod tests { // provider is also unavailable, but its fresh=false signals that its // metrics are stale. // - // Default GeographyFirst sort: admission → locality → score → fresh → tiebreak. - // The degraded local (SameSite, fresh=false) outranks the fresh API - // (CrossRegion, fresh=true) because geography sorts above freshness. + // Default GeographyFirst sort: admission → locality → fresh → score → tiebreak. + // Freshness is evaluated before score after locality, so a fresh remote + // candidate can precede a degraded local candidate. let network = test_network("fallback-net"); let local_degraded = test_provider_with_backend_kind_and_phase("provider-local", "fallback-net", "local", "Degraded"); @@ -2346,8 +2609,8 @@ mod tests { assert!(api_c.fresh, "API provider with absent status must have fresh=true"); assert_eq!( overlay.candidates.first().map(|c| c.cluster.as_str()), - Some("provider-local"), - "Degraded local must rank before API (GeographyFirst: locality outranks freshness)" + Some("provider-api"), + "with equal unknown locality, GeographyFirst freshness precedes score" ); } diff --git a/overlay-sync/src/types.rs b/overlay-sync/src/types.rs index 3e7b459..57fb06f 100644 --- a/overlay-sync/src/types.rs +++ b/overlay-sync/src/types.rs @@ -116,11 +116,35 @@ pub(crate) struct RoutingOverlay { /// Routing candidates ordered by the operator's sort. pub(crate) candidates: Vec, + /// Optional local request-selection policy from the Grid operator. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) selection_policy: Option, + /// RFC 3339 timestamp of when this overlay was rendered. #[serde(skip_serializing_if = "Option::is_none")] pub(crate) generated_at: Option, } +/// Selection policy copied without interpretation by overlay-sync. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct SelectionPolicy { + /// Selection mode consumed by Praxis. + pub(crate) mode: SelectionMode, +} + +/// Wire-level mode validation. Overlay-sync does not choose or execute it. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) enum SelectionMode { + /// Ordered selection. + Deterministic, + /// Equal local rotation. + RoundRobin, + /// Local random selection. + Random, +} + /// A single routing candidate. #[derive(Clone, Debug, Deserialize, Serialize)] pub(crate) struct RoutingCandidate { @@ -166,11 +190,16 @@ pub(crate) struct RoutingCandidate { /// Zero-based position in the final sorted overlay. #[serde(skip_serializing_if = "Option::is_none")] pub(crate) rank: Option, + + /// Producer-assigned active selection group. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) selection_group: Option, } /// Credential reference projected alongside a routing candidate. #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] +#[serde(deny_unknown_fields)] pub(crate) struct ProjectedCredential { /// Authentication strategy. pub(crate) strategy: String, @@ -182,6 +211,7 @@ pub(crate) struct ProjectedCredential { /// A reference to a Kubernetes Secret holding a credential value. #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] +#[serde(deny_unknown_fields)] pub(crate) struct ProjectedCredentialRef { /// Secret name. pub(crate) name: String, diff --git a/overlay-sync/src/validation.rs b/overlay-sync/src/validation.rs index 808cbad..072e7c5 100644 --- a/overlay-sync/src/validation.rs +++ b/overlay-sync/src/validation.rs @@ -8,10 +8,13 @@ use std::{fmt, fmt::Write as _}; +#[cfg(test)] use serde::ser::Error as _; use sha2::{Digest as _, Sha256}; -use crate::types::{OverlayEnvelope, OverlayScope, RoutingOverlay}; +#[cfg(test)] +use crate::types::RoutingOverlay; +use crate::types::{OverlayEnvelope, OverlayScope}; // --------------------------------------------------------------------------- // Constants @@ -126,15 +129,12 @@ pub(crate) fn validate_envelope( }); } - let envelope: OverlayEnvelope = serde_json::from_slice(raw).map_err(|e| ValidationError { - reason: RejectionReason::Malformed, - detail: format!("JSON parse failed: {e}"), - })?; + let (raw_value, envelope) = parse_envelope(raw)?; validate_schema_version(&envelope.schema_version)?; validate_scope(&envelope.scope, expected_scope)?; validate_revision_digest_agreement(&envelope)?; - validate_content_digest(&envelope)?; + validate_content_digest(&raw_value, &envelope)?; let revision = envelope.revision.value; @@ -153,6 +153,20 @@ pub(crate) fn validate_envelope( }) } +/// Parse the raw envelope while converting both parser failures into the +/// bounded validation error type. +fn parse_envelope(raw: &[u8]) -> Result<(serde_json::Value, OverlayEnvelope), ValidationError> { + let raw_value: serde_json::Value = serde_json::from_slice(raw).map_err(|e| ValidationError { + reason: RejectionReason::Malformed, + detail: format!("JSON parse failed: {e}"), + })?; + let envelope: OverlayEnvelope = serde_json::from_value(raw_value.clone()).map_err(|e| ValidationError { + reason: RejectionReason::Malformed, + detail: format!("envelope structure invalid: {e}"), + })?; + Ok((raw_value, envelope)) +} + /// Check that the schema version is supported. fn validate_schema_version(version: &str) -> Result<(), ValidationError> { if version.starts_with(SUPPORTED_MAJOR) { @@ -224,8 +238,15 @@ fn validate_revision_digest_agreement(envelope: &OverlayEnvelope) -> Result<(), } /// Recompute the semantic digest and verify it matches the envelope. -fn validate_content_digest(envelope: &OverlayEnvelope) -> Result<(), ValidationError> { - let recomputed = compute_semantic_digest(&envelope.overlay).map_err(|e| ValidationError { +fn validate_content_digest( + raw_envelope: &serde_json::Value, + envelope: &OverlayEnvelope, +) -> Result<(), ValidationError> { + let raw_overlay = raw_envelope.get("overlay").ok_or_else(|| ValidationError { + reason: RejectionReason::Malformed, + detail: "overlay payload is missing".to_owned(), + })?; + let recomputed = compute_raw_semantic_digest(raw_overlay).map_err(|e| ValidationError { reason: RejectionReason::Malformed, detail: format!("cannot canonicalize overlay: {e}"), })?; @@ -239,21 +260,64 @@ fn validate_content_digest(envelope: &OverlayEnvelope) -> Result<(), ValidationE Ok(()) } +/// Compute the semantic digest from the raw overlay JSON. +/// +/// The operator's semantic payload includes the complete candidate objects. +/// Computing from raw JSON keeps additive routing metadata digest-significant +/// even when this sidecar does not understand the new field yet. +fn compute_raw_semantic_digest(overlay: &serde_json::Value) -> Result { + let object = overlay + .as_object() + .ok_or_else(|| "overlay must be a JSON object".to_owned())?; + let network = object + .get("network") + .ok_or_else(|| "overlay.network is missing".to_owned())?; + let local_site = object + .get("local_site") + .ok_or_else(|| "overlay.local_site is missing".to_owned())?; + let candidates = object + .get("candidates") + .ok_or_else(|| "overlay.candidates is missing".to_owned())?; + let mut semantic_payload = serde_json::json!({ + "candidates": candidates, + "local_site": local_site, + "network": network, + }); + if let Some(selection_policy) = object.get("selection_policy") { + semantic_payload + .as_object_mut() + .ok_or_else(|| "semantic payload is not an object".to_owned())? + .insert("selection_policy".to_owned(), selection_policy.clone()); + } + let canonical = serde_json_canonicalizer::to_vec(&semantic_payload) + .map_err(|e| format!("RFC 8785 canonicalization failed: {e}"))?; + let digest: [u8; 32] = Sha256::digest(&canonical).into(); + Ok(hex_encode(&digest)) +} + // --------------------------------------------------------------------------- // Digest computation (same algorithm as the operator) // --------------------------------------------------------------------------- /// Compute the SHA-256 digest of the RFC 8785 canonical semantic payload. /// -/// The semantic payload includes only `network`, `local_site`, and -/// `candidates` — the same fields used by the operator. +/// The semantic payload includes `network`, `local_site`, and `candidates`, +/// plus `selection_policy` when present — the same fields used by the +/// operator. +#[cfg(test)] fn compute_semantic_digest(overlay: &RoutingOverlay) -> Result { let candidates_value = serde_json::to_value(&overlay.candidates)?; - let semantic_payload = serde_json::json!({ + let mut semantic_payload = serde_json::json!({ "candidates": candidates_value, "local_site": overlay.local_site, "network": overlay.network, }); + if let Some(policy) = &overlay.selection_policy { + let Some(object) = semantic_payload.as_object_mut() else { + return Err(serde_json::Error::custom("semantic payload must be an object")); + }; + object.insert("selection_policy".to_owned(), serde_json::to_value(policy)?); + } let canonical = serde_json_canonicalizer::to_vec(&semantic_payload).map_err(serde_json::Error::custom)?; @@ -280,7 +344,7 @@ fn hex_encode(bytes: &[u8]) -> String { #[allow(clippy::unwrap_used, clippy::expect_used, clippy::indexing_slicing, reason = "tests")] mod tests { use super::*; - use crate::types::{ContentDigest, ContentRevision, OverlayProvenance, RoutingCandidate}; + use crate::types::{ContentDigest, ContentRevision, OverlayProvenance, RoutingCandidate, RoutingOverlay}; fn test_scope() -> ExpectedScope { ExpectedScope { @@ -308,7 +372,9 @@ mod tests { score: None, score_breakdown: None, rank: Some(0), + selection_group: None, }], + selection_policy: None, generated_at: Some("2026-07-29T00:00:00Z".to_owned()), } } @@ -354,6 +420,50 @@ mod tests { assert!(result.is_ok()); } + #[test] + fn additive_candidate_field_is_digest_significant_and_accepted() { + let env = valid_envelope(); + let mut value = serde_json::to_value(env).unwrap(); + value["overlay"]["candidates"][0]["future_field"] = serde_json::json!({"mode": "x"}); + let digest = compute_raw_semantic_digest(&value["overlay"]).unwrap(); + value["revision"]["value"] = serde_json::Value::String(digest.clone()); + value["content_digest"]["value"] = serde_json::Value::String(digest); + + let raw = serde_json::to_vec(&value).unwrap(); + let result = validate_envelope(&raw, &test_scope(), 1_048_576, None); + assert!(result.is_ok(), "additive candidate metadata should survive validation"); + } + + #[test] + fn additive_candidate_field_mutation_without_digest_update_is_rejected() { + let env = valid_envelope(); + let mut value = serde_json::to_value(env).unwrap(); + value["overlay"]["candidates"][0]["future_field"] = serde_json::json!({"mode": "x"}); + + let raw = serde_json::to_vec(&value).unwrap(); + let result = validate_envelope(&raw, &test_scope(), 1_048_576, None); + assert!(matches!(result.unwrap_err().reason, RejectionReason::DigestMismatch)); + } + + #[test] + fn credential_value_fields_remain_rejected() { + let env = valid_envelope(); + let mut value = serde_json::to_value(env).unwrap(); + value["overlay"]["candidates"][0]["credential"] = serde_json::json!({ + "strategy": "bearer_token", + "secretRef": { + "name": "credential", + "namespace": "ns", + "key": "token" + }, + "token": "must-not-enter-overlay" + }); + + let raw = serde_json::to_vec(&value).unwrap(); + let result = validate_envelope(&raw, &test_scope(), 1_048_576, None); + assert!(matches!(result.unwrap_err().reason, RejectionReason::Malformed)); + } + #[test] fn oversized_payload_rejected() { let env = valid_envelope(); @@ -437,6 +547,37 @@ mod tests { ); } + #[test] + fn selection_policy_fixture_digest_matches_operator_contract() { + let fixture = include_str!("../../tests/fixtures/overlay-contract/v1/valid-selection-policy.json"); + let envelope: OverlayEnvelope = serde_json::from_str(fixture).unwrap(); + let recomputed = compute_semantic_digest(&envelope.overlay).unwrap(); + assert_eq!( + recomputed, envelope.revision.value, + "selection policy fixture must match the producer digest; computed={recomputed}" + ); + let scope = test_scope(); + validate_envelope(fixture.as_bytes(), &scope, 1_048_576, None) + .expect("selection policy fixture must pass transparent validation"); + } + + #[test] + fn unknown_selection_policy_field_is_rejected() { + let env = valid_envelope(); + let mut value = serde_json::to_value(env).unwrap(); + value["overlay"]["selection_policy"] = serde_json::json!({ + "mode": "roundRobin", + "unexpected": true + }); + let digest = compute_raw_semantic_digest(&value["overlay"]).unwrap(); + value["revision"]["value"] = serde_json::Value::String(digest.clone()); + value["content_digest"]["value"] = serde_json::Value::String(digest); + + let raw = serde_json::to_vec(&value).unwrap(); + let result = validate_envelope(&raw, &test_scope(), 1_048_576, None); + assert!(matches!(result.unwrap_err().reason, RejectionReason::Malformed)); + } + #[test] fn hex_encode_lowercase() { let bytes = [0xAB, 0xCD, 0x01, 0x23]; diff --git a/overlay-sync/src/watcher.rs b/overlay-sync/src/watcher.rs index 561d56a..62581e1 100644 --- a/overlay-sync/src/watcher.rs +++ b/overlay-sync/src/watcher.rs @@ -482,22 +482,47 @@ mod tests { score: None, score_breakdown: None, rank: Some(0), + selection_group: None, }], + selection_policy: None, generated_at: Some("2026-07-29T00:00:00Z".to_owned()), } } fn overlay_digest(overlay: &crate::types::RoutingOverlay) -> String { - let canonical = serde_json::json!({ + let mut semantic_payload = serde_json::json!({ "candidates": serde_json::to_value(&overlay.candidates).unwrap(), "local_site": overlay.local_site, "network": overlay.network, }); - let canonical_bytes = serde_json_canonicalizer::to_vec(&canonical).unwrap(); + if let Some(policy) = &overlay.selection_policy { + semantic_payload + .as_object_mut() + .unwrap() + .insert("selection_policy".to_owned(), serde_json::to_value(policy).unwrap()); + } + let canonical_bytes = serde_json_canonicalizer::to_vec(&semantic_payload).unwrap(); let digest: [u8; 32] = sha2::Sha256::new().chain_update(&canonical_bytes).finalize().into(); digest.iter().map(|b| format!("{b:02x}")).collect() } + #[test] + fn overlay_digest_includes_selection_policy() { + let scope = ExpectedScope { + network: "test-net".to_owned(), + gateway: "gw".to_owned(), + namespace: "ns".to_owned(), + local_site: "site-a".to_owned(), + }; + let without_policy = test_overlay(&scope); + let mut with_policy = without_policy.clone(); + with_policy.selection_policy = Some(crate::types::SelectionPolicy { + mode: crate::types::SelectionMode::RoundRobin, + }); + + assert_ne!(overlay_digest(&without_policy), overlay_digest(&with_policy)); + } + fn test_provenance() -> crate::types::OverlayProvenance { crate::types::OverlayProvenance { producer: "grid-operator".to_owned(), diff --git a/tests/e2e/topologies/grid-combined-site/forge.yaml b/tests/e2e/topologies/grid-combined-site/forge.yaml index 7686b30..a3bc1cb 100644 --- a/tests/e2e/topologies/grid-combined-site/forge.yaml +++ b/tests/e2e/topologies/grid-combined-site/forge.yaml @@ -210,6 +210,11 @@ spec: gridId: grid-combined-site-v1 region: west zone: west-1 + routingPolicy: scoreFirst + scoringPolicy: + strategy: noMetrics + selectionPolicy: + mode: roundRobin swim: probeInterval: "5s" suspicionTimeout: "15s" @@ -268,6 +273,11 @@ spec: gridId: grid-combined-site-v1 region: central zone: central-1 + routingPolicy: scoreFirst + scoringPolicy: + strategy: noMetrics + selectionPolicy: + mode: roundRobin swim: probeInterval: "5s" suspicionTimeout: "15s" @@ -326,6 +336,11 @@ spec: gridId: grid-combined-site-v1 region: east zone: east-1 + routingPolicy: scoreFirst + scoringPolicy: + strategy: noMetrics + selectionPolicy: + mode: roundRobin swim: probeInterval: "5s" suspicionTimeout: "15s" diff --git a/tests/e2e/topologies/grid-combined-site/resources/common/vcr-provider-workload.yaml b/tests/e2e/topologies/grid-combined-site/resources/common/vcr-provider-workload.yaml index debf2d2..1a68c7f 100644 --- a/tests/e2e/topologies/grid-combined-site/resources/common/vcr-provider-workload.yaml +++ b/tests/e2e/topologies/grid-combined-site/resources/common/vcr-provider-workload.yaml @@ -36,6 +36,8 @@ spec: - ALL readOnlyRootFilesystem: false runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 env: - name: MODEL value: "Qwen/Qwen3-0.6B" diff --git a/tests/e2e/topologies/grid-provider-traffic/configs/provider/praxis.yaml b/tests/e2e/topologies/grid-provider-traffic/configs/provider/praxis.yaml new file mode 100644 index 0000000..aa41394 --- /dev/null +++ b/tests/e2e/topologies/grid-provider-traffic/configs/provider/praxis.yaml @@ -0,0 +1,60 @@ +# Provider gateway fixture used by the provider-traffic xtask tests. +listeners: + - name: provider + address: "0.0.0.0:8443" + filter_chains: + - provider-inference + tls: + certificates: + - cert_path: /etc/praxis/tls/tls.crt + key_path: /etc/praxis/tls/tls.key + client_ca: + ca_path: /etc/praxis/tls/ca.crt + client_cert_mode: require + +filter_chains: + - name: provider-inference + filters: + - filter: peer_identity_trust + trusted_peers: + - organization: ai-grid + - filter: json_body_field + field: model + header: X-Model + - filter: headers + response_set: + - name: X-Grid-Provider-Traffic-Provider-Gateway + value: "SITE_PLACEHOLDER" + - filter: provider_route + provider_id: "SITE_PLACEHOLDER" + model_header: X-Model + emit_demo_attribution: true + routes: + - candidate_id: "CANDIDATE_ID_PLACEHOLDER" + model: Qwen/Qwen3-0.6B + paths: + - /v1/chat/completions + - /v1/responses + cluster: vcr-backend + credential: + strategy: bearer_token + secretRef: + name: vcr-inference-credential + namespace: grid-system + key: token + - filter: credential_inject + credentials: + - strategy: bearer_token + name: vcr-inference-credential + namespace: grid-system + key: token + file: /etc/praxis/credentials/vcr-inference/token + - filter: load_balancer + clusters: + - name: vcr-backend + endpoints: + - "vcr-inference-SITE_PLACEHOLDER.grid-system.svc.cluster.local:8000" + +admin: + address: "127.0.0.1:9901" +shutdown_timeout_secs: 5 diff --git a/tests/fixtures/overlay-contract/v1/valid-selection-policy.json b/tests/fixtures/overlay-contract/v1/valid-selection-policy.json new file mode 100644 index 0000000..1bd8f46 --- /dev/null +++ b/tests/fixtures/overlay-contract/v1/valid-selection-policy.json @@ -0,0 +1,56 @@ +{ + "schema_version": "1.0.0", + "revision": { + "kind": "content_addressed", + "algorithm": "sha256", + "value": "d3321586b5e414b8bd80c07a9d16ac10dce38d6810e3591a8526a6bdd0639007" + }, + "content_digest": { + "algorithm": "sha256", + "value": "d3321586b5e414b8bd80c07a9d16ac10dce38d6810e3591a8526a6bdd0639007" + }, + "scope": { + "network": "test-net", + "gateway": "gw", + "namespace": "ns", + "local_site": "site-a" + }, + "provenance": { + "producer": "grid-operator", + "producer_version": "0.1.3", + "source_name": "test-net", + "source_uid": "uid-selection-policy", + "source_generation": 1, + "rendered_at": "2026-08-12T00:00:00Z" + }, + "overlay": { + "network": "test-net", + "local_site": "site-a", + "selection_policy": { + "mode": "roundRobin" + }, + "candidates": [ + { + "kind": "inference_model", + "name": "model-a", + "site": "site-a", + "cluster": "cluster-a", + "fresh": true, + "stable_id": "abcd1234", + "rank": 0, + "selection_group": 0 + }, + { + "kind": "inference_model", + "name": "model-b", + "site": "site-b", + "cluster": "cluster-b", + "fresh": true, + "stable_id": "efgh5678", + "rank": 1, + "selection_group": 0 + } + ], + "generated_at": "2026-08-12T00:00:00Z" + } +} diff --git a/xtask/src/env/combined_site_demo.rs b/xtask/src/env/combined_site_demo.rs index db95ea0..570ef2c 100644 --- a/xtask/src/env/combined_site_demo.rs +++ b/xtask/src/env/combined_site_demo.rs @@ -3407,34 +3407,6 @@ fn require_local_image(image: &str) -> Result<(), Box> { .into()) } -/// Verify that `image` has a numeric USER (or UID:GID) so Kubernetes can -/// enforce `runAsNonRoot` without `runAsUser` in the pod spec. -fn require_numeric_image_user(image: &str) -> Result<(), Box> { - let output = Command::new("docker") - .args(["inspect", "--format", "{{.Config.User}}", image]) - .output()?; - if !output.status.success() { - return Err(format!( - "cannot inspect image {image:?}: {}", - String::from_utf8_lossy(&output.stderr).trim() - ) - .into()); - } - let user = String::from_utf8_lossy(&output.stdout).trim().to_owned(); - if user.is_empty() { - return Err(format!("image {image:?} has no USER set; runAsNonRoot requires a numeric user").into()); - } - let valid = user.split(':').all(|part| part.parse::().is_ok()); - if !valid { - return Err(format!( - "image {image:?} has non-numeric USER {user:?}; \ - Kubernetes cannot verify runAsNonRoot with a non-numeric user" - ) - .into()); - } - Ok(()) -} - /// Load local container images into all Kind clusters. /// /// Reads image references from the `GRID_XTASK_*_IMAGE` environment variables @@ -3458,9 +3430,6 @@ fn load_images_into_clusters(forge_bin: &Path, resolved_config: &Path) -> Result eprintln!(" verified local image: {image}"); } - require_numeric_image_user(&vcr)?; - eprintln!(" verified numeric USER for {vcr}"); - for cluster in CLUSTERS { for image in [&gateway, &operator, &vcr] { eprintln!(" loading {image} into {cluster}..."); diff --git a/xtask/src/env/image_overrides.rs b/xtask/src/env/image_overrides.rs index 21944a5..a8abb1a 100644 --- a/xtask/src/env/image_overrides.rs +++ b/xtask/src/env/image_overrides.rs @@ -29,6 +29,9 @@ const VCR_IMAGE_ENV: &str = "GRID_XTASK_VCR_IMAGE"; /// Environment variable to override the operator image. const OPERATOR_IMAGE_ENV: &str = "GRID_XTASK_OPERATOR_IMAGE"; +/// Environment variable to override the overlay-sync image. +const OVERLAY_SYNC_IMAGE_ENV: &str = "GRID_XTASK_OVERLAY_SYNC_IMAGE"; + /// Environment variable to override the image pull policy. const IMAGE_PULL_POLICY_ENV: &str = "GRID_XTASK_IMAGE_PULL_POLICY"; @@ -48,6 +51,9 @@ const DEFAULT_MOCK_PROVIDER_IMAGE: &str = "grid-mock-providers:latest"; /// Default operator image (matches operator.rs). const DEFAULT_OPERATOR_IMAGE: &str = "grid-operator:latest"; +/// Default overlay-sync image used by overlay-enabled demo gateways. +const DEFAULT_OVERLAY_SYNC_IMAGE: &str = "grid-overlay-sync:latest"; + /// Default gateway image used by the GLB demo. const DEFAULT_GLB_GATEWAY_IMAGE: &str = "ghcr.io/praxis-proxy/grid-ai-rollup:v0.1.3"; @@ -96,6 +102,11 @@ pub(crate) fn operator_image() -> String { env::var(OPERATOR_IMAGE_ENV).unwrap_or_else(|_| DEFAULT_OPERATOR_IMAGE.to_owned()) } +/// Get the overlay-sync image name, respecting environment overrides. +pub(crate) fn overlay_sync_image() -> String { + env::var(OVERLAY_SYNC_IMAGE_ENV).unwrap_or_else(|_| DEFAULT_OVERLAY_SYNC_IMAGE.to_owned()) +} + /// Get the VCR image name, respecting environment overrides. pub(crate) fn vcr_image() -> String { env::var(VCR_IMAGE_ENV).unwrap_or_else(|_| DEFAULT_VCR_IMAGE.to_owned()) diff --git a/xtask/src/env/mod.rs b/xtask/src/env/mod.rs index c3e0f18..96db010 100644 --- a/xtask/src/env/mod.rs +++ b/xtask/src/env/mod.rs @@ -16,6 +16,7 @@ pub(crate) mod kubectl; pub(crate) mod llmd_pool_metrics_demo; pub(crate) mod operator; pub(crate) mod operator_overlay; +pub(crate) mod provider_traffic_demo; pub(crate) mod providers; pub(crate) mod trust; pub(crate) mod verify; @@ -974,6 +975,19 @@ pub(crate) enum Action { #[arg(long)] kv_cache: bool, }, + + /// Create the focused provider-gateway traffic demo, then prove equal + /// selection across its active provider group. + RunGridProviderTrafficDemo { + /// Path to the public or internal Forge environment config file. + /// This is required because the focused demo currently lives in the + /// public demos repository rather than the Grid source tree. + #[arg(long)] + forge_config: PathBuf, + /// Demo mode and teardown options. Only `--quick` is supported. + #[command(flatten)] + options: GlbDemoOptions, + }, } #[cfg(test)] @@ -1092,6 +1106,9 @@ pub(crate) fn run(action: &Action) -> Result<(), Box> { metrics_mtls, kv_cache, } => llmd_pool_metrics_demo::run(forge_config, options, *metrics_mtls, *kv_cache), + Action::RunGridProviderTrafficDemo { forge_config, options } => { + provider_traffic_demo::run(forge_config, options) + }, } } diff --git a/xtask/src/env/provider_traffic_demo.rs b/xtask/src/env/provider_traffic_demo.rs new file mode 100644 index 0000000..7bd6057 --- /dev/null +++ b/xtask/src/env/provider_traffic_demo.rs @@ -0,0 +1,3047 @@ +//! Narrated, evidence-backed provider-traffic demo scenarios. +use std::{ + collections::{BTreeMap, BTreeSet}, + fs, + path::{Path, PathBuf}, + process::Command, + sync::atomic::{AtomicU64, Ordering}, + time::{Duration, Instant}, +}; + +use serde::Serialize; + +use super::{DemoMode, GlbDemoOptions, certs, glb, kubectl, operator, safe_truncate_str}; + +// ----------------------------------------------------------------------------- +// Constants +// ----------------------------------------------------------------------------- + +/// Directory where generated TLS certificates are stored. +const CERTS_DIR: &str = "tests/env/certs"; + +/// Ordered provider-site cluster names in the provider-traffic scenario. +/// +/// The consumer entrypoint is deployed only in `CONSUMER_SITE`; the other +/// clusters contain provider gateways and backends only. +const CLUSTERS: &[&str] = &["provider-a", "provider-b", "provider-c"]; + +/// The single consumer gateway used by the focused request-routing proof. +const CONSUMER_SITE: &str = "provider-a"; + +/// Consumer gateway TLS secret name (matches Helm `existingSecret` reference). +const CONSUMER_TLS_SECRET: &str = "consumer-gateway-tls"; + +/// Evidence JSON schema version. +const EVIDENCE_SCHEMA_VERSION: &str = "1"; + +/// Kubernetes namespace for all Grid components. +const GRID_SYSTEM_NS: &str = "grid-system"; + +/// Overlay `ConfigMap` name created by the Grid operator for consumer gateways. +const OVERLAY_CONFIGMAP: &str = "grid-overlay-grid-provider-traffic-consumer-gateway"; + +/// Provider credential secret name (matches Helm `credentials[0].name`). +const VCR_INFERENCE_CREDENTIAL: &str = "vcr-inference-credential"; + +/// Stable terminal separator that also remains readable in captured logs. +const OUTPUT_RULE: &str = "==============================================================================="; + +/// Provider gateway service name advertised via SWIM for cross-site discovery. +const PROVIDER_GATEWAY_SERVICE: &str = "provider-gateway"; + +/// Provider gateway port advertised via SWIM for cross-site discovery. +const PROVIDER_GATEWAY_PORT: &str = "8443"; + +/// Provider gateway TLS secret name (matches Helm `existingSecret` reference). +const PROVIDER_TLS_SECRET: &str = "provider-gateway-tls"; + +/// Same-CA client identity with an organization rejected by `peer_identity_trust`. +const WRONG_ORG_TLS_SECRET: &str = "wrong-org-client-tls"; + +/// Number of environment setup phases shown to the user. +const SETUP_PHASES: usize = 14; + +/// Makes retry probe names unique while retaining a recognizable prefix. +static PROBE_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +// ----------------------------------------------------------------------------- +// Context +// ----------------------------------------------------------------------------- + +/// Provider-traffic demo execution context. +struct ProviderTrafficContext { + /// Canonical demo root directory for resolving configs, resources, and + /// other demo-relative assets. + demo_root: PathBuf, + /// Path to the resolved Forge config. + resolved_config: PathBuf, + /// Path to the forge binary. + forge_bin: PathBuf, +} + +// ----------------------------------------------------------------------------- +// Overlay State +// ----------------------------------------------------------------------------- + +/// Per-site overlay snapshot captured from the `ConfigMap`. +#[derive(Clone, Debug, serde::Deserialize, Serialize)] +struct OverlayData { + /// Kubernetes `ConfigMap` `resourceVersion` (per-cluster, never compared + /// across clusters). + resource_version: String, + /// Content-addressed semantic revision from the + /// `grid.praxis-proxy.io/overlay-revision` annotation. Changes only when + /// routing-relevant fields change; safe to compare across clusters. + semantic_revision: String, + /// Candidate name to stable ID mapping. + stable_ids: BTreeMap, + /// Full candidate details for evidence (kind, name, site, cluster, model). + candidates: Vec, +} + +/// Subset of `RoutingCandidate` fields kept for evidence and validation. +#[derive(Clone, Debug, serde::Deserialize, Serialize)] +struct OverlayCandidate { + /// Candidate kind (e.g. `inference_model`). + kind: String, + /// Model name. + name: String, + /// Site name. + site: String, + /// Upstream cluster identifier (used as the candidate key). + cluster: String, + /// Deterministic stable ID for session binding. + stable_id: String, + /// Whether the candidate's metrics snapshot is fresh. + fresh: Option, + /// Admission state emitted by the operator, when present. + admission_state: Option, + /// Explicit priority group emitted by the operator, when present. + selection_group: Option, +} + +/// Pre- and post-SWIM overlay snapshots for evidence. +#[derive(Clone, Debug, Default, serde::Deserialize, Serialize)] +struct OverlayState { + /// Per-site overlays before SWIM seeding (local candidates only). + pre_swim: BTreeMap, + /// Per-site overlays after global convergence. + post_swim: BTreeMap, +} + +// ----------------------------------------------------------------------------- +// Evidence +// ----------------------------------------------------------------------------- + +/// Evidence written to `results.json`. +#[derive(Debug, serde::Deserialize, Serialize)] +struct Evidence { + /// Evidence schema version. + schema_version: String, + /// Demo mode that was executed. + mode: String, + /// Topology name. + topology: String, + /// List of cluster names. + clusters: Vec, + /// Proof results for each assertion. + proof_results: BTreeMap, + /// Exact image references used. + images: BTreeMap, + /// Pre- and post-SWIM overlay snapshots. + overlay_state: OverlayState, + /// Cluster health status. + cluster_health: Vec, + /// Component deployment status. + components: Vec, + /// SWIM membership views. + swim_membership: Vec, + /// Provider response samples. + provider_responses: Vec, + /// Security assertion results. + security_results: Vec, + /// Teardown success. + teardown_success: bool, +} + +/// Evidence for one proof assertion. +#[derive(Debug, serde::Deserialize, Serialize)] +struct ProofResult { + /// Whether the proof passed. + success: bool, + /// Human-readable reason. + reason: String, + /// Observed facts that support this result. + observed_facts: BTreeMap, + /// Duration of the assertion in milliseconds. + duration_ms: u64, +} + +/// Cluster health status. +#[derive(Debug, serde::Deserialize, Serialize)] +struct ClusterHealth { + /// Cluster name. + name: String, + /// Whether the cluster is healthy. + healthy: bool, + /// API server response time in milliseconds. + api_response_ms: Option, + /// Number of ready nodes. + ready_nodes: u32, +} + +/// Component deployment status. +#[derive(Debug, serde::Deserialize, Serialize)] +struct ComponentStatus { + /// Component name. + name: String, + /// Deployment namespace. + namespace: String, + /// Ready replicas. + ready_replicas: u32, + /// Desired replicas. + desired_replicas: u32, + /// Whether the component is ready. + ready: bool, +} + +/// SWIM membership view. +#[derive(Debug, serde::Deserialize, Serialize)] +struct SwimMembership { + /// Site name. + site: String, + /// Local node ID. + local_node: String, + /// List of known peers. + peers: Vec, + /// Membership convergence status. + converged: bool, +} + +/// Provider response metadata. +#[derive(Debug, serde::Deserialize, Serialize)] +struct ProviderResponse { + /// Consumer site that made the request. + consumer_site: String, + /// Provider site that served the request. + provider_site: String, + /// Provider instance ID. + provider_instance: String, + /// Session ID. + session_id: String, + /// Serving revision. + serving_revision: String, + /// Response time in milliseconds. + response_time_ms: u64, + /// Whether the response was successful. + success: bool, +} + +/// Security assertion result. +#[derive(Debug, serde::Deserialize, Serialize)] +struct SecurityResult { + /// Type of security test. + test_type: String, + /// Expected result (allow/deny). + expected: String, + /// Actual result. + actual: String, + /// Whether the test passed. + passed: bool, + /// Additional context. + context: String, +} + +// ----------------------------------------------------------------------------- +// Utility Functions +// ----------------------------------------------------------------------------- + +/// Format current UTC timestamp for run IDs. +fn format_utc_timestamp() -> String { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or_else(|_| "unknown".to_owned(), |duration| format!("{}", duration.as_secs())) +} + +/// Format current UTC timestamp in ISO format. +fn format_utc_iso() -> String { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now().duration_since(UNIX_EPOCH).map_or_else( + |_| "unknown-utc".to_owned(), + |duration| format!("{}-utc", duration.as_secs()), + ) +} + +/// Resolve the evidence directory path. +fn resolve_evidence_dir( + forge_config: &Path, + options: &GlbDemoOptions, + run_id: &str, +) -> Result> { + if let Some(dir) = &options.evidence_dir { + Ok(dir.clone()) + } else { + let config_dir = forge_config + .parent() + .ok_or("forge config should have parent directory")?; + Ok(config_dir.join(format!("evidence-{run_id}"))) + } +} + +// ----------------------------------------------------------------------------- +// Assertion Framework +// ----------------------------------------------------------------------------- + +/// Result of a runtime assertion. +type AssertionResult = Result>; + +/// Create a successful proof result with observed facts. +fn proof_success(reason: &str, observed_facts: BTreeMap, duration: Duration) -> ProofResult { + ProofResult { + success: true, + reason: reason.to_owned(), + observed_facts, + duration_ms: u64::try_from(duration.as_millis().min(u128::from(u64::MAX))).unwrap_or(u64::MAX), + } +} + +/// Create a failed proof result with observed facts. +fn proof_failure(reason: &str, observed_facts: BTreeMap, duration: Duration) -> ProofResult { + ProofResult { + success: false, + reason: reason.to_owned(), + observed_facts, + duration_ms: u64::try_from(duration.as_millis().min(u128::from(u64::MAX))).unwrap_or(u64::MAX), + } +} + +/// Execute an assertion with timing and error handling. +fn run_assertion(name: &str, assertion_fn: F) -> AssertionResult +where + F: FnOnce() -> AssertionResult, +{ + let start = Instant::now(); + eprintln!(" [ASSERT] {name}"); + + let result = assertion_fn(); + let _duration = start.elapsed(); + + match &result { + Ok(proof) => { + if proof.success { + eprintln!(" [OK] {name}: {}", proof.reason); + } else { + eprintln!(" [FAIL] {name}: {}", proof.reason); + } + }, + Err(e) => { + eprintln!(" [ERROR] {name}: {e}"); + }, + } + + result.map_err(|e| { + // Convert assertion errors to proof failures + format!("Assertion {name} failed: {e}").into() + }) +} + +/// Poll for a condition with bounded retries. +fn poll_until(condition: F, timeout: Duration, interval: Duration) -> Result> +where + F: Fn() -> Result, Box>, +{ + let start = Instant::now(); + + loop { + match condition() { + Ok(Some(result)) => return Ok(result), + Ok(None) => { + if start.elapsed() >= timeout { + return Err("Timeout waiting for condition".into()); + } + std::thread::park_timeout(interval); + }, + Err(e) => return Err(e), + } + } +} + +/// Wait for a deployment to be ready. +fn wait_for_deployment(deployment: &str, namespace: &str, context: &str) -> Result<(), Box> { + kubectl::wait_for_rollout_ns(context, deployment, namespace, "deployment")?; + Ok(()) +} + +/// Pod-security overrides for ephemeral curl pods in restricted namespaces. +/// +/// `kubectl run` names the container after the pod, so the container name must +/// match. The curlimages/curl image runs as UID 100. +fn curl_pod_overrides(pod_name: &str, curl_args: &[&str]) -> String { + let args = curl_args.strip_prefix(&["curl"]).unwrap_or(curl_args); + serde_json::json!({ + "spec": { + "automountServiceAccountToken": false, + "securityContext": { + "runAsNonRoot": true, + "seccompProfile": { "type": "RuntimeDefault" } + }, + "containers": [{ + "name": pod_name, + "image": "curlimages/curl:8.12.1", + "command": ["curl"], + "args": args, + "securityContext": { + "runAsUser": 100, + "allowPrivilegeEscalation": false, + "readOnlyRootFilesystem": true, + "capabilities": { "drop": ["ALL"] } + } + }] + } + }) + .to_string() +} + +/// Run an ephemeral curl pod with restricted `PodSecurity` context. +fn run_curl_probe(context: &str, pod_name: &str, curl_args: &[&str]) -> Result { + let sequence = PROBE_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let prefix = pod_name.get(..pod_name.len().min(40)).unwrap_or(pod_name); + let unique_pod_name = format!("{prefix}-{sequence}"); + let overrides = curl_pod_overrides(&unique_pod_name, curl_args); + Command::new("kubectl") + .args([ + "run", + &unique_pod_name, + "--image=curlimages/curl:8.12.1", + "--context", + context, + "-n", + GRID_SYSTEM_NS, + "--rm", + "-i", + "--restart=Never", + "--overrides", + &overrides, + ]) + .output() +} + +/// Run an ephemeral curl pod with additional kubectl flags (e.g. `--labels`). +fn response_header(output: &[u8], name: &str) -> Option { + let expected = name.to_ascii_lowercase(); + String::from_utf8_lossy(output).lines().find_map(|line| { + let (header, value) = line.split_once(':')?; + header.eq_ignore_ascii_case(&expected).then(|| value.trim().to_owned()) + }) +} + +/// Wait for the demo environment to be ready. +fn wait_for_environment_ready() -> Result> { + // Wait for Grid operators to converge + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + eprintln!(" [WAIT] {cluster}: Grid operator convergence"); + + // Wait for deployment to be ready + wait_for_deployment("grid-operator", "grid-system", &context)?; + + // Every site has a provider gateway; only the designated ingress site + // has the consumer gateway used by the request proof. + if *cluster == CONSUMER_SITE { + wait_for_deployment("consumer-gateway", "grid-system", &context)?; + } + wait_for_deployment("provider-gateway", "grid-system", &context)?; + + eprintln!(" [OK] {cluster}: Gateways ready"); + } + + Ok("Three provider sites converged; one consumer entrypoint and all provider gateways are ready".to_owned()) +} + +// ----------------------------------------------------------------------------- +// Runtime Assertions +// ----------------------------------------------------------------------------- + +/// Assert exactly three provider clusters exist and are healthy. +#[expect( + clippy::too_many_lines, + reason = "The proof reports one bounded fact set per cluster." +)] +fn assert_cluster_health() -> AssertionResult { + let start = Instant::now(); + let mut observed_facts = BTreeMap::new(); + let mut cluster_health = Vec::new(); + + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + + // Check API server responsiveness + let api_start = Instant::now(); + let output = Command::new("kubectl") + .args(["cluster-info", "--context", &context]) + .output()?; + + let api_response_ms = + u64::try_from(api_start.elapsed().as_millis().min(u128::from(u64::MAX))).unwrap_or(u64::MAX); + let healthy = output.status.success(); + + // Get node count + let nodes_output = Command::new("kubectl") + .args([ + "get", + "nodes", + "--context", + &context, + "-o", + "jsonpath={.items[*].status.conditions[?(@.type=='Ready')].status}", + ]) + .output()?; + + let ready_nodes = if nodes_output.status.success() { + u32::try_from( + String::from_utf8_lossy(&nodes_output.stdout) + .split_whitespace() + .filter(|s| s == &"True") + .count() + .min(u32::MAX as usize), + ) + .unwrap_or(u32::MAX) + } else { + 0 + }; + + cluster_health.push(ClusterHealth { + name: cluster.to_string(), + healthy, + api_response_ms: Some(api_response_ms), + ready_nodes, + }); + + observed_facts.insert(format!("{cluster}_healthy"), serde_json::Value::Bool(healthy)); + observed_facts.insert( + format!("{cluster}_ready_nodes"), + serde_json::Value::Number(ready_nodes.into()), + ); + } + + let all_healthy = cluster_health.iter().all(|c| c.healthy && c.ready_nodes > 0); + observed_facts.insert( + "total_clusters".to_owned(), + serde_json::Value::Number(CLUSTERS.len().into()), + ); + + if all_healthy { + Ok(proof_success( + &format!("All {} clusters are healthy with ready nodes", CLUSTERS.len()), + observed_facts, + start.elapsed(), + )) + } else { + Ok(proof_failure( + "One or more clusters are unhealthy", + observed_facts, + start.elapsed(), + )) + } +} + +/// Assert one provider stack per site and one consumer entrypoint. +#[expect( + clippy::too_many_lines, + reason = "The proof checks the required deployed components together." +)] +fn assert_component_deployment() -> AssertionResult { + let start = Instant::now(); + let mut observed_facts = BTreeMap::new(); + let mut components = Vec::new(); + + let static_components = ["grid-operator", "provider-gateway"]; + + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + let mock_name = format!("vcr-inference-{cluster}"); + let cluster_components: Vec<&str> = static_components + .iter() + .copied() + .chain(std::iter::once(mock_name.as_str())) + .collect(); + let cluster_components = if *cluster == CONSUMER_SITE { + cluster_components + .into_iter() + .chain(std::iter::once("consumer-gateway")) + .collect::>() + } else { + cluster_components + }; + + for component in &cluster_components { + let output = Command::new("kubectl") + .args([ + "get", + "deployment", + component, + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "-o", + "jsonpath={.status.readyReplicas},{.status.replicas}", + ]) + .output()?; + + if output.status.success() { + let status_str = String::from_utf8_lossy(&output.stdout); + let parts: Vec<&str> = status_str.trim().split(',').collect(); + + let ready_replicas = parts.first().and_then(|s| s.parse::().ok()).unwrap_or(0); + let desired_replicas = parts.get(1).and_then(|s| s.parse::().ok()).unwrap_or(0); + let ready = ready_replicas > 0 && ready_replicas == desired_replicas; + + components.push(ComponentStatus { + name: format!("{cluster}-{component}"), + namespace: GRID_SYSTEM_NS.to_owned(), + ready_replicas, + desired_replicas, + ready, + }); + + observed_facts.insert(format!("{cluster}_{component}_ready"), serde_json::Value::Bool(ready)); + observed_facts.insert( + format!("{cluster}_{component}_replicas"), + serde_json::Value::Number(ready_replicas.into()), + ); + } else { + components.push(ComponentStatus { + name: format!("{cluster}-{component}"), + namespace: GRID_SYSTEM_NS.to_owned(), + ready_replicas: 0, + desired_replicas: 1, + ready: false, + }); + observed_facts.insert(format!("{cluster}_{component}_ready"), serde_json::Value::Bool(false)); + } + } + } + + let expected_count = components.len(); + let all_ready = components.len() == expected_count && components.iter().all(|c| c.ready); + observed_facts.insert( + "total_components".to_owned(), + serde_json::Value::Number(expected_count.into()), + ); + + if all_ready { + Ok(proof_success( + &format!( + "{} components are ready across {} provider sites with one consumer entrypoint", + expected_count, + CLUSTERS.len() + ), + observed_facts, + start.elapsed(), + )) + } else { + Ok(proof_failure( + "One or more components are not ready", + observed_facts, + start.elapsed(), + )) + } +} + +/// Assert every site's `GridNetwork` reports all three remote sites connected. +#[expect( + clippy::too_many_lines, + clippy::unnecessary_wraps, + reason = "The assertion framework requires a fallible, named proof boundary." +)] +fn assert_swim_convergence() -> AssertionResult { + let start = Instant::now(); + let mut observed_facts = BTreeMap::new(); + let expected_remote_sites = CLUSTERS.len() - 1; + let mut all_converged = true; + + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + let status = poll_until( + || { + let output = Command::new("kubectl") + .args([ + "get", + "gridnetwork/grid-provider-traffic", + "--context", + &context, + "-o", + "jsonpath={.status.phase},{.status.connectedSites}", + ]) + .output()?; + if !output.status.success() { + return Ok(None); + } + let value = String::from_utf8_lossy(&output.stdout); + let Some((phase, connected)) = value.trim().split_once(',') else { + return Ok(None); + }; + let connected = connected.parse::().unwrap_or_default(); + Ok((phase == "Active" && connected == expected_remote_sites).then_some((phase.to_owned(), connected))) + }, + Duration::from_secs(90), + Duration::from_secs(3), + ); + + match status { + Ok((phase, connected)) => { + observed_facts.insert(format!("{cluster}_phase"), serde_json::Value::String(phase)); + observed_facts.insert( + format!("{cluster}_connected_sites"), + serde_json::Value::Number(connected.into()), + ); + }, + Err(error) => { + all_converged = false; + observed_facts.insert(format!("{cluster}_error"), serde_json::Value::String(error.to_string())); + }, + } + } + + observed_facts.insert( + "expected_remote_sites".to_owned(), + serde_json::Value::Number(expected_remote_sites.into()), + ); + + if all_converged { + Ok(proof_success( + "Every GridNetwork is Active with all three remote sites connected", + observed_facts, + start.elapsed(), + )) + } else { + Ok(proof_failure( + "One or more GridNetworks did not report all three remote sites connected", + observed_facts, + start.elapsed(), + )) + } +} + +/// Verify that both remote sites are discovered and routing-eligible. +/// +/// The locally declared placement site is not a remote SWIM discovery result +/// and is excluded from this assertion. +#[expect( + clippy::too_many_lines, + clippy::unnecessary_wraps, + reason = "The assertion framework requires a fallible, named proof boundary." +)] +fn assert_site_auto_discovery() -> AssertionResult { + let start = Instant::now(); + let mut observed_facts = BTreeMap::new(); + let expected_remote_count = CLUSTERS.len() - 1; + let mut all_ok = true; + + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + + let result = poll_until( + || { + let output = Command::new("kubectl") + .args([ + "get", "gridsite", + "-l", "grid.praxis-proxy.io/auto-discovered=true", + "--context", &context, + "-n", GRID_SYSTEM_NS, + "-o", "jsonpath={range .items[*]}{.metadata.name}\t{.status.phase}\t{.status.reason}\t{.spec.egress.address}\t{.spec.egress.tls.serverName}\t{.spec.trust.canonicalFingerprints}\n{end}", + ]) + .output()?; + if !output.status.success() { + return Ok(None); + } + let body = String::from_utf8_lossy(&output.stdout); + let mut verified = Vec::new(); + let port_suffix = format!(":{PROVIDER_GATEWAY_PORT}"); + for line in body.lines() { + let fields: Vec<&str> = line.trim().split('\t').collect(); + let [name, phase, reason, addr, server_name, fingerprints, ..] = fields.as_slice() else { + continue; + }; + if *phase == "Active" + && *reason == "TlsVerified" + && addr.ends_with(&port_suffix) + && !server_name.is_empty() + && !fingerprints.is_empty() + { + verified.push(serde_json::json!({ + "name": name, + "phase": phase, + "reason": reason, + "egressAddress": addr, + "serverName": server_name, + "hasFingerprints": true, + })); + } + } + if verified.len() == expected_remote_count { + Ok(Some(verified)) + } else { + Ok(None) + } + }, + Duration::from_secs(120), + Duration::from_secs(5), + ); + + match result { + Ok(remotes) => { + observed_facts.insert(format!("{cluster}_remote_sites"), serde_json::Value::Array(remotes)); + }, + Err(error) => { + all_ok = false; + observed_facts.insert(format!("{cluster}_error"), serde_json::Value::String(error.to_string())); + }, + } + } + + observed_facts.insert( + "expected_remote_count".to_owned(), + serde_json::Value::Number(expected_remote_count.into()), + ); + + if all_ok { + Ok(proof_success( + "Every cluster has two Active/TlsVerified remote `GridSites` with :8443 addresses and trust configuration", + observed_facts, + start.elapsed(), + )) + } else { + Ok(proof_failure( + "One or more clusters lack Active auto-discovered remote GridSites", + observed_facts, + start.elapsed(), + )) + } +} + +/// Assert each consumer receives and accepts a versioned overlay. +/// +/// Validates per-site: `ConfigMap` exists with non-empty data, gateway deployment +/// is ready, and a request is routed through the accepted overlay. +#[expect( + clippy::too_many_lines, + reason = "The proof checks overlay, gateway, and routing acceptance together." +)] +fn assert_overlay_acceptance() -> AssertionResult { + let start = Instant::now(); + let mut observed_facts = BTreeMap::new(); + let mut all_accepted = true; + + { + let cluster = CONSUMER_SITE; + let context = format!("kind-grid-provider-traffic-{cluster}"); + + // Step 1: Overlay ConfigMap exists + let overlay_output = Command::new("kubectl") + .args([ + "get", + "configmap", + "grid-overlay-grid-provider-traffic-consumer-gateway", + "--context", + &context, + "-n", + "grid-system", + "-o", + "jsonpath={.metadata.resourceVersion}", + ]) + .output()?; + + let resource_version = String::from_utf8_lossy(&overlay_output.stdout).trim().to_owned(); + let overlay_exists = overlay_output.status.success() && !resource_version.is_empty(); + + // Step 2: Overlay ConfigMap has non-empty data + let overlay_data_output = Command::new("kubectl") + .args([ + "get", + "configmap", + "grid-overlay-grid-provider-traffic-consumer-gateway", + "--context", + &context, + "-n", + "grid-system", + "-o", + "jsonpath={.data}", + ]) + .output()?; + + let data_str = String::from_utf8_lossy(&overlay_data_output.stdout).trim().to_owned(); + let has_data = overlay_data_output.status.success() && !data_str.is_empty() && data_str != "{}"; + + // Step 3: Consumer gateway deployment is ready + let deploy_output = Command::new("kubectl") + .args([ + "get", + "deployment/consumer-gateway", + "--context", + &context, + "-n", + "grid-system", + "-o", + "jsonpath={.status.readyReplicas}", + ]) + .output()?; + + let gateway_ready = deploy_output.status.success() + && String::from_utf8_lossy(&deploy_output.stdout) + .trim() + .parse::() + .is_ok_and(|n| n > 0); + + // Step 4: A valid request proves that the accepted overlay is serving. + // Praxis health endpoints are exposed only on the pod-local admin listener. + let routing_output = run_curl_probe( + &context, + &format!("overlay-routing-{cluster}"), + &[ + "curl", + "--fail-with-body", + "--silent", + "--show-error", + "--max-time", + "10", + "--header", + "Content-Type: application/json", + "--header", + "Authorization: Bearer consumer-token", + "--data", + r#"{"model":"Qwen/Qwen3-0.6B","messages":[{"role":"user","content":"overlay probe"}],"max_tokens":16}"#, + "http://consumer-gateway.grid-system.svc.cluster.local:8080/v1/chat/completions", + ], + )?; + + let routing_ok = routing_output.status.success(); + + let overlay_accepted = overlay_exists && has_data && gateway_ready && routing_ok; + if !overlay_accepted { + all_accepted = false; + } + + observed_facts.insert( + format!("{cluster}_overlay_configmap_exists"), + serde_json::Value::Bool(overlay_exists), + ); + observed_facts.insert(format!("{cluster}_overlay_has_data"), serde_json::Value::Bool(has_data)); + observed_facts.insert( + format!("{cluster}_gateway_ready"), + serde_json::Value::Bool(gateway_ready), + ); + observed_facts.insert( + format!("{cluster}_overlay_routing_ok"), + serde_json::Value::Bool(routing_ok), + ); + observed_facts.insert( + format!("{cluster}_resource_version"), + serde_json::Value::String(resource_version), + ); + } + + observed_facts.insert( + "all_overlays_accepted".to_owned(), + serde_json::Value::Bool(all_accepted), + ); + + if all_accepted { + Ok(proof_success( + "Consumer entrypoint overlay ConfigMap exists with data, gateway ready, and routing passes", + observed_facts, + start.elapsed(), + )) + } else { + Ok(proof_failure( + "One or more sites failed overlay acceptance (ConfigMap/data/ready/routing)", + observed_facts, + start.elapsed(), + )) + } +} + +/// Assert consumer and operator cannot access provider credentials. +fn require_local_image(image: &str) -> Result<(), Box> { + let status = Command::new("docker") + .args(["image", "inspect", image]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status()?; + if status.success() { + return Ok(()); + } + Err(format!( + "required local image {image:?} is absent; build it or set \ + GRID_XTASK_IMAGE_PULL_POLICY=IfNotPresent with registry image overrides" + ) + .into()) +} + +/// Load local container images into all Kind clusters. +/// +/// Reads image references from the `GRID_XTASK_*_IMAGE` environment variables +/// (same source as `apply_image_overrides`). When `imagePullPolicy` is not +/// `Never`, this is a no-op. +#[expect( + clippy::too_many_lines, + reason = "Image loading is one bounded setup operation across the three clusters." +)] +fn load_images_into_clusters(forge_bin: &Path, resolved_config: &Path) -> Result<(), Box> { + let pull_policy = std::env::var("GRID_XTASK_IMAGE_PULL_POLICY").unwrap_or_else(|_| "Never".to_owned()); + if pull_policy != "Never" { + eprintln!(" skipping Kind image loading (pull policy is {pull_policy})"); + return Ok(()); + } + + let gateway = + std::env::var("GRID_XTASK_GATEWAY_IMAGE").unwrap_or_else(|_| "praxis-ai:provider-traffic-demo".to_owned()); + let operator = + std::env::var("GRID_XTASK_OPERATOR_IMAGE").unwrap_or_else(|_| "grid-operator:provider-traffic-demo".to_owned()); + let overlay_sync = crate::env::image_overrides::overlay_sync_image(); + let vcr = crate::env::image_overrides::vcr_image(); + + for image in [&gateway, &operator, &overlay_sync, &vcr] { + require_local_image(image)?; + eprintln!(" verified local image: {image}"); + } + + for cluster in CLUSTERS { + for image in [&gateway, &operator, &overlay_sync, &vcr] { + eprintln!(" loading {image} into {cluster}..."); + let output = Command::new(forge_bin.as_os_str()) + .arg("--config") + .arg(resolved_config) + .args(["--non-interactive", "cluster", "load-image", cluster, image]) + .output()?; + if !output.status.success() { + return Err(format!( + "failed to load {image} into {cluster}: {}", + String::from_utf8_lossy(&output.stderr).trim() + ) + .into()); + } + } + eprintln!(" [OK] {cluster}: all images loaded"); + } + Ok(()) +} + +/// Generate TLS certificates for all provider-traffic identities. +/// +/// Must be called BEFORE `forge up` so the certificates exist on the host +/// when `install_provider_boundary` creates the Kubernetes Secrets. +fn stage_provider_boundary() -> Result<(), Box> { + let identities: Vec = CLUSTERS.iter().map(|c| (*c).to_owned()).collect(); + certs::generate_all(&identities)?; + + let wrong_ca = ::certs::generate_ca("Combined Site untrusted test CA")?; + fs::write(Path::new(CERTS_DIR).join("untrusted-ca.pem"), wrong_ca.cert_pem)?; + + eprintln!(" [OK] TLS certificates generated for provider-a, provider-b, provider-c"); + Ok(()) +} + +/// Create TLS and credential Secrets in every provider-traffic cluster. +/// +/// Must be called AFTER `forge up` since the clusters must exist. Gateway +/// deployments are restarted so pods pick up the new volume mounts. +fn install_provider_boundary() -> Result<(), Box> { + let certs_dir = Path::new(CERTS_DIR); + + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + + apply_tls_secret(&context, cluster, CONSUMER_TLS_SECRET, certs_dir)?; + apply_tls_secret(&context, cluster, PROVIDER_TLS_SECRET, certs_dir)?; + apply_tls_secret(&context, "wrong-org-client", WRONG_ORG_TLS_SECRET, certs_dir)?; + + eprintln!(" [OK] {cluster}: TLS secrets installed"); + } + + Ok(()) +} + +/// Create a TLS secret from the generated cert, key, and CA files. +#[expect( + clippy::too_many_lines, + reason = "TLS secret creation is one bounded setup operation." +)] +fn apply_tls_secret( + context: &str, + identity: &str, + secret_name: &str, + certs_dir: &Path, +) -> Result<(), Box> { + let output = Command::new("kubectl") + .args([ + "--context", + context, + "-n", + GRID_SYSTEM_NS, + "create", + "secret", + "generic", + secret_name, + &format!( + "--from-file=tls.crt={}", + certs_dir.join(format!("{identity}-cert.pem")).display() + ), + &format!( + "--from-file=tls.key={}", + certs_dir.join(format!("{identity}-key.pem")).display() + ), + &format!("--from-file=ca.crt={}", certs_dir.join("ca.pem").display()), + "--dry-run=client", + "-o", + "yaml", + ]) + .output()?; + if !output.status.success() { + return Err(format!( + "failed to render {identity} Secret/{secret_name}: {}", + String::from_utf8_lossy(&output.stderr).trim() + ) + .into()); + } + kubectl::apply_manifest(context, &String::from_utf8(output.stdout)?) +} + +/// Generate a random 32-byte hex provider credential. +fn generate_provider_credential() -> Result> { + let output = Command::new("openssl").args(["rand", "-hex", "32"]).output()?; + if !output.status.success() { + return Err("openssl failed to generate provider credential".into()); + } + let token = String::from_utf8(output.stdout)?.trim().to_owned(); + if token.len() != 64 || !token.bytes().all(|b| b.is_ascii_hexdigit()) { + return Err("openssl returned an invalid provider credential".into()); + } + Ok(token) +} + +/// Create an Opaque Secret with a `token` key. +fn apply_credential_secret(context: &str, secret_name: &str, token: &str) -> Result<(), Box> { + let manifest = format!( + r#"{{"apiVersion":"v1","kind":"Secret","metadata":{{"name":"{secret_name}","namespace":"{GRID_SYSTEM_NS}"}},"type":"Opaque","stringData":{{"token":"{token}"}}}}"#, + ); + kubectl::apply_manifest(context, &manifest) +} + +/// Data extracted from the operator-created overlay `ConfigMap`. +/// Read a single cluster's overlay `ConfigMap` and return structured data. +/// +/// Captures both the Kubernetes `resourceVersion` (per-cluster) and the +/// semantic revision from the `grid.praxis-proxy.io/overlay-revision` +/// annotation (content-addressed, safe to compare across clusters). +#[expect( + clippy::too_many_lines, + reason = "The reader validates and parses one Kubernetes ConfigMap response." +)] +fn read_cluster_overlay(cluster: &str) -> Result> { + let context = format!("kind-grid-provider-traffic-{cluster}"); + + let output = Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "configmap", + OVERLAY_CONFIGMAP, + "-o", + "json", + ]) + .output()?; + + if !output.status.success() { + return Err(format!("{cluster}: overlay ConfigMap not found").into()); + } + + let cm: serde_json::Value = + serde_json::from_slice(&output.stdout).map_err(|e| format!("{cluster}: overlay ConfigMap invalid: {e}"))?; + + let resource_version = cm + .pointer("/metadata/resourceVersion") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_owned(); + + let semantic_revision = cm + .pointer("/metadata/annotations/grid.praxis-proxy.io~1overlay-revision") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_owned(); + + let routing_json = cm + .pointer("/data/routing-config.json") + .and_then(|v| v.as_str()) + .ok_or_else(|| format!("{cluster}: overlay missing routing-config.json"))?; + + let parsed: serde_json::Value = + serde_json::from_str(routing_json).map_err(|e| format!("{cluster}: routing-config.json invalid: {e}"))?; + + let raw_candidates = parsed + .get("candidates") + .and_then(|v| v.as_array()) + .ok_or_else(|| format!("{cluster}: overlay missing candidates array"))?; + + let mut stable_ids = BTreeMap::new(); + let mut candidates = Vec::new(); + + for c in raw_candidates { + let cluster_field = c.get("cluster").and_then(|v| v.as_str()).unwrap_or("").to_owned(); + let stable_id = c.get("stable_id").and_then(|v| v.as_str()).unwrap_or("").to_owned(); + let kind = c.get("kind").and_then(|v| v.as_str()).unwrap_or("").to_owned(); + let name = c.get("name").and_then(|v| v.as_str()).unwrap_or("").to_owned(); + let site = c.get("site").and_then(|v| v.as_str()).unwrap_or("").to_owned(); + let fresh = c.get("fresh").and_then(serde_json::Value::as_bool); + let admission_state = c + .get("admission_state") + .and_then(serde_json::Value::as_str) + .map(str::to_owned); + let selection_group = c + .get("selection_group") + .and_then(serde_json::Value::as_u64) + .and_then(|value| u32::try_from(value).ok()); + + if !cluster_field.is_empty() && !stable_id.is_empty() { + stable_ids.insert(cluster_field.clone(), stable_id.clone()); + } + + candidates.push(OverlayCandidate { + kind, + name, + site, + cluster: cluster_field, + stable_id, + fresh, + admission_state, + selection_group, + }); + } + + Ok(OverlayData { + resource_version, + semantic_revision, + stable_ids, + candidates, + }) +} + +/// Establish the non-traffic preconditions for the measured picker proof. +/// +/// This gate deliberately performs no request. It verifies one consumer +/// replica, the explicit round-robin policy, three fresh `NewAndExisting` +/// candidates in group zero, and three consecutive identical semantic +/// revisions. The measured request window starts only after this gate passes. +#[expect( + clippy::too_many_lines, + reason = "The readiness barrier checks all non-traffic serving invariants." +)] +fn wait_for_round_robin_readiness() -> Result, Box> { + let context = "kind-grid-provider-traffic-provider-a"; + let replicas = Command::new("kubectl") + .args([ + "--context", + context, + "-n", + GRID_SYSTEM_NS, + "get", + "deployment/consumer-gateway", + "-o", + "jsonpath={.spec.replicas},{.status.replicas},{.status.readyReplicas}", + ]) + .output()?; + let replica_text = String::from_utf8_lossy(&replicas.stdout).trim().to_owned(); + if !replicas.status.success() || replica_text != "1,1,1" { + return Err(format!("consumer gateway is not exactly one ready replica: {replica_text}").into()); + } + + let mut stable_revision: Option = None; + let mut stable_signature: Option = None; + for observation in 1..=3 { + let overlay = read_cluster_overlay("provider-a")?; + let configmap = Command::new("kubectl") + .args([ + "--context", + context, + "-n", + GRID_SYSTEM_NS, + "get", + "configmap", + OVERLAY_CONFIGMAP, + "-o", + "json", + ]) + .output()?; + if !configmap.status.success() { + return Err("consumer overlay ConfigMap could not be read".into()); + } + let configmap_json: serde_json::Value = serde_json::from_slice(&configmap.stdout)?; + let routing_text = configmap_json + .pointer("/data/routing-config.json") + .and_then(serde_json::Value::as_str) + .ok_or("consumer overlay ConfigMap has no routing-config.json data")?; + let routing: serde_json::Value = serde_json::from_str(routing_text)?; + let mode = routing + .pointer("/selection_policy/mode") + .and_then(serde_json::Value::as_str) + .ok_or("overlay does not publish selection_policy.mode")?; + if mode != "roundRobin" { + return Err(format!("overlay selection mode is {mode}, expected roundRobin").into()); + } + + let mut candidate_signature = Vec::new(); + for candidate in &overlay.candidates { + if candidate.selection_group != Some(0) { + return Err(format!( + "candidate {} is not in group 0: {:?}", + candidate.cluster, candidate.selection_group + ) + .into()); + } + if candidate.fresh == Some(false) { + return Err(format!("candidate {} is stale", candidate.cluster).into()); + } + if candidate + .admission_state + .as_deref() + .is_some_and(|state| state != "new_and_existing") + { + return Err(format!( + "candidate {} is not NewAndExisting: {:?}", + candidate.cluster, candidate.admission_state + ) + .into()); + } + candidate_signature.push(format!( + "{}:{}:{:?}:{:?}", + candidate.cluster, candidate.stable_id, candidate.selection_group, candidate.admission_state + )); + } + candidate_signature.sort(); + let signature = candidate_signature.join("|"); + if overlay.candidates.len() != 3 { + return Err(format!("expected three candidates, found {}", overlay.candidates.len()).into()); + } + if stable_revision + .as_ref() + .is_some_and(|revision| revision != &overlay.semantic_revision) + || stable_signature.as_ref().is_some_and(|previous| previous != &signature) + { + return Err("overlay changed while establishing the measured proof precondition".into()); + } + stable_revision = Some(overlay.semantic_revision); + stable_signature = Some(signature); + if observation < 3 { + std::thread::park_timeout(Duration::from_secs(2)); + } + } + + let serving_revision = stable_revision + .as_deref() + .ok_or("round-robin readiness did not establish a semantic revision")?; + wait_for_consumer_gateway_revision(serving_revision)?; + + let mut facts = BTreeMap::new(); + facts.insert("consumer_gateway_replicas".to_owned(), serde_json::json!(replica_text)); + facts.insert("semantic_revision".to_owned(), serde_json::json!(stable_revision)); + facts.insert("candidate_signature".to_owned(), serde_json::json!(stable_signature)); + facts.insert("selection_policy".to_owned(), serde_json::json!("roundRobin")); + facts.insert("candidate_count".to_owned(), serde_json::json!(3)); + facts.insert( + "praxis_serving_revision".to_owned(), + serde_json::json!(serving_revision), + ); + Ok(facts) +} + +/// Read bounded logs from the single consumer gateway used by the measured +/// provider-traffic proof. +fn consumer_gateway_logs() -> Result> { + let output = Command::new("kubectl") + .args([ + "--context", + "kind-grid-provider-traffic-provider-a", + "-n", + GRID_SYSTEM_NS, + "logs", + "deployment/consumer-gateway", + "-c", + "praxis", + "--tail=300", + ]) + .output()?; + if !output.status.success() { + return Err(format!( + "failed to read consumer gateway logs: {}", + safe_truncate_str(String::from_utf8_lossy(&output.stderr).trim(), 160) + ) + .into()); + } + Ok(strip_csi_sgr(&String::from_utf8_lossy(&output.stdout))) +} + +/// Strip the ANSI SGR sequences emitted by the gateway's tracing subscriber. +fn strip_csi_sgr(value: &str) -> String { + let mut output = String::with_capacity(value.len()); + let mut chars = value.chars(); + while let Some(character) = chars.next() { + if character == '\x1b' { + if chars.next() == Some('[') { + for final_byte in chars.by_ref() { + if final_byte.is_ascii_alphabetic() { + break; + } + } + } + } else { + output.push(character); + } + } + output +} + +/// Wait until Praxis reports that the exact final overlay revision is serving. +/// +/// The `ConfigMap` and projected file can be ready before the gateway watcher +/// has accepted the file. Starting the measured window earlier would count +/// requests against the initial local-only snapshot and invalidate the +/// round-robin proof. +fn wait_for_consumer_gateway_revision(revision: &str) -> Result<(), Box> { + const TIMEOUT: Duration = Duration::from_secs(90); + const POLL: Duration = Duration::from_secs(2); + let deadline = Instant::now() + TIMEOUT; + loop { + let logs = consumer_gateway_logs()?; + let accepted = latest_log_field(&logs, "accepted_revision"); + let serving = latest_log_field(&logs, "serving_revision"); + if accepted.as_deref() == Some(revision) && serving.as_deref() == Some(revision) { + return Ok(()); + } + if Instant::now() >= deadline { + let accepted_summary = accepted + .as_deref() + .map_or_else(|| "none".to_owned(), |value| safe_truncate_str(value, 16)); + let serving_summary = serving + .as_deref() + .map_or_else(|| "none".to_owned(), |value| safe_truncate_str(value, 16)); + return Err(format!( + "consumer gateway did not report accepted/serving overlay revision {} within {TIMEOUT:?} (accepted={}, serving={})", + safe_truncate_str(revision, 16), + accepted_summary, + serving_summary + ) + .into()); + } + std::thread::park_timeout(POLL); + } +} + +/// Return the latest exact tracing field value from bounded gateway logs. +fn latest_log_field(logs: &str, field: &str) -> Option { + logs.lines() + .rev() + .find_map(|line| { + let prefix = format!("{field}="); + line.match_indices(&prefix).find_map(|(index, _)| { + let at_boundary = index == 0 + || line + .get(..index) + .and_then(|value| value.chars().next_back()) + .is_some_and(char::is_whitespace); + if !at_boundary { + return None; + } + let value = line.get(index + prefix.len()..)?; + Some( + value + .strip_prefix('"') + .map_or_else( + || value.split_whitespace().next().unwrap_or(""), + |value| value.split('"').next().unwrap_or(""), + ) + .to_owned(), + ) + }) + }) + .filter(|value| !value.is_empty()) +} + +/// Build the expected provider candidate name set. +fn expected_candidates() -> BTreeSet { + let mut expected = BTreeSet::new(); + for cluster in CLUSTERS { + expected.insert(format!("vcr-{cluster}-provider")); + } + + expected +} + +/// Wait for each cluster's operator to produce its expected local candidate. +/// +/// Pre-SWIM, each operator only knows its local `InferenceProvider` resources. +/// Global convergence (all candidates on every cluster) happens after SWIM +/// seeding in a separate phase. +#[expect( + clippy::too_many_lines, + reason = "The readiness poll checks each local cluster's overlay." +)] +fn wait_for_local_overlays() -> Result, Box> { + let timeout = Duration::from_secs(180); + let interval = Duration::from_secs(5); + let start = Instant::now(); + + eprintln!(" Polling for local overlay ConfigMaps (timeout {timeout:?})..."); + + while start.elapsed() < timeout { + let mut overlays = BTreeMap::new(); + let mut all_ready = true; + + for cluster in CLUSTERS { + let expected_local = format!("vcr-{cluster}-provider"); + match read_cluster_overlay(cluster) { + Ok(data) if data.stable_ids.contains_key(&expected_local) => { + eprintln!( + " {cluster}: {expected_local} present (semantic_rev={}, stable_id={})", + data.semantic_revision, + data.stable_ids.get(&expected_local).map_or("?", String::as_str) + ); + overlays.insert((*cluster).to_owned(), data); + }, + Ok(data) => { + eprintln!( + " {cluster}: overlay present but missing {expected_local} (has: {:?})", + data.stable_ids.keys().collect::>() + ); + all_ready = false; + }, + Err(_) => { + all_ready = false; + }, + } + } + + if all_ready && overlays.len() == CLUSTERS.len() { + return Ok(overlays); + } + + std::thread::park_timeout(interval); + } + + collect_overlay_diagnostics(); + Err(format!("Local overlay ConfigMaps not ready after {timeout:?}").into()) +} + +/// Wait until every provider-traffic cluster serves the same candidate set. +#[expect( + clippy::too_many_lines, + reason = "The convergence poll compares the bounded three-cluster overlay set." +)] +fn wait_for_global_overlay_convergence( + expected: &BTreeSet, +) -> Result, Box> { + let timeout = Duration::from_secs(180); + let interval = Duration::from_secs(5); + let start = Instant::now(); + + eprintln!( + " Waiting for global overlay convergence \ + ({} candidates on all clusters, timeout {timeout:?})...", + expected.len() + ); + + while start.elapsed() < timeout { + let mut overlays = BTreeMap::new(); + let mut all_converged = true; + + for cluster in CLUSTERS { + match read_cluster_overlay(cluster) { + Ok(data) => { + let missing: Vec<&String> = expected.iter().filter(|c| !data.stable_ids.contains_key(*c)).collect(); + if missing.is_empty() { + overlays.insert((*cluster).to_owned(), data); + } else { + eprintln!(" {cluster}: missing candidates {missing:?}"); + all_converged = false; + } + }, + Err(_) => { + all_converged = false; + }, + } + } + + if all_converged && overlays.len() == CLUSTERS.len() { + let reference_site = CLUSTERS.first().ok_or("CLUSTERS is empty")?; + let reference = overlays + .get(*reference_site) + .ok_or("reference site missing from overlays")?; + + for cluster in CLUSTERS.iter().skip(1) { + let site_data = overlays + .get(*cluster) + .ok_or_else(|| format!("{cluster} missing from overlays"))?; + for candidate in expected { + let ref_id = reference + .stable_ids + .get(candidate) + .ok_or_else(|| format!("{reference_site}: missing {candidate}"))?; + let site_id = site_data + .stable_ids + .get(candidate) + .ok_or_else(|| format!("{cluster}: missing {candidate}"))?; + if site_id != ref_id { + return Err(format!( + "stable_id mismatch for {candidate}: \ + {reference_site}={ref_id} vs {cluster}={site_id}" + ) + .into()); + } + } + } + + eprintln!( + " Global overlay converged: {} candidates on all {} clusters, stable_ids agree", + expected.len(), + CLUSTERS.len() + ); + return Ok(overlays); + } + + std::thread::park_timeout(interval); + } + + collect_overlay_diagnostics(); + Err(format!("Global overlay convergence failed after {timeout:?}").into()) +} + +/// Collect diagnostic information when the overlay `ConfigMap` fails to converge. +#[expect( + clippy::too_many_lines, + reason = "Diagnostics intentionally report each bounded control-plane boundary." +)] +fn collect_overlay_diagnostics() { + eprintln!(" [DIAG] Collecting overlay failure diagnostics...\n"); + + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + eprintln!(" ======== {cluster} ========"); + + eprintln!(" [DIAG] {cluster}: 1. Operator deployment SWIM env vars"); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "deployment", + "grid-operator", + "-o", + "jsonpath={range .spec.template.spec.containers[0].env[*]}{.name}={.value}{'\\n'}{end}", + ]) + .status(), + ); + + eprintln!("\n [DIAG] {cluster}: 2. SWIM service details"); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "svc", + "grid-operator-swim", + "-o", + "wide", + ]) + .status(), + ); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "endpoints", + "grid-operator-swim", + "-o", + "yaml", + ]) + .status(), + ); + + eprintln!(" [DIAG] {cluster}: 3. Operator pod status"); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "pods", + "-l", + "app.kubernetes.io/name=grid-operator", + "-o", + "wide", + ]) + .status(), + ); + + eprintln!(" [DIAG] {cluster}: 4. Operator logs (last 50 lines, unfiltered)"); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "logs", + "deployment/grid-operator", + "--tail=50", + ]) + .status(), + ); + + eprintln!("\n [DIAG] {cluster}: 5. GridNetwork CRD status"); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "gridnetwork", + "-o", + "yaml", + ]) + .status(), + ); + + eprintln!(" [DIAG] {cluster}: 6. Overlay ConfigMap content"); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "configmap", + OVERLAY_CONFIGMAP, + "-o", + "json", + ]) + .status(), + ); + + eprintln!(" [DIAG] {cluster}: 7. InferenceProvider CRs"); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "inferenceprovider", + "-o", + "yaml", + ]) + .status(), + ); + + eprintln!(" [DIAG] {cluster}: 8. Helm values for grid-operator"); + drop( + Command::new("helm") + .args([ + "get", + "values", + "grid-operator", + "--namespace", + GRID_SYSTEM_NS, + "--kube-context", + &context, + "-o", + "yaml", + ]) + .status(), + ); + + eprintln!(" [DIAG] {cluster}: 9. NetworkPolicy in grid-system"); + drop( + Command::new("kubectl") + .args(["--context", &context, "-n", GRID_SYSTEM_NS, "get", "networkpolicy"]) + .status(), + ); + + eprintln!(); + } + + eprintln!(" [DIAG] 10. Cross-cluster SWIM connectivity check"); + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + for target in CLUSTERS { + if *target == *cluster { + continue; + } + let target_context = format!("kind-grid-provider-traffic-{target}"); + if let Ok(output) = Command::new("kubectl") + .args([ + "--context", + &target_context, + "-n", + GRID_SYSTEM_NS, + "get", + "svc", + "grid-operator-swim", + "-o", + "jsonpath={.status.loadBalancer.ingress[0].ip}", + ]) + .output() + { + let ip = String::from_utf8_lossy(&output.stdout); + eprintln!(" [DIAG] {cluster} -> {target} (SWIM LB {ip}): testing TCP 7946"); + drop( + Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "exec", + "deployment/grid-operator", + "--", + "sh", + "-c", + &format!( + "timeout 3 sh -c 'echo | nc -w 2 {ip} 7946' && echo REACHABLE || echo UNREACHABLE" + ), + ]) + .status(), + ); + } + } + } +} + +/// Materialize provider gateway configuration from the pre-SWIM overlay map. +/// +/// For each cluster, extracts the `stable_id` for the local +/// `vcr-{cluster}-provider` candidate and renders the provider praxis.yaml +/// template with: +/// - `SITE_PLACEHOLDER` → cluster name +/// - `CANDIDATE_ID_PLACEHOLDER` → stable ID from the overlay +/// +/// Creates the `provider-gateway-config` `ConfigMap` in each cluster. +#[expect( + clippy::too_many_lines, + reason = "Provider config materialization is one bounded setup phase." +)] +fn materialize_provider_config( + overlays: &BTreeMap, + demo_root: &Path, +) -> Result<(), Box> { + let template_path = demo_root.join("configs/provider/praxis.yaml"); + let template = + fs::read_to_string(template_path).map_err(|e| format!("failed to read provider config template: {e}"))?; + + for cluster in CLUSTERS { + let provider_name = format!("vcr-{cluster}-provider"); + let overlay = overlays + .get(*cluster) + .ok_or_else(|| format!("no overlay data for cluster {cluster}"))?; + let stable_id = overlay.stable_ids.get(&provider_name).ok_or_else(|| { + format!( + "{cluster}: no candidate {provider_name} in overlay (has: {:?})", + overlay.stable_ids.keys().collect::>() + ) + })?; + + eprintln!(" {cluster}: {provider_name} -> {stable_id}"); + + let rendered = template + .replace("SITE_PLACEHOLDER", cluster) + .replace("CANDIDATE_ID_PLACEHOLDER", stable_id); + + let context = format!("kind-grid-provider-traffic-{cluster}"); + + let create = Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "create", + "configmap", + "provider-gateway-config", + &format!("--from-literal=praxis.yaml={rendered}"), + "--dry-run=client", + "-o", + "yaml", + ]) + .output()?; + + if !create.status.success() { + return Err(format!( + "failed to render provider-gateway-config for {cluster}: {}", + String::from_utf8_lossy(&create.stderr).trim() + ) + .into()); + } + + kubectl::apply_manifest(&context, &String::from_utf8(create.stdout)?)?; + eprintln!(" [OK] {cluster}: provider-gateway-config created (stable_id={stable_id})"); + } + + Ok(()) +} + +/// Materialize the Forge configuration with image overrides. +fn materialize_config(source: &Path) -> Result> { + let content = fs::read_to_string(source)?; + let mut config: serde_yaml::Value = serde_yaml::from_str(&content)?; + apply_image_overrides(&mut config); + let rendered = serde_yaml::to_string(&config)?; + let parent = source.parent().ok_or("source config must have parent directory")?; + let output = parent.join(".forge.resolved.yaml"); + fs::write(&output, rendered)?; + Ok(output) +} + +/// Apply image overrides from environment variables to the Forge configuration. +#[expect( + clippy::too_many_lines, + clippy::collapsible_if, + reason = "Image override application with structured YAML manipulation; nested ifs follow YAML structure hierarchy" +)] +fn apply_image_overrides(config: &mut serde_yaml::Value) { + let gateway_image = + std::env::var("GRID_XTASK_GATEWAY_IMAGE").unwrap_or_else(|_| "praxis-ai:provider-traffic-demo".to_owned()); + let operator_image = + std::env::var("GRID_XTASK_OPERATOR_IMAGE").unwrap_or_else(|_| "grid-operator:provider-traffic-demo".to_owned()); + let overlay_sync_image = crate::env::image_overrides::overlay_sync_image(); + let vcr_image = crate::env::image_overrides::vcr_image(); + let image_pull_policy = std::env::var("GRID_XTASK_IMAGE_PULL_POLICY").unwrap_or_else(|_| "Never".to_owned()); + + let (gateway_repo, gateway_tag) = parse_image_ref(&gateway_image); + let (operator_repo, operator_tag) = parse_image_ref(&operator_image); + let (overlay_sync_repo, overlay_sync_tag) = parse_image_ref(&overlay_sync_image); + + if let Some(spec) = config.get_mut("spec") { + if let Some(clusters) = spec.get_mut("clusters") { + if let Some(clusters_array) = clusters.as_sequence_mut() { + for cluster in clusters_array { + if let Some(properties) = cluster.get_mut("properties") { + if let Some(props_map) = properties.as_mapping_mut() { + let pairs = [ + ("gatewayImage", &gateway_image), + ("operatorImage", &operator_image), + ("vcrImage", &vcr_image), + ("imagePullPolicy", &image_pull_policy), + ("gatewayImageRepo", &gateway_repo), + ("gatewayImageTag", &gateway_tag), + ("operatorImageRepo", &operator_repo), + ("operatorImageTag", &operator_tag), + ("overlaySyncImage", &overlay_sync_image), + ("overlaySyncImageRepo", &overlay_sync_repo), + ("overlaySyncImageTag", &overlay_sync_tag), + ]; + for (key, val) in pairs { + props_map.insert( + serde_yaml::Value::String(key.to_owned()), + serde_yaml::Value::String(val.clone()), + ); + } + } + } + } + } + } + } +} + +/// Parse image reference into (repo, tag) components. +fn parse_image_ref(image: &str) -> (String, String) { + if let Some(colon_pos) = image.rfind(':') { + let (repo, tag) = image.split_at(colon_pos); + // Skip the ':' character + let tag = tag.strip_prefix(':').unwrap_or(tag); + (repo.to_owned(), tag.to_owned()) + } else { + (image.to_owned(), "latest".to_owned()) + } +} + +/// Prepare setup context from configuration. +fn prepare_setup(forge_config: &Path) -> Result> { + let root = super::demo_root(forge_config); + eprintln!("Forge config: {}", forge_config.display()); + eprintln!("Demo root: {}", root.display()); + let resolved_config = materialize_config(forge_config)?; + let forge_bin = glb::resolve_forge_binary() + .ok_or("praxis-forge binary not found")? + .into(); + + Ok(ProviderTrafficContext { + demo_root: root, + resolved_config, + forge_bin, + }) +} + +/// Authorize auto-discovered remote `GridSites` with identity trust material. +/// +/// For each local cluster, waits for the two remote auto-discovered `GridSites`, +/// verifies the SWIM-advertised certificate matches the staged identity, then +/// patches `spec.egress.tls.serverName` and `spec.trust.canonicalFingerprints`. +/// The controller transitions the site to Active naturally after the patch. +fn authorize_discovered_sites() -> Result<(), Box> { + const TRUST_TIMEOUT: Duration = Duration::from_secs(120); + const GRID_NETWORK: &str = "grid-provider-traffic"; + + for local in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{local}"); + eprintln!(); + eprintln!(" {local}: authorizing remote provider sites"); + for remote in CLUSTERS { + if *remote == *local { + continue; + } + let site_name = format!("{GRID_NETWORK}-{remote}"); + operator::wait_for_auto_gridsite(&context, &site_name, GRID_NETWORK, TRUST_TIMEOUT)?; + let canonical_fp = certs::site_certificate_fingerprint(remote)?; + operator::wait_for_expected_site_certificate(&context, &site_name, &canonical_fp, TRUST_TIMEOUT)?; + let server_name = format!("{remote}.grid.internal"); + operator::patch_gridsite_identity_trust(&context, &site_name, &canonical_fp, &server_name)?; + operator::wait_for_gridsite_phase(&context, &site_name, "Active", TRUST_TIMEOUT)?; + } + } + eprintln!(" [OK] All auto-discovered remote GridSites authorized and Active"); + Ok(()) +} + +/// Deploy the provider-traffic environment. +#[expect( + clippy::too_many_lines, + reason = "sequential setup steps: each step depends on the previous; splitting obscures the setup flow" +)] +fn deploy_setup(context: &ProviderTrafficContext) -> Result> { + let total_phases = SETUP_PHASES; + let mut phase = 0; + let mut next = || { + phase += 1; + phase + }; + + eprintln!(); + eprintln!( + "[SETUP {}/{}] Resolving Forge config and building images", + next(), + total_phases + ); + + // Validate the resolved forge configuration + let output = Command::new(&context.forge_bin) + .args(["config", "validate", "--config"]) + .arg(&context.resolved_config) + .output()?; + + if !output.status.success() { + return Err(format!( + "Forge config validation failed: {}", + String::from_utf8_lossy(&output.stderr) + ) + .into()); + } + eprintln!(" [OK] Forge config resolved to {}", context.resolved_config.display()); + + eprintln!(); + eprintln!( + "[SETUP {}/{}] Generating TLS certificates for all sites", + next(), + total_phases + ); + + stage_provider_boundary()?; + + eprintln!(); + eprintln!( + "[SETUP {}/{}] Creating three provider Kind clusters: provider-a, provider-b, provider-c", + next(), + total_phases + ); + + let status = Command::new(&context.forge_bin) + .args(["up", "--config"]) + .arg(&context.resolved_config) + .status()?; + + if !status.success() { + return Err("Failed to create provider-traffic clusters".into()); + } + + eprintln!(" [OK] All three provider clusters created"); + + eprintln!(); + eprintln!( + "[SETUP {}/{}] Loading container images into Kind clusters", + next(), + total_phases + ); + + load_images_into_clusters(&context.forge_bin, &context.resolved_config)?; + + eprintln!(); + eprintln!("[SETUP {}/{}] Deploying infrastructure stacks", next(), total_phases); + + let apply_stack = |forge_bin: &Path, + resolved_config: &Path, + cluster: &str, + stack: &str| + -> Result<(), Box> { + eprintln!(" applying {stack} to {cluster}..."); + let status = Command::new(forge_bin) + .arg("--config") + .arg(resolved_config) + .args(["--non-interactive", "stack", "apply", cluster, stack]) + .status()?; + if !status.success() { + return Err(format!("Failed to apply {stack} to {cluster}").into()); + } + Ok(()) + }; + + for cluster in CLUSTERS { + apply_stack(&context.forge_bin, &context.resolved_config, cluster, "metallb")?; + } + for cluster in CLUSTERS { + let op_stack = format!("{cluster}-operator-base"); + apply_stack(&context.forge_bin, &context.resolved_config, cluster, &op_stack)?; + } + eprintln!(" [OK] Infrastructure stacks applied"); + + eprintln!(); + eprintln!("[SETUP {}/{}] Verifying Grid operators are ready", next(), total_phases); + + for cluster in CLUSTERS { + let ctx = format!("kind-grid-provider-traffic-{cluster}"); + wait_for_deployment("grid-operator", GRID_SYSTEM_NS, &ctx)?; + eprintln!(" [OK] {cluster}: Grid operator ready"); + } + + eprintln!(); + eprintln!("[SETUP {}/{}] Deploying VCR backends", next(), total_phases); + + for cluster in CLUSTERS { + let ctx = format!("kind-grid-provider-traffic-{cluster}"); + let credential = generate_provider_credential()?; + apply_credential_secret(&ctx, VCR_INFERENCE_CREDENTIAL, &credential)?; + eprintln!(" [OK] {cluster}: vcr-inference-credential created"); + apply_stack(&context.forge_bin, &context.resolved_config, cluster, "vcr-backend")?; + } + eprintln!(" [OK] VCR backends deployed"); + + eprintln!(); + eprintln!("[SETUP {}/{}] Deploying grid-site resources", next(), total_phases); + + for cluster in CLUSTERS { + let site_stack = format!("{cluster}-site"); + apply_stack(&context.forge_bin, &context.resolved_config, cluster, &site_stack)?; + } + eprintln!(" [OK] Grid site resources deployed"); + + eprintln!(); + eprintln!( + "[SETUP {}/{}] Waiting for local overlay ConfigMaps", + next(), + total_phases + ); + + let pre_swim_overlays = wait_for_local_overlays()?; + eprintln!(" [OK] Local overlay ConfigMaps ready"); + + eprintln!(); + eprintln!( + "[SETUP {}/{}] Materializing provider config and installing trust", + next(), + total_phases + ); + + install_provider_boundary()?; + + materialize_provider_config(&pre_swim_overlays, &context.demo_root)?; + eprintln!(" [OK] Provider config materialized, trust installed"); + + eprintln!(); + eprintln!("[SETUP {}/{}] Deploying provider gateways", next(), total_phases); + + for cluster in CLUSTERS { + apply_stack( + &context.forge_bin, + &context.resolved_config, + cluster, + "provider-gateway", + )?; + } + eprintln!(" [OK] Provider gateways deployed"); + + eprintln!(); + eprintln!( + "[SETUP {}/{}] Deploying the single consumer gateway", + next(), + total_phases + ); + + apply_stack( + &context.forge_bin, + &context.resolved_config, + CONSUMER_SITE, + "consumer-gateway", + )?; + eprintln!(" [OK] Consumer gateway deployed in {CONSUMER_SITE}"); + + eprintln!(); + eprintln!( + "[SETUP {}/{}] SWIM discovery and trust authorization", + next(), + total_phases + ); + + configure_swim_peers(&context.forge_bin, &context.resolved_config)?; + authorize_discovered_sites()?; + + eprintln!(); + eprintln!( + "[SETUP {}/{}] Waiting for global overlay convergence", + next(), + total_phases + ); + + let expected = expected_candidates(); + let post_swim_overlays = wait_for_global_overlay_convergence(&expected)?; + let environment_status = wait_for_environment_ready()?; + eprintln!(" [OK] {environment_status}"); + + Ok(OverlayState { + pre_swim: pre_swim_overlays, + post_swim: post_swim_overlays, + }) +} + +// ----------------------------------------------------------------------------- +// Demo Scenarios +// ----------------------------------------------------------------------------- + +/// Run the quick-mode proof scenarios using the assertion framework. +/// +/// Run the six focused provider-traffic proof scenarios. +#[expect( + clippy::too_many_lines, + reason = "The focused demo presents its six proof phases in order." +)] +fn run_quick_scenarios() -> BTreeMap { + let mut results = BTreeMap::new(); + let mut scenario_num: usize = 0; + let mut scenario = || { + scenario_num += 1; + scenario_num + }; + + eprintln!(); + eprintln!("=== QUICK MODE SCENARIOS ==="); + eprintln!(); + + eprintln!("[SCENARIO {}] Verify three provider clusters are healthy", scenario()); + run_and_insert(&mut results, "cluster_health", assert_cluster_health); + + eprintln!(); + eprintln!("[SCENARIO {}] Verify component deployment", scenario()); + run_and_insert(&mut results, "component_deployment", assert_component_deployment); + + eprintln!(); + eprintln!("[SCENARIO {}] Verify SWIM convergence", scenario()); + run_and_insert(&mut results, "swim_convergence", assert_swim_convergence); + + eprintln!(); + eprintln!("[SCENARIO {}] Verify site auto-discovery", scenario()); + run_and_insert(&mut results, "site_auto_discovery", assert_site_auto_discovery); + + eprintln!(); + eprintln!("[SCENARIO {}] Verify overlay acceptance", scenario()); + run_and_insert(&mut results, "overlay_acceptance", assert_overlay_acceptance); + + eprintln!(); + eprintln!(); + eprintln!( + "[SCENARIO {}] Prove equal round-robin across distinct provider gateways", + scenario() + ); + run_and_insert( + &mut results, + "provider_gateway_round_robin", + assert_provider_gateway_round_robin, + ); + + results +} + +/// Send serial, unbound requests through one consumer gateway and verify that +/// the active no-metrics round-robin picker distributes them across the +/// distinct provider gateways. This is intentionally a request-path proof: +/// the request itself is the only source of the attribution counts. +#[expect( + clippy::too_many_lines, + clippy::unnecessary_wraps, + reason = "The assertion framework requires a fallible, named traffic proof boundary." +)] +fn assert_provider_gateway_round_robin() -> AssertionResult { + let start = Instant::now(); + let context = "kind-grid-provider-traffic-provider-a"; + let mut counts: BTreeMap = BTreeMap::new(); + let mut sequence = Vec::new(); + let mut failures = Vec::new(); + let mut overlay_changes = Vec::new(); + + let readiness = match wait_for_round_robin_readiness() { + Ok(facts) => facts, + Err(error) => { + return Ok(proof_failure( + &format!("round-robin readiness gate failed: {error}"), + BTreeMap::from([(String::from("readiness_error"), serde_json::json!(error.to_string()))]), + start.elapsed(), + )); + }, + }; + let baseline_overlay = match read_cluster_overlay("provider-a") { + Ok(overlay) => overlay, + Err(error) => { + return Ok(proof_failure( + &format!("could not capture baseline overlay after readiness: {error}"), + BTreeMap::new(), + start.elapsed(), + )); + }, + }; + + for request_number in 1..=60 { + if let Ok(current_overlay) = read_cluster_overlay("provider-a") + && (current_overlay.resource_version != baseline_overlay.resource_version + || current_overlay.semantic_revision != baseline_overlay.semantic_revision) + { + overlay_changes.push(serde_json::json!({ + "request": request_number, + "resource_version": current_overlay.resource_version, + "semantic_revision": current_overlay.semantic_revision, + })); + } + let request_label = format!("provider-traffic-rr-{request_number:03}"); + let output = run_curl_probe( + context, + &format!("rr-{request_number}"), + &[ + "curl", + "--fail-with-body", + "--include", + "--silent", + "--show-error", + "--header", + "Content-Type: application/json", + "--header", + "Authorization: Bearer consumer-token", + "--data", + r#"{"model":"Qwen/Qwen3-0.6B","messages":[{"role":"user","content":"provider traffic proof"}],"max_tokens":8}"#, + "http://consumer-gateway.grid-system.svc.cluster.local:8080/v1/chat/completions", + ], + ); + + match output { + Ok(output) if output.status.success() => { + let provider = response_header(&output.stdout, "x-grid-combined-provider-gateway") + .or_else(|| response_header(&output.stdout, "x-grid-provider-traffic-provider-gateway")) + .or_else(|| response_header(&output.stdout, "x-grid-provider-gateway")); + if let Some(provider) = provider { + *counts.entry(provider.clone()).or_default() += 1; + sequence.push(provider); + } else { + failures.push(format!("{request_label}: provider attribution header missing")); + } + }, + Ok(output) => failures.push(format!( + "{request_label}: HTTP probe failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + )), + Err(error) => failures.push(format!("{request_label}: probe execution failed: {error}")), + } + } + + let canonical = ["provider-a", "provider-b", "provider-c"]; + let exact_counts = canonical + .iter() + .all(|name| counts.get(*name).copied().unwrap_or(0) == 20); + let cycle_start = sequence + .first() + .and_then(|first| canonical.iter().position(|name| *name == first)); + let repeating_cycle = sequence.len() == 60 + && cycle_start.is_some_and(|start_index| { + sequence.iter().enumerate().all(|(index, provider)| { + canonical + .get((start_index + index) % canonical.len()) + .is_some_and(|expected| *expected == provider.as_str()) + }) + }); + let balanced = exact_counts && repeating_cycle && failures.is_empty(); + + let mut facts = BTreeMap::new(); + facts.insert("request_count".to_owned(), serde_json::json!(sequence.len())); + facts.insert("provider_counts".to_owned(), serde_json::json!(counts)); + facts.insert("ordered_provider_sequence".to_owned(), serde_json::json!(sequence)); + facts.insert("cycle_start_provider".to_owned(), serde_json::json!(sequence.first())); + facts.insert("exact_20_each".to_owned(), serde_json::json!(exact_counts)); + facts.insert( + "repeating_three_provider_cycle".to_owned(), + serde_json::json!(repeating_cycle), + ); + facts.insert("failures".to_owned(), serde_json::json!(failures)); + facts.insert("selection_policy".to_owned(), serde_json::json!("roundRobin")); + facts.insert("scoring_strategy".to_owned(), serde_json::json!("noMetrics")); + facts.insert("readiness".to_owned(), serde_json::json!(readiness)); + facts.insert( + "baseline_resource_version".to_owned(), + serde_json::json!(baseline_overlay.resource_version), + ); + facts.insert( + "baseline_semantic_revision".to_owned(), + serde_json::json!(baseline_overlay.semantic_revision), + ); + let overlay_stable = overlay_changes.is_empty(); + facts.insert( + "overlay_changes_during_requests".to_owned(), + serde_json::json!(overlay_changes), + ); + + if balanced && overlay_stable { + Ok(proof_success( + "60 serial requests distributed evenly across three distinct provider gateways", + facts, + start.elapsed(), + )) + } else { + Ok(proof_failure( + &format!( + "provider gateway distribution or overlay stability failed: counts={counts:?}, overlay_changes={overlay_changes:?}" + ), + facts, + start.elapsed(), + )) + } +} + +/// Run one proof assertion and retain failures as structured evidence. +fn run_and_insert(results: &mut BTreeMap, name: &str, assertion_fn: fn() -> AssertionResult) { + match run_assertion(name, assertion_fn) { + Ok(proof) => { + results.insert(name.to_owned(), proof); + }, + Err(error) => { + eprintln!(" [X] {name} failed: {error}"); + results.insert( + name.to_owned(), + proof_failure(&format!("{name} failed: {error}"), BTreeMap::new(), Duration::ZERO), + ); + }, + } +} + +/// Configure SWIM peer discovery by updating each operator with peer seed addresses. +fn configure_swim_peers(_forge_bin: &Path, _resolved_config: &Path) -> Result<(), Box> { + let mut swim_ips = BTreeMap::new(); + + for cluster in CLUSTERS { + let ip = read_swim_lb_ip(cluster)?; + eprintln!(" {cluster}: SWIM LB IP = {ip}"); + swim_ips.insert(*cluster, ip); + } + + for cluster in CLUSTERS { + let this_ip = swim_ips + .get(cluster) + .ok_or_else(|| format!("{cluster}: missing SWIM IP"))?; + let mut peer_parts = Vec::new(); + for c in CLUSTERS { + if *c != *cluster { + let ip = swim_ips.get(c).ok_or_else(|| format!("{c}: missing SWIM IP"))?; + peer_parts.push(format!("{ip}:7946")); + } + } + let peer_seeds = peer_parts.join(","); + update_operator_swim_config(cluster, this_ip, &peer_seeds)?; + } + + for cluster in CLUSTERS { + let ctx = format!("kind-grid-provider-traffic-{cluster}"); + wait_for_deployment("grid-operator", GRID_SYSTEM_NS, &ctx)?; + eprintln!(" [OK] {cluster}: operator restarted with SWIM config"); + } + + Ok(()) +} + +/// Read the SWIM `LoadBalancer` IP for a cluster directly from Kubernetes. +fn read_swim_lb_ip(cluster: &str) -> Result> { + let context = format!("kind-grid-provider-traffic-{cluster}"); + let output = Command::new("kubectl") + .args([ + "--context", + &context, + "-n", + GRID_SYSTEM_NS, + "get", + "svc", + "grid-operator-swim", + "-o", + "jsonpath={.status.loadBalancer.ingress[0].ip}", + ]) + .output()?; + + if !output.status.success() { + return Err(format!( + "{cluster}: cannot read SWIM service: {}", + String::from_utf8_lossy(&output.stderr).trim() + ) + .into()); + } + + let ip = String::from_utf8(output.stdout)?.trim().to_owned(); + if ip.is_empty() { + return Err(format!("{cluster}: SWIM LoadBalancer has no ingress IP").into()); + } + + Ok(ip) +} + +/// Update a single operator's SWIM configuration with peer addresses. +#[expect( + clippy::too_many_lines, + reason = "The Helm upgrade carries the bounded SWIM configuration contract." +)] +fn update_operator_swim_config( + cluster: &str, + advertise_ip: &str, + seeds: &str, +) -> Result<(), Box> { + let context = format!("kind-grid-provider-traffic-{cluster}"); + + let operator_image = + std::env::var("GRID_XTASK_OPERATOR_IMAGE").unwrap_or_else(|_| "grid-operator:provider-traffic-demo".to_owned()); + let image_pull_policy = std::env::var("GRID_XTASK_IMAGE_PULL_POLICY").unwrap_or_else(|_| "Never".to_owned()); + let (operator_repo, operator_tag) = parse_image_ref(&operator_image); + + let seeds_escaped = seeds.replace(',', "\\,"); + + let upgrade_output = Command::new("helm") + .args([ + "upgrade", + "grid-operator", + "charts/grid-operator", + "--version", + "0.1.0", + "--namespace", + "grid-system", + "--kube-context", + &context, + "--reuse-values", + "--set", + &format!("image.repository={operator_repo}"), + "--set", + &format!("image.tag={operator_tag}"), + "--set", + &format!("image.pullPolicy={image_pull_policy}"), + "--set", + &format!("swim.siteName={cluster}"), + "--set", + &format!("swim.advertiseAddress={advertise_ip}:7946"), + "--set", + &format!("swim.seeds={seeds_escaped}"), + "--set", + "swim.service.enabled=true", + "--set", + "swim.service.type=LoadBalancer", + "--set", + &format!("gateway.serviceName={PROVIDER_GATEWAY_SERVICE}"), + "--set-string", + &format!("gateway.port={PROVIDER_GATEWAY_PORT}"), + ]) + .output()?; + + if !upgrade_output.status.success() { + return Err(format!( + "Failed to update SWIM config for {cluster}: {}", + String::from_utf8_lossy(&upgrade_output.stderr) + ) + .into()); + } + + eprintln!(" [OK] {cluster}: SWIM peers configured"); + + assert_operator_gateway_env(&context, cluster)?; + + Ok(()) +} + +/// Post-upgrade assertion: the operator deployment must reflect the expected +/// gateway discovery contract after every Helm upgrade. +fn assert_operator_gateway_env(context: &str, cluster: &str) -> Result<(), Box> { + let output = Command::new("kubectl") + .args([ + "--context", + context, + "-n", + GRID_SYSTEM_NS, + "get", + "deployment/grid-operator", + "-o", + "jsonpath={.spec.template.spec.containers[0].env}", + ]) + .output()?; + + let env_json = String::from_utf8_lossy(&output.stdout); + + let has_service = env_json.contains(&format!( + "\"name\":\"GRID_GATEWAY_SERVICE_NAME\",\"value\":\"{PROVIDER_GATEWAY_SERVICE}\"" + )); + let has_port = env_json.contains(&format!( + "\"name\":\"GRID_GATEWAY_PORT\",\"value\":\"{PROVIDER_GATEWAY_PORT}\"" + )); + + if !has_service || !has_port { + return Err(format!( + "{cluster}: operator gateway env mismatch after helm upgrade \ + (expected GRID_GATEWAY_SERVICE_NAME={PROVIDER_GATEWAY_SERVICE}, \ + GRID_GATEWAY_PORT={PROVIDER_GATEWAY_PORT}); got: {env_json}" + ) + .into()); + } + + Ok(()) +} + +/// Tear down only the provider-traffic Forge environment. +fn teardown_environment(context: &ProviderTrafficContext) -> Result<(), Box> { + eprintln!(); + eprintln!("=== TEARDOWN ==="); + + let status = Command::new(&context.forge_bin) + .args(["down", "--config"]) + .arg(&context.resolved_config) + .status()?; + + if !status.success() { + return Err("failed to tear down provider-traffic environment".into()); + } + + eprintln!(" [OK] Environment torn down successfully"); + Ok(()) +} + +/// Run the focused provider-traffic demo. +#[expect( + clippy::too_many_lines, + reason = "The public demo entrypoint keeps setup, proof, evidence, and teardown visible." +)] +pub(crate) fn run(forge_config: &Path, options: &GlbDemoOptions) -> Result<(), Box> { + if options.mode() != DemoMode::Quick { + return Err("provider-traffic supports only the focused quick proof".into()); + } + let mode = DemoMode::Quick; + let run_id = format_utc_timestamp(); + let wall_start = Instant::now(); + let _started_at = format_utc_iso(); + + let evidence_dir = resolve_evidence_dir(forge_config, options, &run_id)?; + fs::create_dir_all(&evidence_dir)?; + + let setup_ctx = prepare_setup(forge_config); + let mut teardown_success = false; + let mut run_error = None; + let mut overlay_state = OverlayState::default(); + let mut images = BTreeMap::new(); + + let proof_results = match &setup_ctx { + Ok(context) => { + eprintln!("{OUTPUT_RULE}"); + eprintln!("Grid Provider Traffic Demo"); + eprintln!("Mode: {}", if mode == DemoMode::Quick { "quick" } else { "full" }); + eprintln!("Config: {}", forge_config.display()); + eprintln!("{OUTPUT_RULE}"); + + match deploy_setup(context) { + Ok(state) => { + overlay_state = state; + images = match collect_image_evidence() { + Ok(images) => images, + Err(error) => { + run_error = Some(format!("image evidence collection failed: {error}")); + BTreeMap::new() + }, + }; + eprintln!(); + eprintln!("{OUTPUT_RULE}"); + eprintln!("ENVIRONMENT READY - Starting proof scenarios"); + eprintln!("{OUTPUT_RULE}"); + + let scenario_results = run_quick_scenarios(); + + let failed_proofs: Vec<&str> = scenario_results + .iter() + .filter_map(|(name, proof)| (!proof.success).then_some(name.as_str())) + .collect(); + if !failed_proofs.is_empty() { + run_error = Some(format!("runtime proofs failed: {}", failed_proofs.join(", "))); + } + + // Teardown if requested + if options.teardown && (run_error.is_none() || !options.keep_on_failure) { + match teardown_environment(context) { + Ok(()) => teardown_success = true, + Err(error) => { + eprintln!("[WARN] Teardown failed: {error}"); + run_error = Some(match run_error { + Some(previous) => format!("{previous}; teardown failed: {error}"), + None => format!("teardown failed: {error}"), + }); + }, + } + } + + scenario_results + }, + Err(e) => { + eprintln!("[FAIL] Environment setup failed: {e}"); + run_error = Some(format!("environment setup failed: {e}")); + + if options.teardown && !options.keep_on_failure { + if let Err(cleanup_err) = teardown_environment(context) { + eprintln!("[WARN] Cleanup after setup failure also failed: {cleanup_err}"); + run_error = Some(format!("environment setup failed: {e}; cleanup failed: {cleanup_err}")); + } else { + teardown_success = true; + } + } + + BTreeMap::new() + }, + } + }, + Err(e) => { + eprintln!("[FAIL] Setup preparation failed: {e}"); + run_error = Some(format!("setup preparation failed: {e}")); + BTreeMap::new() + }, + }; + + let evidence = Evidence { + schema_version: EVIDENCE_SCHEMA_VERSION.to_owned(), + mode: if mode == DemoMode::Quick { "quick" } else { "full" }.to_owned(), + topology: "provider-traffic".to_owned(), + clusters: CLUSTERS.iter().map(|&s| s.to_owned()).collect(), + proof_results, + images, + overlay_state, + cluster_health: Vec::new(), // Will be populated during runtime assertions + components: Vec::new(), // Will be populated during runtime assertions + swim_membership: Vec::new(), // Will be populated during runtime assertions + provider_responses: Vec::new(), // Will be populated during runtime assertions + security_results: Vec::new(), // Will be populated during runtime assertions + teardown_success, + }; + + // Write evidence + let evidence_file = evidence_dir.join("results.json"); + let evidence_json = serde_json::to_string_pretty(&evidence)?; + fs::write(&evidence_file, evidence_json)?; + + eprintln!(); + eprintln!("{OUTPUT_RULE}"); + eprintln!("Demo completed in {:.1}s", wall_start.elapsed().as_secs_f64()); + eprintln!("Evidence: {}", evidence_file.display()); + eprintln!("{OUTPUT_RULE}"); + + match run_error { + Some(error) => Err(error.into()), + None => Ok(()), + } +} + +/// Collect actual image evidence from the deployed clusters. +#[expect( + clippy::too_many_lines, + reason = "Evidence collection queries the bounded set of deployed component images." +)] +fn collect_image_evidence() -> Result, Box> { + let mut images = BTreeMap::new(); + + for cluster in CLUSTERS { + let context = format!("kind-grid-provider-traffic-{cluster}"); + + // Get grid operator image + let output = Command::new("kubectl") + .args([ + "get", + "deployment/grid-operator", + "--context", + &context, + "-n", + "grid-system", + "-o", + "jsonpath={.spec.template.spec.containers[0].image}", + ]) + .output()?; + + if output.status.success() { + let operator_image = String::from_utf8_lossy(&output.stdout).trim().to_owned(); + images.insert(format!("{cluster}_operator"), operator_image); + } + + // Get consumer gateway image + let output = Command::new("kubectl") + .args([ + "get", + "deployment/consumer-gateway", + "--context", + &context, + "-n", + "grid-system", + "-o", + "jsonpath={.spec.template.spec.containers[0].image}", + ]) + .output()?; + + if output.status.success() { + let consumer_image = String::from_utf8_lossy(&output.stdout).trim().to_owned(); + images.insert(format!("{cluster}_consumer_gateway"), consumer_image); + } + + // Get provider gateway image + let output = Command::new("kubectl") + .args([ + "get", + "deployment/provider-gateway", + "--context", + &context, + "-n", + "grid-system", + "-o", + "jsonpath={.spec.template.spec.containers[0].image}", + ]) + .output()?; + + if output.status.success() { + let provider_image = String::from_utf8_lossy(&output.stdout).trim().to_owned(); + images.insert(format!("{cluster}_provider_gateway"), provider_image); + } + + // Get VCR inference image + let output = Command::new("kubectl") + .args([ + "get", + &format!("deployment/vcr-inference-{cluster}"), + "--context", + &context, + "-n", + "grid-system", + "-o", + "jsonpath={.spec.template.spec.containers[0].image}", + ]) + .output()?; + + if output.status.success() { + let mock_image = String::from_utf8_lossy(&output.stdout).trim().to_owned(); + images.insert(format!("{cluster}_vcr_inference"), mock_image); + } + } + + Ok(images) +} + +#[cfg(test)] +mod tests { + use std::{collections::BTreeMap, time::Duration}; + + use super::*; + + #[test] + fn test_proof_success_creation() { + let mut facts = BTreeMap::new(); + facts.insert("cluster_count".to_owned(), serde_json::Value::Number(3.into())); + facts.insert("all_healthy".to_owned(), serde_json::Value::Bool(true)); + + let proof = proof_success("Test success", facts.clone(), Duration::from_millis(100)); + + assert!(proof.success); + assert_eq!(proof.reason, "Test success"); + assert_eq!(proof.duration_ms, 100); + assert_eq!(proof.observed_facts.len(), 2); + assert_eq!( + proof.observed_facts.get("all_healthy"), + Some(&serde_json::Value::Bool(true)) + ); + } + + #[test] + fn test_proof_failure_creation() { + let mut facts = BTreeMap::new(); + facts.insert("error_code".to_owned(), serde_json::Value::Number(500.into())); + + let proof = proof_failure("Test failure", facts.clone(), Duration::from_millis(50)); + + assert!(!proof.success); + assert_eq!(proof.reason, "Test failure"); + assert_eq!(proof.duration_ms, 50); + assert_eq!(proof.observed_facts.len(), 1); + assert_eq!( + proof.observed_facts.get("error_code"), + Some(&serde_json::Value::Number(500.into())) + ); + } + + #[test] + fn test_assertion_result_error_handling() { + let assertion_fn = || -> AssertionResult { Err("Simulated assertion failure".into()) }; + + let result = run_assertion("test_assertion", assertion_fn); + assert!(result.is_err()); + + let Err(error) = result else { + std::process::abort(); + }; + let error_msg = error.to_string(); + assert!(error_msg.contains("Assertion test_assertion failed")); + assert!(error_msg.contains("Simulated assertion failure")); + } + + #[test] + #[expect(clippy::too_many_lines, reason = "This test exercises the complete evidence schema.")] + fn test_evidence_serialization() { + let evidence = Evidence { + schema_version: "test".to_owned(), + mode: "quick".to_owned(), + topology: "provider-traffic".to_owned(), + clusters: vec![ + "provider-a".to_owned(), + "provider-b".to_owned(), + "provider-c".to_owned(), + ], + proof_results: BTreeMap::new(), + images: BTreeMap::new(), + overlay_state: OverlayState::default(), + cluster_health: vec![ClusterHealth { + name: "provider-a".to_owned(), + healthy: true, + api_response_ms: Some(100), + ready_nodes: 1, + }], + components: vec![ComponentStatus { + name: "provider-a-grid-operator".to_owned(), + namespace: "grid-system".to_owned(), + ready_replicas: 1, + desired_replicas: 1, + ready: true, + }], + swim_membership: vec![SwimMembership { + site: "provider-a".to_owned(), + local_node: "provider-a-operator".to_owned(), + peers: vec!["provider-b-operator".to_owned(), "provider-c-operator".to_owned()], + converged: true, + }], + provider_responses: vec![], + security_results: vec![], + teardown_success: true, + }; + + let Ok(json) = serde_json::to_string(&evidence) else { + std::process::abort(); + }; + assert!(json.contains("\"schema_version\":\"test\"")); + assert!(json.contains("\"topology\":\"provider-traffic\"")); + assert!(json.contains("\"healthy\":true")); + assert!(json.contains("\"ready_replicas\":1")); + assert!(json.contains("\"converged\":true")); + + // Verify deserialization + let Ok(_deserialized) = serde_json::from_str::(&json) else { + std::process::abort(); + }; + } + + #[test] + fn test_proof_count_validation() { + let names = [ + "cluster_health", + "component_deployment", + "swim_convergence", + "site_auto_discovery", + "overlay_acceptance", + "provider_gateway_round_robin", + ]; + assert_eq!(names.len(), 6); + assert_eq!(names[0], "cluster_health"); + assert_eq!(names[5], "provider_gateway_round_robin"); + assert_eq!(CLUSTERS.len(), 3); + assert_eq!(CLUSTERS, &["provider-a", "provider-b", "provider-c"]); + } + + #[test] + fn test_evidence_schema_version() { + assert_eq!(EVIDENCE_SCHEMA_VERSION, "1"); + } + + #[test] + fn curl_pod_overrides_meets_restricted_pod_security() { + let json = curl_pod_overrides("test-probe", &["curl", "--fail", "http://example.test"]); + let actual: serde_json::Value = serde_json::from_str(&json).unwrap_or_else(|_| std::process::abort()); + assert_eq!( + actual, + serde_json::json!({ + "spec": { + "automountServiceAccountToken": false, + "securityContext": { + "runAsNonRoot": true, + "seccompProfile": { "type": "RuntimeDefault" } + }, + "containers": [{ + "name": "test-probe", + "image": "curlimages/curl:8.12.1", + "command": ["curl"], + "args": ["--fail", "http://example.test"], + "securityContext": { + "runAsUser": 100, + "allowPrivilegeEscalation": false, + "readOnlyRootFilesystem": true, + "capabilities": { "drop": ["ALL"] } + } + }] + } + }) + ); + } + + #[test] + fn provider_traffic_constants_describe_focused_topology() { + assert_eq!(CLUSTERS, &["provider-a", "provider-b", "provider-c"]); + assert_eq!(CONSUMER_SITE, "provider-a"); + assert_eq!(EVIDENCE_SCHEMA_VERSION, "1"); + } +}