refactor(proxy): isolate project attribution policy (#1957)

## Description

Extracts pure project attribution policy from the runtime project
context holder. Header classification, project path splitting, and
project-prefixed base URL construction now live in a policy module while
`project_context` keeps the ContextVar and ASGI scope adapter
responsibilities.

Closes #

## Type of Change

- [ ] 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
- [x] Code refactoring (no functional changes)

## Changes Made

- Added `headroom.proxy.project_policy` for pure project attribution
header/path/base-URL helpers.
- Updated `headroom.proxy.project_context` to re-export the pure helpers
and retain only request context binding and ASGI scope mutation.
- Added direct tests for the extracted project attribution policy
boundary.
- Kept the LiteLLM callback compatibility shim required for repo-wide
type checking on fresh branches.

## Testing

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

### Test Output

```text
python -m pytest tests/test_project_policy.py tests/test_proxy_project_savings.py tests/test_litellm_callback.py tests/test_compress_api.py::TestLiteLLMCallback -q
29 passed in 13.70s

python -m ruff check .
All checks passed!

python -m ruff format --check .
1095 files already formatted

python -m mypy headroom --ignore-missing-imports
Success: no issues found in 409 source files

gitleaks protect --staged --no-banner --redact
no leaks found
```

## Real Behavior Proof

- Environment: Windows, Python 3.13.13, local worktree based on
`headroomlabs/main`.
- Exact command / steps: ran focused project policy tests, project
savings tests, LiteLLM callback tests, Ruff lint/format checks, mypy
over `headroom`, and staged gitleaks scan.
- Observed result: all local checks passed; staged secret scan found no
leaks.
- Not tested: full CI matrix and deployment flows; those are covered by
GitHub Actions.

## Review Readiness

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

## Checklist

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

## Screenshots (if applicable)

N/A

## Additional Notes

Documentation and changelog updates are not applicable for this internal
refactor. GitHub reported existing Dependabot alerts on the default
branch during push; this PR does not change dependencies, and the staged
secret scan is clean.
This commit is contained in:
JD Davis 2026-07-11 04:50:27 +00:00 committed by GitHub
parent 740fb9bc16
commit 1c1e360112
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 84 additions and 47 deletions

View file

@ -15,28 +15,23 @@ that request, matching pre-feature behavior.
from __future__ import annotations
from collections.abc import Mapping, MutableMapping
from collections.abc import MutableMapping
from contextvars import ContextVar
from typing import Any
from urllib.parse import quote, unquote, urlsplit, urlunsplit
from urllib.parse import quote
from headroom.proxy.project_policy import (
PROJECT_HEADER,
PROJECT_PATH_PREFIX,
classify_project,
split_project_path,
with_project_prefix,
)
from headroom.proxy.savings_tracker import sanitize_project_name
PROJECT_HEADER = "x-headroom-project"
PROJECT_PATH_PREFIX = "/p/"
_current_project: ContextVar[str | None] = ContextVar("headroom_current_project", default=None)
def classify_project(headers: Mapping[str, Any] | Any) -> str | None:
"""Extract a sanitized project name from request headers, if present."""
get = getattr(headers, "get", None)
if get is None:
return None
value = get(PROJECT_HEADER) or get("X-Headroom-Project")
return sanitize_project_name(value)
def set_current_project(project: str | None) -> None:
"""Bind the active request's project for downstream outcome recording."""
_current_project.set(sanitize_project_name(project))
@ -47,24 +42,6 @@ def get_current_project() -> str | None:
return _current_project.get()
def split_project_path(path: str) -> tuple[str | None, str]:
"""Split ``/p/<name>/rest`` into ``(name, /rest)``.
Clients that cannot send custom headers (aider, Copilot BYOK, Cursor)
are pointed at a project-prefixed base URL instead; the first path
segment after ``/p/`` is the URL-encoded project name. Returns
``(None, path)`` unchanged when the prefix is absent or unusable.
"""
if not path.startswith(PROJECT_PATH_PREFIX):
return None, path
remainder = path[len(PROJECT_PATH_PREFIX) :]
segment, sep, rest = remainder.partition("/")
project = sanitize_project_name(unquote(segment)) if segment else None
if project is None:
return None, path
return project, ("/" + rest) if sep else "/"
def strip_project_path_prefix(scope: MutableMapping[str, Any]) -> str | None:
"""Strip a ``/p/<name>`` prefix from an ASGI scope, returning the name.
@ -79,21 +56,6 @@ def strip_project_path_prefix(scope: MutableMapping[str, Any]) -> str | None:
return project
def with_project_prefix(base_url: str, project: str | None) -> str:
"""Insert ``/p/<name>`` ahead of the path of a local proxy base URL.
Producer-side counterpart of :func:`split_project_path`, used by
``headroom wrap`` for clients that cannot send custom headers.
Returns ``base_url`` unchanged when the project name is unusable.
"""
name = sanitize_project_name(project)
if name is None:
return base_url
parts = urlsplit(base_url)
prefixed = f"{PROJECT_PATH_PREFIX}{quote(name, safe='')}{parts.path}"
return urlunsplit(parts._replace(path=prefixed.rstrip("/")))
__all__ = [
"PROJECT_HEADER",
"PROJECT_PATH_PREFIX",

View file

@ -0,0 +1,43 @@
"""Pure project attribution policy helpers."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from urllib.parse import quote, unquote, urlsplit, urlunsplit
from headroom.proxy.savings_tracker import sanitize_project_name
PROJECT_HEADER = "x-headroom-project"
PROJECT_PATH_PREFIX = "/p/"
def classify_project(headers: Mapping[str, Any] | Any) -> str | None:
"""Extract a sanitized project name from request headers, if present."""
get = getattr(headers, "get", None)
if get is None:
return None
value = get(PROJECT_HEADER) or get("X-Headroom-Project")
return sanitize_project_name(value)
def split_project_path(path: str) -> tuple[str | None, str]:
"""Split ``/p/<name>/rest`` into ``(name, /rest)``."""
if not path.startswith(PROJECT_PATH_PREFIX):
return None, path
remainder = path[len(PROJECT_PATH_PREFIX) :]
segment, sep, rest = remainder.partition("/")
project = sanitize_project_name(unquote(segment)) if segment else None
if project is None:
return None, path
return project, ("/" + rest) if sep else "/"
def with_project_prefix(base_url: str, project: str | None) -> str:
"""Insert ``/p/<name>`` ahead of the path of a local proxy base URL."""
name = sanitize_project_name(project)
if name is None:
return base_url
parts = urlsplit(base_url)
prefixed = f"{PROJECT_PATH_PREFIX}{quote(name, safe='')}{parts.path}"
return urlunsplit(parts._replace(path=prefixed.rstrip("/")))

View file

@ -0,0 +1,32 @@
"""Tests for pure project attribution policy helpers."""
from __future__ import annotations
from headroom.proxy.project_policy import (
classify_project,
split_project_path,
with_project_prefix,
)
def test_classify_project_reads_project_header() -> None:
assert classify_project({"x-headroom-project": " frontend "}) == "frontend"
assert classify_project({"X-Headroom-Project": "api"}) == "api"
assert classify_project({"user-agent": "codex"}) is None
assert classify_project(object()) is None
def test_split_project_path_extracts_sanitized_project_and_path() -> None:
assert split_project_path("/p/frontend/v1/messages") == ("frontend", "/v1/messages")
assert split_project_path("/p/my%20repo/v1") == ("my repo", "/v1")
assert split_project_path("/p/frontend") == ("frontend", "/")
assert split_project_path("/v1/messages") == (None, "/v1/messages")
assert split_project_path("/p/%20%20/v1") == (None, "/p/%20%20/v1")
def test_with_project_prefix_round_trips_with_split_project_path() -> None:
url = with_project_prefix("http://127.0.0.1:8787/v1", "my repo")
assert url == "http://127.0.0.1:8787/p/my%20repo/v1"
assert split_project_path("/p/my%20repo/v1") == ("my repo", "/v1")
assert with_project_prefix("http://127.0.0.1:8787/v1", " ") == ("http://127.0.0.1:8787/v1")