diff --git a/headroom/proxy/output_steering.py b/headroom/proxy/output_steering.py index c9ef1a263..6eb31b64e 100644 --- a/headroom/proxy/output_steering.py +++ b/headroom/proxy/output_steering.py @@ -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 = "" -_STEERING_SUFFIX = "" - -# 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: diff --git a/headroom/proxy/output_verbosity_policy.py b/headroom/proxy/output_verbosity_policy.py new file mode 100644 index 000000000..5c692b797 --- /dev/null +++ b/headroom/proxy/output_verbosity_policy.py @@ -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 = "" +STEERING_SUFFIX = "" + +# 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 diff --git a/tests/test_output_verbosity_policy.py b/tests/test_output_verbosity_policy.py new file mode 100644 index 000000000..79653ac6c --- /dev/null +++ b/tests/test_output_verbosity_policy.py @@ -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}"