diff --git a/backend/packages/harness/deerflow/runtime/runs/store/memory.py b/backend/packages/harness/deerflow/runtime/runs/store/memory.py index cd67bdff6..cbbd42900 100644 --- a/backend/packages/harness/deerflow/runtime/runs/store/memory.py +++ b/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 diff --git a/backend/tests/test_multi_worker_run_ownership.py b/backend/tests/test_multi_worker_run_ownership.py index 09b25b2e4..7b4af73a5 100644 --- a/backend/tests/test_multi_worker_run_ownership.py +++ b/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.