0dcd558c78
test_prompt_row_id_rewind_preserves_scaffold_before_regeneration asserted that a truncate_before_row_id rewind leaves exactly one active row (the hidden compaction scaffold). That reflected the pre-#111868 timing, where the user turn only reached the DB through the agent's own flush — which the test's stub agent never runs. The previous commit persists the submitted prompt at send time, and on the rewind path that write happens after the truncated transcript is written, so the durable transcript is now "scaffold + EDITED ASK" (the archived carrier and its reply stay inactive). This is the same contract the sibling rewind tests in test_tui_gateway_server.py were already updated to; the source needs no change. The test now checks the scaffold survives the cut AND that the edited prompt is the single active user turn after it, with a row id newer than the cut target.
519 lines
17 KiB
Python
519 lines
17 KiB
Python
"""Regression contracts for TUI rewind of live compaction carriers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from agent.context_compressor import (
|
|
HISTORICAL_TASK_HEADING,
|
|
SUMMARY_PREFIX,
|
|
_SUMMARY_END_MARKER,
|
|
)
|
|
from hermes_state import SessionDB
|
|
from tui_gateway import server
|
|
|
|
|
|
def _composite_carrier() -> dict:
|
|
return {
|
|
"role": "user",
|
|
"content": (
|
|
f"{SUMMARY_PREFIX}\n{HISTORICAL_TASK_HEADING}\nold task\n\n"
|
|
f"{_SUMMARY_END_MARKER}\n\nREAL ASK"
|
|
),
|
|
}
|
|
|
|
|
|
@pytest.fixture()
|
|
def carrier_session(tmp_path):
|
|
old_db = server._db
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
installed_ids: list[str] = []
|
|
|
|
def install(history: list[dict]):
|
|
sid = f"carrier-sid-{len(installed_ids)}"
|
|
session_key = f"carrier-session-{len(installed_ids)}"
|
|
installed_ids.append(sid)
|
|
db.create_session(session_key, source="tui")
|
|
for message in history:
|
|
db.append_message(
|
|
session_key,
|
|
message["role"],
|
|
message.get("content"),
|
|
)
|
|
durable = db.get_messages_as_conversation(session_key)
|
|
agent = SimpleNamespace(
|
|
_session_messages=list(durable),
|
|
_last_flushed_db_idx=len(durable),
|
|
_db_flush_scan_prefix=list(durable),
|
|
)
|
|
session = {
|
|
"agent": agent,
|
|
"attached_images": [],
|
|
"history": list(durable),
|
|
"history_lock": threading.Lock(),
|
|
"history_version": 0,
|
|
"running": False,
|
|
"session_key": session_key,
|
|
}
|
|
server._sessions[sid] = session
|
|
return sid, session_key, session
|
|
|
|
server._db = db
|
|
yield db, install
|
|
for sid in installed_ids:
|
|
server._sessions.pop(sid, None)
|
|
server._db = old_db
|
|
db.close()
|
|
|
|
|
|
def _dispatch(sid: str, name: str) -> dict:
|
|
return server._methods["command.dispatch"](
|
|
"request-id",
|
|
{"session_id": sid, "name": name, "arg": ""},
|
|
)
|
|
|
|
|
|
def _session_undo(sid: str) -> dict:
|
|
return server._methods["session.undo"](
|
|
"request-id",
|
|
{"session_id": sid},
|
|
)
|
|
|
|
|
|
def _assert_scaffold_preserved(
|
|
db: SessionDB,
|
|
session_key: str,
|
|
session: dict,
|
|
*,
|
|
prefix_len: int = 0,
|
|
) -> None:
|
|
active = db.get_messages_as_conversation(session_key, include_row_ids=True)
|
|
assert len(active) == prefix_len + 1
|
|
scaffold = active[prefix_len]
|
|
assert scaffold["role"] == "user"
|
|
assert scaffold["display_kind"] == "hidden"
|
|
assert SUMMARY_PREFIX in scaffold["content"]
|
|
assert "REAL ASK" not in scaffold["content"]
|
|
assert session["history"][prefix_len]["content"] == scaffold["content"]
|
|
assert session["history"][prefix_len]["display_kind"] == "hidden"
|
|
|
|
|
|
def test_retry_selects_the_live_ask_inside_a_force_user_leading_carrier(
|
|
carrier_session,
|
|
):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[_composite_carrier(), {"role": "assistant", "content": "failed"}]
|
|
)
|
|
|
|
response = _dispatch(sid, "retry")
|
|
|
|
assert response["result"] == {"type": "send", "message": "REAL ASK"}
|
|
_assert_scaffold_preserved(db, session_key, session)
|
|
|
|
|
|
@pytest.mark.parametrize("command", ["retry", "undo"])
|
|
def test_rewind_matches_cold_sanitized_carrier_to_unchanged_warm_ask(
|
|
carrier_session, command
|
|
):
|
|
db, install = carrier_session
|
|
carrier = _composite_carrier()
|
|
carrier["content"] = carrier["content"].replace(
|
|
"REAL ASK",
|
|
" REAL ASK\n\n<memory-context>\nprivate\n</memory-context> ",
|
|
)
|
|
sid, session_key, session = install(
|
|
[carrier, {"role": "assistant", "content": "failed"}]
|
|
)
|
|
db._conn.execute(
|
|
"UPDATE messages SET api_content = ? "
|
|
"WHERE session_id = ? AND role = 'user'",
|
|
(carrier["content"], session_key),
|
|
)
|
|
db._conn.commit()
|
|
# A live agent still has the raw wire form while a cold DB projection has
|
|
# already applied the role-aware sanitize_context(...).strip() rule and
|
|
# retained the raw provider wire in api_content.
|
|
session["history"][0] = carrier.copy()
|
|
session["agent"]._session_messages = list(session["history"])
|
|
|
|
response = _dispatch(sid, command)
|
|
|
|
assert response["result"]["message"] == "REAL ASK"
|
|
_assert_scaffold_preserved(db, session_key, session)
|
|
|
|
|
|
def test_retry_fails_closed_when_transcript_changes_after_snapshot(
|
|
carrier_session, monkeypatch
|
|
):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[_composite_carrier(), {"role": "assistant", "content": "failed"}]
|
|
)
|
|
sibling = SessionDB(db_path=db.db_path)
|
|
original_rewind = db.rewind_to_message
|
|
|
|
def _append_then_rewind(*args, **kwargs):
|
|
sibling.append_message(session_key, "assistant", "concurrent tail")
|
|
return original_rewind(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(db, "rewind_to_message", _append_then_rewind)
|
|
before_history = [dict(message) for message in session["history"]]
|
|
|
|
response = _dispatch(sid, "retry")
|
|
|
|
assert response["error"]["code"] == 5008
|
|
assert "active transcript changed" in response["error"]["message"]
|
|
assert session["history"] == before_history
|
|
rows = db._conn.execute(
|
|
"SELECT content, active, display_kind FROM messages "
|
|
"WHERE session_id = ? ORDER BY id",
|
|
(session_key,),
|
|
).fetchall()
|
|
assert [tuple(row) for row in rows] == [
|
|
(_composite_carrier()["content"], 1, None),
|
|
("failed", 1, None),
|
|
("concurrent tail", 1, None),
|
|
]
|
|
sibling.close()
|
|
|
|
|
|
@pytest.mark.parametrize("command", ["retry", "undo"])
|
|
def test_rewind_allows_database_only_reaction_metadata_change(
|
|
carrier_session, command
|
|
):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[
|
|
{"role": "user", "content": "OLDER ASK"},
|
|
{"role": "assistant", "content": "older answer"},
|
|
_composite_carrier(),
|
|
{"role": "assistant", "content": "failed"},
|
|
]
|
|
)
|
|
older_answer = next(
|
|
row
|
|
for row in db.get_messages(session_key)
|
|
if row["role"] == "assistant" and row["content"] == "older answer"
|
|
)
|
|
assert db.set_message_reaction(
|
|
session_key, older_answer["id"], "👍", author="user"
|
|
)
|
|
|
|
response = _dispatch(sid, command)
|
|
|
|
assert response["result"]["message"] == "REAL ASK"
|
|
_assert_scaffold_preserved(db, session_key, session, prefix_len=2)
|
|
|
|
|
|
def test_retry_ignores_buried_ephemeral_scaffolding_missing_from_db(
|
|
carrier_session,
|
|
):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[
|
|
{"role": "user", "content": "OLDER ASK"},
|
|
{"role": "assistant", "content": "older answer"},
|
|
_composite_carrier(),
|
|
{"role": "assistant", "content": "failed"},
|
|
]
|
|
)
|
|
session["history"].insert(
|
|
2,
|
|
{
|
|
"role": "user",
|
|
"content": "internal recovery nudge",
|
|
"_dropped_toolcall_nudge": True,
|
|
},
|
|
)
|
|
session["agent"]._session_messages = list(session["history"])
|
|
|
|
response = _dispatch(sid, "retry")
|
|
|
|
assert response["result"] == {"type": "send", "message": "REAL ASK"}
|
|
_assert_scaffold_preserved(db, session_key, session, prefix_len=2)
|
|
|
|
|
|
def test_retry_drops_buried_ephemeral_scaffolding_from_the_warm_prefix(
|
|
carrier_session,
|
|
):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[
|
|
{"role": "user", "content": "OLDER ASK"},
|
|
{"role": "assistant", "content": "candidate answer"},
|
|
{"role": "assistant", "content": "verified answer"},
|
|
_composite_carrier(),
|
|
{"role": "assistant", "content": "failed"},
|
|
]
|
|
)
|
|
session["history"].insert(
|
|
2,
|
|
{
|
|
"role": "user",
|
|
"content": "[System: verify before stopping]",
|
|
"_verification_stop_synthetic": True,
|
|
},
|
|
)
|
|
session["agent"]._session_messages = list(session["history"])
|
|
|
|
response = _dispatch(sid, "retry")
|
|
|
|
assert response["result"] == {"type": "send", "message": "REAL ASK"}
|
|
assert [message.get("content") for message in session["history"][:3]] == [
|
|
"OLDER ASK",
|
|
"candidate answer",
|
|
"verified answer",
|
|
]
|
|
active = db.get_messages_as_conversation(session_key, include_row_ids=True)
|
|
# Alternation repair is a model/memory projection; the two durable source
|
|
# rows remain independently recoverable ahead of the inserted scaffold.
|
|
assert [message.get("content") for message in active[:3]] == [
|
|
"OLDER ASK",
|
|
"candidate answer",
|
|
"verified answer",
|
|
]
|
|
assert active[3]["display_kind"] == "hidden"
|
|
assert "REAL ASK" not in active[3]["content"]
|
|
assert session["history"][3]["display_kind"] == "hidden"
|
|
assert "REAL ASK" not in session["history"][3]["content"]
|
|
|
|
|
|
def test_retry_preserves_older_warm_media_while_targeting_plain_ask(
|
|
carrier_session,
|
|
):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[
|
|
{"role": "user", "content": "look\n[screenshot]"},
|
|
{"role": "assistant", "content": "seen"},
|
|
_composite_carrier(),
|
|
{"role": "assistant", "content": "failed"},
|
|
]
|
|
)
|
|
session["history"][0]["content"] = [
|
|
{"type": "text", "text": "look"},
|
|
{"type": "image_url", "image_url": {"url": "data:image/png;base64,x"}},
|
|
]
|
|
session["agent"]._session_messages = list(session["history"])
|
|
|
|
response = _dispatch(sid, "retry")
|
|
|
|
assert response["result"] == {"type": "send", "message": "REAL ASK"}
|
|
assert isinstance(session["history"][0]["content"], list)
|
|
_assert_scaffold_preserved(db, session_key, session, prefix_len=2)
|
|
|
|
|
|
def test_undo_targets_the_composite_carrier_not_an_older_user_turn(
|
|
carrier_session,
|
|
):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[
|
|
{"role": "user", "content": "OLDER ASK"},
|
|
{"role": "assistant", "content": "older answer"},
|
|
_composite_carrier(),
|
|
{"role": "assistant", "content": "failed"},
|
|
]
|
|
)
|
|
|
|
response = _dispatch(sid, "undo")
|
|
|
|
assert response["result"]["type"] == "prefill"
|
|
assert response["result"]["message"] == "REAL ASK"
|
|
active = db.get_messages_as_conversation(session_key, include_row_ids=True)
|
|
assert [message.get("content") for message in active[:2]] == [
|
|
"OLDER ASK",
|
|
"older answer",
|
|
]
|
|
_assert_scaffold_preserved(db, session_key, session, prefix_len=2)
|
|
|
|
|
|
def test_undo_rewinds_media_placeholder_without_treating_it_as_retry(
|
|
carrier_session,
|
|
):
|
|
db, install = carrier_session
|
|
carrier = _composite_carrier()
|
|
carrier["content"] = carrier["content"].replace(
|
|
"REAL ASK", "look\n[screenshot]"
|
|
)
|
|
sid, session_key, session = install(
|
|
[carrier, {"role": "assistant", "content": "seen"}]
|
|
)
|
|
|
|
response = _dispatch(sid, "undo")
|
|
|
|
assert response["result"]["type"] == "prefill"
|
|
assert response["result"]["message"] == "look\n[screenshot]"
|
|
active = db.get_messages_as_conversation(session_key)
|
|
assert len(active) == 1
|
|
assert active[0]["display_kind"] == "hidden"
|
|
assert "look\n[screenshot]" not in active[0]["content"]
|
|
assert len(session["history"]) == 1
|
|
assert session["history"][0]["content"] == active[0]["content"]
|
|
assert session["history"][0]["display_kind"] == "hidden"
|
|
|
|
|
|
def test_session_undo_preserves_the_composite_carriers_scaffold(carrier_session):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[_composite_carrier(), {"role": "assistant", "content": "answer"}]
|
|
)
|
|
|
|
response = _session_undo(sid)
|
|
|
|
assert response["result"]["removed"] == 2
|
|
_assert_scaffold_preserved(db, session_key, session)
|
|
|
|
|
|
def test_history_projection_unwraps_composite_and_hides_sole_handoff():
|
|
composite = {**_composite_carrier(), "_row_id": 7}
|
|
sole_handoff = {
|
|
**composite,
|
|
"content": composite["content"].split("\n\nREAL ASK", 1)[0],
|
|
}
|
|
|
|
assert server._history_to_messages(
|
|
[composite, {"role": "user", "content": "newer ask", "_row_id": 9}]
|
|
) == [
|
|
{"role": "user", "text": "REAL ASK", "row_id": 7},
|
|
{"role": "user", "text": "newer ask", "row_id": 9},
|
|
]
|
|
assert server._history_to_messages([sole_handoff]) == []
|
|
|
|
|
|
def test_retry_preserves_literal_media_like_text(carrier_session):
|
|
db, install = carrier_session
|
|
carrier = _composite_carrier()
|
|
carrier["content"] = carrier["content"].replace(
|
|
"REAL ASK", "inspect [image|ybres:RID]"
|
|
)
|
|
sid, session_key, session = install(
|
|
[carrier, {"role": "assistant", "content": "failed"}]
|
|
)
|
|
response = _dispatch(sid, "retry")
|
|
|
|
assert response["result"] == {
|
|
"type": "send",
|
|
"message": "inspect [image|ybres:RID]",
|
|
}
|
|
_assert_scaffold_preserved(db, session_key, session)
|
|
|
|
|
|
def test_retry_rejects_durable_media_before_rewind_when_warm_view_is_text(
|
|
carrier_session,
|
|
):
|
|
db, install = carrier_session
|
|
carrier = _composite_carrier()
|
|
handoff = carrier["content"].rsplit("\n\nREAL ASK", 1)[0]
|
|
durable_carrier = carrier.copy()
|
|
durable_carrier["content"] = [
|
|
{"type": "text", "text": handoff},
|
|
{"type": "image_url", "image_url": {"url": "data:image/png;base64,x"}},
|
|
]
|
|
sid, session_key, session = install(
|
|
[durable_carrier, {"role": "assistant", "content": "failed"}]
|
|
)
|
|
# The warm projection can be a degraded text-only view that compares equal
|
|
# to the durable media payload. Durable retryability must still be checked
|
|
# before the physical carrier and tail are archived.
|
|
warm_carrier = carrier.copy()
|
|
warm_carrier["content"] = handoff + "\n\n[screenshot]"
|
|
session["history"][0] = warm_carrier
|
|
session["agent"]._session_messages = list(session["history"])
|
|
before_history = [message.copy() for message in session["history"]]
|
|
|
|
response = _dispatch(sid, "retry")
|
|
|
|
assert response["error"]["code"] == 4018
|
|
assert session["history"] == before_history
|
|
assert len(db.get_messages_as_conversation(session_key)) == 2
|
|
|
|
|
|
def test_retry_rejects_pending_attachments_before_mutating_history(carrier_session):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[_composite_carrier(), {"role": "assistant", "content": "failed"}]
|
|
)
|
|
session["attached_images"] = ["/tmp/pending.png"]
|
|
before_memory = list(session["history"])
|
|
|
|
response = _dispatch(sid, "retry")
|
|
|
|
assert response["error"]["code"] == 4018
|
|
assert session["history"] == before_memory
|
|
assert len(db.get_messages_as_conversation(session_key)) == 2
|
|
|
|
def test_prompt_row_id_rewind_preserves_scaffold_before_regeneration(
|
|
carrier_session, monkeypatch
|
|
):
|
|
db, install = carrier_session
|
|
sid, session_key, session = install(
|
|
[_composite_carrier(), {"role": "assistant", "content": "failed"}]
|
|
)
|
|
target_row_id = db.get_messages_as_conversation(
|
|
session_key, include_row_ids=True
|
|
)[0]["_row_id"]
|
|
seen = {}
|
|
|
|
class _Agent:
|
|
_session_messages = list(session["history"])
|
|
_last_flushed_db_idx = len(_session_messages)
|
|
_db_flush_scan_prefix = list(_session_messages)
|
|
|
|
def run_conversation(
|
|
self, prompt, conversation_history=None, stream_callback=None, **_kwargs
|
|
):
|
|
seen["prompt"] = prompt
|
|
seen["history"] = list(conversation_history or [])
|
|
return {
|
|
"final_response": "regenerated",
|
|
"messages": [
|
|
*(conversation_history or []),
|
|
{"role": "user", "content": prompt},
|
|
{"role": "assistant", "content": "regenerated"},
|
|
],
|
|
}
|
|
|
|
class _ImmediateThread:
|
|
def __init__(self, target=None, daemon=None, args=(), kwargs=None, name=None):
|
|
self._target = lambda: target(*args, **(kwargs or {}))
|
|
|
|
def start(self):
|
|
self._target()
|
|
|
|
session["agent"] = _Agent()
|
|
monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
|
|
monkeypatch.setattr(server, "_get_usage", lambda _agent: {})
|
|
monkeypatch.setattr(server, "render_message", lambda *_args: "")
|
|
monkeypatch.setattr(server, "_emit", lambda *_args: None)
|
|
|
|
response = server._methods["prompt.submit"](
|
|
"request-id",
|
|
{
|
|
"session_id": sid,
|
|
"text": "EDITED ASK",
|
|
"truncate_before_row_id": target_row_id,
|
|
"truncate_before_user_ordinal": 0,
|
|
"confirm_truncate": True,
|
|
},
|
|
)
|
|
|
|
assert response["result"]["status"] == "streaming"
|
|
assert seen["prompt"] == "EDITED ASK"
|
|
assert len(seen["history"]) == 1
|
|
assert seen["history"][0]["display_kind"] == "hidden"
|
|
assert "REAL ASK" not in seen["history"][0]["content"]
|
|
active = db.get_messages_as_conversation(session_key, include_row_ids=True)
|
|
# The cut keeps only the scaffold; the edited prompt is then durable at submit (#111868),
|
|
# written AFTER the truncation, so it is the single active user turn after the scaffold.
|
|
assert [(m.get("display_kind"), m["content"]) for m in active] == [
|
|
("hidden", active[0]["content"]),
|
|
(None, "EDITED ASK"),
|
|
]
|
|
assert "REAL ASK" not in active[0]["content"]
|
|
assert active[1]["_row_id"] > target_row_id
|