Merge pull request #73250 from NousResearch/bb/sys-event-rows

Type a resumed interrupted turn as a timeline event, not a user message
This commit is contained in:
brooklyn! 2026-07-28 04:04:44 -05:00 committed by GitHub
commit d2cdb21633
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 269 additions and 10 deletions

View file

@ -1093,6 +1093,8 @@ def run_conversation(
stream_callback: Optional[callable] = None,
persist_user_message: Optional[Any] = None,
persist_user_timestamp: Optional[float] = None,
persist_user_display_kind: Optional[str] = None,
persist_user_display_metadata: Optional[Dict[str, Any]] = None,
moa_config: Optional[dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
@ -1111,6 +1113,13 @@ def run_conversation(
synthetic prefixes.
persist_user_timestamp: Optional platform event timestamp to store
as metadata on that persisted user message.
persist_user_display_kind: Optional presentation type for a
synthesized user turn (``auto_continue``, ``model_switch``, ).
Display-only: transcript surfaces render the row as a timeline
event instead of a user bubble, while the model still receives
the message unchanged.
persist_user_display_metadata: Optional payload for that event
(e.g. a delegation's task count).
or queuing follow-up prefetch work.
Returns:
@ -1153,6 +1162,8 @@ def run_conversation(
stream_callback,
persist_user_message,
persist_user_timestamp,
persist_user_display_kind=persist_user_display_kind,
persist_user_display_metadata=persist_user_display_metadata,
restore_or_build_system_prompt=_restore_or_build_system_prompt,
install_safe_stdio=_install_safe_stdio,
sanitize_surrogates=_sanitize_surrogates,

View file

@ -336,6 +336,8 @@ def build_turn_context(
persist_user_message: Optional[Any],
persist_user_timestamp: Optional[float] = None,
*,
persist_user_display_kind: Optional[str] = None,
persist_user_display_metadata: Optional[Dict[str, Any]] = None,
restore_or_build_system_prompt,
install_safe_stdio,
sanitize_surrogates,
@ -537,6 +539,19 @@ def build_turn_context(
# Add the current user message after the prompt/session setup has made
# close persistence safe. The handoff above preserves any marker already
# stamped by an earlier close flush.
#
# A synthesized turn (auto-continue recovery note, delegation completion)
# declares how it should READ in a transcript. Stamp that on the live
# message so the crash persist below writes the row already typed. Typing
# it after the turn instead leaves the row untyped for the whole run — and
# forever if the turn crashes — so the raw system note paints as a user
# bubble. The model still receives role/content unchanged; the api_messages
# build strips both fields from every outgoing copy.
if persist_user_display_kind:
user_msg["display_kind"] = persist_user_display_kind
if persist_user_display_metadata:
user_msg["display_metadata"] = persist_user_display_metadata
messages.append(user_msg)
current_turn_user_idx = len(messages) - 1
agent._persist_user_message_idx = current_turn_user_idx

View file

@ -494,7 +494,7 @@ export interface SessionMessage {
reasoning?: null | string
reasoning_content?: null | string
reasoning_details?: unknown
display_kind?: 'async_delegation_complete' | 'hidden' | 'model_switch' | string
display_kind?: 'async_delegation_complete' | 'auto_continue' | 'hidden' | 'model_switch' | string
/**
* A backend older than this app can still serve this as unparsed JSON text,
* so readers must narrow before indexing into it.

View file

@ -594,6 +594,9 @@ class CLIAgentSetupMixin:
if display_kind == "async_delegation_complete":
entries.append(("event", "background delegation completed"))
continue
if display_kind == "auto_continue":
entries.append(("event", "resumed interrupted turn"))
continue
if role == "system":
continue

View file

@ -2112,6 +2112,7 @@ class AIAgent:
and not msg.get("_compressed_summary_has_user_turn")
else msg.get("display_kind")
),
display_metadata=msg.get("display_metadata"),
compression_lock_holder=getattr(
self, "_active_compression_lock_holder", None
),
@ -6842,6 +6843,8 @@ class AIAgent:
stream_callback: Optional[callable] = None,
persist_user_message: Optional[Any] = None,
persist_user_timestamp: Optional[float] = None,
persist_user_display_kind: Optional[str] = None,
persist_user_display_metadata: Optional[Dict[str, Any]] = None,
moa_config: Optional[dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Forwarder — see ``agent.conversation_loop.run_conversation``."""
@ -6886,6 +6889,8 @@ class AIAgent:
stream_callback,
persist_user_message,
persist_user_timestamp=persist_user_timestamp,
persist_user_display_kind=persist_user_display_kind,
persist_user_display_metadata=persist_user_display_metadata,
moa_config=moa_config,
)
finally:

View file

@ -0,0 +1,106 @@
"""A synthesized turn's row is typed when it is WRITTEN, not when it ends.
Synthetic user turns (the crash-recovery note, delegation completions) are
model-facing scaffolding that must read as a timeline event, never as a user
bubble. The type used to be stamped only after ``run_conversation`` returned,
which left two holes:
* the row sits untyped for the whole turn, so any surface reading the
transcript mid-turn paints the raw ``[System note: ]`` text as if the user
had typed it;
* a turn killed before it concludes never reaches the stamp at all and the
auto-continue turn is precisely the one that gets killed twice, so its note
stays a user bubble forever.
``persist_user_display_kind`` moves the typing to turn start, where the
crash-resilience persist writes it in the same insert as the content.
"""
from __future__ import annotations
import shutil
import tempfile
import types
from pathlib import Path
from unittest.mock import patch
import pytest
from agent.turn_context import build_turn_context
from hermes_state import SessionDB
from run_agent import AIAgent
NOTE = "[System note: Your previous turn was interrupted mid-run …]\n\nkeep going"
@pytest.fixture()
def agent_db():
tmp = tempfile.mkdtemp(prefix="synthetic_display_kind_")
db = SessionDB(Path(tmp) / "state.db")
sid = "sess-synthetic"
db.create_session(session_id=sid, source="desktop", model="test-model")
agent = AIAgent(
api_key="test-key",
base_url="https://openrouter.ai/api/v1",
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
session_db=db,
session_id=sid,
)
agent._session_db_created = True
agent._cached_system_prompt = "SYSTEM"
agent._skip_mcp_refresh = True
try:
yield agent, db, sid
finally:
db.close()
shutil.rmtree(tmp, ignore_errors=True)
def _build(agent, **overrides):
kwargs = dict(
agent=agent,
user_message=NOTE,
system_message=None,
conversation_history=None,
task_id=None,
stream_callback=None,
persist_user_message=None,
restore_or_build_system_prompt=lambda *a, **k: None,
install_safe_stdio=lambda: None,
sanitize_surrogates=lambda s: s,
summarize_user_message_for_log=lambda s: s,
set_session_context=lambda _sid: None,
set_current_write_origin=lambda _o: None,
ra=lambda: types.SimpleNamespace(_set_interrupt=lambda *a, **k: None),
)
kwargs.update(overrides)
with patch("agent.auxiliary_client.set_runtime_main", lambda *a, **k: None):
return build_turn_context(**kwargs)
def test_row_is_typed_by_the_turn_start_persist(agent_db):
agent, db, sid = agent_db
_build(
agent,
persist_user_display_kind="auto_continue",
persist_user_display_metadata={"attempt": 2},
)
row, = [r for r in db.get_messages_as_conversation(sid) if r["role"] == "user"]
# Typed before the turn ran — a crash from here on still reads as an event.
assert row["display_kind"] == "auto_continue"
assert row["display_metadata"] == {"attempt": 2}
# The model's copy is untouched: same role, same content.
assert row["content"] == NOTE
def test_a_real_user_turn_stays_untyped(agent_db):
agent, db, sid = agent_db
_build(agent, user_message="keep going")
row, = [r for r in db.get_messages_as_conversation(sid) if r["role"] == "user"]
assert row.get("display_kind") is None

View file

@ -1753,6 +1753,29 @@ def test_history_to_messages_drops_display_hidden_scaffolding():
assert all("api_content" not in m for m in projected)
def test_history_to_messages_types_a_legacy_auto_continue_row():
# A crash-interrupted turn used to be typed only AFTER it finished, so a
# turn killed a second time (or any row written before turn-start typing
# landed) sits in the DB untyped and painted the raw recovery note as a
# user bubble. The projection recognizes the synthetic note's fixed
# prefix so those rows still read as a timeline event.
history = [
{"role": "user", "content": "keep going"},
{"role": "user", "content": server._auto_continue_note("keep going")},
]
projected = server._history_to_messages(history)
assert projected == [
{"role": "user", "text": "keep going"},
{
"role": "user",
"text": server._auto_continue_note("keep going"),
"display_kind": "auto_continue",
},
]
def test_history_to_messages_keeps_real_user_bracket_text():
# Only role=user rows whose text OPENS with the [System: marker sentinel are
# bookkeeping notices. A genuine user turn that merely mentions the token is

View file

@ -245,6 +245,62 @@ def test_continuation_turn_records_attempt_and_original_prompt(
assert "_auto_continue_prompt" not in session
def test_continuation_is_typed_at_turn_start(emits, turn_env, marker_home):
"""The recovery note's row must be typed BEFORE the turn runs.
Typing it after run_conversation returns leaves the row a raw user bubble
for the whole turn and permanently if the continuation is itself killed,
which is exactly the scenario it exists for. The model still receives the
note as an ordinary user message.
"""
seen: list = []
def _run(message, *, persist_user_display_kind=None, **kwargs):
seen.append((message, persist_user_display_kind))
return {"final_response": "done"}
agent = types.SimpleNamespace(
session_id="session-key", run_conversation=_run, clear_interrupt=lambda: None
)
note = server._auto_continue_note("the original prompt")
server._run_prompt_submit(
"rid", "sid", _session(agent=agent, running=True), note,
display_kind="auto_continue",
)
assert seen == [(note, "auto_continue")]
def test_older_agent_still_gets_the_post_turn_stamp(emits, turn_env, marker_home):
"""An agent whose run_conversation predates turn-start typing keeps the
original behavior the row is typed once the turn concludes."""
stamped: list = []
class _LegacyDB:
def set_latest_matching_message_display_kind(self, session_id, **kwargs):
stamped.append((session_id, kwargs["display_kind"]))
return True
def _run(message, conversation_history=None, stream_callback=None, **_kwargs):
return {"final_response": "done"}
agent = types.SimpleNamespace(
session_id="session-key",
run_conversation=_run,
clear_interrupt=lambda: None,
_session_db=_LegacyDB(),
)
note = server._auto_continue_note("the original prompt")
server._run_prompt_submit(
"rid", "sid", _session(agent=agent, running=True), note,
display_kind="auto_continue",
)
assert stamped == [("session-key", "auto_continue")]
# ── Scheduling decision ────────────────────────────────────────────────

View file

@ -6237,6 +6237,28 @@ def _is_display_hidden_marker(role: str | None, text: str) -> bool:
return role == "user" and text.lstrip().startswith("[System:")
# Opening of the crash-recovery note synthesized by _auto_continue_note.
# Matched (not just built) so a row persisted before the display type was
# stamped at turn start still reads as a timeline event, and to recognize the
# messaging gateway's twin note.
_AUTO_CONTINUE_NOTE_PREFIX = "[System note: Your previous turn was interrupted mid-run"
def _legacy_display_kind(role: str, text: str) -> str | None:
"""Infer the display type of a synthetic row persisted without one.
Turn-start typing (see ``persist_user_display_kind``) covers everything
written from here on. Sessions already on disk carry untyped rows and a
turn killed mid-run never reached the post-turn stamp at all, which is
exactly the auto-continue case so the raw recovery note would paint as a
user bubble forever. Sniffing the one fixed synthetic prefix is the
migration for those rows; it is not how new rows get typed.
"""
if role == "user" and text.lstrip().startswith(_AUTO_CONTINUE_NOTE_PREFIX):
return "auto_continue"
return None
def _history_to_messages(history: list[dict]) -> list[dict]:
messages = []
tool_call_args = {}
@ -6304,8 +6326,9 @@ def _history_to_messages(history: list[dict]) -> list[dict]:
# Forward display-only timeline metadata so the TUI can render
# model switches and delegation completions as events instead of
# opaque user messages, and hide compaction handoffs entirely.
if m.get("display_kind"):
msg["display_kind"] = m["display_kind"]
display_kind = m.get("display_kind") or _legacy_display_kind(role, content_text)
if display_kind:
msg["display_kind"] = display_kind
if m.get("display_metadata"):
msg["display_metadata"] = m["display_metadata"]
messages.append(msg)
@ -6515,10 +6538,10 @@ def _auto_continue_note(prompt: str) -> str:
# crash persists nothing of the interrupted turn to the session DB — this
# note is the only copy the model will see.
return (
"[System note: Your previous turn was interrupted mid-run — the app or "
"its backend process stopped before the turn could finish. Some of the "
"work may already be complete; check the current state before redoing "
"anything, then finish the task. The interrupted request was:]\n\n"
f"{_AUTO_CONTINUE_NOTE_PREFIX} — the app or its backend process "
"stopped before the turn could finish. Some of the work may already "
"be complete; check the current state before redoing anything, then "
"finish the task. The interrupted request was:]\n\n"
f"{prompt}"
)
@ -11902,11 +11925,21 @@ def _run_prompt_submit(
_build_persist_user_message(prompt, images, run_message) if images else prompt
),
}
# Type a synthesized turn at turn START so the crash persist writes
# its row as a timeline event, instead of leaving a raw user bubble
# until the turn ends — and forever if it never does, which is
# exactly the auto-continue case. The post-turn stamp below is the
# fallback for an older agent without the parameter; re-stamping
# the same value is a no-op.
try:
if "task_id" in inspect.signature(agent.run_conversation).parameters:
run_kwargs["task_id"] = session["session_key"]
_run_params = inspect.signature(agent.run_conversation).parameters
except (TypeError, ValueError):
pass
_run_params = {}
if "task_id" in _run_params:
run_kwargs["task_id"] = session["session_key"]
if display_kind and "persist_user_display_kind" in _run_params:
run_kwargs["persist_user_display_kind"] = display_kind
run_kwargs["persist_user_display_metadata"] = display_metadata
result = agent.run_conversation(run_message, **run_kwargs)
if display_kind and isinstance(text, str):
db = getattr(agent, "_session_db", None)

View file

@ -69,6 +69,13 @@ export const toTranscriptMessages = (rows: unknown): Msg[] => {
continue
}
if (display_kind === 'auto_continue') {
out.push({ kind: 'event', role: 'system', text: 'resumed interrupted turn' })
pending = []
continue
}
if (display_kind === 'async_delegation_complete') {
const meta = (row as TranscriptRow).display_metadata
const count = meta && typeof meta.task_count === 'number' ? meta.task_count : undefined