From 3077ac81e8ef3ddefebbe308ea37a4e9bb2100e6 Mon Sep 17 00:00:00 2001 From: JD Davis Date: Wed, 12 Aug 2026 23:16:54 -0500 Subject: [PATCH] feat: add deterministic runtime rollout controls (#1490) ## Description Establish one centrally resolved, observable, deterministic, versioned runtime rollout-control mechanism for Headroom. Runtime rollout controls which behaviors an already-built artifact may expose; it does not select or qualify a Headroom release/version. ## Type of Change - [x] New feature (non-breaking change that adds functionality) - [x] Bug fix (non-breaking change that fixes rollout enforcement regressions) - [x] Documentation update - [x] Code refactoring (no functional changes) ## Changes Made - Added `RolloutChannel`, `HEADROOM_ROLLOUT_CHANNEL`, `--rollout-channel`, and a versioned immutable `RolloutSnapshot` shared by Python configuration boundaries. - Added schema/policy versions, canonical registry and snapshot SHA-256 identities, per-feature decision reasons, disable precedence, unsafe qualification poisoning, strict CLI validation, and fail-closed environment handling. - Added `headroom rollout status --json`, Python `/stats.rollout`, and Rust `/rollout/status` runtime provenance. - Added equivalent Rust snapshot semantics and shared Python/Rust policy vectors while retaining language-specific feature registries. - Enforced rollout policy at alternate Python server composition roots so `HEADROOM_READ_MATURATION=1` cannot bypass its beta gate. - Preserved typed rollout snapshots across multi-worker serialization with schema, policy, registry, snapshot-digest, type, and feature-name validation. - Made loopback runtime output-shaper updates replace the immutable snapshot atomically for request readers, retain explicit request/disable provenance, preserve channel and kill-switch precedence, invalidate cached stats, and return the effective rollout decision. - Made `headroom learn --verbosity --apply` report a channel-blocked update instead of claiming the shaper is live. - Made explicit CLI feature flags fail loudly when their current channel blocks them. - Made persistent interceptor installation select canary automatically, or reject an explicitly insufficient channel unless the break-glass override is set. - Updated architecture, proxy, rollout, learn, and output-shaper documentation with required channels and hot-reload semantics. ## Testing - [x] Unit tests pass - [x] Linting passes (`ruff check .` and `ruff format --check .`) - [x] Type checking passes (`mypy headroom --ignore-missing-imports`) - [x] New regression tests added for every corrected behavior - [x] Rust tests and production-target Clippy pass - [x] Documentation build passes ### Test Output ```text Focused rollout coverage suite 57 passed; headroom.rollout + rollout CLI: 98% coverage Affected proxy/rollout/transform/governance suites 222 passed; 0 failed Final changed regression suites 100 passed; 0 failed Cross-module hot-reload isolation regression 6 passed; 0 failed cargo test -p headroom-core -p headroom-proxy --quiet headroom-core: 924 passed; 1 ignored headroom-proxy and integration suites: all passed cargo clippy -p headroom-core -p headroom-proxy --lib --bins -- -D warnings cargo fmt --all -- --check ruff check . ruff format --check . mypy headroom --ignore-missing-imports git diff --check All passed cd docs && npm run build Compiled successfully; 164 static pages generated ``` The unsharded Windows-only CI selection exposed unrelated baseline failures, principally the existing `sqlite:///C:\\...` URL parser producing an invalid `\\C:\\...` path. At commit `8e793a80`, all 52 completed GitHub checks passed; the only other conclusions are expected skips and superseded governance jobs. ## Real Behavior Proof - **Environment:** Windows checkout on Python 3.13.3 and the current Rust workspace, based on upstream `main` at `93f2d7a2`. - **Exact command / steps:** Exercised canary and beta feature requests through CLI status, Python `/stats.rollout`, Rust `/rollout/status`, multi-worker payload round trips, loopback `/admin/runtime-env`, real proxy request shaping before/after hot reload, installer manifest generation, and shared Python/Rust policy vectors. - **Observed result:** Stable blocks unstable requests; disable wins over explicit/default/legacy/unsafe paths; unsafe state reports `qualification_eligible=false`; worker handoff rejects tampering; running output shaping changes only when the effective beta policy permits it; explicit blocked flags fail with actionable diagnostics. - **Not tested:** Live production traffic requiring provider credentials, or future artifact qualification/promotion automation (intentionally out of scope). ## Runtime Rollout Safety - **Rollout-managed features:** Python `tool_result_interceptors`, `proxy_output_shaper`, `read_maturation`; Rust `native_bedrock`, `openai_responses_streaming`, `canary_probe`. - **Minimum rollout channel:** Registry-defined per feature; process default is `stable`. - **Stable/default behavior changed:** No unstable feature becomes enabled by default. Explicit blocked CLI flags now fail instead of silently doing nothing. - **Kill switch / disable path:** `HEADROOM_DISABLE_FEATURES=`; explicit disable has highest precedence, including over the unsafe override. - **Unsafe override required:** No. `HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES=1` is break-glass only and makes qualification evidence ineligible. - **Qualification impact:** Adds machine-readable policy/snapshot identities and eligibility; does not implement qualification itself. - **Rollback path:** Set the named disable list for operational rollback, lower the channel, or revert this PR. ## Review Readiness - [x] I have performed a full diff review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented hard-to-understand areas - [x] I have made corresponding documentation changes - [x] My changes generate no new warnings - [x] I added tests that reproduce and prevent every regression fixed during review - [x] New and existing affected tests pass locally - [x] I did **not** edit `CHANGELOG.md`; release-please generates it from the Conventional Commit PR title ## Additional Notes Out of scope: artifact candidates, benchmark orchestration, qualification manifests/gates, promotion automation, release branches, publication guards, and release-risk classification. Those workflows can consume the rollout registry digest, runtime snapshot digest, decision reasons, and qualification eligibility through supported black-box interfaces. --------- Co-authored-by: JD Davis Co-authored-by: JD Davis --- .github/PULL_REQUEST_TEMPLATE.md | 10 + .github/act/pr-governance-valid.json | 32 +- crates/headroom-core/src/lib.rs | 1 + crates/headroom-core/src/rollout.rs | 439 +++++++++++++++++ crates/headroom-proxy/src/config.rs | 153 +++++- crates/headroom-proxy/src/health.rs | 20 + crates/headroom-proxy/src/main.rs | 7 + crates/headroom-proxy/src/proxy.rs | 3 +- docs/content/docs/architecture.mdx | 2 +- docs/content/docs/configuration.mdx | 14 + docs/content/docs/meta.json | 1 + docs/content/docs/opencode-deepseek.mdx | 2 +- docs/content/docs/proxy.mdx | 4 +- docs/content/docs/runtime-rollouts.mdx | 174 +++++++ headroom/cli/__init__.py | 1 + headroom/cli/install.py | 3 +- headroom/cli/learn.py | 42 +- headroom/cli/main.py | 1 + headroom/cli/output_savings.py | 5 +- headroom/cli/proxy.py | 45 +- headroom/cli/rollout.py | 66 +++ headroom/config.py | 13 +- headroom/dashboard/templates/dashboard.html | 2 +- headroom/install/planner.py | 21 + headroom/proxy/handlers/anthropic.py | 8 +- headroom/proxy/handlers/openai.py | 42 +- headroom/proxy/models.py | 18 + headroom/proxy/output_shaper.py | 25 +- headroom/proxy/runtime_env.py | 6 +- headroom/proxy/server.py | 67 ++- headroom/rollout.py | 487 +++++++++++++++++++ headroom/transforms/pipeline.py | 13 +- scripts/pr-governance.py | 16 + scripts/tests/test_pr_governance.py | 21 + tests/fixtures/rollout_policy_vectors.json | 8 + tests/test_cli_learn.py | 57 +++ tests/test_install/test_planner.py | 11 + tests/test_openai_responses_output_shaper.py | 52 ++ tests/test_proxy_scalability.py | 1 + tests/test_proxy_stats_recent_requests.py | 30 ++ tests/test_read_maturation_handler_nobust.py | 16 + tests/test_rollout.py | 465 ++++++++++++++++++ tests/test_runtime_env.py | 68 +++ tests/test_tool_result_interceptors.py | 16 +- wiki/configuration.md | 20 + wiki/learn.md | 9 +- 46 files changed, 2435 insertions(+), 82 deletions(-) create mode 100644 crates/headroom-core/src/rollout.rs create mode 100644 docs/content/docs/runtime-rollouts.mdx create mode 100644 headroom/cli/rollout.py create mode 100644 headroom/rollout.py create mode 100644 tests/fixtures/rollout_policy_vectors.json create mode 100644 tests/test_rollout.py diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md index 2014d8dc8..3dd7ca601 100644 --- a/.github/PULL_REQUEST_TEMPLATE.md +++ b/.github/PULL_REQUEST_TEMPLATE.md @@ -40,6 +40,16 @@ Closes # - Observed result: - Not tested: +## Runtime Rollout Safety + +- Rollout-managed feature(s): +- Minimum rollout channel: +- Stable/default behavior changed: +- Kill switch / disable path: +- Unsafe override required: +- Qualification impact: +- Rollback path: + ## Review Readiness - [ ] I have performed a self-review diff --git a/.github/act/pr-governance-valid.json b/.github/act/pr-governance-valid.json index 8d1b07d16..a38b4b005 100644 --- a/.github/act/pr-governance-valid.json +++ b/.github/act/pr-governance-valid.json @@ -1,19 +1,13 @@ -{ - "action": "ready_for_review", - "number": 42, - "pull_request": { - "number": 42, - "draft": false, - "title": "feat: add PR governance", - "body": "## Description\n\nAdd a required PR governance check and commit-msg enforcement.\n\nCloses #123\n\n## Type of Change\n\n- [x] New feature (non-breaking change that adds functionality)\n\n## Changes Made\n\n- Added workflow validation for PR template completeness.\n- Added a commit-msg hook that runs commitlint locally.\n\n## Testing\n\n- [x] Unit tests pass (`pytest`)\n- [x] Manual testing performed\n\n### Test Output\n\n```text\npytest scripts/tests/test_pr_governance.py -q\n```\n\n## Real Behavior Proof\n\n- Environment: Ubuntu runner, Python 3.12\n- Exact command / steps: Opened a PR with an incomplete template, then fixed the body.\n- Observed result: The governance check failed until the template and readiness boxes were complete.\n- Not tested: Repository-level automatic Copilot rulesets.\n\n## Review Readiness\n\n- [x] I have performed a self-review\n- [x] This PR is ready for human review\n", - "user": { - "login": "octocat" - }, - "base": { - "sha": "dff6a199" - } - }, - "repository": { - "full_name": "JerrettDavis/headroom" - } -} +{ + "action": "ready_for_review", + "number": 42, + "pull_request": { + "number": 42, + "draft": false, + "title": "feat: add PR governance", + "body": "## Description\n\nAdd a required PR governance check and commit-msg enforcement.\n\n## Type of Change\n\n- [x] New feature (non-breaking change that adds functionality)\n\n## Changes Made\n\n- Added workflow validation for PR template completeness.\n\n## Testing\n\n- [x] Unit tests pass (`pytest`)\n\n### Test Output\n\n```text\npytest scripts/tests/test_pr_governance.py -q\n```\n\n## Real Behavior Proof\n\n- Environment: Ubuntu runner, Python 3.12\n- Exact command / steps: Opened a PR and ran governance.\n- Observed result: The check passed with complete facts.\n- Not tested: Repository settings.\n\n## Runtime Rollout Safety\n\n- Rollout-managed feature(s): None.\n- Minimum rollout channel: Stable.\n- Stable/default behavior changed: No.\n- Kill switch / disable path: Not applicable.\n- Unsafe override required: No.\n- Qualification impact: None.\n- Rollback path: Revert the workflow and script changes.\n\n## Review Readiness\n\n- [x] I have performed a self-review\n- [x] This PR is ready for human review\n", + "user": {"login": "octocat"}, + "base": {"sha": "dff6a199"} + }, + "repository": {"full_name": "JerrettDavis/headroom"} +} diff --git a/crates/headroom-core/src/lib.rs b/crates/headroom-core/src/lib.rs index 8a907045a..a9417bd49 100644 --- a/crates/headroom-core/src/lib.rs +++ b/crates/headroom-core/src/lib.rs @@ -7,6 +7,7 @@ pub mod compression_policy; #[cfg(feature = "ml")] mod onnx_cpu; pub mod relevance; +pub mod rollout; pub mod signals; pub mod tokenizer; pub mod transforms; diff --git a/crates/headroom-core/src/rollout.rs b/crates/headroom-core/src/rollout.rs new file mode 100644 index 000000000..fc9d4fb92 --- /dev/null +++ b/crates/headroom-core/src/rollout.rs @@ -0,0 +1,439 @@ +//! Deterministic runtime-rollout policy and provenance. +//! +//! Rollout channels control behavior in an already-built artifact. They do not +//! select a package, release candidate, or distribution version. Composition +//! roots resolve one immutable snapshot and inject its concrete decisions. + +use serde::Serialize; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::collections::BTreeSet; +use std::str::FromStr; + +pub const ROLLOUT_SCHEMA_VERSION: u32 = 1; +pub const ROLLOUT_POLICY_VERSION: &str = "1"; + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum RolloutChannel { + #[default] + Stable, + Beta, + Canary, + Dev, +} + +impl RolloutChannel { + pub fn as_str(self) -> &'static str { + match self { + Self::Stable => "stable", + Self::Beta => "beta", + Self::Canary => "canary", + Self::Dev => "dev", + } + } + + pub fn allows(self, required: Self) -> bool { + self >= required + } +} + +impl FromStr for RolloutChannel { + type Err = (); + + fn from_str(value: &str) -> Result { + match value.trim().to_ascii_lowercase().replace('-', "_").as_str() { + "" | "stable" | "prod" | "production" => Ok(Self::Stable), + "beta" | "preview" => Ok(Self::Beta), + "canary" | "nightly" => Ok(Self::Canary), + "dev" | "development" => Ok(Self::Dev), + _ => Err(()), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum Feature { + NativeBedrock, + OpenAiResponsesStreaming, + CanaryProbe, +} + +const ALL_FEATURES: [Feature; 3] = [ + Feature::CanaryProbe, + Feature::NativeBedrock, + Feature::OpenAiResponsesStreaming, +]; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct FeatureSpec { + pub name: &'static str, + pub available_in: RolloutChannel, + pub default_enabled_in: Option, +} + +impl Feature { + pub fn spec(self) -> FeatureSpec { + match self { + Self::NativeBedrock => FeatureSpec { + name: "native_bedrock", + available_in: RolloutChannel::Stable, + default_enabled_in: Some(RolloutChannel::Stable), + }, + Self::OpenAiResponsesStreaming => FeatureSpec { + name: "openai_responses_streaming", + available_in: RolloutChannel::Stable, + default_enabled_in: Some(RolloutChannel::Stable), + }, + Self::CanaryProbe => FeatureSpec { + name: "canary_probe", + available_in: RolloutChannel::Canary, + default_enabled_in: None, + }, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum FeatureDecisionReason { + Default, + Explicit, + LegacyAlias, + Disabled, + BlockedByChannel, + UnsafeOverride, + NotRequested, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct RolloutConfig { + pub channel: RolloutChannel, + pub requested: BTreeSet, + pub disabled: BTreeSet, + pub unsafe_allow_unstable: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct FeatureDecision { + pub name: &'static str, + pub available_in: RolloutChannel, + pub default_enabled_in: Option, + pub requested: bool, + pub disabled: bool, + pub enabled: bool, + #[serde(rename = "decision")] + pub reason: FeatureDecisionReason, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RolloutSnapshot { + pub schema_version: u32, + pub policy_version: &'static str, + pub registry_digest: String, + pub config: RolloutConfig, + pub decisions: Vec, +} + +impl Default for RolloutSnapshot { + fn default() -> Self { + Self::from_parts("stable", "", "", false) + } +} + +impl RolloutSnapshot { + pub fn from_parts( + channel: &str, + requested: &str, + disabled: &str, + unsafe_allow_unstable: bool, + ) -> Self { + Self::from_parts_with_explicit(channel, requested, disabled, unsafe_allow_unstable, &[]) + } + + pub fn from_parts_with_explicit( + channel: &str, + requested: &str, + disabled: &str, + unsafe_allow_unstable: bool, + explicit: &[Feature], + ) -> Self { + let parsed_channel = RolloutChannel::from_str(channel).unwrap_or_else(|_| { + tracing::warn!(channel, "unknown rollout channel; falling back to stable"); + RolloutChannel::Stable + }); + let valid_names: BTreeSet<_> = ALL_FEATURES + .iter() + .map(|feature| feature.spec().name.to_owned()) + .collect(); + let mut requested_names = validated_names(requested, "requested", &valid_names); + requested_names.extend( + explicit + .iter() + .map(|feature| feature.spec().name.to_owned()), + ); + let disabled_names = validated_names(disabled, "disabled", &valid_names); + let config = RolloutConfig { + channel: parsed_channel, + requested: requested_names, + disabled: disabled_names, + unsafe_allow_unstable, + }; + let decisions = ALL_FEATURES + .iter() + .map(|feature| resolve_feature(*feature, &config)) + .collect(); + Self { + schema_version: ROLLOUT_SCHEMA_VERSION, + policy_version: ROLLOUT_POLICY_VERSION, + registry_digest: registry_digest(), + config, + decisions, + } + } + + pub fn decision(&self, feature: Feature) -> &FeatureDecision { + let name = feature.spec().name; + self.decisions + .iter() + .find(|decision| decision.name == name) + .expect("every registered feature has a decision") + } + + pub fn is_enabled(&self, feature: Feature, _explicit: bool) -> bool { + self.decision(feature).enabled + } + + pub fn enabled(&self) -> BTreeSet { + self.decisions + .iter() + .filter(|decision| decision.enabled) + .map(|decision| decision.name.to_owned()) + .collect() + } + + pub fn qualification_eligible(&self) -> bool { + !self.config.unsafe_allow_unstable + } + + fn canonical_value(&self) -> Value { + json!({ + "schema_version": self.schema_version, + "policy_version": self.policy_version, + "channel": self.config.channel, + "unsafe_override": self.config.unsafe_allow_unstable, + "registry_digest": self.registry_digest, + "features": self.decisions, + }) + } + + pub fn snapshot_digest(&self) -> String { + digest_value(&self.canonical_value()) + } + + pub fn to_value(&self) -> Value { + let mut value = self.canonical_value(); + let object = value + .as_object_mut() + .expect("rollout snapshot is an object"); + object.insert("snapshot_digest".into(), json!(self.snapshot_digest())); + object.insert( + "qualification_eligible".into(), + json!(self.qualification_eligible()), + ); + if !self.qualification_eligible() { + object.insert( + "qualification_ineligible_reason".into(), + json!("unsafe_rollout_override_active"), + ); + } + value + } +} + +fn resolve_feature(feature: Feature, config: &RolloutConfig) -> FeatureDecision { + let spec = feature.spec(); + let requested = config.requested.contains(spec.name); + let disabled = config.disabled.contains(spec.name); + let normally_available = config.channel.allows(spec.available_in); + let (enabled, reason) = if disabled { + (false, FeatureDecisionReason::Disabled) + } else if requested && !normally_available && !config.unsafe_allow_unstable { + (false, FeatureDecisionReason::BlockedByChannel) + } else if requested && !normally_available { + (true, FeatureDecisionReason::UnsafeOverride) + } else if requested { + (true, FeatureDecisionReason::Explicit) + } else if spec + .default_enabled_in + .is_some_and(|minimum| config.channel.allows(minimum)) + { + (true, FeatureDecisionReason::Default) + } else { + (false, FeatureDecisionReason::NotRequested) + }; + FeatureDecision { + name: spec.name, + available_in: spec.available_in, + default_enabled_in: spec.default_enabled_in, + requested, + disabled, + enabled, + reason, + } +} + +fn validated_names(raw: &str, source: &str, valid: &BTreeSet) -> BTreeSet { + let names: BTreeSet<_> = split_feature_names(raw).into_iter().collect(); + for unknown in names.difference(valid) { + tracing::warn!( + feature = unknown, + source, + "unknown rollout feature; ignoring (fail-closed)" + ); + } + names.intersection(valid).cloned().collect() +} + +pub fn split_feature_names(raw: &str) -> Vec { + raw.replace(';', ",") + .split(',') + .filter_map(|part| { + let normalized = normalize_feature_name(part); + (!normalized.is_empty()).then_some(normalized) + }) + .collect() +} + +pub fn normalize_feature_name(raw: impl AsRef) -> String { + raw.as_ref().trim().to_ascii_lowercase().replace('-', "_") +} + +pub fn registry_digest() -> String { + let registry: Vec<_> = ALL_FEATURES.iter().map(|feature| feature.spec()).collect(); + digest_value(&serde_json::to_value(registry).expect("registry is serializable")) +} + +pub fn feature_names() -> BTreeSet<&'static str> { + ALL_FEATURES + .iter() + .map(|feature| feature.spec().name) + .collect() +} + +fn digest_value(value: &Value) -> String { + let canonical = serde_json::to_vec(value).expect("rollout provenance is serializable"); + format!("sha256:{:x}", Sha256::digest(canonical)) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde::Deserialize; + + #[derive(Deserialize)] + struct PolicyVector { + channel: String, + requested: bool, + disabled: bool, + #[serde(rename = "unsafe")] + unsafe_override: bool, + enabled: bool, + decision: String, + } + + #[test] + fn channel_order_matches_python_policy() { + assert!(RolloutChannel::Dev.allows(RolloutChannel::Canary)); + assert!(RolloutChannel::Canary.allows(RolloutChannel::Beta)); + assert!(!RolloutChannel::Stable.allows(RolloutChannel::Canary)); + } + + #[test] + fn stable_blocks_explicit_canary_feature_with_reason() { + let rollout = RolloutSnapshot::from_parts("stable", "canary_probe", "", false); + let decision = rollout.decision(Feature::CanaryProbe); + assert!(!decision.enabled); + assert_eq!(decision.reason, FeatureDecisionReason::BlockedByChannel); + } + + #[test] + fn default_enabled_feature_has_default_reason() { + let rollout = RolloutSnapshot::default(); + let decision = rollout.decision(Feature::NativeBedrock); + assert!(decision.enabled); + assert_eq!(decision.reason, FeatureDecisionReason::Default); + } + + #[test] + fn unsafe_override_crosses_boundary_and_is_ineligible() { + let rollout = RolloutSnapshot::from_parts("stable", "canary_probe", "", true); + assert_eq!( + rollout.decision(Feature::CanaryProbe).reason, + FeatureDecisionReason::UnsafeOverride + ); + assert!(!rollout.qualification_eligible()); + assert_eq!( + rollout.to_value()["qualification_ineligible_reason"], + "unsafe_rollout_override_active" + ); + } + + #[test] + fn disable_beats_default_explicit_and_unsafe() { + for unsafe_override in [false, true] { + let rollout = RolloutSnapshot::from_parts( + "stable", + "native_bedrock", + "native-bedrock", + unsafe_override, + ); + assert_eq!( + rollout.decision(Feature::NativeBedrock).reason, + FeatureDecisionReason::Disabled + ); + } + } + + #[test] + fn provenance_digests_are_deterministic_and_policy_sensitive() { + let first = RolloutSnapshot::from_parts("canary", "canary_probe", "", false); + let second = RolloutSnapshot::from_parts("canary", "canary_probe", "", false); + let changed = RolloutSnapshot::from_parts("stable", "canary_probe", "", false); + assert_eq!(first.registry_digest, second.registry_digest); + assert_eq!(first.snapshot_digest(), second.snapshot_digest()); + assert_ne!(first.snapshot_digest(), changed.snapshot_digest()); + } + + #[test] + fn invalid_inputs_fail_closed() { + let rollout = RolloutSnapshot::from_parts("stabel", "unknown", "unknown", false); + assert_eq!(rollout.config.channel, RolloutChannel::Stable); + assert!(rollout.config.requested.is_empty()); + assert!(rollout.config.disabled.is_empty()); + } + + #[test] + fn shared_python_rust_policy_vectors() { + let vectors: Vec = serde_json::from_str(include_str!( + "../../../tests/fixtures/rollout_policy_vectors.json" + )) + .unwrap(); + for vector in vectors { + let requested = if vector.requested { "canary_probe" } else { "" }; + let disabled = if vector.disabled { "canary_probe" } else { "" }; + let rollout = RolloutSnapshot::from_parts( + &vector.channel, + requested, + disabled, + vector.unsafe_override, + ); + let decision = rollout.decision(Feature::CanaryProbe); + assert_eq!(decision.enabled, vector.enabled); + assert_eq!( + serde_json::to_value(decision.reason).unwrap(), + vector.decision + ); + } + } +} diff --git a/crates/headroom-proxy/src/config.rs b/crates/headroom-proxy/src/config.rs index 5c2dfffc2..181589754 100644 --- a/crates/headroom-proxy/src/config.rs +++ b/crates/headroom-proxy/src/config.rs @@ -1,6 +1,9 @@ //! Configuration for the proxy: CLI flags + env vars. use clap::{Parser, ValueEnum}; +use headroom_core::rollout::{ + feature_names, split_feature_names, Feature, RolloutChannel, RolloutSnapshot, +}; use std::net::SocketAddr; use std::time::Duration; use url::Url; @@ -230,6 +233,49 @@ impl BetaHeaderSticky { about = "Headroom transparent reverse proxy" )] pub struct CliArgs { + /// Runtime rollout channel that bounds which managed features may run. + /// + /// `stable` admits only features that have completed bake time. `beta` and + /// `canary` admit progressively newer features. `dev` is for local work. + /// Explicit feature requests still cannot cross this boundary unless the + /// unsafe override is set. + #[arg( + long = "rollout-channel", + env = "HEADROOM_ROLLOUT_CHANNEL", + default_value = "stable", + value_parser = parse_rollout_channel, + )] + pub rollout_channel: String, + + /// Comma-separated rollout features to request explicitly. + #[arg( + long = "features", + env = "HEADROOM_FEATURES", + default_value = "", + value_parser = parse_rollout_features, + )] + pub features: String, + + /// Comma-separated rollout features to force off. Disable wins over defaults + /// and explicit enable requests. + #[arg( + long = "disable-features", + env = "HEADROOM_DISABLE_FEATURES", + default_value = "", + value_parser = parse_rollout_features, + )] + pub disable_features: String, + + /// Break-glass override that allows unstable features below their channel. + /// Intended only for emergency mitigation and should be visible in logs. + #[arg( + long = "unsafe-allow-unstable-features", + env = "HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES", + default_value_t = false, + action = clap::ArgAction::Set, + )] + pub unsafe_allow_unstable_features: bool, + /// Address the proxy listens on (e.g. 0.0.0.0:8787). #[arg(long, env = "HEADROOM_PROXY_LISTEN", default_value = "0.0.0.0:8787")] pub listen: SocketAddr, @@ -539,6 +585,32 @@ fn parse_duration(s: &str) -> Result { humantime::parse_duration(s).map_err(|e| format!("invalid duration `{s}`: {e}")) } +fn parse_rollout_channel(value: &str) -> Result { + value + .parse::() + .map(|channel| channel.as_str().to_owned()) + .map_err(|_| { + format!("unknown rollout channel `{value}` (valid: stable, beta, canary, dev)") + }) +} + +fn parse_rollout_features(value: &str) -> Result { + let valid = feature_names(); + let unknown: Vec<_> = split_feature_names(value) + .into_iter() + .filter(|name| !valid.contains(name.as_str())) + .collect(); + if unknown.is_empty() { + Ok(value.to_owned()) + } else { + Err(format!( + "unknown rollout feature(s): {}; valid: {}", + unknown.join(", "), + valid.into_iter().collect::>().join(", ") + )) + } +} + fn parse_bytes(s: &str) -> Result { s.parse::() .map(|b| b.as_u64()) @@ -548,6 +620,8 @@ fn parse_bytes(s: &str) -> Result { /// Resolved configuration used by the running server. #[derive(Debug, Clone)] pub struct Config { + /// Runtime rollout state resolved from CLI/env. + pub rollout: RolloutSnapshot, pub listen: SocketAddr, pub upstream: Url, pub upstream_timeout: Duration, @@ -622,6 +696,30 @@ pub struct Config { impl Config { pub fn from_cli(args: CliArgs) -> Self { + let mut explicit_features = Vec::new(); + if args.enable_responses_streaming { + explicit_features.push(Feature::OpenAiResponsesStreaming); + } + if args.enable_bedrock_native { + explicit_features.push(Feature::NativeBedrock); + } + // Preserve the pre-rollout rollback controls as legacy disables. Both + // features are stable defaults in the registry, so merely omitting a + // false flag from `explicit_features` would turn it straight back on. + let mut disabled_features = split_feature_names(&args.disable_features); + if !args.enable_responses_streaming { + disabled_features.push(Feature::OpenAiResponsesStreaming.spec().name.to_owned()); + } + if !args.enable_bedrock_native { + disabled_features.push(Feature::NativeBedrock.spec().name.to_owned()); + } + let rollout = RolloutSnapshot::from_parts_with_explicit( + &args.rollout_channel, + &args.features, + &disabled_features.join(","), + args.unsafe_allow_unstable_features, + &explicit_features, + ); let rewrite_host = if args.no_rewrite_host { false } else { @@ -631,6 +729,7 @@ impl Config { .compression_max_body_bytes .unwrap_or(args.max_body_bytes); Self { + rollout: rollout.clone(), listen: args.listen, upstream: args.upstream, upstream_timeout: args.upstream_timeout, @@ -646,9 +745,13 @@ impl Config { auth_mode_policy_enforcement: args.auth_mode_policy_enforcement, strip_internal_headers: args.strip_internal_headers, beta_header_sticky: args.beta_header_sticky, - enable_responses_streaming: args.enable_responses_streaming, + enable_responses_streaming: rollout.is_enabled( + Feature::OpenAiResponsesStreaming, + args.enable_responses_streaming, + ), enable_conversations_passthrough: args.enable_conversations_passthrough, - enable_bedrock_native: args.enable_bedrock_native, + enable_bedrock_native: rollout + .is_enabled(Feature::NativeBedrock, args.enable_bedrock_native), bedrock_region: args.bedrock_region, bedrock_endpoint: args.bedrock_endpoint, aws_profile: args.aws_profile, @@ -662,6 +765,7 @@ impl Config { /// production-default behaviour so existing tests stay unchanged. pub fn for_test(upstream: Url) -> Self { Self { + rollout: RolloutSnapshot::default(), listen: "127.0.0.1:0".parse().unwrap(), upstream, upstream_timeout: Duration::from_secs(60), @@ -715,3 +819,48 @@ impl Config { } } } + +#[cfg(test)] +mod rollout_input_tests { + use super::*; + + #[test] + fn explicit_rollout_inputs_are_strict_and_diagnosable() { + assert_eq!(parse_rollout_channel("CANARY").unwrap(), "canary"); + assert!(parse_rollout_channel("stabel") + .unwrap_err() + .contains("unknown rollout channel")); + assert!(parse_rollout_features("native-bedrock").is_ok()); + let error = parse_rollout_features("native_bedrok").unwrap_err(); + assert!(error.contains("native_bedrok")); + assert!(error.contains("native_bedrock")); + } + + #[test] + fn legacy_false_flags_remain_effective_rollout_disables() { + let args = CliArgs::try_parse_from([ + "headroom-proxy", + "--upstream", + "http://127.0.0.1:9", + "--enable-responses-streaming", + "false", + "--enable-bedrock-native", + "false", + ]) + .unwrap(); + + let config = Config::from_cli(args); + + for feature in [Feature::OpenAiResponsesStreaming, Feature::NativeBedrock] { + let decision = config.rollout.decision(feature); + assert!(!decision.enabled); + assert!(decision.disabled); + assert_eq!( + decision.reason, + headroom_core::rollout::FeatureDecisionReason::Disabled + ); + } + assert!(!config.enable_responses_streaming); + assert!(!config.enable_bedrock_native); + } +} diff --git a/crates/headroom-proxy/src/health.rs b/crates/headroom-proxy/src/health.rs index 683b917a6..5cf17a7f2 100644 --- a/crates/headroom-proxy/src/health.rs +++ b/crates/headroom-proxy/src/health.rs @@ -13,6 +13,11 @@ pub async fn healthz() -> impl IntoResponse { Json(json!({ "ok": true, "service": "headroom-proxy" })) } +/// Effective rollout state of this running Rust proxy process. +pub async fn rollout_status(State(state): State) -> Json { + Json(state.config.rollout.to_value()) +} + /// Upstream health: GETs upstream `/healthz`. Returns 200 when reachable + /// 2xx, 503 otherwise. The endpoint name is reserved by the proxy and is /// not forwarded; operators must not name a real upstream route this. @@ -39,3 +44,18 @@ pub async fn healthz_upstream(State(state): State) -> Response { .into_response(), } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::Config; + + #[tokio::test] + async fn rollout_status_exposes_running_snapshot() { + let state = AppState::new(Config::for_test("http://127.0.0.1:9".parse().unwrap())).unwrap(); + let expected = state.config.rollout.snapshot_digest(); + let Json(payload) = rollout_status(State(state)).await; + assert_eq!(payload["snapshot_digest"], expected); + assert_eq!(payload["qualification_eligible"], true); + } +} diff --git a/crates/headroom-proxy/src/main.rs b/crates/headroom-proxy/src/main.rs index 3a19d98ab..f21d0117f 100644 --- a/crates/headroom-proxy/src/main.rs +++ b/crates/headroom-proxy/src/main.rs @@ -28,6 +28,13 @@ async fn main() -> Result<(), Box> { max_body_bytes = config.max_body_bytes, rewrite_host = config.rewrite_host, graceful_shutdown_timeout_s = config.graceful_shutdown_timeout.as_secs(), + rollout_channel = config.rollout.config.channel.as_str(), + rollout_features_enabled = ?config.rollout.enabled(), + rollout_features_disabled = ?config.rollout.config.disabled, + unsafe_allow_unstable_features = config.rollout.config.unsafe_allow_unstable, + rollout_registry_digest = %config.rollout.registry_digest, + rollout_snapshot_digest = %config.rollout.snapshot_digest(), + qualification_eligible = config.rollout.qualification_eligible(), "headroom-proxy starting" ); diff --git a/crates/headroom-proxy/src/proxy.rs b/crates/headroom-proxy/src/proxy.rs index 4c14f9900..1005fb03f 100644 --- a/crates/headroom-proxy/src/proxy.rs +++ b/crates/headroom-proxy/src/proxy.rs @@ -25,7 +25,7 @@ use crate::compression; use crate::config::Config; use crate::error::ProxyError; use crate::headers::{build_forward_request_headers, filter_response_headers}; -use crate::health::{healthz, healthz_upstream}; +use crate::health::{healthz, healthz_upstream, rollout_status}; use crate::websocket::ws_handler; // Phase F PR-F1: imported as `classify_auth_mode` to make the call // site self-documenting. `AuthMode` is re-exported under the same @@ -157,6 +157,7 @@ pub fn build_app(state: AppState) -> Router { let mut router = Router::new() .route("/healthz", get(healthz)) .route("/healthz/upstream", get(healthz_upstream)) + .route("/rollout/status", get(rollout_status)) // PR-D3: Prometheus scrape endpoint. Renders the global // registry in text format. The handler is stateless — no // `AppState` needed — and idempotent across concurrent diff --git a/docs/content/docs/architecture.mdx b/docs/content/docs/architecture.mdx index 089cfca7c..d8f754372 100644 --- a/docs/content/docs/architecture.mdx +++ b/docs/content/docs/architecture.mdx @@ -47,7 +47,7 @@ In proxy mode the server is a FastAPI app with per-provider handlers (Anthropic, The proxy assembles a small, ordered pipeline. Every transform is independent, safe to skip, and **fails open** — on any error it returns the content unchanged and the request still goes through. -1. **Tool-result interceptor** *(opt-in)* — light structural interceptors such as ast-grep Read outlining. Off unless you pass `--intercept-tool-results`. +1. **Tool-result interceptor** *(canary opt-in)* — light structural interceptors such as ast-grep Read outlining. Requires `HEADROOM_ROLLOUT_CHANNEL=canary` plus `--intercept-tool-results`. 2. **CacheAligner** *(off by default)* — a detector that reports dynamic-prefix drift (dates, UUIDs, session tokens). It **never mutates, moves, or rewrites** content. It is disabled by default and hard-disabled inside the proxy; it exists to surface prefix-stability metrics, not to change your messages. 3. **ContentRouter** — the workhorse that does essentially all of the compression. See below. diff --git a/docs/content/docs/configuration.mdx b/docs/content/docs/configuration.mdx index f4d2956c1..c8a4fca82 100644 --- a/docs/content/docs/configuration.mdx +++ b/docs/content/docs/configuration.mdx @@ -5,6 +5,20 @@ description: All configuration options for the Headroom Python and TypeScript SD Headroom can be configured via the SDK constructor, proxy command line, environment variables, or per-request overrides. +## Runtime Rollout Channels + +Headroom uses rollout channels to control which behaviors an already-installed +artifact may expose. They do not select a package or released version. + +| Variable | Default | Purpose | +|----------|---------|---------| +| `HEADROOM_ROLLOUT_CHANNEL` | `stable` | Selects `stable`, `beta`, `canary`, or `dev`. | +| `HEADROOM_FEATURES` | unset | Comma-separated feature names to request explicitly. | +| `HEADROOM_DISABLE_FEATURES` | unset | Comma-separated feature names to force off. Disable wins over every enable path. | +| `HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES` | unset | Break-glass override for emergency mitigation only. | + +See [Runtime Rollouts](/docs/runtime-rollouts) for policy, provenance, and +contributor rules. If Codex history disappeared after using an older wrapper, see [Recover Codex State](/docs/codex-recovery) before wrapping Codex again. ## SDK Modes (`default_mode` / `headroom_mode`) diff --git a/docs/content/docs/meta.json b/docs/content/docs/meta.json index 554918cc6..d716df852 100644 --- a/docs/content/docs/meta.json +++ b/docs/content/docs/meta.json @@ -58,6 +58,7 @@ "architecture", "ci-cd-flows", "releases", + "runtime-rollouts", "benchmarks", "limitations", "---Help---", diff --git a/docs/content/docs/opencode-deepseek.mdx b/docs/content/docs/opencode-deepseek.mdx index 83c21a982..a9a3594af 100644 --- a/docs/content/docs/opencode-deepseek.mdx +++ b/docs/content/docs/opencode-deepseek.mdx @@ -71,7 +71,7 @@ curl -s http://127.0.0.1:8787/v1/models \ Output shaping makes the model's responses shorter — fewer tokens, lower cost: ```bash -HEADROOM_OUTPUT_SHAPER=1 HEADROOM_VERBOSITY_LEVEL=2 \ +HEADROOM_ROLLOUT_CHANNEL=beta HEADROOM_OUTPUT_SHAPER=1 HEADROOM_VERBOSITY_LEVEL=2 \ headroom proxy --port 8787 --openai-api-url https://api.deepseek.com/v1 ``` diff --git a/docs/content/docs/proxy.mdx b/docs/content/docs/proxy.mdx index 0351e3181..35bb7a77f 100644 --- a/docs/content/docs/proxy.mdx +++ b/docs/content/docs/proxy.mdx @@ -68,7 +68,7 @@ Avoid setting process-wide variables such as `HTTP_PROXY`, `HTTPS_PROXY`, `ALL_P |--------|---------|-------------| | `--mode token` | | Prioritize token compression; prior turns may be rewritten for maximum savings. | | `--mode cache` | default | Freeze prior turns to maximize provider prefix-cache hit rate. This is the effective default (see [Savings profiles](#savings-profiles)). | -| `--intercept-tool-results` | `false` | Opt into tool-result interceptors such as ast-grep Read outlining. | +| `--intercept-tool-results` | `false` | Opt into canary tool-result interceptors such as ast-grep Read outlining. Requires `HEADROOM_ROLLOUT_CHANNEL=canary` (or `dev`). | | `--no-read-lifecycle` | `false` | Disable stale/superseded Read-output compression. | | `--code-aware` / `--no-code-aware` | disabled | Enable or disable AST-based code compression. Requires `headroom-ai[code]`. | | `--code-graph` | `false` | Enable the proxy's live code-graph file watcher for the current project. | @@ -249,7 +249,7 @@ Coding agents re-read the same files repeatedly; these control how stale reads a | Flag / env | Default | Effect | |---|---|---| | `--no-read-lifecycle` | lifecycle on | Stop replacing stale/superseded file reads with CCR markers. | -| `--read-maturation` / `HEADROOM_READ_MATURATION` | `false` | *(Experimental)* Hold freshly-read files out of the prefix cache until the file quiesces. | +| `--read-maturation` / `HEADROOM_READ_MATURATION` | `false` | *(Beta)* Hold freshly-read files out of the prefix cache until the file quiesces. Requires `HEADROOM_ROLLOUT_CHANNEL=beta` (or `dev`). | | `--read-maturation-quiesce-turns` | `5` | Turns of no change before a held read is admitted. | ### Reliability: timeouts, retries, limits diff --git a/docs/content/docs/runtime-rollouts.mdx b/docs/content/docs/runtime-rollouts.mdx new file mode 100644 index 000000000..8dead2da9 --- /dev/null +++ b/docs/content/docs/runtime-rollouts.mdx @@ -0,0 +1,174 @@ +--- +title: Runtime Rollouts +description: Deterministic runtime feature control for installed Headroom artifacts. +--- + +Runtime rollout answers one question: **which behaviors may this already-built +Headroom artifact expose in this process?** It is separate from the source and +distribution lifecycle, which decides which commit/artifact is qualified, +released, packaged, and published. + +```bash +HEADROOM_ROLLOUT_CHANNEL=canary headroom proxy +``` + +This runs the installed artifact with canary-eligible runtime features available +according to that artifact's rollout policy. It does **not** install, select, or +run a canary release/version of Headroom. + +## Channels and feature policy + +Channels are ordered `stable < beta < canary < dev`. + +| Channel | Purpose | +|---------|---------| +| `stable` | Default; behavior eligible for normal production use. | +| `beta` | Opt-in behavior backed by automated and limited production evidence. | +| `canary` | Early dogfood behavior still gathering evidence. | +| `dev` | Local development and maintainer experiments. | + +Availability and default enablement are separate registry fields. A feature can +be available in `canary` but remain off until explicitly requested; another can +be available and default-enabled in `stable`. + +Request a named feature: + +```bash +HEADROOM_ROLLOUT_CHANNEL=canary \ +HEADROOM_FEATURES=tool_result_interceptors \ +headroom proxy --intercept-tool-results +``` + +Force it off with the kill switch: + +```bash +HEADROOM_DISABLE_FEATURES=tool_result_interceptors headroom proxy +``` + +## Resolution and precedence + +CLI arguments, environment variables, and typed configuration are resolved once +at configuration construction. The immutable snapshot is injected into the +proxy and transform pipelines; changing the process environment afterward does +not alter a running proxy. + +The existing loopback-only `/admin/runtime-env` endpoint is one narrow +exception: hot-reloading the legacy `HEADROOM_OUTPUT_SHAPER` alias replaces the +proxy's immutable snapshot with a newly resolved snapshot. Channel bounds and +`HEADROOM_DISABLE_FEATURES` still win, and `/stats.rollout` changes with the +effective running decision. Because these overrides are process-local, the +endpoint rejects updates when the built-in server uses multiple workers; restart +the proxy with the desired environment instead. Ambient environment mutation +remains ignored. + +Precedence is deterministic: + +| Condition | Result | +|-----------|--------| +| Explicit disable | Off, even if defaulted, requested, aliased, or unsafe override is active. | +| Requested below its availability channel, unsafe override active | On with `unsafe_override`. | +| Requested below its availability channel | Off with `blocked_by_channel`. | +| Explicit request in an allowed channel | On with `explicit`. | +| Enabled legacy alias in an allowed channel | On with `legacy_alias`. | +| Default-enabled in the active channel | On with `default`. | +| Otherwise | Off with `not_requested`. | + +Legacy feature-specific variables are narrow compatibility aliases only. They +obey channel bounds and explicit disable precedence. + +## Unsafe override and invalid input + +`HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES=1` is a break-glass mechanism. It can +cross a channel boundary for a requested feature, but cannot beat an explicit +disable. The runtime remains usable for debugging and emergency reproduction, +while its snapshot reports: + +```json +{ + "unsafe_override": true, + "qualification_eligible": false, + "qualification_ineligible_reason": "unsafe_rollout_override_active" +} +``` + +The Python resolver logs a warning and falls back to `stable` for an unknown +channel; unknown feature names are warned and ignored (fail-closed). Explicit +Python diagnostics (`headroom rollout status`) and the Rust front proxy's typed +CLI/environment parser reject unknown channels/features and list valid values +before startup. + +## Machine-readable status and provenance + +Inspect a supplied configuration without starting the proxy: + +```bash +headroom rollout status --json +``` + +Inspect the actual running process through the supported black-box endpoint: + +```bash +curl http://127.0.0.1:8787/stats +``` + +The Python proxy publishes the object at `/stats.rollout`. The Rust front proxy, +when deployed, publishes its own effective snapshot at `/rollout/status`; this +keeps each process's distinct feature registry and decisions independently +observable. + +The `/stats.rollout` object and CLI output contain no secrets. They include: + +```json +{ + "schema_version": 1, + "policy_version": "1", + "channel": "stable", + "unsafe_override": false, + "registry_digest": "sha256:...", + "snapshot_digest": "sha256:...", + "qualification_eligible": true, + "features": [ + { + "name": "tool_result_interceptors", + "available_in": "canary", + "default_enabled_in": null, + "requested": false, + "disabled": false, + "enabled": false, + "decision": "not_requested" + } + ] +} +``` + +`schema_version` versions the external JSON contract. `policy_version` versions +the rollout rules. `registry_digest` is SHA-256 over canonical, ordered feature +definitions. `snapshot_digest` identifies the complete effective runtime state. +Equivalent policies/configurations produce equal digests; material policy or +decision changes do not. + +These identities deliberately remain separate from source SHA, artifact SHA-256, +runtime payload SHA-256, and future qualification-policy identities. An external +benchmark can compare `/stats.rollout.registry_digest` and `snapshot_digest` +between A1 passthrough and B Headroom arms without importing Headroom internals. +A mismatch makes the future experiment invalid; benchmark logic itself is out of +scope for runtime rollout. + +## Evidence-backed graduation and rollback + +Features progress from canary through beta toward stable only with linked +deterministic, integration, and benchmark evidence. **Bake time is evidence, not +qualification by itself.** Stable eligibility is followed by release +qualification before behavior becomes a stable default. + +Every rollout-managed behavior must have a fast disable path. Operational +rollback uses `HEADROOM_DISABLE_FEATURES`; source rollback reverts the defining +change. The unsafe override is for diagnostics, not promotion or passing release +evidence. + +Contributors should add named registry entries and tests for default behavior, +explicit request, channel blocking, disable precedence, unsafe behavior, +decision reasons, and provenance rather than reading rollout variables inside +implementation components. Python and Rust registries contain features relevant +to their own runtimes, but share channel ordering, precedence, decision reasons, +fail-closed invalid-input semantics, and deterministic identity semantics. diff --git a/headroom/cli/__init__.py b/headroom/cli/__init__.py index 529a20d65..5901456a0 100644 --- a/headroom/cli/__init__.py +++ b/headroom/cli/__init__.py @@ -25,6 +25,7 @@ from . import ( # noqa: F401 perf, proxy, recover, + rollout, tools, update, wrap, diff --git a/headroom/cli/install.py b/headroom/cli/install.py index bf6b79991..417f73e37 100644 --- a/headroom/cli/install.py +++ b/headroom/cli/install.py @@ -495,7 +495,8 @@ def _echo_installed(manifest: DeploymentManifest, *, prefix: str = "Installed pe is_flag=True, help=( "Opt in to tool_result interceptors (ast-grep Read outliner, etc.) in the " - "persistent runtime. Off by default while this feature ships." + "persistent runtime. This also selects the required canary rollout channel " + "unless --env HEADROOM_ROLLOUT_CHANNEL=... is supplied." ), ) @click.option( diff --git a/headroom/cli/learn.py b/headroom/cli/learn.py index bcbf51466..b17a339ec 100644 --- a/headroom/cli/learn.py +++ b/headroom/cli/learn.py @@ -400,8 +400,9 @@ def _activate_output_shaper(port: int | None = None) -> tuple[str, int]: When a proxy is already running locally we hot-enable it via ``/admin/runtime-env`` (no restart, the same channel ``wrap`` uses), so ``--apply`` actually takes effect. Returns ``(status, port)`` where status is - ``"live"`` (enabled on a running proxy), ``"absent"`` (no reachable proxy), - or ``"error"``. + ``"live"`` (enabled on a running proxy), ``"blocked"`` (the proxy's + rollout channel rejected it), ``"absent"`` (no reachable proxy), or + ``"error"``. """ import json as _json import os as _os @@ -417,7 +418,22 @@ def _activate_output_shaper(port: int | None = None) -> tuple[str, int]: ) try: with urllib.request.urlopen(request, timeout=2) as response: - response.read() + raw_response = response.read() + payload = _json.loads(raw_response) if raw_response else {} + rollout = payload.get("rollout") if isinstance(payload, dict) else None + if isinstance(rollout, dict): + decisions = rollout.get("features") + if isinstance(decisions, list): + output_shaper = next( + ( + item + for item in decisions + if isinstance(item, dict) and item.get("name") == "proxy_output_shaper" + ), + None, + ) + if isinstance(output_shaper, dict) and not output_shaper.get("enabled", False): + return "blocked", resolved_port return "live", resolved_port except (urllib.error.URLError, OSError): # ConnectionRefused (no proxy) or 404 (proxy predates the endpoint). @@ -555,8 +571,17 @@ def _run_verbosity( f"level {best_profile.level} is live now (while HEADROOM_VERBOSITY_LEVEL is unset)." ) click.echo( - " To keep it on across restarts: export HEADROOM_OUTPUT_SHAPER=1 " - "before `headroom wrap ...` (wrap pushes it to the proxy)." + " To keep it on across restarts: export HEADROOM_ROLLOUT_CHANNEL=beta " + "and HEADROOM_OUTPUT_SHAPER=1 before `headroom wrap ...`." + ) + elif status == "blocked": + click.echo( + "\n ⚠ Level written, but the running proxy's rollout channel blocks the " + "beta output shaper." + ) + click.echo( + " Restart it with HEADROOM_ROLLOUT_CHANNEL=beta and " + "HEADROOM_OUTPUT_SHAPER=1; the learned level will be used automatically." ) else: click.echo( @@ -564,9 +589,10 @@ def _run_verbosity( "NOT shaping output yet." ) click.echo( - " Enable it: export HEADROOM_OUTPUT_SHAPER=1 then `headroom wrap ...` " - "(or start `headroom proxy` with it set). The learned level is then used " - "automatically while HEADROOM_VERBOSITY_LEVEL is unset." + " Enable it: export HEADROOM_ROLLOUT_CHANNEL=beta and " + "HEADROOM_OUTPUT_SHAPER=1, then run `headroom wrap ...` (or restart " + "`headroom proxy`). The learned level is then used automatically while " + "HEADROOM_VERBOSITY_LEVEL is unset." ) else: click.echo("\n Dry run — use --apply to persist the level and baseline.") diff --git a/headroom/cli/main.py b/headroom/cli/main.py index 1b74e22df..74fdaad8a 100644 --- a/headroom/cli/main.py +++ b/headroom/cli/main.py @@ -76,6 +76,7 @@ def _register_commands() -> None: perf, # noqa: F401 proxy, # noqa: F401 recover, # noqa: F401 + rollout, # noqa: F401 savings, # noqa: F401 tools, # noqa: F401 update, # noqa: F401 diff --git a/headroom/cli/output_savings.py b/headroom/cli/output_savings.py index a341b2eea..c3598fc3b 100644 --- a/headroom/cli/output_savings.py +++ b/headroom/cli/output_savings.py @@ -28,7 +28,10 @@ def output_savings() -> None: if not path.exists(): click.echo("No output-savings data yet.") click.echo("Run `headroom learn --verbosity --apply` to seed the baseline,") - click.echo("then enable the shaper (HEADROOM_OUTPUT_SHAPER=1) and send traffic.") + click.echo( + "then enable the beta shaper (HEADROOM_ROLLOUT_CHANNEL=beta " + "HEADROOM_OUTPUT_SHAPER=1) and send traffic." + ) return ledger = SavingsLedger.load(path) diff --git a/headroom/cli/proxy.py b/headroom/cli/proxy.py index 29e964abd..e9ba52686 100644 --- a/headroom/cli/proxy.py +++ b/headroom/cli/proxy.py @@ -258,7 +258,7 @@ def dashboard(port: int, no_open: bool) -> None: is_flag=True, help=( "Opt in to tool_result interceptors (ast-grep Read outliner, etc.). " - "Off by default while this feature ships." + "Requires HEADROOM_ROLLOUT_CHANNEL=canary (or dev)." ), ) @click.option("--no-optimize", is_flag=True, help="Disable optimization (passthrough mode)") @@ -641,7 +641,8 @@ def dashboard(port: int, no_open: bool) -> None: help=( "EXPERIMENTAL: activity-based read maturation — hold fresh Reads " "out of the provider prefix cache and compress them once their " - "file quiesces (env: HEADROOM_READ_MATURATION=1)" + "file quiesces. Requires HEADROOM_ROLLOUT_CHANNEL=beta (or dev); " + "env: HEADROOM_READ_MATURATION=1." ), ) @click.option( @@ -1080,12 +1081,46 @@ def proxy( err=True, ) + # Resolve rollout inputs once before constructing any rollout-managed + # behavior. The immutable snapshot is injected into ProxyConfig and is also + # what /stats later exposes. + from headroom.rollout import resolve_rollout + + rollout_requests = [] + if intercept_tool_results: + rollout_requests.append("tool_result_interceptors") + if read_maturation: + rollout_requests.append("read_maturation") + rollout_snapshot = resolve_rollout(os.environ, requested=rollout_requests) + + if read_maturation and not rollout_snapshot.is_enabled("read_maturation"): + click.secho( + "error: --read-maturation is not available in the current rollout channel " + f"({rollout_snapshot.channel.value}). Set HEADROOM_ROLLOUT_CHANNEL=beta " + "(or dev), or use HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES=1 for an " + "emergency override.", + fg="red", + err=True, + ) + sys.exit(1) + # Opt-in: turn on tool_result interceptors (ast-grep Read outline, etc.). # Only fetch the bundled CLI tool binaries when the feature is enabled — # otherwise we'd pay a network round-trip and risk a readonly-FS failure # for capabilities the user hasn't asked for. The TransformPipeline reads - # this env var at construction time. + # the resolved snapshot says it is active. if intercept_tool_results: + if not rollout_snapshot.is_enabled("tool_result_interceptors"): + click.secho( + "error: --intercept-tool-results is not available in the current " + f"rollout channel ({rollout_snapshot.channel.value}). Set " + "HEADROOM_ROLLOUT_CHANNEL=canary to dogfood it, or use " + "HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES=1 for emergency override.", + fg="red", + err=True, + ) + sys.exit(1) + from headroom.binaries import ensure_tools resolved_tools = ensure_tools() @@ -1103,7 +1138,6 @@ def proxy( err=True, ) sys.exit(1) - os.environ["HEADROOM_INTERCEPT_ENABLED"] = "1" try: resolved_anthropic_extra_headers = resolve_extra_headers( @@ -1185,6 +1219,7 @@ def proxy( config = ProxyConfig( host=host, port=port, + rollout=rollout_snapshot, anthropic_api_url=provider_api_overrides.anthropic, anthropic_extra_headers=resolved_anthropic_extra_headers, openai_extra_headers=resolved_openai_extra_headers, @@ -1291,7 +1326,7 @@ def proxy( # Read lifecycle: ON by default (use --no-read-lifecycle to disable) read_lifecycle=not no_read_lifecycle, # Read maturation (Mechanism B): experimental, OFF by default - read_maturation=read_maturation, + read_maturation=rollout_snapshot.is_enabled("read_maturation"), read_maturation_quiesce_turns=read_maturation_quiesce_turns, read_maturation_max_hold_turns=read_maturation_max_hold_turns, read_maturation_min_size_bytes=read_maturation_min_size_bytes, diff --git a/headroom/cli/rollout.py b/headroom/cli/rollout.py new file mode 100644 index 000000000..fce624174 --- /dev/null +++ b/headroom/cli/rollout.py @@ -0,0 +1,66 @@ +"""Runtime rollout diagnostics commands.""" + +from __future__ import annotations + +import json +import os + +import click + +from headroom.rollout import RolloutConfigurationError, resolve_rollout + +from .main import main + + +@main.group("rollout") +def rollout_group() -> None: + """Inspect runtime feature-rollout policy (not package releases).""" + + +@rollout_group.command("status") +@click.option("--channel", envvar="HEADROOM_ROLLOUT_CHANNEL") +@click.option("--features", envvar="HEADROOM_FEATURES") +@click.option("--disable-features", envvar="HEADROOM_DISABLE_FEATURES") +@click.option( + "--unsafe-allow-unstable-features", + is_flag=True, + envvar="HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES", +) +@click.option("--json", "json_output", is_flag=True, help="Emit the versioned JSON snapshot.") +def rollout_status( + channel: str | None, + features: str | None, + disable_features: str | None, + unsafe_allow_unstable_features: bool, + json_output: bool, +) -> None: + """Resolve and print the supplied runtime rollout configuration.""" + + env = dict(os.environ) + if channel is not None: + env["HEADROOM_ROLLOUT_CHANNEL"] = channel + if features is not None: + env["HEADROOM_FEATURES"] = features + if disable_features is not None: + env["HEADROOM_DISABLE_FEATURES"] = disable_features + if unsafe_allow_unstable_features: + env["HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES"] = "1" + try: + snapshot = resolve_rollout(env, strict=True) + except RolloutConfigurationError as exc: + raise click.ClickException(str(exc)) from exc + + payload = snapshot.to_dict() + if json_output: + click.echo(json.dumps(payload, sort_keys=True, separators=(",", ":"))) + return + + click.echo(f"Rollout channel: {snapshot.channel.value}") + click.echo(f"Policy: {snapshot.policy_version} ({snapshot.registry_digest})") + click.echo(f"Snapshot: {snapshot.snapshot_digest}") + click.echo(f"Qualification eligible: {str(snapshot.qualification_eligible).lower()}") + for decision in snapshot.decisions: + click.echo( + f" {decision.name}: enabled={str(decision.enabled).lower()} " + f"decision={decision.reason.value}" + ) diff --git a/headroom/config.py b/headroom/config.py index 2b01d6bd6..123cc9e93 100644 --- a/headroom/config.py +++ b/headroom/config.py @@ -11,6 +11,7 @@ from enum import Enum from typing import Any, Literal from headroom.models.config import ML_MODEL_DEFAULTS +from headroom.rollout import RolloutSnapshot, resolve_rollout class HeadroomMode(str, Enum): @@ -672,9 +673,14 @@ class HeadroomConfig: content_router_enabled: InitVar[bool | None] = None # Tool-result interceptors (ast-grep Read outline, etc.). Opt-in for now. - # Env var HEADROOM_INTERCEPT_ENABLED=1 also enables (for CLI `--intercept-tool-results`). + # The legacy env alias and this typed request still obey the canary rollout gate. intercept_tool_results: bool = False + # Immutable runtime rollout state. ``None`` is resolved once here so every + # pipeline built from this config observes the same decisions even if the + # process environment later changes. + rollout: RolloutSnapshot | None = None + # Debugging - opt-in diff artifact generation generate_diff_artifact: bool = False # Enable to get detailed transform diffs @@ -682,6 +688,11 @@ class HeadroomConfig: pipeline_extensions: list[Any] = field(default_factory=list) discover_pipeline_extensions: bool = True + def __post_init__(self, content_router_enabled: bool | None = None) -> None: + if self.rollout is None: + requested = ("tool_result_interceptors",) if self.intercept_tool_results else () + self.rollout = resolve_rollout(requested=requested) + def get_context_limit(self, model: str) -> int | None: """ Get context limit for a model from user overrides. diff --git a/headroom/dashboard/templates/dashboard.html b/headroom/dashboard/templates/dashboard.html index f195eec6f..ca0c9efc3 100644 --- a/headroom/dashboard/templates/dashboard.html +++ b/headroom/dashboard/templates/dashboard.html @@ -264,7 +264,7 @@ diff --git a/headroom/install/planner.py b/headroom/install/planner.py index 9b6d26fdd..960a5b4ee 100644 --- a/headroom/install/planner.py +++ b/headroom/install/planner.py @@ -9,6 +9,7 @@ import click from headroom import paths as _paths from headroom.providers.install_registry import build_install_target_envs +from headroom.rollout import RolloutChannel from .models import ( ConfigScope, @@ -169,6 +170,26 @@ def build_manifest( # defaults above (e.g. a custom HEADROOM_WORKSPACE_DIR). if extra_env: base_env.update(extra_env) + if intercept_tool_results: + configured_channel = base_env.get("HEADROOM_ROLLOUT_CHANNEL") + if configured_channel is None: + # The flag is an explicit canary opt-in. Persist the matching + # channel so the generated service can actually start. + base_env["HEADROOM_ROLLOUT_CHANNEL"] = RolloutChannel.CANARY.value + else: + channel = RolloutChannel.parse(configured_channel) + unsafe = base_env.get("HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES", "").lower() in { + "1", + "true", + "yes", + "on", + "enabled", + } + if not channel.allows(RolloutChannel.CANARY) and not unsafe: + raise click.ClickException( + "--intercept-tool-results requires HEADROOM_ROLLOUT_CHANNEL=canary " + "(or dev), unless the unsafe rollout override is explicitly enabled" + ) proxy_args = [ "--host", diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py index ac608f4b9..6d6848c43 100644 --- a/headroom/proxy/handlers/anthropic.py +++ b/headroom/proxy/handlers/anthropic.py @@ -2763,7 +2763,13 @@ class AnthropicHandlerMixin: shape_request, ) - _shaper_settings = OutputShaperSettings.from_env() + _shaper_settings = OutputShaperSettings.from_env( + enabled=( + self.config.rollout.is_enabled("proxy_output_shaper") + if getattr(self.config, "rollout", None) is not None + else None + ) + ) if _shaper_settings.enabled: # Conversation-stable holdout assignment: a whole # conversation is treatment or control. This keeps the A/B diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index ce89c0d2c..79a0fa986 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -674,6 +674,7 @@ def _shape_openai_responses_payload( *, model: str, request_id: str, + output_shaper_enabled: bool | None = None, ) -> tuple[list[str], bool]: """Output shaping for a Responses payload (opt-in, HEADROOM_OUTPUT_SHAPER). @@ -704,7 +705,7 @@ def _shape_openai_responses_payload( shape_responses_request, ) - settings = OutputShaperSettings.from_env() + settings = OutputShaperSettings.from_env(enabled=output_shaper_enabled) if not settings.enabled: return [], False @@ -1196,6 +1197,7 @@ def _shape_openai_responses_for_output( input_tokens: int, model: str, conversation_key: str | None = None, + output_shaper_enabled: bool | None = None, ) -> Any: """Apply OpenAI Responses output shaping and attach holdout labels.""" from headroom.proxy.output_savings import ( @@ -1212,7 +1214,7 @@ def _shape_openai_responses_for_output( shape_openai_responses_request, ) - settings = OutputShaperSettings.from_env() + settings = OutputShaperSettings.from_env(enabled=output_shaper_enabled) result = ShapeResult() if not settings.enabled: return result @@ -1279,6 +1281,7 @@ def _shape_openai_response_create_frame( *, input_tokens: int, conversation_key: str | None = None, + output_shaper_enabled: bool | None = None, ) -> tuple[str, bool, list[str], str | None]: try: parsed = json.loads(raw_msg) @@ -1297,6 +1300,7 @@ def _shape_openai_response_create_frame( input_tokens=input_tokens, model=str(payload.get("model") or ""), conversation_key=conversation_key, + output_shaper_enabled=output_shaper_enabled, ) labels = list(result.labels or []) if not result.changed: @@ -2799,7 +2803,16 @@ class OpenAIHandlerMixin: # closure so the extra payload serialization stays off the event # loop. shape_labels, shape_mutated = _shape_openai_responses_payload( - payload, model=model, request_id=request_id + payload, + model=model, + request_id=request_id, + output_shaper_enabled=( + getattr(getattr(self, "config", None), "rollout", None).is_enabled( + "proxy_output_shaper" + ) + if getattr(getattr(self, "config", None), "rollout", None) is not None + else None + ), ) compression_kwargs: dict[str, Any] = { "model": model, @@ -3980,7 +3993,13 @@ class OpenAIHandlerMixin: shape_openai_chat_request, ) - _shaper_settings = OutputShaperSettings.from_env() + _shaper_settings = OutputShaperSettings.from_env( + enabled=( + self.config.rollout.is_enabled("proxy_output_shaper") + if getattr(self.config, "rollout", None) is not None + else None + ) + ) if _shaper_settings.enabled: # Conversation-stable holdout: a whole conversation is treatment # or control, which keeps the A/B comparison clean and the @@ -5412,6 +5431,11 @@ class OpenAIHandlerMixin: if _http_conversation_key else None ), + output_shaper_enabled=( + self.config.rollout.is_enabled("proxy_output_shaper") + if getattr(self.config, "rollout", None) is not None + else None + ), ) _append_unique_transforms(transforms_applied, _shape_result.labels) if _shape_result.changed: @@ -7293,6 +7317,11 @@ class OpenAIHandlerMixin: self.openai_provider, ), conversation_key=f"ws:{session_id}", + output_shaper_enabled=( + self.config.rollout.is_enabled("proxy_output_shaper") + if getattr(self.config, "rollout", None) is not None + else None + ), ) _append_unique_transforms(transforms_applied, _shape_labels) if _shape_modified: @@ -7719,6 +7748,11 @@ class OpenAIHandlerMixin: self.openai_provider, ), conversation_key=f"ws:{session_id}", + output_shaper_enabled=( + self.config.rollout.is_enabled("proxy_output_shaper") + if getattr(self.config, "rollout", None) is not None + else None + ), ) _append_unique_transforms( transforms_applied, diff --git a/headroom/proxy/models.py b/headroom/proxy/models.py index e44493bbe..3edb7ad60 100644 --- a/headroom/proxy/models.py +++ b/headroom/proxy/models.py @@ -14,6 +14,7 @@ from typing import Any, Literal from headroom.memory import qdrant_env from headroom.providers.registry import ProviderApiOverrides from headroom.proxy.model_router import ModelRouterConfig +from headroom.rollout import RolloutSnapshot, resolve_rollout logger = logging.getLogger(__name__) @@ -133,6 +134,8 @@ class ProxyConfig: # Server host: str = "127.0.0.1" port: int = 8787 + # Resolved at this configuration boundary and then injected unchanged. + rollout: RolloutSnapshot | None = None anthropic_api_url: str | None = None # Custom Anthropic API URL override openai_api_url: str | None = None # Custom OpenAI API URL override # Display label for the OpenAI-compatible upstream (dashboard/stats only). @@ -493,7 +496,22 @@ class ProxyConfig: # ``HeadroomProxy._run_compression_in_executor``. compression_max_workers: int | None = None + # Number of built-in uvicorn worker processes sharing this listen socket. + # Kept at the end to avoid shifting existing positional constructor fields. + # Process-local runtime hot reload is unsafe above one worker because only + # the worker receiving the admin request would observe the update. + worker_processes: int = 1 + def __post_init__(self, smart_routing: bool | None = None) -> None: + if self.rollout is None: + self.rollout = resolve_rollout() + # ``read_maturation`` remains a concrete, already-resolved runtime + # setting for programmatic/config-file callers. The CLI composition + # root derives it from this same snapshot before constructing the + # config; rewriting it here would resolve policy a second time and + # break explicit non-CLI configuration. + if self.worker_processes < 1: + raise ValueError("worker_processes must be >= 1") if self.retry_enabled and self.retry_max_attempts < 1: raise ValueError("retry_max_attempts must be >= 1 when retry_enabled=True") # A 0 (or negative) requests-per-minute limit divides by zero in the diff --git a/headroom/proxy/output_shaper.py b/headroom/proxy/output_shaper.py index daa84f256..7dd9663ba 100644 --- a/headroom/proxy/output_shaper.py +++ b/headroom/proxy/output_shaper.py @@ -97,11 +97,7 @@ _replace_or_append_steering_block = replace_or_append_steering_block @dataclass(frozen=True) class OutputShaperSettings: - """Runtime settings, resolved once per request from the environment. - - Env-driven (like HEADROOM_INTERCEPT_ENABLED) so the proxy picks it up - without config plumbing through the server. Off by default. - """ + """Output-shaping settings with rollout enablement injected by the proxy.""" enabled: bool = False verbosity_level: int = 2 @@ -109,12 +105,19 @@ class OutputShaperSettings: mechanical_effort: str = "low" @classmethod - def from_env(cls) -> OutputShaperSettings: - enabled = runtime_env.getenv("HEADROOM_OUTPUT_SHAPER", "").lower() in ( - "1", - "true", - "yes", - ) + def from_env(cls, *, enabled: bool | None = None) -> OutputShaperSettings: + """Resolve tuning; running proxies always inject the resolved gate. + + ``None`` preserves the helper's direct-call compatibility for SDK/tests, + but proxy request paths never use it and therefore never re-resolve the + rollout alias. + """ + if enabled is None: + enabled = runtime_env.getenv("HEADROOM_OUTPUT_SHAPER", "").lower() in ( + "1", + "true", + "yes", + ) try: level = int(runtime_env.getenv("HEADROOM_VERBOSITY_LEVEL", "2")) except ValueError: diff --git a/headroom/proxy/runtime_env.py b/headroom/proxy/runtime_env.py index 22a84a8ec..bc4c03bc2 100644 --- a/headroom/proxy/runtime_env.py +++ b/headroom/proxy/runtime_env.py @@ -2,7 +2,7 @@ Most Headroom settings are read once at proxy startup into ``Config`` and are visible in ``/health``. A second, smaller class of environment variables is -read *live* — on every request (the output-shaper family) or captured at module +read *live* — on every request (output-shaper tuning) or captured at module import (the ast-grep read-rewrite threshold). The proxy reads these from its own process environment, so a *reused* proxy — one ``headroom wrap`` attaches to rather than starting fresh — never sees values a user exports afterwards. The @@ -18,6 +18,10 @@ instead of ``os.environ.get`` so an override wins over the launch-time environment; with no override set, behaviour is byte-for-byte identical to reading the environment directly. +The output-shaper master switch is rollout-managed. Hot-reloading that legacy +alias also replaces the proxy's immutable rollout snapshot; the active channel +and named kill switch still bound the resulting decision. + Scope rule: a variable belongs here only if the proxy reads it *after* startup (or captures it at import) AND it is not already reflected in the ``/health`` ``config`` block that ``wrap`` compares for reuse. Startup-captured settings diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index e47332eb9..1c9fae65e 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -957,7 +957,8 @@ class HeadroomProxy( self._code_aware_status = "lazy" if config.code_aware_enabled else "disabled" _intercept_prefix: list = [] - if os.environ.get("HEADROOM_INTERCEPT_ENABLED"): + assert config.rollout is not None + if config.rollout.is_enabled("tool_result_interceptors"): from headroom.proxy.interceptors import ToolResultInterceptorTransform _intercept_prefix = [ToolResultInterceptorTransform()] @@ -3497,8 +3498,8 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI: Loopback-only. The body is a flat ``{ENV_NAME: "value"}`` map; unknown keys and non-string values are ignored. Returns what was applied plus - the resulting live config. Last writer wins (overrides are global to the - proxy, which is inherent — every wrapper shares one process). + the resulting live config. Last writer wins in a single-worker proxy; + multi-worker proxies reject the update because overrides are process-local. """ try: body = await request.json() @@ -3509,7 +3510,27 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI: status_code=400, content={"error": "expected a JSON object of {ENV_NAME: value}"}, ) + if proxy.config.worker_processes > 1: + return JSONResponse( + status_code=409, + content={ + "error": ( + "runtime environment hot reload is unavailable with multiple " + "worker processes; restart the proxy with the desired environment" + ), + "worker_processes": proxy.config.worker_processes, + }, + ) applied = runtime_env.set_overrides(body) + rollout_aliases = { + key: value for key, value in applied.items() if key == "HEADROOM_OUTPUT_SHAPER" + } + if rollout_aliases: + assert proxy.config.rollout is not None + proxy.config.rollout = proxy.config.rollout.with_legacy_env(rollout_aliases) + async with _stats_snapshot_lock: + _stats_snapshot["value"] = None + _stats_snapshot["expires_at"] = 0.0 if applied: logger.info("runtime-env hot-reload applied: %s", sorted(applied)) # Record which runtime-env keys changed (the "what" of a config @@ -3524,7 +3545,11 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI: ) return JSONResponse( status_code=200, - content={"applied": applied, "runtime_env": runtime_env.effective_runtime_env()}, + content={ + "applied": applied, + "runtime_env": runtime_env.effective_runtime_env(), + "rollout": proxy.config.rollout.to_dict() if proxy.config.rollout else None, + }, ) # Vendored dashboard JS (tailwind/htmx/alpine). Mounted before @@ -4283,6 +4308,9 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI: "log_full_messages": proxy.config.log_full_messages if proxy else False, **get_quota_registry().get_all_stats(), "throughput": throughput, + # Effective state from the running process. This is the supported + # black-box provenance surface for benchmark/qualification tools. + "rollout": proxy.config.rollout.to_dict() if proxy.config.rollout else None, } def _dashboard_config_payload() -> dict[str, Any]: @@ -5058,6 +5086,10 @@ def _json_ready(value: Any) -> Any: def _proxy_config_payload(config: ProxyConfig) -> dict[str, Any]: payload: dict[str, Any] = {} for field in fields(config): + if field.name == "rollout": + assert config.rollout is not None + payload["_rollout_snapshot"] = config.rollout.to_internal_dict() + continue value = _json_ready(getattr(config, field.name)) try: json.dumps(value) @@ -5071,13 +5103,25 @@ def _proxy_config_from_env() -> ProxyConfig: raw_config = os.environ.get(_MULTI_WORKER_CONFIG_ENV) if raw_config: try: - return ProxyConfig(**json.loads(raw_config)) - except (TypeError, ValueError, json.JSONDecodeError): + values = json.loads(raw_config) + if not isinstance(values, dict): + raise TypeError("proxy config JSON must be an object") + if "_rollout_snapshot" in values: + rollout_value = values.pop("_rollout_snapshot") + from headroom.rollout import RolloutSnapshot + + values["rollout"] = RolloutSnapshot.from_internal_dict(rollout_value) + return ProxyConfig(**values) + except (KeyError, TypeError, ValueError, json.JSONDecodeError): logger.warning( "Invalid %s; falling back to HEADROOM_* env vars", _MULTI_WORKER_CONFIG_ENV ) + from headroom.rollout import resolve_rollout + + rollout = resolve_rollout() return ProxyConfig( + rollout=rollout, host=_get_env_str("HEADROOM_HOST", "127.0.0.1"), port=_get_env_int("HEADROOM_PORT", 8787), openai_api_url=os.environ.get("OPENAI_TARGET_API_URL"), @@ -5115,7 +5159,7 @@ def _proxy_config_from_env() -> ProxyConfig: # posture (compress_user, protect_recent, min_tokens). HEADROOM_SAVINGS_PROFILE # overrides. savings_profile=os.environ.get("HEADROOM_SAVINGS_PROFILE") or "coding", - read_maturation=_get_env_bool("HEADROOM_READ_MATURATION", False), + read_maturation=rollout.is_enabled("read_maturation"), read_maturation_quiesce_turns=_get_env_int("HEADROOM_READ_MATURATION_QUIESCE_TURNS", 5), read_maturation_max_hold_turns=_get_env_int("HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", 25), read_maturation_min_size_bytes=_get_env_int( @@ -5205,6 +5249,9 @@ def run_server( seed_proxy_env_defaults() config = config or ProxyConfig() + if workers < 1: + raise ValueError("workers must be >= 1") + config.worker_processes = workers code_aware_status = _get_code_aware_banner_status(config) # Format connection pool info @@ -5781,7 +5828,11 @@ if __name__ == "__main__": args.protect_tool_results or os.environ.get("HEADROOM_PROTECT_TOOL_RESULTS") ) + from headroom.rollout import resolve_rollout + + rollout = resolve_rollout() config = ProxyConfig( + rollout=rollout, host=_get_env_str("HEADROOM_HOST", args.host), port=_get_env_int("HEADROOM_PORT", args.port), openai_api_url=_get_env_str("OPENAI_TARGET_API_URL", args.openai_api_url), @@ -5835,7 +5886,7 @@ if __name__ == "__main__": keepalive_expiry=_get_env_float("HEADROOM_KEEPALIVE_EXPIRY", args.keepalive_expiry), http2=not args.no_http2 and _get_env_bool("HEADROOM_HTTP2", True), http_proxy=_get_env_str("HEADROOM_HTTP_PROXY", args.http_proxy or "") or None, - read_maturation=_get_env_bool("HEADROOM_READ_MATURATION", False), + read_maturation=rollout.is_enabled("read_maturation"), read_maturation_quiesce_turns=_get_env_int("HEADROOM_READ_MATURATION_QUIESCE_TURNS", 5), read_maturation_max_hold_turns=_get_env_int("HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", 25), read_maturation_min_size_bytes=_get_env_int( diff --git a/headroom/rollout.py b/headroom/rollout.py new file mode 100644 index 000000000..859b6e80c --- /dev/null +++ b/headroom/rollout.py @@ -0,0 +1,487 @@ +"""Deterministic runtime rollout policy and provenance. + +Rollout channels control behavior exposed by an already-installed artifact. +They do not select a Headroom release, package, or distribution version. +Environment access is confined to :func:`resolve_rollout`; downstream code +receives the resulting immutable :class:`RolloutSnapshot`. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from enum import Enum + +logger = logging.getLogger(__name__) + +ROLLOUT_SCHEMA_VERSION = 1 +ROLLOUT_POLICY_VERSION = "1" + +_TRUE_VALUES = {"1", "true", "yes", "on", "enabled"} +_FALSE_VALUES = {"0", "false", "no", "off", "disabled"} + + +class RolloutConfigurationError(ValueError): + """A supplied rollout configuration is invalid.""" + + +class RolloutChannel(str, Enum): + """Ordered runtime-behavior channels; unrelated to artifact releases.""" + + STABLE = "stable" + BETA = "beta" + CANARY = "canary" + DEV = "dev" + + @classmethod + def parse(cls, value: str | None, *, strict: bool = False) -> RolloutChannel: + if not value: + return cls.STABLE + normalized = value.strip().lower().replace("-", "_") + aliases = { + "prod": cls.STABLE, + "production": cls.STABLE, + "preview": cls.BETA, + "nightly": cls.CANARY, + "development": cls.DEV, + } + if normalized in aliases: + return aliases[normalized] + try: + return cls(normalized) + except ValueError: + message = f"unknown rollout channel {value!r}" + if strict: + raise RolloutConfigurationError(message) from None + logger.warning("%s; falling back to 'stable'", message) + return cls.STABLE + + @property + def order(self) -> int: + return { + RolloutChannel.STABLE: 0, + RolloutChannel.BETA: 1, + RolloutChannel.CANARY: 2, + RolloutChannel.DEV: 3, + }[self] + + def allows(self, required: RolloutChannel) -> bool: + return self.order >= required.order + + +class FeatureDecisionReason(str, Enum): + DEFAULT = "default" + EXPLICIT = "explicit" + LEGACY_ALIAS = "legacy_alias" + DISABLED = "disabled" + BLOCKED_BY_CHANNEL = "blocked_by_channel" + UNSAFE_OVERRIDE = "unsafe_override" + NOT_REQUESTED = "not_requested" + + +@dataclass(frozen=True) +class FeatureSpec: + name: str + available_in: RolloutChannel + default_enabled_in: RolloutChannel | None = None + legacy_env: tuple[str, ...] = () + description: str = "" + + def default_enabled(self, channel: RolloutChannel) -> bool: + return self.default_enabled_in is not None and channel.allows(self.default_enabled_in) + + +FEATURES: dict[str, FeatureSpec] = { + "tool_result_interceptors": FeatureSpec( + name="tool_result_interceptors", + available_in=RolloutChannel.CANARY, + legacy_env=("HEADROOM_INTERCEPT_ENABLED",), + description="AST-aware Read/tool-result interceptors used before compression.", + ), + "proxy_output_shaper": FeatureSpec( + name="proxy_output_shaper", + available_in=RolloutChannel.BETA, + legacy_env=("HEADROOM_OUTPUT_SHAPER",), + description="Proxy output-shaping path for response-side experiments.", + ), + "read_maturation": FeatureSpec( + name="read_maturation", + available_in=RolloutChannel.BETA, + legacy_env=("HEADROOM_READ_MATURATION",), + description="Hold-back Read maturation before provider cache entry.", + ), +} + + +def _split_names(raw: str | None) -> set[str]: + if not raw: + return set() + return { + part.strip().lower().replace("-", "_") + for part in raw.replace(";", ",").split(",") + if part.strip() + } + + +def _truthy(value: str | None) -> bool: + return bool(value and value.strip().lower() in _TRUE_VALUES) + + +def _falsey(value: str | None) -> bool: + return bool(value and value.strip().lower() in _FALSE_VALUES) + + +def _canonical_json(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +def _sha256(value: object) -> str: + return "sha256:" + hashlib.sha256(_canonical_json(value).encode()).hexdigest() + + +def registry_digest(registry: Mapping[str, FeatureSpec] = FEATURES) -> str: + """Return a stable identity for all behavior-affecting registry fields.""" + + canonical = [ + { + "name": spec.name, + "available_in": spec.available_in.value, + "default_enabled_in": ( + spec.default_enabled_in.value if spec.default_enabled_in is not None else None + ), + "legacy_env": sorted(spec.legacy_env), + } + for _, spec in sorted(registry.items()) + ] + return _sha256(canonical) + + +@dataclass(frozen=True) +class RolloutConfig: + channel: RolloutChannel + requested: frozenset[str] + disabled: frozenset[str] + unsafe_allow_unstable: bool + # Retain source provenance so a supported live compatibility alias can be + # re-resolved without erasing a generic kill switch or converting an + # explicit request into an alias request. These stay out of the JSON schema. + explicit_requested: frozenset[str] = frozenset() + explicit_disabled: frozenset[str] = frozenset() + legacy_requested: frozenset[str] = frozenset() + legacy_disabled: frozenset[str] = frozenset() + + +@dataclass(frozen=True) +class FeatureDecision: + name: str + available_in: RolloutChannel + default_enabled_in: RolloutChannel | None + requested: bool + disabled: bool + enabled: bool + reason: FeatureDecisionReason + + def to_dict(self) -> dict[str, object]: + return { + "name": self.name, + "available_in": self.available_in.value, + "default_enabled_in": ( + self.default_enabled_in.value if self.default_enabled_in is not None else None + ), + "requested": self.requested, + "disabled": self.disabled, + "enabled": self.enabled, + "decision": self.reason.value, + } + + +@dataclass(frozen=True) +class RolloutSnapshot: + schema_version: int + policy_version: str + registry_digest: str + config: RolloutConfig + decisions: tuple[FeatureDecision, ...] + + @property + def channel(self) -> RolloutChannel: + return self.config.channel + + @property + def unsafe_allow_unstable(self) -> bool: + return self.config.unsafe_allow_unstable + + @property + def qualification_eligible(self) -> bool: + return not self.unsafe_allow_unstable + + @property + def snapshot_digest(self) -> str: + return _sha256(self._canonical_dict()) + + def decision(self, feature: str) -> FeatureDecision: + normalized = feature.strip().lower().replace("-", "_") + for decision in self.decisions: + if decision.name == normalized: + return decision + raise KeyError(feature) + + def is_available(self, feature: str) -> bool: + decision = self.decision(feature) + return self.channel.allows(decision.available_in) or self.unsafe_allow_unstable + + def is_enabled(self, feature: str, **_: object) -> bool: + """Return the pre-resolved decision; extra legacy kwargs are ignored.""" + + return self.decision(feature).enabled + + @property + def enabled(self) -> frozenset[str]: + return frozenset(item.name for item in self.decisions if item.enabled) + + @property + def disabled(self) -> frozenset[str]: + return self.config.disabled + + def _canonical_dict(self) -> dict[str, object]: + return { + "schema_version": self.schema_version, + "policy_version": self.policy_version, + "channel": self.channel.value, + "unsafe_override": self.unsafe_allow_unstable, + "registry_digest": self.registry_digest, + "features": [item.to_dict() for item in self.decisions], + } + + def to_dict(self) -> dict[str, object]: + result = self._canonical_dict() + result["snapshot_digest"] = self.snapshot_digest + result["qualification_eligible"] = self.qualification_eligible + if not self.qualification_eligible: + result["qualification_ineligible_reason"] = "unsafe_rollout_override_active" + return result + + def to_internal_dict(self) -> dict[str, object]: + """Serialize source-separated state for trusted worker handoff.""" + + return { + "schema_version": self.schema_version, + "policy_version": self.policy_version, + "registry_digest": self.registry_digest, + "snapshot_digest": self.snapshot_digest, + "channel": self.channel.value, + "unsafe_allow_unstable": self.unsafe_allow_unstable, + "explicit_requested": sorted(self.config.explicit_requested), + "explicit_disabled": sorted(self.config.explicit_disabled), + "legacy_requested": sorted(self.config.legacy_requested), + "legacy_disabled": sorted(self.config.legacy_disabled), + } + + @classmethod + def from_internal_dict(cls, value: Mapping[str, object]) -> RolloutSnapshot: + """Validate and restore a snapshot serialized for worker handoff.""" + + if not isinstance(value, Mapping): + raise RolloutConfigurationError("invalid rollout worker snapshot") + try: + if value.get("schema_version") != ROLLOUT_SCHEMA_VERSION: + raise RolloutConfigurationError("unsupported rollout worker schema version") + if value.get("policy_version") != ROLLOUT_POLICY_VERSION: + raise RolloutConfigurationError("rollout worker policy version mismatch") + channel = RolloutChannel.parse(str(value["channel"]), strict=True) + unsafe = value["unsafe_allow_unstable"] + if not isinstance(unsafe, bool): + raise RolloutConfigurationError("invalid rollout worker unsafe override") + + def names(field: str) -> set[str]: + raw = value[field] + if not isinstance(raw, list) or not all(isinstance(item, str) for item in raw): + raise RolloutConfigurationError(f"invalid rollout worker field {field!r}") + return set(_validate_names(set(raw), source=field, strict=True)) + + snapshot = _resolve_snapshot( + channel=channel, + explicit_requested=names("explicit_requested"), + explicit_disabled=names("explicit_disabled"), + legacy_requested=names("legacy_requested"), + legacy_disabled=names("legacy_disabled"), + unsafe=unsafe, + ) + except (KeyError, TypeError) as exc: + raise RolloutConfigurationError("invalid rollout worker snapshot") from exc + if value.get("registry_digest") != snapshot.registry_digest: + raise RolloutConfigurationError("rollout worker registry digest mismatch") + if value.get("snapshot_digest") != snapshot.snapshot_digest: + raise RolloutConfigurationError("rollout worker snapshot digest mismatch") + return snapshot + + def with_legacy_env(self, environ: Mapping[str, str]) -> RolloutSnapshot: + """Return a new snapshot after applying supplied legacy alias values. + + This intentionally supports existing hot-reloadable aliases without + re-reading ambient process state or weakening named disable precedence. + Both the old and new snapshots remain immutable, so requests observe a + complete policy rather than partially updated fields. + """ + + legacy_requested = set(self.config.legacy_requested) + legacy_disabled = set(self.config.legacy_disabled) + for spec in FEATURES.values(): + for alias in spec.legacy_env: + if alias not in environ: + continue + legacy_requested.discard(spec.name) + legacy_disabled.discard(spec.name) + if _truthy(environ[alias]): + legacy_requested.add(spec.name) + elif _falsey(environ[alias]): + legacy_disabled.add(spec.name) + return _resolve_snapshot( + channel=self.channel, + explicit_requested=set(self.config.explicit_requested), + explicit_disabled=set(self.config.explicit_disabled), + legacy_requested=legacy_requested, + legacy_disabled=legacy_disabled, + unsafe=self.unsafe_allow_unstable, + ) + + +def _validate_names(names: set[str], *, source: str, strict: bool) -> frozenset[str]: + unknown = sorted(names - FEATURES.keys()) + if unknown: + valid = ", ".join(sorted(FEATURES)) + message = f"unknown rollout feature(s) in {source}: {', '.join(unknown)}; valid: {valid}" + if strict: + raise RolloutConfigurationError(message) + logger.warning("%s; ignoring unknown names (fail-closed)", message) + return frozenset(names & FEATURES.keys()) + + +def resolve_rollout( + environ: Mapping[str, str] | None = None, + *, + requested: Iterable[str] = (), + disabled: Iterable[str] = (), + strict: bool = False, +) -> RolloutSnapshot: + """Resolve all rollout inputs exactly once into an immutable snapshot.""" + + env = os.environ if environ is None else environ + channel = RolloutChannel.parse(env.get("HEADROOM_ROLLOUT_CHANNEL"), strict=strict) + requested_names = _split_names(env.get("HEADROOM_FEATURES")) | { + normalized for name in requested if (normalized := name.strip().lower().replace("-", "_")) + } + disabled_names = _split_names(env.get("HEADROOM_DISABLE_FEATURES")) | { + normalized for name in disabled if (normalized := name.strip().lower().replace("-", "_")) + } + requested_names = set( + _validate_names(requested_names, source="requested features", strict=strict) + ) + disabled_names = set(_validate_names(disabled_names, source="disabled features", strict=strict)) + legacy_requested: set[str] = set() + legacy_disabled: set[str] = set() + for spec in FEATURES.values(): + for alias in spec.legacy_env: + if _truthy(env.get(alias)): + legacy_requested.add(spec.name) + elif _falsey(env.get(alias)): + legacy_disabled.add(spec.name) + + unsafe = _truthy(env.get("HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES")) + return _resolve_snapshot( + channel=channel, + explicit_requested=requested_names, + explicit_disabled=disabled_names, + legacy_requested=legacy_requested, + legacy_disabled=legacy_disabled, + unsafe=unsafe, + ) + + +def _resolve_snapshot( + *, + channel: RolloutChannel, + explicit_requested: set[str], + explicit_disabled: set[str], + legacy_requested: set[str], + legacy_disabled: set[str], + unsafe: bool, +) -> RolloutSnapshot: + """Resolve validated, source-separated inputs into one snapshot.""" + + requested_names = explicit_requested | legacy_requested + disabled_names = explicit_disabled | legacy_disabled + config = RolloutConfig( + channel=channel, + requested=frozenset(requested_names), + disabled=frozenset(disabled_names), + unsafe_allow_unstable=unsafe, + explicit_requested=frozenset(explicit_requested), + explicit_disabled=frozenset(explicit_disabled), + legacy_requested=frozenset(legacy_requested), + legacy_disabled=frozenset(legacy_disabled), + ) + decisions: list[FeatureDecision] = [] + for name, spec in sorted(FEATURES.items()): + is_requested = name in config.requested + is_disabled = name in config.disabled + normally_available = channel.allows(spec.available_in) + if is_disabled: + enabled, reason = False, FeatureDecisionReason.DISABLED + elif is_requested and not normally_available and not unsafe: + enabled, reason = False, FeatureDecisionReason.BLOCKED_BY_CHANNEL + elif is_requested and not normally_available and unsafe: + enabled, reason = True, FeatureDecisionReason.UNSAFE_OVERRIDE + elif name in legacy_requested: + enabled, reason = True, FeatureDecisionReason.LEGACY_ALIAS + elif name in explicit_requested: + enabled, reason = True, FeatureDecisionReason.EXPLICIT + elif spec.default_enabled(channel): + enabled, reason = True, FeatureDecisionReason.DEFAULT + else: + enabled, reason = False, FeatureDecisionReason.NOT_REQUESTED + decisions.append( + FeatureDecision( + name=name, + available_in=spec.available_in, + default_enabled_in=spec.default_enabled_in, + requested=is_requested, + disabled=is_disabled, + enabled=enabled, + reason=reason, + ) + ) + return RolloutSnapshot( + schema_version=ROLLOUT_SCHEMA_VERSION, + policy_version=ROLLOUT_POLICY_VERSION, + registry_digest=registry_digest(), + config=config, + decisions=tuple(decisions), + ) + + +def current_rollout(environ: Mapping[str, str] | None = None) -> RolloutSnapshot: + """Compatibility name for resolving a snapshot at a composition boundary.""" + + return resolve_rollout(environ) + + +def feature_enabled( + feature: str, + *, + explicit: bool = False, + environ: Mapping[str, str] | None = None, +) -> bool: + """Compatibility helper for composition roots; do not use in deep components.""" + + requested = (feature,) if explicit else () + return resolve_rollout(environ, requested=requested).is_enabled(feature) + + +# The PR was never released, but this narrow source alias keeps in-branch callers +# importable while the correction migrates them. It is intentionally undocumented. +Rollout = RolloutSnapshot diff --git a/headroom/transforms/pipeline.py b/headroom/transforms/pipeline.py index 769580d42..305e888f6 100644 --- a/headroom/transforms/pipeline.py +++ b/headroom/transforms/pipeline.py @@ -138,15 +138,10 @@ class TransformPipeline: # 0. Tool-result interceptors (ast-grep Read outline, etc.) run first # so downstream compressors operate on the already-shrunk content. - # OPT-IN: enable via HeadroomConfig.intercept_tool_results, or for - # non-config callers (CLI / SDK / tests) the env var - # HEADROOM_INTERCEPT_ENABLED=1. Off by default while this ships — lets - # users try it and compare before we make it the default. - import os as _os - - if getattr(self.config, "intercept_tool_results", False) or _os.environ.get( - "HEADROOM_INTERCEPT_ENABLED" - ): + # Rollout was resolved once by HeadroomConfig. Never re-read process + # environment here: this pipeline must match its recorded provenance. + assert self.config.rollout is not None + if self.config.rollout.is_enabled("tool_result_interceptors"): from headroom.proxy.interceptors import ToolResultInterceptorTransform transforms.append(ToolResultInterceptorTransform()) diff --git a/scripts/pr-governance.py b/scripts/pr-governance.py index cfda86b3a..8c5165e39 100644 --- a/scripts/pr-governance.py +++ b/scripts/pr-governance.py @@ -22,6 +22,7 @@ REQUIRED_SECTIONS = ( "Changes Made", "Testing", "Real Behavior Proof", + "Runtime Rollout Safety", "Review Readiness", ) PROOF_FIELDS = ( @@ -30,6 +31,15 @@ PROOF_FIELDS = ( "Observed result", "Not tested", ) +ROLLOUT_FIELDS = ( + "Rollout-managed feature(s)", + "Minimum rollout channel", + "Stable/default behavior changed", + "Kill switch / disable path", + "Unsafe override required", + "Qualification impact", + "Rollback path", +) SECTION_RE = re.compile(r"^##\s+(.+?)\s*$", re.MULTILINE) CHECKBOX_RE = re.compile(r"^- \[(?P[ xX])\] (?P