refactor(report-context): extract UPSTREAM_IS_STREAM_KEY constant

The "upstream_is_stream" JSON key flows from the AI execution report
context producer (aether-ai-serving::report_context) through several
consumers — usage runtime metadata copy/move, gateway watchdog, sync
execution decision, observability handlers, and the per-driver usage
repositories. Each site spelled the key as a bare string literal, so a
producer-side rename would silently degrade every consumer to its
fallback (typically assuming streaming) with no compile-time signal.

Introduce a single pub const UPSTREAM_IS_STREAM_KEY in
aether-ai-formats (the lowest crate every consumer already depends on),
re-export from the crate root, and route producer + all map-style
consumers through it. The change is purely a string-literal → constant
swap; behaviour is identical.

Sites left as literals (intentional):
- `json!({"upstream_is_stream": ...})` macro keys, which must be string
  literals at the macro layer; these are also API-response payload
  field names (an external contract that should not silently track
  internal report-context renames).
- SQL column accessors (`try_get::<...>("upstream_is_stream")`), which
  refer to the database schema column, not the JSON key.
- Test fixtures and assertions, which validate the on-the-wire contract
  and should keep verifying the actual string.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

apps/aether-gateway/src/execution_runtime/sync/execution.rs

@@ -3,6 +3,7 @@ use std::io::Error as IoError;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+use aether_ai_serving::UPSTREAM_IS_STREAM_KEY;
 use aether_contracts::{
     ExecutionError, ExecutionErrorKind, ExecutionPhase, ExecutionPlan, ExecutionResult,
     ExecutionTelemetry,
@@ -1005,7 +1006,7 @@ fn resolve_openai_image_sync_total_timeout_ms(plan: &ExecutionPlan) -> u64 {
 
 fn report_context_upstream_is_stream(report_context: Option<&Value>) -> bool {
     report_context
-        .and_then(|value| value.get("upstream_is_stream"))
+        .and_then(|value| value.get(UPSTREAM_IS_STREAM_KEY))
         .and_then(Value::as_bool)
         .unwrap_or(false)
 }

apps/aether-gateway/src/handlers/public/support/user_me_usage.rs

@@ -1,5 +1,6 @@
 use std::collections::{BTreeMap, BTreeSet};
 
+use aether_ai_serving::UPSTREAM_IS_STREAM_KEY;
 use aether_billing::{
     normalize_input_tokens_for_billing, normalize_total_input_context_for_cache_hit_rate,
 };
@@ -314,7 +315,7 @@ fn users_me_usage_upstream_is_stream(item: &StoredRequestUsageAudit) -> bool {
     item.request_metadata
         .as_ref()
         .and_then(serde_json::Value::as_object)
-        .and_then(|metadata| metadata.get("upstream_is_stream"))
+        .and_then(|metadata| metadata.get(UPSTREAM_IS_STREAM_KEY))
         .and_then(serde_json::Value::as_bool)
         .or_else(|| users_me_usage_headers_stream_flag(item.response_headers.as_ref()))
         .or_else(|| users_me_usage_infer_upstream_stream_from_captured_bodies(item))

crates/aether-admin/src/observability/usage.rs

@@ -1,5 +1,6 @@
 use crate::observability::stats::{aggregate_usage_stats, parse_bounded_u32, round_to};
 use aether_ai_formats::api::request_path_implies_stream_request;
+use aether_ai_formats::UPSTREAM_IS_STREAM_KEY;
 use aether_billing::{
     normalize_input_tokens_for_billing, normalize_total_input_context_for_cache_hit_rate,
 };
@@ -1052,7 +1053,7 @@ fn admin_usage_upstream_is_stream(item: &StoredRequestUsageAudit) -> bool {
     item.request_metadata
         .as_ref()
         .and_then(Value::as_object)
-        .and_then(|metadata| metadata.get("upstream_is_stream"))
+        .and_then(|metadata| metadata.get(UPSTREAM_IS_STREAM_KEY))
         .and_then(Value::as_bool)
         .or_else(|| admin_usage_headers_stream_flag(item.response_headers.as_ref()))
         .or_else(|| admin_usage_infer_upstream_stream_from_captured_bodies(item))
@@ -1256,7 +1257,10 @@ pub fn admin_usage_record_json(
         .as_object_mut()
         .expect("admin usage record payload should be an object");
     object.insert("is_stream".to_string(), json!(item.is_stream));
-    object.insert("upstream_is_stream".to_string(), json!(upstream_is_stream));
+    object.insert(
+        UPSTREAM_IS_STREAM_KEY.to_string(),
+        json!(upstream_is_stream),
+    );
     object.insert(
         "client_requested_stream".to_string(),
         json!(client_is_stream),

crates/aether-ai-formats/src/formats/shared/request.rs

@@ -2,6 +2,14 @@ use base64::Engine as _;
 
 use crate::formats::id::api_format_uses_body_stream_field;
 
+/// JSON key under which `upstream_is_stream` is written into the AI execution
+/// report context and propagated into usage metadata. Shared by the producer
+/// (`aether-ai-serving::report_context`) and every downstream consumer so that
+/// renames cannot silently desync them — a string-literal mismatch here would
+/// degrade to default values (e.g. assuming streaming) without any compile-time
+/// signal.
+pub const UPSTREAM_IS_STREAM_KEY: &str = "upstream_is_stream";
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum UpstreamStreamPolicy {
     Auto,

crates/aether-ai-formats/src/formats/shared/response.rs

@@ -3,6 +3,7 @@ use std::collections::BTreeMap;
 use serde_json::Value;
 
 use crate::contracts::core_success_background_report_kind;
+use crate::formats::shared::request::UPSTREAM_IS_STREAM_KEY;
 
 #[derive(Debug, Clone, PartialEq)]
 pub struct LocalSyncReportParts {
@@ -66,7 +67,7 @@ fn should_capture_client_sync_success_body(payload: &LocalSyncReportParts) -> bo
         .report_context
         .as_ref()
         .and_then(Value::as_object)
-        .and_then(|context| context.get("upstream_is_stream"))
+        .and_then(|context| context.get(UPSTREAM_IS_STREAM_KEY))
         .and_then(Value::as_bool)
         .unwrap_or(false)
 }

crates/aether-ai-formats/src/lib.rs

@@ -29,7 +29,7 @@ pub use formats::shared::model_directives::{
 };
 pub use formats::shared::request::{
     endpoint_config_forces_upstream_stream_policy, enforce_request_body_stream_field,
-    resolve_upstream_is_stream_from_endpoint_config,
+    resolve_upstream_is_stream_from_endpoint_config, UPSTREAM_IS_STREAM_KEY,
 };
 pub use protocol::canonical::{
     canonical_request_unknown_block_count, canonical_response_unknown_block_count,

crates/aether-ai-serving/src/lib.rs

@@ -22,6 +22,7 @@ pub mod request_body_diagnostics;
 pub mod runtime_miss;
 pub mod surface_spec;
 
+pub use aether_ai_formats::UPSTREAM_IS_STREAM_KEY;
 pub use aether_pool_core::{
     normalize_enabled_pool_presets, run_pool_scheduler, PoolCandidateFacts, PoolCandidateInput,
     PoolCandidateOrchestration, PoolMemberSignals, PoolRuntimeState, PoolScheduledCandidate,

crates/aether-ai-serving/src/report_context.rs

@@ -1,6 +1,7 @@
 use std::collections::BTreeMap;
 
 use aether_ai_formats::api::ExecutionRuntimeAuthContext;
+use aether_ai_formats::UPSTREAM_IS_STREAM_KEY;
 use aether_scheduler_core::SchedulerRankingOutcome;
 use serde_json::{Map, Value};
 
@@ -140,7 +141,7 @@ pub fn build_ai_execution_report_context(parts: AiExecutionReportContextParts<'_
         Value::Bool(parts.client_requested_stream),
     );
     object.insert(
-        "upstream_is_stream".to_string(),
+        UPSTREAM_IS_STREAM_KEY.to_string(),
         Value::Bool(parts.upstream_is_stream),
     );
     object.insert("has_envelope".to_string(), Value::Bool(parts.has_envelope));

crates/aether-data/src/repository/usage/memory.rs

@@ -2,6 +2,7 @@ use std::collections::BTreeMap;
 use std::sync::Arc;
 use std::sync::RwLock;
 
+use aether_ai_formats::UPSTREAM_IS_STREAM_KEY;
 use aether_data_contracts::repository::usage::{
     parse_usage_body_ref, usage_body_ref, StoredUsageAuditAggregation, StoredUsageAuditSummary,
     StoredUsageBreakdownSummaryRow, StoredUsageCacheAffinityHitSummary,
@@ -952,7 +953,7 @@ fn usage_output_tps_uses_generation_time(item: &StoredRequestUsageAudit) -> bool
     item.request_metadata
         .as_ref()
         .and_then(Value::as_object)
-        .and_then(|metadata| metadata.get("upstream_is_stream"))
+        .and_then(|metadata| metadata.get(UPSTREAM_IS_STREAM_KEY))
         .and_then(Value::as_bool)
         .unwrap_or(item.is_stream)
 }

crates/aether-data/src/repository/usage/mysql.rs

@@ -1,5 +1,6 @@
 use std::collections::{BTreeMap, HashSet};
 
+use aether_ai_formats::UPSTREAM_IS_STREAM_KEY;
 use async_trait::async_trait;
 use sqlx::{mysql::MySqlRow, Row};
 
@@ -881,7 +882,7 @@ fn usage_upstream_is_stream(usage: &UpsertUsageRecord) -> bool {
         .request_metadata
         .as_ref()
         .and_then(serde_json::Value::as_object)
-        .and_then(|metadata| metadata.get("upstream_is_stream"))
+        .and_then(|metadata| metadata.get(UPSTREAM_IS_STREAM_KEY))
         .and_then(serde_json::Value::as_bool)
         .unwrap_or_else(|| usage.is_stream.unwrap_or(false))
 }
@@ -895,7 +896,7 @@ fn merge_usage_stream_metadata(metadata: &mut Option<serde_json::Value>, upstrea
         return;
     };
     object
-        .entry("upstream_is_stream")
+        .entry(UPSTREAM_IS_STREAM_KEY)
         .or_insert(serde_json::Value::Bool(upstream));
 }
 

crates/aether-data/src/repository/usage/sqlite.rs

@@ -1,6 +1,7 @@
 use std::collections::{BTreeMap, HashSet};
 use std::io::Read;
 
+use aether_ai_formats::UPSTREAM_IS_STREAM_KEY;
 use aether_data_contracts::repository::usage::{parse_usage_body_ref, UsageBodyField};
 use async_trait::async_trait;
 use flate2::read::GzDecoder;
@@ -3874,7 +3875,7 @@ fn usage_upstream_is_stream(usage: &UpsertUsageRecord) -> bool {
         .request_metadata
         .as_ref()
         .and_then(serde_json::Value::as_object)
-        .and_then(|metadata| metadata.get("upstream_is_stream"))
+        .and_then(|metadata| metadata.get(UPSTREAM_IS_STREAM_KEY))
         .and_then(serde_json::Value::as_bool)
         .unwrap_or_else(|| usage.is_stream.unwrap_or(false))
 }
@@ -3888,7 +3889,7 @@ fn merge_usage_stream_metadata(metadata: &mut Option<serde_json::Value>, upstrea
         return;
     };
     object
-        .entry("upstream_is_stream")
+        .entry(UPSTREAM_IS_STREAM_KEY)
         .or_insert(serde_json::Value::Bool(upstream));
 }
 

crates/aether-usage-runtime/src/request_metadata.rs

@@ -1,6 +1,7 @@
 use aether_ai_formats::api::{
     sanitize_request_path, sanitize_request_path_and_query, sanitize_request_query_string,
 };
+use aether_ai_formats::UPSTREAM_IS_STREAM_KEY;
 use aether_contracts::ExecutionPlan;
 use serde_json::{json, Map, Value};
 
@@ -74,7 +75,7 @@ fn copy_allowed_metadata_fields(source: &Map<String, Value>, target: &mut Map<St
     copy_non_empty_string(source, target, "user_agent");
     copy_non_empty_string(source, target, "client_family");
     copy_bool(source, target, "client_requested_stream");
-    copy_bool(source, target, "upstream_is_stream");
+    copy_bool(source, target, UPSTREAM_IS_STREAM_KEY);
     copy_non_null_value(source, target, "client_session_affinity");
     copy_bool(source, target, "api_key_is_standalone");
     copy_non_empty_string(source, target, "request_path");
@@ -114,7 +115,7 @@ fn move_allowed_metadata_fields(mut source: Map<String, Value>, target: &mut Map
     remove_non_empty_string(&mut source, target, "user_agent");
     remove_non_empty_string(&mut source, target, "client_family");
     remove_bool(&mut source, target, "client_requested_stream");
-    remove_bool(&mut source, target, "upstream_is_stream");
+    remove_bool(&mut source, target, UPSTREAM_IS_STREAM_KEY);
     remove_non_null_value(&mut source, target, "client_session_affinity");
     remove_bool(&mut source, target, "api_key_is_standalone");
     remove_non_empty_string(&mut source, target, "request_path");

crates/aether-usage-runtime/src/write.rs

@@ -1,5 +1,6 @@
 use std::collections::BTreeMap;
 
+use aether_ai_formats::UPSTREAM_IS_STREAM_KEY;
 use aether_contracts::{ExecutionPlan, ExecutionTelemetry};
 use aether_data_contracts::repository::usage::{UpsertUsageRecord, UsageBodyCaptureState};
 use aether_data_contracts::DataLayerError;
@@ -718,7 +719,7 @@ pub fn build_sync_terminal_usage_payload_seed(
         .report_context
         .as_ref()
         .and_then(Value::as_object)
-        .and_then(|context| context.get("upstream_is_stream"))
+        .and_then(|context| context.get(UPSTREAM_IS_STREAM_KEY))
         .and_then(Value::as_bool)
         .unwrap_or(false);
     let provider_response_full = if upstream_is_stream && payload.body_base64.is_some() {
@@ -1592,9 +1593,9 @@ fn build_runtime_request_metadata_seed_from_parts(
             Value::Bool(client_requested_stream),
         );
     }
-    if let Some(upstream_is_stream) = context_bool(context, "upstream_is_stream") {
+    if let Some(upstream_is_stream) = context_bool(context, UPSTREAM_IS_STREAM_KEY) {
         metadata.insert(
-            "upstream_is_stream".to_string(),
+            UPSTREAM_IS_STREAM_KEY.to_string(),
             Value::Bool(upstream_is_stream),
         );
     }