▚ MODEL AUDIT // MLX

// KAT-Coder v2.5 Dev XL

CRIT — Critical Bugs 52/100 tests: CRASH

tok/sec65.26

tokens—

TTFT—

▮ PILLAR BREAKDOWN

Complexity (O(1))

15/20

Concurrency / Races

7/20

Error Handling

13/20

Resource & State Safety

7/20

Test Integrity

10/20

✓ WENT RIGHT

No pillar reached 16+ — no standout strengths.

✗ WENT WRONG

Error Handling (13/20)
Test Integrity (10/20)
Concurrency / Races (7/20)
Resource & State Safety (7/20)

▮ CRITICAL BUGS

In-flight cancellation is a no-op: cancel() only sets _cancelled=True for STARTED jobs, but every cancellation check ALSO requires job.state==CANCELLED, which cancel() never sets for in-flight jobs (only QUEUED). In-flight jobs always run to completion — the spec's core requirement is unmet.
Hard 30s timeout: Test B drains 100 jobs at max_concurrency=1 (sequential); expected ~20.6s, p95 ~28s, with 10% retry backoffs it routinely exceeds 30s — a self-inflicted wall-clock sink.
Leaked tasks on shutdown: stop() cancels _worker_task but does NOT track or cancel the fire-and-forget _process_job tasks (asyncio.create_task). In-flight synthesis tasks are orphaned — no gather, no cancellation, semaphore slots held until they finish.
Latent deadlock: _process_job is spawned untracked; if it raises outside its try/except, task_done() is never called AND the job stays STARTED forever, so drain()'s terminal-state poll loops infinitely.
Cancellation test passes for the wrong reason: it asserts completed+cancelled+failed==20, but since in-flight cancel is a no-op, 'cancelled' jobs actually COMPLETE.

▮ RECOMMENDED USE

Scaffolding async pipeline structure (queue+semaphore+retry+callbacks) when you intend to fix the cancellation and shutdown paths yourself.

▮ REFACTORED PATCH

# FIX 1 — cancel() must set state=CANCELLED for in-flight jobs too:
async def cancel(self, job_id):
    async with self._lock:
        job = self._jobs.get(job_id)
        if job is None: raise KeyError(f"Unknown job: {job_id}")
        if job.state in (COMPLETED, FAILED, CANCELLED): return
        job._cancelled = True
        job.state = JobState.CANCELLED   # set for BOTH queued AND in-flight
    await self._notify(job, "cancelled")

# FIX 2 — gate cancellation checks on the flag, not the state:
# replace every `if job._cancelled and job.state == JobState.CANCELLED`
# with `if job._cancelled:`.

# FIX 3 — track spawned tasks and cancel them on stop():
#   __init__: self._tasks = set()
#   _worker_loop:
#     t = asyncio.create_task(self._process_job(job))
#     self._tasks.add(t); t.add_done_callback(self._tasks.discard)
#   stop():
#     for t in list(self._tasks): t.cancel()
#     await asyncio.gather(*self._tasks, return_exceptions=True)

# FIX 4 — don't self-DOS the backpressure test: use max_concurrency=1 AND
# max_queue_size=10, submit 11, assert 11th rejected, then stop WITHOUT draining.