⚠ SPEED CAVEAT: Slow deep-thinker: ~17-19 t/s, 5-9 min/prompt, ~5-9k tokens incl. reasoning. Fast for quality, not throughput.
tok/sec17.18
tokens9012
TTFT3.44s
▮ PILLAR BREAKDOWN
Idempotency
18/20
Retry / Backoff
19/20
Checkpointing
19/20
Signal Handling
16/20
Test Integrity
17/20
✓ WENT RIGHT
Retry / Backoff (19/20)
Checkpointing (19/20)
Idempotency (18/20)
Test Integrity (17/20)
Signal Handling (16/20)
✗ WENT WRONG
No pillar fell below 14 — solid across the board.
▮ CRITICAL BUGS
No explicit checkpoint flush on the SIGINT exit path — relied on per-item save being always-current, which is correct, but a worker that crashes between process()-success and the lock/save block could lose a completed item's state on hard kill mid-critical-section (narrow window).
Signal handling does not cancel in-flight tasks — a worker past the sem-acquire running process() at SIGINT time runs to completion including all retry/backoff sleeps (up to ~0.7s), delaying exit. Spec allows 'finish' but exit latency can exceed expectations under load.
Test asserts convergence, concurrency, and corruption-freedom but does NOT assert that a rerun avoids re-invoking process() on already-completed items — the idempotency no-rework guarantee is structurally enforced by the pending filter but unverified by an explicit call-count-on-rerun assertion.
Queue.task_done() is called but q.join() is never awaited — cosmetic dead code.
▮ RECOMMENDED USE
Reliable production-grade async batch processing with atomic checkpointing (temp+fsync+rename) where correctness of summary output and crash-safe state matter more than raw throughput. First model in the benchmark to print a correct, meaningful automation summary (98/2/0/100).
▮ REFACTORED PATCH
# 1. Explicit checkpoint flush + clean exit on SIGINT
async def run_batch(items):
try:
await asyncio.gather(*workers, return_exceptions=True)
finally:
save_checkpoint(completed, failed) # explicit final flush covers the gap
try: loop.remove_signal_handler(signal.SIGINT)
except (NotImplementedError, RuntimeError): pass
# 2. Cancel in-flight on SIGINT for prompt exit (add cooperative bail between retries)
async def handle_item(item, ...):
async with sem:
if stop_event.is_set(): return
backoff = 0.1
for attempt in range(3):
if stop_event.is_set(): return # bail between retries
try:
await process(item)
async with lock: completed.add(item); save_checkpoint(completed, failed)
return
except ProcessingError:
if attempt == 2:
async with lock: failed.add(item); save_checkpoint(completed, failed)
return
await asyncio.sleep(backoff); backoff *= 2
# 3. Assert no-rework on rerun
second_calls = {k: v for k, v in calls.items()}
await run_batch(items) # third run, all should skip
reprocess = {k: calls[k]-second_calls[k] for k in calls if calls[k] > second_calls.get(k, 0)}
assert not reprocess, f"idempotency violated, reprocessed: {reprocess}"