simplify(agent): sidecar backfill — drop the hasattr guard and the duplicated row-id predicate; tests 7→6
_session_db is always a SessionDB (agent_init / delegate_tool), so the "fail closed on a store wrapper" hasattr was defense around code that cannot fail; the store's own guard binds the value into SQL, so the prologue only needs the sibling idiom isinstance(_row_id, int) that session_persistence and transcript_repair already use. The positional hazard is explained once, on set_latest_user_api_content. The in-place compaction test duplicated test_api_content_sidecar's test_inplace_compaction_backfills_sidecar_into_db verbatim (its row_id parameter was never varied); dropped, as was the positional-helper tail of test_older_identical_row_is_untouched already covered there.
This commit is contained in:
kshitijk4poor
@@ -760,9 +760,7 @@ def _stamp_api_content_sidecar(
|
||||
# clean content and the request prefix diverges here. Both writers stamp ``_row_id`` on the live
|
||||
# dict, which is at once the proof a row exists and the address to update.
|
||||
#
|
||||
# Do NOT widen this to an unconditional backfill: without a row id the store can only target the
|
||||
# newest active user row, and a repeated user turn ("ok", "y", "continue") makes the PREVIOUS
|
||||
# turn's row compare equal — this turn's bytes would overwrite its sidecar for good.
|
||||
# Never widen this to an unconditional positional backfill — see set_latest_user_api_content.
|
||||
#
|
||||
# ``_row_id`` is read under ``_session_persist_lock``: a close flush holds it while it commits
|
||||
# the row and only then writes ``_row_id`` back (``sync_flushed_message_markers``). Read outside
|
||||
@@ -770,17 +768,13 @@ def _stamp_api_content_sidecar(
|
||||
# persisted with ``api_content = NULL``, leaving no writer to correct the row.
|
||||
with _persist_lock(agent):
|
||||
_row_id = _turn_user_msg.get("_row_id")
|
||||
_has_valid_row_id = isinstance(_row_id, int) and not isinstance(_row_id, bool) and _row_id > 0
|
||||
_in_place_compacted = preflight_compressed and bool(getattr(agent, "_last_compaction_in_place", False))
|
||||
_db = getattr(agent, "_session_db", None)
|
||||
if _db is None or not (_has_valid_row_id or _in_place_compacted):
|
||||
if _db is None or not (isinstance(_row_id, int) or _in_place_compacted):
|
||||
return
|
||||
try:
|
||||
if _has_valid_row_id:
|
||||
# Fail closed on a store wrapper without the row-addressed method: a positional
|
||||
# fallback here is exactly the wrong-row write this function exists to prevent.
|
||||
if hasattr(_db, "set_message_api_content"):
|
||||
_db.set_message_api_content(agent.session_id, _row_id, durable_content, _api_content)
|
||||
if isinstance(_row_id, int):
|
||||
_db.set_message_api_content(agent.session_id, _row_id, durable_content, _api_content)
|
||||
else:
|
||||
# Compacted copies carry no row id; positional is safe only because
|
||||
# archive_and_compact just made this message the newest active user row.
|
||||
|
||||
@@ -57,12 +57,6 @@ class TestSetMessageApiContent:
|
||||
assert rows[turn_1_id]["api_content"] == "ok\n\nTURN-1"
|
||||
assert rows[turn_2_id]["api_content"] == "ok\n\nTURN-2"
|
||||
|
||||
# The positional helper is only safe when the caller already knows
|
||||
# the newest active user row is its own message.
|
||||
db.set_latest_user_api_content("s1", "ok", "ok\n\nTURN-3")
|
||||
rows = {r["id"]: r for r in db.get_messages("s1")}
|
||||
assert rows[turn_2_id]["api_content"] == "ok\n\nTURN-3"
|
||||
assert rows[turn_1_id]["api_content"] == "ok\n\nTURN-1"
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@@ -106,19 +100,6 @@ class TestPrologueRowAddressedBackfill:
|
||||
agent._session_db.set_message_api_content.assert_not_called()
|
||||
agent._session_db.set_latest_user_api_content.assert_not_called()
|
||||
|
||||
def test_compaction_without_row_id_keeps_positional_fallback(self):
|
||||
agent = _make_in_place_compaction_agent(row_id=None)
|
||||
with patch(
|
||||
"hermes_cli.plugins.invoke_hook",
|
||||
return_value=[{"context": "PLUGIN-CTX"}],
|
||||
):
|
||||
_build(agent)
|
||||
agent._session_db.set_latest_user_api_content.assert_called_once_with(
|
||||
"sess-1", "hello", "hello\n\nPLUGIN-CTX"
|
||||
)
|
||||
agent._session_db.set_message_api_content.assert_not_called()
|
||||
|
||||
|
||||
class _RealPersistenceAgent(SessionPersistenceMixin, _FakeAgent):
|
||||
"""Stand-in agent with the real SessionPersistenceMixin flush implementation."""
|
||||
|
||||
@@ -255,43 +236,3 @@ class TestRealEarlyFlushAndOverrideLifecycle:
|
||||
assert rows[t2_user_row["id"]]["api_content"] == "ok\n\nTURN-2-CTX"
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def _make_in_place_compaction_agent(*, row_id):
|
||||
"""Agent whose preflight compression compacts in place, mirroring
|
||||
``archive_and_compact``: the current-turn user dict is replaced by a fresh
|
||||
copy whose row already exists (and carries ``_row_id`` when the insert
|
||||
stamped one)."""
|
||||
agent = _FakeAgent()
|
||||
agent.compression_enabled = True
|
||||
agent._session_db = MagicMock()
|
||||
|
||||
calls = {"n": 0}
|
||||
|
||||
def _should_compress(_tokens):
|
||||
calls["n"] += 1
|
||||
return calls["n"] == 1
|
||||
|
||||
agent.context_compressor = types.SimpleNamespace(
|
||||
protect_first_n=0,
|
||||
protect_last_n=0,
|
||||
threshold_tokens=1,
|
||||
context_length=1000,
|
||||
last_prompt_tokens=-1,
|
||||
should_compress=_should_compress,
|
||||
should_defer_preflight_to_real_usage=lambda _t: False,
|
||||
get_active_compression_failure_cooldown=lambda: None,
|
||||
)
|
||||
|
||||
def _compress(messages, _system, approx_tokens=None, task_id=None):
|
||||
agent._last_compaction_in_place = True
|
||||
survivor = dict(messages[-1])
|
||||
if row_id is not None:
|
||||
survivor["_row_id"] = row_id
|
||||
return (
|
||||
[{"role": "assistant", "content": "compaction summary"}, survivor],
|
||||
"SYSTEM",
|
||||
)
|
||||
|
||||
agent._compress_context = _compress
|
||||
return agent
|
||||
|
||||
Reference in New Issue
Block a user