feat(router): Bind graph-specific configuration to selected supergraphs

.changeset/add_persisted_documents_support_to_websockets.md

@@ -0,0 +1,9 @@
+---
+hive-router: patch
+---
+
+# Add persisted documents support to WebSocket operations
+
+WebSocket `subscribe` payloads can now omit the GraphQL query and provide a persisted document ID through their extensions. The router resolves the document before parsing, validation, and execution.
+
+Persisted-document extraction, ID enforcement, resolution, metrics, and missing-ID logging use the supergraph selected for the WebSocket connection. This prevents an operation from resolving a document from one supergraph's manifest and executing it against another supergraph.

.changeset/isolate-plugin-supergraph-options.md

@@ -0,0 +1,30 @@
+---
+hive-router: minor
+---
+
+# Isolate plugin-selected supergraph configuration
+
+Plugin-selected supergraphs no longer inherit graph-bound settings from the router's configured supergraph. Requests, WebSocket connections, persisted-document resolution, subgraph execution, usage reports, and Hive traces now use the options attached to the selected `Supergraph` snapshot.
+
+Persisted-document reloaders and Hive usage agents now use separate background-task groups scoped to the selected supergraph runtime. Cancelling a runtime waits for any active Hive flush and explicitly flushes the remaining report buffer before removing its worker. Router shutdown also waits for these graceful background tasks to finish.
+
+## Migration
+
+Configured supergraphs require no YAML changes. The router continues deriving their graph-bound options from the existing configuration.
+
+Plugins that construct supergraph variants must import `SupergraphOptions`, provide the graph-bound settings each variant needs, and retain the owner while it remains selectable:
+
+```rust
+use std::sync::Arc;
+
+use hive_router::plugins::hooks::on_supergraph_load::{Supergraph, SupergraphOptions};
+
+let mut options = SupergraphOptions::default();
+options.traffic_shaping.all.forward_operation_name = true;
+options.error_masking.redacted_error_message = "Variant error".to_string();
+options.hive_target = Some("organization/project/variant".to_string());
+
+let variant = Arc::new(Supergraph::from_sdl(sdl, options)?);
+```
+
+Move any existing `QueryPlannerOptions` into `SupergraphOptions::query_planner`. Also copy every graph-specific setting the variant previously inherited from router configuration, including subgraph traffic shaping, URL overrides, headers, override labels, demand control, subscription transports, error masking, persisted documents, and its Hive target. Omitted fields use `SupergraphOptions::default()` and no longer inherit values from the configured supergraph.

.changeset/run_graphql_params_hooks_for_websockets.md

@@ -0,0 +1,9 @@
+---
+hive-router: patch
+---
+
+# Run GraphQL parameters hooks for WebSocket operations
+
+WebSocket `subscribe` payloads now run the `on_graphql_params` start hook and its registered end callbacks before parsing, validation, and execution. The start hook receives the synthetic WebSocket operation request and the already-decoded parameters in `OnGraphQLParamsStartHookPayload.graphql_params`.
+
+When a hook ends preparation with an early response, the router sends the response's GraphQL body as a WebSocket `next` message followed by `complete`. HTTP-only response metadata, including its status and headers, cannot be represented by the GraphQL over WebSocket protocol and is not forwarded.

.changeset/stop-usage-flush-cancellation-race.md

@@ -0,0 +1,7 @@
+---
+hive-console-sdk: patch
+---
+
+# Stop cancellation from interrupting an active usage flush
+
+The usage agent now observes cancellation while waiting for the next flush interval instead of checking only after the full interval has elapsed. Cancellation remains pending while an active flush completes, preventing a drained report batch from being lost when its send future is interrupted.

.changeset/supergraph-options-api.md

@@ -0,0 +1,44 @@
+---
+hive-router-plan-executor: major
+hive-router: patch
+---
+
+# Bind graph-specific configuration to `Supergraph`
+
+`Supergraph::from_sdl` and `Supergraph::from_document` now accept `SupergraphOptions` instead of `QueryPlannerOptions`. The immutable options snapshot includes planner, executor, subgraph subscription, persisted-document, error-masking, and Hive target settings.
+
+## Migration
+
+Import `SupergraphOptions` with `Supergraph`:
+
+```rust
+use hive_router::plugins::hooks::on_supergraph_load::{Supergraph, SupergraphOptions};
+```
+
+Replace the query-planner options argument with a complete supergraph options value:
+
+```diff
+-use hive_router::plugins::hooks::on_supergraph_load::Supergraph;
++use hive_router::plugins::hooks::on_supergraph_load::{Supergraph, SupergraphOptions};
+ use hive_router::query_planner::planner::QueryPlannerOptions;
+
+ let query_planner = QueryPlannerOptions {
+     experimental_abstract_type_folding: true,
+ };
+-let supergraph = Supergraph::from_sdl(sdl, query_planner)?;
++let supergraph = Supergraph::from_sdl(
++    sdl,
++    SupergraphOptions {
++        query_planner,
++        ..SupergraphOptions::default()
++    },
++)?;
+```
+
+Callers that used `Default::default()` can migrate directly:
+
+```rust
+use hive_router::plugins::hooks::on_supergraph_load::{Supergraph, SupergraphOptions};
+
+let supergraph = Supergraph::from_sdl(sdl, SupergraphOptions::default())?;
+```

bin/router/Cargo.toml

@@ -103,6 +103,7 @@ object_store = { workspace = true }
 [dev-dependencies]
 criterion = { workspace = true }
 insta = { workspace = true }
+mockito = { workspace = true }
 
 [[bench]]
 name = "persisted_documents_matcher_benches"

bin/router/src/error.rs

@@ -2,9 +2,9 @@ use hive_router_config::RouterConfigError;
 use hive_router_plan_executor::executors::error::TlsCertificatesError;
 
 use crate::{
-    jwt::jwks_manager::JwksSourceError, pipeline::usage_reporting::UsageReportingError,
-    plugins::registry::PluginRegistryError, schema_state::SupergraphManagerError,
-    shared_state::SharedStateError, storage::error::StorageError, telemetry::TelemetryInitError,
+    jwt::jwks_manager::JwksSourceError, plugins::registry::PluginRegistryError,
+    schema_state::SupergraphManagerError, shared_state::SharedStateError,
+    storage::error::StorageError, telemetry::TelemetryInitError,
 };
 
 #[derive(Debug, thiserror::Error)]
@@ -21,16 +21,12 @@ pub enum RouterInitError {
     HttpServerStartError(std::io::Error),
     #[error(transparent)]
     JwksSourceError(#[from] JwksSourceError),
-    #[error("Usage Reporting - {0}")]
-    UsageReportingError(#[from] UsageReportingError),
     #[error(transparent)]
     SharedStateError(#[from] SharedStateError),
     #[error(transparent)]
     TelemetryInitError(#[from] TelemetryInitError),
     #[error(transparent)]
     PluginRegistryError(#[from] PluginRegistryError),
-    #[error("Persisted documents endpoint incompatible: {0}")]
-    PersistedDocumentsEndpointIncompatible(String),
     #[error("Endpoints of '{endpoint_name_one}' and '{endpoint_name_two}' cannot both use the same endpoint: {endpoint}")]
     EndpointConflict {
         endpoint_name_one: String,

bin/router/src/http_utils/probes.rs

@@ -12,7 +12,7 @@ pub async fn readiness_check_handler(
     req: HttpRequest,
     schema_state: web::types::State<Arc<SchemaState>>,
 ) -> impl Responder {
-    if schema_state.is_ready(&req) {
+    if schema_state.is_ready(&req).await {
         web::HttpResponse::Ok()
     } else {
         web::HttpResponse::ServiceUnavailable()

bin/router/src/lib.rs

@@ -39,15 +39,15 @@ use crate::{
         header::ResponseMode,
         http_callback::handler,
         long_lived_client_limit::LongLivedClientLimitService,
-        persisted_documents::PersistedDocumentsRuntime,
+        persisted_documents::PersistedDocumentsBackgroundTasks,
         request_extensions::{
             read_graphql_operation_metric_identity, read_graphql_response_metric_status,
             write_graphql_response_metric_status,
         },
         request_identifiers::RequestIdentifiersService,
         request_summary::RequestSummaryService,
         timeout::handle_timeout,
-        usage_reporting::init_hive_usage_agent,
+        usage_reporting::HiveUsageReportingBackgroundTasks,
         validation::{
             max_aliases_rule::MaxAliasesRule, max_depth_rule::MaxDepthRule,
             max_directives_rule::MaxDirectivesRule,
@@ -515,7 +515,7 @@ pub async fn router_entrypoint(plugin_registry: PluginRegistry) -> Result<(), Ro
     .map_err(RouterInitError::HttpServerStartError);
 
     info!(target: targets::CORE, "router stopped, clearing background tasks");
-    bg_tasks_manager.shutdown();
+    bg_tasks_manager.graceful_shutdown().await;
     telemetry.graceful_shutdown().await;
 
     invoke_shutdown_hooks(&shared_state_clone).await;
@@ -544,12 +544,6 @@ pub async fn configure_app_from_config(
         false => None,
     };
 
-    let hive_usage_agent = match router_config.telemetry.hive.as_ref() {
-        Some(hive_config) if hive_config.usage_reporting.enabled => {
-            Some(init_hive_usage_agent(bg_tasks_manager, hive_config)?)
-        }
-        _ => None,
-    };
     let plugins_arc = plugin_registry.initialize_plugins(&router_config, bg_tasks_manager)?;
 
     let active_subscriptions =
@@ -558,13 +552,22 @@ pub async fn configure_app_from_config(
     let router_config_arc = Arc::new(router_config);
     let telemetry_context_arc = Arc::new(telemetry_context);
 
+    let (persisted_documents_background_tasks, persisted_documents_background_task) =
+        PersistedDocumentsBackgroundTasks::new();
+    bg_tasks_manager.register_graceful_task(persisted_documents_background_task);
+    let (hive_usage_reporting_background_tasks, hive_usage_reporting_background_task) =
+        HiveUsageReportingBackgroundTasks::new();
+    bg_tasks_manager.register_graceful_task(hive_usage_reporting_background_task);
+
     let schema_state = SchemaState::new_from_config(
         bg_tasks_manager,
         telemetry_context_arc.clone(),
         router_config_arc.clone(),
         plugins_arc.clone(),
         active_subscriptions.clone(),
         storage_manager.clone(),
+        persisted_documents_background_tasks,
+        hive_usage_reporting_background_tasks,
     )
     .await?;
     let schema_state_arc = Arc::new(schema_state);
@@ -585,32 +588,10 @@ pub async fn configure_app_from_config(
             config: max_aliases_config.clone(),
         }));
     }
-    let persisted_documents_runtime = PersistedDocumentsRuntime::init(
-        &router_config_arc.persisted_documents,
-        &router_config_arc.http.graphql_endpoint,
-        bg_tasks_manager,
-        &storage_manager,
-    )
-    .await
-    .map_err(|err| crate::shared_state::SharedStateError::PersistedDocuments(Box::new(err)))?;
-
-    if !persisted_documents_runtime
-        .supports_graphql_endpoint(&router_config_arc.http.graphql_endpoint)
-    {
-        // url_path_param extractor depends on path segments relative to graphql endpoint.
-        // Root endpoint would make all routes ambiguous for persisted-document extraction.
-        // Even /health could be treated as a graphql request with document id == "health".
-        return Err(RouterInitError::PersistedDocumentsEndpointIncompatible(
-            "http.graphql_endpoint='/' is not allowed when persisted_documents.selectors contains type=url_path_param. Use a non-root endpoint like '/graphql'.".to_string(),
-        ));
-    }
-
     let metrics_enabled = router_config_arc.telemetry.metrics.is_enabled();
     let shared_state = Arc::new(RouterSharedState::new(
         router_config_arc,
-        persisted_documents_runtime,
         jwt_runtime,
-        hive_usage_agent,
         validation_plan,
         telemetry_context_arc.clone(),
         plugins_arc,

bin/router/src/pipeline/execution.rs

@@ -127,7 +127,7 @@ pub async fn execute_plan<'exec>(
                 .operation_for_plan
                 .clone(),
             projection_plan: planned_request.normalized_payload.projection_plan.clone(),
-            headers_plan: app_state.headers_plan.clone(),
+            headers_plan: supergraph.runtime.headers_plan.clone(),
             extensions_plan: app_state.extensions_plan.clone(),
             variable_values: planned_request.variable_payload.clone(),
             extensions,
@@ -149,7 +149,7 @@ pub async fn execute_plan<'exec>(
                 operation_name,
             ),
             response_header_sink,
-            error_masking_runtime: app_state.error_masking.clone(),
+            error_masking_runtime: supergraph.runtime.error_masking.clone(),
         })
         .await?;