fix(chat): preserve submitted image turns on stale recovery
When a dead stream is reconciled from state.db, retain the submitted prompt and attachment in the sidecar while keeping the Agent-enriched row in model context. Use one parent snapshot for full-session conflict projection so the sidecar owner is stable across the display merge. Assisted-by: Hermes Agent:gpt-6-sol Assisted-by: Codex:gpt-6-luna Assisted-by: Claude Code:claude-opus-5-5
This commit is contained in:
1 parent
4f2bdd2dff
commit
d07c60a92a
5 files changed
+224
-32
No files matched your search
+77
-11
@@ -4033,6 +4033,68 @@ def _sync_sidecar_from_state_db_if_newer(session) -> bool:
|
||||
)
|
||||
if not state_messages:
|
||||
return False
|
||||
|
||||
# The Agent row is intentionally hidden from display while the saved
|
||||
# handoff proof is active. Materialize the exact WebUI-owned prompt
|
||||
# before reconciling that row away, so clearing pending state cannot
|
||||
# erase the only durable copy of the submitted text/attachments.
|
||||
pending_identity = _validated_webui_pending_user_timestamp_identity(
|
||||
locked,
|
||||
getattr(locked, '_webui_pending_user_timestamp_identity', None),
|
||||
)
|
||||
# When a legacy sidecar has no explicit context_messages, the shared
|
||||
# context reconciler falls back to messages. Capture that model-facing
|
||||
# merge before adding the display-only WebUI owner row below.
|
||||
merged_context = (
|
||||
reconciled_state_db_messages_for_session(
|
||||
locked,
|
||||
prefer_context=True,
|
||||
state_messages=state_messages,
|
||||
)
|
||||
if pending_identity is not None
|
||||
else None
|
||||
)
|
||||
if pending_identity is not None:
|
||||
pending_timestamp = pending_identity[1]
|
||||
pending_text = locked.pending_user_message
|
||||
pending_source = getattr(locked, 'pending_user_source', None) or 'webui'
|
||||
pending_attachments = list(getattr(locked, 'pending_attachments', None) or [])
|
||||
pending_row = {
|
||||
'role': 'user',
|
||||
'content': pending_text,
|
||||
'timestamp': pending_timestamp,
|
||||
}
|
||||
stamp_message_source(pending_row, pending_source)
|
||||
if str(pending_source or '').strip().lower() == 'fork':
|
||||
pending_row['_fork_child_turn'] = locked.session_id
|
||||
if pending_attachments:
|
||||
pending_row['attachments'] = pending_attachments
|
||||
|
||||
existing_pending = next((
|
||||
message for message in locked_messages
|
||||
if isinstance(message, dict)
|
||||
and (
|
||||
_message_exact_timestamp_details(message)
|
||||
== (pending_timestamp, True)
|
||||
and (message.get('_source') or 'webui') == pending_source
|
||||
and _message_matches_pending_text(message, pending_text)
|
||||
)
|
||||
), None)
|
||||
if existing_pending is not None:
|
||||
# A prior eager checkpoint already owns this exact proved turn;
|
||||
# refresh it from the still-authoritative pending fields.
|
||||
existing_pending.update(pending_row)
|
||||
if pending_attachments:
|
||||
existing_pending['attachments'] = pending_attachments
|
||||
else:
|
||||
existing_pending.pop('attachments', None)
|
||||
else:
|
||||
# The proof makes this submitted row authoritative even if a
|
||||
# clock adjustment places it before the surviving sidecar tail.
|
||||
if not _insert_state_message_chronologically(locked_messages, pending_row):
|
||||
locked_messages.insert(0, pending_row)
|
||||
locked.messages = locked_messages
|
||||
|
||||
merged_messages = reconciled_state_db_messages_for_session(
|
||||
locked,
|
||||
state_messages=state_messages,
|
||||
@@ -4043,11 +4105,12 @@ def _sync_sidecar_from_state_db_if_newer(session) -> bool:
|
||||
# recover — leave the sidecar untouched rather than rewriting in place.
|
||||
if len(merged_messages) <= locked_count:
|
||||
return False
|
||||
merged_context = reconciled_state_db_messages_for_session(
|
||||
locked,
|
||||
prefer_context=True,
|
||||
state_messages=state_messages,
|
||||
)
|
||||
if merged_context is None:
|
||||
merged_context = reconciled_state_db_messages_for_session(
|
||||
locked,
|
||||
prefer_context=True,
|
||||
state_messages=state_messages,
|
||||
)
|
||||
|
||||
# Mutate + persist the freshly-loaded, locked object. Because we hold the
|
||||
# lock and reloaded under it, this save cannot clobber a concurrent
|
||||
@@ -10787,12 +10850,15 @@ def _project_native_image_payload_conflicts_for_display(
|
||||
if row_id_valid and row_id is not None and timestamp_valid
|
||||
else None
|
||||
)
|
||||
if (
|
||||
conflict is not None
|
||||
and message is not conflict[1]
|
||||
and _message_identity_compatible(message, conflict[0])
|
||||
):
|
||||
continue
|
||||
if conflict is not None and _message_identity_compatible(message, conflict[0]):
|
||||
incoming, sidecar_owner = conflict
|
||||
is_sidecar_owner = (
|
||||
_message_identity_compatible(message, sidecar_owner)
|
||||
and _session_message_api_content_key(message)
|
||||
== _session_message_api_content_key(sidecar_owner)
|
||||
)
|
||||
if not is_sidecar_owner:
|
||||
continue
|
||||
visible_messages.append(message)
|
||||
return visible_messages
|
||||
|
||||
|
||||
+39
-20
@@ -10065,7 +10065,35 @@ def _merged_session_messages_for_display(session, cli_messages=None) -> list:
|
||||
|
||||
|
||||
|
||||
def _merged_webui_lineage_messages_for_display(session, messages=None) -> list:
|
||||
_LINEAGE_PARENT_SESSION_UNSET = object()
|
||||
|
||||
|
||||
def _webui_lineage_parent_session_for_display(session):
|
||||
"""Load the immediate parent only for display-eligible continuations."""
|
||||
parent_id = str(getattr(session, "parent_session_id", "") or "").strip()
|
||||
if not parent_id:
|
||||
return None
|
||||
if (
|
||||
str(getattr(session, "compression_recovery_source_session_id", "") or "").strip()
|
||||
and str(getattr(session, "compression_recovery_action", "") or "").strip()
|
||||
):
|
||||
return None
|
||||
source = str(getattr(session, "session_source", "") or "").strip().lower()
|
||||
relationship = str(getattr(session, "relationship_type", "") or "").strip().lower()
|
||||
if source == "fork" or relationship == "child_session":
|
||||
return None
|
||||
try:
|
||||
return get_session(parent_id, metadata_only=False)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _merged_webui_lineage_messages_for_display(
|
||||
session,
|
||||
messages=None,
|
||||
*,
|
||||
parent_session=_LINEAGE_PARENT_SESSION_UNSET,
|
||||
) -> list:
|
||||
"""Include immediate parent-only rows when a WebUI continuation sidecar is partial.
|
||||
|
||||
Compression/continuation sessions should render as one conversation. Most
|
||||
@@ -10076,23 +10104,9 @@ def _merged_webui_lineage_messages_for_display(session, messages=None) -> list:
|
||||
subset of their parent.
|
||||
"""
|
||||
primary_messages = list(messages if messages is not None else (getattr(session, "messages", []) or []))
|
||||
parent_id = str(getattr(session, "parent_session_id", "") or "").strip()
|
||||
if not parent_id:
|
||||
return primary_messages
|
||||
if (
|
||||
str(getattr(session, "compression_recovery_source_session_id", "") or "").strip()
|
||||
and str(getattr(session, "compression_recovery_action", "") or "").strip()
|
||||
):
|
||||
return primary_messages
|
||||
source = str(getattr(session, "session_source", "") or "").strip().lower()
|
||||
relationship = str(getattr(session, "relationship_type", "") or "").strip().lower()
|
||||
if source == "fork" or relationship == "child_session":
|
||||
return primary_messages
|
||||
try:
|
||||
parent = get_session(parent_id, metadata_only=False)
|
||||
except Exception:
|
||||
return primary_messages
|
||||
parent_messages = list(getattr(parent, "messages", []) or [])
|
||||
if parent_session is _LINEAGE_PARENT_SESSION_UNSET:
|
||||
parent_session = _webui_lineage_parent_session_for_display(session)
|
||||
parent_messages = list(getattr(parent_session, "messages", []) or [])
|
||||
if not parent_messages:
|
||||
return primary_messages
|
||||
if _messages_start_with_visible_prefix(primary_messages, parent_messages):
|
||||
@@ -10572,7 +10586,6 @@ from api.models import (
|
||||
new_session,
|
||||
all_sessions,
|
||||
title_from,
|
||||
_write_session_index,
|
||||
SESSION_INDEX_FILE,
|
||||
_active_state_db_path,
|
||||
load_projects,
|
||||
@@ -13333,9 +13346,11 @@ def _handle_session_get(handler, parsed) -> bool:
|
||||
state_db_messages,
|
||||
)
|
||||
sidecar_messages = _webui_sidecar_lineage_messages_for_display(s)
|
||||
lineage_parent = _webui_lineage_parent_session_for_display(s)
|
||||
projection_sidecar_messages = _merged_webui_lineage_messages_for_display(
|
||||
s,
|
||||
sidecar_messages,
|
||||
parent_session=lineage_parent,
|
||||
)
|
||||
_all_msgs = merge_session_messages_append_only(
|
||||
sidecar_messages,
|
||||
@@ -13343,7 +13358,11 @@ def _handle_session_get(handler, parsed) -> bool:
|
||||
truncation_watermark=getattr(s, "truncation_watermark", None),
|
||||
truncation_boundary=getattr(s, "truncation_boundary", None),
|
||||
)
|
||||
_all_msgs = _merged_webui_lineage_messages_for_display(s, _all_msgs)
|
||||
_all_msgs = _merged_webui_lineage_messages_for_display(
|
||||
s,
|
||||
_all_msgs,
|
||||
parent_session=lineage_parent,
|
||||
)
|
||||
_all_msgs = _project_native_image_payload_conflicts_for_display(
|
||||
projection_sidecar_messages,
|
||||
state_db_messages,
|
||||
|
||||
@@ -183,7 +183,12 @@ and 5; it does not mark every run-state boundary implemented.
|
||||
native-image and scalar text-attachment rows, and never changes model
|
||||
context. If multiple user rows share that timestamp, omit the whole
|
||||
ambiguous display bucket until the turn settles; keep all rows in model
|
||||
context. Do not identify the row by its text.
|
||||
context. Do not identify the row by its text. If a stream dies before
|
||||
settlement, state.db self-heal must save the submitted prompt and attachments
|
||||
as a visible sidecar row before clearing pending metadata; the Agent row
|
||||
remains available in model context. A partial continuation must use one
|
||||
consistent parent snapshot when projecting a conflicting provider payload
|
||||
onto its sidecar-owned display row.
|
||||
Match settled native-image scalar projections only with trusted turn and
|
||||
durable-row identity, never the marker alone. A durable row ID proves row
|
||||
identity, not provider-payload freshness: when the sidecar and state.db have
|
||||
|
||||
@@ -433,11 +433,23 @@ def test_get_session_projects_parent_only_payload_conflict_without_losing_parent
|
||||
response.update(payload=payload, status=status) or payload
|
||||
),
|
||||
)
|
||||
original_get_session = routes.get_session
|
||||
parent_loads = []
|
||||
|
||||
def get_session_with_fresh_parent(requested_id, metadata_only=False):
|
||||
if requested_id == parent_id:
|
||||
parent_session = models.Session.load(parent_id)
|
||||
parent_loads.append(parent_session)
|
||||
return parent_session
|
||||
return original_get_session(requested_id, metadata_only=metadata_only)
|
||||
|
||||
monkeypatch.setattr(routes, "get_session", get_session_with_fresh_parent)
|
||||
routes._handle_session_get(
|
||||
None,
|
||||
SimpleNamespace(path="/api/session", query=f"session_id={session_id}&resolve_model=0"),
|
||||
)
|
||||
|
||||
assert len(parent_loads) == 1
|
||||
assert response["status"] == 200
|
||||
public_messages = response["payload"]["session"]["messages"]
|
||||
mirror_rows = [message for message in public_messages if message.get("content") == mirror]
|
||||
@@ -452,6 +464,9 @@ def test_get_session_projects_parent_only_payload_conflict_without_losing_parent
|
||||
prefer_context=True,
|
||||
state_messages=state_rows,
|
||||
)
|
||||
assert "Unrelated parent-only row" not in [
|
||||
message.get("content") for message in replay_context
|
||||
]
|
||||
replay_rows = [
|
||||
message for message in _sanitize_messages_for_agent(replay_context)
|
||||
if message.get("content") == mirror
|
||||
|
||||
@@ -818,6 +818,83 @@ def test_get_session_syncs_sidecar_from_newer_state_db_even_when_stream_not_term
|
||||
assert reloaded.messages[-1]["content"] == "latest live progress"
|
||||
|
||||
|
||||
def test_get_session_sync_keeps_pending_image_prompt_when_agent_row_is_proven(monkeypatch, hermes_home):
|
||||
sid = "state_newer_pending_image_sid"
|
||||
stream_id = "dead_pending_image_stream"
|
||||
timestamp = time.time() - models._REPAIR_STALE_PENDING_GRACE_SECONDS - 5
|
||||
prompt = "Describe the attached image"
|
||||
attachment = {"name": "sample.png", "mime": "image/png", "is_image": True}
|
||||
agent_only_note = "Agent-only recall note"
|
||||
agent_user_content = [
|
||||
{"type": "text", "text": prompt},
|
||||
{"type": "image_url", "image_url": {"url": "data:image/png;base64,YWJj"}},
|
||||
{"type": "text", "text": agent_only_note},
|
||||
]
|
||||
previous_messages = [
|
||||
{"role": "user", "content": "old question", "timestamp": timestamp - 2},
|
||||
{"role": "assistant", "content": "old answer", "timestamp": timestamp - 1},
|
||||
]
|
||||
state_messages = [
|
||||
*previous_messages,
|
||||
{"role": "user", "content": agent_user_content, "timestamp": timestamp},
|
||||
{"role": "assistant", "content": "recovered assistant tail", "timestamp": timestamp + 1},
|
||||
]
|
||||
s = Session(
|
||||
session_id=sid,
|
||||
title="Pending image recovery",
|
||||
messages=previous_messages,
|
||||
context_messages=previous_messages,
|
||||
active_stream_id=stream_id,
|
||||
pending_user_message=prompt,
|
||||
pending_attachments=[attachment],
|
||||
pending_started_at=timestamp,
|
||||
pending_user_source="webui",
|
||||
_webui_pending_user_timestamp_identity=(stream_id, timestamp),
|
||||
)
|
||||
s.save()
|
||||
models.SESSIONS.pop(sid, None)
|
||||
|
||||
monkeypatch.setattr(
|
||||
models,
|
||||
"get_state_db_session_summary",
|
||||
lambda sid_arg, profile=None: {
|
||||
"message_count": len(state_messages),
|
||||
"last_message_at": timestamp + 1,
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
models,
|
||||
"get_state_db_session_messages",
|
||||
lambda sid_arg, **kwargs: list(state_messages),
|
||||
)
|
||||
|
||||
models.get_session(sid)
|
||||
reloaded = Session.load(sid)
|
||||
|
||||
submitted_rows = [
|
||||
message for message in reloaded.messages
|
||||
if message.get("role") == "user" and message.get("content") == prompt
|
||||
]
|
||||
assert len(submitted_rows) == 1
|
||||
assert submitted_rows[0]["timestamp"] == timestamp
|
||||
assert submitted_rows[0]["attachments"] == [attachment]
|
||||
assert reloaded.messages[-1] == state_messages[-1]
|
||||
assert agent_only_note not in str(reloaded.messages)
|
||||
|
||||
context_rows = models.reconciled_state_db_messages_for_session(
|
||||
reloaded,
|
||||
prefer_context=True,
|
||||
state_messages=state_messages,
|
||||
)
|
||||
enriched_rows = [
|
||||
message for message in context_rows
|
||||
if message.get("role") == "user" and message.get("timestamp") == timestamp
|
||||
]
|
||||
assert len(enriched_rows) == 1
|
||||
assert enriched_rows[0]["content"] == agent_user_content
|
||||
assert agent_only_note in str(enriched_rows[0]["content"])
|
||||
|
||||
|
||||
def test_get_session_does_not_sync_while_stream_is_still_live(monkeypatch):
|
||||
"""A still-running worker owns its own writeback; do not race it.
|
||||
|
||||
@@ -884,6 +961,7 @@ def test_sync_save_failure_does_not_mutate_session(monkeypatch):
|
||||
"""
|
||||
sid = "state_sync_save_fail_sid"
|
||||
stream_id = "stream_save_fail_no_done"
|
||||
attachment = {"name": "sample.png", "mime": "image/png", "is_image": True}
|
||||
s = Session(
|
||||
session_id=sid,
|
||||
title="Save failure",
|
||||
@@ -893,7 +971,10 @@ def test_sync_save_failure_does_not_mutate_session(monkeypatch):
|
||||
],
|
||||
active_stream_id=stream_id,
|
||||
pending_user_message="new request",
|
||||
pending_attachments=[attachment],
|
||||
pending_started_at=102.0,
|
||||
pending_user_source="webui",
|
||||
_webui_pending_user_timestamp_identity=(stream_id, 102.0),
|
||||
)
|
||||
s.save()
|
||||
|
||||
@@ -926,7 +1007,13 @@ def test_sync_save_failure_does_not_mutate_session(monkeypatch):
|
||||
assert result is False
|
||||
assert s.active_stream_id == stream_id
|
||||
assert s.pending_user_message == "new request"
|
||||
assert s.pending_attachments == [attachment]
|
||||
assert [m["content"] for m in s.messages] == ["old question", "old answer"]
|
||||
reloaded = Session.load(sid)
|
||||
assert reloaded.active_stream_id == stream_id
|
||||
assert reloaded.pending_user_message == "new request"
|
||||
assert reloaded.pending_attachments == s.pending_attachments
|
||||
assert [m["content"] for m in reloaded.messages] == ["old question", "old answer"]
|
||||
|
||||
|
||||
def test_sync_persists_recovered_state_db_tail_when_stream_dead(monkeypatch):
|
||||
|
||||
Reference in new issue
Block a user