From 61f841692280dbb33d3180c39d51095e7650198c Mon Sep 17 00:00:00 2001 From: Guangyu Date: Wed, 29 Jul 2026 05:37:43 +0000 Subject: [PATCH 1/3] fix(playbook): restore apply_playbook_edit, deleted while still imported MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #385 removed `playbook_edit_apply.py` as part of "remove the obsolete split-edit application path", but that module was not the split path — it is the shared atomic supersede primitive, and `review_service.py` still imports it at module scope and calls it at line 297. `apply_playbook_edit` exists nowhere else on main, so no replacement was shipped. `review_service.py` is imported by `server/routes/playbooks.py`, which loads at app startup, making this a guaranteed ModuleNotFoundError on boot. It was caught when a production deploy built from a clean checkout of main: the task started, failed its container health checks on the API port, and the deployment circuit breaker rolled the service back. Builds from working trees that still had the file on disk masked it. The whole `tests/server/services/playbook` directory currently fails at collection on main for the same reason. Restored verbatim from 4268ad1^; every dependency it uses (`supersede_record`, `commit_scope`, `save_user_playbooks`, `precompute_user_playbook_embeddings`, `LineageContext`) still exists. --- .../services/playbook/playbook_edit_apply.py | 87 +++++++++++++++++++ 1 file changed, 87 insertions(+) create mode 100644 reflexio/server/services/playbook/playbook_edit_apply.py diff --git a/reflexio/server/services/playbook/playbook_edit_apply.py b/reflexio/server/services/playbook/playbook_edit_apply.py new file mode 100644 index 00000000..7dbc784b --- /dev/null +++ b/reflexio/server/services/playbook/playbook_edit_apply.py @@ -0,0 +1,87 @@ +"""Shared atomic supersede primitive for applying a playbook edit. + +Online and background playbook repair paths share one lifecycle +(insert-then-supersede, no orphan). +""" + +from typing import TYPE_CHECKING + +from reflexio.models.api_schema.domain.entities import LineageContext, UserPlaybook + +if TYPE_CHECKING: + from reflexio.server.services.storage.storage_base import BaseStorage + + +class _LostSupersedeRaceError(Exception): + """Internal signal used to roll back a provisional successor.""" + + +def apply_playbook_edit( + storage: "BaseStorage", + *, + incumbent_id: int, + new_playbook: UserPlaybook, + source: str, + request_id: str, + skip_embedding: bool = False, + revise_context: LineageContext | None = None, +) -> int: + """Insert a replacement playbook then atomically supersede the incumbent. + + Uses ``storage.supersede_record`` (atomic conditional CAS) so a lost race + never leaves an orphan CURRENT row: + + - Insert the new playbook as CURRENT. + - Call ``supersede_record(incumbent_id → new_id)``, which only succeeds when + the incumbent is still CURRENT (``status IS NULL``). + - If ``supersede_record`` returns ``False`` (incumbent already gone), roll + back the transaction and return ``-1``. + + Args: + storage: A BaseStorage instance providing ``save_user_playbooks``, + ``supersede_record``, and ``commit_scope``. + incumbent_id: ``user_playbook_id`` of the playbook being replaced. + new_playbook: The replacement playbook (inserted as CURRENT, i.e. + ``status=None``). + source: Provenance label stored on the new playbook row and in the + lineage event actor field. + request_id: Operation-run correlation id for the lineage event. Must be + non-empty; use an operation-scoped id. Raises ``ValueError`` + immediately (before any storage write) when empty, preventing + orphaned successor rows. + skip_embedding: Forwarded to ``save_user_playbooks``. Defaults to + ``False`` (precompute the embedding before opening the transaction). + + Returns: + The ``user_playbook_id`` of the newly inserted playbook, or ``-1`` if + the incumbent was not CURRENT (no mutation; no orphan left behind). + + Raises: + ValueError: If ``request_id`` is empty or None. + """ + if not request_id: + raise ValueError( + "apply_playbook_edit: request_id must be non-empty (operation-run correlation id)" + ) + new_playbook.source = source + if not skip_embedding: + storage.precompute_user_playbook_embeddings([new_playbook]) + + ctx = revise_context or LineageContext( + op_kind="revise", actor=source, request_id=request_id + ) + try: + with storage.commit_scope(): + storage.save_user_playbooks([new_playbook], skip_embedding=True) + new_id = new_playbook.user_playbook_id + if not storage.supersede_record( + entity_type="user_playbook", + incumbent_id=str(incumbent_id), + successor_id=str(new_id), + context=ctx, + ): + raise _LostSupersedeRaceError + except _LostSupersedeRaceError: + new_playbook.user_playbook_id = 0 + return -1 + return new_id From 15bcd29be3e2191c3a6d9e74bb5a428b5821dae2 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Wed, 29 Jul 2026 05:43:59 +0000 Subject: [PATCH 2/3] test(playbook): restore the apply_playbook_edit unit tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #385 deleted `tests/server/services/playbook/test_playbook_edit_apply.py` alongside the module it covers. Restoring only the source left `apply_playbook_edit` with no direct unit coverage — its behaviour was exercised only indirectly, through `review_service`. Restored verbatim from 4268ad1^; all 8 tests pass unmodified against current main, which independently confirms the supersede CAS semantics the restored module relies on are unchanged. --- .../playbook/test_playbook_edit_apply.py | 263 ++++++++++++++++++ 1 file changed, 263 insertions(+) create mode 100644 tests/server/services/playbook/test_playbook_edit_apply.py diff --git a/tests/server/services/playbook/test_playbook_edit_apply.py b/tests/server/services/playbook/test_playbook_edit_apply.py new file mode 100644 index 00000000..cee4198e --- /dev/null +++ b/tests/server/services/playbook/test_playbook_edit_apply.py @@ -0,0 +1,263 @@ +"""Tests for apply_playbook_edit() — the shared archive+insert primitive.""" + +import tempfile +from unittest.mock import patch + +import pytest + +from reflexio.models.api_schema.domain.entities import UserPlaybook +from reflexio.server.services.storage.sqlite_storage import SQLiteStorage +from reflexio.server.services.storage.sqlite_storage._lineage import ( + _EMPTY_REQUEST_ID_MSG, +) + + +def _storage(tmp: str) -> SQLiteStorage: + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + return SQLiteStorage(org_id="org_apply_1", db_path=f"{tmp}/t.db") + + +def _playbook(user_id: str = "u1", content: str = "old") -> UserPlaybook: + return UserPlaybook( + user_id=user_id, + agent_version="v1", + request_id="req_test", + playbook_name="refund", + content=content, + trigger="refund", + ) + + +def test_apply_inserts_new_and_archives_incumbent(): + from reflexio.server.services.playbook.playbook_edit_apply import ( + apply_playbook_edit, + ) + + with tempfile.TemporaryDirectory() as tmp: + s = _storage(tmp) + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + old = _playbook(content="old") + s.save_user_playbooks([old]) + old_id = old.user_playbook_id + assert old_id > 0 + + new = _playbook(content="new") + new_id = apply_playbook_edit( + s, + incumbent_id=old_id, + new_playbook=new, + source="offline_optimizer", + request_id="run-abc", + ) + assert new_id > 0 + + # Only new_id should be CURRENT (status=None); old_id should be archived + all_pbs = s.get_user_playbooks(user_id="u1") + current_ids = {p.user_playbook_id for p in all_pbs if p.status is None} + assert new_id in current_ids + assert old_id not in current_ids + + +def test_apply_skips_archive_when_incumbent_not_current(): + from reflexio.server.services.playbook.playbook_edit_apply import ( + apply_playbook_edit, + ) + + with tempfile.TemporaryDirectory() as tmp: + s = _storage(tmp) + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + old = _playbook(content="old") + s.save_user_playbooks([old]) + old_id = old.user_playbook_id + assert old_id > 0 + + # Someone else archived it first + s.archive_user_playbook_by_id(user_id="u1", user_playbook_id=old_id) + + new = _playbook(content="new") + new_id = apply_playbook_edit( + s, + incumbent_id=old_id, + new_playbook=new, + source="offline_optimizer", + request_id="run-abc", + ) + # Optimistic-concurrency: incumbent was already archived → skip and return -1 + assert new_id == -1 + + +def test_apply_expect_current_false_archives(): + """With expect_current=False, new playbook is inserted and incumbent is archived.""" + from reflexio.server.services.playbook.playbook_edit_apply import ( + apply_playbook_edit, + ) + + with tempfile.TemporaryDirectory() as tmp: + s = _storage(tmp) + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + old = _playbook(content="old") + s.save_user_playbooks([old]) + old_id = old.user_playbook_id + assert old_id > 0 + + new = _playbook(content="new") + new_id = apply_playbook_edit( + s, + incumbent_id=old_id, + new_playbook=new, + source="offline_optimizer", + request_id="run-abc", + ) + assert new_id > 0 + + # Only new_id should be CURRENT (status=None); old_id should be archived + all_pbs = s.get_user_playbooks(user_id="u1") + current_ids = {p.user_playbook_id for p in all_pbs if p.status is None} + assert new_id in current_ids + assert old_id not in current_ids + + +def test_apply_expect_current_false_returns_minus1_and_no_orphan(): + """When incumbent is already archived, supersede_record returns False. + + The transaction rolls back the provisional successor and its create event, + so the -1 return value indicates the lost race, not an orphan. + """ + from reflexio.server.services.playbook.playbook_edit_apply import ( + apply_playbook_edit, + ) + + with tempfile.TemporaryDirectory() as tmp: + s = _storage(tmp) + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + old = _playbook(content="old") + s.save_user_playbooks([old]) + old_id = old.user_playbook_id + assert old_id > 0 + + # Archive first so supersede_record will return False + s.archive_user_playbook_by_id(user_id="u1", user_playbook_id=old_id) + event_ids_before = { + event.event_id for event in s.get_lineage_events(org_id="org_apply_1") + } + + new = _playbook(content="new") + new_id = apply_playbook_edit( + s, + incumbent_id=old_id, + new_playbook=new, + source="offline_optimizer", + request_id="run-abc", + ) + # supersede_record returned False → -1, transaction rolled back. + assert new_id == -1 + + # No orphan: the inserted successor was deleted + all_pbs = s.get_user_playbooks(user_id="u1") + current_ids = {p.user_playbook_id for p in all_pbs if p.status is None} + assert len(current_ids) == 0 + assert { + event.event_id for event in s.get_lineage_events(org_id="org_apply_1") + } == event_ids_before + + +def test_apply_raises_on_empty_request_id_before_write(): + """apply_playbook_edit raises ValueError on empty request_id before any storage write. + + The I2 (orphan) guard: an empty request_id is rejected immediately so no + successor row is ever inserted when the caller forgets to supply a run id. + """ + from reflexio.server.services.playbook.playbook_edit_apply import ( + apply_playbook_edit, + ) + + with tempfile.TemporaryDirectory() as tmp: + s = _storage(tmp) + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + old = _playbook(content="old") + s.save_user_playbooks([old]) + old_id = old.user_playbook_id + + # Patch save_user_playbooks to confirm it is never reached on empty request_id. + with patch.object(s, "save_user_playbooks") as mock_save: + with pytest.raises(ValueError, match=_EMPTY_REQUEST_ID_MSG): + apply_playbook_edit( + s, + incumbent_id=old_id, + new_playbook=_playbook(content="new"), + source="offline_optimizer", + request_id="", + ) + mock_save.assert_not_called() + + # No orphan: no successor row was inserted (incumbent still CURRENT, count==1). + count = s.conn.execute( + "SELECT COUNT(*) FROM user_playbooks WHERE status IS NULL" + ).fetchone()[0] + assert count == 1, ( + "no orphan successor row should be inserted on empty request_id" + ) + + +@pytest.mark.parametrize("bad_request_id", ["", None]) +def test_apply_raises_on_empty_or_none_request_id(bad_request_id): + """apply_playbook_edit raises ValueError for both empty string and None request_id.""" + from reflexio.server.services.playbook.playbook_edit_apply import ( + apply_playbook_edit, + ) + + with tempfile.TemporaryDirectory() as tmp: + s = _storage(tmp) + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + old = _playbook(content="old") + s.save_user_playbooks([old]) + old_id = old.user_playbook_id + + with pytest.raises((ValueError, TypeError)): + apply_playbook_edit( + s, + incumbent_id=old_id, + new_playbook=_playbook(content="new"), + source="offline_optimizer", + request_id=bad_request_id, # type: ignore[arg-type] + ) + + +def test_apply_lineage_event_carries_operation_run_id(): + """apply_playbook_edit records the operation-run request_id on the revise event. + + The lineage event must carry the operation request_id, + NOT the incumbent's birth request_id. This enables correct run-correlation. + """ + from reflexio.server.services.playbook.playbook_edit_apply import ( + apply_playbook_edit, + ) + + with tempfile.TemporaryDirectory() as tmp: + s = _storage(tmp) + with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): + old = _playbook(content="old") + s.save_user_playbooks([old]) + old_id = old.user_playbook_id + assert old_id > 0 + + operation_run_id = "optimizer_run_xyz" + new = _playbook(content="new") + new_id = apply_playbook_edit( + s, + incumbent_id=old_id, + new_playbook=new, + source="offline_optimizer", + request_id=operation_run_id, + ) + assert new_id > 0 + + events = s.get_lineage_events( + entity_type="user_playbook", entity_id=str(new_id) + ) + assert [event.op for event in events] == ["create", "revise"] + revise_event = events[1] + assert revise_event.request_id == operation_run_id, ( + f"lineage event must carry the operation run id {operation_run_id!r}, " + f"not the incumbent's birth request_id {old.request_id!r}" + ) From dbb01dc036b2b3c25f7bfcb2fc6daad4561491a6 Mon Sep 17 00:00:00 2001 From: Guangyu Date: Wed, 29 Jul 2026 05:49:46 +0000 Subject: [PATCH 3/3] test(playbook): drop a dead duplicate and a stale expect_current name MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The restored test file carried two references to an `expect_current` parameter that `apply_playbook_edit` has never had: - `test_apply_expect_current_false_archives` was byte-identical to `test_apply_inserts_new_and_archives_incumbent` (both bodies 1009 chars), so it added no coverage while implying an API that does not exist. Removed. - `test_apply_expect_current_false_returns_minus1_and_no_orphan` is a genuine lost-race test — it asserts the -1 return, the rollback, and the absence of an orphan lineage event — and only its name was stale. Renamed to `test_apply_returns_minus1_and_leaves_no_orphan_on_lost_race`; body and docstring are unchanged. `grep -rn expect_current reflexio/` returns nothing, confirming the parameter is not part of any current signature. --- .../playbook/test_playbook_edit_apply.py | 33 +------------------ 1 file changed, 1 insertion(+), 32 deletions(-) diff --git a/tests/server/services/playbook/test_playbook_edit_apply.py b/tests/server/services/playbook/test_playbook_edit_apply.py index cee4198e..4034fd91 100644 --- a/tests/server/services/playbook/test_playbook_edit_apply.py +++ b/tests/server/services/playbook/test_playbook_edit_apply.py @@ -86,38 +86,7 @@ def test_apply_skips_archive_when_incumbent_not_current(): assert new_id == -1 -def test_apply_expect_current_false_archives(): - """With expect_current=False, new playbook is inserted and incumbent is archived.""" - from reflexio.server.services.playbook.playbook_edit_apply import ( - apply_playbook_edit, - ) - - with tempfile.TemporaryDirectory() as tmp: - s = _storage(tmp) - with patch.object(SQLiteStorage, "_get_embedding", return_value=[0.0] * 512): - old = _playbook(content="old") - s.save_user_playbooks([old]) - old_id = old.user_playbook_id - assert old_id > 0 - - new = _playbook(content="new") - new_id = apply_playbook_edit( - s, - incumbent_id=old_id, - new_playbook=new, - source="offline_optimizer", - request_id="run-abc", - ) - assert new_id > 0 - - # Only new_id should be CURRENT (status=None); old_id should be archived - all_pbs = s.get_user_playbooks(user_id="u1") - current_ids = {p.user_playbook_id for p in all_pbs if p.status is None} - assert new_id in current_ids - assert old_id not in current_ids - - -def test_apply_expect_current_false_returns_minus1_and_no_orphan(): +def test_apply_returns_minus1_and_leaves_no_orphan_on_lost_race(): """When incumbent is already archived, supersede_record returns False. The transaction rolls back the provisional successor and its create event,