fix(#4767): guard empty-sidecar reconstruction against not-advanced watermark (Opus gate)
Opus deep-gate found a latent data-loss the Codex harnesses missed: the empty-sidecar branch kept at_or_after (ts>=watermark) UNCONDITIONALLY, which is only valid once the watermark was ADVANCED strictly past the original cutoff (a new turn committed -> boundary < watermark). In the not-advanced states the content above the watermark is the deleted suffix, so it got resurrected — the same data-loss class this fix exists to kill. Reachable via crash/cold-load metadata-vs-sidecar divergence. - Keep at_or_after ONLY when boundary < watermark (genuinely advanced). - boundary == watermark (just truncated) OR boundary is None (legacy: a positive watermark always meant 'frozen at cutoff' pre-#4767) -> conservative ts<=watermark filter; never resurrects. - Replaced the backward-scan heuristic with the provable boundary guard. - Removed the now-dead 'or (not sidecar_messages and watermark>0)' clause in the non-empty-sidecar path (unreachable; empty sidecar returns earlier). - Added 2 regression tests (boundary==watermark, legacy None-boundary), both proven non-vacuous; updated 2 contributor tests that assumed advanced-watermark-without-boundary (an unreachable state) to pass the boundary the real persistence path always writes. Co-authored-by: AlexeyDsov <alexeydsov@gmail.com>
This commit is contained in:
1 parent
6ce9533d53
commit
7317e29d82
4 files changed
+114
-45
No files matched your search
+36
-39
@@ -5135,45 +5135,48 @@ def merge_session_messages_append_only(
|
|||||||
# Truncate-to-empty sentinel (#2914) — block all replay.
|
# Truncate-to-empty sentinel (#2914) — block all replay.
|
||||||
return []
|
return []
|
||||||
else:
|
else:
|
||||||
# Positive watermark advanced after edit/retry/undo (#4767).
|
# Positive watermark after edit/retry/undo (#4767). Without a
|
||||||
# Without a sidecar there's no seen_content_keys to check against,
|
# sidecar there's no seen_content_keys to check against, so we
|
||||||
# so we reconstruct the correct transcript from state.db alone.
|
# reconstruct the correct transcript from state.db alone.
|
||||||
#
|
#
|
||||||
# Use truncation_boundary (the original truncate cutoff) to
|
# `at_or_after` (ts >= watermark) is legitimate POST-EDIT content
|
||||||
# distinguish legitimate prefix from deleted suffix. Without it,
|
# ONLY when the watermark was ADVANCED strictly past the original
|
||||||
# fall back to the backward-scan heuristic (drops last user+assistant
|
# truncate cutoff — i.e. a new turn was committed after the edit, so
|
||||||
# pair) which works for the common single-turn case.
|
# truncation_boundary (the original cutoff) is strictly below the
|
||||||
|
# advanced watermark. In that state we keep the legitimate prefix
|
||||||
|
# (ts <= boundary) plus the post-edit tail (ts >= watermark) and drop
|
||||||
|
# the deleted (boundary, watermark) suffix.
|
||||||
|
#
|
||||||
|
# In every OTHER state the content above the watermark is the deleted
|
||||||
|
# suffix, NOT post-edit content, so keeping it would resurrect deleted
|
||||||
|
# turns (the exact data-loss this fix exists to kill):
|
||||||
|
# * boundary == watermark — just truncated, no new turn committed
|
||||||
|
# yet (e.g. crash/cold-load with metadata-vs-sidecar divergence);
|
||||||
|
# * boundary is None — legacy session saved before this field
|
||||||
|
# existed. In the pre-#4767 model committing a turn CLEARED the
|
||||||
|
# watermark to None, so a persisted positive watermark always
|
||||||
|
# meant "frozen at cutoff, not advanced".
|
||||||
|
# For all of those, fall back to the conservative pre-#4767 filter
|
||||||
|
# `ts <= watermark`, which never resurrects a deleted suffix.
|
||||||
boundary_ts = _message_timestamp_as_float({"timestamp": truncation_boundary})
|
boundary_ts = _message_timestamp_as_float({"timestamp": truncation_boundary})
|
||||||
at_or_after = []
|
if boundary_ts is not None and boundary_ts < watermark_timestamp:
|
||||||
pre_watermark = []
|
|
||||||
for msg in state_messages:
|
|
||||||
ts = _message_timestamp_as_float(msg)
|
|
||||||
if ts is not None and ts >= watermark_timestamp:
|
|
||||||
at_or_after.append(msg)
|
|
||||||
else:
|
|
||||||
pre_watermark.append(msg)
|
|
||||||
if boundary_ts is not None:
|
|
||||||
# Use the persisted boundary: keep only messages at or before it.
|
|
||||||
pre_legitimate = [
|
pre_legitimate = [
|
||||||
m for m in pre_watermark
|
m for m in state_messages
|
||||||
if (ts := _message_timestamp_as_float(m)) is not None
|
if (ts := _message_timestamp_as_float(m)) is not None
|
||||||
and ts <= boundary_ts
|
and ts <= boundary_ts
|
||||||
]
|
]
|
||||||
|
at_or_after = [
|
||||||
|
m for m in state_messages
|
||||||
|
if (ts := _message_timestamp_as_float(m)) is not None
|
||||||
|
and ts >= watermark_timestamp
|
||||||
|
]
|
||||||
filtered = pre_legitimate + at_or_after
|
filtered = pre_legitimate + at_or_after
|
||||||
else:
|
else:
|
||||||
# Fallback: backward scan (works when exactly one turn was
|
filtered = [
|
||||||
# deleted — the common edit/retry/undo case).
|
m for m in state_messages
|
||||||
i = len(pre_watermark) - 1
|
if (ts := _message_timestamp_as_float(m)) is not None
|
||||||
while i >= 0:
|
and ts <= watermark_timestamp
|
||||||
role = str(pre_watermark[i].get("role", "")).lower()
|
]
|
||||||
if role == "assistant":
|
|
||||||
i -= 1
|
|
||||||
elif role == "user":
|
|
||||||
i -= 1 # skip this user message too (the replaced prompt)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
i -= 1 # tool messages etc. — skip
|
|
||||||
filtered = pre_watermark[:i + 1] + at_or_after
|
|
||||||
|
|
||||||
# Deduplicate true duplicates (same role, content, exact timestamp)
|
# Deduplicate true duplicates (same role, content, exact timestamp)
|
||||||
# without collapsing legitimately-repeated identical turns (#3346).
|
# without collapsing legitimately-repeated identical turns (#3346).
|
||||||
@@ -5270,14 +5273,8 @@ def merge_session_messages_append_only(
|
|||||||
# advanced), allow state rows newer than the sidecar tail to merge.
|
# advanced), allow state rows newer than the sidecar tail to merge.
|
||||||
sidecar_advanced_past_watermark = (
|
sidecar_advanced_past_watermark = (
|
||||||
watermark_timestamp is not None
|
watermark_timestamp is not None
|
||||||
and (
|
and max_sidecar_timestamp is not None
|
||||||
(max_sidecar_timestamp is not None
|
and max_sidecar_timestamp > watermark_timestamp
|
||||||
and max_sidecar_timestamp > watermark_timestamp)
|
|
||||||
# If the sidecar is empty but watermark > 0, the session has
|
|
||||||
# advanced (a new user turn was committed). Treat this as
|
|
||||||
# advanced so post-edit state.db rows are not dropped.
|
|
||||||
or (not sidecar_messages and watermark_timestamp > 0)
|
|
||||||
)
|
|
||||||
)
|
)
|
||||||
if (
|
if (
|
||||||
watermark_timestamp is not None
|
watermark_timestamp is not None
|
||||||
|
|||||||
@@ -73,7 +73,9 @@ def test_core_a_empty_sidecar_resurrects_deleted_turns():
|
|||||||
|
|
||||||
|
|
||||||
def test_core_a_single_deleted_turn_still_works():
|
def test_core_a_single_deleted_turn_still_works():
|
||||||
"""Sanity check: backward-scan handles one deleted turn correctly."""
|
"""Sanity check: an advanced watermark with the original cutoff persisted
|
||||||
|
as truncation_boundary keeps the prefix + post-edit tail and drops the one
|
||||||
|
deleted turn."""
|
||||||
state = [
|
state = [
|
||||||
_msg("user", "first msg", 50.0),
|
_msg("user", "first msg", 50.0),
|
||||||
_msg("assistant", "first reply", 51.0),
|
_msg("assistant", "first reply", 51.0),
|
||||||
@@ -83,8 +85,10 @@ def test_core_a_single_deleted_turn_still_works():
|
|||||||
_msg("assistant", "post-edit reply", 201.0),
|
_msg("assistant", "post-edit reply", 201.0),
|
||||||
]
|
]
|
||||||
|
|
||||||
|
# Original cutoff kept through the first reply (@51); a new turn was then
|
||||||
|
# committed (@200), advancing the watermark. boundary (51) < watermark (200).
|
||||||
merged = models.merge_session_messages_append_only(
|
merged = models.merge_session_messages_append_only(
|
||||||
[], state, truncation_watermark=200.0
|
[], state, truncation_watermark=200.0, truncation_boundary=51.0
|
||||||
)
|
)
|
||||||
|
|
||||||
contents = [m["content"] for m in merged]
|
contents = [m["content"] for m in merged]
|
||||||
@@ -96,6 +100,59 @@ def test_core_a_single_deleted_turn_still_works():
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_core_a_not_advanced_watermark_equals_boundary_does_not_resurrect():
|
||||||
|
"""Just-truncated state (boundary == watermark, no new turn committed yet):
|
||||||
|
everything above the watermark is the deleted suffix and must NOT be kept.
|
||||||
|
|
||||||
|
Reachable via crash/cold-load metadata-vs-sidecar divergence — the exact
|
||||||
|
scenario the empty-sidecar branch serves. Opus gate found that the prior
|
||||||
|
code swept the suffix into `at_or_after` here, resurrecting deleted turns
|
||||||
|
(the same data-loss class this fix exists to kill)."""
|
||||||
|
state = [
|
||||||
|
_msg("user", "u1", 50.0),
|
||||||
|
_msg("assistant", "a1", 51.0),
|
||||||
|
_msg("user", "deleted-u2", 100.0),
|
||||||
|
_msg("assistant", "deleted-a2", 101.0),
|
||||||
|
_msg("user", "deleted-u3", 150.0),
|
||||||
|
_msg("assistant", "deleted-a3", 151.0),
|
||||||
|
]
|
||||||
|
|
||||||
|
merged = models.merge_session_messages_append_only(
|
||||||
|
[], state, truncation_watermark=51.0, truncation_boundary=51.0
|
||||||
|
)
|
||||||
|
|
||||||
|
contents = [m["content"] for m in merged]
|
||||||
|
assert contents == ["u1", "a1"], (
|
||||||
|
f"watermark==boundary must keep only ts<=watermark, got {contents}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_core_a_legacy_none_boundary_frozen_watermark_does_not_resurrect():
|
||||||
|
"""Legacy session (truncation_boundary is None) with a positive watermark:
|
||||||
|
in the pre-#4767 model a persisted positive watermark always meant
|
||||||
|
'frozen at cutoff' (committing a turn cleared it to None), so the safe
|
||||||
|
behavior is the conservative ts<=watermark filter — no resurrection, and
|
||||||
|
the legitimately-kept first turn survives."""
|
||||||
|
state = [
|
||||||
|
_msg("user", "u1", 50.0),
|
||||||
|
_msg("assistant", "a1", 51.0),
|
||||||
|
_msg("user", "deleted-u2", 100.0),
|
||||||
|
_msg("assistant", "deleted-a2", 101.0),
|
||||||
|
_msg("user", "deleted-u3", 150.0),
|
||||||
|
_msg("assistant", "deleted-a3", 151.0),
|
||||||
|
]
|
||||||
|
|
||||||
|
merged = models.merge_session_messages_append_only(
|
||||||
|
[], state, truncation_watermark=51.0 # boundary defaults to None
|
||||||
|
)
|
||||||
|
|
||||||
|
contents = [m["content"] for m in merged]
|
||||||
|
assert contents == ["u1", "a1"], (
|
||||||
|
f"legacy None-boundary frozen watermark must keep only ts<=watermark "
|
||||||
|
f"(no resurrection, keep u1), got {contents}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
# ─── Same-second assistant reply dropped ──────────────────────────────────────
|
# ─── Same-second assistant reply dropped ──────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -138,7 +138,13 @@ def test_advanced_watermark_keeps_post_edit_state_rows():
|
|||||||
def test_active_watermark_still_filters_replaced_tail_empty_sidecar():
|
def test_active_watermark_still_filters_replaced_tail_empty_sidecar():
|
||||||
"""A positive watermark (advanced to the new user-turn timestamp) with an
|
"""A positive watermark (advanced to the new user-turn timestamp) with an
|
||||||
empty sidecar must suppress the replaced pre-edit tail while keeping
|
empty sidecar must suppress the replaced pre-edit tail while keeping
|
||||||
post-edit rows (#4767 / CORE finding #2)."""
|
post-edit rows (#4767 / CORE finding #2).
|
||||||
|
|
||||||
|
A genuinely-advanced session persists the original cutoff as
|
||||||
|
``truncation_boundary`` (@100, the last kept reply before the replaced tail)
|
||||||
|
alongside the advanced watermark (@200, the new committed turn). The
|
||||||
|
reconstruction keeps ts <= boundary plus ts >= watermark and drops the
|
||||||
|
replaced (boundary, watermark) suffix."""
|
||||||
state = _rows(
|
state = _rows(
|
||||||
("user", "q1", 50),
|
("user", "q1", 50),
|
||||||
("assistant", "a1", 100),
|
("assistant", "a1", 100),
|
||||||
@@ -148,7 +154,7 @@ def test_active_watermark_still_filters_replaced_tail_empty_sidecar():
|
|||||||
("assistant", "a2-new", 300), # post-edit reply -> kept
|
("assistant", "a2-new", 300), # post-edit reply -> kept
|
||||||
)
|
)
|
||||||
merged = models.merge_session_messages_append_only(
|
merged = models.merge_session_messages_append_only(
|
||||||
[], state, truncation_watermark=200.0
|
[], state, truncation_watermark=200.0, truncation_boundary=100.0
|
||||||
)
|
)
|
||||||
assert [m["content"] for m in merged] == ["q1", "a1", "edited-q2", "a2-new"]
|
assert [m["content"] for m in merged] == ["q1", "a1", "edited-q2", "a2-new"]
|
||||||
|
|
||||||
|
|||||||
@@ -353,7 +353,16 @@ def test_empty_sidecar_advanced_watermark_no_ghost_no_data_loss():
|
|||||||
resurrecting stale pre-edit ghosts (#4767).
|
resurrecting stale pre-edit ghosts (#4767).
|
||||||
|
|
||||||
Reproduction: cold reload / crash recovery before sidecar persists.
|
Reproduction: cold reload / crash recovery before sidecar persists.
|
||||||
state.db has the full lineage including replaced pre-edit rows."""
|
state.db has the full lineage including replaced pre-edit rows.
|
||||||
|
|
||||||
|
A genuinely-advanced session persists the original cutoff as
|
||||||
|
``truncation_boundary`` (here @51, the last kept reply) while the watermark
|
||||||
|
is advanced to the new committed turn (@200). The reconstruction keeps the
|
||||||
|
legitimate prefix (ts <= boundary) plus the post-edit tail (ts >= watermark)
|
||||||
|
and drops the replaced (boundary, watermark) suffix. Passing the boundary is
|
||||||
|
required — an advanced watermark with NO boundary is the unreachable
|
||||||
|
legacy/frozen state (see the not-advanced regression tests in
|
||||||
|
test_core_data_loss_cases.py)."""
|
||||||
state = [
|
state = [
|
||||||
{"role": "user", "content": "first msg", "timestamp": 50},
|
{"role": "user", "content": "first msg", "timestamp": 50},
|
||||||
{"role": "assistant", "content": "first reply", "timestamp": 51},
|
{"role": "assistant", "content": "first reply", "timestamp": 51},
|
||||||
@@ -363,7 +372,7 @@ def test_empty_sidecar_advanced_watermark_no_ghost_no_data_loss():
|
|||||||
{"role": "assistant", "content": "post-edit reply", "timestamp": 201},
|
{"role": "assistant", "content": "post-edit reply", "timestamp": 201},
|
||||||
]
|
]
|
||||||
merged = models.merge_session_messages_append_only(
|
merged = models.merge_session_messages_append_only(
|
||||||
[], state, truncation_watermark=200.0
|
[], state, truncation_watermark=200.0, truncation_boundary=51.0
|
||||||
)
|
)
|
||||||
contents = [m["content"] for m in merged]
|
contents = [m["content"] for m in merged]
|
||||||
wanted = ["first msg", "first reply", "edited/new turn", "post-edit reply"]
|
wanted = ["first msg", "first reply", "edited/new turn", "post-edit reply"]
|
||||||
|
|||||||
Reference in new issue
Block a user