fix(runtime): share one change position across an atomic thread operation (#5663)

* fix(runtime): share one change position across an atomic thread operation

MemoryRunStore.create_thread_operation_atomic marked each row it touched
separately, so one interrupt-and-replace consumed two positions in the
(change_seq, run_id) cursor that list_changed consumers page with. The SQL
store allocates a single position for the whole set, and runtime/AGENTS.md
documents that as the contract for an atomic thread operation.

The memory store is the backend an install runs whenever no durable database
is configured, so a cursor reader could observe the interrupted row and its
replacement under two positions instead of one, ordered by run_id.

Allocate one position after the raise-only scan -- so a rejected operation
still consumes nothing -- and reuse it for the claimed rows and the new row.

* test(runtime): pin that a rejected thread operation consumes no position

create_thread_operation_atomic allocates its change position only after the
raise-only candidate scan, so a ConflictError consumes nothing -- the memory
store's counterpart of the SQL store rolling back and leaving its clock
untouched. Nothing asserted that ordering.

A consumed-but-unused position leaves no trace in the rows themselves, so a
test comparing list_changed output passes even with the allocation hoisted
above the scan. Assert instead on the position the next accepted operation
lands on: that surfaces a hoist as a gap.

backend/packages/harness/deerflow/runtime/runs/store/memory.py

@@ -40,9 +40,12 @@ class MemoryRunStore(RunStore):
             if not bucket:
                 self._runs_by_thread.pop(thread_id, None)
 
-    def _mark_changed(self, run: dict[str, Any]) -> None:
+    def _next_change_seq(self) -> int:
         self._change_seq += 1
-        run["change_seq"] = self._change_seq
+        return self._change_seq
+
+    def _mark_changed(self, run: dict[str, Any]) -> None:
+        run["change_seq"] = self._next_change_seq()
 
     async def put(
         self,
@@ -502,6 +505,7 @@ class MemoryRunStore(RunStore):
         # interrupted state on raise, diverging from SQL where a raise rolls
         # the whole transaction back.
         claimed = []
+        change_seq: int | None = None
         if multitask_strategy in ("interrupt", "rollback"):
             candidates: list[dict[str, Any]] = []
             for r in self._runs.values():
@@ -533,12 +537,20 @@ class MemoryRunStore(RunStore):
                 if r.get("operation_kind", "run") != "run" and not lease_expired:
                     raise ConflictError(f"Thread {thread_id} has an active checkpoint write")
                 candidates.append(r)
+            # One position covers this atomic set of changes, with ``run_id``
+            # ordering ties. The SQL store allocates the same single value for
+            # the set from its singleton clock (``runtime/AGENTS.md``); marking
+            # each row separately split one interrupt-and-replace into two
+            # positions in the ``(change_seq, run_id)`` cursor consumers page
+            # with. Allocate after the raise-only scan above so a rejected
+            # operation does not consume a position.
+            change_seq = self._next_change_seq()
             for r in candidates:
                 r["status"] = "interrupted"
                 r["error"] = "Cancelled by newer run"
                 r["owner_worker_id"] = owner_worker_id
                 r["updated_at"] = now
-                self._mark_changed(r)
+                r["change_seq"] = change_seq
                 claimed.append(r)
 
         new_row = {
@@ -561,7 +573,9 @@ class MemoryRunStore(RunStore):
             "created_at": created_at or now,
             "updated_at": now,
         }
+        if change_seq is None:
+            change_seq = self._next_change_seq()
+        new_row["change_seq"] = change_seq
         self._runs[run_id] = new_row
-        self._mark_changed(new_row)
         self._index_run(run_id, thread_id)
         return new_row, claimed

backend/tests/test_multi_worker_run_ownership.py

@@ -1296,6 +1296,97 @@ async def test_create_thread_operation_atomic_interrupt_claims_and_creates():
     assert old_row["status"] == "interrupted"
 
 
+@pytest.mark.anyio
+async def test_create_thread_operation_atomic_uses_one_change_position():
+    """One atomic thread operation advances the change cursor once.
+
+    ``runtime/AGENTS.md`` documents that an atomic thread operation uses one
+    position for its interrupted rows and its new row, with ``run_id``
+    ordering ties, and ``test_run_evidence_reader`` pins that for the SQL
+    store. The memory store must agree: it is the backend an install runs
+    whenever no durable database is configured. Advancing once per row splits
+    a single interrupt-and-replace into two positions in the
+    ``(change_seq, run_id)`` cursor extensions page with.
+    """
+    store = MemoryRunStore()
+    config = _lease_config()
+    expired_lease = (datetime.now(UTC) - timedelta(seconds=60)).isoformat()
+
+    await store.create_thread_operation_atomic(
+        run_id="run-old",
+        thread_id="thread-1",
+        owner_worker_id="w1",
+        lease_expires_at=expired_lease,
+        multitask_strategy="reject",
+        grace_seconds=config.grace_seconds,
+    )
+
+    new_row, claimed = await store.create_thread_operation_atomic(
+        run_id="run-new",
+        thread_id="thread-1",
+        owner_worker_id="w2",
+        lease_expires_at=(datetime.now(UTC) + timedelta(seconds=30)).isoformat(),
+        multitask_strategy="interrupt",
+        grace_seconds=config.grace_seconds,
+    )
+
+    assert [row["run_id"] for row in claimed] == ["run-old"]
+    assert claimed[0]["change_seq"] == new_row["change_seq"]
+
+    # The shared position is what a paging consumer observes: the interrupted
+    # row and its replacement surface under one cursor value, ordered by
+    # ``run_id``, exactly as the SQL store reports them.
+    changed = await store.list_changed(after_change_seq=-1, after_run_id="")
+    positions = {row["run_id"]: row["change_seq"] for row in changed}
+    assert positions["run-old"] == positions["run-new"]
+    assert [row["run_id"] for row in changed if row["change_seq"] == positions["run-new"]] == ["run-new", "run-old"]
+
+
+@pytest.mark.anyio
+async def test_create_thread_operation_atomic_rejection_consumes_no_change_position():
+    """A ``ConflictError``-rejected operation advances no change position.
+
+    ``create_thread_operation_atomic`` allocates its position only after the
+    raise-only candidate scan, so a rejected operation consumes nothing — the
+    memory-store counterpart of the SQL store's rollback leaving the clock
+    untouched. A consumed-but-unused position leaves no trace in the rows
+    themselves, so the assertion is on the position the next accepted
+    operation lands on. Hoisting the allocation above the scan keeps every
+    other test in this file green and shows up here as a gap.
+    """
+    store = MemoryRunStore()
+    config = _lease_config(grace_seconds=10)
+
+    accepted, _ = await store.create_thread_operation_atomic(
+        run_id="valid-lease-run",
+        thread_id="thread-1",
+        owner_worker_id="other-worker",
+        lease_expires_at=(datetime.now(UTC) + timedelta(seconds=30)).isoformat(),
+        multitask_strategy="reject",
+        grace_seconds=config.grace_seconds,
+    )
+
+    with pytest.raises(ConflictError, match="another worker"):
+        await store.create_thread_operation_atomic(
+            run_id="run-new",
+            thread_id="thread-1",
+            owner_worker_id="w2",
+            lease_expires_at=(datetime.now(UTC) + timedelta(seconds=30)).isoformat(),
+            multitask_strategy="interrupt",
+            grace_seconds=config.grace_seconds,
+        )
+
+    next_row, _ = await store.create_thread_operation_atomic(
+        run_id="run-after",
+        thread_id="thread-2",
+        owner_worker_id="w2",
+        lease_expires_at=(datetime.now(UTC) + timedelta(seconds=30)).isoformat(),
+        multitask_strategy="reject",
+        grace_seconds=config.grace_seconds,
+    )
+    assert next_row["change_seq"] == accepted["change_seq"] + 1
+
+
 @pytest.mark.anyio
 async def test_create_thread_operation_atomic_interrupt_rejects_other_worker_valid_lease():
     """Interrupt must raise ConflictError when a valid-lease run is owned by another worker.