fix(streaming): extract inner steer text from typed steer rows in place (#7600)

This commit is contained in:
webtecnica committed 2026-09-23 11:19:25 -03:00
1 parent 9a6a9acffa
commit 6d1753bb6c
3 files changed
+174 -257

No files matched your search

+79 -39
View File
@@ -5441,52 +5441,92 @@ def _strip_oob_blocks(content):
return content
def _content_has_oob_marker(content) -> bool:
"""True when ``content`` holds a complete [OUT-OF-BAND USER MESSAGE] block.
_OOB_ANY_OPEN_RE = re.compile(
r'\[OUT-OF-BAND\s+USER\s+MESSAGE(?:\s*(?:—|-)\s*.*?)?\]',
re.IGNORECASE,
)
_OOB_ANY_CLOSE_RE = re.compile(
r'\[/OUT-OF-BAND\s+USER\s+MESSAGE\]',
re.IGNORECASE,
)
Cheap pre-check so a scrub pass over a settled transcript only copies the
rows that actually carry the control wrapper.
def _unwrap_single_oob_frame(content: str) -> str | None:
"""Unwrap exactly one fully-validated [OUT-OF-BAND USER MESSAGE] frame.
Returns the extracted inner user text if and only if ``content`` consists of
exactly one valid opening tag and one valid closing tag wrapping the user
content. If markers are multiple, nested, incomplete, or ambiguous, returns
None so caller preserves the row byte-for-byte.
"""
if not isinstance(content, str):
return None
stripped = content.strip()
if not stripped:
return None
open_matches = list(_OOB_ANY_OPEN_RE.finditer(stripped))
close_matches = list(_OOB_ANY_CLOSE_RE.finditer(stripped))
# Must have exactly one opening marker and one closing marker
if len(open_matches) != 1 or len(close_matches) != 1:
return None
open_m = open_matches[0]
close_m = close_matches[0]
# Opening marker must be at the very start of stripped content
if open_m.start() != 0:
return None
# Closing marker must be at the very end of stripped content
if close_m.end() != len(stripped):
return None
# Opening marker must end before closing marker starts
if open_m.end() > close_m.start():
return None
inner = stripped[open_m.end():close_m.start()]
# Strip surrounding whitespace/newlines from the extracted user text
return inner.strip('\r\n').strip()
def _unwrap_steer_row_oob_marker(message: dict) -> None:
"""Extract inner steer text from a typed steer row in place (#7600).
The Hermes Agent emits mid-turn steers as standalone typed user rows
(`role == 'user'`, `display_kind == 'steer'`). The control wrapper
`[OUT-OF-BAND USER MESSAGE ...] ... [/OUT-OF-BAND USER MESSAGE]` is
extracted to preserve only the user-authored instruction.
Mutates caller-row in place to maintain object identity. If the marker
frame is malformed, nested, multiple, or legacy, preserves the row
byte-for-byte.
"""
if not isinstance(message, dict):
return
if message.get('role') != 'user' or message.get('display_kind') != 'steer':
return
content = message.get('content')
if isinstance(content, str):
return bool(_OOB_USER_MESSAGE_BLOCK_RE.search(content))
if isinstance(content, list):
return any(_content_has_oob_marker(part) for part in content)
if isinstance(content, dict):
return any(_content_has_oob_marker(value) for value in content.values())
return False
unwrapped = _unwrap_single_oob_frame(content)
if unwrapped is not None:
message['content'] = unwrapped
elif isinstance(content, list):
if len(content) == 1 and isinstance(content[0], dict):
part = content[0]
if part.get('type') == 'text' and isinstance(part.get('text'), str):
unwrapped = _unwrap_single_oob_frame(part['text'])
if unwrapped is not None:
part['text'] = unwrapped
def _strip_oob_markers_from_messages(messages):
"""Drop consumed OOB steer wrappers from the tool rows that carry them.
A mid-turn ``/steer`` is delivered as an ``[OUT-OF-BAND USER MESSAGE ...]``
block appended to the turn's last tool result. The wrapper is agent control
data: the gateway history builder already strips it from the model-facing
copy, and the settled transcript must not keep it either — ``session.messages``
is rendered verbatim in the UI (#7600).
Only ``role == 'tool'`` rows are candidates, because that is the only place
the transport ever appends a wrapper. User and assistant rows are
user-visible content: a message that quotes a complete marker — a pasted
example, a log excerpt — must survive verbatim, so the scrub never reaches
those rows. Only the carrier's ``content`` is rebuilt; every other field
stays the object it already is, so stable ids, reasoning metadata, and turn
bookkeeping are untouched and the scrubbed rows still compare equal to the
marker-free rows already persisted from earlier turns.
"""
cleaned = []
"""Unwrap OOB steer markers from typed steer rows in place (#7600)."""
for message in messages or []:
if not isinstance(message, dict) or message.get('role') != 'tool':
cleaned.append(message)
continue
content = message.get('content')
if not _content_has_oob_marker(content):
cleaned.append(message)
continue
scrubbed = dict(message)
scrubbed['content'] = _strip_oob_blocks(content)
cleaned.append(scrubbed)
return cleaned
_unwrap_steer_row_oob_marker(message)
return messages
def _content_has_reasoning_only_parts(content) -> bool:
+7 -12
View File
@@ -215,18 +215,13 @@ settle.
Steer must not exist only as a toast or transient DOM state. It must survive
refresh, session switch, replay, and settled render.
**The transport wrapper is not the Steer row.** A Steer reaches the agent as an
`[OUT-OF-BAND USER MESSAGE …] … [/OUT-OF-BAND USER MESSAGE]` block appended to
the run's last tool result. That block is transport-level control data, not user
content: settlement drops it from both written copies — the display transcript
(`session.messages`) and the persisted context state
(`session.context_messages`) — so the wrapper never renders in the settled view
and is never replayed to the model. The Steer's own user-visible row is not
touched by that scrub, which is scoped to the tool row that carried the wrapper:
user text that merely quotes a complete marker (a pasted example, a log excerpt)
stays byte-for-byte and appears exactly once. A session whose transcript was
stored before this behavior existed keeps the wrapper in place until its next
settlement, when the merged transcript is scrubbed again.
**The transport wrapper is not the Steer row.** A Steer is emitted as a standalone
typed user row (`role: user`, `display_kind: "steer"`). The surrounding
`[OUT-OF-BAND USER MESSAGE …] … [/OUT-OF-BAND USER MESSAGE]` block is transport-level
control data: settlement unwraps the single validated frame in place, extracting
and preserving the user's authored steer instruction while dropping the wrapper.
Legacy tool rows and rows where markers are multiple, nested, incomplete, or
contain ambiguous delimiters are preserved byte-for-byte.
### Relationship to system control events
+88 -206
View File
@@ -1,16 +1,14 @@
"""Settled writeback must drop consumed OUT-OF-BAND steer wrappers (#7600).
"""Settled writeback must extract inner steer text from typed steer rows (#7600).
A mid-turn ``/steer`` is delivered to the agent as an
``[OUT-OF-BAND USER MESSAGE ...] ... [/OUT-OF-BAND USER MESSAGE]`` block appended
to the turn's last tool result. The model-facing path strips that wrapper
(``_strip_oob_blocks`` in the gateway history builder), but the writeback in
``_prepare_marker_clean_writeback`` kept the raw text, so the settled display
transcript (``session.messages``) and the persisted context
(``session.context_messages``) both stored the wrapper and the frontend rendered
it verbatim.
A mid-turn ``/steer`` is emitted as a standalone typed user row
(``role: 'user'``, ``display_kind: 'steer'``). The surrounding
``[OUT-OF-BAND USER MESSAGE ...] ... [/OUT-OF-BAND USER MESSAGE]`` block is
transport-level control data. Settlement unwraps the single validated frame
in place, preserving the user's authored steer instruction while dropping
the wrapper.
These tests drive the real settle path (``_settle_result_messages``) with the
production message shape and assert on what actually gets persisted.
Legacy tool rows and rows where markers are multiple, nested, incomplete,
or contain ambiguous delimiters are preserved byte-for-byte.
"""
from __future__ import annotations
@@ -18,7 +16,7 @@ import copy
import json
from api.models import Session
from api.streaming import _settle_result_messages
from api.streaming import _settle_result_messages, _unwrap_single_oob_frame, _unwrap_steer_row_oob_marker
from api import streaming as _streaming
PROMPT = "please run the smoke checks"
@@ -60,15 +58,15 @@ def _session_with_prior_turn() -> Session:
return session
def _settle_turn_with_oob(session, monkeypatch, *, tool_content: str, steer_row=None):
"""Settle one turn whose last tool result carries the consumed steer block."""
def _settle_turn_with_steer(session, monkeypatch, *, tool_content: str = "checks passed", steer_rows=None):
"""Settle one turn with an optional steer row emitted by the Agent."""
monkeypatch.setattr(
_streaming, "_annotate_media_snapshots_for_settled_messages", lambda messages: None
)
previous = list(session.messages)
previous_context = list(session.context_messages)
ts = 1788440000
steer_rows = list(steer_row or [])
steers = list(steer_rows or [])
result = copy.deepcopy(previous_context) + [
{"role": "user", "content": PROMPT, "timestamp": ts},
{
@@ -83,222 +81,106 @@ def _settle_turn_with_oob(session, monkeypatch, *, tool_content: str, steer_row=
"content": tool_content,
"timestamp": ts + 2,
},
*steer_rows,
*steers,
{"role": "assistant", "content": "Smoke checks passed.", "timestamp": ts + 3},
]
_settle_result_messages(session, previous, previous_context, result, PROMPT, "webui", None)
return previous
def test_settle_drops_oob_wrapper_from_display_and_context(monkeypatch):
"""The consumed steer wrapper must not survive into either written copy."""
def test_settle_unwraps_steer_row_in_place_preserving_user_text(monkeypatch):
"""A typed steer row is unwrapped in place: inner text preserved, wrapper gone."""
session = _session_with_prior_turn()
_settle_turn_with_oob(
session, monkeypatch, tool_content=f"checks passed\n\n{OOB_BLOCK}"
)
steer_row = {
"role": "user",
"display_kind": "steer",
"content": OOB_BLOCK,
"timestamp": 1788440002.5,
}
_settle_turn_with_steer(session, monkeypatch, steer_rows=[steer_row])
# The outer wrapper is dropped from display and context
assert not _contains_oob(session.messages)
assert not _contains_oob(session.context_messages)
# The user's inner steer text is preserved exactly
steer_contents = [
m.get("content")
for m in session.messages
if isinstance(m, dict) and m.get("display_kind") == "steer"
]
assert steer_contents == ["steer: use the staging bucket this time"]
# In-place mutation preserved the dictionary object identity
assert steer_row["content"] == "steer: use the staging bucket this time"
def test_settle_unwraps_list_based_content_steer_row(monkeypatch):
"""List-based content parts on a steer row are unwrapped too."""
session = _session_with_prior_turn()
steer_row = {
"role": "user",
"display_kind": "steer",
"content": [{"type": "text", "text": OOB_VARIANT}],
"timestamp": 1788440002.5,
}
_settle_turn_with_steer(session, monkeypatch, steer_rows=[steer_row])
assert not _contains_oob(session.messages)
assert not _contains_oob(session.context_messages)
assert steer_row["content"] == [{"type": "text", "text": "steer: also bump the timeout"}]
def test_settle_leaves_legacy_tool_rows_alone(monkeypatch):
"""Legacy tool rows carrying raw markers are left alone (not truncated/deleted)."""
session = _session_with_prior_turn()
tool_text = f"checks passed\n\n{OOB_BLOCK}"
_settle_turn_with_steer(session, monkeypatch, tool_content=tool_text)
assert not _contains_oob(session.messages), (
"raw [OUT-OF-BAND USER MESSAGE] wrapper leaked into the display transcript"
)
assert not _contains_oob(session.context_messages), (
"raw [OUT-OF-BAND USER MESSAGE] wrapper leaked into the model context"
)
# The tool result it was appended to keeps its own output.
tool_bodies = [
m.get("content") for m in session.messages if isinstance(m, dict) and m.get("role") == "tool"
]
assert "checks passed" in json.dumps(tool_bodies)
assert session.messages[-1]["content"] == "Smoke checks passed."
assert tool_text in tool_bodies
def test_settle_drops_oob_wrapper_in_list_content_parts(monkeypatch):
"""List-based content parts are stripped too (the helper already supports them)."""
def test_settle_preserves_malformed_nested_and_multiple_blocks_byte_for_byte(monkeypatch):
"""Nested markers, multiple frames, and incomplete markers degrade gracefully."""
session = _session_with_prior_turn()
_settle_turn_with_oob(
session,
monkeypatch,
tool_content=[
{"type": "text", "text": f"checks passed\n{OOB_BLOCK}\nrevision 41"},
{"type": "text", "text": "no marker here"},
],
)
nested = f"{OOB_OPEN}\nhello {OOB_VARIANT} world\n{OOB_CLOSE}"
multiple = f"{OOB_VARIANT} and {OOB_BLOCK}"
incomplete = "[OUT-OF-BAND USER MESSAGE — truncated"
assert not _contains_oob(session.messages)
assert not _contains_oob(session.context_messages)
assert "checks passed" in json.dumps(session.messages)
assert "no marker here" in json.dumps(session.messages)
for malformed in (nested, multiple, incomplete):
steer_row = {
"role": "user",
"display_kind": "steer",
"content": malformed,
"timestamp": 1788440002.5,
}
_unwrap_steer_row_oob_marker(steer_row)
assert steer_row["content"] == malformed, f"malformed content was corrupted: {malformed!r}"
def test_settle_drops_every_oob_block_in_one_row(monkeypatch):
session = _session_with_prior_turn()
_settle_turn_with_oob(
session, monkeypatch, tool_content=f"before {OOB_BLOCK} middle {OOB_VARIANT} end"
)
assert not _contains_oob(session.messages)
assert not _contains_oob(session.context_messages)
joined = json.dumps(session.messages)
assert "before" in joined and "middle" in joined and "end" in joined
def test_settle_keeps_transcript_without_markers_intact(monkeypatch):
"""No marker: every previously settled row keeps its exact content."""
session = _session_with_prior_turn()
previous = _settle_turn_with_oob(
session, monkeypatch, tool_content="checks passed\nrevision 41"
)
assert not _contains_oob(session.messages)
assert [m.get("content") for m in session.messages[: len(previous)]] == [
m.get("content") for m in previous
]
assert [m.get("content") for m in session.context_messages[: len(previous)]] == [
m.get("content") for m in previous
]
def test_settle_keeps_prose_that_merely_names_the_marker(monkeypatch):
"""A user row that *talks about* the marker (no brackets) must not be trimmed."""
session = _session_with_prior_turn()
prose = "the docs describe the OUT-OF-BAND USER MESSAGE wrapper used by /steer"
previous = list(session.messages)
previous_context = list(session.context_messages)
monkeypatch.setattr(
_streaming, "_annotate_media_snapshots_for_settled_messages", lambda messages: None
)
result = copy.deepcopy(previous_context) + [
{"role": "user", "content": PROMPT, "timestamp": 1788440000},
{"role": "assistant", "content": prose, "timestamp": 1788440001},
]
_settle_result_messages(session, previous, previous_context, result, PROMPT, "webui", None)
assert prose in json.dumps(session.messages)
assert [m.get("content") for m in session.messages[: len(previous)]] == [
m.get("content") for m in previous
]
def test_settle_keeps_unclosed_marker_as_is(monkeypatch):
"""Unclosed wrappers stay untouched — stripping only complete, consumed blocks.
This is the contract ``_strip_oob_blocks`` already documents (a half-written
marker means the block was never completed), and the settle path must not
invent a different one.
"""
session = _session_with_prior_turn()
_settle_turn_with_oob(
session, monkeypatch, tool_content="checks passed\n[OUT-OF-BAND USER MESSAGE — truncated"
)
assert session.messages[-1]["content"] == "Smoke checks passed."
assert "truncated" in json.dumps(session.messages)
def test_settle_heals_a_transcript_that_already_stored_the_wrapper(monkeypatch):
"""A transcript polluted by an earlier settle is scrubbed on the next one.
The display merge carries earlier rows across turns verbatim, so without a
scrub on the merged transcript the wrapper written before this fix would stay
visible for the rest of the session (and the agent replays it with the
wrapper still in place).
"""
session = _session_with_prior_turn()
# Turn settled before the guard existed: the tool row kept the raw wrapper.
session.messages[2]["content"] = f"deploy finished: revision 41\n{OOB_BLOCK}"
session.context_messages = copy.deepcopy(session.messages)
monkeypatch.setattr(
_streaming, "_annotate_media_snapshots_for_settled_messages", lambda messages: None
)
previous = list(session.messages)
previous_context = list(session.context_messages)
result = copy.deepcopy(previous_context) + [
{"role": "user", "content": PROMPT, "timestamp": 1788440000},
{"role": "assistant", "content": "Nothing else changed.", "timestamp": 1788440001},
]
_settle_result_messages(session, previous, previous_context, result, PROMPT, "webui", None)
assert not _contains_oob(session.messages)
assert not _contains_oob(session.context_messages)
assert sum(1 for m in session.messages if m.get("role") == "tool") == 1
assert "deploy finished: revision 41" in json.dumps(session.messages)
def test_settle_keeps_literal_markers_in_user_text_intact_and_unduplicated(monkeypatch):
"""A complete marker inside user text is content, not transport control data.
The scrub may only touch the tool row the transport appended the wrapper to.
A user row that quotes a *complete* marker — a pasted log excerpt, or the very
prompt asking about this feature — is user-visible content: it must survive
byte-for-byte and exactly once in both written copies. Before the role gate,
the pre-match scrub also shortened the current turn's own prompt row, which
broke active-turn identity matching and duplicated the turn.
"""
def test_settle_keeps_normal_user_pasted_marker_intact(monkeypatch):
"""A normal user row that quotes a complete marker survives byte-for-byte."""
pasted = (
"our bot log shows this block, is that normal?\n"
f"{OOB_BLOCK}\n"
"the docs say the gateway adds it"
)
asked = f"is this wrapper expected?\n{OOB_VARIANT}\nit showed up mid-turn"
session = _session_with_prior_turn()
# Prior turn, already settled: its user row quotes the wrapper as an example.
session.messages[0]["content"] = pasted
session.context_messages = copy.deepcopy(session.messages)
previous = list(session.messages)
previous_context = list(session.context_messages)
monkeypatch.setattr(
_streaming, "_annotate_media_snapshots_for_settled_messages", lambda messages: None
)
ts = 1788440000
token = "direct-stream:2"
identity = {
"token": token,
"text": asked,
"timestamp": float(ts),
"source": "webui",
"attachments": [],
"current_turn_user_idx": len(previous_context),
"turn_id": "turn-2",
"agent_turn_boundary_resolved": True,
steer_row = {
"role": "user",
"display_kind": "steer",
"content": OOB_BLOCK,
"timestamp": 1788440002.5,
}
result = copy.deepcopy(previous_context) + [
{"role": "user", "content": asked, "timestamp": ts, "_active_turn_token": token},
{
"role": "assistant",
"content": "",
"tool_calls": [{"id": "call-2"}],
"timestamp": ts + 1,
},
{
"role": "tool",
"tool_call_id": "call-2",
"content": f"checks passed\n\n{OOB_BLOCK}",
"timestamp": ts + 2,
},
{"role": "assistant", "content": "Smoke checks passed.", "timestamp": ts + 3},
]
_settle_turn_with_steer(session, monkeypatch, steer_rows=[steer_row])
_settle_result_messages(
session, previous, previous_context, result, asked, "webui", identity
)
expected_user_rows = [pasted, asked]
for written in (session.messages, session.context_messages):
user_texts = [
m.get("content")
for m in written
if isinstance(m, dict) and m.get("role") == "user"
]
assert user_texts == expected_user_rows, (
"a literal marker in user text must survive intact and exactly once, "
f"got {user_texts!r}"
)
assert session.messages[-1]["content"] == "Smoke checks passed."
# The consumed steer wrapper on the carrier tool row is still dropped.
tool_bodies = [
m.get("content") for m in session.messages if isinstance(m, dict) and m.get("role") == "tool"
]
assert not _contains_oob(tool_bodies), "wrapper leaked into a tool row"
assert "checks passed" in json.dumps(tool_bodies)
# The normal user row still carries its quoted block intact
assert session.messages[0]["content"] == pasted
# The steer row had its wrapper unwrapped
assert session.messages[-2]["content"] == "steer: use the staging bucket this time"