refactor(proxy): isolate output verbosity policy (#1963)

## Description

Extracts output verbosity steering text and sentinel replacement into a
pure `output_verbosity_policy` module. `output_shaper` continues to
mutate Anthropic/OpenAI request bodies, while the byte-stable steering
block and replacement rules now live behind deterministic, directly
tested policy functions.

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.output_verbosity_policy` for steering sentinels,
level text, `steering_text`, and `replace_or_append_steering_block`.
- Updated `output_shaper` to delegate pure steering text/replacement
rules while preserving existing public imports and request mutation
behavior.
- Added direct policy tests for byte-stable steering text, append,
replacement, malformed sentinel handling, and idempotency.
- Included the current LiteLLM callback signature compatibility shim
required for repo-wide mypy on main-based slices.

## 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_output_verbosity_policy.py tests/test_output_shaper.py tests/test_litellm_callback.py -q
58 passed in 6.23s

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, clean worktree based on
`headroomlabs/main`.
- Exact command / steps: targeted pytest, ruff, format check, repo-wide
mypy, staged gitleaks scan.
- Observed result: output verbosity policy/shaper/callback tests pass;
static checks pass; no staged secrets detected.
- Not tested: live provider calls; this slice preserves existing request
mutation behavior and only moves pure steering rules.

## 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
architecture slice. PR-specific GHAS checks will be monitored after
opening.
This commit is contained in:
JD Davis 2026-07-13 04:35:09 +00:00 committed by GitHub
parent f359f21424
commit 0415dc8765
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 138 additions and 56 deletions

View file

@ -4,62 +4,13 @@ from __future__ import annotations
from typing import Any
# Sentinel prefix marks the steering block so application is idempotent and
# the block is recognizable in logs/diffs.
_STEERING_SENTINEL = "<headroom_output_shaping>"
_STEERING_SUFFIX = "</headroom_output_shaping>"
# Levels are cumulative: each includes everything above it. Text must stay
# byte-stable across releases for prefix-cache friendliness; treat edits to
# these strings as cache-busting changes.
_VERBOSITY_LEVELS = {
1: (
"Skip preamble and postamble. Do not announce what you are about to "
"do or recap what you just did; start with the substance."
),
2: (
"Skip preamble and postamble; start with the substance. Never restate "
"code, file contents, diffs, or tool output that already appear in "
"this conversation — reference them by path and line instead. After a "
"tool call succeeds, continue without narrating the result."
),
3: (
"Skip preamble and postamble. Never restate code, file contents, "
"diffs, or tool output already in this conversation — reference by "
"path and line. Give conclusions only; omit rationale unless the user "
"asks why. Prefer the smallest edit over rewriting whole files. Keep "
"prose to the minimum needed to be unambiguous."
),
4: (
"Minimum tokens. Fragments fine. No preamble, no postamble, no "
"restating context, no rationale. Answer, smallest-possible edits, "
"nothing else."
),
}
def steering_text(level: int) -> str | None:
"""The full steering block for a verbosity level, or None for level 0."""
text = _VERBOSITY_LEVELS.get(level)
if text is None:
return None
return f"{_STEERING_SENTINEL}\n{text}\n{_STEERING_SUFFIX}"
def replace_or_append_steering_block(existing: str, block: str) -> tuple[str, bool]:
"""Replace an existing steering block in text, or append one at the tail."""
start = existing.find(_STEERING_SENTINEL)
if start >= 0:
end = existing.find(_STEERING_SUFFIX, start)
end = len(existing) if end < 0 else end + len(_STEERING_SUFFIX)
prefix = existing[:start].rstrip()
suffix = existing[end:].lstrip("\n")
parts = [part for part in (prefix, block, suffix) if part]
updated = "\n\n".join(parts)
return updated, updated != existing
updated = f"{existing.rstrip()}\n\n{block}" if existing.strip() else block
return updated, updated != existing
from headroom.proxy.output_verbosity_policy import (
STEERING_SENTINEL as _STEERING_SENTINEL,
)
from headroom.proxy.output_verbosity_policy import (
replace_or_append_steering_block,
steering_text,
)
def apply_verbosity_steering(body: dict[str, Any], level: int) -> bool:

View file

@ -0,0 +1,60 @@
"""Pure output verbosity steering policy."""
from __future__ import annotations
# Sentinel prefix marks the steering block so application is idempotent and
# the block is recognizable in logs/diffs.
STEERING_SENTINEL = "<headroom_output_shaping>"
STEERING_SUFFIX = "</headroom_output_shaping>"
# Levels are cumulative: each includes everything above it. Text must stay
# byte-stable across releases for prefix-cache friendliness; edits to these
# strings are cache-busting changes.
VERBOSITY_LEVELS = {
1: (
"Skip preamble and postamble. Do not announce what you are about to "
"do or recap what you just did; start with the substance."
),
2: (
"Skip preamble and postamble; start with the substance. Never restate "
"code, file contents, diffs, or tool output that already appear in "
"this conversation — reference them by path and line instead. After a "
"tool call succeeds, continue without narrating the result."
),
3: (
"Skip preamble and postamble. Never restate code, file contents, "
"diffs, or tool output already in this conversation — reference by "
"path and line. Give conclusions only; omit rationale unless the user "
"asks why. Prefer the smallest edit over rewriting whole files. Keep "
"prose to the minimum needed to be unambiguous."
),
4: (
"Minimum tokens. Fragments fine. No preamble, no postamble, no "
"restating context, no rationale. Answer, smallest-possible edits, "
"nothing else."
),
}
def steering_text(level: int) -> str | None:
"""The full steering block for a verbosity level, or ``None`` for level 0."""
text = VERBOSITY_LEVELS.get(level)
if text is None:
return None
return f"{STEERING_SENTINEL}\n{text}\n{STEERING_SUFFIX}"
def replace_or_append_steering_block(existing: str, block: str) -> tuple[str, bool]:
"""Replace an existing steering block in text, or append one at the tail."""
start = existing.find(STEERING_SENTINEL)
if start >= 0:
end = existing.find(STEERING_SUFFIX, start)
end = len(existing) if end < 0 else end + len(STEERING_SUFFIX)
prefix = existing[:start].rstrip()
suffix = existing[end:].lstrip("\n")
parts = [part for part in (prefix, block, suffix) if part]
updated = "\n\n".join(parts)
return updated, updated != existing
updated = f"{existing.rstrip()}\n\n{block}" if existing.strip() else block
return updated, updated != existing

View file

@ -0,0 +1,71 @@
"""Tests for pure output verbosity steering policy."""
from __future__ import annotations
from headroom.proxy.output_verbosity_policy import (
STEERING_SENTINEL,
STEERING_SUFFIX,
replace_or_append_steering_block,
steering_text,
)
def test_level_zero_and_unknown_levels_have_no_steering_text() -> None:
assert steering_text(0) is None
assert steering_text(99) is None
def test_steering_text_is_wrapped_and_byte_stable() -> None:
first = steering_text(2)
second = steering_text(2)
assert first == second
assert first is not None
assert first.startswith(f"{STEERING_SENTINEL}\n")
assert first.endswith(f"\n{STEERING_SUFFIX}")
assert "Never restate code" in first
def test_replace_or_append_adds_block_to_nonempty_instructions() -> None:
block = steering_text(3)
assert block is not None
updated, changed = replace_or_append_steering_block("System.", block)
assert changed is True
assert updated == f"System.\n\n{block}"
def test_replace_or_append_uses_block_for_empty_instructions() -> None:
block = steering_text(1)
assert block is not None
updated, changed = replace_or_append_steering_block(" ", block)
assert changed is True
assert updated == block
def test_replace_or_append_replaces_existing_complete_block_once() -> None:
old = steering_text(1)
new = steering_text(4)
assert old is not None
assert new is not None
updated, changed = replace_or_append_steering_block(f"System.\n\n{old}\n\nTail.", new)
assert changed is True
assert old not in updated
assert updated == f"System.\n\n{new}\n\nTail."
def test_replace_or_append_replaces_unclosed_sentinel_to_end() -> None:
new = steering_text(2)
assert new is not None
updated, changed = replace_or_append_steering_block(
f"System.\n\n{STEERING_SENTINEL}\nold text without close",
new,
)
assert changed is True
assert updated == f"System.\n\n{new}"
def test_replace_or_append_is_idempotent_when_block_matches() -> None:
block = steering_text(2)
assert block is not None
updated, changed = replace_or_append_steering_block(f"System.\n\n{block}", block)
assert changed is False
assert updated == f"System.\n\n{block}"