FATAL: file does not parse. 'async with self._callbacks_lock' (L103) and 'await coro' (L98) are inside on_event, a SYNC def. SyntaxError before any test runs.
Bounded-concurrency design is broken: only ONE worker task is created (start() does a single create_task), and the Semaphore(4) is acquired inside that single loop. Effective concurrency is 1, not 4 — the semaphore is decorative.
Test A asserts max_concurrency <= 4, but since real concurrency is ~1 it passes trivially and proves NOTHING about the cap actually holding under saturation.
Cooperative cancel only sets a flag; it cannot interrupt an in-flight mock_synthesize mid-sleep (acceptable, but not 'cancel at next safe checkpoint' for a long synth).
submit() after stop() silently enqueues to a dead worker — no rejection, jobs leak (never processed, never failed).
_final_states dict grows unbounded (no eviction) — memory leak for a long-running service.
Callback errors are silently swallowed (broad try/except Exception) — a silent-failure pattern; debug visibility lost.
▮ RECOMMENDED USE
NOT a safe offload for async-pipeline / queue work despite scoring 82 on the LFU exam. Same model+quant, different prompt: 82 -> 49. Use it for data-structure/ACID tasks; route queue/backpressure/retry work elsewhere (or to the cloud model).
▮ REFACTORED PATCH
# FIX 1 (the parse error): make on_event async (or don't await inside it).
# Simplest correct version:
async def on_event(self, callback):
async with self._callbacks_lock:
self._callbacks.append(callback)
# FIX 2 (real bounded concurrency): spawn N workers OR create_task per job
# gated by the semaphore. Option B (concurrency from the semaphore itself):
async def _worker_loop(self):
while self._running:
job = await self._queue.get()
# do NOT hold the semaphore in the single worker; instead launch
# each job as its own task, gated so at most max_concurrency run:
async def _run(j):
async with self._semaphore:
await self._process_job(j)
self._queue.task_done()
asyncio.create_task(_run(job))
# (and add a test that submits >max_concurrency long jobs and asserts
# exactly max_concurrency run at once.)
# FIX 3: reject submit() after stop() (guard on self._running).
# FIX 4: bound _final_states (e.g. keep last N, or evict terminal >TTL).
# FIX 5: log callback errors instead of swallowing them silently.