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=<comma-separated feature names>`; 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 <jd@jds-macbook-air.tail2a279.ts.net> Co-authored-by: JD Davis <jd@JDH-AIR-00.local>
This commit is contained in:
JD Davis
@@ -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
|
||||
|
||||
@@ -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"}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<Self, Self::Err> {
|
||||
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<RolloutChannel>,
|
||||
}
|
||||
|
||||
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<String>,
|
||||
pub disabled: BTreeSet<String>,
|
||||
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<RolloutChannel>,
|
||||
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<FeatureDecision>,
|
||||
}
|
||||
|
||||
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<String> {
|
||||
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<String>) -> BTreeSet<String> {
|
||||
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<String> {
|
||||
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<str>) -> 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<PolicyVector> = 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
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<Duration, String> {
|
||||
humantime::parse_duration(s).map_err(|e| format!("invalid duration `{s}`: {e}"))
|
||||
}
|
||||
|
||||
fn parse_rollout_channel(value: &str) -> Result<String, String> {
|
||||
value
|
||||
.parse::<RolloutChannel>()
|
||||
.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<String, String> {
|
||||
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::<Vec<_>>().join(", ")
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_bytes(s: &str) -> Result<u64, String> {
|
||||
s.parse::<bytesize::ByteSize>()
|
||||
.map(|b| b.as_u64())
|
||||
@@ -548,6 +620,8 @@ fn parse_bytes(s: &str) -> Result<u64, String> {
|
||||
/// 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<AppState>) -> Json<serde_json::Value> {
|
||||
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<AppState>) -> 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,6 +28,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
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"
|
||||
);
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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`)
|
||||
|
||||
@@ -58,6 +58,7 @@
|
||||
"architecture",
|
||||
"ci-cd-flows",
|
||||
"releases",
|
||||
"runtime-rollouts",
|
||||
"benchmarks",
|
||||
"limitations",
|
||||
"---Help---",
|
||||
|
||||
@@ -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
|
||||
```
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
@@ -25,6 +25,7 @@ from . import ( # noqa: F401
|
||||
perf,
|
||||
proxy,
|
||||
recover,
|
||||
rollout,
|
||||
tools,
|
||||
update,
|
||||
wrap,
|
||||
|
||||
@@ -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(
|
||||
|
||||
+34
-8
@@ -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.")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
+40
-5
@@ -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,
|
||||
|
||||
@@ -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}"
|
||||
)
|
||||
+12
-1
@@ -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.
|
||||
|
||||
@@ -264,7 +264,7 @@
|
||||
<template x-if="!stats.tokens?.output_reduction?.available">
|
||||
<div class="mt-1 text-xs text-gray-500 leading-relaxed">
|
||||
<span class="text-2xl font-light tabular-nums text-gray-600">—</span>
|
||||
<div class="mt-1">Enable the output shaper (HEADROOM_OUTPUT_SHAPER=1) and run
|
||||
<div class="mt-1">Enable the beta output shaper (HEADROOM_ROLLOUT_CHANNEL=beta HEADROOM_OUTPUT_SHAPER=1) and run
|
||||
<code class="text-gray-400">headroom learn --verbosity --apply</code> to start measuring.</div>
|
||||
</div>
|
||||
</template>
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
@@ -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())
|
||||
|
||||
@@ -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<checked>[ xX])\] (?P<label>.+)$", re.MULTILINE)
|
||||
@@ -193,6 +203,12 @@ def validate_pull_request(event: dict[str, Any]) -> GovernanceReport:
|
||||
if proof_section and not proof_values.get(field_name):
|
||||
problems.append(f"Fill in `Real Behavior Proof` → `{field_name}`.")
|
||||
|
||||
rollout_section = sections.get("Runtime Rollout Safety", "")
|
||||
rollout_values = proof_field_values(rollout_section)
|
||||
for field_name in ROLLOUT_FIELDS:
|
||||
if rollout_section and not rollout_values.get(field_name):
|
||||
problems.append(f"Fill in `Runtime Rollout Safety` → `{field_name}`.")
|
||||
|
||||
readiness_checked = normalize_checkbox_map(checked_items(sections.get("Review Readiness", "")))
|
||||
has_self_review = "i have performed a self-review" in readiness_checked
|
||||
has_ready_checkbox = "this pr is ready for human review" in readiness_checked
|
||||
|
||||
@@ -64,6 +64,16 @@ pytest scripts/tests/test_pr_governance.py -q
|
||||
- Observed result: The governance check fails and the PR gets a needs-author-action label.
|
||||
- Not tested: Automatic Copilot review rulesets in repository settings.
|
||||
|
||||
## Runtime Rollout Safety
|
||||
|
||||
- Rollout-managed feature(s): None.
|
||||
- Minimum rollout channel: Stable.
|
||||
- Stable/default behavior changed: No.
|
||||
- Kill switch / disable path: Not applicable.
|
||||
- Unsafe override required: No.
|
||||
- Qualification impact: None.
|
||||
- Rollback path: Revert the workflow and script changes.
|
||||
|
||||
## Review Readiness
|
||||
|
||||
- [x] I have performed a self-review
|
||||
@@ -186,6 +196,16 @@ Fixes #123
|
||||
- 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
|
||||
@@ -201,6 +221,7 @@ Fixes #123
|
||||
assert any("Type of Change" in problem for problem in report.problems)
|
||||
assert any("Test Output" in problem for problem in report.problems)
|
||||
assert any("Real Behavior Proof" in problem for problem in report.problems)
|
||||
assert any("Runtime Rollout Safety" in problem for problem in report.problems)
|
||||
|
||||
|
||||
def test_validate_pull_request_skips_bot_authored_prs() -> None:
|
||||
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
[
|
||||
{"channel":"stable","requested":false,"disabled":false,"unsafe":false,"enabled":false,"decision":"not_requested"},
|
||||
{"channel":"stable","requested":true,"disabled":false,"unsafe":false,"enabled":false,"decision":"blocked_by_channel"},
|
||||
{"channel":"canary","requested":true,"disabled":false,"unsafe":false,"enabled":true,"decision":"explicit"},
|
||||
{"channel":"stable","requested":true,"disabled":false,"unsafe":true,"enabled":true,"decision":"unsafe_override"},
|
||||
{"channel":"stable","requested":true,"disabled":true,"unsafe":true,"enabled":false,"decision":"disabled"},
|
||||
{"channel":"dev","requested":false,"disabled":false,"unsafe":false,"enabled":false,"decision":"not_requested"}
|
||||
]
|
||||
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
@@ -472,3 +473,59 @@ def test_learn_target_ignored_for_unsupported_agent(
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Note: --target is not supported for codex" in result.output
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("enabled", "expected"), [(True, "live"), (False, "blocked")])
|
||||
def test_activate_output_shaper_reports_effective_rollout_decision(
|
||||
monkeypatch: pytest.MonkeyPatch, enabled: bool, expected: str
|
||||
) -> None:
|
||||
import urllib.request
|
||||
|
||||
from headroom.cli.learn import _activate_output_shaper
|
||||
|
||||
class Response:
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
return False
|
||||
|
||||
def read(self) -> bytes:
|
||||
return json.dumps(
|
||||
{
|
||||
"rollout": {
|
||||
"features": [
|
||||
{"name": "proxy_output_shaper", "enabled": enabled},
|
||||
]
|
||||
}
|
||||
}
|
||||
).encode()
|
||||
|
||||
monkeypatch.setattr(urllib.request, "urlopen", lambda *args, **kwargs: Response())
|
||||
|
||||
status, port = _activate_output_shaper(9876)
|
||||
|
||||
assert status == expected
|
||||
assert port == 9876
|
||||
|
||||
|
||||
def test_activate_output_shaper_handles_malformed_response(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
import urllib.request
|
||||
|
||||
from headroom.cli.learn import _activate_output_shaper
|
||||
|
||||
class Response:
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
return False
|
||||
|
||||
def read(self) -> bytes:
|
||||
return b"not-json"
|
||||
|
||||
monkeypatch.setattr(urllib.request, "urlopen", lambda *args, **kwargs: Response())
|
||||
|
||||
assert _activate_output_shaper(9876) == ("error", 9876)
|
||||
|
||||
@@ -265,6 +265,17 @@ def test_build_manifest_persists_intercept_tool_results() -> None:
|
||||
manifest = build_manifest(**_base_manifest_kwargs(intercept_tool_results=True))
|
||||
|
||||
assert "--intercept-tool-results" in manifest.proxy_args
|
||||
assert manifest.base_env["HEADROOM_ROLLOUT_CHANNEL"] == "canary"
|
||||
|
||||
|
||||
def test_build_manifest_rejects_interceptor_below_required_rollout_channel() -> None:
|
||||
with pytest.raises(click.ClickException, match="requires HEADROOM_ROLLOUT_CHANNEL=canary"):
|
||||
build_manifest(
|
||||
**_base_manifest_kwargs(
|
||||
intercept_tool_results=True,
|
||||
extra_env={"HEADROOM_ROLLOUT_CHANNEL": "stable"},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_build_manifest_persists_protect_tool_results() -> None:
|
||||
|
||||
@@ -10,10 +10,19 @@ pytest.importorskip("fastapi")
|
||||
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
|
||||
from headroom.proxy import runtime_env # noqa: E402
|
||||
from headroom.proxy.loopback_guard import require_loopback # noqa: E402
|
||||
from headroom.proxy.server import ProxyConfig, create_app # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolate_runtime_env_overrides():
|
||||
"""Keep loopback hot-reload state from leaking into later test modules."""
|
||||
runtime_env.clear_overrides()
|
||||
yield
|
||||
runtime_env.clear_overrides()
|
||||
|
||||
|
||||
def _make_client() -> TestClient:
|
||||
app = create_app(
|
||||
ProxyConfig(
|
||||
@@ -49,6 +58,7 @@ async def _ok_response(
|
||||
|
||||
def test_http_responses_output_shaper_rewrites_and_labels(monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_OUTPUT_SHAPER", "1")
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "beta")
|
||||
monkeypatch.setenv("HEADROOM_VERBOSITY_LEVEL", "2")
|
||||
monkeypatch.delenv("HEADROOM_OUTPUT_HOLDOUT", raising=False)
|
||||
captured: dict[str, Any] = {}
|
||||
@@ -104,6 +114,7 @@ def test_http_responses_output_shaper_rewrites_and_labels(monkeypatch):
|
||||
|
||||
def test_http_responses_output_shaper_respects_bypass(monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_OUTPUT_SHAPER", "1")
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "beta")
|
||||
captured: dict[str, Any] = {}
|
||||
payload = {"model": "gpt-5", "input": "hi"}
|
||||
|
||||
@@ -131,6 +142,7 @@ def test_http_responses_output_shaper_respects_bypass(monkeypatch):
|
||||
|
||||
def test_http_responses_output_shaper_holdout_labels_without_rewrite(monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_OUTPUT_SHAPER", "1")
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "beta")
|
||||
monkeypatch.setenv("HEADROOM_OUTPUT_HOLDOUT", "1")
|
||||
captured: dict[str, Any] = {}
|
||||
outcomes: list[Any] = []
|
||||
@@ -160,3 +172,43 @@ def test_http_responses_output_shaper_holdout_labels_without_rewrite(monkeypatch
|
||||
transforms = outcomes[-1].transforms_applied
|
||||
assert any(t.startswith("output_shaper:control:") for t in transforms)
|
||||
assert "output_shaper:verbosity:L2" not in transforms
|
||||
|
||||
|
||||
def test_http_output_shaper_hot_reload_changes_the_running_request_path(monkeypatch):
|
||||
"""The admin endpoint must not report success while traffic stays unchanged."""
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "beta")
|
||||
monkeypatch.delenv("HEADROOM_OUTPUT_SHAPER", raising=False)
|
||||
payload = {
|
||||
"model": "gpt-5",
|
||||
"input": [{"type": "function_call_output", "call_id": "call_1", "output": "ok"}],
|
||||
"reasoning": {"effort": "high"},
|
||||
"text": {"verbosity": "medium"},
|
||||
}
|
||||
sent: list[dict[str, Any]] = []
|
||||
|
||||
with _make_client() as client:
|
||||
proxy = client.app.state.proxy
|
||||
|
||||
async def _fake_retry(*args: Any, **kwargs: Any) -> httpx.Response:
|
||||
sent.append(copy.deepcopy(args[3]))
|
||||
return await _ok_response(*args, **kwargs)
|
||||
|
||||
proxy._retry_request = _fake_retry
|
||||
first = client.post(
|
||||
"/v1/responses", headers={"authorization": "Bearer test-key"}, json=payload
|
||||
)
|
||||
update = client.post("/admin/runtime-env", json={"HEADROOM_OUTPUT_SHAPER": "1"})
|
||||
second = client.post(
|
||||
"/v1/responses", headers={"authorization": "Bearer test-key"}, json=payload
|
||||
)
|
||||
|
||||
assert first.status_code == second.status_code == update.status_code == 200
|
||||
assert sent[0] == payload
|
||||
assert "<headroom_output_shaping>" in sent[1]["instructions"]
|
||||
decision = next(
|
||||
item
|
||||
for item in update.json()["rollout"]["features"]
|
||||
if item["name"] == "proxy_output_shaper"
|
||||
)
|
||||
assert decision["enabled"] is True
|
||||
assert decision["decision"] == "legacy_alias"
|
||||
|
||||
@@ -275,6 +275,7 @@ class TestWorkerConfiguration:
|
||||
payload = json.loads(os.environ[_MULTI_WORKER_CONFIG_ENV])
|
||||
assert payload["host"] == "0.0.0.0"
|
||||
assert payload["port"] == 8787
|
||||
assert payload["worker_processes"] == 4
|
||||
assert payload["max_connections"] == 200
|
||||
assert payload["http_proxy"] == "http://proxy.local:8080"
|
||||
finally:
|
||||
|
||||
@@ -5,6 +5,7 @@ from fastapi.testclient import TestClient
|
||||
|
||||
from headroom.proxy.models import ProxyConfig
|
||||
from headroom.proxy.server import create_app
|
||||
from headroom.rollout import resolve_rollout
|
||||
|
||||
|
||||
class FakeRequestLogger:
|
||||
@@ -28,6 +29,35 @@ class FakeLogEntry(dict[str, object]):
|
||||
return self.get(name)
|
||||
|
||||
|
||||
def test_stats_exposes_actual_running_rollout_snapshot() -> None:
|
||||
rollout = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "canary",
|
||||
"HEADROOM_FEATURES": "tool_result_interceptors",
|
||||
}
|
||||
)
|
||||
app = create_app(
|
||||
ProxyConfig(
|
||||
rollout=rollout,
|
||||
optimize=False,
|
||||
cache_enabled=False,
|
||||
rate_limit_enabled=False,
|
||||
cost_tracking_enabled=False,
|
||||
log_requests=False,
|
||||
ccr_inject_tool=False,
|
||||
ccr_handle_responses=False,
|
||||
ccr_context_tracking=False,
|
||||
http2=False,
|
||||
)
|
||||
)
|
||||
|
||||
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
|
||||
payload = client.get("/stats").json()["rollout"]
|
||||
|
||||
assert payload == rollout.to_dict()
|
||||
assert payload["qualification_eligible"] is True
|
||||
|
||||
|
||||
def test_stats_refreshes_recent_requests_when_cached() -> None:
|
||||
app = create_app(
|
||||
ProxyConfig(
|
||||
|
||||
@@ -360,6 +360,7 @@ def test_read_maturation_knobs_from_env(monkeypatch):
|
||||
monkeypatch.delenv(_MULTI_WORKER_CONFIG_ENV, raising=False)
|
||||
|
||||
monkeypatch.setenv("HEADROOM_READ_MATURATION", "1")
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "beta")
|
||||
monkeypatch.setenv("HEADROOM_READ_MATURATION_QUIESCE_TURNS", "3")
|
||||
monkeypatch.setenv("HEADROOM_READ_MATURATION_MAX_HOLD_TURNS", "10")
|
||||
monkeypatch.setenv("HEADROOM_READ_MATURATION_MIN_SIZE_BYTES", "4096")
|
||||
@@ -370,3 +371,18 @@ def test_read_maturation_knobs_from_env(monkeypatch):
|
||||
assert cfg.read_maturation_quiesce_turns == 3
|
||||
assert cfg.read_maturation_max_hold_turns == 10
|
||||
assert cfg.read_maturation_min_size_bytes == 4096
|
||||
|
||||
|
||||
def test_read_maturation_env_cannot_bypass_stable_rollout(monkeypatch):
|
||||
"""Every env-driven server composition root must enforce the beta gate."""
|
||||
from headroom.proxy.server import _MULTI_WORKER_CONFIG_ENV, _proxy_config_from_env
|
||||
|
||||
monkeypatch.delenv(_MULTI_WORKER_CONFIG_ENV, raising=False)
|
||||
monkeypatch.setenv("HEADROOM_READ_MATURATION", "1")
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "stable")
|
||||
|
||||
cfg = _proxy_config_from_env()
|
||||
|
||||
assert cfg.read_maturation is False
|
||||
assert cfg.rollout is not None
|
||||
assert cfg.rollout.decision("read_maturation").reason.value == "blocked_by_channel"
|
||||
|
||||
@@ -0,0 +1,465 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from headroom.cli.main import main
|
||||
from headroom.config import HeadroomConfig
|
||||
from headroom.rollout import (
|
||||
FEATURES,
|
||||
FeatureDecisionReason,
|
||||
FeatureSpec,
|
||||
RolloutChannel,
|
||||
RolloutConfigurationError,
|
||||
RolloutSnapshot,
|
||||
current_rollout,
|
||||
feature_enabled,
|
||||
registry_digest,
|
||||
resolve_rollout,
|
||||
)
|
||||
from headroom.transforms.pipeline import TransformPipeline
|
||||
|
||||
|
||||
def test_default_stable_resolution_is_versioned_and_eligible() -> None:
|
||||
snapshot = resolve_rollout({})
|
||||
|
||||
assert snapshot.channel is RolloutChannel.STABLE
|
||||
assert snapshot.schema_version == 1
|
||||
assert snapshot.policy_version == "1"
|
||||
assert snapshot.qualification_eligible is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("channel", ["beta", "canary", "dev"])
|
||||
def test_valid_rollout_channels(channel: str) -> None:
|
||||
assert resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": channel}).channel.value == channel
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("alias", "expected"),
|
||||
[
|
||||
("prod", RolloutChannel.STABLE),
|
||||
("production", RolloutChannel.STABLE),
|
||||
("preview", RolloutChannel.BETA),
|
||||
("nightly", RolloutChannel.CANARY),
|
||||
("development", RolloutChannel.DEV),
|
||||
],
|
||||
)
|
||||
def test_channel_aliases(alias: str, expected: RolloutChannel) -> None:
|
||||
assert RolloutChannel.parse(alias) is expected
|
||||
|
||||
|
||||
def test_strict_channel_configuration_rejects_unknown_input() -> None:
|
||||
with pytest.raises(RolloutConfigurationError, match="unknown rollout channel"):
|
||||
resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": "stabel"}, strict=True)
|
||||
|
||||
|
||||
def test_unknown_channel_fails_closed_with_diagnostic(caplog: pytest.LogCaptureFixture) -> None:
|
||||
with caplog.at_level(logging.WARNING):
|
||||
snapshot = resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": "stabel"})
|
||||
|
||||
assert snapshot.channel is RolloutChannel.STABLE
|
||||
assert "unknown rollout channel 'stabel'; falling back to 'stable'" in caplog.text
|
||||
|
||||
|
||||
def test_unknown_requested_and_disabled_features_fail_closed_and_warn(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
with caplog.at_level(logging.WARNING):
|
||||
snapshot = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_FEATURES": "typo_requested",
|
||||
"HEADROOM_DISABLE_FEATURES": "typo_disabled",
|
||||
}
|
||||
)
|
||||
|
||||
assert snapshot.config.requested == frozenset()
|
||||
assert snapshot.config.disabled == frozenset()
|
||||
assert "typo_requested" in caplog.text
|
||||
assert "typo_disabled" in caplog.text
|
||||
|
||||
|
||||
def test_strict_configuration_rejects_unknown_input() -> None:
|
||||
with pytest.raises(RolloutConfigurationError, match="unknown rollout feature"):
|
||||
resolve_rollout({"HEADROOM_FEATURES": "typo"}, strict=True)
|
||||
|
||||
|
||||
def test_stable_blocks_explicit_canary_feature() -> None:
|
||||
snapshot = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "stable",
|
||||
"HEADROOM_FEATURES": "tool-result-interceptors",
|
||||
}
|
||||
)
|
||||
decision = snapshot.decision("tool_result_interceptors")
|
||||
|
||||
assert decision.enabled is False
|
||||
assert decision.reason is FeatureDecisionReason.BLOCKED_BY_CHANNEL
|
||||
|
||||
|
||||
def test_canary_allows_explicit_request() -> None:
|
||||
snapshot = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "canary",
|
||||
"HEADROOM_FEATURES": "tool_result_interceptors",
|
||||
}
|
||||
)
|
||||
|
||||
assert snapshot.decision("tool_result_interceptors").reason is FeatureDecisionReason.EXPLICIT
|
||||
|
||||
|
||||
def test_non_default_feature_remains_off_when_not_requested() -> None:
|
||||
decision = resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": "dev"}).decision(
|
||||
"tool_result_interceptors"
|
||||
)
|
||||
assert decision.enabled is False
|
||||
assert decision.reason is FeatureDecisionReason.NOT_REQUESTED
|
||||
|
||||
|
||||
def test_legacy_alias_obeys_channel_and_has_distinct_reason() -> None:
|
||||
stable = resolve_rollout(
|
||||
{"HEADROOM_ROLLOUT_CHANNEL": "stable", "HEADROOM_INTERCEPT_ENABLED": "1"}
|
||||
)
|
||||
canary = resolve_rollout(
|
||||
{"HEADROOM_ROLLOUT_CHANNEL": "canary", "HEADROOM_INTERCEPT_ENABLED": "1"}
|
||||
)
|
||||
|
||||
assert (
|
||||
stable.decision("tool_result_interceptors").reason
|
||||
is FeatureDecisionReason.BLOCKED_BY_CHANNEL
|
||||
)
|
||||
assert canary.decision("tool_result_interceptors").reason is FeatureDecisionReason.LEGACY_ALIAS
|
||||
|
||||
|
||||
@pytest.mark.parametrize("request_source", ["HEADROOM_FEATURES", "HEADROOM_INTERCEPT_ENABLED"])
|
||||
def test_disable_beats_explicit_and_legacy_request(request_source: str) -> None:
|
||||
snapshot = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "canary",
|
||||
request_source: "tool_result_interceptors"
|
||||
if request_source.endswith("FEATURES")
|
||||
else "1",
|
||||
"HEADROOM_DISABLE_FEATURES": "tool_result_interceptors",
|
||||
}
|
||||
)
|
||||
decision = snapshot.decision("tool_result_interceptors")
|
||||
|
||||
assert decision.enabled is False
|
||||
assert decision.reason is FeatureDecisionReason.DISABLED
|
||||
|
||||
|
||||
def test_unsafe_override_crosses_channel_and_poisons_qualification() -> None:
|
||||
snapshot = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "stable",
|
||||
"HEADROOM_FEATURES": "tool_result_interceptors",
|
||||
"HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES": "1",
|
||||
}
|
||||
)
|
||||
payload = snapshot.to_dict()
|
||||
|
||||
assert (
|
||||
snapshot.decision("tool_result_interceptors").reason
|
||||
is FeatureDecisionReason.UNSAFE_OVERRIDE
|
||||
)
|
||||
assert payload["qualification_eligible"] is False
|
||||
assert payload["qualification_ineligible_reason"] == "unsafe_rollout_override_active"
|
||||
|
||||
|
||||
def test_disable_still_beats_unsafe_override() -> None:
|
||||
snapshot = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_FEATURES": "tool_result_interceptors",
|
||||
"HEADROOM_DISABLE_FEATURES": "tool_result_interceptors",
|
||||
"HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES": "1",
|
||||
}
|
||||
)
|
||||
assert snapshot.decision("tool_result_interceptors").reason is FeatureDecisionReason.DISABLED
|
||||
|
||||
|
||||
def test_live_legacy_reresolution_preserves_channel_and_named_kill_switch() -> None:
|
||||
eligible = resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": "beta"})
|
||||
enabled = eligible.with_legacy_env({"HEADROOM_OUTPUT_SHAPER": "1"})
|
||||
disabled_again = enabled.with_legacy_env({"HEADROOM_OUTPUT_SHAPER": "0"})
|
||||
killed = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "beta",
|
||||
"HEADROOM_DISABLE_FEATURES": "proxy_output_shaper",
|
||||
}
|
||||
).with_legacy_env({"HEADROOM_OUTPUT_SHAPER": "1"})
|
||||
blocked = resolve_rollout({}).with_legacy_env({"HEADROOM_OUTPUT_SHAPER": "1"})
|
||||
|
||||
assert enabled.decision("proxy_output_shaper").reason is FeatureDecisionReason.LEGACY_ALIAS
|
||||
assert disabled_again.decision("proxy_output_shaper").reason is FeatureDecisionReason.DISABLED
|
||||
assert killed.decision("proxy_output_shaper").reason is FeatureDecisionReason.DISABLED
|
||||
assert (
|
||||
blocked.decision("proxy_output_shaper").reason is FeatureDecisionReason.BLOCKED_BY_CHANNEL
|
||||
)
|
||||
assert enabled.snapshot_digest != eligible.snapshot_digest
|
||||
|
||||
|
||||
def test_empty_programmatic_feature_names_are_ignored() -> None:
|
||||
snapshot = resolve_rollout({}, requested=["", " "], disabled=[""])
|
||||
|
||||
assert snapshot.config.requested == frozenset()
|
||||
assert snapshot.config.disabled == frozenset()
|
||||
|
||||
|
||||
def test_multi_worker_config_round_trip_preserves_typed_rollout(monkeypatch) -> None:
|
||||
from headroom.proxy.models import ProxyConfig
|
||||
from headroom.proxy.server import (
|
||||
_MULTI_WORKER_CONFIG_ENV,
|
||||
_proxy_config_from_env,
|
||||
_proxy_config_payload,
|
||||
)
|
||||
|
||||
rollout = resolve_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "beta",
|
||||
"HEADROOM_OUTPUT_SHAPER": "1",
|
||||
"HEADROOM_DISABLE_FEATURES": "read_maturation",
|
||||
}
|
||||
)
|
||||
original = ProxyConfig(rollout=rollout, worker_processes=2)
|
||||
monkeypatch.setenv(_MULTI_WORKER_CONFIG_ENV, json.dumps(_proxy_config_payload(original)))
|
||||
|
||||
restored = _proxy_config_from_env()
|
||||
|
||||
assert restored.rollout is not None
|
||||
assert restored.rollout.to_internal_dict() == rollout.to_internal_dict()
|
||||
assert restored.rollout.is_enabled("proxy_output_shaper") is True
|
||||
assert restored.rollout.is_enabled("read_maturation") is False
|
||||
assert restored.worker_processes == 2
|
||||
|
||||
|
||||
def test_documented_proxy_json_without_internal_snapshot_is_preserved(monkeypatch) -> None:
|
||||
from headroom.proxy.server import _MULTI_WORKER_CONFIG_ENV, _proxy_config_from_env
|
||||
|
||||
monkeypatch.setenv(
|
||||
_MULTI_WORKER_CONFIG_ENV,
|
||||
json.dumps(
|
||||
{
|
||||
"port": 39099,
|
||||
"rate_limit_enabled": False,
|
||||
"proxy_token": "required-token",
|
||||
"offline": True,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
restored = _proxy_config_from_env()
|
||||
|
||||
assert restored.port == 39099
|
||||
assert restored.rate_limit_enabled is False
|
||||
assert restored.proxy_token == "required-token"
|
||||
assert restored.offline is True
|
||||
assert restored.rollout is not None
|
||||
|
||||
|
||||
def test_internal_proxy_json_still_rejects_tampered_rollout_snapshot(monkeypatch) -> None:
|
||||
from headroom.proxy.models import ProxyConfig
|
||||
from headroom.proxy.server import (
|
||||
_MULTI_WORKER_CONFIG_ENV,
|
||||
_proxy_config_from_env,
|
||||
_proxy_config_payload,
|
||||
)
|
||||
|
||||
payload = _proxy_config_payload(ProxyConfig(port=39099))
|
||||
payload["_rollout_snapshot"]["snapshot_digest"] = "sha256:tampered" # type: ignore[index]
|
||||
monkeypatch.setenv(_MULTI_WORKER_CONFIG_ENV, json.dumps(payload))
|
||||
monkeypatch.setenv("HEADROOM_PORT", "39100")
|
||||
|
||||
restored = _proxy_config_from_env()
|
||||
|
||||
assert restored.port == 39100
|
||||
assert restored.rollout is not None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw_config", ["null", "[]", '"not-an-object"'])
|
||||
def test_non_object_proxy_json_falls_back_without_crashing(monkeypatch, raw_config: str) -> None:
|
||||
from headroom.proxy.server import _MULTI_WORKER_CONFIG_ENV, _proxy_config_from_env
|
||||
|
||||
monkeypatch.setenv(_MULTI_WORKER_CONFIG_ENV, raw_config)
|
||||
monkeypatch.setenv("HEADROOM_PORT", "39100")
|
||||
|
||||
restored = _proxy_config_from_env()
|
||||
|
||||
assert restored.port == 39100
|
||||
assert restored.rollout is not None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("mutation", "message"),
|
||||
[
|
||||
({"schema_version": 999}, "schema version"),
|
||||
({"policy_version": "999"}, "policy version"),
|
||||
({"unsafe_allow_unstable": "yes"}, "unsafe override"),
|
||||
({"explicit_requested": "proxy_output_shaper"}, "explicit_requested"),
|
||||
({"registry_digest": "sha256:tampered"}, "registry digest"),
|
||||
({"snapshot_digest": "sha256:tampered"}, "snapshot digest"),
|
||||
],
|
||||
)
|
||||
def test_worker_rollout_handoff_rejects_invalid_or_tampered_state(
|
||||
mutation: dict[str, object], message: str
|
||||
) -> None:
|
||||
payload = resolve_rollout({}).to_internal_dict()
|
||||
payload.update(mutation)
|
||||
|
||||
with pytest.raises(RolloutConfigurationError, match=message):
|
||||
RolloutSnapshot.from_internal_dict(payload)
|
||||
|
||||
|
||||
def test_worker_rollout_handoff_rejects_non_object_state() -> None:
|
||||
with pytest.raises(RolloutConfigurationError, match="invalid rollout worker snapshot"):
|
||||
RolloutSnapshot.from_internal_dict([]) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_snapshot_query_and_compatibility_helpers() -> None:
|
||||
snapshot = current_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "canary",
|
||||
"HEADROOM_FEATURES": "tool_result_interceptors",
|
||||
"HEADROOM_DISABLE_FEATURES": "read_maturation",
|
||||
}
|
||||
)
|
||||
|
||||
assert snapshot.is_available("tool-result-interceptors") is True
|
||||
assert snapshot.enabled == frozenset({"tool_result_interceptors"})
|
||||
assert snapshot.disabled == frozenset({"read_maturation"})
|
||||
assert feature_enabled(
|
||||
"tool_result_interceptors",
|
||||
explicit=True,
|
||||
environ={"HEADROOM_ROLLOUT_CHANNEL": "canary"},
|
||||
)
|
||||
assert not feature_enabled("tool_result_interceptors", environ={})
|
||||
with pytest.raises(KeyError, match="missing"):
|
||||
snapshot.decision("missing")
|
||||
|
||||
|
||||
def test_registry_and_snapshot_digests_are_deterministic_and_policy_sensitive() -> None:
|
||||
first = resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": "canary"})
|
||||
second = resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": "canary"})
|
||||
equivalent = dict(reversed(list(FEATURES.items())))
|
||||
changed = dict(FEATURES)
|
||||
changed["tool_result_interceptors"] = FeatureSpec(
|
||||
"tool_result_interceptors", RolloutChannel.BETA
|
||||
)
|
||||
|
||||
assert first.registry_digest == second.registry_digest == registry_digest(equivalent)
|
||||
assert first.snapshot_digest == second.snapshot_digest
|
||||
assert registry_digest(changed) != first.registry_digest
|
||||
assert json.dumps(first.to_dict(), sort_keys=True) == json.dumps(
|
||||
second.to_dict(), sort_keys=True
|
||||
)
|
||||
|
||||
|
||||
def test_pipeline_uses_config_snapshot_after_environment_mutation(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.delenv("HEADROOM_ROLLOUT_CHANNEL", raising=False)
|
||||
monkeypatch.delenv("HEADROOM_FEATURES", raising=False)
|
||||
config = HeadroomConfig()
|
||||
original_digest = config.rollout.snapshot_digest if config.rollout else None
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "canary")
|
||||
monkeypatch.setenv("HEADROOM_FEATURES", "tool_result_interceptors")
|
||||
|
||||
pipeline = TransformPipeline(config)
|
||||
|
||||
assert config.rollout is not None
|
||||
assert config.rollout.snapshot_digest == original_digest
|
||||
assert all(
|
||||
type(transform).__name__ != "ToolResultInterceptorTransform"
|
||||
for transform in pipeline.transforms
|
||||
)
|
||||
|
||||
|
||||
def test_cli_json_status_and_strict_error() -> None:
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(
|
||||
main,
|
||||
[
|
||||
"rollout",
|
||||
"status",
|
||||
"--channel",
|
||||
"canary",
|
||||
"--features",
|
||||
"tool_result_interceptors",
|
||||
"--json",
|
||||
],
|
||||
)
|
||||
invalid = runner.invoke(main, ["rollout", "status", "--features", "typo", "--json"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
payload = json.loads(result.output)
|
||||
assert payload["channel"] == "canary"
|
||||
assert payload["features"][2]["name"] == "tool_result_interceptors"
|
||||
assert invalid.exit_code != 0
|
||||
assert "unknown rollout feature" in invalid.output
|
||||
|
||||
|
||||
def test_cli_human_status_exercises_disable_and_unsafe_options() -> None:
|
||||
result = CliRunner().invoke(
|
||||
main,
|
||||
[
|
||||
"rollout",
|
||||
"status",
|
||||
"--channel",
|
||||
"stable",
|
||||
"--features",
|
||||
"tool_result_interceptors",
|
||||
"--disable-features",
|
||||
"read_maturation",
|
||||
"--unsafe-allow-unstable-features",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "Rollout channel: stable" in result.output
|
||||
assert "Qualification eligible: false" in result.output
|
||||
assert "tool_result_interceptors: enabled=true decision=unsafe_override" in result.output
|
||||
assert "read_maturation: enabled=false decision=disabled" in result.output
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("option", "message", "required_channel"),
|
||||
[
|
||||
("--read-maturation", "--read-maturation is not available", "beta"),
|
||||
(
|
||||
"--intercept-tool-results",
|
||||
"--intercept-tool-results is not available",
|
||||
"canary",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_proxy_cli_fails_loudly_when_explicit_feature_is_channel_blocked(
|
||||
option: str, message: str, required_channel: str
|
||||
) -> None:
|
||||
result = CliRunner().invoke(
|
||||
main,
|
||||
["proxy", option],
|
||||
env={"HEADROOM_ROLLOUT_CHANNEL": "stable"},
|
||||
)
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert message in result.output
|
||||
assert f"HEADROOM_ROLLOUT_CHANNEL={required_channel}" in result.output
|
||||
|
||||
|
||||
def test_shared_python_rust_policy_vectors() -> None:
|
||||
vectors = json.loads(
|
||||
(Path(__file__).parent / "fixtures" / "rollout_policy_vectors.json").read_text()
|
||||
)
|
||||
for vector in vectors:
|
||||
env = {"HEADROOM_ROLLOUT_CHANNEL": vector["channel"]}
|
||||
if vector["requested"]:
|
||||
env["HEADROOM_FEATURES"] = "tool_result_interceptors"
|
||||
if vector["disabled"]:
|
||||
env["HEADROOM_DISABLE_FEATURES"] = "tool_result_interceptors"
|
||||
if vector["unsafe"]:
|
||||
env["HEADROOM_UNSAFE_ALLOW_UNSTABLE_FEATURES"] = "1"
|
||||
decision = resolve_rollout(env).decision("tool_result_interceptors")
|
||||
assert decision.enabled is vector["enabled"]
|
||||
assert decision.reason.value == vector["decision"]
|
||||
@@ -14,6 +14,7 @@ pytest.importorskip("httpx")
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
|
||||
from headroom.proxy.server import ProxyConfig, create_app # noqa: E402
|
||||
from headroom.rollout import resolve_rollout # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@@ -155,11 +156,78 @@ def test_admin_runtime_env_applies_and_reflects_in_health(loopback_client):
|
||||
assert health["HEADROOM_VERBOSITY_LEVEL"] == "3"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("rollout", "expected_enabled", "expected_reason"),
|
||||
[
|
||||
(resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": "beta"}), True, "legacy_alias"),
|
||||
(resolve_rollout({}), False, "blocked_by_channel"),
|
||||
(
|
||||
resolve_rollout(
|
||||
{
|
||||
"HEADROOM_ROLLOUT_CHANNEL": "beta",
|
||||
"HEADROOM_DISABLE_FEATURES": "proxy_output_shaper",
|
||||
}
|
||||
),
|
||||
False,
|
||||
"disabled",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_admin_runtime_env_reresolves_running_rollout_without_weakening_policy(
|
||||
rollout, expected_enabled, expected_reason
|
||||
):
|
||||
app = create_app(
|
||||
ProxyConfig(
|
||||
rollout=rollout,
|
||||
optimize=False,
|
||||
cache_enabled=False,
|
||||
rate_limit_enabled=False,
|
||||
cost_tracking_enabled=False,
|
||||
)
|
||||
)
|
||||
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
|
||||
before = client.get("/stats?cached=1").json()["rollout"]
|
||||
response = client.post("/admin/runtime-env", json={"HEADROOM_OUTPUT_SHAPER": "1"})
|
||||
after = client.get("/stats?cached=1").json()["rollout"]
|
||||
|
||||
decision = next(item for item in after["features"] if item["name"] == "proxy_output_shaper")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["rollout"] == after
|
||||
assert decision["enabled"] is expected_enabled
|
||||
assert decision["decision"] == expected_reason
|
||||
assert after["snapshot_digest"] != before["snapshot_digest"]
|
||||
|
||||
|
||||
def test_admin_runtime_env_rejects_non_object(loopback_client):
|
||||
resp = loopback_client.post("/admin/runtime-env", json=["not", "a", "dict"])
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_admin_runtime_env_rejects_process_local_update_with_multiple_workers(monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
|
||||
rollout = resolve_rollout({"HEADROOM_ROLLOUT_CHANNEL": "beta"})
|
||||
config = ProxyConfig(
|
||||
worker_processes=2,
|
||||
rollout=rollout,
|
||||
optimize=False,
|
||||
cache_enabled=False,
|
||||
rate_limit_enabled=False,
|
||||
cost_tracking_enabled=False,
|
||||
)
|
||||
app = create_app(config)
|
||||
before_digest = rollout.snapshot_digest
|
||||
|
||||
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
|
||||
response = client.post("/admin/runtime-env", json={"HEADROOM_OUTPUT_SHAPER": "1"})
|
||||
after = client.get("/stats").json()["rollout"]
|
||||
|
||||
assert response.status_code == 409
|
||||
assert response.json()["worker_processes"] == 2
|
||||
assert "restart" in response.json()["error"]
|
||||
assert rt.getenv("HEADROOM_OUTPUT_SHAPER") is None
|
||||
assert after["snapshot_digest"] == before_digest
|
||||
|
||||
|
||||
def test_admin_runtime_env_is_loopback_only():
|
||||
config = ProxyConfig(
|
||||
optimize=False,
|
||||
|
||||
@@ -704,8 +704,9 @@ def test_transform_adapter_tokens_before_is_baseline_not_reconstruction(tokenize
|
||||
|
||||
|
||||
def test_proxy_pipeline_includes_interceptor_when_env_enabled(monkeypatch):
|
||||
"""When HEADROOM_INTERCEPT_ENABLED=1, ToolResultInterceptorTransform is at index 0 in both pipelines."""
|
||||
"""An eligible legacy request installs the interceptor in both pipelines."""
|
||||
monkeypatch.setenv("HEADROOM_INTERCEPT_ENABLED", "1")
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "canary")
|
||||
from headroom.proxy.interceptors import ToolResultInterceptorTransform
|
||||
from headroom.proxy.models import ProxyConfig
|
||||
from headroom.proxy.server import HeadroomProxy
|
||||
@@ -717,6 +718,19 @@ def test_proxy_pipeline_includes_interceptor_when_env_enabled(monkeypatch):
|
||||
assert isinstance(transforms[0], ToolResultInterceptorTransform)
|
||||
|
||||
|
||||
def test_proxy_pipeline_blocks_interceptor_below_rollout_channel(monkeypatch):
|
||||
"""A legacy request cannot bypass the stable rollout-channel boundary."""
|
||||
monkeypatch.setenv("HEADROOM_INTERCEPT_ENABLED", "1")
|
||||
monkeypatch.setenv("HEADROOM_ROLLOUT_CHANNEL", "stable")
|
||||
from headroom.proxy.interceptors import ToolResultInterceptorTransform
|
||||
from headroom.proxy.models import ProxyConfig
|
||||
from headroom.proxy.server import HeadroomProxy
|
||||
|
||||
proxy = HeadroomProxy(ProxyConfig())
|
||||
for pipeline in (proxy.anthropic_pipeline, proxy.openai_pipeline):
|
||||
assert not any(isinstance(t, ToolResultInterceptorTransform) for t in pipeline.transforms)
|
||||
|
||||
|
||||
def test_proxy_pipeline_excludes_interceptor_when_env_not_set(monkeypatch):
|
||||
"""When HEADROOM_INTERCEPT_ENABLED is unset, no interceptor in either pipeline."""
|
||||
monkeypatch.delenv("HEADROOM_INTERCEPT_ENABLED", raising=False)
|
||||
|
||||
@@ -2,6 +2,26 @@
|
||||
|
||||
Headroom can be configured via the SDK, proxy command line, or per-request overrides.
|
||||
|
||||
## Runtime Rollout Channels
|
||||
|
||||
Rollout channels control behaviors in an already-installed artifact. They do
|
||||
not install or select a Headroom release/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. |
|
||||
|
||||
Example:
|
||||
|
||||
```bash
|
||||
export HEADROOM_ROLLOUT_CHANNEL=canary
|
||||
export HEADROOM_FEATURES=tool_result_interceptors
|
||||
headroom proxy --intercept-tool-results
|
||||
```
|
||||
|
||||
## SDK Configuration
|
||||
|
||||
```python
|
||||
|
||||
+5
-4
@@ -169,11 +169,12 @@ Options:
|
||||
|
||||
`headroom learn --verbosity` analyzes past sessions to infer the ideal output verbosity level for your project and writes a `verbosity.json` profile.
|
||||
|
||||
**Important**: the output shaper is **off by default**. Running `--verbosity --apply` will either:
|
||||
- Hot-enable the output shaper on a running proxy (`POST /admin/runtime-env`), OR
|
||||
- Print instructions to set `HEADROOM_OUTPUT_SHAPER=1` before `headroom wrap ...`
|
||||
**Important**: the output shaper is **off by default** and requires the `beta`
|
||||
runtime rollout channel. Running `--verbosity --apply` will either:
|
||||
- Hot-enable the output shaper on an eligible running proxy (`POST /admin/runtime-env`), OR
|
||||
- Print instructions to set `HEADROOM_ROLLOUT_CHANNEL=beta` and `HEADROOM_OUTPUT_SHAPER=1` before `headroom wrap ...`
|
||||
|
||||
To keep the shaper on across proxy restarts, add `export HEADROOM_OUTPUT_SHAPER=1` to your shell profile before starting the proxy.
|
||||
To keep the shaper on across proxy restarts, export both variables before starting the proxy.
|
||||
|
||||
**Flag interactions**:
|
||||
- `--all` and `--project` are mutually exclusive
|
||||
|
||||
Reference in New Issue
Block a user