akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(proxy): dedupe Codex WS request logging for accurate mixed-provider dashboards (#2189)

Browse source 
## Description

Running Claude Code (Anthropic) and Codex (OpenAI) against the **same**
Headroom proxy instance on one port produced incorrect, unstable
dashboard data. The proxy core is provider-isolated and
multi-provider-safe by design; the defect was in the observability
layer. The Codex `/v1/responses` **WebSocket** handler was the only path
in the proxy that wrote to the request logger by hand instead of through
the unified `emit_request_outcome` funnel, and it did so twice per
session close: the per-turn funnel record plus an unconditional
cumulative session-summary `RequestLog`. This PR removes the duplicate
summary log so Codex WS emits exactly one request log per turn, matching
the HTTP provider paths.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Dropped the duplicate cumulative session-summary `RequestLog` in the
Codex WS handler while preserving the per-turn `emit_request_outcome`
path.
- Preserved gated `request_messages` and `turn_id` on residual outcomes
so dashboard telemetry keeps the useful attribution without
double-counting tokens.
- Ensured explicit `--anyllm-provider` wins over a leaked
`HEADROOM_ANYLLM_PROVIDER` environment variable.
- Registered retry delay settings that had drifted out of the settings
registry.
- Hardened tests against developer-shell `HEADROOM_*` /
`ANTHROPIC_CUSTOM_HEADERS` leakage and stabilized several focused
proxy/wrap test fixtures.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ .venv/bin/pytest tests/ -q -p no:cacheprovider
8529 passed, 537 skipped, 5831 warnings in 276.25s (0:04:36)

$ .venv/bin/ruff check <touched files>
All checks passed!
```

## Real Behavior Proof

- Environment: macOS (Darwin 25.4.0), Python 3.13.14, pytest 9.0.3, ruff
via project venv, branch `fix/multi-provider-runtime`.
- Exact command / steps: Ran the full test suite without pytest cache
provider and Ruff on all touched files; used `git stash` to confirm the
stale fake-config failures pre-existed this change.
- Observed result: Full suite passed with no failures; Ruff passed;
Codex WS now routes end-of-session logging through
`emit_request_outcome`, emitting one request log per turn with the same
accounting model as Anthropic HTTP turns.
- Not tested: Live simultaneous Claude + Codex dashboard run. `mypy
headroom` was not run to completion; a scoped run reported one
pre-existing `settings_store.py:470` coercion error outside this diff.

## Review Readiness

- [x] I have performed a self-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 my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A - server-side observability fix; no UI markup changed.

## Additional Notes

- The proxy's multi-provider routing, header/auth isolation, and
per-model cache keying are already correct and unchanged here; only the
WS observability write path was double-counting.
- Architectural assessment:
`plans/reports/research-260714-0004-multi-provider-upstream-compression-report.md`;
root-cause + resolution trail:
`plans/reports/debug-assessment-260714-0011-dashboard-instability-mixed-claude-codex-report.md`.
- No live simultaneous Claude + Codex dashboard run was performed;
validation is from test coverage and code review of the WS logging path.

---------

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
This commit is contained in:
nangsontay 2026-07-16 01:18:34 +07:00 • committed by GitHub
parent a352fa0168
commit e5b3a634df
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
33 changed files with 2694 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

13
headroom/cli/main.py
View file

@ -32,6 +32,19 @@ def main(ctx: click.Context) -> None:
"""
ctx.ensure_object(dict)
# Apply file-backed settings (settings.json) to the process environment
# BEFORE Click parses any subcommand's ``envvar=`` options (Click resolves
# those when it builds the subcommand context, which happens after this
# group callback runs). ``os.environ.setdefault`` keeps explicit shell
# exports authoritative over the stored file. Fail-open so a corrupt
# settings.json can never block the CLI.
try:
from headroom import settings_store
settings_store.apply_to_environ(settings_store.load())
except Exception: # noqa: BLE001 — settings load must never break the CLI
pass
# Fire a rate-limited, opt-out background check for newer releases so other
# surfaces (e.g. the proxy banner) can show an "update available" notice.
# Never blocks, never raises; skipped for `update` (it checks explicitly).

50
headroom/cli/proxy.py
View file

@ -9,7 +9,11 @@ from typing import Any, Literal, cast
import click
from headroom import paths as _paths
from headroom.providers.registry import resolve_api_overrides, resolve_api_targets
from headroom.providers.registry import (
resolve_api_overrides,
resolve_api_targets,
resolve_extra_headers,
)
from headroom.proxy.modes import PROXY_MODE_CACHE, normalize_proxy_mode
from .main import main
@ -872,6 +876,22 @@ def dashboard(port: int, no_open: bool) -> None:
"Default: /tmp/headroom-embed-{port}.sock. "
"(env: HEADROOM_EMBEDDING_SERVER_SOCKET)",
)
@click.option(
"--anthropic-extra-headers",
default=None,
help=(
"JSON object of extra headers merged into (and overriding) headers forwarded to "
'the Anthropic endpoint, e.g. \'{"Api-Key": "..."}\' (env: ANTHROPIC_TARGET_API_HEADERS)'
),
)
@click.option(
"--openai-extra-headers",
default=None,
help=(
"JSON object of extra headers merged into (and overriding) headers forwarded to "
"the OpenAI endpoint (env: OPENAI_TARGET_API_HEADERS)"
),
)
@click.pass_context
def proxy(
ctx: click.Context,
@ -943,6 +963,8 @@ def proxy(
backend: str,
anyllm_provider: str,
anthropic_api_url: str | None,
anthropic_extra_headers: str | None,
openai_extra_headers: str | None,
openai_api_url: str | None,
gemini_api_url: str | None,
cloudcode_api_url: str | None,
@ -1047,6 +1069,17 @@ def proxy(
sys.exit(1)
os.environ["HEADROOM_INTERCEPT_ENABLED"] = "1"
try:
resolved_anthropic_extra_headers = resolve_extra_headers(
anthropic_extra_headers, "ANTHROPIC_TARGET_API_HEADERS"
)
resolved_openai_extra_headers = resolve_extra_headers(
openai_extra_headers, "OPENAI_TARGET_API_HEADERS"
)
except ValueError as exc:
click.secho(f"error: {exc}", fg="red", err=True)
sys.exit(1)
provider_api_overrides = resolve_api_overrides(
anthropic_api_url=anthropic_api_url,
openai_api_url=openai_api_url,
@ -1056,8 +1089,17 @@ def proxy(
environ=os.environ,
)
# Resolve anyllm provider: env var takes precedence over CLI default (matches argparse path)
effective_anyllm_provider = os.environ.get("HEADROOM_ANYLLM_PROVIDER") or anyllm_provider
# Resolve anyllm provider. An explicit --anyllm-provider flag always wins;
# otherwise honor HEADROOM_ANYLLM_PROVIDER, which the settings store may
# have exported into os.environ after Click parsed the option (so the
# already-parsed param can't see it).
_anyllm_source = click.get_current_context().get_parameter_source("anyllm_provider")
if _anyllm_source is click.core.ParameterSource.COMMANDLINE:
effective_anyllm_provider = anyllm_provider
else:
effective_anyllm_provider = (
os.environ.get("HEADROOM_ANYLLM_PROVIDER") or anyllm_provider
)
# Resolve mode: CLI flag > env var > default. Default is CACHE (Headroom's
# coding posture): delta-only compression at ~0 prefix-cache busts.
@ -1110,6 +1152,8 @@ def proxy(
host=host,
port=port,
anthropic_api_url=provider_api_overrides.anthropic,
anthropic_extra_headers=resolved_anthropic_extra_headers,
openai_extra_headers=resolved_openai_extra_headers,
openai_api_url=provider_api_overrides.openai,
gemini_api_url=provider_api_overrides.gemini,
cloudcode_api_url=provider_api_overrides.cloudcode,

6
headroom/dashboard/__init__.py
View file

@ -10,3 +10,9 @@ def get_dashboard_html() -> str:
"""Load the dashboard HTML template."""
template_path = TEMPLATES_DIR / "dashboard.html"
return template_path.read_text(encoding="utf-8")
def get_settings_html() -> str:
"""Load the settings GUI HTML template."""
template_path = TEMPLATES_DIR / "settings.html"
return template_path.read_text(encoding="utf-8")

5
headroom/dashboard/templates/dashboard.html
View file

@ -164,6 +164,11 @@
<div class="text-xs text-gray-500">
Updated <span x-text="lastUpdate"></span>
</div>
<a href="/dashboard/settings" id="settings-link"
class="px-3 py-1.5 text-sm rounded-md border border-border bg-surface text-gray-400 hover:text-gray-200 transition-colors"
title="Settings">
&#9881; Settings
</a>
<button onclick="toggleTheme()" id="theme-toggle"
class="px-3 py-1.5 text-sm rounded-md border border-border bg-surface text-gray-400 hover:text-gray-200 transition-colors"
title="Toggle light/dark mode">

314
headroom/dashboard/templates/settings.html Normal file
View file

@ -0,0 +1,314 @@
<!DOCTYPE html>
<html lang="en" class="dark">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Headroom Settings</title>
<script src="https://cdn.tailwindcss.com"></script>
<script src="https://unpkg.com/alpinejs@3.13.3/dist/cdn.min.js" defer></script>
<script>
(function () {
const saved = localStorage.getItem('headroom-theme');
const prefersDark = window.matchMedia('(prefers-color-scheme: dark)').matches;
if (saved === 'light' || (!saved && !prefersDark)) {
document.documentElement.classList.remove('dark');
} else {
document.documentElement.classList.add('dark');
}
})();
tailwind.config = { darkMode: 'class', theme: { extend: { colors: { accent: '#22d3ee' } } } };
</script>
<style>
:root { --bg: #f9fafb; --surface: #ffffff; --border: #e5e7eb; --muted: #6b7280; }
.dark { --bg: #0b0f14; --surface: #111827; --border: #1f2937; --muted: #9ca3af; }
body { background: var(--bg); }
.card { background: var(--surface); border: 1px solid var(--border); }
.field-help { color: var(--muted); }
input, select { background: var(--surface); border: 1px solid var(--border); }
input:disabled, select:disabled { opacity: .55; cursor: not-allowed; }
</style>
</head>
<body class="text-gray-800 dark:text-gray-100 min-h-screen" x-data="settingsPage()" x-init="init()">
<div class="max-w-3xl mx-auto px-4 py-8">
<div class="flex items-center justify-between mb-6">
<div>
<h1 class="text-2xl font-semibold">Settings</h1>
<p class="text-sm field-help">Configure Headroom runtime knobs. Changes need a restart to apply.</p>
</div>
<a href="/dashboard" class="text-sm text-accent hover:underline">&larr; Dashboard</a>
</div>
<!-- Status / error surface -->
<template x-if="loadError">
<div class="card rounded-lg p-4 mb-4 border-red-500/50">
<p class="text-red-500 text-sm" x-text="loadError"></p>
</div>
</template>
<!-- Restart-required banner -->
<template x-if="banner">
<div class="card rounded-lg p-4 mb-4 border-amber-500/50" x-cloak>
<p class="text-amber-500 text-sm font-medium" x-text="banner"></p>
<template x-if="applyMode === 'docker' && applyCommand">
<div class="mt-2">
<p class="text-xs field-help mb-1">Run this on the host to apply:</p>
<code class="block text-xs bg-black/40 rounded px-3 py-2 select-all" x-text="applyCommand"></code>
</div>
</template>
</div>
</template>
<!-- Tier tabs -->
<div class="flex items-center gap-2 mb-6" role="tablist">
<button type="button" role="tab" @click="activeTab = 'basic'"
:aria-selected="activeTab === 'basic'"
:class="activeTab === 'basic' ? 'bg-accent text-black' : 'card field-help'"
class="px-3 py-1.5 rounded text-sm font-medium">Settings</button>
<button type="button" role="tab" @click="activeTab = 'advanced'"
:aria-selected="activeTab === 'advanced'"
:class="activeTab === 'advanced' ? 'bg-accent text-black' : 'card field-help'"
class="px-3 py-1.5 rounded text-sm font-medium">Advanced</button>
</div>
<!-- Grouped fields -->
<template x-if="loaded">
<form @submit.prevent>
<template x-for="group in groups.filter(g => fieldsIn(g).length > 0)" :key="group">
<div class="card rounded-lg p-5 mb-4">
<h2 class="text-sm font-semibold uppercase tracking-wide field-help mb-4" x-text="group"></h2>
<template x-for="field in fieldsIn(group)" :key="field.key">
<div class="mb-4 last:mb-0">
<label class="flex items-center justify-between gap-4">
<span class="text-sm font-medium" x-text="field.label"></span>
<!-- bool -->
<template x-if="field.type === 'bool'">
<input type="checkbox" class="h-5 w-5 accent-cyan-400"
:disabled="isLocked(field)"
x-model="values[field.key]">
</template>
<!-- enum -->
<template x-if="field.type === 'enum'">
<select class="rounded px-2 py-1 text-sm w-56"
:disabled="isLocked(field)"
x-model="values[field.key]">
<template x-for="choice in field.choices" :key="choice">
<option :value="choice" x-text="choice"></option>
</template>
</select>
</template>
<!-- int / float -->
<template x-if="field.type === 'int' || field.type === 'float'">
<input type="number" class="rounded px-2 py-1 text-sm w-56"
:step="field.type === 'float' ? 'any' : '1'"
:min="field.minimum" :max="field.maximum"
:disabled="isLocked(field)"
x-model.number="values[field.key]">
</template>
<!-- str / secret / header-map -->
<template x-if="field.type === 'str' || field.type === 'header-map'">
<input :type="field.secret ? 'password' : 'text'"
class="rounded px-2 py-1 text-sm w-56"
:disabled="isLocked(field)"
x-model="values[field.key]"
@input="delete clearedKeys[field.key]">
</template>
</label>
<template x-if="field.secret && !isLocked(field)">
<button type="button" @click="values[field.key] = ''; clearedKeys[field.key] = true"
class="text-xs text-red-500/80 hover:underline mt-1">
Clear stored value
</button>
</template>
<p class="text-xs field-help mt-1" x-text="field.help"></p>
<template x-if="field.manifest_managed && manifestManaged">
<p class="text-xs text-amber-500/80 mt-1">
Managed by the install manifest — change via <code>headroom install</code>.
</p>
</template>
<template x-if="field.env_override && !(field.manifest_managed && manifestManaged)">
<p class="text-xs text-amber-500/80 mt-1">
Overridden by environment variable <code x-text="field.env"></code> -- edits here have no effect until it's unset.
</p>
</template>
<template x-if="fieldErrors[field.key]">
<p class="text-xs text-red-500 mt-1" x-text="fieldErrors[field.key]"></p>
</template>
</div>
</template>
</div>
</template>
<div class="flex items-center gap-3 mt-6">
<button type="button" @click="save()"
:disabled="busy"
class="px-4 py-2 rounded bg-gray-700 hover:bg-gray-600 text-white text-sm disabled:opacity-50">
Save
</button>
<button type="button" @click="applyAndRestart()"
:disabled="busy"
class="px-4 py-2 rounded bg-cyan-500 hover:bg-cyan-400 text-black font-medium text-sm disabled:opacity-50">
Apply &amp; Restart
</button>
<span class="text-sm field-help" x-text="status"></span>
</div>
</form>
</template>
</div>
<script>
function settingsPage() {
return {
loaded: false,
loadError: '',
fields: [],
groups: [],
values: {},
storedValues: {},
manifestManaged: false, // true on supervised installs (locks manifest fields)
fieldErrors: {},
banner: '',
status: '',
busy: false,
applyMode: '',
applyCommand: '',
activeTab: 'basic', // 'basic' (Settings tab) | 'advanced'
clearedKeys: {}, // secret fields explicitly cleared this session (send null on save)
async init() {
try {
const res = await fetch('/settings/schema');
if (!res.ok) throw new Error('HTTP ' + res.status);
const schema = await res.json();
this.fields = schema.fields || [];
this.groups = schema.groups || [];
// Seed the editable model from the effective values.
for (const f of this.fields) {
this.values[f.key] = schema.values ? schema.values[f.key] : f.value;
this.storedValues[f.key] = f.stored;
}
// The server tells us whether this is a supervised install
// (docker/service); there manifest-managed knobs are baked
// into the install manifest and cannot be changed via the GUI.
this.manifestManaged = !!schema.supervised;
this.loaded = true;
} catch (e) {
this.loadError = 'Failed to load settings: ' + e.message;
}
},
fieldsIn(group) { return this.fields.filter(f => f.group === group && f.tier === this.activeTab); },
isLocked(field) { return (!!field.manifest_managed && this.manifestManaged) || !!field.env_override; },
isEmptyValue(value) { return value === '' || value === null || value === undefined; },
normalizedComparable(value) {
return this.isEmptyValue(value) ? null : value;
},
editableValues() {
// Send only changed, non-locked values. The server merges partial
// updates; posting every effective default would freeze those
// defaults into settings.json and change future upgrade behavior.
// Explicit `null` clears a previously stored value.
const out = {};
for (const f of this.fields) {
if (this.isLocked(f)) continue;
if (this.clearedKeys[f.key]) { out[f.key] = null; continue; }
let v = this.values[f.key];
const stored = this.storedValues[f.key];
const baseline = this.isEmptyValue(stored) ? f.default : stored;
if (this.isEmptyValue(v)) {
if (!this.isEmptyValue(stored)) out[f.key] = null;
continue;
}
if (this.normalizedComparable(v) === this.normalizedComparable(baseline)) continue;
out[f.key] = v;
}
return out;
},
async save() {
this.busy = true; this.status = 'Saving…'; this.fieldErrors = {};
try {
const res = await fetch('/settings', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ values: this.editableValues() }),
});
const data = await res.json();
if (res.status === 200) {
this.status = 'Saved.';
const changed = data.changed_keys || [];
this.banner = changed.length ? ('Restart required to apply: ' + changed.join(', ')) : 'Saved — no changes to apply.';
this.applyMode = ''; this.applyCommand = '';
this.clearedKeys = {};
} else if (res.status === 422) {
this.fieldErrors = data.field_errors || {};
this.status = 'Fix the highlighted fields.';
} else {
this.status = data.error || ('Error ' + res.status);
}
} catch (e) {
this.status = 'Save failed: ' + e.message;
} finally {
this.busy = false;
}
},
async applyAndRestart() {
this.busy = true; this.status = 'Applying…'; this.fieldErrors = {};
try {
const res = await fetch('/settings/apply', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ values: this.editableValues() }),
});
const data = await res.json();
if (res.status === 422) {
this.fieldErrors = data.field_errors || {};
this.status = 'Fix the highlighted fields.'; return;
}
if (res.status === 400) {
this.status = data.error || 'Bad request'; return;
}
this.applyMode = data.mode || '';
if (data.restarted && data.mode === 'service') {
this.banner = 'Restarting…';
this.status = 'Waiting for the proxy to come back…';
await this.pollHealth();
} else if (data.mode === 'docker') {
this.applyCommand = data.command || '';
this.banner = 'Saved. To apply, run the command on the host:';
this.status = '';
} else {
this.banner = data.instruction || 'Saved. Restart the proxy to apply.';
this.status = '';
}
} catch (e) {
this.status = 'Apply failed: ' + e.message;
} finally {
this.busy = false;
}
},
async pollHealth() {
// /health is the single source of truth that the proxy is back.
const deadline = 40; // ~60s at 1.5s intervals
for (let i = 0; i < deadline; i++) {
await new Promise(r => setTimeout(r, 1500));
try {
const res = await fetch('/health', { cache: 'no-store' });
if (res.ok) {
this.banner = '';
this.status = 'Applied — proxy restarted.';
return;
}
} catch (e) { /* proxy still down: keep polling */ }
}
this.status = 'Timed out waiting for the proxy. Check the proxy logs.';
},
};
}
</script>
</body>
</html>

85
headroom/install/runtime.py
View file

@ -16,8 +16,9 @@ from typing import Any, cast
from headroom._subprocess import pid_alive, run
from .health import probe_ready
from .models import DeploymentManifest, InstallPreset, RuntimeKind
from .models import DeploymentManifest, InstallPreset, RuntimeKind, SupervisorKind
from .paths import log_path, pid_path, profile_root
from .state import load_manifest
# Inside the container the proxy must listen on every interface so the
# host-side published port (127.0.0.1:<port>) can reach it.
@ -370,3 +371,85 @@ def runtime_status(manifest: DeploymentManifest) -> str:
# as a SystemError against the detached agent, crashing status and taking the
# live proxy down with it (#1544).
return "running" if pid_alive(pid) else "stopped"
def detect_current_deployment() -> tuple[DeploymentManifest | None, str]:
"""Detect how THIS running proxy was launched.
Returns ``(manifest_or_none, mode)`` where ``mode`` is one of:
* ``"docker"`` — persistent-docker deployment. Cannot self-restart:
there is no docker socket/CLI inside the container.
* ``"service"`` — any other persistent (supervised) deployment; can
self-restart via ``headroom install restart``.
* ``"foreground"`` — a plain ``headroom proxy`` (or unknown); not
self-restartable.
Keys off the ``HEADROOM_DEPLOYMENT_*`` env vars the supervisor injects at
launch (see :func:`_deployment_env`); a foreground proxy has none set.
"""
profile = os.environ.get("HEADROOM_DEPLOYMENT_PROFILE")
preset = os.environ.get("HEADROOM_DEPLOYMENT_PRESET")
if not profile:
return None, "foreground"
manifest = load_manifest(profile)
if preset == InstallPreset.PERSISTENT_DOCKER.value:
return manifest, "docker"
if manifest is None:
return None, "foreground"
if manifest.supervisor_kind == SupervisorKind.TASK.value:
return manifest, "task"
return manifest, "service"
def _spawn_detached_restart(profile: str) -> None:
"""Spawn a detached ``headroom install restart --profile <p>`` process.
Detached (``start_new_session`` on POSIX) so it outlives this process being
torn down by the very restart it triggers.
"""
command = [*resolve_headroom_command(), "install", "restart", "--profile", profile]
popen_kwargs: dict[str, Any] = {"stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL}
if not _is_windows():
popen_kwargs["start_new_session"] = True
subprocess.Popen(command, **popen_kwargs)
def restart_current_deployment() -> dict[str, Any]:
"""Restart the current deployment so new settings take effect.
* service -> spawn a detached restart, return ``{restarted: True, ...}``.
* docker -> not restartable in-container; return the host command to run.
* task -> not restartable via the CLI (``headroom install`` rejects
lifecycle ops for task-scheduled deployments); return an instruction.
* foreground/unknown -> return a manual-restart instruction.
"""
manifest, mode = detect_current_deployment()
profile = os.environ.get("HEADROOM_DEPLOYMENT_PROFILE") or (
manifest.profile if manifest else "default"
)
if mode == "service":
_spawn_detached_restart(profile)
return {"restarted": True, "mode": "service", "profile": profile}
if mode == "docker":
return {
"restarted": False,
"mode": "docker",
"command": f"headroom install restart --profile {profile}",
}
if mode == "task":
return {
"restarted": False,
"mode": "task",
"instruction": (
"This deployment is managed by an OS task scheduler, not "
"`headroom install`; stop the running process so it is "
"relaunched (with the new settings) on its next scheduled "
"trigger, or restart it via your OS task scheduler."
),
}
return {
"restarted": False,
"mode": "foreground",
"instruction": "Restart the proxy to apply the new settings.",
}

14
headroom/paths.py
View file

@ -50,6 +50,7 @@ HEADROOM_SAVINGS_PATH_ENV = "HEADROOM_SAVINGS_PATH"
HEADROOM_SAVINGS_EVENTS_PATH_ENV = "HEADROOM_SAVINGS_EVENTS_PATH"
HEADROOM_TOIN_PATH_ENV = "HEADROOM_TOIN_PATH"
HEADROOM_SUBSCRIPTION_STATE_PATH_ENV = "HEADROOM_SUBSCRIPTION_STATE_PATH"
HEADROOM_SETTINGS_PATH_ENV = "HEADROOM_SETTINGS_PATH"
# ---------------------------------------------------------------------------
# Default sub-path fragments
@ -60,6 +61,7 @@ _CONFIG_DIR_DEFAULT_SUFFIX = "config"
# Resource file/sub-dir names (kept here so nothing else has to hardcode them)
_SAVINGS_FILE = "proxy_savings.json"
_SETTINGS_FILE = "settings.json"
_TOIN_FILE = "toin.json"
_MODELS_FILE = "models.json"
_SUBSCRIPTION_FILE = "subscription_state.json"
@ -210,6 +212,16 @@ def savings_path(explicit: str | os.PathLike[str] | None = None) -> Path:
)
def settings_path(explicit: str | os.PathLike[str] | None = None) -> Path:
"""Return the path for the dashboard-managed settings JSON file."""
return _resolve(
explicit,
HEADROOM_SETTINGS_PATH_ENV,
workspace_dir() / _SETTINGS_FILE,
)
def toin_path(explicit: str | os.PathLike[str] | None = None) -> Path:
"""Return the path for the TOIN telemetry JSON file.
@ -412,6 +424,7 @@ __all__ = [
"HEADROOM_SAVINGS_EVENTS_PATH_ENV",
"HEADROOM_TOIN_PATH_ENV",
"HEADROOM_SUBSCRIPTION_STATE_PATH_ENV",
"HEADROOM_SETTINGS_PATH_ENV",
"set_process_stateless",
"process_is_stateless",
"config_dir",
@ -426,6 +439,7 @@ __all__ = [
"license_cache_path",
"session_stats_path",
"savings_events_path",
"settings_path",
"sync_state_path",
"bridge_state_path",
"log_dir",

27
headroom/providers/registry.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
import json
import logging
import os
from collections.abc import Callable, Mapping
@ -130,6 +131,32 @@ def resolve_api_overrides(
)
def resolve_extra_headers(
cli_value: str | None,
env_var: str,
*,
environ: Mapping[str, str] | None = None,
) -> dict[str, str] | None:
"""Resolve extra headers to merge into (and override) forwarded provider requests.
Accepts a JSON object string from CLI or env (CLI wins). Returns ``None`` if unset.
Raises ``ValueError`` on invalid JSON or a non-string-keyed/valued object.
"""
env = environ or os.environ
raw = cli_value or env.get(env_var)
if not raw:
return None
try:
parsed = json.loads(raw)
except (ValueError, TypeError) as exc:
raise ValueError(f"{env_var} must be a JSON object of header name/value strings") from exc
if not isinstance(parsed, dict) or not all(
isinstance(k, str) and isinstance(v, str) for k, v in parsed.items()
):
raise ValueError(f"{env_var} must be a JSON object of header name/value strings")
return parsed or None
def resolve_api_targets(overrides: ProviderApiOverrides) -> ProviderApiTargets:
"""Resolve normalized upstream provider targets from configured overrides."""
return ProviderApiTargets(

33
headroom/proxy/handlers/anthropic.py
View file

@ -798,10 +798,15 @@ class AnthropicHandlerMixin:
# uses `request.headers.get(...)` directly above; memory user-id
# is read from `request.headers` below if needed. From this
# point on, `headers` is the upstream-bound copy.
from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
from headroom.proxy.helpers import (
_strip_internal_headers,
log_outbound_headers,
merge_extra_headers,
)
_pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
headers = _strip_internal_headers(headers)
headers = merge_extra_headers(headers, self.config.anthropic_extra_headers)
log_outbound_headers(
forwarder="anthropic_messages",
stripped_count=_pre_strip_count
@ -2815,8 +2820,11 @@ class AnthropicHandlerMixin:
# Sanitize headers (redact API keys)
safe_headers = {}
_sensitive_header_names = {"x-api-key", "authorization"} | {
k.lower() for k in (self.config.anthropic_extra_headers or {})
}
for k, v in headers.items():
if k.lower() in ("x-api-key", "authorization"):
if k.lower() in _sensitive_header_names:
safe_headers[k] = v[:12] + "..." if v else ""
else:
safe_headers[k] = v
@ -3558,10 +3566,15 @@ class AnthropicHandlerMixin:
client = classify_client(headers, default="claude")
tags = extract_tags(headers)
# PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
from headroom.proxy.helpers import (
_strip_internal_headers,
log_outbound_headers,
merge_extra_headers,
)
_pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
headers = _strip_internal_headers(headers)
headers = merge_extra_headers(headers, self.config.anthropic_extra_headers)
log_outbound_headers(
forwarder="anthropic_batch",
stripped_count=_pre_strip_count,
@ -3837,10 +3850,15 @@ class AnthropicHandlerMixin:
client = classify_client(headers, default="claude")
tags = extract_tags(headers)
# PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
from headroom.proxy.helpers import (
_strip_internal_headers,
log_outbound_headers,
merge_extra_headers,
)
_pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
headers = _strip_internal_headers(headers)
headers = merge_extra_headers(headers, self.config.anthropic_extra_headers)
log_outbound_headers(
forwarder="anthropic_batch_passthrough",
stripped_count=_pre_strip_count,
@ -3968,10 +3986,15 @@ class AnthropicHandlerMixin:
client = classify_client(headers, default="claude")
tags = extract_tags(headers)
# PR-A5 (P5-49): strip internal x-headroom-* before forwarding upstream.
from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
from headroom.proxy.helpers import (
_strip_internal_headers,
log_outbound_headers,
merge_extra_headers,
)
_pre_strip_count = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
headers = _strip_internal_headers(headers)
headers = merge_extra_headers(headers, self.config.anthropic_extra_headers)
log_outbound_headers(
forwarder="anthropic_batch_results",
stripped_count=_pre_strip_count,

79
headroom/proxy/handlers/openai.py
View file

@ -2504,10 +2504,15 @@ class OpenAIHandlerMixin:
# uses `request.headers.get(...)` above; memory user-id reads
# `request.headers` below. From this point on, `headers` is the
# upstream-bound copy.
from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
from headroom.proxy.helpers import (
_strip_internal_headers,
log_outbound_headers,
merge_extra_headers,
)
_pre_strip_count_chat = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
headers = _strip_internal_headers(headers)
headers = merge_extra_headers(headers, self.config.openai_extra_headers)
log_outbound_headers(
forwarder="openai_chat_completions",
stripped_count=_pre_strip_count_chat,
@ -4015,10 +4020,15 @@ class OpenAIHandlerMixin:
# PR-A5 (P5-49): strip internal x-headroom-* from upstream-bound
# headers AFTER `_extract_tags` reads them. Memory user-id reads
# `request.headers` below.
from headroom.proxy.helpers import _strip_internal_headers, log_outbound_headers
from headroom.proxy.helpers import (
_strip_internal_headers,
log_outbound_headers,
merge_extra_headers,
)
_pre_strip_count_resp = sum(1 for k in headers if k.lower().startswith("x-headroom-"))
headers = _strip_internal_headers(headers)
headers = merge_extra_headers(headers, self.config.openai_extra_headers)
# Mirror the WS handler: never forward Codex's client-only lite header
# upstream. OpenAI rejects newer Codex models when it leaks, and the HTTP
# POST path (unlike the WS path) otherwise forwards request headers verbatim.
@ -5280,6 +5290,9 @@ class OpenAIHandlerMixin:
_upstream_connect_recorded = False
_upstream_first_event_started: float | None = None
upstream: Any = None
from headroom.proxy.helpers import merge_extra_headers
upstream_headers = merge_extra_headers(upstream_headers, self.config.openai_extra_headers)
for ws_attempt in range(ws_connect_attempts):
try:
@ -7109,10 +7122,23 @@ class OpenAIHandlerMixin:
# Session-end residual: tokens not captured by any
# per-turn record (e.g. signaling frames after the
# last response.completed). The funnel emits the full
# bookkeeping quartet for the residual; the explicit
# session-summary RequestLog below remains a separate
# entry (different semantics — cumulative session
# totals vs delta residual).
# bookkeeping quartet for the residual. This is the
# session's ONLY end-of-session RequestLog row: per-turn
# rows already carry the session's tokens as deltas, so a
# separate cumulative session-summary row would double-count
# Codex WS traffic in the recent-requests feed and every
# log-derived stat (savings, throughput).
from headroom.proxy.helpers import compute_turn_id
ws_messages_for_log: list[dict[str, Any]] = []
ws_input_for_log = ws_inner_for_telemetry.get("input")
ws_instructions_for_log = ws_inner_for_telemetry.get("instructions")
if isinstance(ws_instructions_for_log, str) and ws_instructions_for_log:
ws_messages_for_log.append(
{"role": "system", "content": ws_instructions_for_log}
)
if isinstance(ws_input_for_log, str) and ws_input_for_log:
ws_messages_for_log.append({"role": "user", "content": ws_input_for_log})
await self._record_request_outcome(
RequestOutcome(
request_id=request_id,
@ -7133,44 +7159,6 @@ class OpenAIHandlerMixin:
transforms_applied=tuple(transforms_applied),
tags=ws_session_tags,
client=client,
)
)
ws_recorded_overhead_ms_total = _current_ws_overhead_ms()
if final_ttfb_ms > 0:
ws_recorded_ttfb_ms = True
if getattr(self, "logger", None) is not None:
from headroom.proxy.helpers import compute_turn_id
from headroom.proxy.models import RequestLog
ws_messages_for_log: list[dict[str, Any]] = []
ws_input_for_log = ws_inner_for_telemetry.get("input")
ws_instructions_for_log = ws_inner_for_telemetry.get("instructions")
if isinstance(ws_instructions_for_log, str) and ws_instructions_for_log:
ws_messages_for_log.append(
{"role": "system", "content": ws_instructions_for_log}
)
if isinstance(ws_input_for_log, str) and ws_input_for_log:
ws_messages_for_log.append({"role": "user", "content": ws_input_for_log})
self.logger.log(
RequestLog(
request_id=request_id,
timestamp=datetime.now().isoformat(),
provider="openai",
model=model_name,
input_tokens_original=ws_input_tokens_total + tokens_saved,
input_tokens_optimized=ws_input_tokens_total,
output_tokens=ws_output_tokens_total,
tokens_saved=tokens_saved,
savings_percent=(
tokens_saved / (ws_input_tokens_total + tokens_saved) * 100
)
if ws_input_tokens_total + tokens_saved > 0
else 0.0,
optimization_latency_ms=_current_ws_overhead_ms(),
total_latency_ms=ws_session_duration_ms,
tags=ws_session_tags,
cache_hit=False,
transforms_applied=transforms_applied,
request_messages=ws_messages_for_log
if getattr(self.config, "log_full_messages", False)
else None,
@ -7181,6 +7169,9 @@ class OpenAIHandlerMixin:
),
)
)
ws_recorded_overhead_ms_total = _current_ws_overhead_ms()
if final_ttfb_ms > 0:
ws_recorded_ttfb_ms = True
except Exception as e:
if "WebSocketDisconnect" in type(e).__name__:

18
headroom/proxy/helpers.py
View file

@ -1517,6 +1517,24 @@ def _strip_internal_headers(headers: dict[str, str]) -> dict[str, str]:
return strip_internal_headers(headers, mode=get_strip_internal_headers_mode())
def merge_extra_headers(headers: dict[str, str], extra: dict[str, str] | None) -> dict[str, str]:
"""Merge configured extra headers into ``headers``, overriding same-named keys.
``extra`` comes from ``ProxyConfig.anthropic_extra_headers``/``openai_extra_headers``
(settings-panel/CLI-configured, for gateways that need one extra header alongside the
client's own auth). Returns ``headers`` unchanged (no copy) when nothing is configured.
"""
if not extra:
return headers
# HTTP header names are case-insensitive: drop any existing key that
# case-insensitively collides with a configured extra so the extra wins.
# A plain {**headers, **extra} would emit both casings upstream.
lowered = {k.lower() for k in extra}
merged = {k: v for k, v in headers.items() if k.lower() not in lowered}
merged.update(extra)
return merged
def log_outbound_headers(
*,
forwarder: str,

33
headroom/proxy/loopback_guard.py
View file

@ -54,6 +54,7 @@ __all__ = [
"is_loopback_host",
"is_loopback_host_header",
"require_loopback",
"require_same_origin",
]
@ -213,3 +214,35 @@ def require_loopback(request: Request) -> None: # type: ignore[valid-type]
host_header = None
if not is_loopback_host_header(host_header):
raise HTTPException(status_code=404)
def require_same_origin(request: Request) -> None: # type: ignore[valid-type]
"""FastAPI dependency: reject cross-origin browser requests on mutating routes.
``require_loopback``'s Host-header check stops DNS-rebinding, but not a
plain CSRF where a remote page's JS targets a known
``http://127.0.0.1:<port>`` URL directly with a non-preflighted "simple"
request (e.g. ``Content-Type: text/plain`` carrying a JSON body) -- the
browser's ``Host:`` header still reads the real destination (loopback),
but its ``Origin:`` header reflects the page's actual origin. CORS alone
does not stop this: CORS only blocks the attacker's JS from *reading* the
response, not the server from acting on the request.
Reject when ``Origin`` is present and does not itself name a loopback
host, or is the opaque literal ``"null"`` (sandboxed iframe / ``file://``
page). Requests with no ``Origin`` header (CLI tools, curl, ``TestClient``,
same-origin simple navigations) pass through unchanged -- a real browser
always sets ``Origin`` on cross-origin fetch/XHR.
"""
if HTTPException is None: # pragma: no cover - defensive
raise RuntimeError("FastAPI is required for the same-origin guard")
headers = getattr(request, "headers", None)
origin = headers.get("origin") if headers is not None else None
if not origin:
return
if origin == "null":
raise HTTPException(status_code=403, detail="cross-origin request rejected")
host_part = origin.split("://", 1)[-1].split("/", 1)[0]
if not is_loopback_host_header(host_part):
raise HTTPException(status_code=403, detail="cross-origin request rejected")

5
headroom/proxy/models.py
View file

@ -127,6 +127,11 @@ class ProxyConfig:
gemini_api_url: str | None = None # Custom Gemini API URL override
cloudcode_api_url: str | None = None # Custom Cloud Code Assist API URL override
vertex_api_url: str | None = None # Custom Vertex AI regional API URL override
# Extra headers merged into (and overriding) forwarded Anthropic/OpenAI requests.
# JSON-object config knobs; see settings_store's anthropic_extra_headers/
# openai_extra_headers and providers.registry.resolve_extra_headers.
anthropic_extra_headers: dict[str, str] | None = None
openai_extra_headers: dict[str, str] | None = None
# Backend: "anthropic" (direct API), "litellm-*" (via LiteLLM), or "anyllm" (via any-llm)
backend: str = "anthropic"

158
headroom/proxy/server.py
View file

@ -2135,6 +2135,17 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
config = config or ProxyConfig()
# Defensive re-apply of file-backed settings for embedded/non-CLI callers
# that construct the app without going through the `headroom` CLI entrypoint
# (which already applies them before Click parsing). setdefault keeps
# explicit env exports authoritative; fail-open so it never blocks startup.
try:
from headroom import settings_store
settings_store.apply_to_environ(settings_store.load())
except Exception: # noqa: BLE001 — settings load must never break startup
pass
# Air-gap master switch. Propagate config.offline to the env so the
# env-based egress predicates (telemetry, update check, license) all honor
# it, force HF/transformers offline before any model code loads, and
@ -2933,6 +2944,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
collect_tasks as _collect_tasks,
)
from headroom.proxy.loopback_guard import require_loopback as _require_loopback
from headroom.proxy.loopback_guard import require_same_origin as _require_same_origin
@app.get("/admin/upstream", dependencies=[Depends(_require_loopback)])
async def get_upstream():
@ -3025,6 +3037,152 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
"""Serve the Headroom dashboard UI."""
return get_dashboard_html()
# --- Dashboard settings API (loopback-gated, registry-validated) ---------
# Read/write the curated HEADROOM_* knobs the settings GUI manages. Writes
# reuse the same loopback guard as the /admin and /debug endpoints and only
# ever touch keys in the settings_store registry allowlist. All curated
# knobs are startup-captured, so a write always requires a restart to apply
# (Phase 3's /settings/apply drives that).
from headroom import settings_store
@app.get("/settings/schema", dependencies=[Depends(_require_loopback)])
async def settings_schema(_request: Request):
"""Registry + grouped fields + effective values for the settings form."""
schema = settings_store.to_schema()
# Tell the UI whether this is a supervised (docker/service) install, where
# manifest-baked knobs (HEADROOM_PORT/HEADROOM_HOST) are owned by the
# install manifest and must be rendered read-only. Foreground proxies
# can edit everything. Fail-open to "not supervised".
try:
from headroom.install import runtime as install_runtime
_manifest, mode = install_runtime.detect_current_deployment()
schema["supervised"] = mode != "foreground"
except Exception: # noqa: BLE001 — schema must render even if detection fails
schema["supervised"] = False
return JSONResponse(status_code=200, content=schema)
@app.get("/settings", dependencies=[Depends(_require_loopback)])
async def settings_get(_request: Request):
"""Return stored (file) values only; secret fields masked."""
return JSONResponse(status_code=200, content=settings_store.stored_values())
@app.post("/settings", dependencies=[Depends(_require_loopback), Depends(_require_same_origin)])
async def settings_post(request: Request):
"""Persist settings. Unknown key -> 400; bad type/enum/range -> 422.
Loopback-only (mutates process configuration; treated as an admin
action). The audit log records which keys changed — never their values,
so no secret is leaked.
"""
try:
body = await request.json()
except (ValueError, UnicodeDecodeError):
body = None
if not isinstance(body, dict) or not isinstance(body.get("values"), dict):
return JSONResponse(
status_code=400,
content={"error": "expected a JSON object with a `values` object"},
)
values = body["values"]
before = settings_store.load()
try:
settings_store.save(values)
except settings_store.SettingsValidationError as exc:
if exc.unknown_keys:
return JSONResponse(
status_code=400,
content={"error": "unknown settings key(s)", "unknown_keys": exc.unknown_keys},
)
return JSONResponse(
status_code=422,
content={"error": "invalid settings value(s)", "field_errors": exc.field_errors},
)
after = settings_store.load()
changed_keys = sorted(k for k in set(before) | set(after) if before.get(k) != after.get(k))
record_admin_action(
request=request,
action="settings_update",
status_code=200,
details={"changed_keys": changed_keys},
)
return JSONResponse(
status_code=200,
content={"ok": True, "needs_restart": bool(changed_keys), "changed_keys": changed_keys},
)
@app.post(
"/settings/apply", dependencies=[Depends(_require_loopback), Depends(_require_same_origin)]
)
async def settings_apply(request: Request):
"""Persist settings (optional body) then restart the proxy to apply them.
Loopback-only. Service deployments self-restart (one-click): we flush a
202 BEFORE the detached restarter tears this process down, and the UI
polls /health to detect the proxy returning. Docker deployments cannot
self-restart (no docker socket in-container), so we surface the host
command instead. Foreground `headroom proxy` returns a manual-restart
instruction. /health is the single source of truth for "proxy is back".
"""
from starlette.background import BackgroundTask
from headroom.install import runtime as install_runtime
try:
body = await request.json()
except (ValueError, UnicodeDecodeError):
body = None
# `values` is optional — the UI may have already saved via POST /settings.
# When present, persist it under the same validation contract.
if isinstance(body, dict) and isinstance(body.get("values"), dict):
try:
settings_store.save(body["values"])
except settings_store.SettingsValidationError as exc:
if exc.unknown_keys:
return JSONResponse(
status_code=400,
content={
"error": "unknown settings key(s)",
"unknown_keys": exc.unknown_keys,
},
)
return JSONResponse(
status_code=422,
content={
"error": "invalid settings value(s)",
"field_errors": exc.field_errors,
},
)
_manifest, mode = install_runtime.detect_current_deployment()
record_admin_action(
request=request,
action="settings_apply",
status_code=202 if mode == "service" else 200,
details={"mode": mode},
)
if mode == "service":
# Flush the 202 first; run the detached restart only after the
# response body is sent (Starlette runs BackgroundTask post-response),
# so restarting our own process never drops this response.
return JSONResponse(
status_code=202,
content={"restarted": True, "mode": "service"},
background=BackgroundTask(install_runtime.restart_current_deployment),
)
result = install_runtime.restart_current_deployment()
return JSONResponse(status_code=200, content=result)
@app.get(
"/dashboard/settings",
response_class=HTMLResponse,
dependencies=[Depends(_require_loopback)],
)
async def dashboard_settings():
"""Serve the Headroom settings GUI."""
from headroom.dashboard import get_settings_html
return get_settings_html()
@app.get("/favicon.ico")
async def favicon() -> Response:
# Registered before register_provider_routes' catch-all passthrough

955
headroom/settings_store.py Normal file
View file

@ -0,0 +1,955 @@
"""File-backed store for a curated subset of Headroom's runtime knobs (mostly ``HEADROOM_*``,
The dashboard settings GUI persists these knobs to ``settings.json`` in the
workspace dir and this module applies them to ``os.environ`` at CLI startup
with ``os.environ.setdefault`` — so an explicit shell export always wins over
the stored file. Precedence: ``export > settings.json > code default``.
Deliberately dependency-light (stdlib + ``headroom.paths`` only, no FastAPI or
proxy imports) so the early CLI apply hook — which must run
before Click parses ``envvar=`` options — stays cheap and import-safe.
``load()`` is fail-open: a corrupt or unreadable ``settings.json`` yields
``{}`` (defaults) rather than raising, so it can never crash-loop the proxy on
startup. ``save()`` writes atomically (temp file + ``os.replace``).
"""
from __future__ import annotations
import json
import logging
import math
import os
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from headroom import paths
logger = logging.getLogger(__name__)
_MASK = "••••••" # ●●●●●● for masked secret values
@dataclass(frozen=True)
class SettingField:
"""One curated, GUI-editable env knob.
``env`` is the ``HEADROOM_*`` variable the knob maps to; ``key`` is the
JSON/API key. ``type`` drives coercion, validation and the UI control.
``manifest_managed`` marks a knob baked into the install manifest on
supervised (docker/service) deploys — ``settings.json`` cannot change it
there, so the UI renders it read-only.
"""
env: str
key: str
label: str
group: str
type: str # "bool" | "int" | "float" | "str" | "enum" | "optional-bool" | "csv-list"
default: Any = None
choices: tuple[str, ...] = ()
help: str = ""
secret: bool = False
manifest_managed: bool = False
minimum: float | None = None
maximum: float | None = None
tier: str = "advanced" # "basic" | "advanced" - Settings vs Advanced tab placement
# Curated registry. Env formats verified against each knob's Click option in
# headroom/cli/proxy.py (bools serialize to "1"/"0", which Click's BOOL type and
# the body-resolved HEADROOM_CODE_AWARE_ENABLED reader both accept).
SETTINGS: tuple[SettingField, ...] = (
# --- Compression ---
SettingField(
"HEADROOM_SAVINGS_PROFILE",
"savings_profile",
"Savings profile",
"Compression",
"enum",
default="coding",
choices=("agent-90", "balanced", "coding", "general"),
help="Named compression posture applied at startup.",
tier="basic",
),
SettingField(
"HEADROOM_TARGET_RATIO",
"target_ratio",
"Target keep-ratio",
"Compression",
"float",
default=None,
minimum=0.0,
maximum=1.0,
help="Kompress keep-ratio 0-1 (lower = more aggressive). Unset = adaptive.",
tier="basic",
),
SettingField(
"HEADROOM_DISABLE_KOMPRESS",
"disable_kompress",
"Disable Kompress",
"Compression",
"bool",
default=False,
help="Disable Kompress ML compression (structural compression stays on).",
tier="basic",
),
SettingField(
"HEADROOM_LOSSLESS",
"lossless",
"Lossless mode",
"Compression",
"bool",
default=False,
help="No-CCR lossless compaction; no retrieval marker emitted.",
tier="basic",
),
SettingField(
"HEADROOM_CODE_AWARE_ENABLED",
"code_aware_enabled",
"Code-aware compression",
"Compression",
"bool",
default=True,
help="AST-based code compression (requires the [code] extra).",
tier="basic",
),
SettingField(
"HEADROOM_PROTECT_TOOL_RESULTS",
"protect_tool_results",
"Protect tool results",
"Compression",
"str",
default=None,
help="Comma-separated tool names whose results are never lossy-compressed.",
tier="advanced",
),
SettingField(
"HEADROOM_NO_CCR",
"no_ccr",
"Disable CCR",
"Compression",
"bool",
default=False,
help="Disable CCR entirely (no markers, no injected retrieve tool).",
tier="basic",
),
# --- Limits ---
SettingField(
"HEADROOM_RPM",
"rpm",
"Requests / min",
"Limits",
"int",
default=None,
minimum=1,
help="Max requests per minute.",
tier="basic",
),
SettingField(
"HEADROOM_TPM",
"tpm",
"Tokens / min",
"Limits",
"int",
default=None,
minimum=1,
help="Max tokens per minute.",
tier="basic",
),
SettingField(
"HEADROOM_LIMIT_CONCURRENCY",
"limit_concurrency",
"Concurrency limit",
"Limits",
"int",
default=1000,
minimum=1,
help="Max concurrent connections before Uvicorn returns 503.",
tier="advanced",
),
SettingField(
"HEADROOM_WORKERS",
"workers",
"Worker processes",
"Limits",
"int",
default=1,
minimum=1,
help="Uvicorn worker processes.",
tier="advanced",
),
# --- Budget ---
SettingField(
"HEADROOM_BUDGET",
"budget",
"Budget (USD)",
"Budget",
"float",
default=None,
minimum=0.0,
help="Budget limit per period; requests are rejected with 429 once reached.",
tier="basic",
),
SettingField(
"HEADROOM_BUDGET_PERIOD",
"budget_period",
"Budget period",
"Budget",
"enum",
default="daily",
choices=("hourly", "daily", "monthly"),
help="Period the budget applies to.",
tier="basic",
),
# --- Networking (baked into the install manifest on supervised deploys) ---
SettingField(
"HEADROOM_HOST",
"host",
"Host",
"Networking",
"str",
default="127.0.0.1",
manifest_managed=True,
help="Bind host. Managed by the install manifest on docker/service installs.",
tier="advanced",
),
SettingField(
"HEADROOM_PORT",
"port",
"Port",
"Networking",
"int",
default=8787,
minimum=1,
maximum=65535,
manifest_managed=True,
help="Bind port. Managed by the install manifest on docker/service installs.",
tier="advanced",
),
# --- Logging ---
SettingField(
"HEADROOM_LOG_MESSAGES",
"log_messages",
"Log message content",
"Logging",
"bool",
default=False,
help="Log full request/response content. WARNING: may log sensitive data.",
tier="basic",
),
SettingField(
"HEADROOM_LOG_FILE",
"log_file",
"Log file path",
"Logging",
"str",
default=None,
help="Path for the message log file.",
tier="basic",
),
# --- Networking (upstream connection pool tuning) ---
SettingField(
"HEADROOM_MAX_CONNECTIONS",
"max_connections",
"Max upstream connections",
"Networking",
"int",
default=500,
minimum=1,
help="Maximum upstream HTTP connections in the shared httpx pool.",
tier="advanced",
),
SettingField(
"HEADROOM_MAX_KEEPALIVE",
"max_keepalive_connections",
"Max keep-alive connections",
"Networking",
"int",
default=100,
minimum=0,
help="Maximum upstream keep-alive connections in the shared httpx pool.",
tier="advanced",
),
SettingField(
"HEADROOM_HTTP2",
"http2",
"HTTP/2 upstream",
"Networking",
"bool",
default=True,
help="Use HTTP/2 for upstream provider connections.",
tier="advanced",
),
SettingField(
"HEADROOM_HTTP_PROXY",
"http_proxy",
"Outbound HTTP proxy",
"Networking",
"str",
default=None,
help="HTTP proxy URL for upstream provider requests only (HTTPS uses CONNECT).",
tier="advanced",
),
SettingField(
"HEADROOM_KEEPALIVE_EXPIRY",
"keepalive_expiry",
"Keep-alive expiry (s)",
"Networking",
"float",
default=90.0,
minimum=0.0,
help="Seconds an idle upstream keep-alive connection is kept open.",
tier="advanced",
),
# --- Compression (additional internals) ---
SettingField(
"HEADROOM_NO_CCR_PROACTIVE_EXPANSION",
"no_ccr_proactive_expansion",
"Disable CCR proactive expansion",
"Compression",
"bool",
default=False,
help="Disable proactive expansion of previously compressed content.",
tier="advanced",
),
SettingField(
"HEADROOM_COMPRESSION_MAX_WORKERS",
"compression_max_workers",
"Compression worker pool size",
"Compression",
"int",
default=None,
help="Bound the dedicated compression threadpool (CPU-bound Kompress work). Unset = cpu_count.",
tier="advanced",
),
SettingField(
"HEADROOM_DISABLE_KOMPRESS_FALLBACK",
"disable_kompress_fallback",
"Disable Kompress fallback",
"Compression",
"bool",
default=False,
help="With disable-kompress, route fall-through content to passthrough instead of the Kompress fallback.",
tier="advanced",
),
SettingField(
"HEADROOM_DISABLE_KOMPRESS_ANTHROPIC",
"disable_kompress_anthropic",
"Disable Kompress (Anthropic)",
"Compression",
"optional-bool",
default=None,
help="Disable (false) or force-enable (true) Kompress for the Anthropic pipeline only. Unset = inherit.",
tier="advanced",
),
SettingField(
"HEADROOM_DISABLE_KOMPRESS_OPENAI",
"disable_kompress_openai",
"Disable Kompress (OpenAI)",
"Compression",
"optional-bool",
default=None,
help="Disable (false) or force-enable (true) Kompress for the OpenAI/Codex pipeline only. Unset = inherit.",
tier="advanced",
),
# --- CCR (experimental read-maturation) ---
SettingField(
"HEADROOM_READ_MATURATION",
"read_maturation",
"Read maturation",
"CCR",
"bool",
default=False,
help="EXPERIMENTAL: hold fresh Reads out of compression until the file quiesces.",
tier="advanced",
),
SettingField(
"HEADROOM_READ_MATURATION_QUIESCE_TURNS",
"read_maturation_quiesce_turns",
"Maturation quiesce turns",
"CCR",
"int",
default=5,
minimum=1,
help="Turns a file must stay quiet before a held Read is matured.",
tier="advanced",
),
SettingField(
"HEADROOM_READ_MATURATION_MAX_HOLD_TURNS",
"read_maturation_max_hold_turns",
"Maturation max hold turns",
"CCR",
"int",
default=25,
minimum=1,
help="Force-mature a held Read after this many turns even if the file stays active.",
tier="advanced",
),
SettingField(
"HEADROOM_READ_MATURATION_MIN_SIZE_BYTES",
"read_maturation_min_size_bytes",
"Maturation min size (bytes)",
"CCR",
"int",
default=2048,
minimum=0,
help="Only hold/mature Read outputs at least this many bytes.",
tier="advanced",
),
# --- Extensions ---
SettingField(
"HEADROOM_PROXY_EXTENSIONS",
"proxy_extensions",
"Enabled proxy extensions",
"Extensions",
"csv-list",
default=None,
help="Comma-separated opt-in proxy extension entry-point names ('*' enables all discovered).",
tier="advanced",
),
# --- Backend ---
SettingField(
"HEADROOM_NO_SUBSCRIPTION_TRACKING",
"no_subscription_tracking",
"Disable subscription tracking",
"Backend",
"bool",
default=False,
help="Disable the Anthropic Claude subscription usage poller (GET /api/oauth/usage).",
tier="advanced",
),
SettingField(
"HEADROOM_SUBSCRIPTION_POLL_INTERVAL",
"subscription_poll_interval",
"Subscription poll interval (s)",
"Backend",
"int",
default=None,
minimum=1,
maximum=3600,
help="Seconds between Anthropic subscription usage polls. Default: 300.",
tier="advanced",
),
SettingField(
"HEADROOM_BACKEND",
"backend",
"Upstream backend",
"Backend",
"str",
default="anthropic",
help="API backend: anthropic, bedrock, openrouter, anyllm, or litellm-<provider>.",
tier="basic",
),
SettingField(
"HEADROOM_ANYLLM_PROVIDER",
"anyllm_provider",
"any-llm provider",
"Backend",
"str",
default="openai",
help="Provider for the any-llm backend: openai, mistral, groq, ollama, etc.",
tier="advanced",
),
SettingField(
"HEADROOM_REGION",
"region",
"Cloud region",
"Backend",
"str",
default="us-west-2",
help="Cloud region for Bedrock/Vertex/etc backends.",
tier="advanced",
),
# --- Timeouts ---
SettingField(
"HEADROOM_RETRY_MAX_ATTEMPTS",
"retry_max_attempts",
"Upstream retry attempts",
"Timeouts",
"int",
default=None,
minimum=1,
maximum=10,
help="Maximum upstream retry attempts on connect/read/5xx failures. Default: 3.",
tier="advanced",
),
SettingField(
"HEADROOM_RETRY_BASE_DELAY_MS",
"retry_base_delay_ms",
"Retry base delay (ms)",
"Timeouts",
"int",
default=1000,
minimum=0,
help="Initial upstream retry delay in milliseconds. Default: 1000.",
tier="advanced",
),
SettingField(
"HEADROOM_RETRY_MAX_DELAY_MS",
"retry_max_delay_ms",
"Retry max delay (ms)",
"Timeouts",
"int",
default=30000,
minimum=0,
help="Maximum upstream retry delay in milliseconds. Default: 30000.",
tier="advanced",
),
SettingField(
"HEADROOM_REQUEST_TIMEOUT",
"request_timeout",
"Request timeout (s)",
"Timeouts",
"int",
default=None,
help="Overall upstream request timeout in seconds. Default: 300.",
tier="advanced",
),
SettingField(
"HEADROOM_CONNECT_TIMEOUT_SECONDS",
"connect_timeout_seconds",
"Connect timeout (s)",
"Timeouts",
"int",
default=None,
minimum=1,
maximum=300,
help="Upstream connection timeout in seconds. Default: 10.",
tier="advanced",
),
SettingField(
"HEADROOM_ANTHROPIC_BUFFERED_REQUEST_TIMEOUT_SECONDS",
"anthropic_buffered_request_timeout_seconds",
"Anthropic buffered timeout (s)",
"Timeouts",
"int",
default=None,
minimum=1,
help="Buffered Anthropic read timeout for non-streaming message batch paths. Default: 600.",
tier="advanced",
),
SettingField(
"HEADROOM_ANTHROPIC_PRE_UPSTREAM_CONCURRENCY",
"anthropic_pre_upstream_concurrency",
"Pre-upstream concurrency gate",
"Timeouts",
"int",
default=None,
help="Cap concurrent Anthropic pre-upstream work. Default: max(2, min(8, cpu_count)).",
tier="advanced",
),
SettingField(
"HEADROOM_ANTHROPIC_PRE_UPSTREAM_ACQUIRE_TIMEOUT_SECONDS",
"anthropic_pre_upstream_acquire_timeout_seconds",
"Pre-upstream acquire timeout (s)",
"Timeouts",
"float",
default=None,
help="Fail-fast timeout waiting on the pre-upstream semaphore. Default: 15.0.",
tier="advanced",
),
SettingField(
"HEADROOM_ANTHROPIC_PRE_UPSTREAM_MEMORY_CONTEXT_TIMEOUT_SECONDS",
"anthropic_pre_upstream_memory_context_timeout_seconds",
"Pre-upstream memory-context timeout (s)",
"Timeouts",
"float",
default=None,
help="Fail-open timeout for memory-context lookup while holding a pre-upstream slot. Default: 2.0.",
tier="advanced",
),
# --- Memory ---
SettingField(
"HEADROOM_MEMORY_DB_PATH",
"memory_db_path",
"Memory DB path",
"Memory",
"str",
default="",
help="Path to the legacy single-file memory SQLite DB. Default: {cwd}/.headroom/memory.db.",
tier="advanced",
),
SettingField(
"HEADROOM_MEMORY_PROJECT_ROOT",
"memory_project_root",
"Memory project root",
"Memory",
"str",
default="",
help="Override the project root used for --memory-storage=project.",
tier="advanced",
),
SettingField(
"HEADROOM_NO_MEMORY_TOOLS",
"no_memory_tools",
"Disable memory tools",
"Memory",
"bool",
default=False,
help="Disable automatic injection of memory_save/memory_search tools.",
tier="basic",
),
SettingField(
"HEADROOM_NO_MEMORY_CONTEXT",
"no_memory_context",
"Disable memory context injection",
"Memory",
"bool",
default=False,
help="Disable automatic injection of relevant memories into the system prompt.",
tier="basic",
),
SettingField(
"HEADROOM_MEMORY_TOP_K",
"memory_top_k",
"Memory retrieval top-K",
"Memory",
"int",
default=10,
minimum=1,
maximum=100,
help="Number of semantically-relevant memories to retrieve.",
tier="advanced",
),
SettingField(
"HEADROOM_MIN_EVIDENCE",
"min_evidence",
"Minimum evidence count",
"Memory",
"int",
default=None,
minimum=1,
help="Minimum times a pattern must be observed before it is persisted to memory. Default: 5.",
tier="advanced",
),
# --- Endpoints (custom Anthropic/OpenAI upstream) ---
SettingField(
"ANTHROPIC_TARGET_API_URL",
"anthropic_base_url",
"Anthropic base URL",
"Endpoints",
"str",
default=None,
help="Custom Anthropic API base URL (e.g. Azure Foundry, corporate gateway). Overrides https://api.anthropic.com.",
tier="advanced",
),
SettingField(
"OPENAI_TARGET_API_URL",
"openai_base_url",
"OpenAI base URL",
"Endpoints",
"str",
default=None,
help="Custom OpenAI API base URL (e.g. corporate gateway). Overrides https://api.openai.com.",
tier="advanced",
),
SettingField(
"ANTHROPIC_TARGET_API_HEADERS",
"anthropic_extra_headers",
"Anthropic extra headers",
"Endpoints",
"header-map",
default=None,
secret=True,
help='JSON object of extra headers merged into (and overriding) forwarded Anthropic requests, e.g. {"Api-Key": "..."}.',
tier="advanced",
),
SettingField(
"OPENAI_TARGET_API_HEADERS",
"openai_extra_headers",
"OpenAI extra headers",
"Endpoints",
"header-map",
default=None,
secret=True,
help="JSON object of extra headers merged into (and overriding) forwarded OpenAI requests.",
tier="advanced",
),
)
_BY_KEY: dict[str, SettingField] = {f.key: f for f in SETTINGS}
class SettingsValidationError(Exception):
"""Raised when a settings payload has unknown keys or invalid values.
Carries structured detail so the API layer can map unknown keys to 400 and
per-field type/range errors to 422.
"""
def __init__(self, unknown_keys: list[str], field_errors: dict[str, str]) -> None:
self.unknown_keys = unknown_keys
self.field_errors = field_errors
super().__init__(
f"settings validation failed: unknown={unknown_keys} errors={field_errors}"
)
def _coerce(field: SettingField, value: Any) -> Any:
"""Coerce a raw JSON/env value to the field's Python type.
Returns ``None`` for null and empty values (empty coerces to ``None`` for
every type except a plain ``bool``, which becomes ``False``). Raises
``ValueError`` on bad input so callers can surface a per-field message.
"""
if value is None:
return None
if field.type in ("bool", "optional-bool"):
if isinstance(value, bool):
return value
token = str(value).strip().lower()
if field.type == "optional-bool" and token == "":
return None
if token in ("1", "true", "yes", "on"):
return True
if token in ("0", "false", "no", "off", ""):
return False
raise ValueError(f"expected a boolean, got {value!r}")
if field.type in ("int", "float"):
if isinstance(value, bool): # bool is an int subclass — reject explicitly
raise ValueError(f"expected a number, got {value!r}")
number: int | float
if field.type == "int":
if isinstance(value, float) and not value.is_integer():
raise ValueError(f"expected an integer, got {value!r}")
number = int(value)
else:
number = float(value)
if not math.isfinite(number):
raise ValueError(f"expected a finite number, got {value!r}")
if field.minimum is not None and number < field.minimum:
raise ValueError(f"must be >= {field.minimum}")
if field.maximum is not None and number > field.maximum:
raise ValueError(f"must be <= {field.maximum}")
return number
if field.type == "enum":
token = str(value)
if token not in field.choices:
raise ValueError(f"{token!r} not one of {list(field.choices)}")
return token
if field.type == "csv-list":
tokens = value if isinstance(value, list | tuple) else str(value).split(",")
tokens = [str(token).strip() for token in tokens]
tokens = [token for token in tokens if token]
return ",".join(tokens) if tokens else None
if field.type == "header-map":
if isinstance(value, dict):
parsed = value
else:
try:
parsed = json.loads(str(value))
except (ValueError, TypeError) as exc:
raise ValueError("expected a JSON object of header name/value strings") from exc
if not isinstance(parsed, dict) or not all(
isinstance(k, str) and isinstance(v, str) for k, v in parsed.items()
):
raise ValueError("expected a JSON object of header name/value strings")
return json.dumps(parsed, sort_keys=True) if parsed else None
# str
token = str(value)
return token if token != "" else None
def _serialize(field: SettingField, value: Any) -> str:
"""Serialize a coerced value to the exact string its env var expects."""
if field.type in ("bool", "optional-bool"):
return "1" if value else "0"
return str(value)
def validate(values: dict[str, Any]) -> dict[str, Any]:
"""Validate/coerce ``values`` against the registry.
Raises :class:`SettingsValidationError` when any key is unknown or any value
fails coercion. Returns the coerced dict (``None`` values dropped) on success.
"""
unknown = [key for key in values if key not in _BY_KEY]
field_errors: dict[str, str] = {}
coerced: dict[str, Any] = {}
for key, value in values.items():
field = _BY_KEY.get(key)
if field is None:
continue
try:
result = _coerce(field, value)
except (ValueError, TypeError) as exc:
field_errors[key] = str(exc)
continue
if result is not None:
coerced[key] = result
if unknown or field_errors:
raise SettingsValidationError(unknown, field_errors)
return coerced
def load() -> dict[str, Any]:
"""Return validated stored values. Fail-open: ``{}`` if missing or corrupt."""
path = paths.settings_path()
try:
raw = path.read_text(encoding="utf-8")
except FileNotFoundError:
return {}
except OSError as exc:
logger.warning("settings_store: cannot read %s: %s", path, exc)
return {}
try:
data = json.loads(raw)
except (ValueError, UnicodeDecodeError) as exc:
logger.warning("settings_store: ignoring corrupt settings.json: %s", exc)
return {}
if not isinstance(data, dict):
logger.warning("settings_store: settings.json is not a JSON object; ignoring")
return {}
out: dict[str, Any] = {}
for key, value in data.items():
field = _BY_KEY.get(key)
if field is None:
continue # drop unknown keys
try:
result = _coerce(field, value)
except (ValueError, TypeError) as exc:
logger.warning("settings_store: dropping invalid %s: %s", key, exc)
continue
if result is not None:
out[key] = result
return out
def _atomic_write_text(path: Path, data: str) -> None:
"""Write ``data`` to ``path`` atomically (temp file + ``os.replace``).
A crash mid-write leaves either the previous file or the complete new one on
disk — never a truncated settings.json that would fail to parse and (via the
startup apply hook) crash-loop a supervised proxy. ``load()`` is also
fail-open as a second line of defence.
"""
fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
tmp_path = Path(tmp_name)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(data)
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp_path, path)
except BaseException:
tmp_path.unlink(missing_ok=True)
raise
def save(values: dict[str, Any]) -> None:
"""Validate ``values`` and merge them into the existing stored settings.
A merge, not a wholesale replace -- callers submit only the fields that
changed, and anything already on disk for other keys is preserved so a
first save doesn't permanently pin every field's current default.
Three submission shapes per key, beyond the default "absent -> unchanged":
explicit ``None`` (JSON ``null``) clears the key from stored settings;
a secret field resent as the mask sentinel (``_MASK``) is retained as-is
and never coerced/overwritten, since the GUI always resends a masked
secret's display value verbatim when the user hasn't touched it; anything
else is validated/coerced and stored.
"""
clear_keys = {key for key, value in values.items() if value is None and key in _BY_KEY}
retained_keys = {
key
for key, value in values.items()
if key in _BY_KEY and _BY_KEY[key].secret and value == _MASK
}
to_validate = {
key: value
for key, value in values.items()
if key not in clear_keys and key not in retained_keys
}
validated = validate(to_validate)
merged = {**load(), **validated}
for key in clear_keys:
merged.pop(key, None)
payload = json.dumps(merged, indent=2, sort_keys=True) + "\n"
paths.ensure_workspace_dir()
_atomic_write_text(paths.settings_path(), payload)
def apply_to_environ(values: dict[str, Any]) -> None:
"""``setdefault`` each stored value into ``os.environ`` (explicit export wins)."""
for key, value in values.items():
field = _BY_KEY.get(key)
if field is None or value is None:
continue
os.environ.setdefault(field.env, _serialize(field, value))
def effective_values(stored: dict[str, Any] | None = None) -> dict[str, Any]:
"""The value actually active now for each knob: default ← file ← environ."""
if stored is None:
stored = load()
result: dict[str, Any] = {}
for field in SETTINGS:
value = stored[field.key] if field.key in stored else field.default
env_raw = os.environ.get(field.env)
if env_raw is not None and env_raw != "":
try:
value = _coerce(field, env_raw)
except (ValueError, TypeError):
pass # unparseable env: keep the file/default value
result[field.key] = value
return result
def _mask(field: SettingField, value: Any) -> Any:
if field.secret and value not in (None, ""):
return _MASK
return value
def stored_values(mask_secrets: bool = True) -> dict[str, Any]:
"""Stored file values (for ``GET /settings``); secret values masked."""
values = load()
if not mask_secrets:
return values
return {key: _mask(_BY_KEY[key], value) for key, value in values.items()}
def to_schema() -> dict[str, Any]:
"""Registry + grouped fields + effective values for the UI. Secrets masked.
All curated knobs are startup-captured, so every key is restart-required;
``needs_restart_keys`` lists them for the UI's "restart to apply" banner.
"""
stored = load()
effective = effective_values(stored)
fields: list[dict[str, Any]] = []
for field in SETTINGS:
fields.append(
{
"key": field.key,
"env": field.env,
"label": field.label,
"group": field.group,
"type": field.type,
"choices": list(field.choices),
"default": field.default,
"help": field.help,
"secret": field.secret,
"manifest_managed": field.manifest_managed,
"minimum": field.minimum,
"maximum": field.maximum,
"tier": field.tier,
"env_override": bool(os.environ.get(field.env)),
"value": _mask(field, effective.get(field.key)),
"stored": _mask(field, stored.get(field.key)),
}
)
groups: list[str] = []
for field in SETTINGS:
if field.group not in groups:
groups.append(field.group)
return {
"groups": groups,
"fields": fields,
"values": {f["key"]: f["value"] for f in fields},
"needs_restart_keys": [field.key for field in SETTINGS],
}

6
pyproject.toml
View file

@ -7,7 +7,7 @@ name = "headroom-ai"
version = "0.32.0"
description = "The Context Optimization Layer for LLM Applications - Cut costs by 50-90%"
readme = "README.md"
license = "Apache-2.0"
license = "Apache-2.0"
requires-python = ">=3.10"
authors = [
{ name = "Headroom Contributors" }
@ -54,7 +54,7 @@ dependencies = [
# ImportError-guarded. Marking it 3.14-optional lets headroom install on Python 3.14
# (core compression + the Anthropic proxy path never import litellm). See GH #956.
"litellm>=1.86.2,<2.0; python_version < '3.14'", # model registry, pricing, providers (lazy)
"click>=8.3.3", # CLI framework
"click>=8.3.3", # CLI framework; PYSEC-2026-2132 fix (command injection in click.edit())
"rich>=13.0.0", # Rich terminal output
"opentelemetry-api>=1.24.0", # Safe no-op OTEL API for instrumentation
"ast-grep-cli>=0.30.0", # AST-aware code slicing (CodeCompressor); binary wheel
@ -165,7 +165,7 @@ relevance = [
# `headroom/image/compressor.py` adapts both API shapes at runtime via
# a try/except cascade. See issue #372 for context.
image = [
"pillow>=12.3.0",
"pillow>=12.3.0", # PYSEC-2026-2253/2254/2255/2256/2257 fixes (decompression-bomb + cmd-injection)
"sentencepiece>=0.1.99", # Required by SigLIP tokenizer (SiglipTokenizer)
# Python 3.6–3.12: keep the proven ORT-bundled package directly.
# ~15 MB ONNX models auto-downloaded on first use.

18
tests/conftest.py
View file

@ -16,6 +16,19 @@ import pytest
from tests._skip_helpers import external_model_skip_reason
# A live `headroom` dev session exports HEADROOM_* into the shell (and the
# Claude wrap adds ANTHROPIC_CUSTOM_HEADERS). Click `envvar=` options pick
# those up inside CliRunner, so assertions would see the developer's proxy
# config instead of the test's. Scrub them so local runs match CI; tests
# that need a value set it explicitly via monkeypatch or CliRunner env.
@pytest.fixture(autouse=True)
def _scrub_developer_headroom_env(monkeypatch):
for key in list(os.environ):
if key.startswith("HEADROOM_"):
monkeypatch.delenv(key, raising=False)
monkeypatch.delenv("ANTHROPIC_CUSTOM_HEADERS", raising=False)
# =============================================================================
# Global test hooks
# =============================================================================
@ -67,6 +80,11 @@ def _reset_headroom_logger_propagation():
if _name == "headroom" or _name.startswith("headroom."):
logger = _logging.getLogger(_name)
logger.disabled = False
# The benchmark also raises the level to CRITICAL; children
# inherit it (effective level), so a WARNING would be filtered
# at the logger before it can propagate to caplog. Reset to
# NOTSET so the subtree inherits root's level deterministically.
logger.setLevel(_logging.NOTSET)
logger.propagate = True
yield

12
tests/test_cli/test_wrap_claude_base_url.py
View file

@ -5,6 +5,8 @@ from __future__ import annotations
import json
from pathlib import Path
import pytest
from headroom.cli import wrap as wrap_cli
@ -242,11 +244,17 @@ def test_wrap_marker_is_not_stale_for_live_pid(tmp_path: Path) -> None:
assert wrap_cli._wrap_marker_is_stale(marker) is False
def test_wrap_marker_is_stale_when_pid_reused(tmp_path: Path) -> None:
def test_wrap_marker_is_stale_when_pid_reused(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
# Inject a deterministic PID identity: _proc_identity returns None on
# macOS without psutil, where reuse detection is deliberately best-effort
# and this scenario would be undetectable.
monkeypatch.setattr(wrap_cli, "_proc_identity", lambda pid: ("test", 50_000.0))
path = _settings(tmp_path)
wrap_cli._write_claude_wrap_base_url("http://127.0.0.1:8787", settings_path=path, port=8787)
marker = json.loads(_marker(tmp_path).read_text(encoding="utf-8"))
marker["start_time"] = (marker["start_time"] or 0) - 10_000 # fabricate a mismatched identity
marker["start_time"] = marker["start_time"] - 10_000 # fabricate a mismatched identity
assert wrap_cli._wrap_marker_is_stale(marker) is True

8
tests/test_cli/test_wrap_codex.py
View file

@ -1594,7 +1594,13 @@ def test_wrap_codex_prepare_only_registers_serena_when_uvx_exists(
with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None):
with patch("headroom.cli.wrap.shutil.which", side_effect=fake_which):
result = runner.invoke(main, ["wrap", "codex", "--prepare-only"])
# tokensave is the primary code-graph compressor; Serena is only
# the backup, registered when tokensave is unavailable. Force it
# unavailable so this test deterministically exercises the Serena
# path regardless of whether a real tokensave binary was installed
# in the shared bin dir by an earlier test in the suite.
with patch("headroom.cli.wrap._ensure_tokensave_binary", return_value=None):
result = runner.invoke(main, ["wrap", "codex", "--prepare-only"])
assert result.exit_code == 0, result.output
content = config_file.read_text(encoding="utf-8")

56
tests/test_cli_proxy_env.py
View file

@ -1375,3 +1375,59 @@ class TestCLIProxyRpmTpm:
assert result.exit_code == 0, result.output
assert captured_config["config"].rate_limit_tokens_per_minute == 80000
class TestSettingsFileToEnv:
"""settings.json is applied to os.environ before Click parses envvar options.
Proves the file reaches both a parse-time ``envvar=`` option (HEADROOM_PORT)
and a body-resolved env read (HEADROOM_CODE_AWARE_ENABLED, which has no
Click ``envvar=``), and that an explicit shell export still wins.
"""
def test_settings_file_reaches_parse_time_and_body_options(self, runner, tmp_path):
(tmp_path / "settings.json").write_text(
'{"port": 9898, "code_aware_enabled": false}', encoding="utf-8"
)
captured_config = {}
def mock_run_server(config, **kwargs):
captured_config["config"] = config
with patch("headroom.proxy.server.run_server", mock_run_server):
result = runner.invoke(
main,
["proxy"],
env={
"HEADROOM_WORKSPACE_DIR": str(tmp_path),
# Ensure nothing ambient shadows the file-applied values.
"HEADROOM_PORT": None,
"HEADROOM_CODE_AWARE_ENABLED": None,
},
catch_exceptions=False,
)
assert result.exit_code == 0, result.output
assert captured_config["config"].port == 9898
assert captured_config["config"].code_aware_enabled is False
def test_explicit_export_overrides_settings_file(self, runner, tmp_path):
(tmp_path / "settings.json").write_text('{"port": 9898}', encoding="utf-8")
captured_config = {}
def mock_run_server(config, **kwargs):
captured_config["config"] = config
with patch("headroom.proxy.server.run_server", mock_run_server):
result = runner.invoke(
main,
["proxy"],
env={
"HEADROOM_WORKSPACE_DIR": str(tmp_path),
"HEADROOM_PORT": "7777",
},
catch_exceptions=False,
)
assert result.exit_code == 0, result.output
assert captured_config["config"].port == 7777

3
tests/test_copilot_auth.py
View file

@ -413,6 +413,9 @@ def test_read_cached_oauth_token_falls_back_to_gh_cli(monkeypatch: pytest.Monkey
monkeypatch.setattr(copilot_auth, "_read_windows_copilot_cli_oauth_token", lambda: None)
monkeypatch.setattr(copilot_auth, "_read_macos_keychain_oauth_token", lambda: None)
monkeypatch.setattr(copilot_auth, "_read_gh_cli_oauth_token", lambda: "gho-gh-cli")
# No cached token files — a developer machine may have a real
# ~/.config/github-copilot token that would win before the gh fallback.
monkeypatch.setattr(copilot_auth, "_resolve_token_file_paths", lambda: [])
assert copilot_auth.read_cached_oauth_token() == "gho-gh-cli"

69
tests/test_header_isolation.py
View file

@ -34,6 +34,7 @@ from fastapi.testclient import TestClient
from headroom.proxy.helpers import (
_strip_internal_headers,
get_strip_internal_headers_mode,
merge_extra_headers,
)
from headroom.proxy.server import ProxyConfig, create_app
@ -225,7 +226,7 @@ class _FakePrefixTracker:
return None
def _make_anthropic_app() -> tuple[TestClient, _CapturingTransport]:
def _make_anthropic_app(**config_overrides) -> tuple[TestClient, _CapturingTransport]:
config = ProxyConfig(
optimize=False,
cache_enabled=False,
@ -236,6 +237,7 @@ def _make_anthropic_app() -> tuple[TestClient, _CapturingTransport]:
ccr_handle_responses=False,
ccr_context_tracking=False,
image_optimize=False,
**config_overrides,
)
app = create_app(config)
proxy = app.state.proxy
@ -569,3 +571,68 @@ def test_openai_chat_x_headroom_bypass_not_forwarded() -> None:
assert "x-headroom-bypass" not in sent_headers
assert "x-headroom-user-id" not in sent_headers
assert sent_headers.get("authorization") == "Bearer sk-test"
# ---------------------------------------------------------------------------
# Configured extra headers (anthropic_extra_headers / openai_extra_headers)
# ---------------------------------------------------------------------------
def test_anthropic_extra_headers_merged_and_override_client_header() -> None:
"""Configured extra headers reach upstream and override same-named client headers."""
client, transport = _make_anthropic_app(
anthropic_extra_headers={"x-api-key": "gateway-key", "x-gateway-id": "gw-1"}
)
resp = client.post(
"/v1/messages",
headers={
"x-api-key": "client-key",
"anthropic-version": "2023-06-01",
},
json={
"model": "claude-sonnet-4-6",
"max_tokens": 16,
"messages": [{"role": "user", "content": "hi"}],
},
)
assert resp.status_code == 200, resp.text
assert transport.captured_headers is not None
upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
# Configured header wins over the client-sent value for the same key.
assert upstream.get("x-api-key") == "gateway-key"
# A configured header not sent by the client is still added.
assert upstream.get("x-gateway-id") == "gw-1"
def test_anthropic_no_extra_headers_configured_is_unchanged() -> None:
"""With no extra headers configured, forwarding behavior is unchanged."""
client, transport = _make_anthropic_app()
resp = client.post(
"/v1/messages",
headers={"x-api-key": "client-key", "anthropic-version": "2023-06-01"},
json={
"model": "claude-sonnet-4-6",
"max_tokens": 16,
"messages": [{"role": "user", "content": "hi"}],
},
)
assert resp.status_code == 200, resp.text
upstream = {k.lower(): v for k, v in transport.captured_headers.items()}
assert upstream.get("x-api-key") == "client-key"
assert "x-gateway-id" not in upstream
def test_merge_extra_headers_overrides_case_insensitively() -> None:
"""A configured extra header wins even when the client used different casing."""
out = merge_extra_headers(
{"Authorization": "client", "keep": "v"}, {"authorization": "gateway"}
)
assert out == {"authorization": "gateway", "keep": "v"}
# Exactly one authorization header survives (no duplicate casings upstream).
assert [k for k in out if k.lower() == "authorization"] == ["authorization"]
def test_merge_extra_headers_none_returns_same_object() -> None:
"""No configured extras -> caller's dict is returned unchanged (no copy)."""
headers = {"a": "b"}
assert merge_extra_headers(headers, None) is headers

94
tests/test_install/test_runtime.py
View file

@ -694,3 +694,97 @@ def test_runtime_status_survives_winerror87_systemerror(monkeypatch, tmp_path: P
)
assert runtime_status(_python_service_manifest()) == "stopped"
class TestRestartCurrentDeployment:
"""detect_current_deployment / restart_current_deployment (settings apply)."""
def _clear_deployment_env(self, monkeypatch) -> None:
monkeypatch.delenv("HEADROOM_DEPLOYMENT_PROFILE", raising=False)
monkeypatch.delenv("HEADROOM_DEPLOYMENT_PRESET", raising=False)
def test_foreground_when_no_deployment_env(self, monkeypatch) -> None:
from headroom.install import runtime as rt
self._clear_deployment_env(monkeypatch)
popen_calls: list = []
monkeypatch.setattr(rt.subprocess, "Popen", lambda *a, **k: popen_calls.append(a))
manifest, mode = rt.detect_current_deployment()
assert manifest is None
assert mode == "foreground"
result = rt.restart_current_deployment()
assert result["restarted"] is False
assert result["mode"] == "foreground"
assert "instruction" in result
assert popen_calls == [] # never restarts a foreground proxy
def test_docker_returns_host_command_without_restarting(self, monkeypatch) -> None:
from headroom.install import runtime as rt
monkeypatch.setenv("HEADROOM_DEPLOYMENT_PROFILE", "default")
monkeypatch.setenv("HEADROOM_DEPLOYMENT_PRESET", InstallPreset.PERSISTENT_DOCKER.value)
monkeypatch.setattr(rt, "load_manifest", lambda profile: None)
popen_calls: list = []
monkeypatch.setattr(rt.subprocess, "Popen", lambda *a, **k: popen_calls.append(a))
_manifest, mode = rt.detect_current_deployment()
assert mode == "docker"
result = rt.restart_current_deployment()
assert result["restarted"] is False
assert result["mode"] == "docker"
assert result["command"] == "headroom install restart --profile default"
assert popen_calls == [] # cannot run docker from inside the container
def test_task_mode_detected_and_not_restarted(self, monkeypatch) -> None:
"""A persistent-task deployment must not be told it's a self-restartable 'service'.
``headroom install start/stop/restart`` all reject SupervisorKind.TASK
deployments (see cli/install.py:_reject_task_lifecycle); a detached
``headroom install restart`` against a task manifest would previously
fail silently (stdout/stderr to DEVNULL) after the API had already
returned ``{"restarted": true}``.
"""
from headroom.install import runtime as rt
monkeypatch.setenv("HEADROOM_DEPLOYMENT_PROFILE", "default")
monkeypatch.setenv("HEADROOM_DEPLOYMENT_PRESET", "persistent-task")
stub = types.SimpleNamespace(profile="default", supervisor_kind="task")
monkeypatch.setattr(rt, "load_manifest", lambda profile: stub)
popen_calls: list = []
monkeypatch.setattr(rt.subprocess, "Popen", lambda *a, **k: popen_calls.append(a))
_manifest, mode = rt.detect_current_deployment()
assert mode == "task"
result = rt.restart_current_deployment()
assert result["restarted"] is False
assert result["mode"] == "task"
assert "instruction" in result
assert popen_calls == [] # never spawns `headroom install restart` for task deployments
def test_service_spawns_detached_restart(self, monkeypatch) -> None:
from headroom.install import runtime as rt
monkeypatch.setenv("HEADROOM_DEPLOYMENT_PROFILE", "default")
monkeypatch.setenv("HEADROOM_DEPLOYMENT_PRESET", "persistent-service")
stub = types.SimpleNamespace(profile="default", supervisor_kind="service")
monkeypatch.setattr(rt, "load_manifest", lambda profile: stub)
recorded: dict = {}
def fake_popen(command, **kwargs):
recorded["command"] = command
recorded["kwargs"] = kwargs
return object()
monkeypatch.setattr(rt.subprocess, "Popen", fake_popen)
_manifest, mode = rt.detect_current_deployment()
assert mode == "service"
result = rt.restart_current_deployment()
assert result["restarted"] is True
assert result["mode"] == "service"
assert recorded["command"][-4:] == ["install", "restart", "--profile", "default"]

1
tests/test_openai_codex_routing.py
View file

@ -169,6 +169,7 @@ class _DummyOpenAIHandler(OpenAIHandlerMixin):
retry_base_delay_ms=10,
retry_max_delay_ms=50,
connect_timeout_seconds=10,
openai_extra_headers=None,
)
self.usage_reporter = None
self.openai_provider = SimpleNamespace(get_context_limit=lambda model: 128_000)

1
tests/test_openai_codex_ws_lifecycle.py
View file

@ -82,6 +82,7 @@ class _DummyOpenAIHandler(OpenAIHandlerMixin):
retry_base_delay_ms=1,
retry_max_delay_ms=1,
connect_timeout_seconds=10,
openai_extra_headers=None,
)
self.usage_reporter = None
self.openai_provider = SimpleNamespace(

1
tests/test_openai_codex_ws_timings.py
View file

@ -38,6 +38,7 @@ class _DummyOpenAIHandler(OpenAIHandlerMixin):
retry_base_delay_ms=1,
retry_max_delay_ms=1,
connect_timeout_seconds=10,
openai_extra_headers=None,
)
self.usage_reporter = None
self.openai_provider = SimpleNamespace(get_context_limit=lambda model: 128_000)

35
tests/test_provider_registry.py
View file

@ -2,6 +2,8 @@ from __future__ import annotations
import logging
import pytest
from headroom.providers.registry import (
ProviderApiOverrides,
build_proxy_provider_runtime,
@ -9,6 +11,7 @@ from headroom.providers.registry import (
format_backend_status,
resolve_api_overrides,
resolve_api_targets,
resolve_extra_headers,
)
from headroom.proxy.models import ProxyConfig
@ -418,3 +421,35 @@ def test_proxy_provider_runtime_openai_transport_handles_prompt_details_without_
assert metrics.tokens_output == 9
assert metrics.cached_tokens == 0
assert len(client._storage.saved) == 1
def test_resolve_extra_headers_cli_wins_over_env(monkeypatch) -> None:
monkeypatch.setenv("ANTHROPIC_TARGET_API_HEADERS", '{"Env-Header": "env-value"}')
result = resolve_extra_headers('{"Cli-Header": "cli-value"}', "ANTHROPIC_TARGET_API_HEADERS")
assert result == {"Cli-Header": "cli-value"}
def test_resolve_extra_headers_falls_back_to_env(monkeypatch) -> None:
monkeypatch.setenv("OPENAI_TARGET_API_HEADERS", '{"Env-Header": "env-value"}')
result = resolve_extra_headers(None, "OPENAI_TARGET_API_HEADERS")
assert result == {"Env-Header": "env-value"}
def test_resolve_extra_headers_unset_returns_none(monkeypatch) -> None:
monkeypatch.delenv("ANTHROPIC_TARGET_API_HEADERS", raising=False)
assert resolve_extra_headers(None, "ANTHROPIC_TARGET_API_HEADERS") is None
def test_resolve_extra_headers_invalid_json_raises(monkeypatch) -> None:
with pytest.raises(ValueError):
resolve_extra_headers("not json", "ANTHROPIC_TARGET_API_HEADERS")
def test_resolve_extra_headers_non_object_raises(monkeypatch) -> None:
with pytest.raises(ValueError):
resolve_extra_headers('["a", "b"]', "ANTHROPIC_TARGET_API_HEADERS")
def test_resolve_extra_headers_non_string_value_raises(monkeypatch) -> None:
with pytest.raises(ValueError):
resolve_extra_headers('{"Key": 123}', "ANTHROPIC_TARGET_API_HEADERS")

57
tests/test_proxy/test_settings_fresh_process_precedence.py Normal file
View file

@ -0,0 +1,57 @@
"""Prove env > file > default precedence in a genuinely separate process.
A mocked ``restart_current_deployment`` proves dispatch only -- it can't
prove settings actually take effect the way a real restarted proxy would
pick them up. This spawns a real subprocess that imports
``headroom.settings_store`` cold and reports what it observes, closing the
gap a Codex red-team pass flagged in the original (mock-only) test plan.
"""
import json
import os
import subprocess
import sys
import pytest
from headroom import paths, settings_store
@pytest.fixture
def workspace(tmp_path, monkeypatch):
"""Point the workspace dir (settings.json) at an isolated tmp dir."""
monkeypatch.setenv(paths.HEADROOM_WORKSPACE_DIR_ENV, str(tmp_path))
monkeypatch.delenv(paths.HEADROOM_SETTINGS_PATH_ENV, raising=False)
return tmp_path
def test_env_beats_file_beats_default_in_subprocess(workspace, monkeypatch):
for field in settings_store.SETTINGS:
monkeypatch.delenv(field.env, raising=False)
settings_store.save({"target_ratio": 0.3, "rpm": 20})
script = (
"import os, json\n"
"from headroom import settings_store\n"
"settings_store.apply_to_environ(settings_store.load())\n"
"print(json.dumps({'rpm': os.environ.get('HEADROOM_RPM'), "
"'target_ratio': os.environ.get('HEADROOM_TARGET_RATIO')}))\n"
)
env = dict(os.environ)
env["HEADROOM_WORKSPACE_DIR"] = str(workspace)
env["HEADROOM_RPM"] = "999" # explicit export: must win over the file's 20
env.pop(
"HEADROOM_TARGET_RATIO", None
) # not exported: file's 0.3 must win over the code default
result = subprocess.run(
[sys.executable, "-c", script],
capture_output=True,
text=True,
env=env,
timeout=30,
)
assert result.returncode == 0, result.stderr
out = json.loads(result.stdout)
assert out["rpm"] == "999"
assert out["target_ratio"] == "0.3"

290
tests/test_proxy/test_settings_store.py Normal file
View file

@ -0,0 +1,290 @@
"""Tests for the file-backed HEADROOM_* settings store (Phase 1).
Covers: JSON round-trip with coercion, unknown-key drop, fail-open load on a
corrupt file, atomic save, setdefault precedence (explicit export wins), the
effective-value resolution order, and secret masking in the schema/GET views.
"""
import json
import os
from dataclasses import replace
import pytest
from headroom import paths, settings_store
@pytest.fixture
def workspace(tmp_path, monkeypatch):
"""Point the workspace dir (and thus settings.json) at an isolated tmp dir."""
monkeypatch.setenv(paths.HEADROOM_WORKSPACE_DIR_ENV, str(tmp_path))
# Ensure no per-resource override leaks in from the ambient environment.
monkeypatch.delenv(settings_store.paths.HEADROOM_SETTINGS_PATH_ENV, raising=False)
return tmp_path
def _clear_env(monkeypatch):
for field in settings_store.SETTINGS:
monkeypatch.delenv(field.env, raising=False)
class TestRoundTrip:
def test_save_then_load_coerces_types(self, workspace, monkeypatch):
_clear_env(monkeypatch)
settings_store.save(
{
"port": "9898", # str in → int out
"target_ratio": "0.4", # str in → float out
"disable_kompress": "1", # str in → bool out
"savings_profile": "balanced",
}
)
loaded = settings_store.load()
assert loaded == {
"port": 9898,
"target_ratio": 0.4,
"disable_kompress": True,
"savings_profile": "balanced",
}
def test_load_drops_unknown_keys(self, workspace, monkeypatch):
_clear_env(monkeypatch)
path = paths.settings_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text('{"port": 9000, "bogus_key": 1}', encoding="utf-8")
assert settings_store.load() == {"port": 9000}
def test_save_is_atomic_no_temp_left_behind(self, workspace, monkeypatch):
_clear_env(monkeypatch)
settings_store.save({"port": 9000})
leftovers = [p.name for p in workspace.iterdir() if p.suffix == ".tmp"]
assert leftovers == []
assert paths.settings_path().exists()
class TestValidation:
def test_save_rejects_unknown_key(self, workspace):
with pytest.raises(settings_store.SettingsValidationError) as exc:
settings_store.save({"nope": 1})
assert exc.value.unknown_keys == ["nope"]
def test_save_rejects_out_of_range(self, workspace):
with pytest.raises(settings_store.SettingsValidationError) as exc:
settings_store.save({"target_ratio": 5.0})
assert "target_ratio" in exc.value.field_errors
def test_save_rejects_bad_enum(self, workspace):
with pytest.raises(settings_store.SettingsValidationError) as exc:
settings_store.save({"savings_profile": "nonsense"})
assert "savings_profile" in exc.value.field_errors
def test_save_rejects_non_numeric(self, workspace):
with pytest.raises(settings_store.SettingsValidationError) as exc:
settings_store.save({"rpm": "abc"})
assert "rpm" in exc.value.field_errors
def test_save_rejects_non_finite_float(self, workspace):
for bad in (float("nan"), float("inf"), float("-inf")):
with pytest.raises(settings_store.SettingsValidationError) as exc:
settings_store.save({"budget": bad})
assert "budget" in exc.value.field_errors
def test_optional_bool_empty_string_coerces_to_none(self, workspace):
# Empty optional-bool means "inherit" -> dropped, not forced to False.
assert settings_store.validate({"disable_kompress_anthropic": ""}) == {}
# A plain bool empty string still coerces to False.
assert settings_store.validate({"disable_kompress": ""}) == {"disable_kompress": False}
class TestFailOpen:
def test_corrupt_json_returns_empty(self, workspace):
path = paths.settings_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("{not valid json", encoding="utf-8")
assert settings_store.load() == {}
def test_non_object_json_returns_empty(self, workspace):
path = paths.settings_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("[1, 2, 3]", encoding="utf-8")
assert settings_store.load() == {}
def test_missing_file_returns_empty(self, workspace):
assert settings_store.load() == {}
class TestApplyToEnviron:
def test_setdefault_fills_unset_env(self, workspace, monkeypatch):
_clear_env(monkeypatch)
settings_store.apply_to_environ({"port": 9898, "disable_kompress": True})
assert os.environ["HEADROOM_PORT"] == "9898"
assert os.environ["HEADROOM_DISABLE_KOMPRESS"] == "1"
def test_explicit_export_wins(self, workspace, monkeypatch):
_clear_env(monkeypatch)
monkeypatch.setenv("HEADROOM_PORT", "7777")
settings_store.apply_to_environ({"port": 9898})
assert os.environ["HEADROOM_PORT"] == "7777"
def test_bool_false_serializes_to_zero(self, workspace, monkeypatch):
_clear_env(monkeypatch)
settings_store.apply_to_environ({"code_aware_enabled": False})
assert os.environ["HEADROOM_CODE_AWARE_ENABLED"] == "0"
class TestEffectiveValues:
def test_default_then_file_then_env(self, workspace, monkeypatch):
_clear_env(monkeypatch)
# default
assert settings_store.effective_values()["savings_profile"] == "coding"
# file overrides default
settings_store.save({"savings_profile": "balanced"})
assert settings_store.effective_values()["savings_profile"] == "balanced"
# env overrides file
monkeypatch.setenv("HEADROOM_SAVINGS_PROFILE", "general")
assert settings_store.effective_values()["savings_profile"] == "general"
class TestSecretMasking:
def test_schema_and_stored_mask_secret(self, workspace, monkeypatch):
_clear_env(monkeypatch)
# No curated knob is secret; synthesize one to exercise the masking path.
base_field = next(f for f in settings_store.SETTINGS if f.key == "log_file")
secret_field = replace(base_field, secret=True)
registry = tuple(
secret_field if f.key == "log_file" else f for f in settings_store.SETTINGS
)
monkeypatch.setattr(settings_store, "SETTINGS", registry)
monkeypatch.setattr(settings_store, "_BY_KEY", {f.key: f for f in registry})
settings_store.save({"log_file": "/tmp/secret.log"})
stored = settings_store.stored_values()
assert stored["log_file"] == settings_store._MASK
schema = settings_store.to_schema()
log_field = next(f for f in schema["fields"] if f["key"] == "log_file")
assert log_field["value"] == settings_store._MASK
assert log_field["stored"] == settings_store._MASK
# unmasked read still returns the real value for internal callers
assert settings_store.stored_values(mask_secrets=False)["log_file"] == "/tmp/secret.log"
def test_schema_lists_all_keys_as_restart_required(self, workspace, monkeypatch):
_clear_env(monkeypatch)
schema = settings_store.to_schema()
assert schema["needs_restart_keys"] == [f.key for f in settings_store.SETTINGS]
assert "Compression" in schema["groups"]
def test_anthropic_extra_headers_retain_on_mask(self, workspace, monkeypatch):
"""Saving _MASK for anthropic_extra_headers retains the stored value."""
_clear_env(monkeypatch)
settings_store.save({"anthropic_extra_headers": '{"Api-Key": "secret123"}'})
stored = settings_store.load()
assert stored.get("anthropic_extra_headers") == '{"Api-Key": "secret123"}'
settings_store.save({"anthropic_extra_headers": settings_store._MASK})
stored = settings_store.load()
assert stored.get("anthropic_extra_headers") == '{"Api-Key": "secret123"}', (
"Saving _MASK should retain the stored value, not overwrite it"
)
def test_anthropic_extra_headers_clear_on_none(self, workspace, monkeypatch):
"""Saving None for anthropic_extra_headers removes it."""
_clear_env(monkeypatch)
settings_store.save({"anthropic_extra_headers": '{"Api-Key": "secret123"}'})
stored = settings_store.load()
assert "anthropic_extra_headers" in stored
settings_store.save({"anthropic_extra_headers": None})
stored = settings_store.load()
assert "anthropic_extra_headers" not in stored
def test_anthropic_extra_headers_json_validation_roundtrip(self, workspace, monkeypatch):
"""JSON header-map values are canonical and round-trip correctly."""
_clear_env(monkeypatch)
settings_store.save({"anthropic_extra_headers": '{"Z-Header": "val", "A-Header": "val2"}'})
stored = settings_store.load()
canonical = json.dumps({"A-Header": "val2", "Z-Header": "val"}, sort_keys=True)
assert stored.get("anthropic_extra_headers") == canonical
def test_anthropic_extra_headers_invalid_json_raises(self, workspace, monkeypatch):
"""Invalid JSON in anthropic_extra_headers field raises SettingsValidationError."""
_clear_env(monkeypatch)
from headroom.settings_store import SettingsValidationError
with pytest.raises(SettingsValidationError) as exc_info:
settings_store.save({"anthropic_extra_headers": "not json"})
assert "anthropic_extra_headers" in exc_info.value.field_errors
def test_anthropic_extra_headers_non_string_values_raises(self, workspace, monkeypatch):
"""Header-map JSON with non-string values raises SettingsValidationError."""
_clear_env(monkeypatch)
from headroom.settings_store import SettingsValidationError
with pytest.raises(SettingsValidationError) as exc_info:
settings_store.save({"anthropic_extra_headers": '{"header": 123}'})
assert "anthropic_extra_headers" in exc_info.value.field_errors
def test_anthropic_extra_headers_non_object_raises(self, workspace, monkeypatch):
"""Header-map with non-object JSON raises SettingsValidationError."""
_clear_env(monkeypatch)
from headroom.settings_store import SettingsValidationError
with pytest.raises(SettingsValidationError) as exc_info:
settings_store.save({"anthropic_extra_headers": '["header", "value"]'})
assert "anthropic_extra_headers" in exc_info.value.field_errors
def test_openai_extra_headers_retain_on_mask(self, workspace, monkeypatch):
"""Saving _MASK for openai_extra_headers retains the stored value."""
_clear_env(monkeypatch)
settings_store.save({"openai_extra_headers": '{"Authorization": "Bearer token"}'})
stored = settings_store.load()
assert stored.get("openai_extra_headers") == '{"Authorization": "Bearer token"}'
settings_store.save({"openai_extra_headers": settings_store._MASK})
stored = settings_store.load()
assert stored.get("openai_extra_headers") == '{"Authorization": "Bearer token"}'
def test_anthropic_base_url_plain_str_field(self, workspace, monkeypatch):
"""anthropic_base_url behaves as a plain str field."""
_clear_env(monkeypatch)
settings_store.save({"anthropic_base_url": "https://custom.example.com/v1"})
stored = settings_store.load()
assert stored.get("anthropic_base_url") == "https://custom.example.com/v1"
settings_store.save({"anthropic_base_url": "https://other.example.com"})
stored = settings_store.load()
assert stored.get("anthropic_base_url") == "https://other.example.com"
def test_openai_base_url_plain_str_field(self, workspace, monkeypatch):
"""openai_base_url behaves as a plain str field."""
_clear_env(monkeypatch)
settings_store.save({"openai_base_url": "https://custom.openai.example.com/v1"})
stored = settings_store.load()
assert stored.get("openai_base_url") == "https://custom.openai.example.com/v1"
class TestRegistryDriftAgainstClick:
"""Guards against the registry silently falling behind the real CLI surface.
Introspects the ``proxy`` Click command's own parameter objects (not a
regex over the source) so a future contributor who adds a new
``@click.option(..., envvar="HEADROOM_...")`` to cli/proxy.py without
adding a matching SettingField gets a failing test, not a silent gap.
"""
def test_every_headroom_click_envvar_is_in_the_registry(self):
from headroom.cli.proxy import proxy
click_envvars = {
param.envvar
for param in proxy.params
if isinstance(getattr(param, "envvar", None), str)
and param.envvar.startswith("HEADROOM_")
}
registry_envvars = {field.env for field in settings_store.SETTINGS}
missing = click_envvars - registry_envvars
assert not missing, (
f"New HEADROOM_* Click option(s) not covered by settings_store.SETTINGS: "
f"{sorted(missing)}. Add a SettingField for each, or document why it's "
"deliberately excluded (e.g. a secret)."
)

286
tests/test_proxy_settings_endpoints.py Normal file
View file

@ -0,0 +1,286 @@
"""Tests for the dashboard settings API endpoints (Phases 2-4).
Covers GET /settings/schema, GET /settings, POST /settings, POST /settings/apply
and the GET /dashboard/settings route: loopback gating on all read/write settings
endpoints, registry validation (400 unknown key / 422 bad value), secret masking,
and the deployment-aware apply/restart dispatch (service 202 + background restart,
docker host command, foreground instruction).
"""
from __future__ import annotations
from dataclasses import replace
import pytest
pytest.importorskip("fastapi")
from fastapi.testclient import TestClient # noqa: E402
from headroom import settings_store # noqa: E402
from headroom.proxy.server import ProxyConfig, create_app # noqa: E402
def _make_app():
return create_app(
ProxyConfig(
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,
image_optimize=False,
)
)
@pytest.fixture
def workspace(tmp_path, monkeypatch):
"""Isolate settings.json under a tmp workspace for each test."""
monkeypatch.setenv("HEADROOM_WORKSPACE_DIR", str(tmp_path))
monkeypatch.delenv("HEADROOM_SETTINGS_PATH", raising=False)
# Make deployment detection deterministic: no supervisor env -> foreground.
monkeypatch.delenv("HEADROOM_DEPLOYMENT_PROFILE", raising=False)
monkeypatch.delenv("HEADROOM_DEPLOYMENT_PRESET", raising=False)
return tmp_path
@pytest.fixture
def client(workspace):
"""Loopback client — passes both the client-IP and Host-header gates."""
return TestClient(_make_app(), base_url="http://127.0.0.1", client=("127.0.0.1", 12345))
@pytest.fixture
def network_client(workspace):
"""Default TestClient presents client.host='testclient' — treated as non-loopback."""
return TestClient(_make_app())
class TestSchemaAndRead:
def test_schema_returns_grouped_fields(self, client):
resp = client.get("/settings/schema")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["groups"]
assert body["fields"]
assert body["needs_restart_keys"]
assert "supervised" in body # server-added deployment flag
assert body["supervised"] is False # foreground in tests
def test_get_settings_reflects_saved_values(self, client, workspace):
settings_store.save({"target_ratio": 0.5, "savings_profile": "balanced"})
resp = client.get("/settings")
assert resp.status_code == 200, resp.text
assert resp.json() == {"target_ratio": 0.5, "savings_profile": "balanced"}
class TestEndpointsGroup:
def test_schema_includes_endpoints_group_and_secret_flags(self, client):
resp = client.get("/settings/schema")
assert resp.status_code == 200, resp.text
body = resp.json()
assert "Endpoints" in body["groups"]
by_key = {f["key"]: f for f in body["fields"]}
for key in (
"anthropic_base_url",
"openai_base_url",
"anthropic_extra_headers",
"openai_extra_headers",
):
assert key in by_key, f"missing field {key}"
assert by_key["anthropic_extra_headers"]["secret"] is True
assert by_key["openai_extra_headers"]["secret"] is True
assert by_key["anthropic_base_url"]["secret"] is False
def test_get_settings_masks_extra_headers(self, client, workspace):
settings_store.save({"anthropic_extra_headers": '{"Api-Key": "super-secret"}'})
resp = client.get("/settings")
assert resp.status_code == 200, resp.text
assert resp.json()["anthropic_extra_headers"] == settings_store._MASK
def test_schema_values_mask_extra_headers(self, client, workspace):
settings_store.save({"openai_extra_headers": '{"Api-Key": "super-secret"}'})
resp = client.get("/settings/schema")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["values"]["openai_extra_headers"] == settings_store._MASK
class TestWriteValidation:
def test_valid_write_persists(self, client, workspace):
resp = client.post("/settings", json={"values": {"target_ratio": 0.4, "rpm": 30}})
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["ok"] is True
assert body["needs_restart"] is True
assert set(body["changed_keys"]) == {"target_ratio", "rpm"}
assert settings_store.load() == {"target_ratio": 0.4, "rpm": 30}
def test_unknown_key_400(self, client):
resp = client.post("/settings", json={"values": {"nope": 1}})
assert resp.status_code == 400, resp.text
assert "nope" in resp.json()["unknown_keys"]
def test_bad_value_422(self, client):
resp = client.post("/settings", json={"values": {"target_ratio": 5}})
assert resp.status_code == 422, resp.text
assert "target_ratio" in resp.json()["field_errors"]
def test_missing_values_object_400(self, client):
resp = client.post("/settings", json={"foo": 1})
assert resp.status_code == 400, resp.text
class TestLoopbackGating:
def test_reads_and_writes_rejected_off_loopback(self, network_client):
assert network_client.get("/settings/schema").status_code == 404
assert network_client.get("/settings").status_code == 404
assert network_client.post("/settings", json={"values": {}}).status_code == 404
assert network_client.post("/settings/apply", json={}).status_code == 404
def test_settings_page_rejected_off_loopback(self, network_client):
assert network_client.get("/dashboard/settings").status_code == 404
class TestSameOriginGuard:
"""CSRF: a same-machine (loopback) attacker page can still send a
non-preflighted 'simple' request straight to a known 127.0.0.1 URL --
the Host header reads the real (loopback) destination either way, but
a real browser's Origin header reflects the page's actual origin.
require_loopback's Host-header check alone does not catch this.
"""
def test_foreign_origin_rejected_on_settings_post(self, client, workspace):
resp = client.post(
"/settings",
json={"values": {"target_ratio": 0.4}},
headers={"Origin": "https://evil.example"},
)
assert resp.status_code == 403, resp.text
def test_null_origin_rejected(self, client, workspace):
resp = client.post(
"/settings",
json={"values": {"target_ratio": 0.4}},
headers={"Origin": "null"},
)
assert resp.status_code == 403, resp.text
def test_foreign_origin_rejected_on_settings_apply(self, client, monkeypatch):
from headroom.install import runtime as rt
monkeypatch.setattr(
rt, "restart_current_deployment", lambda: {"restarted": False, "mode": "foreground"}
)
resp = client.post(
"/settings/apply",
json={},
headers={"Origin": "https://evil.example"},
)
assert resp.status_code == 403, resp.text
def test_loopback_origin_allowed(self, client, workspace):
resp = client.post(
"/settings",
json={"values": {"target_ratio": 0.4}},
headers={"Origin": "http://127.0.0.1:8787"},
)
assert resp.status_code == 200, resp.text
def test_absent_origin_allowed(self, client, workspace):
# No Origin header at all (CLI tools, curl, TestClient's default) must pass.
resp = client.post("/settings", json={"values": {"target_ratio": 0.4}})
assert resp.status_code == 200, resp.text
class TestSecretMasking:
def test_secret_never_returned_unmasked(self, client, workspace, monkeypatch):
# No curated knob is secret; flag one to exercise the masking path end-to-end.
base_field = next(f for f in settings_store.SETTINGS if f.key == "log_file")
secret_field = replace(base_field, secret=True)
registry = tuple(
secret_field if f.key == "log_file" else f for f in settings_store.SETTINGS
)
monkeypatch.setattr(settings_store, "SETTINGS", registry)
monkeypatch.setattr(settings_store, "_BY_KEY", {f.key: f for f in registry})
settings_store.save({"log_file": "/tmp/secret.log"})
assert client.get("/settings").json()["log_file"] == settings_store._MASK
schema = client.get("/settings/schema").json()
log_field = next(f for f in schema["fields"] if f["key"] == "log_file")
assert log_field["value"] == settings_store._MASK
assert log_field["stored"] == settings_store._MASK
class TestApplyRestart:
def _patch_mode(self, monkeypatch, mode, restart_result):
from headroom.install import runtime as rt
monkeypatch.setattr(rt, "detect_current_deployment", lambda: (None, mode))
calls = []
def fake_restart():
calls.append(True)
return restart_result
monkeypatch.setattr(rt, "restart_current_deployment", fake_restart)
return calls
def test_service_returns_202_and_runs_background_restart(self, client, monkeypatch):
calls = self._patch_mode(monkeypatch, "service", {"restarted": True, "mode": "service"})
resp = client.post("/settings/apply", json={})
assert resp.status_code == 202, resp.text
assert resp.json()["restarted"] is True
# TestClient runs the BackgroundTask after sending the response.
assert calls == [True]
def test_docker_returns_host_command(self, client, monkeypatch):
self._patch_mode(
monkeypatch,
"docker",
{
"restarted": False,
"mode": "docker",
"command": "headroom install restart --profile default",
},
)
resp = client.post("/settings/apply", json={})
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["mode"] == "docker"
assert "headroom install restart" in body["command"]
def test_foreground_returns_instruction(self, client, monkeypatch):
self._patch_mode(
monkeypatch,
"foreground",
{
"restarted": False,
"mode": "foreground",
"instruction": "Restart the proxy to apply the new settings.",
},
)
resp = client.post("/settings/apply", json={})
assert resp.status_code == 200, resp.text
assert "instruction" in resp.json()
def test_apply_persists_provided_values(self, client, workspace, monkeypatch):
self._patch_mode(
monkeypatch,
"foreground",
{"restarted": False, "mode": "foreground", "instruction": "x"},
)
resp = client.post("/settings/apply", json={"values": {"rpm": 42}})
assert resp.status_code == 200, resp.text
assert settings_store.load() == {"rpm": 42}
class TestSettingsPageRoute:
def test_dashboard_settings_serves_html(self, client):
resp = client.get("/dashboard/settings")
assert resp.status_code == 200, resp.text
assert "<" in resp.text # rendered HTML page

4
wiki/cli.md
View file

@ -268,6 +268,8 @@ headroom proxy --mode cache
| `--anyllm-provider` | `openai` | Provider name for `anyllm` |
| `--anthropic-api-url` | unset | Custom Anthropic passthrough API URL |
| `--openai-api-url` | unset | Custom OpenAI passthrough API URL |
| `--anthropic-extra-headers` | unset | JSON object of extra headers merged into (and overriding) forwarded Anthropic requests |
| `--openai-extra-headers` | unset | JSON object of extra headers merged into (and overriding) forwarded OpenAI requests |
| `--gemini-api-url` | unset | Custom Gemini passthrough API URL |
| `--region` | `us-west-2` | Cloud region for Bedrock / Vertex / related backends |
| `--bedrock-region` | unset | Deprecated Bedrock region override |
@ -278,7 +280,7 @@ headroom proxy --mode cache
Notes:
- `--learn` implies memory unless `--no-learn` is also set.
- Proxy startup can also read environment variables such as `HEADROOM_HOST`, `HEADROOM_PORT`, `HEADROOM_BUDGET`, `HEADROOM_MODE`, `HEADROOM_ANYLLM_PROVIDER`, `HEADROOM_ANTHROPIC_PRE_UPSTREAM_CONCURRENCY`, `HEADROOM_ANTHROPIC_PRE_UPSTREAM_ACQUIRE_TIMEOUT_SECONDS`, `HEADROOM_REQUEST_TIMEOUT`, `HEADROOM_ANTHROPIC_PRE_UPSTREAM_MEMORY_CONTEXT_TIMEOUT_SECONDS`, `ANTHROPIC_TARGET_API_URL`, `OPENAI_TARGET_API_URL`, and `GEMINI_TARGET_API_URL`. CLI flags take precedence over environment variables.
- Proxy startup can also read environment variables such as `HEADROOM_HOST`, `HEADROOM_PORT`, `HEADROOM_BUDGET`, `HEADROOM_MODE`, `HEADROOM_ANYLLM_PROVIDER`, `HEADROOM_ANTHROPIC_PRE_UPSTREAM_CONCURRENCY`, `HEADROOM_ANTHROPIC_PRE_UPSTREAM_ACQUIRE_TIMEOUT_SECONDS`, `HEADROOM_REQUEST_TIMEOUT`, `HEADROOM_ANTHROPIC_PRE_UPSTREAM_MEMORY_CONTEXT_TIMEOUT_SECONDS`, `ANTHROPIC_TARGET_API_URL`, `OPENAI_TARGET_API_URL`, `GEMINI_TARGET_API_URL`, `ANTHROPIC_TARGET_API_HEADERS`, and `OPENAI_TARGET_API_HEADERS`. CLI flags take precedence over environment variables.
- The default Anthropic pre-upstream cap is intentionally conservative for CPU/ONNX-heavy work. Larger containers may want to raise it after checking the resolved runtime values on `/readyz` or `/debug/warmup`.
See also: [Proxy Server](proxy.md), [Configuration](configuration.md)

17
wiki/configuration.md
View file

@ -229,6 +229,23 @@ Some settings can be configured via environment variables:
| `HEADROOM_BETA_HEADER_STICKY` | Controls per-session `anthropic-beta` / `OpenAI-Beta` re-echo. `enabled` (default): the proxy unions beta tokens across turns within a session — if the client sends a token in turn N and omits it in turn N+1, the proxy re-injects it to preserve prefix-cache stability. `disabled`: the client's value is forwarded verbatim with no accumulation. Any other value raises at request time. See [Session Beta Header Tracking](#session-beta-header-tracking). | `enabled` |
| `HEADROOM_BETA_TRACKER_MAX_SESSIONS` | LRU capacity of the in-memory session beta tracker. Once full, the oldest session entry is evicted. | `1000` |
## Settings GUI
A web-based settings interface is available at `http://127.0.0.1:<port>/dashboard/settings` for configuring every safe `HEADROOM_*` proxy knob without hand-exporting environment variables, plus an **Endpoints** group for custom Anthropic/OpenAI upstream base URLs (`ANTHROPIC_TARGET_API_URL` / `OPENAI_TARGET_API_URL`) and extra headers merged into (and overriding) forwarded requests -- e.g. for a corporate gateway or Azure Foundry deployment that needs a different endpoint plus one extra auth header. Fields are split into a **Settings** tab (commonly-tuned: compression ratio, budget, rate limits, verbosity) and an **Advanced** tab (everything else, including Endpoints). Third-party credentials such as `OPENAI_API_KEY`/`AWS_*` are never exposed here; the two extra-headers fields are the only secret-typed fields in the panel and render masked once set, with a "Clear stored value" action to remove them -- resaving the page without touching a masked field never overwrites the real stored value.
- **Persistence**: Settings are saved to `~/.headroom/settings.json` (merged with existing values, not replaced) and loaded into the process environment at startup.
- **Precedence** (highest to lowest):
- Explicit shell export (`export HEADROOM_FOO=bar`)
- Settings from `~/.headroom/settings.json`
- Code default
- **Activation**: Click "Save" to persist without restarting, or "Apply & Restart" to persist and take effect immediately. Apply & Restart behavior depends on how the proxy is running:
- **Service** (supervised launchd/systemd install): self-restarts in one click.
- **Docker**: cannot self-restart from inside the container; the GUI surfaces the host-side `headroom install restart --profile <p>` command to run instead.
- **Task** (Windows Task Scheduler / cron-managed install): `headroom install` does not support lifecycle operations for task deployments; the GUI shows an instruction to restart via the OS task scheduler or by stopping the process so it relaunches on its next trigger.
- **Foreground** (plain `headroom proxy`): shows a manual-restart instruction.
- **Provenance / locking**: a field currently shadowed by an explicit environment variable export is rendered read-only with a tooltip, since editing it here would have no effect until the env var is unset. Manifest-baked settings (`HEADROOM_PORT`, `HEADROOM_HOST`) are similarly locked on supervised (Docker/Service) installs — managed by the install manifest, not the settings interface.
- **CSRF protection**: `/settings` and `/settings/apply` reject requests whose `Origin` header (when present) doesn't resolve to a loopback host, in addition to the existing loopback-only + Host-header DNS-rebinding guard shared by all admin endpoints.
## Session Beta Header Tracking
When running as a proxy, Headroom maintains a per-session union of `anthropic-beta` (and `OpenAI-Beta`) tokens via `SessionBetaTracker`. The session key is derived from the `x-headroom-session-id` header if present, otherwise from `md5(model + system_prompt[:500])[:16]` — stable across turns of the same conversation.

2
wiki/proxy.md
View file

@ -77,6 +77,8 @@ When configured, Headroom emits OTLP traces for the shared compression pipeline
| `--code-aware` / `--no-code-aware` | disabled | Enable or disable AST-based code compression. Requires `headroom-ai[code]` (env: HEADROOM_CODE_AWARE_ENABLED=1 to enable) |
| `--anthropic-api-url` | `https://api.anthropic.com` | Custom Anthropic API URL endpoint |
| `--openai-api-url` | `https://api.openai.com` | Custom OpenAI API URL endpoint |
| `--anthropic-extra-headers` | unset | JSON object of extra headers merged into (and overriding) forwarded Anthropic requests, e.g. `'{"Api-Key": "..."}'` |
| `--openai-extra-headers` | unset | JSON object of extra headers merged into (and overriding) forwarded OpenAI requests |
### Run Modes