starship-s
0be6a1f454
Record pending-turn proof under caller-owned session locks during normal runs and credential recovery. Preserve both conflicting provider payloads for replay, but display one sidecar-owned bubble when row identity and visible content are unambiguous. Assisted-by: Hermes Agent:gpt-6-sol Assisted-by: Codex:gpt-6-luna
1504 lines
53 KiB
Python
1504 lines
53 KiB
Python
"""Durable transcript revision coverage for WebUI compression handoff."""
|
|
|
|
import queue
|
|
import sqlite3
|
|
import sys
|
|
import types
|
|
from collections import OrderedDict
|
|
|
|
import pytest
|
|
|
|
from api import models, streaming
|
|
from api.models import Session
|
|
|
|
pytestmark = pytest.mark.requires_agent_modules
|
|
|
|
|
|
@pytest.mark.parametrize("revision", [None, {"session_id": "session-1"}])
|
|
def test_run_conversation_revision_kwarg_is_omitted_for_legacy_agent(revision):
|
|
class LegacyAgent:
|
|
def run_conversation(
|
|
self,
|
|
user_message,
|
|
system_message,
|
|
conversation_history,
|
|
task_id,
|
|
persist_user_message,
|
|
):
|
|
return None
|
|
|
|
kwargs = {}
|
|
supported = streaming._add_supported_run_conversation_kwarg(
|
|
LegacyAgent().run_conversation,
|
|
kwargs,
|
|
"conversation_history_revision",
|
|
revision,
|
|
)
|
|
|
|
assert supported is False
|
|
assert "conversation_history_revision" not in kwargs
|
|
|
|
|
|
@pytest.mark.parametrize("revision", [None, {"session_id": "session-1"}])
|
|
@pytest.mark.parametrize("mode", ["explicit", "variadic"])
|
|
def test_run_conversation_revision_kwarg_is_added_for_capable_agent(revision, mode):
|
|
class ExplicitAgent:
|
|
def run_conversation(self, *, conversation_history_revision=None):
|
|
return None
|
|
|
|
class VariadicAgent:
|
|
def run_conversation(self, **kwargs):
|
|
return None
|
|
|
|
agent = ExplicitAgent() if mode == "explicit" else VariadicAgent()
|
|
kwargs = {}
|
|
supported = streaming._add_supported_run_conversation_kwarg(
|
|
agent.run_conversation,
|
|
kwargs,
|
|
"conversation_history_revision",
|
|
revision,
|
|
)
|
|
|
|
assert supported is True
|
|
assert kwargs["conversation_history_revision"] is revision
|
|
|
|
|
|
def test_build_run_conversation_kwargs_omits_revision_for_strict_legacy_agent():
|
|
class LegacyAgent:
|
|
def run_conversation(
|
|
self,
|
|
user_message,
|
|
system_message,
|
|
conversation_history,
|
|
task_id,
|
|
persist_user_message,
|
|
persist_user_timestamp,
|
|
):
|
|
return None
|
|
|
|
agent = LegacyAgent()
|
|
kwargs = streaming._build_run_conversation_kwargs(
|
|
agent.run_conversation,
|
|
user_message="hello",
|
|
system_message="system",
|
|
conversation_history=[{"role": "user", "content": "prior"}],
|
|
conversation_history_revision={"session_id": "session-1"},
|
|
task_id="session-1",
|
|
persist_user_message="hello",
|
|
persist_user_timestamp=123.0,
|
|
)
|
|
|
|
assert kwargs == {
|
|
"user_message": "hello",
|
|
"system_message": "system",
|
|
"conversation_history": [{"role": "user", "content": "prior"}],
|
|
"task_id": "session-1",
|
|
"persist_user_message": "hello",
|
|
"persist_user_timestamp": 123.0,
|
|
}
|
|
|
|
|
|
def _make_state_db(path, sid, rows):
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_id TEXT NOT NULL,
|
|
role TEXT,
|
|
content TEXT,
|
|
timestamp REAL,
|
|
active INTEGER
|
|
)
|
|
"""
|
|
)
|
|
conn.executemany(
|
|
"INSERT INTO messages (session_id, role, content, timestamp, active) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
[
|
|
(
|
|
sid,
|
|
row["role"],
|
|
row["content"],
|
|
row.get("timestamp"),
|
|
row.get("active", 1),
|
|
)
|
|
for row in rows
|
|
],
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _append_state_row(path, sid, *, role, content, timestamp, active=1):
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO messages (session_id, role, content, timestamp, active) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
(sid, role, content, timestamp, active),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _install_streaming_session(monkeypatch, tmp_path, *, sid, stream_id, messages, context_messages):
|
|
import api.config as config
|
|
|
|
session_dir = tmp_path / "sessions"
|
|
session_dir.mkdir(exist_ok=True)
|
|
index_file = session_dir / "_index.json"
|
|
|
|
monkeypatch.setattr(models, "SESSION_DIR", session_dir)
|
|
monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file)
|
|
monkeypatch.setattr(models, "SESSIONS", OrderedDict(), raising=False)
|
|
monkeypatch.setattr(config, "SESSION_DIR", session_dir, raising=False)
|
|
monkeypatch.setattr(config, "SESSION_INDEX_FILE", index_file, raising=False)
|
|
monkeypatch.setattr(streaming, "SESSION_DIR", session_dir, raising=False)
|
|
monkeypatch.setattr(streaming, "SESSIONS", OrderedDict(), raising=False)
|
|
monkeypatch.setattr(models, "_active_state_db_path", lambda: tmp_path / "state.db")
|
|
|
|
config.STREAMS.clear()
|
|
config.CANCEL_FLAGS.clear()
|
|
config.AGENT_INSTANCES.clear()
|
|
config.SESSION_AGENT_LOCKS.clear()
|
|
|
|
session = Session(
|
|
session_id=sid,
|
|
title="Durable revision",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
messages=list(messages),
|
|
context_messages=list(context_messages),
|
|
)
|
|
session.active_stream_id = stream_id
|
|
session.pending_user_message = "new webui turn"
|
|
session.pending_started_at = 10.0
|
|
session.save(touch_updated_at=False)
|
|
models.SESSIONS[sid] = session
|
|
streaming.SESSIONS[sid] = session
|
|
|
|
event_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = event_queue
|
|
streaming.STREAMS[stream_id] = event_queue
|
|
|
|
monkeypatch.setattr(streaming, "get_session", lambda _sid: session)
|
|
monkeypatch.setattr(
|
|
streaming,
|
|
"resolve_model_provider",
|
|
lambda *_args, **_kwargs: ("test-model", "test-provider", None),
|
|
)
|
|
monkeypatch.setattr(streaming, "get_config", lambda: {})
|
|
monkeypatch.setattr(config, "get_config", lambda: {})
|
|
monkeypatch.setattr(config, "_resolve_cli_toolsets", lambda *_args, **_kwargs: [])
|
|
|
|
fake_hermes_state = types.ModuleType("hermes_state")
|
|
fake_hermes_state.__dict__["SessionDB"] = lambda *_args, **_kwargs: object()
|
|
monkeypatch.setitem(sys.modules, "hermes_state", fake_hermes_state)
|
|
|
|
return session, event_queue
|
|
|
|
|
|
def _drain_events(event_queue):
|
|
events = []
|
|
while not event_queue.empty():
|
|
events.append(event_queue.get_nowait())
|
|
return events
|
|
|
|
|
|
def test_state_db_reader_returns_active_messages_with_matching_revision(tmp_path, monkeypatch):
|
|
sid = "revision-reader"
|
|
db_path = tmp_path / "state.db"
|
|
_make_state_db(
|
|
db_path,
|
|
sid,
|
|
[
|
|
{"role": "user", "content": "inactive", "timestamp": 1.0, "active": 0},
|
|
{"role": "user", "content": "active user", "timestamp": 2.0},
|
|
{"role": "assistant", "content": "active answer", "timestamp": 3.0},
|
|
],
|
|
)
|
|
monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
|
|
|
|
snapshot = models.get_state_db_session_messages(sid, with_revision=True)
|
|
|
|
assert [message["content"] for message in snapshot.messages] == [
|
|
"active user",
|
|
"active answer",
|
|
]
|
|
assert snapshot.revision == {
|
|
"session_id": sid,
|
|
"active_message_count": 2,
|
|
"max_active_message_id": 3,
|
|
}
|
|
assert isinstance(models.get_state_db_session_messages(sid), list)
|
|
|
|
|
|
def test_state_db_reader_revision_includes_api_content_digest(
|
|
tmp_path, monkeypatch
|
|
):
|
|
"""The WebUI revision must carry the Agent's model-facing sidecar fence."""
|
|
sid = "revision-api-content-digest"
|
|
db_path = tmp_path / "state.db"
|
|
conn = sqlite3.connect(db_path)
|
|
try:
|
|
conn.execute(
|
|
"CREATE TABLE messages ("
|
|
"id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, "
|
|
"role TEXT, content TEXT, timestamp REAL, active INTEGER, "
|
|
"api_content TEXT)"
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO messages "
|
|
"(session_id, role, content, timestamp, active, api_content) "
|
|
"VALUES (?, ?, ?, ?, ?, ?)",
|
|
(sid, "user", "clean", 1.0, 1, "wire-v1"),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
|
|
|
|
first = models.get_state_db_session_messages(sid, with_revision=True)
|
|
assert first.revision == {
|
|
"session_id": sid,
|
|
"active_message_count": 1,
|
|
"max_active_message_id": 1,
|
|
"active_rows_digest": (
|
|
"50101e75ded18d78bc21646a6fcd4cddb8f07901838635dda6eac85d219c1359"
|
|
),
|
|
}
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
try:
|
|
conn.execute(
|
|
"UPDATE messages SET api_content = ? WHERE session_id = ?",
|
|
("wire-v2", sid),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
second = models.get_state_db_session_messages(sid, with_revision=True)
|
|
assert second.revision == {
|
|
"session_id": sid,
|
|
"active_message_count": 1,
|
|
"max_active_message_id": 1,
|
|
"active_rows_digest": (
|
|
"de45994b9f134be94b99ce03a47da3b553c86a27e5cdcfd67885644a8372102a"
|
|
),
|
|
}
|
|
|
|
|
|
def test_state_db_revision_is_unavailable_when_selected_row_has_null_active(
|
|
tmp_path, monkeypatch
|
|
):
|
|
sid = "revision-null-active"
|
|
db_path = tmp_path / "state.db"
|
|
_make_state_db(
|
|
db_path,
|
|
sid,
|
|
[{"role": "user", "content": "legacy active row", "timestamp": 1.0, "active": None}],
|
|
)
|
|
monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
|
|
|
|
snapshot = models.get_state_db_session_messages(sid, with_revision=True)
|
|
|
|
assert [message["content"] for message in snapshot.messages] == ["legacy active row"]
|
|
assert snapshot.revision is None
|
|
|
|
|
|
def test_state_db_revision_is_unavailable_without_active_column(tmp_path, monkeypatch):
|
|
sid = "revision-no-active-column"
|
|
db_path = tmp_path / "state.db"
|
|
conn = sqlite3.connect(db_path)
|
|
try:
|
|
conn.execute(
|
|
"CREATE TABLE messages ("
|
|
"id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, "
|
|
"role TEXT, content TEXT, timestamp REAL)"
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
|
|
(sid, "user", "legacy schema row", 1.0),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
|
|
|
|
snapshot = models.get_state_db_session_messages(sid, with_revision=True)
|
|
|
|
assert [message["content"] for message in snapshot.messages] == ["legacy schema row"]
|
|
assert snapshot.revision is None
|
|
|
|
|
|
def test_snapshot_reader_missing_explicit_profile_does_not_fall_back_to_active_db(
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
sid = "missing-profile-snapshot"
|
|
foreign_db = tmp_path / "active-home" / "state.db"
|
|
foreign_db.parent.mkdir()
|
|
_make_state_db(
|
|
foreign_db,
|
|
sid,
|
|
[{"role": "user", "content": "foreign profile row", "timestamp": 1.0}],
|
|
)
|
|
missing_home = tmp_path / "missing-profile"
|
|
monkeypatch.setattr(models, "_get_profile_home", lambda _profile: missing_home)
|
|
monkeypatch.setattr(models, "_active_state_db_path", lambda: foreign_db)
|
|
|
|
snapshot = models.get_state_db_session_messages(
|
|
sid,
|
|
profile="missing",
|
|
with_revision=True,
|
|
)
|
|
|
|
assert snapshot.messages == []
|
|
assert snapshot.revision is None
|
|
|
|
|
|
def test_reconciled_reader_missing_explicit_profile_does_not_fall_back_to_active_db(
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
sid = "missing-profile-reconcile"
|
|
foreign_db = tmp_path / "active-home" / "state.db"
|
|
foreign_db.parent.mkdir()
|
|
_make_state_db(
|
|
foreign_db,
|
|
sid,
|
|
[{"role": "user", "content": "foreign profile row", "timestamp": 1.0}],
|
|
)
|
|
missing_home = tmp_path / "missing-profile"
|
|
monkeypatch.setattr(models, "_get_profile_home", lambda _profile: missing_home)
|
|
monkeypatch.setattr(models, "_active_state_db_path", lambda: foreign_db)
|
|
session = Session(
|
|
session_id=sid,
|
|
profile="missing",
|
|
messages=[],
|
|
context_messages=[],
|
|
)
|
|
|
|
reconciled = models.reconciled_state_db_messages_for_session(
|
|
session,
|
|
with_revision=True,
|
|
)
|
|
|
|
assert reconciled.messages == []
|
|
assert reconciled.revision is None
|
|
|
|
|
|
def test_stale_partial_requires_authoritative_current_turn_boundary():
|
|
session = Session(
|
|
session_id="partial-authority",
|
|
messages=[],
|
|
context_messages=[],
|
|
)
|
|
result = {
|
|
"partial": True,
|
|
"messages": [
|
|
{"role": "user", "content": "repeat prompt"},
|
|
{"role": "assistant", "content": "historical answer"},
|
|
{"role": "user", "content": "later unrelated turn"},
|
|
{"role": "assistant", "content": "later answer"},
|
|
]
|
|
}
|
|
identity = {
|
|
"token": "stream:123",
|
|
"text": "repeat prompt",
|
|
# This index belonged to the replaced projection but now happens to
|
|
# point at an identical historical prompt. It is not current-turn proof.
|
|
"current_turn_user_idx": 0,
|
|
"turn_id": "session:turn-current",
|
|
}
|
|
|
|
streaming._append_result_partial_on_error(
|
|
session,
|
|
result,
|
|
[{"role": "user", "content": "unrelated baseline"}],
|
|
"repeat prompt",
|
|
active_turn_identity=identity,
|
|
)
|
|
|
|
assert not any(message.get("_partial") for message in session.messages)
|
|
|
|
|
|
def test_stale_non_prefix_partial_uses_token_and_stops_at_next_user():
|
|
session = Session(
|
|
session_id="stale-partial-token-boundary",
|
|
messages=[],
|
|
context_messages=[],
|
|
)
|
|
result = {
|
|
"partial": True,
|
|
"messages": [
|
|
{"role": "user", "content": "historical"},
|
|
{"role": "assistant", "content": "historical answer"},
|
|
{
|
|
"role": "user",
|
|
"content": "current prompt",
|
|
"_active_turn_token": "current-turn-token",
|
|
},
|
|
{"role": "assistant", "content": "current partial"},
|
|
{"role": "user", "content": "later turn"},
|
|
{"role": "assistant", "content": "later answer"},
|
|
],
|
|
}
|
|
|
|
appended = streaming._append_result_partial_on_error(
|
|
session,
|
|
result,
|
|
[{"role": "user", "content": "different baseline"}],
|
|
"current prompt",
|
|
active_turn_identity={
|
|
"token": "current-turn-token",
|
|
"turn_id": "turn-current",
|
|
"current_turn_user_idx": 2,
|
|
"text": "current prompt",
|
|
},
|
|
)
|
|
|
|
assert appended is not None
|
|
assert appended["content"] == "current partial"
|
|
assert [message["content"] for message in session.messages] == [
|
|
"current partial"
|
|
]
|
|
|
|
|
|
def test_non_prefix_partial_uses_agent_turn_boundary_without_webui_token():
|
|
session = Session(session_id="agent-boundary-partial", messages=[], context_messages=[])
|
|
result = {
|
|
"partial": True,
|
|
"turn_id": "agent-turn-42",
|
|
"current_turn_user_idx": 2,
|
|
"messages": [
|
|
{"role": "user", "content": "repeat prompt"},
|
|
{"role": "assistant", "content": "historical answer"},
|
|
{"role": "user", "content": " repeat prompt "},
|
|
{"role": "assistant", "content": "current partial"},
|
|
{"role": "user", "content": "later turn"},
|
|
{"role": "assistant", "content": "later answer"},
|
|
],
|
|
}
|
|
identity = streaming._resolve_active_turn_authority(
|
|
{"token": "webui-token", "text": "repeat prompt", "turn_id": "", "current_turn_user_idx": None},
|
|
result=result,
|
|
)
|
|
|
|
appended = streaming._append_result_partial_on_error(
|
|
session, result, [{"role": "user", "content": "replaced baseline"}],
|
|
"repeat prompt", active_turn_identity=identity,
|
|
)
|
|
|
|
assert appended["content"] == "current partial"
|
|
|
|
|
|
@pytest.mark.parametrize("terminal", ["returned", "raised"])
|
|
def test_non_prefix_self_heal_accepts_agent_turn_boundary_without_webui_token(terminal):
|
|
result = {
|
|
"completed": True,
|
|
"turn_id": "agent-turn-43",
|
|
"current_turn_user_idx": 2,
|
|
"messages": [
|
|
{"role": "user", "content": "repeat prompt"},
|
|
{"role": "assistant", "content": "historical answer"},
|
|
{"role": "user", "content": " repeat\n prompt "},
|
|
{"role": "assistant", "content": f"healed after {terminal}"},
|
|
{"role": "user", "content": "later turn"},
|
|
{"role": "assistant", "content": "later answer"},
|
|
],
|
|
}
|
|
identity = streaming._resolve_active_turn_authority(
|
|
{"token": "webui-token", "text": "repeat prompt", "turn_id": "", "current_turn_user_idx": None},
|
|
result=result,
|
|
)
|
|
|
|
assert streaming._self_heal_result_succeeded(
|
|
result, [{"role": "user", "content": "replaced baseline"}], identity, "repeat prompt"
|
|
)
|
|
|
|
|
|
def test_non_prefix_agent_boundary_rejects_wrong_normalized_user():
|
|
result = {
|
|
"completed": True,
|
|
"turn_id": "agent-turn-stale",
|
|
"current_turn_user_idx": 0,
|
|
"messages": [
|
|
{"role": "user", "content": "different prompt"},
|
|
{"role": "assistant", "content": "historical answer"},
|
|
],
|
|
}
|
|
identity = streaming._resolve_active_turn_authority(
|
|
{"token": "absent", "text": "repeat prompt", "turn_id": "", "current_turn_user_idx": None},
|
|
result=result,
|
|
)
|
|
|
|
assert not streaming._self_heal_result_succeeded(
|
|
result, [{"role": "user", "content": "baseline"}], identity, "repeat prompt"
|
|
)
|
|
|
|
|
|
def test_result_partial_does_not_mutate_identical_prior_turn():
|
|
identity = {"token": "current-turn-token"}
|
|
session = Session(
|
|
session_id="result-partial-identical-content",
|
|
messages=[
|
|
{"role": "user", "content": "historical"},
|
|
{"role": "assistant", "content": "same answer"},
|
|
{
|
|
"role": "user",
|
|
"content": "current prompt",
|
|
"_active_turn_token": identity["token"],
|
|
},
|
|
],
|
|
context_messages=[],
|
|
)
|
|
|
|
appended = streaming._append_result_partial_on_error(
|
|
session,
|
|
{
|
|
"partial": True,
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": "current prompt",
|
|
"_active_turn_token": identity["token"],
|
|
},
|
|
{"role": "assistant", "content": "same answer"},
|
|
],
|
|
},
|
|
[{"role": "user", "content": "different baseline"}],
|
|
"current prompt",
|
|
active_turn_identity=identity,
|
|
)
|
|
|
|
assert appended is session.messages[-1]
|
|
assert appended["content"] == "same answer"
|
|
assert appended["_partial"] is True
|
|
assert session.messages[1] == {"role": "assistant", "content": "same answer"}
|
|
|
|
|
|
def test_stream_partial_does_not_mutate_identical_prior_turn(monkeypatch):
|
|
identity = {"token": "current-turn-token"}
|
|
session = Session(
|
|
session_id="stream-partial-identical-content",
|
|
messages=[
|
|
{"role": "user", "content": "historical"},
|
|
{"role": "assistant", "content": "same answer"},
|
|
{
|
|
"role": "user",
|
|
"content": "current prompt",
|
|
"_active_turn_token": identity["token"],
|
|
},
|
|
],
|
|
context_messages=[],
|
|
)
|
|
stream_id = "stream-identical-content"
|
|
monkeypatch.setitem(streaming.STREAM_PARTIAL_TEXT, stream_id, "same answer")
|
|
monkeypatch.setitem(streaming.STREAM_REASONING_TEXT, stream_id, "")
|
|
monkeypatch.setitem(streaming.STREAM_LIVE_TOOL_CALLS, stream_id, [])
|
|
|
|
appended = streaming._snapshot_and_append_partial_on_error(
|
|
session,
|
|
stream_id,
|
|
active_turn_identity=identity,
|
|
)
|
|
|
|
assert appended is session.messages[-1]
|
|
assert appended["content"] == "same answer"
|
|
assert appended["_partial"] is True
|
|
assert session.messages[1] == {"role": "assistant", "content": "same answer"}
|
|
|
|
|
|
def test_resolved_boundary_is_rebound_to_fresh_self_heal_agent():
|
|
class HealedAgent:
|
|
_persist_user_message_idx = 2
|
|
_current_turn_id = "fresh-heal-turn"
|
|
|
|
stale_identity = {
|
|
"token": "webui-token",
|
|
"text": "current prompt",
|
|
"current_turn_user_idx": 7,
|
|
"turn_id": "failed-agent-turn",
|
|
"agent_turn_boundary_resolved": True,
|
|
"agent_turn_boundary_source": "agent",
|
|
}
|
|
|
|
resolved = streaming._resolve_active_turn_authority(
|
|
stale_identity,
|
|
result={"messages": []},
|
|
agent=HealedAgent(),
|
|
)
|
|
|
|
assert resolved["current_turn_user_idx"] == 2
|
|
assert resolved["turn_id"] == "fresh-heal-turn"
|
|
assert resolved["agent_turn_boundary_source"] == "agent"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("live_snapshot", "result_snapshot", "expected"),
|
|
[
|
|
("partial response with newer suffix", "partial response", "partial response with newer suffix"),
|
|
("partial response", "partial response with newer suffix", "partial response with newer suffix"),
|
|
],
|
|
)
|
|
def test_differing_live_and_result_snapshots_are_exact_once(
|
|
monkeypatch,
|
|
live_snapshot,
|
|
result_snapshot,
|
|
expected,
|
|
):
|
|
token = "current-turn-token"
|
|
stream_id = "differing-partial-sources"
|
|
session = Session(
|
|
session_id="differing-partial-sources",
|
|
messages=[
|
|
{"role": "user", "content": "historical"},
|
|
{"role": "assistant", "content": "historical answer"},
|
|
{
|
|
"role": "user",
|
|
"content": "current prompt",
|
|
"_active_turn_token": token,
|
|
},
|
|
],
|
|
context_messages=[],
|
|
)
|
|
result = {
|
|
"partial": True,
|
|
"turn_id": "agent-current-turn",
|
|
"current_turn_user_idx": 1,
|
|
"messages": [
|
|
{"role": "assistant", "content": "[compacted] history"},
|
|
{"role": "user", "content": " current\n prompt "},
|
|
{"role": "assistant", "content": result_snapshot},
|
|
{"role": "user", "content": "later turn"},
|
|
{"role": "assistant", "content": "later wrong"},
|
|
],
|
|
}
|
|
identity = streaming._resolve_active_turn_authority(
|
|
{"token": token, "text": "current prompt"},
|
|
result=result,
|
|
)
|
|
monkeypatch.setitem(streaming.STREAM_PARTIAL_TEXT, stream_id, live_snapshot)
|
|
monkeypatch.setitem(streaming.STREAM_REASONING_TEXT, stream_id, "")
|
|
monkeypatch.setitem(streaming.STREAM_LIVE_TOOL_CALLS, stream_id, [])
|
|
|
|
live_row = streaming._snapshot_and_append_partial_on_error(
|
|
session,
|
|
stream_id,
|
|
active_turn_identity=identity,
|
|
)
|
|
result_row = streaming._append_result_partial_on_error(
|
|
session,
|
|
result,
|
|
[{"role": "user", "content": "replaced baseline"}],
|
|
"current prompt",
|
|
active_turn_identity=identity,
|
|
)
|
|
|
|
partials = [message for message in session.messages if message.get("_partial")]
|
|
assert result_row is live_row
|
|
assert [message["content"] for message in partials] == [expected]
|
|
assert "later wrong" not in [message.get("content") for message in session.messages]
|
|
|
|
|
|
def test_reconciled_compressed_projection_keeps_durable_snapshot_revision(tmp_path, monkeypatch):
|
|
sid = "revision-reconcile"
|
|
db_path = tmp_path / "state.db"
|
|
_make_state_db(
|
|
db_path,
|
|
sid,
|
|
[
|
|
{"role": "user", "content": "durable user", "timestamp": 1.0},
|
|
{"role": "assistant", "content": "durable answer", "timestamp": 2.0},
|
|
{"role": "user", "content": "durable follow-up", "timestamp": 3.0},
|
|
],
|
|
)
|
|
monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
|
|
compacted_context = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "[CONTEXT COMPACTION — REFERENCE ONLY]\nCompressed task summary.",
|
|
"timestamp": 4.0,
|
|
}
|
|
]
|
|
session = Session(
|
|
session_id=sid,
|
|
messages=[{"role": "user", "content": "full visible transcript"}],
|
|
context_messages=compacted_context,
|
|
)
|
|
|
|
state_snapshot = models.get_state_db_session_messages(sid, with_revision=True)
|
|
reconciled = models.reconciled_state_db_messages_for_session(
|
|
session,
|
|
prefer_context=True,
|
|
state_messages=state_snapshot,
|
|
with_revision=True,
|
|
)
|
|
|
|
assert reconciled.messages == compacted_context
|
|
assert len(reconciled.messages) < reconciled.revision["active_message_count"]
|
|
assert reconciled.revision == state_snapshot.revision
|
|
|
|
|
|
def test_webui_run_passes_revision_from_original_snapshot_when_sqlite_changes_before_agent(
|
|
tmp_path, monkeypatch
|
|
):
|
|
import api.profiles as profiles
|
|
|
|
sid = "revision-stream"
|
|
stream_id = "stream-revision"
|
|
# Keep a conflicting default-profile DB to prove the worker uses the
|
|
# session's explicit profile rather than process-global active state.
|
|
_make_state_db(
|
|
tmp_path / "state.db",
|
|
sid,
|
|
[{"role": "user", "content": "wrong profile row", "timestamp": 1.0}],
|
|
)
|
|
profile_home = tmp_path / "named-profile-home"
|
|
profile_home.mkdir()
|
|
db_path = profile_home / "state.db"
|
|
_make_state_db(
|
|
db_path,
|
|
sid,
|
|
[
|
|
{"role": "user", "content": "durable user", "timestamp": 1.0},
|
|
{"role": "assistant", "content": "durable answer", "timestamp": 2.0},
|
|
{"role": "user", "content": "durable follow-up", "timestamp": 3.0},
|
|
],
|
|
)
|
|
compacted_context = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "[CONTEXT COMPACTION — REFERENCE ONLY]\nCompressed task summary.",
|
|
"timestamp": 4.0,
|
|
}
|
|
]
|
|
session, _event_queue = _install_streaming_session(
|
|
monkeypatch,
|
|
tmp_path,
|
|
sid=sid,
|
|
stream_id=stream_id,
|
|
messages=[{"role": "user", "content": "visible user", "timestamp": 1.0}],
|
|
context_messages=compacted_context,
|
|
)
|
|
session.profile = "named-profile"
|
|
session.save(touch_updated_at=False)
|
|
monkeypatch.setattr(profiles, "get_hermes_home_for_profile", lambda _profile: profile_home)
|
|
monkeypatch.setattr(profiles, "get_profile_runtime_env", lambda _home: {})
|
|
|
|
original_reconcile = streaming.reconciled_state_db_messages_for_session
|
|
reconciliation_calls = 0
|
|
|
|
def reconcile_then_mutate(*args, **kwargs):
|
|
nonlocal reconciliation_calls
|
|
result = original_reconcile(*args, **kwargs)
|
|
reconciliation_calls += 1
|
|
if reconciliation_calls == 1:
|
|
_append_state_row(
|
|
db_path,
|
|
sid,
|
|
role="assistant",
|
|
content="concurrent durable row",
|
|
timestamp=5.0,
|
|
)
|
|
return result
|
|
|
|
monkeypatch.setattr(
|
|
streaming,
|
|
"reconciled_state_db_messages_for_session",
|
|
reconcile_then_mutate,
|
|
)
|
|
captured = {}
|
|
|
|
class FakeAgent:
|
|
def __init__(self, session_id=None, **_kwargs):
|
|
self.session_id = session_id
|
|
self.context_compressor = None
|
|
self.ephemeral_system_prompt = None
|
|
self._last_error = None
|
|
|
|
def run_conversation(self, **kwargs):
|
|
captured.update(kwargs)
|
|
return {
|
|
"completed": True,
|
|
"final_response": "ok",
|
|
"messages": [
|
|
{"role": "user", "content": kwargs["persist_user_message"]},
|
|
{"role": "assistant", "content": "ok"},
|
|
],
|
|
}
|
|
|
|
monkeypatch.setattr(streaming, "_get_ai_agent", lambda: FakeAgent)
|
|
|
|
streaming._run_agent_streaming(
|
|
session_id=sid,
|
|
msg_text="new webui turn",
|
|
model="test-model",
|
|
workspace=str(tmp_path),
|
|
stream_id=stream_id,
|
|
attachments=[],
|
|
)
|
|
|
|
assert captured["task_id"] == sid
|
|
assert captured["conversation_history_revision"] == {
|
|
"session_id": sid,
|
|
"active_message_count": 3,
|
|
"max_active_message_id": 3,
|
|
}
|
|
assert captured["conversation_history_revision"]["active_message_count"] == 3
|
|
assert len(models.get_state_db_session_messages(sid, profile="named-profile")) == 4
|
|
|
|
|
|
def test_webui_run_missing_explicit_profile_passes_no_foreign_revision(
|
|
tmp_path,
|
|
monkeypatch,
|
|
):
|
|
import api.profiles as profiles
|
|
|
|
sid = "missing-profile-stream"
|
|
stream_id = "stream-missing-profile"
|
|
foreign_db = tmp_path / "state.db"
|
|
_make_state_db(
|
|
foreign_db,
|
|
sid,
|
|
[{"role": "user", "content": "foreign profile row", "timestamp": 1.0}],
|
|
)
|
|
missing_home = tmp_path / "missing-profile-home"
|
|
session, event_queue = _install_streaming_session(
|
|
monkeypatch,
|
|
tmp_path,
|
|
sid=sid,
|
|
stream_id=stream_id,
|
|
messages=[{"role": "user", "content": "local projection", "timestamp": 1.0}],
|
|
context_messages=[],
|
|
)
|
|
session.profile = "missing"
|
|
session.save(touch_updated_at=False)
|
|
monkeypatch.setattr(profiles, "get_hermes_home_for_profile", lambda _profile: missing_home)
|
|
monkeypatch.setattr(profiles, "get_profile_runtime_env", lambda _home: {})
|
|
captured = {}
|
|
|
|
class FakeAgent:
|
|
def __init__(self, session_id=None, **_kwargs):
|
|
self.session_id = session_id
|
|
self.context_compressor = None
|
|
self.ephemeral_system_prompt = None
|
|
self._last_error = None
|
|
|
|
def run_conversation(self, **kwargs):
|
|
captured.update(kwargs)
|
|
return {
|
|
"completed": True,
|
|
"final_response": "ok",
|
|
"messages": [
|
|
{"role": "user", "content": kwargs["persist_user_message"]},
|
|
{"role": "assistant", "content": "ok"},
|
|
],
|
|
}
|
|
|
|
monkeypatch.setattr(streaming, "_get_ai_agent", lambda: FakeAgent)
|
|
|
|
streaming._run_agent_streaming(
|
|
session_id=sid,
|
|
msg_text="new webui turn",
|
|
model="test-model",
|
|
workspace=str(tmp_path),
|
|
stream_id=stream_id,
|
|
attachments=[],
|
|
)
|
|
|
|
assert captured["conversation_history_revision"] is None
|
|
events = _drain_events(event_queue)
|
|
assert not any(event == "apperror" for event, _payload in events)
|
|
|
|
|
|
@pytest.mark.parametrize("delivery", ["exception", "result"])
|
|
def test_stale_compression_snapshot_emits_actionable_error_without_replaying_turn(
|
|
tmp_path, monkeypatch, delivery
|
|
):
|
|
sid = f"revision-stale-{delivery}"
|
|
stream_id = f"stream-revision-stale-{delivery}"
|
|
_make_state_db(
|
|
tmp_path / "state.db",
|
|
sid,
|
|
[{"role": "user", "content": "durable user", "timestamp": 1.0}],
|
|
)
|
|
session, event_queue = _install_streaming_session(
|
|
monkeypatch,
|
|
tmp_path,
|
|
sid=sid,
|
|
stream_id=stream_id,
|
|
messages=[],
|
|
context_messages=[],
|
|
)
|
|
calls = 0
|
|
|
|
class CompressionSnapshotStaleError(RuntimeError):
|
|
pass
|
|
|
|
compression_module = types.ModuleType("agent.conversation_compression")
|
|
compression_module.__dict__["CompressionSnapshotStaleError"] = CompressionSnapshotStaleError
|
|
monkeypatch.setitem(sys.modules, "agent.conversation_compression", compression_module)
|
|
|
|
class FakeAgent:
|
|
def __init__(self, session_id=None, stream_delta_callback=None, **_kwargs):
|
|
self.session_id = session_id
|
|
self.stream_delta_callback = stream_delta_callback
|
|
self.context_compressor = None
|
|
self.ephemeral_system_prompt = None
|
|
self._last_error = None
|
|
self._persist_user_message_idx = 0
|
|
self._current_turn_id = f"paired-turn-{delivery}"
|
|
|
|
def run_conversation(self, **_kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if self.stream_delta_callback:
|
|
self.stream_delta_callback(
|
|
"Partial work completed before the conflict with a newer suffix."
|
|
)
|
|
if delivery == "exception":
|
|
raise CompressionSnapshotStaleError(
|
|
"expected revision count=1 max_id=1; observed count=2 max_id=9"
|
|
)
|
|
return {
|
|
"completed": False,
|
|
"final_response": "Partial work completed before the conflict.",
|
|
"messages": [
|
|
{"role": "user", "content": "new webui turn"},
|
|
{
|
|
"role": "assistant",
|
|
"content": "Partial work completed before the conflict.",
|
|
},
|
|
],
|
|
"error": "compression_snapshot_stale",
|
|
"partial": True,
|
|
"failed": False,
|
|
"compression_snapshot_stale": True,
|
|
}
|
|
|
|
monkeypatch.setattr(streaming, "_get_ai_agent", lambda: FakeAgent)
|
|
|
|
streaming._run_agent_streaming(
|
|
session_id=sid,
|
|
msg_text="new webui turn",
|
|
model="test-model",
|
|
workspace=str(tmp_path),
|
|
stream_id=stream_id,
|
|
attachments=[],
|
|
)
|
|
|
|
events = _drain_events(event_queue)
|
|
apperrors = [payload for event, payload in events if event == "apperror"]
|
|
assert calls == 1
|
|
assert apperrors and apperrors[-1]["type"] == "compression_snapshot_stale"
|
|
assert "next message" in apperrors[-1]["hint"].lower()
|
|
assert "expected revision" not in apperrors[-1]["message"].lower()
|
|
assert not any(event == "done" for event, _payload in events)
|
|
|
|
reloaded = Session.load(sid)
|
|
assert reloaded is not None
|
|
assert reloaded.active_stream_id is None
|
|
assert reloaded.pending_user_message is None
|
|
assert any(
|
|
message.get("_partial")
|
|
and message.get("content")
|
|
== "Partial work completed before the conflict with a newer suffix."
|
|
for message in reloaded.messages
|
|
)
|
|
partial_messages = [
|
|
message for message in reloaded.messages if message.get("_partial")
|
|
]
|
|
assert len(partial_messages) == 1, [
|
|
(
|
|
message.get("role"),
|
|
message.get("content"),
|
|
message.get("_active_turn_token"),
|
|
message.get("_partial"),
|
|
)
|
|
for message in reloaded.messages
|
|
]
|
|
assert sum(
|
|
"Partial work completed" in str(message.get("content") or "")
|
|
for message in reloaded.messages
|
|
if message.get("role") == "assistant"
|
|
) == 1
|
|
assert reloaded.messages[-1]["_error"] is True
|
|
assert "next message" in reloaded.messages[-1]["content"].lower()
|
|
assert session.active_stream_id is None
|
|
|
|
|
|
@pytest.mark.parametrize("failure_mode", ["result", "exception"])
|
|
@pytest.mark.parametrize(
|
|
"second_result",
|
|
[
|
|
"recovered",
|
|
"stale_error",
|
|
"stale_flag",
|
|
"stale_without_current_assistant",
|
|
"stale_non_prefix_compacted",
|
|
"stale_non_prefix_owned_partial",
|
|
"stale_non_prefix_without_current_user",
|
|
"stale_non_prefix_repeated_prompt_without_authority",
|
|
"no_response_after_streamed_token",
|
|
],
|
|
)
|
|
def test_auth_self_heal_refreshes_revision_after_first_agent_persists_user(
|
|
tmp_path, monkeypatch, failure_mode, second_result
|
|
):
|
|
sid = f"revision-self-heal-{failure_mode}-{second_result}"
|
|
stream_id = f"stream-revision-self-heal-{failure_mode}-{second_result}"
|
|
db_path = tmp_path / "state.db"
|
|
prior_messages = [
|
|
{"role": "user", "content": "prior user", "timestamp": 1.0},
|
|
{"role": "assistant", "content": "prior answer", "timestamp": 2.0},
|
|
]
|
|
_make_state_db(db_path, sid, prior_messages)
|
|
_session, event_queue = _install_streaming_session(
|
|
monkeypatch,
|
|
tmp_path,
|
|
sid=sid,
|
|
stream_id=stream_id,
|
|
messages=prior_messages,
|
|
context_messages=prior_messages,
|
|
)
|
|
_session.pending_user_source = "webui"
|
|
revisions = []
|
|
|
|
class AuthThenRecoverAgent:
|
|
runs = 0
|
|
|
|
def __init__(self, session_id=None, **_kwargs):
|
|
self.session_id = session_id
|
|
self.context_compressor = None
|
|
self.ephemeral_system_prompt = None
|
|
self._last_error = None
|
|
self.stream_delta_callback = _kwargs.get("stream_delta_callback")
|
|
|
|
def run_conversation(self, persist_user_timestamp=None, **kwargs):
|
|
type(self).runs += 1
|
|
assert persist_user_timestamp == 10.0
|
|
persisted = Session.load(sid)
|
|
assert persisted._webui_pending_user_timestamp_identity == (
|
|
stream_id,
|
|
persist_user_timestamp,
|
|
)
|
|
revisions.append(kwargs["conversation_history_revision"])
|
|
history = list(kwargs.get("conversation_history") or [])
|
|
if type(self).runs == 1:
|
|
_append_state_row(
|
|
db_path,
|
|
sid,
|
|
role="user",
|
|
content=kwargs["persist_user_message"],
|
|
timestamp=3.0,
|
|
)
|
|
if failure_mode == "exception":
|
|
raise RuntimeError("401 unauthorized")
|
|
return {
|
|
"messages": history,
|
|
"error": {
|
|
"type": "authentication_error",
|
|
"status_code": 401,
|
|
"message": "token invalid",
|
|
},
|
|
}
|
|
if second_result == "no_response_after_streamed_token":
|
|
if self.stream_delta_callback is not None:
|
|
self.stream_delta_callback("streamed but not finalized")
|
|
return {"completed": True, "messages": history}
|
|
if second_result != "recovered":
|
|
if second_result == "stale_non_prefix_owned_partial":
|
|
stale_messages = [
|
|
{"role": "assistant", "content": "compacted history"},
|
|
{
|
|
"role": "user",
|
|
"content": "new webui turn",
|
|
"_active_turn_token": streaming.build_active_turn_token(
|
|
stream_id, 10.0
|
|
),
|
|
},
|
|
{"role": "assistant", "content": "owned partial"},
|
|
{"role": "user", "content": "later user"},
|
|
{"role": "assistant", "content": "later wrong"},
|
|
]
|
|
elif second_result == "stale_non_prefix_without_current_user":
|
|
stale_messages = [
|
|
{"role": "user", "content": "unrelated historical user"},
|
|
{"role": "assistant", "content": "historical answer"},
|
|
]
|
|
elif second_result == "stale_non_prefix_compacted":
|
|
# Compacted/replayed result that REPLACES the pre-call
|
|
# baseline instead of appending to it: historical
|
|
# assistant rows (including "prior answer") sit before the
|
|
# last current-user row and one of them lands in the
|
|
# numeric suffix messages[len(baseline):]. There is no
|
|
# assistant row for the current turn.
|
|
stale_messages = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "[compacted] summary of earlier context",
|
|
},
|
|
{"role": "user", "content": "prior user"},
|
|
{"role": "assistant", "content": "intermediate answer"},
|
|
{"role": "assistant", "content": "prior answer"},
|
|
{"role": "user", "content": "new webui turn"},
|
|
]
|
|
elif second_result == "stale_without_current_assistant":
|
|
stale_messages = history + [
|
|
{"role": "user", "content": "new webui turn"}
|
|
]
|
|
elif second_result == "stale_non_prefix_repeated_prompt_without_authority":
|
|
stale_messages = [
|
|
{"role": "user", "content": "new webui turn"},
|
|
{"role": "assistant", "content": "prior answer"},
|
|
]
|
|
else:
|
|
stale_messages = history + [
|
|
{"role": "user", "content": "new webui turn"},
|
|
{"role": "assistant", "content": "partial stale"},
|
|
]
|
|
stale_result = {
|
|
"completed": False,
|
|
"final_response": (
|
|
"owned partial"
|
|
if second_result == "stale_non_prefix_owned_partial"
|
|
else (
|
|
""
|
|
if second_result
|
|
in (
|
|
"stale_without_current_assistant",
|
|
"stale_non_prefix_compacted",
|
|
"stale_non_prefix_without_current_user",
|
|
"stale_non_prefix_repeated_prompt_without_authority",
|
|
)
|
|
else "partial stale"
|
|
)
|
|
),
|
|
"messages": stale_messages,
|
|
"partial": True,
|
|
"failed": False,
|
|
}
|
|
if second_result == "stale_error":
|
|
stale_result["error"] = "compression_snapshot_stale"
|
|
elif second_result == "stale_flag":
|
|
stale_result["compression_snapshot_stale"] = True
|
|
else:
|
|
stale_result["error"] = "compression_snapshot_stale"
|
|
stale_result["compression_snapshot_stale"] = True
|
|
return stale_result
|
|
return {
|
|
"completed": True,
|
|
"final_response": "recovered",
|
|
"messages": history + [
|
|
{"role": "user", "content": "new webui turn"},
|
|
{"role": "assistant", "content": "recovered"},
|
|
],
|
|
}
|
|
|
|
monkeypatch.setattr(streaming, "_get_ai_agent", lambda: AuthThenRecoverAgent)
|
|
monkeypatch.setattr(
|
|
streaming,
|
|
"_attempt_credential_self_heal",
|
|
lambda *_args, **_kwargs: {
|
|
"api_key": "refreshed-key",
|
|
"provider": "test-provider",
|
|
"base_url": None,
|
|
"credential_pool": None,
|
|
},
|
|
)
|
|
|
|
streaming._run_agent_streaming(
|
|
session_id=sid,
|
|
msg_text="new webui turn",
|
|
model="test-model",
|
|
workspace=str(tmp_path),
|
|
stream_id=stream_id,
|
|
attachments=[],
|
|
)
|
|
|
|
assert revisions == [
|
|
{
|
|
"session_id": sid,
|
|
"active_message_count": 2,
|
|
"max_active_message_id": 2,
|
|
},
|
|
{
|
|
"session_id": sid,
|
|
"active_message_count": 3,
|
|
"max_active_message_id": 3,
|
|
},
|
|
]
|
|
events = _drain_events(event_queue)
|
|
apperrors = [payload for event, payload in events if event == "apperror"]
|
|
reloaded = Session.load(sid)
|
|
assert reloaded is not None
|
|
assert AuthThenRecoverAgent.runs == 2
|
|
if second_result == "recovered":
|
|
assert not apperrors
|
|
assert any(message.get("content") == "recovered" for message in reloaded.messages)
|
|
elif second_result == "no_response_after_streamed_token":
|
|
assert len(apperrors) == 1
|
|
assert apperrors[0]["type"] == "auth_mismatch"
|
|
assert not any(event == "done" for event, _payload in events)
|
|
assert reloaded.active_stream_id is None
|
|
assert reloaded.pending_user_message is None
|
|
else:
|
|
assert len(apperrors) == 1
|
|
assert apperrors[0]["type"] == "compression_snapshot_stale"
|
|
assert "next message" in apperrors[0]["hint"].lower()
|
|
assert not any(event == "done" for event, _payload in events)
|
|
assert reloaded.active_stream_id is None
|
|
assert reloaded.pending_user_message is None
|
|
assert reloaded.pending_attachments == []
|
|
partials = [
|
|
message
|
|
for message in reloaded.messages
|
|
if message.get("_partial") is True
|
|
]
|
|
if second_result in (
|
|
"stale_without_current_assistant",
|
|
"stale_non_prefix_compacted",
|
|
"stale_non_prefix_without_current_user",
|
|
"stale_non_prefix_repeated_prompt_without_authority",
|
|
):
|
|
assert not partials
|
|
assert not any(
|
|
message.get("content") == "prior answer" and message.get("_partial") is True
|
|
for message in reloaded.messages
|
|
)
|
|
else:
|
|
expected_partial = (
|
|
"owned partial"
|
|
if second_result == "stale_non_prefix_owned_partial"
|
|
else "partial stale"
|
|
)
|
|
assert [message.get("content") for message in partials] == [expected_partial]
|
|
if second_result == "stale_non_prefix_owned_partial":
|
|
assert not any(
|
|
message.get("content") == "later wrong"
|
|
for message in reloaded.messages
|
|
)
|
|
assert reloaded.messages[-1]["_error"] is True
|
|
assert "next message" in reloaded.messages[-1]["content"].lower()
|
|
|
|
|
|
def _repeated_prompt_collision_result_messages(prompt, *, current_answer):
|
|
"""Non-prefix Agent projection where BOTH candidate positions are user rows
|
|
with the same normalized text.
|
|
|
|
Against the 2-row heal baseline (prior user + prior assistant) the legacy
|
|
shifted probe ``current_turn_user_idx - len(previous_context)`` = 3 - 2 = 1
|
|
lands on the HISTORICAL same-text user row, which is followed only by
|
|
historical assistant prose. The declared Agent index 3 addresses
|
|
``result["messages"]`` directly and lands on the CURRENT same-text user
|
|
row, followed only by current output (when any exists).
|
|
"""
|
|
rows = [
|
|
{"role": "assistant", "content": "[compacted] summary of earlier context"},
|
|
{"role": "user", "content": prompt}, # shifted probe target (historical)
|
|
{"role": "assistant", "content": "historical answer"}, # historical-only prose
|
|
{"role": "user", "content": prompt}, # exact Agent index target (current)
|
|
]
|
|
if current_answer is not None:
|
|
rows.append({"role": "assistant", "content": current_answer})
|
|
return rows
|
|
|
|
|
|
@pytest.mark.parametrize("terminal", ["returned", "raised"])
|
|
@pytest.mark.parametrize(
|
|
"collision",
|
|
[
|
|
"current_output_after_exact_row",
|
|
"historical_output_after_shifted_row_only",
|
|
"historical_only_stale_partial",
|
|
],
|
|
)
|
|
def test_self_heal_repeated_prompt_never_accepts_shifted_historical_row(
|
|
tmp_path, monkeypatch, terminal, collision
|
|
):
|
|
"""Required regression (2026-08-13 review): one declared index domain.
|
|
|
|
A repeated prompt makes both legacy candidate positions (shifted
|
|
``current_turn_user_idx - len(previous_context)`` and exact
|
|
``current_turn_user_idx``) same-text user rows in the healed non-prefix
|
|
projection. Historical assistant output follows only the shifted row;
|
|
current output follows only the exact row. Composed through BOTH
|
|
production self-heal entries — returned auth-failure result and raised
|
|
401 — the historical row must never be accepted by
|
|
``_self_heal_result_succeeded`` nor persisted by
|
|
``_append_result_partial_on_error``; settlement/done/apperror must fire
|
|
exactly once.
|
|
"""
|
|
prompt = "new webui turn"
|
|
sid = f"repeat-collision-{terminal}-{collision}"
|
|
stream_id = f"stream-{sid}"
|
|
db_path = tmp_path / "state.db"
|
|
prior_messages = [
|
|
{"role": "user", "content": prompt, "timestamp": 1.0},
|
|
{"role": "assistant", "content": "historical answer", "timestamp": 2.0},
|
|
]
|
|
_make_state_db(db_path, sid, prior_messages)
|
|
session, event_queue = _install_streaming_session(
|
|
monkeypatch,
|
|
tmp_path,
|
|
sid=sid,
|
|
stream_id=stream_id,
|
|
messages=prior_messages,
|
|
context_messages=prior_messages,
|
|
)
|
|
session.pending_user_message = prompt
|
|
|
|
class RepeatedPromptAgent:
|
|
runs = 0
|
|
|
|
def __init__(self, session_id=None, **_kwargs):
|
|
self.session_id = session_id
|
|
self.context_compressor = None
|
|
self.ephemeral_system_prompt = None
|
|
self._last_error = None
|
|
self.stream_delta_callback = _kwargs.get("stream_delta_callback")
|
|
|
|
def run_conversation(self, **kwargs):
|
|
type(self).runs += 1
|
|
history = list(kwargs.get("conversation_history") or [])
|
|
if type(self).runs == 1:
|
|
_append_state_row(
|
|
db_path,
|
|
sid,
|
|
role="user",
|
|
content=kwargs["persist_user_message"],
|
|
timestamp=3.0,
|
|
)
|
|
if terminal == "raised":
|
|
raise RuntimeError("401 unauthorized")
|
|
return {
|
|
"messages": history,
|
|
"error": {
|
|
"type": "authentication_error",
|
|
"status_code": 401,
|
|
"message": "token invalid",
|
|
},
|
|
}
|
|
# Heal retry: the 2-row refreshed baseline makes the legacy
|
|
# shifted probe (3 - 2 = 1) collide with the historical user row.
|
|
heal = {
|
|
"turn_id": "agent-heal-turn",
|
|
"current_turn_user_idx": 3,
|
|
"messages": _repeated_prompt_collision_result_messages(
|
|
prompt,
|
|
current_answer=(
|
|
"healed current answer"
|
|
if collision == "current_output_after_exact_row"
|
|
else None
|
|
),
|
|
),
|
|
}
|
|
if collision == "current_output_after_exact_row":
|
|
heal.update(
|
|
{"completed": True, "final_response": "healed current answer"}
|
|
)
|
|
elif collision == "historical_output_after_shifted_row_only":
|
|
# Silent success shape: if the shifted historical row were
|
|
# accepted, "historical answer" would satisfy the retry.
|
|
heal.update({"completed": True, "final_response": ""})
|
|
else:
|
|
# Stale partial shape: if the shifted historical row were
|
|
# accepted, "historical answer" would persist as the current
|
|
# turn's partial.
|
|
heal.update(
|
|
{
|
|
"completed": False,
|
|
"partial": True,
|
|
"failed": False,
|
|
"final_response": "",
|
|
"error": "compression_snapshot_stale",
|
|
"compression_snapshot_stale": True,
|
|
}
|
|
)
|
|
return heal
|
|
|
|
monkeypatch.setattr(streaming, "_get_ai_agent", lambda: RepeatedPromptAgent)
|
|
monkeypatch.setattr(
|
|
streaming,
|
|
"_attempt_credential_self_heal",
|
|
lambda *_args, **_kwargs: {
|
|
"api_key": "refreshed-key",
|
|
"provider": "test-provider",
|
|
"base_url": None,
|
|
"credential_pool": None,
|
|
},
|
|
)
|
|
|
|
streaming._run_agent_streaming(
|
|
session_id=sid,
|
|
msg_text=prompt,
|
|
model="test-model",
|
|
workspace=str(tmp_path),
|
|
stream_id=stream_id,
|
|
attachments=[],
|
|
)
|
|
|
|
assert RepeatedPromptAgent.runs == 2
|
|
events = _drain_events(event_queue)
|
|
done_events = [payload for event, payload in events if event == "done"]
|
|
apperrors = [payload for event, payload in events if event == "apperror"]
|
|
reloaded = Session.load(sid)
|
|
assert reloaded is not None
|
|
|
|
# Exact-once turn settlement in every arm: the pending turn is consumed.
|
|
assert reloaded.active_stream_id is None
|
|
assert reloaded.pending_user_message is None
|
|
assert reloaded.pending_attachments == []
|
|
|
|
current_turn_assistants = [
|
|
str(message.get("content") or "")
|
|
for message in reloaded.messages[2:]
|
|
if isinstance(message, dict) and message.get("role") == "assistant"
|
|
]
|
|
if collision == "current_output_after_exact_row":
|
|
# Success arm: exactly one done, no apperror, and the settled current
|
|
# turn carries only the exact-row answer — persisted exactly once.
|
|
assert len(done_events) == 1
|
|
assert not apperrors
|
|
assert current_turn_assistants == ["healed current answer"]
|
|
done_messages = done_events[0]["session"]["messages"]
|
|
assert [
|
|
(message["role"], message.get("content")) for message in done_messages
|
|
] == [
|
|
("user", prompt),
|
|
("assistant", "historical answer"),
|
|
("user", prompt),
|
|
("assistant", "healed current answer"),
|
|
]
|
|
assert not any(message.get("_error") for message in reloaded.messages)
|
|
else:
|
|
# Failure arms: the shifted historical row must never satisfy the
|
|
# retry — exactly one apperror, never a done.
|
|
assert not done_events
|
|
assert len(apperrors) == 1
|
|
if collision == "historical_output_after_shifted_row_only":
|
|
assert apperrors[0]["type"] == "auth_mismatch"
|
|
else:
|
|
assert apperrors[0]["type"] == "compression_snapshot_stale"
|
|
# Historical prose is never accepted as this turn's output nor
|
|
# persisted as its partial.
|
|
assert not any(
|
|
message.get("_partial") for message in reloaded.messages
|
|
)
|
|
assert "historical answer" not in current_turn_assistants
|
|
assert reloaded.messages[-1]["_error"] is True
|
|
# The historical assistant row itself survives untouched exactly once, at
|
|
# its original position in the transcript.
|
|
assert [
|
|
str(message.get("content") or "")
|
|
for message in reloaded.messages
|
|
if isinstance(message, dict) and message.get("role") == "assistant"
|
|
].count("historical answer") == 1
|