Merge upstream/main into fix/copilot-oauth-runtime

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
JerrettDavis 2026-04-21 23:44:58 -05:00
commit 1d440023b6
34 changed files with 2556 additions and 41 deletions

View file

@ -0,0 +1,30 @@
{
"name": "headroom-marketplace",
"owner": {
"name": "Headroom Contributors"
},
"metadata": {
"description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.",
"version": "0.9.2"
},
"plugins": [
{
"name": "headroom",
"source": "./plugins/headroom-agent-hooks",
"description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
"version": "0.9.2",
"author": {
"name": "Headroom Contributors",
"url": "https://github.com/chopratejas/headroom"
},
"homepage": "https://github.com/chopratejas/headroom",
"repository": "https://github.com/chopratejas/headroom",
"keywords": [
"headroom",
"hooks",
"claude-code",
"copilot-cli"
]
}
]
}

View file

@ -1,6 +1,9 @@
# VCS
.git
.github
.github/*
!.github/plugin/
!.github/plugin/**
.gitignore
# Python artifacts
@ -38,6 +41,8 @@ plugins/
!plugins/
!plugins/openclaw/
!plugins/openclaw/**
!plugins/headroom-agent-hooks/
!plugins/headroom-agent-hooks/**
node_modules/
*.tgz
@ -61,3 +66,8 @@ docker-compose*.yml
.superpowers/
examples/
node-compile-cache/
!e2e/
!e2e/init/
!e2e/init/**
!.claude-plugin/
!.claude-plugin/**

42
.github/copilot-instructions.md vendored Normal file
View file

@ -0,0 +1,42 @@
<!-- headroom:rtk-instructions -->
# RTK (Rust Token Killer) - Token-Optimized Commands
When running shell commands, **always prefix with `rtk`**. This reduces context
usage by 60-90% with zero behavior change. If rtk has no filter for a command,
it passes through unchanged — so it is always safe to use.
## Key Commands
```bash
# Git (59-80% savings)
rtk git status rtk git diff rtk git log
# Files & Search (60-75% savings)
rtk ls <path> rtk read <file> rtk grep <pattern>
rtk find <pattern> rtk diff <file>
# Test (90-99% savings) — shows failures only
rtk pytest tests/ rtk cargo test rtk test <cmd>
# Build & Lint (80-90% savings) — shows errors only
rtk tsc rtk lint rtk cargo build
rtk prettier --check rtk mypy rtk ruff check
# Analysis (70-90% savings)
rtk err <cmd> rtk log <file> rtk json <file>
rtk summary <cmd> rtk deps rtk env
# GitHub (26-87% savings)
rtk gh pr view <n> rtk gh run list rtk gh issue list
# Infrastructure (85% savings)
rtk docker ps rtk kubectl get rtk docker logs <c>
# Package managers (70-90% savings)
rtk pip list rtk pnpm install rtk npm run <script>
```
## Rules
- In command chains, prefix each segment: `rtk git add . && rtk git commit -m "msg"`
- For debugging, use raw command without rtk prefix
- `rtk proxy <cmd>` runs command without filtering but tracks usage
<!-- /headroom:rtk-instructions -->

30
.github/plugin/marketplace.json vendored Normal file
View file

@ -0,0 +1,30 @@
{
"name": "headroom-marketplace",
"owner": {
"name": "Headroom Contributors"
},
"metadata": {
"description": "Headroom marketplace for Claude Code and GitHub Copilot CLI plugins.",
"version": "0.9.2"
},
"plugins": [
{
"name": "headroom",
"source": "./plugins/headroom-agent-hooks",
"description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
"version": "0.9.2",
"author": {
"name": "Headroom Contributors",
"url": "https://github.com/chopratejas/headroom"
},
"homepage": "https://github.com/chopratejas/headroom",
"repository": "https://github.com/chopratejas/headroom",
"keywords": [
"headroom",
"hooks",
"claude-code",
"copilot-cli"
]
}
]
}

View file

@ -49,12 +49,12 @@ jobs:
- name: Run tests
run: |
pytest -v --tb=short
pytest -v --tb=short tests scripts/tests
- name: Run tests with coverage
if: matrix.python-version == '3.11'
run: |
pytest --cov=headroom --cov-report=xml --cov-report=term-missing
pytest tests scripts/tests --cov=headroom --cov-report=xml --cov-report=term-missing
- name: Upload coverage to Codecov
if: matrix.python-version == '3.11'
@ -146,6 +146,11 @@ jobs:
docker build -f e2e/wrap/Dockerfile -t headroom-wrap-e2e .
docker run --rm headroom-wrap-e2e
- name: Run Docker-native init e2e
run: |
docker build -f e2e/init/Dockerfile -t headroom-init-e2e .
docker run --rm headroom-init-e2e
windows-native-wrapper:
runs-on: windows-latest
steps:
@ -175,11 +180,11 @@ jobs:
with:
python-version: "3.11"
- name: Install bash and test dependencies
run: |
brew install bash
python -m pip install --upgrade pip
pip install pytest
- name: Install bash and test dependencies
run: |
brew install bash
python -m pip install --upgrade pip
python -m pip install --retries 10 --timeout 60 pytest
- name: Run native installer wrapper tests
run: |
@ -215,10 +220,10 @@ jobs:
name: dist
path: dist/
commitlint:
if: github.event_name != 'push' || !startsWith(github.event.head_commit.message, 'Merge pull request ')
runs-on: ubuntu-latest
steps:
commitlint:
if: github.event_name != 'push' || !startsWith(github.event.head_commit.message, 'Merge pull request ')
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0

22
.github/workflows/init-e2e.yml vendored Normal file
View file

@ -0,0 +1,22 @@
name: Init E2E
on:
pull_request:
branches: [main]
push:
branches: [main]
workflow_dispatch:
jobs:
docker-init-e2e:
runs-on: ubuntu-latest
timeout-minutes: 45
steps:
- uses: actions/checkout@v4
- name: Build init e2e image
run: docker build -f e2e/init/Dockerfile -t headroom-init-e2e .
- name: Run init e2e container
run: docker run --rm headroom-init-e2e

View file

@ -395,15 +395,28 @@ jobs:
run: |
ls -la release-assets
- name: Create GitHub Release
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
TAG="v${{ needs.detect-version.outputs.version }}"
TITLE="Release v${{ needs.detect-version.outputs.version }}"
- name: Create or update GitHub Release
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
TAG="v${{ needs.detect-version.outputs.version }}"
TITLE="Release v${{ needs.detect-version.outputs.version }}"
if gh release view "$TAG" > /dev/null 2>&1; then
gh release edit "$TAG" --title "$TITLE" --notes-file .changelog.md
else
gh release create "$TAG" --title "$TITLE" --notes-file .changelog.md
fi
gh release upload "$TAG" release-assets/* --clobber
else
gh release create "$TAG" --title "$TITLE" --notes-file .changelog.md
fi
- name: Publish ${{ env.PYPI_PACKAGE }} Python distributions to GitHub Release
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
TAG="v${{ needs.detect-version.outputs.version }}"
gh release upload "$TAG" release-assets/*.whl release-assets/*.tar.gz --clobber
- name: Publish Node package tarballs to GitHub Release
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
TAG="v${{ needs.detect-version.outputs.version }}"
gh release upload "$TAG" release-assets/*.tgz --clobber

1
.gitignore vendored
View file

@ -5,6 +5,7 @@ scripts/*
!scripts/install.sh
!scripts/install.ps1
!scripts/version-sync.py
!scripts/sync-plugin-versions.py
!scripts/changelog-gen.py
!scripts/verify-versions.py
!scripts/tests/

View file

@ -1,4 +1,12 @@
repos:
- repo: local
hooks:
- id: sync-plugin-versions
name: Sync plugin versions
entry: python scripts/sync-plugin-versions.py
language: system
pass_filenames: false
always_run: true
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.9.4
hooks:

View file

@ -30,6 +30,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
savings, logs, and telemetry resolve to the bind-mounted `.headroom` path.
See [`wiki/filesystem-contract.md`](wiki/filesystem-contract.md).
### Changed
- **`/stats-history` now returns compact checkpoint history by default** — the
JSON response keeps recent checkpoints dense while evenly sampling older
checkpoints so long-running installs do not return ever-growing payloads.
Add `history_mode=full` to fetch the full retained checkpoint list, or
`history_mode=none` to skip it entirely while still receiving the derived
hourly/daily/weekly/monthly rollups. Responses now include a
`history_summary` block describing stored versus returned points.
## [0.5.22] - 2026-04-11
### Added

View file

@ -5,6 +5,7 @@
**Compress everything your AI agent reads. Same answers, fraction of the tokens.**
[![CI](https://github.com/chopratejas/headroom/actions/workflows/ci.yml/badge.svg)](https://github.com/chopratejas/headroom/actions/workflows/ci.yml)
[![codecov](https://codecov.io/gh/chopratejas/headroom/graph/badge.svg)](https://app.codecov.io/gh/chopratejas/headroom)
[![PyPI](https://img.shields.io/pypi/v/headroom-ai.svg)](https://pypi.org/project/headroom-ai/)
[![npm](https://img.shields.io/npm/v/headroom-ai.svg)](https://www.npmjs.com/package/headroom-ai)
[![Model: Kompress-base](https://img.shields.io/badge/model-Kompress--base-yellow.svg)](https://huggingface.co/chopratejas/kompress-base)
@ -42,6 +43,14 @@ headroom wrap aider # Aider
headroom wrap copilot # GitHub Copilot CLI
```
**Prefer a one-time durable install instead of wrapping every launch:**
```bash
headroom init -g # Detect installed user-scoped agents and wire them to Headroom
headroom init claude # Install repo-local Claude hooks for just this project
headroom init copilot -g # Install user-scoped Copilot hooks and provider routing
```
**Drop it into your own code — Python or TypeScript:**
```python
@ -146,14 +155,14 @@ python -m headroom.evals suite --tier 1
## Built for coding agents
| Agent | One-command wrap | Notes |
| Agent | Durable init / one-shot wrap | Notes |
|--------------------|------------------------------------|------------------------------------------------------------------|
| **Claude Code** | `headroom wrap claude` | `--memory` for cross-agent memory, `--code-graph` for codebase intel |
| **Codex** | `headroom wrap codex --memory` | Shares the same memory store as Claude |
| **Cursor** | `headroom wrap cursor` | Prints Cursor config — paste once, done |
| **Claude Code** | `headroom init claude -g` / `headroom wrap claude` | `init` installs user or repo-local hooks; `wrap` is still useful for ad hoc sessions |
| **Codex** | `headroom init codex -g` / `headroom wrap codex --memory` | `init` installs provider config plus lifecycle hooks where supported |
| **Cursor** | `headroom wrap cursor` | Prints Cursor config — durable init not available yet |
| **Aider** | `headroom wrap aider` | Starts proxy, launches Aider |
| **Copilot CLI** | `headroom wrap copilot` | Starts proxy, launches Copilot |
| **OpenClaw** | `headroom wrap openclaw` | Installs Headroom as ContextEngine plugin |
| **Copilot CLI** | `headroom init copilot -g` / `headroom wrap copilot` | `init` installs hooks and BYOK provider routing for the current user |
| **OpenClaw** | `headroom init openclaw -g` / `headroom wrap openclaw` | Installs Headroom as ContextEngine plugin |
MCP-native too — `headroom mcp install` exposes `headroom_compress`, `headroom_retrieve`, and `headroom_stats` to any MCP client.

View file

@ -18,7 +18,7 @@ The release workflow also calls `.github/workflows/docker.yml` as a reusable wor
| `headroom-openclaw` | TypeScript plugin | npmjs.org | `NPM_OPENCLAW_PACKAGE` |
| `@{owner}/headroom-ai` | TypeScript SDK | GitHub Package Registry | — |
| `@{owner}/headroom-openclaw` | TypeScript plugin | GitHub Package Registry | — |
| `headroom-ai-{version}.tar.gz` / `headroom_ai-{version}-py3-none-any.whl` | Python release assets | GitHub Release (`{owner}/headroom`) | — |
| `headroom-ai-{version}.tar.gz` / `headroom_ai-{version}-py3-none-any.whl` | Python package distributions | GitHub Release (`{owner}/headroom`) | — |
| `headroom-ai-{version}.tgz` / `headroom-openclaw-{version}.tgz` | Node release assets | GitHub Release (`{owner}/headroom`) | — |
| `ghcr.io/{owner}/headroom` | Docker image | GitHub Container Registry | — |
@ -102,7 +102,7 @@ Publishes both Node packages to GitHub Package Registry (`npm.pkg.github.com`) u
- `plugins/openclaw/` as `@{owner}/headroom-openclaw`
### GitHub release assets
Uploads the built Python distributions and both npm tarballs to the GitHub Release created in the current repository. This is what makes fork-owned main-branch builds downloadable for local validation even when consumers are not pulling from PyPI or npmjs.org.
Uploads the built Python distributions and both npm tarballs to the GitHub Release created in the current repository. GitHub Packages does not provide a PyPI-compatible package registry, so the workflow publishes Python wheels and sdists to GitHub as release assets while npm packages go to GitHub Package Registry and Docker images go to GHCR.
### publish-docker
Calls the reusable Docker workflow to publish GHCR images with the same semantic version and synced package metadata as the rest of the release.

33
e2e/init/Dockerfile Normal file
View file

@ -0,0 +1,33 @@
FROM node:22-bookworm
ENV DEBIAN_FRONTEND=noninteractive \
PATH="/opt/headroom-venv/bin:${PATH}" \
PIP_DISABLE_PIP_VERSION_CHECK=1 \
PIP_NO_CACHE_DIR=1 \
PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1
RUN apt-get update && \
apt-get install -y --no-install-recommends \
ca-certificates \
git \
python3 \
python3-pip \
python3-venv && \
ln -sf /usr/bin/python3 /usr/local/bin/python && \
rm -rf /var/lib/apt/lists/*
WORKDIR /workspace
COPY pyproject.toml README.md uv.lock ./
COPY headroom ./headroom
COPY .claude-plugin ./.claude-plugin
COPY .github/plugin ./.github/plugin
COPY plugins/headroom-agent-hooks ./plugins/headroom-agent-hooks
COPY e2e/init ./e2e/init
RUN python -m venv /opt/headroom-venv && \
/opt/headroom-venv/bin/python -m pip install --upgrade pip && \
/opt/headroom-venv/bin/python -m pip install -e ".[proxy]"
CMD ["python", "e2e/init/run.py"]

236
e2e/init/run.py Normal file
View file

@ -0,0 +1,236 @@
from __future__ import annotations
import json
import os
import stat
import subprocess
import sys
import tempfile
import textwrap
from pathlib import Path
from headroom.cli import init as init_cli
REPO_ROOT = Path("/workspace")
HEADROOM = "headroom"
def log(message: str) -> None:
print(f"[init-e2e] {message}", flush=True)
def run(
cmd: list[str],
*,
env: dict[str, str],
cwd: Path,
timeout: int = 180,
) -> subprocess.CompletedProcess[str]:
log(f"$ {' '.join(cmd)}")
result = subprocess.run(
cmd,
env=env,
cwd=str(cwd),
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=timeout,
)
if result.stdout.strip():
print(result.stdout.rstrip(), flush=True)
if result.stderr.strip():
print(result.stderr.rstrip(), file=sys.stderr, flush=True)
if result.returncode != 0:
raise RuntimeError(f"Command failed with exit code {result.returncode}: {' '.join(cmd)}")
return result
def assert_true(condition: bool, message: str) -> None:
if not condition:
raise AssertionError(message)
def write_executable(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
def read_jsonl(path: Path) -> list[dict[str, object]]:
if not path.exists():
return []
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]
def create_agent_shims(shim_dir: Path, log_path: Path) -> None:
shim = textwrap.dedent(
"""\
#!/usr/bin/env python3
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
record = {
"tool": Path(sys.argv[0]).name,
"argv": sys.argv[1:],
"cwd": os.getcwd(),
}
log_path = Path(os.environ["HEADROOM_INIT_E2E_LOG"])
log_path.parent.mkdir(parents=True, exist_ok=True)
with log_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(record) + "\\n")
print(f"{record['tool']} shim executed")
raise SystemExit(0)
"""
)
shim_dir.mkdir(parents=True, exist_ok=True)
for name in ("claude", "copilot"):
write_executable(shim_dir / name, shim)
def expect_hook_command(command: str, profile: str) -> None:
assert_true("init hook ensure" in command, f"missing init hook ensure in: {command}")
assert_true(f"--profile {profile}" in command, f"missing profile {profile} in: {command}")
def read_manifest(home_dir: Path, profile: str) -> dict[str, object]:
path = home_dir / ".headroom" / "deploy" / profile / "manifest.json"
assert_true(path.exists(), f"Expected manifest at {path}")
return json.loads(path.read_text(encoding="utf-8"))
def verify_claude_local(home_dir: Path, project_dir: Path, shim_log: Path) -> None:
settings = json.loads(
(project_dir / ".claude" / "settings.local.json").read_text(encoding="utf-8")
)
assert_true(
settings["env"]["ANTHROPIC_BASE_URL"] == "http://127.0.0.1:9011",
"Claude local settings should point at the requested proxy port",
)
session_start = settings["hooks"]["SessionStart"][0]["hooks"][0]["command"]
pre_tool = settings["hooks"]["PreToolUse"][0]["hooks"][0]["command"]
profile = init_cli._local_profile(project_dir)
expect_hook_command(session_start, profile)
expect_hook_command(pre_tool, profile)
manifest = read_manifest(home_dir, profile)
assert_true("claude" in manifest["targets"], "Claude init should register the claude target")
claude_calls = [record["argv"] for record in read_jsonl(shim_log) if record["tool"] == "claude"]
assert_true(
claude_calls
== [
["plugin", "marketplace", "add", str(REPO_ROOT)],
["plugin", "install", "headroom@headroom-marketplace", "--scope", "local"],
],
f"Unexpected Claude install commands: {claude_calls}",
)
def verify_copilot_global(home_dir: Path, shim_log: Path) -> None:
config = json.loads((home_dir / ".copilot" / "config.json").read_text(encoding="utf-8"))
assert_true(
"SessionStart" in config["hooks"], "Copilot config should include SessionStart hooks"
)
assert_true("PreToolUse" in config["hooks"], "Copilot config should include PreToolUse hooks")
session_start = config["hooks"]["SessionStart"][0]["command"]
expect_hook_command(session_start, "init-user")
for shell_file in (home_dir / ".bashrc", home_dir / ".zshrc", home_dir / ".profile"):
content = shell_file.read_text(encoding="utf-8")
assert_true(
'export COPILOT_PROVIDER_TYPE="openai"' in content,
f"{shell_file.name} should contain the Copilot provider type",
)
assert_true(
'export COPILOT_PROVIDER_BASE_URL="http://127.0.0.1:9005/v1"' in content,
f"{shell_file.name} should contain the Copilot provider base URL",
)
assert_true(
'export COPILOT_PROVIDER_WIRE_API="completions"' in content,
f"{shell_file.name} should contain the Copilot wire API",
)
copilot_calls = [
record["argv"] for record in read_jsonl(shim_log) if record["tool"] == "copilot"
]
assert_true(
copilot_calls
== [
["plugin", "marketplace", "add", str(REPO_ROOT)],
["plugin", "install", "headroom@headroom-marketplace"],
],
f"Unexpected Copilot install commands: {copilot_calls}",
)
def verify_codex_local(home_dir: Path, project_dir: Path) -> None:
config_path = project_dir / ".codex" / "config.toml"
hooks_path = project_dir / ".codex" / "hooks.json"
config = config_path.read_text(encoding="utf-8")
hooks = json.loads(hooks_path.read_text(encoding="utf-8"))
profile = init_cli._local_profile(project_dir)
assert_true(
'base_url = "http://127.0.0.1:9012/v1"' in config,
"Codex config should point at the requested proxy port",
)
assert_true(
config.count("[features]") == 1, "Codex config should keep a single [features] table"
)
assert_true("codex_hooks = true" in config, "Codex config should enable codex_hooks")
command = hooks["hooks"]["SessionStart"][0]["hooks"][0]["command"]
expect_hook_command(command, profile)
manifest = read_manifest(home_dir, profile)
targets = manifest["targets"]
assert_true(set(targets) == {"claude", "codex"}, f"Unexpected merged targets: {targets}")
def main() -> None:
with tempfile.TemporaryDirectory(prefix="headroom-init-e2e-") as temp_root_raw:
temp_root = Path(temp_root_raw)
home_dir = temp_root / "home"
project_dir = temp_root / "project"
shim_dir = temp_root / "bin"
shim_log = temp_root / "shim-log.jsonl"
home_dir.mkdir(parents=True)
project_dir.mkdir(parents=True)
create_agent_shims(shim_dir, shim_log)
env = os.environ.copy()
env["HOME"] = str(home_dir)
env["USERPROFILE"] = str(home_dir)
env["HEADROOM_INIT_E2E_LOG"] = str(shim_log)
env["PATH"] = f"{shim_dir}:{env['PATH']}"
run([HEADROOM, "init", "--port", "9011", "claude"], env=env, cwd=project_dir)
verify_claude_local(home_dir, project_dir, shim_log)
run(
[
HEADROOM,
"init",
"-g",
"--port",
"9005",
"--backend",
"openai",
"copilot",
],
env=env,
cwd=project_dir,
)
verify_copilot_global(home_dir, shim_log)
run([HEADROOM, "init", "--port", "9012", "codex"], env=env, cwd=project_dir)
verify_codex_local(home_dir, project_dir)
log("Init e2e completed successfully")
if __name__ == "__main__":
main()

679
headroom/cli/init.py Normal file
View file

@ -0,0 +1,679 @@
"""Durable agent initialization commands."""
from __future__ import annotations
import json
import os
import shlex
import shutil
import subprocess
from hashlib import sha1
from pathlib import Path
from typing import Any
import click
from headroom.install.models import ConfigScope, InstallPreset, RuntimeKind, SupervisorKind
from headroom.install.paths import claude_settings_path, codex_config_path, validate_profile_name
from headroom.install.planner import build_manifest
from headroom.install.providers import _apply_unix_env_scope, _apply_windows_env_scope
from headroom.install.runtime import (
resolve_headroom_command,
start_detached_agent,
start_persistent_docker,
stop_runtime,
wait_ready,
)
from headroom.install.state import load_manifest, save_manifest
from headroom.install.supervisors import start_supervisor
from .main import main
_GLOBAL_PROFILE = "init-user"
_CLAUDE_HOOK_MARKER = "headroom-init-claude"
_COPILOT_HOOK_MARKER = "headroom-init-copilot"
_CODEX_HOOK_MARKER = "headroom-init-codex"
_CODEX_PROVIDER_MARKER_START = "# --- Headroom init provider ---"
_CODEX_PROVIDER_MARKER_END = "# --- end Headroom init provider ---"
_CODEX_FEATURE_MARKER_START = "# --- Headroom init features ---"
_CODEX_FEATURE_MARKER_END = "# --- end Headroom init features ---"
_SUPPORTED_TARGETS = ("claude", "copilot", "codex", "openclaw")
_LOCAL_TARGETS = {"claude", "codex"}
_GLOBAL_TARGETS = {"claude", "copilot", "codex", "openclaw"}
def _command_string(parts: list[str]) -> str:
if os.name == "nt":
return subprocess.list2cmdline(parts)
return shlex.join(parts)
def _hook_command(*parts: str) -> str:
return _command_string([*resolve_headroom_command(), "init", "hook", "ensure", *parts])
def _powershell_matcher() -> str:
return "Bash|PowerShell" if os.name == "nt" else "Bash"
def _local_profile(cwd: Path | None = None) -> str:
root = (cwd or Path.cwd()).resolve()
slug = "".join(ch if ch.isalnum() or ch in "-._" else "-" for ch in root.name.lower()).strip(
"-"
)
digest = sha1(str(root).encode("utf-8")).hexdigest()[:8]
return validate_profile_name(f"init-{slug or 'repo'}-{digest}")
def _runtime_profile(global_scope: bool, cwd: Path | None = None) -> str:
return _GLOBAL_PROFILE if global_scope else _local_profile(cwd)
def _copilot_config_path() -> Path:
return Path.home() / ".copilot" / "config.json"
def _codex_hooks_path(global_scope: bool) -> Path:
return (Path.home() if global_scope else Path.cwd()) / ".codex" / "hooks.json"
def _claude_scope_path(global_scope: bool) -> Path:
if global_scope:
return claude_settings_path()
return Path.cwd() / ".claude" / "settings.local.json"
def _codex_scope_path(global_scope: bool) -> Path:
if global_scope:
return codex_config_path()
return Path.cwd() / ".codex" / "config.toml"
def _json_file(path: Path) -> dict[str, Any]:
if not path.exists():
return {}
content = path.read_text(encoding="utf-8").strip()
if not content:
return {}
payload = json.loads(content)
return payload if isinstance(payload, dict) else {}
def _write_json(path: Path, payload: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
def _ensure_claude_hooks(path: Path, profile: str, port: int) -> None:
payload = _json_file(path)
env_map = dict(payload.get("env") or {}) if isinstance(payload.get("env"), dict) else {}
env_map["ANTHROPIC_BASE_URL"] = f"http://127.0.0.1:{port}"
payload["env"] = env_map
hooks = dict(payload.get("hooks") or {}) if isinstance(payload.get("hooks"), dict) else {}
command = _hook_command("--profile", profile)
for event, matcher in (
("SessionStart", "startup|resume"),
("PreToolUse", _powershell_matcher()),
):
entries = list(hooks.get(event) or []) if isinstance(hooks.get(event), list) else []
retained: list[dict[str, Any]] = []
for entry in entries:
if not isinstance(entry, dict):
retained.append(entry)
continue
hook_items = entry.get("hooks")
if not isinstance(hook_items, list):
retained.append(entry)
continue
has_headroom = any(
isinstance(item, dict)
and item.get("command")
and _CLAUDE_HOOK_MARKER in str(item.get("command"))
for item in hook_items
)
if not has_headroom:
retained.append(entry)
retained.append(
{
"matcher": matcher,
"hooks": [
{
"type": "command",
"command": f"{command} --marker {_CLAUDE_HOOK_MARKER}",
"timeout": 15,
}
],
}
)
hooks[event] = retained
payload["hooks"] = hooks
_write_json(path, payload)
def _ensure_copilot_hooks(path: Path, profile: str) -> None:
payload = _json_file(path)
hooks = dict(payload.get("hooks") or {}) if isinstance(payload.get("hooks"), dict) else {}
command = f"{_hook_command('--profile', profile)} --marker {_COPILOT_HOOK_MARKER}"
for event in ("SessionStart", "PreToolUse"):
entries = list(hooks.get(event) or []) if isinstance(hooks.get(event), list) else []
retained = [
entry
for entry in entries
if not (
isinstance(entry, dict) and _COPILOT_HOOK_MARKER in str(entry.get("command", ""))
)
]
retained.append({"type": "command", "command": command, "cwd": ".", "timeout": 15})
hooks[event] = retained
payload["hooks"] = hooks
_write_json(path, payload)
def _replace_marker_block(content: str, marker_start: str, marker_end: str, block: str) -> str:
if marker_start in content and marker_end in content:
start = content.index(marker_start)
end = content.index(marker_end) + len(marker_end)
content = content[:start].rstrip() + "\n\n" + content[end:].lstrip()
return (content.rstrip() + "\n\n" + block.strip() + "\n").lstrip()
def _ensure_codex_provider(path: Path, port: int) -> None:
block = (
f"{_CODEX_PROVIDER_MARKER_START}\n"
'model_provider = "headroom"\n\n'
"[model_providers.headroom]\n"
'name = "Headroom init proxy"\n'
f'base_url = "http://127.0.0.1:{port}/v1"\n'
'env_key = "OPENAI_API_KEY"\n'
"requires_openai_auth = true\n"
"supports_websockets = true\n"
f"{_CODEX_PROVIDER_MARKER_END}"
)
content = path.read_text(encoding="utf-8") if path.exists() else ""
content = _replace_marker_block(
content, _CODEX_PROVIDER_MARKER_START, _CODEX_PROVIDER_MARKER_END, block
)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
def _ensure_codex_feature_flag(path: Path) -> None:
content = path.read_text(encoding="utf-8") if path.exists() else ""
if _CODEX_FEATURE_MARKER_START in content and _CODEX_FEATURE_MARKER_END in content:
block = f"{_CODEX_FEATURE_MARKER_START}\ncodex_hooks = true\n{_CODEX_FEATURE_MARKER_END}"
content = _replace_marker_block(
content,
_CODEX_FEATURE_MARKER_START,
_CODEX_FEATURE_MARKER_END,
block,
)
elif "[features]" in content:
lines = content.splitlines()
inserted = False
for index, line in enumerate(lines):
if line.strip() != "[features]":
continue
section_end = index + 1
while section_end < len(lines) and not (
lines[section_end].startswith("[") and lines[section_end].endswith("]")
):
if "codex_hooks" in lines[section_end]:
inserted = True
break
section_end += 1
if not inserted:
lines[index + 1 : index + 1] = [
_CODEX_FEATURE_MARKER_START,
"codex_hooks = true",
_CODEX_FEATURE_MARKER_END,
]
inserted = True
break
content = "\n".join(lines).rstrip() + "\n"
if not inserted:
content = (
content.rstrip()
+ "\n\n[features]\n"
+ _CODEX_FEATURE_MARKER_START
+ "\n"
+ "codex_hooks = true\n"
+ _CODEX_FEATURE_MARKER_END
+ "\n"
)
else:
content = (
content.rstrip()
+ "\n\n[features]\n"
+ _CODEX_FEATURE_MARKER_START
+ "\n"
+ "codex_hooks = true\n"
+ _CODEX_FEATURE_MARKER_END
+ "\n"
).lstrip()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
def _ensure_codex_hooks(path: Path, profile: str) -> None:
command = f"{_hook_command('--profile', profile)} --marker {_CODEX_HOOK_MARKER}"
payload = {
"hooks": {
"SessionStart": [
{
"matcher": "startup|resume",
"hooks": [{"type": "command", "command": command, "timeout": 15}],
}
],
"PreToolUse": [
{
"matcher": "Bash",
"hooks": [{"type": "command", "command": command, "timeout": 15}],
}
],
}
}
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
def _manifest_changed(
existing: Any,
*,
port: int,
backend: str,
anyllm_provider: str | None,
region: str | None,
memory: bool,
) -> bool:
return any(
[
getattr(existing, "port", port) != port,
getattr(existing, "backend", backend) != backend,
getattr(existing, "anyllm_provider", anyllm_provider) != anyllm_provider,
getattr(existing, "region", region) != region,
getattr(existing, "memory_enabled", memory) != memory,
]
)
def _ensure_runtime_manifest(
*,
global_scope: bool,
targets: list[str],
port: int,
backend: str,
anyllm_provider: str | None,
region: str | None,
memory: bool,
) -> str:
profile = _runtime_profile(global_scope)
existing = load_manifest(profile)
merged_targets = sorted(set(existing.targets if existing else []).union(targets))
manifest = build_manifest(
profile=profile,
preset=InstallPreset.PERSISTENT_TASK.value,
runtime_kind=RuntimeKind.PYTHON.value,
scope=ConfigScope.USER.value,
provider_mode="manual",
targets=merged_targets,
port=port,
backend=backend,
anyllm_provider=anyllm_provider,
region=region,
proxy_mode="token",
memory_enabled=memory,
telemetry_enabled=True,
image="ghcr.io/chopratejas/headroom:latest",
)
manifest.supervisor_kind = SupervisorKind.NONE.value
manifest.artifacts = []
manifest.mutations = existing.mutations if existing else []
if existing is not None and _manifest_changed(
existing,
port=port,
backend=backend,
anyllm_provider=anyllm_provider,
region=region,
memory=memory,
):
try:
stop_runtime(existing)
except Exception:
pass
save_manifest(manifest)
return profile
def _env_manifest(values: dict[str, str]) -> Any:
return build_manifest(
profile="init-env",
preset=InstallPreset.PERSISTENT_TASK.value,
runtime_kind=RuntimeKind.PYTHON.value,
scope=ConfigScope.USER.value,
provider_mode="manual",
targets=["copilot"],
port=8787,
backend="anthropic",
anyllm_provider=None,
region=None,
proxy_mode="token",
memory_enabled=False,
telemetry_enabled=True,
image="ghcr.io/chopratejas/headroom:latest",
)
def _apply_user_env(values: dict[str, str]) -> None:
manifest = _env_manifest(values)
manifest.base_env = {}
manifest.tool_envs = {"copilot": values}
if os.name == "nt":
_apply_windows_env_scope(manifest)
else:
_apply_unix_env_scope(manifest)
def _resolve_copilot_env(port: int, backend: str) -> dict[str, str]:
if backend == "anthropic":
return {
"COPILOT_PROVIDER_TYPE": "anthropic",
"COPILOT_PROVIDER_BASE_URL": f"http://127.0.0.1:{port}",
}
return {
"COPILOT_PROVIDER_TYPE": "openai",
"COPILOT_PROVIDER_BASE_URL": f"http://127.0.0.1:{port}/v1",
"COPILOT_PROVIDER_WIRE_API": "completions",
}
def _marketplace_source() -> str:
override = os.environ.get("HEADROOM_MARKETPLACE_SOURCE")
if override:
return override
repo_root = Path(__file__).resolve().parents[2]
if (repo_root / ".claude-plugin" / "marketplace.json").exists():
return str(repo_root)
return "chopratejas/headroom"
def _run_checked(command: list[str], *, action: str) -> None:
result = subprocess.run(
command,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
)
if result.returncode == 0:
return
detail = "\n".join(part for part in (result.stderr.strip(), result.stdout.strip()) if part)
if "already" in detail.lower() or "exists" in detail.lower():
return
raise click.ClickException(f"{action} failed: {detail or result.returncode}")
def _install_claude_marketplace(scope: str) -> None:
claude_bin = shutil.which("claude")
if not claude_bin:
raise click.ClickException("'claude' not found in PATH. Install Claude Code first.")
source = _marketplace_source()
_run_checked(
[claude_bin, "plugin", "marketplace", "add", source], action="claude marketplace add"
)
_run_checked(
[claude_bin, "plugin", "install", "headroom@headroom-marketplace", "--scope", scope],
action="claude plugin install",
)
def _install_copilot_marketplace() -> None:
copilot_bin = shutil.which("copilot")
if not copilot_bin:
raise click.ClickException("'copilot' not found in PATH. Install GitHub Copilot CLI first.")
source = _marketplace_source()
_run_checked(
[copilot_bin, "plugin", "marketplace", "add", source],
action="copilot marketplace add",
)
_run_checked(
[copilot_bin, "plugin", "install", "headroom@headroom-marketplace"],
action="copilot plugin install",
)
def _ensure_profile_running(profile: str) -> None:
manifest = load_manifest(profile)
if manifest is None:
return
if wait_ready(manifest, timeout_seconds=1):
return
try:
if manifest.preset == InstallPreset.PERSISTENT_DOCKER.value:
start_persistent_docker(manifest)
elif manifest.supervisor_kind == SupervisorKind.SERVICE.value:
start_supervisor(manifest)
else:
start_detached_agent(manifest.profile)
wait_ready(manifest, timeout_seconds=45)
except Exception:
return
def detect_init_targets(global_scope: bool) -> list[str]:
allowed = _GLOBAL_TARGETS if global_scope else _LOCAL_TARGETS
detected: list[str] = []
for target in _SUPPORTED_TARGETS:
if target not in allowed:
continue
if shutil.which(target):
detected.append(target)
return detected
def _init_claude(*, global_scope: bool, profile: str, port: int) -> None:
_ensure_claude_hooks(_claude_scope_path(global_scope), profile, port)
_install_claude_marketplace("user" if global_scope else "local")
click.echo(f"Configured Claude Code ({'user' if global_scope else 'local'} scope).")
click.echo("Restart Claude Code to activate Headroom hooks and provider routing.")
def _init_copilot(*, global_scope: bool, profile: str, port: int, backend: str) -> None:
if not global_scope:
raise click.ClickException(
"Copilot durable init currently requires -g (current-user scope)."
)
_ensure_copilot_hooks(_copilot_config_path(), profile)
_apply_user_env(_resolve_copilot_env(port, backend))
_install_copilot_marketplace()
click.echo("Configured GitHub Copilot CLI (user scope).")
click.echo("Restart Copilot CLI to activate Headroom hooks and provider routing.")
def _init_codex(*, global_scope: bool, profile: str, port: int) -> None:
config_path = _codex_scope_path(global_scope)
_ensure_codex_provider(config_path, port)
_ensure_codex_feature_flag(config_path)
_ensure_codex_hooks(_codex_hooks_path(global_scope), profile)
click.echo(f"Configured Codex ({'user' if global_scope else 'local'} scope).")
if os.name == "nt":
click.echo(
"Codex hooks are currently disabled upstream on Windows; provider routing was still installed."
)
click.echo("Restart Codex to activate Headroom configuration.")
def _init_openclaw(*, global_scope: bool, port: int) -> None:
if not global_scope:
raise click.ClickException(
"OpenClaw durable init currently requires -g (current-user scope)."
)
command = [*resolve_headroom_command(), "wrap", "openclaw", "--proxy-port", str(port)]
result = subprocess.run(command)
if result.returncode != 0:
raise SystemExit(result.returncode)
def _run_init_targets(
*,
targets: list[str],
global_scope: bool,
port: int,
backend: str,
anyllm_provider: str | None,
region: str | None,
memory: bool,
) -> None:
runtime_targets = [target for target in targets if target != "openclaw"]
profile = _ensure_runtime_manifest(
global_scope=global_scope,
targets=runtime_targets,
port=port,
backend=backend,
anyllm_provider=anyllm_provider,
region=region,
memory=memory,
)
for target in targets:
if target == "claude":
_init_claude(global_scope=global_scope, profile=profile, port=port)
elif target == "copilot":
_init_copilot(global_scope=global_scope, profile=profile, port=port, backend=backend)
elif target == "codex":
_init_codex(global_scope=global_scope, profile=profile, port=port)
elif target == "openclaw":
_init_openclaw(global_scope=global_scope, port=port)
@main.group(invoke_without_command=True)
@click.option("-g", "--global", "global_scope", is_flag=True, help="Install for the current user.")
@click.option("--port", default=8787, type=int, show_default=True, help="Headroom proxy port.")
@click.option("--backend", default="anthropic", show_default=True, help="Proxy backend.")
@click.option("--anyllm-provider", default=None, help="Provider for any-llm backends.")
@click.option("--region", default=None, help="Cloud region for Bedrock / Vertex style backends.")
@click.option("--memory", is_flag=True, help="Enable persistent memory in the proxy runtime.")
@click.pass_context
def init(
ctx: click.Context,
global_scope: bool,
port: int,
backend: str,
anyllm_provider: str | None,
region: str | None,
memory: bool,
) -> None:
"""Install durable Headroom integrations for supported agents."""
if ctx.invoked_subcommand is not None:
ctx.obj = {
"global_scope": global_scope,
"port": port,
"backend": backend,
"anyllm_provider": anyllm_provider,
"region": region,
"memory": memory,
}
return
targets = detect_init_targets(global_scope)
if not targets:
scope_label = "user" if global_scope else "local"
raise click.ClickException(
f"No supported {scope_label} init targets were auto-detected. Specify one explicitly."
)
_run_init_targets(
targets=targets,
global_scope=global_scope,
port=port,
backend=backend,
anyllm_provider=anyllm_provider,
region=region,
memory=memory,
)
def _ctx_value(ctx: click.Context, key: str) -> Any:
return (ctx.obj or {}).get(key)
@init.command("claude")
@click.pass_context
def init_claude(ctx: click.Context) -> None:
"""Install Claude Code durable hooks and provider routing."""
_run_init_targets(
targets=["claude"],
global_scope=bool(_ctx_value(ctx, "global_scope")),
port=int(_ctx_value(ctx, "port") or 8787),
backend=str(_ctx_value(ctx, "backend") or "anthropic"),
anyllm_provider=_ctx_value(ctx, "anyllm_provider"),
region=_ctx_value(ctx, "region"),
memory=bool(_ctx_value(ctx, "memory")),
)
@init.command("copilot")
@click.pass_context
def init_copilot(ctx: click.Context) -> None:
"""Install GitHub Copilot CLI durable hooks and provider routing."""
_run_init_targets(
targets=["copilot"],
global_scope=bool(_ctx_value(ctx, "global_scope")),
port=int(_ctx_value(ctx, "port") or 8787),
backend=str(_ctx_value(ctx, "backend") or "anthropic"),
anyllm_provider=_ctx_value(ctx, "anyllm_provider"),
region=_ctx_value(ctx, "region"),
memory=bool(_ctx_value(ctx, "memory")),
)
@init.command("codex")
@click.pass_context
def init_codex(ctx: click.Context) -> None:
"""Install Codex durable hooks and provider routing."""
_run_init_targets(
targets=["codex"],
global_scope=bool(_ctx_value(ctx, "global_scope")),
port=int(_ctx_value(ctx, "port") or 8787),
backend=str(_ctx_value(ctx, "backend") or "anthropic"),
anyllm_provider=_ctx_value(ctx, "anyllm_provider"),
region=_ctx_value(ctx, "region"),
memory=bool(_ctx_value(ctx, "memory")),
)
@init.command("openclaw")
@click.pass_context
def init_openclaw(ctx: click.Context) -> None:
"""Install the durable OpenClaw Headroom plugin."""
_run_init_targets(
targets=["openclaw"],
global_scope=bool(_ctx_value(ctx, "global_scope")),
port=int(_ctx_value(ctx, "port") or 8787),
backend=str(_ctx_value(ctx, "backend") or "anthropic"),
anyllm_provider=_ctx_value(ctx, "anyllm_provider"),
region=_ctx_value(ctx, "region"),
memory=bool(_ctx_value(ctx, "memory")),
)
@init.group("hook", hidden=True)
def init_hook() -> None:
"""Internal hook helpers."""
@init_hook.command("ensure")
@click.option("--profile", default=None, help="Explicit deployment profile to ensure.")
@click.option("--marker", default=None, hidden=True)
def init_hook_ensure(profile: str | None, marker: str | None) -> None:
"""Best-effort ensure used by installed agent hooks."""
del marker
profiles: list[str] = []
if profile:
profiles.append(profile)
else:
local_profile = _local_profile()
if load_manifest(local_profile) is not None:
profiles.append(local_profile)
elif load_manifest(_GLOBAL_PROFILE) is not None:
profiles.append(_GLOBAL_PROFILE)
for name in profiles:
_ensure_profile_running(name)

View file

@ -37,6 +37,7 @@ def _register_commands() -> None:
"""Register all subcommand groups."""
from . import (
evals, # noqa: F401
init, # noqa: F401
install, # noqa: F401
learn, # noqa: F401
mcp, # noqa: F401

View file

@ -1616,6 +1616,9 @@
async downloadHistory(format = 'json', series = null) {
const selectedSeries = series || this.historySelectedSeriesKey;
const params = new URLSearchParams({ format, series: selectedSeries });
if (format === 'json' && selectedSeries === 'history') {
params.set('history_mode', 'full');
}
const response = await fetch('/stats-history?' + params.toString());
if (!response.ok) throw new Error('Failed to export history');

View file

@ -29,6 +29,7 @@ DEFAULT_SAVINGS_FILE = "proxy_savings.json"
SCHEMA_VERSION = 2
DEFAULT_MAX_HISTORY_POINTS = 5000
DEFAULT_MAX_HISTORY_AGE_DAYS = 365
DEFAULT_MAX_RESPONSE_HISTORY_POINTS = 500
DEFAULT_DISPLAY_SESSION_INACTIVITY_MINUTES = 60
LITELLM_AVAILABLE = importlib.util.find_spec("litellm") is not None
@ -311,11 +312,19 @@ class SavingsTracker:
path: str | None = None,
max_history_points: int = DEFAULT_MAX_HISTORY_POINTS,
max_history_age_days: int = DEFAULT_MAX_HISTORY_AGE_DAYS,
max_response_history_points: int = DEFAULT_MAX_RESPONSE_HISTORY_POINTS,
display_session_inactivity_minutes: int = (DEFAULT_DISPLAY_SESSION_INACTIVITY_MINUTES),
) -> None:
self._path = Path(path or get_default_savings_storage_path())
self._max_history_points = max_history_points
self._max_history_age_days = max_history_age_days
self._max_response_history_points = max(
_coerce_int(
max_response_history_points,
DEFAULT_MAX_RESPONSE_HISTORY_POINTS,
),
1,
)
self._display_session_inactivity_minutes = max(
_coerce_int(
display_session_inactivity_minutes,
@ -524,16 +533,17 @@ class SavingsTracker:
"retention": snapshot["retention"],
}
def history_response(self) -> dict[str, Any]:
def history_response(self, history_mode: str = "compact") -> dict[str, Any]:
"""Return frontend-friendly historical data for `/stats-history`."""
snapshot = self.snapshot()
history = snapshot["history"]
raw_history = snapshot["history"]
series = {
"hourly": self._build_rollup(history, bucket="hour"),
"daily": self._build_rollup(history, bucket="day"),
"weekly": self._build_rollup(history, bucket="week"),
"monthly": self._build_rollup(history, bucket="month"),
"hourly": self._build_rollup(raw_history, bucket="hour"),
"daily": self._build_rollup(raw_history, bucket="day"),
"weekly": self._build_rollup(raw_history, bucket="week"),
"monthly": self._build_rollup(raw_history, bucket="month"),
}
history = self._history_for_response(raw_history, mode=history_mode)
return {
"schema_version": snapshot["schema_version"],
"generated_at": _to_utc_iso(_utc_now()),
@ -549,6 +559,12 @@ class SavingsTracker:
"available_series": ["history", *series.keys()],
},
"retention": snapshot["retention"],
"history_summary": {
"mode": history_mode,
"stored_points": len(raw_history),
"returned_points": len(history),
"compacted": len(history) < len(raw_history),
},
}
def export_rows(self, series: str = "history") -> list[dict[str, Any]]:
@ -604,6 +620,7 @@ class SavingsTracker:
"retention": {
"max_history_points": self._max_history_points,
"max_history_age_days": self._max_history_age_days,
"max_response_history_points": self._max_response_history_points,
},
}
@ -727,6 +744,53 @@ class SavingsTracker:
self._state["history"] = history
def _history_for_response(
self,
history: list[dict[str, Any]],
*,
mode: str,
) -> list[dict[str, Any]]:
if mode == "none":
return []
if mode == "full":
return [dict(item) for item in history]
return self._compact_history(history)
def _compact_history(self, history: list[dict[str, Any]]) -> list[dict[str, Any]]:
if len(history) <= self._max_response_history_points:
return [dict(item) for item in history]
# Keep the recent tail dense for charts while evenly sampling older
# checkpoints so long-running installs don't return unbounded payloads.
recent_points = min(
max(self._max_response_history_points // 3, 50),
self._max_response_history_points - 1,
)
recent = history[-recent_points:]
older = history[:-recent_points]
older_slots = self._max_response_history_points - len(recent)
if older_slots <= 0 or not older:
return [dict(item) for item in recent[-self._max_response_history_points :]]
if older_slots == 1:
sampled_older = [older[0]]
else:
sampled_older = [
older[((len(older) - 1) * index) // (older_slots - 1)]
for index in range(older_slots)
]
compacted: list[dict[str, Any]] = []
seen_timestamps: set[str] = set()
for point in [*sampled_older, *recent]:
timestamp = point.get("timestamp")
if not isinstance(timestamp, str) or timestamp in seen_timestamps:
continue
seen_timestamps.add(timestamp)
compacted.append(dict(point))
return compacted
def _save_locked(self) -> None:
try:
self._path.parent.mkdir(parents=True, exist_ok=True)

View file

@ -1668,6 +1668,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
async def stats_history(
format: Literal["json", "csv"] = "json",
series: Literal["history", "hourly", "daily", "weekly", "monthly"] = "history",
history_mode: Literal["compact", "full", "none"] = "compact",
):
"""Get durable proxy compression history plus display-session state."""
if format == "csv":
@ -1678,7 +1679,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
return proxy.metrics.savings_tracker.history_response()
return proxy.metrics.savings_tracker.history_response(history_mode=history_mode)
@app.get("/transformations/feed")
async def transformations_feed(limit: int = 20):

View file

@ -0,0 +1,17 @@
{
"name": "headroom",
"version": "0.9.2",
"description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
"author": {
"name": "Headroom Contributors",
"url": "https://github.com/chopratejas/headroom"
},
"homepage": "https://github.com/chopratejas/headroom",
"repository": "https://github.com/chopratejas/headroom",
"keywords": [
"headroom",
"hooks",
"claude-code",
"copilot-cli"
]
}

View file

@ -0,0 +1,18 @@
{
"name": "headroom",
"version": "0.9.2",
"description": "Headroom startup hooks for Claude Code and GitHub Copilot CLI.",
"author": {
"name": "Headroom Contributors",
"url": "https://github.com/chopratejas/headroom"
},
"homepage": "https://github.com/chopratejas/headroom",
"repository": "https://github.com/chopratejas/headroom",
"keywords": [
"headroom",
"hooks",
"claude-code",
"copilot-cli"
],
"hooks": "./hooks"
}

View file

@ -0,0 +1,11 @@
# Headroom agent hooks
This plugin exposes lightweight startup hooks for Claude Code and GitHub Copilot CLI.
The hooks call:
```bash
headroom init hook ensure
```
That hidden helper checks for a matching durable `headroom init` deployment and starts it if needed.

View file

@ -0,0 +1,29 @@
{
"description": "Headroom plugin hooks — ensure the local Headroom runtime is available for initialized agents.",
"hooks": {
"SessionStart": [
{
"matcher": "startup|resume",
"hooks": [
{
"type": "command",
"command": "headroom init hook ensure",
"timeout": 15
}
]
}
],
"PreToolUse": [
{
"matcher": "Bash|PowerShell",
"hooks": [
{
"type": "command",
"command": "headroom init hook ensure",
"timeout": 15
}
]
}
]
}
}

View file

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "headroom-ai"
version = "0.5.25"
version = "0.9.1"
description = "The Context Optimization Layer for LLM Applications - Cut costs by 50-90%"
readme = "README.md"
license = "Apache-2.0"
@ -52,6 +52,7 @@ dependencies = [
"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
"tomli>=2.0.0; python_version < '3.11'", # tomllib backport for helper scripts
]
[project.optional-dependencies]

View file

@ -0,0 +1,55 @@
"""Sync plugin manifest versions to the repo's computed release semver."""
from __future__ import annotations
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from headroom.release_version import ( # noqa: E402
compute_release_version,
determine_bump_level,
find_latest_release_tag,
get_canonical_version,
list_release_commits,
list_release_tags,
)
def compute_repo_semver(root: Path) -> str:
"""Return the npm-style semver for the repo's next release."""
tags = list_release_tags(root)
previous_tag = find_latest_release_tag(tags) or ""
level = determine_bump_level(list_release_commits(root, previous_tag))
info = compute_release_version(
canonical_version=get_canonical_version(root),
level=level,
tags=tags,
)
return info.npm_version
def main() -> None:
root = ROOT
version = compute_repo_semver(root)
subprocess.run(
[
sys.executable,
str(root / "scripts" / "version-sync.py"),
"--root",
str(root),
"--version",
version,
"--plugin-manifests-only",
],
cwd=root,
check=True,
)
if __name__ == "__main__":
main()

View file

@ -0,0 +1,68 @@
"""Tests for sync-plugin-versions.py."""
from __future__ import annotations
import importlib.util
from pathlib import Path
def _load_module():
script = Path(__file__).parent.parent / "sync-plugin-versions.py"
spec = importlib.util.spec_from_file_location("sync_plugin_versions", script)
assert spec is not None
assert spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_compute_repo_semver_uses_release_helpers(monkeypatch) -> None:
module = _load_module()
calls: dict[str, object] = {}
monkeypatch.setattr(module, "list_release_tags", lambda root: ["v0.9.0"])
monkeypatch.setattr(module, "find_latest_release_tag", lambda tags: "v0.9.0")
monkeypatch.setattr(module, "list_release_commits", lambda root, tag: ["feat: add init"])
monkeypatch.setattr(module, "determine_bump_level", lambda commits: "minor")
monkeypatch.setattr(module, "get_canonical_version", lambda root: "0.5.25")
def fake_compute_release_version(*, canonical_version: str, level: str, tags: list[str]):
calls["canonical_version"] = canonical_version
calls["level"] = level
calls["tags"] = tags
return type("Info", (), {"npm_version": "0.10.0"})()
monkeypatch.setattr(module, "compute_release_version", fake_compute_release_version)
assert module.compute_repo_semver(Path("repo")) == "0.10.0"
assert calls == {
"canonical_version": "0.5.25",
"level": "minor",
"tags": ["v0.9.0"],
}
def test_main_runs_plugin_only_version_sync(monkeypatch) -> None:
module = _load_module()
commands: list[list[str]] = []
monkeypatch.setattr(module, "compute_repo_semver", lambda root: "0.10.0")
monkeypatch.setattr(
module.subprocess,
"run",
lambda command, cwd, check: commands.append(command),
)
module.main()
assert commands == [
[
module.sys.executable,
str(module.ROOT / "scripts" / "version-sync.py"),
"--root",
str(module.ROOT),
"--version",
"0.10.0",
"--plugin-manifests-only",
]
]

View file

@ -15,9 +15,17 @@ def temp_project(tmp_path: Path) -> dict[str, Path]:
root = tmp_path / "project"
headroom = root / "headroom"
headroom.mkdir(parents=True)
repo_claude_plugin = root / ".claude-plugin"
repo_claude_plugin.mkdir(parents=True)
repo_github_plugin = root / ".github" / "plugin"
repo_github_plugin.mkdir(parents=True)
plugins = root / "plugins"
openclaw = plugins / "openclaw"
openclaw.mkdir(parents=True)
agent_hooks_claude = plugins / "headroom-agent-hooks" / ".claude-plugin"
agent_hooks_claude.mkdir(parents=True)
agent_hooks_github = plugins / "headroom-agent-hooks" / ".github" / "plugin"
agent_hooks_github.mkdir(parents=True)
sdk = root / "sdk"
typescript = sdk / "typescript"
typescript.mkdir(parents=True)
@ -34,6 +42,32 @@ def temp_project(tmp_path: Path) -> dict[str, Path]:
openclaw_pkg = openclaw / "package.json"
openclaw_pkg.write_text(json.dumps({"name": "test", "version": "0.5.25"}))
repo_claude_marketplace = repo_claude_plugin / "marketplace.json"
repo_claude_marketplace.write_text(
json.dumps(
{
"metadata": {"name": "claude-marketplace", "version": "0.1.0"},
"plugins": [{"name": "headroom-agent-hooks", "version": "0.1.0"}],
}
)
)
repo_github_marketplace = repo_github_plugin / "marketplace.json"
repo_github_marketplace.write_text(
json.dumps(
{
"metadata": {"name": "copilot-marketplace", "version": "0.1.0"},
"plugins": [{"name": "headroom-agent-hooks", "version": "0.1.0"}],
}
)
)
claude_plugin = agent_hooks_claude / "plugin.json"
claude_plugin.write_text(json.dumps({"name": "headroom-agent-hooks", "version": "0.1.0"}))
github_plugin = agent_hooks_github / "plugin.json"
github_plugin.write_text(json.dumps({"name": "headroom-agent-hooks", "version": "0.1.0"}))
# sdk/typescript/package.json
typescript_pkg = typescript / "package.json"
typescript_pkg.write_text(json.dumps({"name": "test", "version": "0.5.25"}))
@ -43,6 +77,10 @@ def temp_project(tmp_path: Path) -> dict[str, Path]:
"pyproject": pyproject,
"version_py": version_py,
"openclaw_pkg": openclaw_pkg,
"repo_claude_marketplace": repo_claude_marketplace,
"repo_github_marketplace": repo_github_marketplace,
"claude_plugin": claude_plugin,
"github_plugin": github_plugin,
"typescript_pkg": typescript_pkg,
}
@ -76,6 +114,20 @@ def test_version_sync_explicit_version(temp_project: dict[str, Path]) -> None:
typescript_pkg = json.loads(temp_project["typescript_pkg"].read_text())
assert typescript_pkg["version"] == "0.7.0"
repo_claude_marketplace = json.loads(temp_project["repo_claude_marketplace"].read_text())
assert repo_claude_marketplace["metadata"]["version"] == "0.7.0"
assert repo_claude_marketplace["plugins"][0]["version"] == "0.7.0"
repo_github_marketplace = json.loads(temp_project["repo_github_marketplace"].read_text())
assert repo_github_marketplace["metadata"]["version"] == "0.7.0"
assert repo_github_marketplace["plugins"][0]["version"] == "0.7.0"
claude_plugin = json.loads(temp_project["claude_plugin"].read_text())
assert claude_plugin["version"] == "0.7.0"
github_plugin = json.loads(temp_project["github_plugin"].read_text())
assert github_plugin["version"] == "0.7.0"
# Verify .releaseetadata was created
release_metadata = root / ".releaseetadata"
assert release_metadata.exists()
@ -84,6 +136,7 @@ def test_version_sync_explicit_version(temp_project: dict[str, Path]) -> None:
assert metadata["packages"]["pypi"] == "0.7.0"
assert metadata["packages"]["npm-sdk"] == "0.7.0"
assert metadata["packages"]["npm-openclaw"] == "0.7.0"
assert metadata["packages"]["agent-hooks-plugin"] == "0.7.0"
def test_bump_patch(temp_project: dict[str, Path]) -> None:
@ -112,6 +165,9 @@ def test_bump_patch(temp_project: dict[str, Path]) -> None:
typescript_pkg = json.loads(temp_project["typescript_pkg"].read_text())
assert typescript_pkg["version"] == "0.5.26"
claude_plugin = json.loads(temp_project["claude_plugin"].read_text())
assert claude_plugin["version"] == "0.5.26"
def test_bump_minor(temp_project: dict[str, Path]) -> None:
"""Test --bump minor bumps 0.5.25 to 0.6.0."""
@ -139,6 +195,9 @@ def test_bump_minor(temp_project: dict[str, Path]) -> None:
typescript_pkg = json.loads(temp_project["typescript_pkg"].read_text())
assert typescript_pkg["version"] == "0.6.0"
github_plugin = json.loads(temp_project["github_plugin"].read_text())
assert github_plugin["version"] == "0.6.0"
def test_bump_major(temp_project: dict[str, Path]) -> None:
"""Test --bump major bumps 0.5.25 to 1.0.0."""
@ -166,6 +225,9 @@ def test_bump_major(temp_project: dict[str, Path]) -> None:
typescript_pkg = json.loads(temp_project["typescript_pkg"].read_text())
assert typescript_pkg["version"] == "1.0.0"
repo_claude_marketplace = json.loads(temp_project["repo_claude_marketplace"].read_text())
assert repo_claude_marketplace["metadata"]["version"] == "1.0.0"
def test_release_metadata_written(temp_project: dict[str, Path]) -> None:
"""Test .releaseetadata is written correctly."""
@ -190,5 +252,40 @@ def test_release_metadata_written(temp_project: dict[str, Path]) -> None:
"pypi": "0.6.0",
"npm-sdk": "0.6.0",
"npm-openclaw": "0.6.0",
"agent-hooks-plugin": "0.6.0",
},
}
def test_plugin_manifests_only_leaves_package_versions_unchanged(
temp_project: dict[str, Path],
) -> None:
"""Test plugin-only sync leaves canonical package versions alone."""
root = temp_project["root"]
script = Path(__file__).parent.parent / "version-sync.py"
result = subprocess.run(
[
sys.executable,
str(script),
"--root",
str(root),
"--version",
"0.8.0",
"--plugin-manifests-only",
],
capture_output=True,
text=True,
)
assert result.returncode == 0, f"Script failed: {result.stderr}"
assert 'version = "0.5.25"' in temp_project["pyproject"].read_text()
assert '__version__ = "0.5.25"' in temp_project["version_py"].read_text()
assert json.loads(temp_project["openclaw_pkg"].read_text())["version"] == "0.5.25"
assert json.loads(temp_project["typescript_pkg"].read_text())["version"] == "0.5.25"
assert json.loads(temp_project["claude_plugin"].read_text())["version"] == "0.8.0"
assert (
json.loads(temp_project["repo_github_marketplace"].read_text())["metadata"]["version"]
== "0.8.0"
)
assert not (root / ".releaseetadata").exists()

View file

@ -4,7 +4,10 @@
import json
from pathlib import Path
import tomllib
try:
import tomllib
except ImportError: # pragma: no cover - Python 3.10 fallback
import tomli as tomllib
ROOT = Path(__file__).parent.parent

View file

@ -8,7 +8,10 @@ import json
import re
from pathlib import Path
import tomllib
try:
import tomllib
except ImportError: # pragma: no cover - Python 3.10 fallback
import tomli as tomllib
def get_version_from_pyproject(root: Path) -> str:
@ -56,6 +59,46 @@ def update_package_json(file_path: Path, version: str) -> None:
f.write("\n")
def update_plugin_manifest(file_path: Path, version: str) -> None:
"""Update a plugin.json version field."""
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
data["version"] = version
with open(file_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
def update_marketplace_manifest(file_path: Path, version: str) -> None:
"""Update marketplace metadata and plugin entry versions."""
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
metadata = data.get("metadata")
if isinstance(metadata, dict):
metadata["version"] = version
plugins = data.get("plugins")
if isinstance(plugins, list):
for plugin in plugins:
if isinstance(plugin, dict):
plugin["version"] = version
with open(file_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
def update_plugin_versions(root: Path, version: str) -> None:
"""Update marketplace and plugin manifest versions."""
update_marketplace_manifest(root / ".claude-plugin" / "marketplace.json", version)
update_marketplace_manifest(root / ".github" / "plugin" / "marketplace.json", version)
update_plugin_manifest(
root / "plugins" / "headroom-agent-hooks" / ".claude-plugin" / "plugin.json", version
)
update_plugin_manifest(
root / "plugins" / "headroom-agent-hooks" / ".github" / "plugin" / "plugin.json",
version,
)
def update_openclaw_package_json(file_path: Path, version: str, sdk_version: str) -> None:
"""Update openclaw package.json version and headroom-ai dependency range."""
with open(file_path, encoding="utf-8") as f:
@ -89,6 +132,7 @@ def write_release_metadata(root: Path, version: str) -> None:
"pypi": version,
"npm-sdk": version,
"npm-openclaw": version,
"agent-hooks-plugin": version,
},
}
metadata_path = root / ".releaseetadata"
@ -112,6 +156,11 @@ def main() -> None:
choices=["major", "minor", "patch"],
help="Bump version from pyproject.toml",
)
parser.add_argument(
"--plugin-manifests-only",
action="store_true",
help="Only update marketplace/plugin manifest versions",
)
args = parser.parse_args()
if args.version:
@ -122,6 +171,11 @@ def main() -> None:
else:
version = get_version_from_pyproject(args.root)
if args.plugin_manifests_only:
update_plugin_versions(args.root, version)
print(f"Plugin versions synchronized to {version}")
return
# Update all versioned files
update_pyproject_version(args.root, version)
update_version_py(args.root, version)
@ -129,6 +183,7 @@ def main() -> None:
args.root / "plugins" / "openclaw" / "package.json", version, version
)
update_package_json(args.root / "sdk" / "typescript" / "package.json", version)
update_plugin_versions(args.root, version)
write_release_metadata(args.root, version)
print(f"Version synchronized to {version}")

View file

@ -0,0 +1,829 @@
from __future__ import annotations
import importlib
import json
import sys
import types
from pathlib import Path
from types import SimpleNamespace
import click
import pytest
from click.testing import CliRunner
def _load_init_module(monkeypatch):
monkeypatch.delitem(sys.modules, "headroom.cli.init", raising=False)
monkeypatch.delitem(sys.modules, "headroom.cli.main", raising=False)
fake_main_module = types.ModuleType("headroom.cli.main")
@click.group()
def fake_main() -> None:
pass
fake_main_module.main = fake_main
monkeypatch.setitem(sys.modules, "headroom.cli.main", fake_main_module)
importlib.invalidate_caches()
init_cli = importlib.import_module("headroom.cli.init")
monkeypatch.delitem(sys.modules, "headroom.cli.init", raising=False)
return init_cli, fake_main
def test_init_auto_detects_targets(monkeypatch) -> None:
init_cli, fake_main = _load_init_module(monkeypatch)
runner = CliRunner()
captured: dict[str, object] = {}
monkeypatch.setattr(init_cli, "detect_init_targets", lambda global_scope: ["claude", "codex"])
monkeypatch.setattr(init_cli, "_run_init_targets", lambda **kwargs: captured.update(kwargs))
result = runner.invoke(fake_main, ["init", "-g"])
assert result.exit_code == 0, result.output
assert captured["targets"] == ["claude", "codex"]
assert captured["global_scope"] is True
def test_init_fails_when_auto_detection_empty(monkeypatch) -> None:
init_cli, fake_main = _load_init_module(monkeypatch)
runner = CliRunner()
monkeypatch.setattr(init_cli, "detect_init_targets", lambda global_scope: [])
result = runner.invoke(fake_main, ["init"])
assert result.exit_code != 0
assert "auto-detected" in result.output
def test_init_copilot_requires_global(monkeypatch) -> None:
init_cli, fake_main = _load_init_module(monkeypatch)
runner = CliRunner()
monkeypatch.setattr(init_cli, "_ensure_runtime_manifest", lambda **kwargs: "init-local-test")
result = runner.invoke(fake_main, ["init", "copilot"])
assert result.exit_code != 0
assert "requires -g" in result.output
def test_init_claude_local_writes_settings_and_installs_marketplace(
monkeypatch, tmp_path: Path
) -> None:
init_cli, fake_main = _load_init_module(monkeypatch)
runner = CliRunner()
monkeypatch.chdir(tmp_path)
marketplace_calls: list[str] = []
monkeypatch.setattr(init_cli, "_ensure_runtime_manifest", lambda **kwargs: "init-local-demo")
monkeypatch.setattr(
init_cli,
"_install_claude_marketplace",
lambda scope: marketplace_calls.append(scope),
)
result = runner.invoke(fake_main, ["init", "claude"])
assert result.exit_code == 0, result.output
settings_path = tmp_path / ".claude" / "settings.local.json"
payload = json.loads(settings_path.read_text(encoding="utf-8"))
assert payload["env"]["ANTHROPIC_BASE_URL"] == "http://127.0.0.1:8787"
assert marketplace_calls == ["local"]
assert any(
"--profile init-local-demo" in hook["command"] and "init hook ensure" in hook["command"]
for entry in payload["hooks"]["SessionStart"]
for hook in entry["hooks"]
)
def test_init_codex_merges_feature_flag_into_existing_table(monkeypatch, tmp_path: Path) -> None:
init_cli, _ = _load_init_module(monkeypatch)
monkeypatch.chdir(tmp_path)
config_path = tmp_path / ".codex" / "config.toml"
config_path.parent.mkdir(parents=True)
config_path.write_text("[features]\nshell_tool = true\n", encoding="utf-8")
init_cli._init_codex(global_scope=False, profile="init-local-demo", port=9000)
content = config_path.read_text(encoding="utf-8")
assert 'base_url = "http://127.0.0.1:9000/v1"' in content
assert content.count("[features]") == 1
assert "codex_hooks = true" in content
hooks = json.loads((tmp_path / ".codex" / "hooks.json").read_text(encoding="utf-8"))
assert "--profile init-local-demo" in hooks["hooks"]["SessionStart"][0]["hooks"][0]["command"]
assert "init hook ensure" in hooks["hooks"]["SessionStart"][0]["hooks"][0]["command"]
def test_init_claude_uses_custom_port(monkeypatch, tmp_path: Path) -> None:
init_cli, _ = _load_init_module(monkeypatch)
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(init_cli, "_install_claude_marketplace", lambda scope: None)
init_cli._init_claude(global_scope=False, profile="init-local-demo", port=9011)
payload = json.loads((tmp_path / ".claude" / "settings.local.json").read_text(encoding="utf-8"))
assert payload["env"]["ANTHROPIC_BASE_URL"] == "http://127.0.0.1:9011"
def test_init_copilot_global_writes_hooks_and_env(monkeypatch, tmp_path: Path) -> None:
init_cli, _ = _load_init_module(monkeypatch)
captured_env: dict[str, str] = {}
monkeypatch.setattr(init_cli, "_copilot_config_path", lambda: tmp_path / "copilot-config.json")
monkeypatch.setattr(init_cli, "_apply_user_env", lambda values: captured_env.update(values))
monkeypatch.setattr(init_cli, "_install_copilot_marketplace", lambda: None)
init_cli._init_copilot(global_scope=True, profile="init-user", port=9005, backend="openai")
payload = json.loads((tmp_path / "copilot-config.json").read_text(encoding="utf-8"))
assert "SessionStart" in payload["hooks"]
assert "PreToolUse" in payload["hooks"]
assert "--profile init-user" in payload["hooks"]["SessionStart"][0]["command"]
assert captured_env == {
"COPILOT_PROVIDER_TYPE": "openai",
"COPILOT_PROVIDER_BASE_URL": "http://127.0.0.1:9005/v1",
"COPILOT_PROVIDER_WIRE_API": "completions",
}
def test_init_hook_ensure_prefers_local_profile(monkeypatch) -> None:
init_cli, fake_main = _load_init_module(monkeypatch)
ensured: list[str] = []
def fake_load(profile: str):
return object() if profile == "init-repo-12345678" else None
monkeypatch.setattr(init_cli, "_local_profile", lambda cwd=None: "init-repo-12345678")
monkeypatch.setattr(init_cli, "load_manifest", fake_load)
monkeypatch.setattr(
init_cli, "_ensure_profile_running", lambda profile: ensured.append(profile)
)
runner = CliRunner()
result = runner.invoke(fake_main, ["init", "hook", "ensure"])
assert result.exit_code == 0, result.output
assert ensured == ["init-repo-12345678"]
def test_init_openclaw_requires_global(monkeypatch) -> None:
_, fake_main = _load_init_module(monkeypatch)
runner = CliRunner()
result = runner.invoke(fake_main, ["init", "openclaw"])
assert result.exit_code != 0
assert "requires -g" in result.output
def test_init_openclaw_delegates_to_wrap(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
calls: list[list[str]] = []
class _Result:
returncode = 0
monkeypatch.setattr(init_cli, "resolve_headroom_command", lambda: ["headroom"])
monkeypatch.setattr(
init_cli.subprocess,
"run",
lambda cmd: calls.append(cmd) or _Result(),
)
init_cli._init_openclaw(global_scope=True, port=9999)
assert calls == [["headroom", "wrap", "openclaw", "--proxy-port", "9999"]]
def test_detect_init_targets_respects_scope(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
monkeypatch.setattr(
init_cli.shutil,
"which",
lambda name: name if name in {"claude", "copilot", "codex", "openclaw"} else None,
)
assert init_cli.detect_init_targets(False) == ["claude", "codex"]
assert init_cli.detect_init_targets(True) == ["claude", "copilot", "codex", "openclaw"]
def test_marketplace_source_prefers_env_override(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
monkeypatch.setenv("HEADROOM_MARKETPLACE_SOURCE", "custom/source")
assert init_cli._marketplace_source() == "custom/source"
def test_run_checked_treats_existing_install_as_success(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
class _Result:
returncode = 1
stderr = "plugin already exists"
stdout = ""
monkeypatch.setattr(init_cli.subprocess, "run", lambda *args, **kwargs: _Result())
init_cli._run_checked(["claude", "plugin", "install"], action="claude plugin install")
def test_command_string_and_matcher_on_windows(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
monkeypatch.setattr(init_cli, "os", SimpleNamespace(name="nt"))
monkeypatch.setattr(init_cli.subprocess, "list2cmdline", lambda parts: "joined-command")
assert init_cli._command_string(["headroom", "init"]) == "joined-command"
assert init_cli._powershell_matcher() == "Bash|PowerShell"
def test_json_file_handles_missing_empty_and_non_mapping(monkeypatch, tmp_path: Path) -> None:
init_cli, _ = _load_init_module(monkeypatch)
missing = tmp_path / "missing.json"
empty = tmp_path / "empty.json"
array_payload = tmp_path / "payload.json"
empty.write_text(" \n", encoding="utf-8")
array_payload.write_text('["value"]\n', encoding="utf-8")
assert init_cli._json_file(missing) == {}
assert init_cli._json_file(empty) == {}
assert init_cli._json_file(array_payload) == {}
def test_ensure_claude_hooks_rewrites_existing_entries(monkeypatch, tmp_path: Path) -> None:
init_cli, _ = _load_init_module(monkeypatch)
settings_path = tmp_path / "settings.json"
settings_path.write_text(
json.dumps(
{
"env": {"KEEP": "1"},
"hooks": {
"SessionStart": [
"not-a-dict",
{"hooks": "not-a-list"},
{
"matcher": "startup|resume",
"hooks": [{"type": "command", "command": "echo keep-me"}],
},
{
"matcher": "startup|resume",
"hooks": [
{
"type": "command",
"command": "headroom init hook ensure --marker headroom-init-claude",
}
],
},
]
},
}
),
encoding="utf-8",
)
monkeypatch.setattr(init_cli, "_hook_command", lambda *parts: "headroom init hook ensure")
init_cli._ensure_claude_hooks(settings_path, "init-local-demo", 9001)
payload = json.loads(settings_path.read_text(encoding="utf-8"))
assert payload["env"] == {"KEEP": "1", "ANTHROPIC_BASE_URL": "http://127.0.0.1:9001"}
session_entries = payload["hooks"]["SessionStart"]
assert session_entries[0] == "not-a-dict"
assert session_entries[1] == {"hooks": "not-a-list"}
assert session_entries[2]["hooks"][0]["command"] == "echo keep-me"
assert session_entries[-1]["hooks"][0]["command"].endswith("--marker headroom-init-claude")
def test_ensure_copilot_hooks_replaces_existing_marker(monkeypatch, tmp_path: Path) -> None:
init_cli, _ = _load_init_module(monkeypatch)
config_path = tmp_path / "copilot.json"
config_path.write_text(
json.dumps(
{
"hooks": {
"SessionStart": [
{"type": "command", "command": "echo keep"},
{
"type": "command",
"command": "headroom init hook ensure --marker headroom-init-copilot",
},
]
}
}
),
encoding="utf-8",
)
monkeypatch.setattr(init_cli, "_hook_command", lambda *parts: "headroom init hook ensure")
init_cli._ensure_copilot_hooks(config_path, "init-user")
payload = json.loads(config_path.read_text(encoding="utf-8"))
commands = [entry["command"] for entry in payload["hooks"]["SessionStart"]]
assert commands == ["echo keep", "headroom init hook ensure --marker headroom-init-copilot"]
def test_replace_marker_block_replaces_existing_block(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
content = "before\n# start\nold\n# end\nafter\n"
replaced = init_cli._replace_marker_block(content, "# start", "# end", "# start\nnew\n# end")
assert replaced == "before\n\nafter\n\n# start\nnew\n# end\n"
def test_ensure_codex_provider_replaces_existing_marker(monkeypatch, tmp_path: Path) -> None:
init_cli, _ = _load_init_module(monkeypatch)
path = tmp_path / "config.toml"
path.write_text(
f"prefix\n{init_cli._CODEX_PROVIDER_MARKER_START}\nold = true\n{init_cli._CODEX_PROVIDER_MARKER_END}\n",
encoding="utf-8",
)
init_cli._ensure_codex_provider(path, 9100)
content = path.read_text(encoding="utf-8")
assert content.count(init_cli._CODEX_PROVIDER_MARKER_START) == 1
assert 'base_url = "http://127.0.0.1:9100/v1"' in content
assert "old = true" not in content
def test_ensure_codex_feature_flag_replaces_existing_marker(monkeypatch, tmp_path: Path) -> None:
init_cli, _ = _load_init_module(monkeypatch)
path = tmp_path / "config.toml"
path.write_text(
f"[features]\n{init_cli._CODEX_FEATURE_MARKER_START}\ncodex_hooks = false\n{init_cli._CODEX_FEATURE_MARKER_END}\n",
encoding="utf-8",
)
init_cli._ensure_codex_feature_flag(path)
content = path.read_text(encoding="utf-8")
assert content.count(init_cli._CODEX_FEATURE_MARKER_START) == 1
assert "codex_hooks = true" in content
def test_ensure_codex_feature_flag_skips_duplicate_existing_setting(
monkeypatch, tmp_path: Path
) -> None:
init_cli, _ = _load_init_module(monkeypatch)
path = tmp_path / "config.toml"
path.write_text("[features]\ncodex_hooks = true\nshell_tool = true\n", encoding="utf-8")
init_cli._ensure_codex_feature_flag(path)
content = path.read_text(encoding="utf-8")
assert content.count("codex_hooks = true") == 1
assert init_cli._CODEX_FEATURE_MARKER_START not in content
def test_ensure_codex_feature_flag_creates_features_section_when_missing(
monkeypatch, tmp_path: Path
) -> None:
init_cli, _ = _load_init_module(monkeypatch)
path = tmp_path / "config.toml"
path.write_text('model = "gpt-5"\n', encoding="utf-8")
init_cli._ensure_codex_feature_flag(path)
content = path.read_text(encoding="utf-8")
assert "[features]" in content
assert "codex_hooks = true" in content
def test_manifest_changed_detects_differences(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
existing = SimpleNamespace(
port=8787,
backend="anthropic",
anyllm_provider=None,
region=None,
memory_enabled=False,
)
assert not init_cli._manifest_changed(
existing,
port=8787,
backend="anthropic",
anyllm_provider=None,
region=None,
memory=False,
)
assert init_cli._manifest_changed(
existing,
port=9000,
backend="anthropic",
anyllm_provider=None,
region=None,
memory=False,
)
def test_ensure_runtime_manifest_merges_targets_and_stops_changed_runtime(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
existing = SimpleNamespace(
targets=["claude"],
mutations=["mutation"],
port=8787,
backend="anthropic",
anyllm_provider=None,
region=None,
memory_enabled=False,
)
saved: list[object] = []
stopped: list[object] = []
built = SimpleNamespace(supervisor_kind="", artifacts=[], mutations=[], targets=[])
monkeypatch.setattr(init_cli, "_runtime_profile", lambda global_scope, cwd=None: "init-user")
monkeypatch.setattr(init_cli, "load_manifest", lambda profile: existing)
monkeypatch.setattr(
init_cli,
"build_manifest",
lambda **kwargs: built.__dict__.update(kwargs) or built,
)
monkeypatch.setattr(init_cli, "save_manifest", lambda manifest: saved.append(manifest))
monkeypatch.setattr(init_cli, "stop_runtime", lambda manifest: stopped.append(manifest))
profile = init_cli._ensure_runtime_manifest(
global_scope=True,
targets=["codex"],
port=9001,
backend="anthropic",
anyllm_provider=None,
region=None,
memory=False,
)
assert profile == "init-user"
assert stopped == [existing]
assert saved == [built]
assert built.targets == ["claude", "codex"]
assert built.mutations == ["mutation"]
assert built.supervisor_kind == init_cli.SupervisorKind.NONE.value
assert built.artifacts == []
def test_ensure_runtime_manifest_ignores_stop_runtime_errors(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
existing = SimpleNamespace(
targets=[],
mutations=[],
port=8787,
backend="anthropic",
anyllm_provider=None,
region=None,
memory_enabled=False,
)
saved: list[object] = []
built = SimpleNamespace(supervisor_kind="", artifacts=[], mutations=[], targets=[])
monkeypatch.setattr(init_cli, "_runtime_profile", lambda global_scope, cwd=None: "init-user")
monkeypatch.setattr(init_cli, "load_manifest", lambda profile: existing)
monkeypatch.setattr(
init_cli,
"build_manifest",
lambda **kwargs: built.__dict__.update(kwargs) or built,
)
monkeypatch.setattr(init_cli, "save_manifest", lambda manifest: saved.append(manifest))
monkeypatch.setattr(
init_cli, "stop_runtime", lambda manifest: (_ for _ in ()).throw(RuntimeError("boom"))
)
init_cli._ensure_runtime_manifest(
global_scope=True,
targets=["claude"],
port=9001,
backend="anthropic",
anyllm_provider=None,
region=None,
memory=False,
)
assert saved == [built]
def test_apply_user_env_routes_by_platform(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
manifest = SimpleNamespace(base_env={"OLD": "1"}, tool_envs={})
windows_calls: list[object] = []
unix_calls: list[object] = []
monkeypatch.setattr(init_cli, "_env_manifest", lambda values: manifest)
monkeypatch.setattr(
init_cli, "_apply_windows_env_scope", lambda value: windows_calls.append(value)
)
monkeypatch.setattr(init_cli, "_apply_unix_env_scope", lambda value: unix_calls.append(value))
monkeypatch.setattr(init_cli, "os", SimpleNamespace(name="nt"))
init_cli._apply_user_env({"COPILOT_PROVIDER_TYPE": "openai"})
monkeypatch.setattr(init_cli, "os", SimpleNamespace(name="posix"))
init_cli._apply_user_env({"COPILOT_PROVIDER_TYPE": "anthropic"})
assert manifest.base_env == {}
assert manifest.tool_envs == {"copilot": {"COPILOT_PROVIDER_TYPE": "anthropic"}}
assert windows_calls == [manifest]
assert unix_calls == [manifest]
def test_resolve_copilot_env_supports_anthropic(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
assert init_cli._resolve_copilot_env(9010, "anthropic") == {
"COPILOT_PROVIDER_TYPE": "anthropic",
"COPILOT_PROVIDER_BASE_URL": "http://127.0.0.1:9010",
}
def test_marketplace_source_prefers_repo_checkout(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
monkeypatch.delenv("HEADROOM_MARKETPLACE_SOURCE", raising=False)
assert init_cli._marketplace_source() == str(Path(init_cli.__file__).resolve().parents[2])
def test_run_checked_raises_on_failure(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
class _Result:
returncode = 2
stderr = "bad stderr"
stdout = "bad stdout"
monkeypatch.setattr(init_cli.subprocess, "run", lambda *args, **kwargs: _Result())
with pytest.raises(
click.ClickException, match="claude plugin install failed: bad stderr\nbad stdout"
):
init_cli._run_checked(["claude", "plugin", "install"], action="claude plugin install")
def test_install_claude_marketplace_errors_without_binary(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
monkeypatch.setattr(init_cli.shutil, "which", lambda name: None)
with pytest.raises(click.ClickException, match="'claude' not found"):
init_cli._install_claude_marketplace("local")
def test_install_claude_marketplace_runs_expected_commands(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
calls: list[tuple[list[str], str]] = []
monkeypatch.setattr(init_cli.shutil, "which", lambda name: "claude")
monkeypatch.setattr(init_cli, "_marketplace_source", lambda: "repo/source")
monkeypatch.setattr(
init_cli, "_run_checked", lambda command, action: calls.append((command, action))
)
init_cli._install_claude_marketplace("user")
assert calls == [
(["claude", "plugin", "marketplace", "add", "repo/source"], "claude marketplace add"),
(
["claude", "plugin", "install", "headroom@headroom-marketplace", "--scope", "user"],
"claude plugin install",
),
]
def test_install_copilot_marketplace_handles_missing_binary(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
monkeypatch.setattr(init_cli.shutil, "which", lambda name: None)
with pytest.raises(click.ClickException, match="'copilot' not found"):
init_cli._install_copilot_marketplace()
def test_install_copilot_marketplace_runs_expected_commands(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
calls: list[tuple[list[str], str]] = []
monkeypatch.setattr(init_cli.shutil, "which", lambda name: "copilot")
monkeypatch.setattr(init_cli, "_marketplace_source", lambda: "repo/source")
monkeypatch.setattr(
init_cli, "_run_checked", lambda command, action: calls.append((command, action))
)
init_cli._install_copilot_marketplace()
assert calls == [
(["copilot", "plugin", "marketplace", "add", "repo/source"], "copilot marketplace add"),
(
["copilot", "plugin", "install", "headroom@headroom-marketplace"],
"copilot plugin install",
),
]
def test_ensure_profile_running_covers_runtime_modes(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
docker_manifest = SimpleNamespace(
preset=init_cli.InstallPreset.PERSISTENT_DOCKER.value,
supervisor_kind=init_cli.SupervisorKind.NONE.value,
profile="docker-profile",
)
service_manifest = SimpleNamespace(
preset=init_cli.InstallPreset.PERSISTENT_TASK.value,
supervisor_kind=init_cli.SupervisorKind.SERVICE.value,
profile="service-profile",
)
task_manifest = SimpleNamespace(
preset=init_cli.InstallPreset.PERSISTENT_TASK.value,
supervisor_kind=init_cli.SupervisorKind.NONE.value,
profile="task-profile",
)
manifests = {
"docker-profile": docker_manifest,
"service-profile": service_manifest,
"task-profile": task_manifest,
}
docker_calls: list[object] = []
service_calls: list[object] = []
detached_calls: list[str] = []
wait_calls: list[tuple[str, int]] = []
monkeypatch.setattr(init_cli, "load_manifest", lambda profile: manifests.get(profile))
def fake_wait_ready(manifest, timeout_seconds: int) -> bool:
wait_calls.append((manifest.profile, timeout_seconds))
return False
monkeypatch.setattr(init_cli, "wait_ready", fake_wait_ready)
monkeypatch.setattr(
init_cli, "start_persistent_docker", lambda manifest: docker_calls.append(manifest)
)
monkeypatch.setattr(
init_cli, "start_supervisor", lambda manifest: service_calls.append(manifest)
)
monkeypatch.setattr(
init_cli,
"start_detached_agent",
lambda profile: detached_calls.append(profile),
)
init_cli._ensure_profile_running("missing")
init_cli._ensure_profile_running("docker-profile")
init_cli._ensure_profile_running("service-profile")
init_cli._ensure_profile_running("task-profile")
assert docker_calls == [docker_manifest]
assert service_calls == [service_manifest]
assert detached_calls == ["task-profile"]
assert ("docker-profile", 1) in wait_calls
assert ("docker-profile", 45) in wait_calls
def test_ensure_profile_running_returns_when_ready_or_on_exception(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
manifest = SimpleNamespace(
preset=init_cli.InstallPreset.PERSISTENT_TASK.value,
supervisor_kind=init_cli.SupervisorKind.NONE.value,
profile="task-profile",
)
detached_calls: list[str] = []
monkeypatch.setattr(init_cli, "load_manifest", lambda profile: manifest)
monkeypatch.setattr(init_cli, "wait_ready", lambda manifest, timeout_seconds: True)
monkeypatch.setattr(
init_cli,
"start_detached_agent",
lambda profile: detached_calls.append(profile),
)
init_cli._ensure_profile_running("task-profile")
assert detached_calls == []
monkeypatch.setattr(init_cli, "wait_ready", lambda manifest, timeout_seconds: False)
monkeypatch.setattr(
init_cli,
"start_detached_agent",
lambda profile: (_ for _ in ()).throw(RuntimeError("boom")),
)
init_cli._ensure_profile_running("task-profile")
def test_init_codex_windows_warns_about_upstream_hook_limitation(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
messages: list[str] = []
monkeypatch.setattr(init_cli, "os", SimpleNamespace(name="nt"))
monkeypatch.setattr(init_cli, "_codex_scope_path", lambda global_scope: Path("config.toml"))
monkeypatch.setattr(init_cli, "_codex_hooks_path", lambda global_scope: Path("hooks.json"))
monkeypatch.setattr(init_cli, "_ensure_codex_provider", lambda path, port: None)
monkeypatch.setattr(init_cli, "_ensure_codex_feature_flag", lambda path: None)
monkeypatch.setattr(init_cli, "_ensure_codex_hooks", lambda path, profile: None)
monkeypatch.setattr(init_cli.click, "echo", lambda message: messages.append(message))
init_cli._init_codex(global_scope=True, profile="init-user", port=9000)
assert any("disabled upstream on Windows" in message for message in messages)
def test_init_openclaw_propagates_nonzero_exit(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
class _Result:
returncode = 9
monkeypatch.setattr(init_cli, "resolve_headroom_command", lambda: ["headroom"])
monkeypatch.setattr(init_cli.subprocess, "run", lambda command: _Result())
with pytest.raises(SystemExit) as exc:
init_cli._init_openclaw(global_scope=True, port=9999)
assert exc.value.code == 9
def test_run_init_targets_dispatches_supported_targets(monkeypatch) -> None:
init_cli, _ = _load_init_module(monkeypatch)
calls: list[tuple[str, tuple[object, ...]]] = []
monkeypatch.setattr(init_cli, "_ensure_runtime_manifest", lambda **kwargs: "init-profile")
monkeypatch.setattr(
init_cli,
"_init_claude",
lambda **kwargs: calls.append(
("claude", (kwargs["global_scope"], kwargs["profile"], kwargs["port"]))
),
)
monkeypatch.setattr(
init_cli,
"_init_copilot",
lambda **kwargs: calls.append(
("copilot", (kwargs["global_scope"], kwargs["profile"], kwargs["port"]))
),
)
monkeypatch.setattr(
init_cli,
"_init_codex",
lambda **kwargs: calls.append(
("codex", (kwargs["global_scope"], kwargs["profile"], kwargs["port"]))
),
)
monkeypatch.setattr(
init_cli,
"_init_openclaw",
lambda **kwargs: calls.append(("openclaw", (kwargs["global_scope"], kwargs["port"]))),
)
init_cli._run_init_targets(
targets=["claude", "copilot", "codex", "openclaw"],
global_scope=True,
port=9000,
backend="openai",
anyllm_provider="provider",
region="us-east-1",
memory=True,
)
assert calls == [
("claude", (True, "init-profile", 9000)),
("copilot", (True, "init-profile", 9000)),
("codex", (True, "init-profile", 9000)),
("openclaw", (True, 9000)),
]
def test_init_subcommand_uses_group_options(monkeypatch) -> None:
init_cli, fake_main = _load_init_module(monkeypatch)
runner = CliRunner()
captured: dict[str, object] = {}
monkeypatch.setattr(init_cli, "_run_init_targets", lambda **kwargs: captured.update(kwargs))
result = runner.invoke(
fake_main,
["init", "-g", "--port", "9007", "--backend", "openai", "--memory", "claude"],
)
assert result.exit_code == 0, result.output
assert captured == {
"targets": ["claude"],
"global_scope": True,
"port": 9007,
"backend": "openai",
"anyllm_provider": None,
"region": None,
"memory": True,
}
def test_init_hook_ensure_prefers_global_when_local_missing(monkeypatch) -> None:
init_cli, fake_main = _load_init_module(monkeypatch)
ensured: list[str] = []
monkeypatch.setattr(init_cli, "_local_profile", lambda cwd=None: "init-repo-12345678")
monkeypatch.setattr(
init_cli,
"load_manifest",
lambda profile: object() if profile == init_cli._GLOBAL_PROFILE else None,
)
monkeypatch.setattr(
init_cli, "_ensure_profile_running", lambda profile: ensured.append(profile)
)
runner = CliRunner()
result = runner.invoke(fake_main, ["init", "hook", "ensure"])
assert result.exit_code == 0, result.output
assert ensured == [init_cli._GLOBAL_PROFILE]
def test_init_hook_ensure_uses_explicit_profile(monkeypatch) -> None:
init_cli, fake_main = _load_init_module(monkeypatch)
ensured: list[str] = []
monkeypatch.setattr(
init_cli, "_ensure_profile_running", lambda profile: ensured.append(profile)
)
runner = CliRunner()
result = runner.invoke(fake_main, ["init", "hook", "ensure", "--profile", "init-explicit"])
assert result.exit_code == 0, result.output
assert ensured == ["init-explicit"]

View file

@ -0,0 +1,55 @@
from __future__ import annotations
import json
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
def _load_json(relative_path: str) -> object:
return json.loads((REPO_ROOT / relative_path).read_text(encoding="utf-8"))
def test_marketplace_manifests_match() -> None:
assert _load_json(".claude-plugin/marketplace.json") == _load_json(
".github/plugin/marketplace.json"
)
def test_plugin_manifests_share_core_metadata() -> None:
claude = _load_json("plugins/headroom-agent-hooks/.claude-plugin/plugin.json")
copilot = _load_json("plugins/headroom-agent-hooks/.github/plugin/plugin.json")
assert isinstance(claude, dict)
assert isinstance(copilot, dict)
for key in ("name", "version", "description", "author", "homepage", "repository", "keywords"):
assert claude[key] == copilot[key]
assert "hooks" not in claude
assert copilot["hooks"] == "./hooks"
def test_marketplace_entry_points_to_plugin_root() -> None:
marketplace = _load_json(".claude-plugin/marketplace.json")
assert isinstance(marketplace, dict)
plugins = marketplace["plugins"]
assert isinstance(plugins, list)
plugin = plugins[0]
assert plugin["name"] == "headroom"
plugin_root = (REPO_ROOT / plugin["source"]).resolve()
assert plugin_root.is_dir()
assert (plugin_root / ".claude-plugin" / "plugin.json").is_file()
assert (plugin_root / "hooks" / "hooks.json").is_file()
def test_plugin_metadata_points_to_upstream_repo() -> None:
expected_repo = "https://github.com/chopratejas/headroom"
marketplace = _load_json(".claude-plugin/marketplace.json")
claude = _load_json("plugins/headroom-agent-hooks/.claude-plugin/plugin.json")
assert isinstance(marketplace, dict)
assert isinstance(claude, dict)
plugin = marketplace["plugins"][0]
assert plugin["author"]["url"] == expected_repo
assert plugin["homepage"] == expected_repo
assert plugin["repository"] == expected_repo
assert claude["author"]["url"] == expected_repo
assert claude["homepage"] == expected_repo
assert claude["repository"] == expected_repo

View file

@ -130,6 +130,7 @@ def test_savings_tracker_sanitizes_legacy_state_and_applies_retention(tmp_path):
assert snapshot["retention"] == {
"max_history_points": 1,
"max_history_age_days": 2,
"max_response_history_points": 500,
}
@ -484,6 +485,60 @@ def test_savings_tracker_rollups_preserve_spend_and_input_history(tmp_path, monk
]
def test_stats_history_defaults_to_compact_history_but_can_return_full_history(
tmp_path, monkeypatch
):
path = tmp_path / "proxy_savings.json"
tracker = SavingsTracker(
path=str(path),
max_history_points=100,
max_history_age_days=30,
max_response_history_points=5,
)
monkeypatch.setattr(
"headroom.proxy.savings_tracker._estimate_compression_savings_usd",
lambda model, tokens_saved: tokens_saved / 1000.0,
)
for i in range(8):
tracker.record_compression_savings(
model="gpt-4o",
tokens_saved=10,
total_input_tokens=(i + 1) * 100,
total_input_cost_usd=(i + 1) * 0.1,
timestamp=f"2026-03-27T09:{i:02d}:00Z",
)
compact = tracker.history_response()
assert compact["history_summary"] == {
"mode": "compact",
"stored_points": 8,
"returned_points": 5,
"compacted": True,
}
assert len(compact["history"]) == 5
assert compact["history"][0]["timestamp"] == "2026-03-27T09:00:00Z"
assert compact["history"][-1]["timestamp"] == "2026-03-27T09:07:00Z"
full = tracker.history_response(history_mode="full")
assert full["history_summary"] == {
"mode": "full",
"stored_points": 8,
"returned_points": 8,
"compacted": False,
}
assert len(full["history"]) == 8
none = tracker.history_response(history_mode="none")
assert none["history"] == []
assert none["history_summary"] == {
"mode": "none",
"stored_points": 8,
"returned_points": 0,
"compacted": True,
}
def test_stats_history_persists_across_restarts_and_stats_stays_compatible(tmp_path, monkeypatch):
savings_path = tmp_path / "proxy_savings.json"
monkeypatch.setenv("HEADROOM_SAVINGS_PATH", str(savings_path))
@ -533,6 +588,12 @@ def test_stats_history_persists_across_restarts_and_stats_stays_compatible(tmp_p
assert history_data["series"]["hourly"][0]["total_input_cost_usd_delta"] == pytest.approx(
0.24
)
assert history_data["history_summary"] == {
"mode": "compact",
"stored_points": 1,
"returned_points": 1,
"compacted": False,
}
assert stats_data["display_session"] == history_data["display_session"]
assert (
@ -560,6 +621,11 @@ def test_stats_history_persists_across_restarts_and_stats_stays_compatible(tmp_p
assert updated["series"]["daily"][0]["total_input_tokens_delta"] == 240
assert updated["series"]["daily"][0]["total_input_cost_usd_delta"] == pytest.approx(0.48)
full = client.get("/stats-history?history_mode=full").json()
assert full["history_summary"]["mode"] == "full"
assert full["history_summary"]["stored_points"] == 2
assert full["history_summary"]["returned_points"] == 2
persisted = json.loads(savings_path.read_text())
assert persisted["lifetime"]["tokens_saved"] == 55
assert persisted["lifetime"]["total_input_tokens"] == 240

View file

@ -100,7 +100,7 @@ curl http://localhost:8787/stats-history
```json
{
"schema_version": 1,
"schema_version": 2,
"generated_at": "2026-03-27T09:10:00Z",
"lifetime": {
"tokens_saved": 12500,
@ -123,6 +123,12 @@ curl http://localhost:8787/stats-history
"default_format": "json",
"available_formats": ["json", "csv"],
"available_series": ["history", "hourly", "daily", "weekly", "monthly"]
},
"history_summary": {
"mode": "compact",
"stored_points": 2048,
"returned_points": 500,
"compacted": true
}
}
```
@ -132,11 +138,17 @@ compression history. It survives proxy restarts, tolerates missing or malformed
state files, and powers the historical view in `/dashboard`. It now includes
hourly, daily, weekly, and monthly chart-ready rollups.
By default, the `history` array is compacted for transport efficiency. Use
`history_mode=full` when you explicitly need the full retained checkpoint list,
or `history_mode=none` when you only need the aggregate rollups and lifetime
totals.
For export-friendly downloads:
```bash
curl "http://localhost:8787/stats-history?format=csv&series=daily"
curl "http://localhost:8787/stats-history?format=csv&series=monthly"
curl "http://localhost:8787/stats-history?history_mode=full"
```
CSV exports are available for `history`, `hourly`, `daily`, `weekly`, and

View file

@ -241,8 +241,10 @@ curl http://localhost:8787/stats-history
other Headroom frontends. It returns:
- lifetime proxy compression totals
- bounded persisted checkpoint history
- compact checkpoint history by default, with `history_mode=full` available for
export/debug flows
- derived hourly, daily, weekly, and monthly rollups for charts
- a `history_summary` block describing stored versus returned checkpoint counts
- UTC timestamps throughout
By default the proxy stores this history at
@ -258,6 +260,7 @@ daily/weekly/monthly rollups and built-in JSON / CSV export buttons.
```bash
curl "http://localhost:8787/stats-history?format=csv&series=weekly"
curl "http://localhost:8787/stats-history?format=csv&series=monthly"
curl "http://localhost:8787/stats-history?history_mode=full"
```
### Prometheus Metrics