Recreate the generator only when it actually latched

[creslinux]

generate_gen's exception handler unconditionally schedules create_generator(), which cancels every in-flight job via wait_for_jobs(skip_wait=True). Any per-request generation error therefore kills all concurrent requests — and their consumers return partial or empty completions with HTTP 200, because cancellation ends the stream cleanly rather than as an error.

On the current pin (exllamav3 1.4.6) every engine exception that reaches this handler arrives via the AsyncGenerator latch, so recreation is the only recovery there and this change leaves that path untouched. exllamav3 dev now contains start/prefill failures per job (turboderp-org/exllamav3#319, merged, next release): those errors are delivered to the failing request alone and the generator stays healthy — at which point unconditional recreation tears down concurrent requests for failures the engine already handled.

Change

• Recreate, and record the unhealthy event, only when AsyncGenerator.error is set, i.e. the wrapper actually latched. Engines without the latch attribute keep the historical always-recreate behaviour.
• A contained error logs a warning, cancels its own job, and re-raises for that request only. Recreation used to cancel the job as a side effect; without it, an error raised by this consumer rather than the engine would leave the job generating into a queue nobody drains. cancel() is a no-op for a job the engine has already reaped. Note this also applies today: a consumer-side exception on 1.4.6 now cancels its own job instead of recreating the generator and cancelling everyone else's.
• create_generator() now close()s the previous generator before building its replacement. Cancelling the jobs alone left the old iteration task parked forever on an emptied job condition, keeping the old sync Generator reachable for the life of the process. After a latch the task has already exited, so this is a no-op on that path.

Evidence

Fault-injection A/B on a 4-slot recurrent pool, three-layer matrix in turboderp-org/exllamav3#318:

stack	concurrent healthy request
stock engine + stock TabbyAPI	killed by recreation — HTTP 200, empty content
engine with #319 + stock TabbyAPI	still killed — recreation is unconditional
engine with #319 + this change	completes (1,495 tokens); failing request 503s; zero recreations

The guard is certified live by that matrix. close() on a latched generator and cancel() on an already-reaped job are verified against the real classes rather than in a live incident, since the certified incident is engine-contained and never latches. The script below runs against the installed wheel — no model or GPU needed:

verify_close_cancel.py

import asyncio
import torch
from exllamav3 import ArgmaxSampler
from exllamav3.generator.async_generator import AsyncGenerator, AsyncJob, _CANCELLED_SENTINEL
from exllamav3.generator.generator import Generator
from exllamav3.generator.job import Job

async def main():
    # 1. close() on a generator whose iteration task already exited (the latch path:
    #    _run_iteration catches the exception internally and returns normally)
    async def exited():
        return None
    w = AsyncGenerator.__new__(AsyncGenerator)
    w.jobs = {}
    w.condition = asyncio.Condition()
    w.error = RuntimeError("latched")
    w.iteration_task = asyncio.create_task(exited())
    await w.iteration_task
    await w.close()
    print("check1  close() on exited (latched) task: OK")

    # 2. close() idempotency (two racing recreations both closing the old wrapper)
    await w.close()
    print("check2  close() idempotent: OK")

    # 3. cancel() on a job the engine already reaped (absent everywhere)
    g = Generator.__new__(Generator)
    g.pending_jobs = []
    g.active_jobs = []
    g.on_queue_drained = lambda: None
    g.num_remaining_jobs = lambda: 0
    w3 = AsyncGenerator.__new__(AsyncGenerator)
    w3.jobs = {}
    w3.condition = asyncio.Condition()
    w3.error = None
    w3.generator = g
    aj = AsyncJob.__new__(AsyncJob)
    aj.generator = w3
    aj.job = Job(input_ids=torch.tensor([[1, 2, 3]], dtype=torch.long),
                 max_new_tokens=4, sampler=ArgmaxSampler())
    aj.queue = asyncio.Queue()
    aj.cancelled = False
    await aj.cancel()
    assert aj.cancelled and w3.jobs == {} and aj.queue.empty()
    print("check3  cancel() on engine-reaped job: no-op, no sentinel: OK")

    # 4. cancel() on a live job still removes it and delivers the sentinel
    inner = Job(input_ids=torch.tensor([[1, 2, 3]], dtype=torch.long),
                max_new_tokens=4, sampler=ArgmaxSampler())
    aj2 = AsyncJob.__new__(AsyncJob)
    w4 = AsyncGenerator.__new__(AsyncGenerator)
    aj2.generator = w4
    aj2.job = inner
    aj2.queue = asyncio.Queue()
    aj2.cancelled = False
    w4.jobs = {inner: aj2}
    w4.condition = asyncio.Condition()
    w4.error = None
    g2 = Generator.__new__(Generator)
    g2.pending_jobs = [inner]
    g2.active_jobs = []
    g2.on_queue_drained = lambda: None
    g2.num_remaining_jobs = lambda: 1
    w4.generator = g2
    await aj2.cancel()
    assert aj2.queue._queue[0] is _CANCELLED_SENTINEL and inner not in g2.pending_jobs
    print("check4  cancel() on live job: sentinel delivered, job removed: OK")

asyncio.run(main())
print("ALL CHECKS PASSED")

Deliberate choice worth a maintainer's eye

Contained errors no longer feed HealthManager. That's correct for a healthy generator, but it means a device that OOMs on routine requests becomes invisible to /health while each affected request 503s. Happy to add a lighter "degraded" signal if you'd prefer that visibility kept.