Skip to content

Commit a6d4921

Browse files
authored
feat(proxy/hooks): run fold-only (stream-safe) turn hooks on streaming OpenAI chat (headroomlabs-ai#2549)
## Description Fixes the last harness gap in the turn-hook seam (the "B4" finding from the savings audit). The OpenAI chat handler gated hooks on `not stream`, so **streamed** `/v1/chat/completions` requests ran **no** turn hooks — the lossless-guard plugin's on_request fold and tool-schema shrink were skipped, unlike the Anthropic path (hooks run unconditionally). Affects opencode / Cursor / older OpenAI SDKs / some Copilot flows; **not** Claude Code (Anthropic path). The gate existed for a real reason: hooks that **re-drive** the model in `on_response` (defer a tool, reload it when asked) can't run mid-stream. But an **on_request fold** mutates the outbound request before the send — safe on a stream. ## Change - Add an opt-in `stream_safe` hook attribute (fold-only hooks set it). `run_request_hooks(ctx, stream_safe_only=…)` filters to stream-safe hooks when set. - OpenAI chat handler runs `on_request` on streaming with `stream_safe_only=stream`; buffered runs all hooks; the `on_response` re-drive (buffered response path) is untouched. - **Default off = conservative:** a hook is buffered-only unless it declares `stream_safe`, so **no behavior change** until a hook opts in. ## Type of Change - [x] Bug fix / feature (opt-in, backward-compatible) ## Testing ```text pytest tests/test_turn_hooks.py tests/test_openai_chat_turn_hooks.py -q → 25 passed ruff + mypy → clean ``` New test pins the filter: streaming runs only stream-safe hooks' on_request; buffered runs all. ## Notes The companion plugin PR (headroom-lossless-guard) sets `stream_safe = True` on its fold-only hook to actually claim the streaming savings. Anthropic path already ran hooks on streaming, so it's unaffected. ## Checklist - [x] Self-reviewed; tests pass; no CHANGELOG edit
1 parent c990cfb commit a6d4921

3 files changed

Lines changed: 59 additions & 8 deletions

File tree

headroom/proxy/handlers/openai.py

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3471,17 +3471,19 @@ async def handle_openai_chat(
34713471
tokens_saved = max(0, original_tokens - optimized_tokens)
34723472

34733473
# Turn hooks (opt-in extensions): a registered hook may rewrite the
3474-
# outbound tools/messages before we send. Buffered requests only — a
3475-
# streamed turn can't be re-driven to resolve whatever the model asks to
3476-
# load. Gated on the registry so it is a no-op when none are registered;
3477-
# the net tool-schema token delta is recorded so it shows up as a saving.
3474+
# outbound tools/messages before we send. on_response re-drive (below, in
3475+
# the buffered response path) can't run on a stream, but on_request folds
3476+
# can — so on streaming we run only stream-safe (fold-only) hooks, and
3477+
# buffered runs all of them. Gated on the registry so it's a no-op when
3478+
# none are registered; the net tool-schema token delta is recorded as a
3479+
# saving.
34783480
from headroom.proxy.turn_hooks import (
34793481
TurnContext,
34803482
registered_turn_hooks,
34813483
run_request_hooks,
34823484
)
34833485

3484-
if registered_turn_hooks() and not stream:
3486+
if registered_turn_hooks():
34853487
_th_tools_before = body.get("tools")
34863488
_th_tok_before = (
34873489
tokenizer.count_text(json.dumps(_th_tools_before, default=str))
@@ -3502,7 +3504,7 @@ async def handle_openai_chat(
35023504
_th_msg_before: int | None = tokenizer.count_messages(body["messages"])
35033505
except Exception:
35043506
_th_msg_before = None
3505-
run_request_hooks(_th_ctx)
3507+
run_request_hooks(_th_ctx, stream_safe_only=stream)
35063508
# A hook may either replace ctx.messages/ctx.tools or mutate them in
35073509
# place (the contract allows both). Use object identity only to decide
35083510
# whether body needs reassignment; measure the saving from the FINAL

headroom/proxy/turn_hooks.py

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,12 @@ class TurnHook(Protocol):
5656

5757
name: str
5858

59+
# Optional: set True if this hook's ``on_request`` needs no later re-drive
60+
# (a fold-only hook). Such hooks run on streaming turns too; hooks that may
61+
# re-drive in ``on_response`` leave it unset and run buffered-only. Absent ⇒
62+
# False (conservative). See :func:`run_request_hooks`.
63+
stream_safe: bool
64+
5965
def on_request(self, ctx: TurnContext) -> None:
6066
"""Inspect / mutate ``ctx`` (e.g. ``ctx.tools``) before it goes upstream."""
6167

@@ -86,9 +92,22 @@ def clear_turn_hooks() -> None:
8692
_hooks.clear()
8793

8894

89-
def run_request_hooks(ctx: TurnContext) -> None:
90-
"""Run every hook's ``on_request``. Inert when none are registered; never raises."""
95+
def run_request_hooks(ctx: TurnContext, *, stream_safe_only: bool = False) -> None:
96+
"""Run each hook's ``on_request``. Inert when none registered; never raises.
97+
98+
``on_request`` mutates the outbound request before the upstream send, so it is
99+
safe on a streamed turn *as long as the hook needs no later re-drive*. A hook
100+
that may re-drive the model in ``on_response`` (e.g. defer a tool, then reload
101+
it when the model asks) can't run on a stream — the bytes are already flowing,
102+
there's nothing to re-drive. So on streaming turns the handler passes
103+
``stream_safe_only=True`` and only hooks that opt in via a truthy ``stream_safe``
104+
attribute run; fold-only hooks (which never re-drive) set it and thus keep
105+
working on streamed OpenAI-compatible traffic. Default off ⇒ conservative:
106+
a hook is treated as buffered-only unless it declares itself stream-safe.
107+
"""
91108
for hook in _hooks:
109+
if stream_safe_only and not getattr(hook, "stream_safe", False):
110+
continue
92111
fn = getattr(hook, "on_request", None)
93112
if fn is None:
94113
continue

tests/test_turn_hooks.py

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,36 @@ async def test_response_runner_returns_input_unchanged_when_empty():
5959
# --- on_request mutation -----------------------------------------------------
6060

6161

62+
def test_stream_safe_filter_runs_only_optted_in_hooks_on_stream():
63+
"""On a streamed turn only ``stream_safe`` hooks' on_request runs (fold-only,
64+
no re-drive); buffered runs all. A hook that may re-drive stays buffered-only."""
65+
ran: list[str] = []
66+
67+
class Fold:
68+
name = "fold"
69+
stream_safe = True # opts in — safe on streaming
70+
71+
def on_request(self, ctx: TurnContext) -> None:
72+
ran.append("fold")
73+
74+
class Redrive: # no stream_safe attr → buffered-only (default)
75+
name = "redrive"
76+
77+
def on_request(self, ctx: TurnContext) -> None:
78+
ran.append("redrive")
79+
80+
register_turn_hook(Fold())
81+
register_turn_hook(Redrive())
82+
83+
ran.clear()
84+
run_request_hooks(_ctx(), stream_safe_only=True) # streaming turn
85+
assert ran == ["fold"]
86+
87+
ran.clear()
88+
run_request_hooks(_ctx()) # buffered turn (default)
89+
assert ran == ["fold", "redrive"]
90+
91+
6292
def test_on_request_may_mutate_ctx():
6393
class Shrink:
6494
name = "shrink"

0 commit comments

Comments
 (0)