fix(mcp): show lifetime totals and label rolling session scope in headroom_stats (#1428)

## Description

`headroom_stats` currently formats only the rolling session view from
`/stats`, so users see session numbers with no explicit scope label and
no lifetime totals even though the proxy already exposes lifetime
savings data.

This PR keeps the current session summary, labels it as rolling-session
output, and appends lifetime totals from `persistent_savings.lifetime`.
It stays formatting-only on an existing payload surface.

Closes #1166

## Type of Change

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

## Changes Made

- label the existing `headroom_stats` session block as rolling-session
output
- append lifetime totals from the existing stats payload
- add focused formatter regressions and fallback coverage
- update `CHANGELOG.md`

## Testing

- [ ] Unit tests pass (`uv run pytest tests/test_ccr_mcp_server.py -v`)
- [ ] Linting passes (`uv run ruff check .`)
- [ ] Type checking passes (`uv run mypy headroom`) or explain N/A
truthfully
- [x] New tests added for new functionality when applicable
- [ ] Manual testing performed

### Test Output

```text
Focused local commands passed:
- uv run pytest tests/test_ccr_mcp_server.py -x -v
  9 passed, 1 skipped
- uv run ruff check headroom/ccr/mcp_server.py tests/test_ccr_mcp_server.py
  All checks passed
- uv run ruff format headroom/ccr/mcp_server.py tests/test_ccr_mcp_server.py --check
  2 files already formatted

Base proof on origin/main with the updated regression file:
- pytest -k "window_scoped"
  failed because the output still says "Headroom Session Summary"
- pytest -k "includes_lifetime_totals_from_persistent_savings"
  failed because the formatted text still has no "Lifetime Savings:" section

Not run locally:
- uv run mypy headroom
- Template-level broader commands `uv run pytest tests/test_ccr_mcp_server.py -v` and `uv run ruff check .`
```

## Real Behavior Proof

- Environment: focused `HeadroomMCPServer._handle_stats()` test payloads
with and without `persistent_savings.lifetime`
- Exact command / steps: run `uv run pytest tests/test_ccr_mcp_server.py
-x -v`, specifically the new `_handle_stats()` regressions that feed
summary-only, summary-plus-lifetime, missing-lifetime, and zero-lifetime
payloads through the MCP stats formatter
- Observed result: output contains `Headroom Window-Scoped Session
Summary`, appends `Lifetime Savings:` when lifetime data is present, and
omits that section cleanly when lifetime data is absent
- Not tested: broader MCP output redesign beyond this formatter

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [ ] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Additional Notes

- Scoped to the MCP text surface only; dashboard and broader
savings-window work stay out of scope.
- Attribution: the issue body identified the exact mismatch between
current `headroom_stats` output and the already-live lifetime stats
payload.
This commit is contained in:
Rod Boev 2026-06-30 09:39:34 -04:00 committed by GitHub
parent a9322477e3
commit 1c0e15243e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 114 additions and 3 deletions

View file

@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
* **telemetry:** anonymous usage telemetry is now **opt-in** (off by default) instead of opt-out. Nothing is collected or sent unless you set `HEADROOM_TELEMETRY=on` or pass `--telemetry` to `headroom proxy` / `headroom install apply`. `is_telemetry_enabled()` is fail-closed — only explicit on-values (`on`/`true`/`1`/`yes`/`enable`/`enabled`) enable it; unset, empty, or unrecognized values stay disabled. The existing `--no-telemetry` flag and `HEADROOM_TELEMETRY=off` remain accepted for back-compat, and install manifests now write the `HEADROOM_TELEMETRY` value explicitly so generated deployments are unambiguous.
* **ccr:** `headroom_stats` now labels its formatted proxy output as a rolling/window-scoped session and adds a lifetime savings section from `/stats persistent_savings.lifetime` when present, while keeping existing summary structure and fallback JSON output behavior.
### Features

View file

@ -87,10 +87,14 @@ _READ_ENABLED = os.environ.get("HEADROOM_MCP_READ", "off").lower().strip() in (
DEFAULT_PROXY_URL = os.environ.get("HEADROOM_PROXY_URL", "http://127.0.0.1:8787")
def _format_session_summary(summary: dict[str, Any], local_stats: dict[str, Any]) -> str:
def _format_session_summary(
summary: dict[str, Any],
local_stats: dict[str, Any],
persistent_lifetime: dict[str, Any] | None = None,
) -> str:
"""Format the proxy summary + local MCP stats into clean readable text."""
lines: list[str] = []
lines.append("Headroom Session Summary")
lines.append("Headroom Window-Scoped Session Summary")
lines.append("=" * 40)
mode = summary.get("mode", "token")
@ -157,6 +161,15 @@ def _format_session_summary(summary: dict[str, Any], local_stats: dict[str, Any]
lines.append(f"MCP Tool: {local_compressions} compressions, {local_saved:,} tokens saved")
lines.append("")
# Lifetime proxy savings (cross-session)
if isinstance(persistent_lifetime, dict):
lifetime_tokens = persistent_lifetime.get("tokens_saved", 0) or 0
lifetime_usd = persistent_lifetime.get("compression_savings_usd", 0.0) or 0.0
lines.append("Lifetime Savings:")
lines.append(f" Tokens saved: {lifetime_tokens:,}")
lines.append(f" Compression savings: ${lifetime_usd:.2f}")
lines.append("")
# Tip
tip = summary.get("tip")
if tip:
@ -740,8 +753,14 @@ class HeadroomMCPServer:
if proxy_data:
summary = proxy_data.get("summary")
if summary:
lifetime = None
persistent_savings = proxy_data.get("persistent_savings")
if isinstance(persistent_savings, dict):
lifetime_block = persistent_savings.get("lifetime")
if isinstance(lifetime_block, dict):
lifetime = lifetime_block
# Return clean formatted summary instead of raw JSON
formatted = _format_session_summary(summary, stats)
formatted = _format_session_summary(summary, stats, lifetime)
return [TextContent(type="text", text=formatted)]
# Fallback: add proxy stats to local stats
proxy_stats = self._extract_proxy_stats(proxy_data)

View file

@ -110,3 +110,94 @@ def test_mcp_retrieve_missing_hash_still_errors(fresh_store) -> None:
server = mcp_server.HeadroomMCPServer(check_proxy=False)
result = asyncio.run(server._retrieve_content("nonexistent_hash"))
assert "Content not found" in result.get("error", "")
def test_handle_stats_session_output_is_window_scoped() -> None:
"""window-scoped stats output should be explicitly labeled after this change."""
async def fetch_stats() -> dict[str, object]:
return {
"summary": {
"mode": "token",
"api_requests": 3,
"compression": {},
}
}
server = mcp_server.HeadroomMCPServer(check_proxy=True)
server._fetch_full_proxy_stats = fetch_stats
response = asyncio.run(server._handle_stats())
text = response[0].kwargs["text"]
assert "Headroom Window-Scoped Session Summary" in text
assert "Headroom Session Summary" not in text
def test_handle_stats_includes_lifetime_totals_from_persistent_savings() -> None:
"""Lifetime savings are appended from /stats persistent_savings.lifetime."""
async def fetch_stats() -> dict[str, object]:
return {
"summary": {
"mode": "token",
"api_requests": 3,
"compression": {},
},
"persistent_savings": {
"lifetime": {"tokens_saved": 12345, "compression_savings_usd": 7.25}
},
}
server = mcp_server.HeadroomMCPServer(check_proxy=True)
server._fetch_full_proxy_stats = fetch_stats
response = asyncio.run(server._handle_stats())
text = response[0].kwargs["text"]
assert "Lifetime Savings:" in text
assert "Tokens saved: 12,345" in text
assert "Compression savings: $7.25" in text
def test_handle_stats_falls_back_gracefully_without_persistent_lifetime() -> None:
"""Missing lifetime data should still return a valid session summary."""
async def fetch_stats() -> dict[str, object]:
return {
"summary": {
"mode": "token",
"api_requests": 3,
"compression": {},
},
"persistent_savings": {"lifetime": None},
}
server = mcp_server.HeadroomMCPServer(check_proxy=True)
server._fetch_full_proxy_stats = fetch_stats
response = asyncio.run(server._handle_stats())
text = response[0].kwargs["text"]
assert "Headroom Window-Scoped Session Summary" in text
assert "Lifetime Savings:" not in text
def test_handle_stats_shows_zero_lifetime_totals_when_present() -> None:
"""A present lifetime payload should still render explicit zero totals."""
async def fetch_stats() -> dict[str, object]:
return {
"summary": {
"mode": "token",
"api_requests": 3,
"compression": {},
},
"persistent_savings": {"lifetime": {"tokens_saved": 0, "compression_savings_usd": 0.0}},
}
server = mcp_server.HeadroomMCPServer(check_proxy=True)
server._fetch_full_proxy_stats = fetch_stats
response = asyncio.run(server._handle_stats())
text = response[0].kwargs["text"]
assert "Lifetime Savings:" in text
assert "Tokens saved: 0" in text
assert "Compression savings: $0.00" in text