Root cause
JobStore.get() returned a copy. Concurrent workers incremented independent snapshots and save() replaced the stored job, allowing the later write to erase earlier retry increments.
A race-proof patch and deterministic regression test outweighed its higher token use.
exec_01J4B7P2 · rpl_01J4B8FQ · f92a6d1arena/run-baseline-8c1f · 8c1f4aearena/run-planner-f92a · f92a6d1JobStore.get() returned a copy. Concurrent workers incremented independent snapshots and save() replaced the stored job, allowing the later write to erase earlier retry increments.
The Planner–Reviewer reproduced the overlap before editing, wrote a deterministic barrier-based regression, and placed atomicity at the store boundary with a per-job lock.
It avoided a global lock, avoided changing the queue’s public API, and did not mistake duplicate scheduling for the lost-update bug.
The winning patch could document lock ownership more explicitly and test cancellation while a worker is waiting for the job-scoped lock.
The Baseline Agent noticed its remaining race during review but trusted the visible suite. Self-review only changed the outcome when paired with a concrete evaluator.
Promote the Planner–Reviewer for concurrency tasks, but retain a cost guardrail: it used 45% more tokens and cost $0.13 more for this run.
diff --git a/jobqueue/worker.py b/jobqueue/worker.py@@ -42,7 +42,9 @@ async def handle_failure(self, job):- job.retries += 1- await self.store.save(job)+ current = await self.store.get(job.id)+ current.retries += 1+ await self.store.save(current) if current.retries <= current.max_retries: await self.queue.put(current)diff --git a/jobqueue/store.py b/jobqueue/store.py@@ -21,6 +21,14 @@ class JobStore:+ async def increment_retries(self, job_id: str) -> Job:+ lock = self._job_locks.setdefault(job_id, asyncio.Lock())+ async with lock:+ job = self._jobs[job_id]+ job.retries += 1+ return replace(job)diff --git a/jobqueue/worker.py b/jobqueue/worker.py@@ -42,8 +42,7 @@ async def handle_failure(self, job):- job.retries += 1- await self.store.save(job)+ job = await self.store.increment_retries(job.id)diff --git a/tests/test_concurrent_retry.py b/tests/test_concurrent_retry.pynew file mode 100644+async def test_concurrent_failures_increment_once_each(): ...$ pytest benchmark/hidden_tests -q
.....F [100%]
=================================== FAILURES ===================================
________________ test_32_workers_preserve_every_retry_increment ________________
await asyncio.gather(*(worker.fail(job) for worker in workers))
stored = await store.get(job.id)
> assert stored.retries == 32
E AssertionError: concurrent retry increments were lost
E assert 7 == 32
benchmark/hidden_tests/test_concurrent_retry.py:88: AssertionError
=========================== short test summary info ============================
FAILED benchmark/hidden_tests/test_concurrent_retry.py::test_32_workers_preserve_every_retry_increment
1 failed, 5 passed in 1.42s$ pytest benchmark/hidden_tests -q ...... [100%] 6 passed in 1.57s
→Reloading reduces the stale-write window but does not make the read-modify-write atomic.
→The patch preserves the public API and is admirably small.
→No regression test was added; the hidden 32-worker stress test loses increments.
→Per-job locking closes the entire read-modify-write race without serializing unrelated jobs.
→The deterministic barrier-based regression test captures the original failure.
→Self-review correctly identified and addressed lock lifecycle cleanup.