Preserve processed Worklog duration during message merges

This commit is contained in:
Frank Song committed 2026-06-24 11:32:27 +08:00
1 parent e182a0ec54
commit cafeda63ca
5 files changed
+122

No files matched your search

+4
View File
@@ -3,6 +3,10 @@
## [Unreleased]
### Fixed
- **Settled Compact Worklog summaries keep their processed duration after session recovery or lineage merge.** When WebUI deduped duplicate assistant rows from the sidecar, state.db, or parent lineage, it could keep the copy without display-only turn metadata and drop the copy that carried `_turnDuration` / anchor scene metadata. Duplicate merges now backfill missing display metadata before discarding the duplicate, so the settled Processed Worklog can still show elapsed time.
## [v0.51.614] — 2026-06-23 — Release VU (Kanban consolidated view toggle)
### Added
+62
View File
@@ -4677,6 +4677,38 @@ def _session_message_merge_key(msg: dict):
)
_SESSION_MESSAGE_DISPLAY_METADATA_KEYS = (
"_turnDuration",
"_turnTps",
"_turnUsage",
"_firstTokenMs",
"_gatewayRouting",
"_statusCard",
"_anchor_stream_id",
"_anchor_activity_scene",
)
def _message_display_metadata_value_present(value) -> bool:
if value is None or value == "":
return False
if isinstance(value, (dict, list, tuple, set)) and not value:
return False
return True
def _merge_session_display_metadata(target: dict | None, source: dict | None) -> None:
"""Preserve display-only turn metadata when duplicate transcript rows merge."""
if not isinstance(target, dict) or not isinstance(source, dict):
return
for key in _SESSION_MESSAGE_DISPLAY_METADATA_KEYS:
if _message_display_metadata_value_present(target.get(key)):
continue
value = source.get(key)
if _message_display_metadata_value_present(value):
target[key] = copy.deepcopy(value)
def _session_message_dedup_key(msg: dict):
"""Like _session_message_merge_key but preserves full-precision timestamp.
@@ -5100,12 +5132,16 @@ def merge_session_messages_append_only(
# collapsing same-second distinct turns would be worse than retaining
# a compaction-restamped duplicate.
seen = set()
seen_messages = {}
deduped = []
for msg in filtered:
key = _session_message_dedup_key(msg)
if key not in seen:
seen.add(key)
seen_messages[key] = msg
deduped.append(msg)
else:
_merge_session_display_metadata(seen_messages.get(key), msg)
return deduped
merged_messages = []
@@ -5114,9 +5150,21 @@ def merge_session_messages_append_only(
seen_content_keys = set()
seen_visible_keys = set()
sidecar_visible_sequence = []
sidecar_visible_messages = []
sidecar_visible_keys = set()
sidecar_visible_counts = {}
merged_by_message_key = {}
merged_by_dedup_key = {}
merged_by_visible_key = {}
max_sidecar_timestamp = None
def _remember_merged_message(message):
if not isinstance(message, dict):
return
merged_by_message_key.setdefault(_session_message_merge_key(message), message)
merged_by_dedup_key.setdefault(_session_message_dedup_key(message), message)
merged_by_visible_key.setdefault(_session_message_visible_key(message), message)
for msg in sidecar_messages:
timestamp = _message_timestamp_as_float(msg)
if timestamp is not None:
@@ -5130,7 +5178,9 @@ def merge_session_messages_append_only(
sidecar_visible_keys.add(visible_key)
sidecar_visible_counts[visible_key] = sidecar_visible_counts.get(visible_key, 0) + 1
sidecar_visible_sequence.append(visible_key)
sidecar_visible_messages.append(msg)
merged_messages.append(msg)
_remember_merged_message(msg)
if _sidecar_has_terminal_partial_error(sidecar_messages):
return merged_messages
sidecar_visible_lookup = _build_visible_duplicate_lookup(sidecar_visible_keys)
@@ -5141,14 +5191,17 @@ def merge_session_messages_append_only(
key = _session_message_merge_key(msg)
visible_key = _session_message_visible_key(msg)
replays_sidecar_prefix = False
replay_target = None
if state_replay_idx < len(sidecar_visible_sequence):
expected_visible_key = sidecar_visible_sequence[state_replay_idx]
if visible_key == expected_visible_key or _has_visible_duplicate(
visible_key, {expected_visible_key}
):
replays_sidecar_prefix = True
replay_target = sidecar_visible_messages[state_replay_idx]
state_replay_idx += 1
if replays_sidecar_prefix:
_merge_session_display_metadata(replay_target, msg)
matched_visible_key = _matching_visible_duplicate(
visible_key,
sidecar_visible_keys,
@@ -5206,6 +5259,7 @@ def merge_session_messages_append_only(
# not.
dedup_key = _session_message_dedup_key(msg)
if dedup_key in seen_dedup_keys:
_merge_session_display_metadata(merged_by_dedup_key.get(dedup_key), msg)
continue
if max_sidecar_timestamp is not None and timestamp is not None and timestamp <= max_sidecar_timestamp:
# For message_id keys the merge key is authoritative — skip if
@@ -5213,6 +5267,7 @@ def merge_session_messages_append_only(
# handled true duplicates; same-second distinct messages must
# fall through.
if key in seen_message_keys and key[0] == "message_id":
_merge_session_display_metadata(merged_by_message_key.get(key), msg)
continue
if not (isinstance(key, tuple) and key[:1] == ("message_id",)):
# Legacy key within sidecar timestamp range — only skip if
@@ -5221,8 +5276,10 @@ def merge_session_messages_append_only(
# identical content/timestamp, so an unchecked continue here
# would drop legitimately distinct turns. (#3346 / PR #3665)
if key in seen_message_keys:
_merge_session_display_metadata(merged_by_message_key.get(key), msg)
continue
if key in seen_message_keys and key[0] == "message_id":
_merge_session_display_metadata(merged_by_message_key.get(key), msg)
continue
matched_visible_key = _matching_visible_duplicate(
visible_key,
@@ -5234,6 +5291,7 @@ def merge_session_messages_append_only(
sidecar_count = sidecar_visible_counts.get(matched_visible_key, 0)
if skipped_count < sidecar_count:
skipped_state_visible_counts[matched_visible_key] = skipped_count + 1
_merge_session_display_metadata(merged_by_visible_key.get(matched_visible_key), msg)
continue
# State rows at or before the newest sidecar timestamp are normally
# assumed to have already been observed by the sidecar. The <= gate
@@ -5262,6 +5320,7 @@ def merge_session_messages_append_only(
if _ck in seen_content_keys and dedup_key not in seen_dedup_keys:
pass # different tool_calls from sidecar — preserve
else:
_merge_session_display_metadata(merged_by_message_key.get(key), msg)
continue
else:
if msg.get("role") == "user" and _session_message_content_key(msg) not in seen_content_keys:
@@ -5270,13 +5329,16 @@ def merge_session_messages_append_only(
seen_dedup_keys.add(dedup_key)
seen_content_keys.add(_session_message_content_key(msg))
seen_visible_keys.add(visible_key)
_remember_merged_message(msg)
continue
_merge_session_display_metadata(merged_by_message_key.get(key), msg)
continue
seen_message_keys.add(key)
seen_dedup_keys.add(dedup_key)
seen_content_keys.add(_session_message_content_key(msg))
seen_visible_keys.add(visible_key)
merged_messages.append(msg)
_remember_merged_message(msg)
return merged_messages
+4
View File
@@ -6129,6 +6129,7 @@ def _merged_webui_lineage_messages_for_display(session, messages=None) -> list:
return primary_messages
merged_messages = []
seen_message_keys = set()
seen_messages_by_key = {}
for msg in sorted(list(parent_messages) + list(primary_messages), key=lambda m: (
float(m.get("timestamp") or 0),
str(m.get("role") or ""),
@@ -6136,8 +6137,10 @@ def _merged_webui_lineage_messages_for_display(session, messages=None) -> list:
)):
key = _session_message_merge_key(msg)
if key in seen_message_keys:
_merge_session_display_metadata(seen_messages_by_key.get(key), msg)
continue
seen_message_keys.add(key)
seen_messages_by_key[key] = msg
merged_messages.append(msg)
return merged_messages
@@ -6573,6 +6576,7 @@ from api.models import (
merge_session_messages_append_only,
_enrich_sidebar_lineage_metadata,
_active_stream_ids,
_merge_session_display_metadata,
_session_message_merge_key,
_session_message_visible_key,
_is_empty_partial_activity_message,
+31
View File
@@ -44,6 +44,37 @@ def test_webui_lineage_display_merge_includes_parent_only_rows(monkeypatch):
]
def test_webui_lineage_display_merge_preserves_duplicate_turn_duration(monkeypatch):
parent = SimpleNamespace(
session_id="parent",
messages=[
_msg("assistant", "final answer", 2000.0),
],
)
tip = SimpleNamespace(
session_id="tip",
parent_session_id="parent",
session_source="webui",
messages=[
{
"role": "assistant",
"content": "final answer",
"timestamp": 2000.0,
"_turnDuration": 502.765,
},
],
truncation_watermark=None,
)
monkeypatch.setattr(routes, "get_session", lambda sid, metadata_only=False: parent)
merged = routes._merged_webui_lineage_messages_for_display(tip, tip.messages)
assert len(merged) == 1
assert merged[0]["content"] == "final answer"
assert merged[0]["_turnDuration"] == 502.765
def test_webui_lineage_display_keeps_cumulative_child_tail_without_timestamps(monkeypatch):
parent = SimpleNamespace(
session_id="parent",
@@ -152,6 +152,27 @@ def test_historical_cancelled_partial_does_not_disable_later_state_db_merge():
]
def test_state_db_duplicate_backfills_turn_duration():
from api.models import merge_session_messages_append_only
sidecar = [
{"role": "assistant", "content": "final answer", "timestamp": 1001.0},
]
state = [
{
"role": "assistant",
"content": "final answer",
"timestamp": 1001.0,
"_turnDuration": 42.5,
},
]
merged = merge_session_messages_append_only(sidecar, state)
assert len(merged) == 1
assert merged[0]["_turnDuration"] == 42.5
def test_api_session_includes_state_db_messages_newer_than_webui_sidecar(monkeypatch, tmp_path):
import api.routes as routes