// KAT-Coder v2.5 Dev XL

▮ PILLAR BREAKDOWN

▮ CRITICAL BUGS

• No tests at all: process() is defined but never called by any async main/harness; file requires CLI `items` arg so `python file.py` errors out (tests_pass=false, violates self-contained-runnable contract).
• Checkpoint error-path double-close: `os.close(fd) if not None else None` — `not None` is always True so it always calls os.close(fd); if the fd was already closed this re-raises OSError [Errno 9] and masks the original exception, also skipping the os.unlink(tmp) cleanup. Should be `if fd is not None` with a closed-flag.
• SIGINT drain logic is broken: run() does ONE asyncio.wait(FIRST_COMPLETED), then cancels pending tasks and swallows CancelledError via gather(return_exceptions=True) rather than letting in-flight work finish.
• No explicit exit 0 on graceful shutdown: _on_signal sets an Event and run() returns normally, but a second SIGINT calls sys.exit(1) (hard exit, no checkpoint flush).

▮ RECOMMENDED USE

▮ REFACTORED PATCH

# 1. Fix checkpoint error-path close (use a closed flag)
async def save_checkpoint(cp, path):
    parent = path.parent; parent.mkdir(parents=True, exist_ok=True)
    loop = asyncio.get_event_loop()
    def _write():
        fd, tmp = tempfile.mkstemp(suffix='.tmp', dir=parent); closed = False
        try:
            payload = json.dumps(asdict(cp), indent=2, ensure_ascii=False) + '\n'
            os.write(fd, payload.encode('utf-8')); os.close(fd); closed = True
            os.replace(tmp, str(path))
        except BaseException:
            if not closed:
                try: os.close(fd)
                except OSError: pass
            try: os.unlink(tmp)
            except OSError: pass
            raise
    await loop.run_in_executor(None, _write)

# 2. Fix SIGINT drain: let in-flight finish, flush, exit 0
async def run(self):
    self._start_time = time.monotonic()
    self._install_signal_handlers()
    pending = [it for it in self.items if not self.cp.is_done(it)]
    skipped = len(self.items) - len(pending)
    self._semaphore = asyncio.Semaphore(self.max_concurrency)
    async def _bounded(item):
        async with self._semaphore:
            if self._shutdown_requested.is_set(): return
            await self._process_one(item)
    workers = [asyncio.create_task(_bounded(it)) for it in pending]
    try:
        await asyncio.gather(*workers, return_exceptions=True)
        await save_checkpoint(self.cp, self.checkpoint_path)
    finally:
        self._remove_signal_handlers()
    elapsed_ms = int((time.monotonic() - self._start_time) * 1000)
    return {'succeeded': len(self.cp.succeeded), 'failed': len(self.cp.failed),
            'skipped': skipped, 'total': len(self.items), 'elapsed_ms': elapsed_ms}

# 3. Add a self-contained async main() test harness (no CLI args)
async def main():
    items = [f'job-{i}' for i in range(20)]
    bp = BatchProcessor(items, checkpoint_path='cp.json')
    s = await bp.run(); print(json.dumps(s))
if __name__ == '__main__': asyncio.run(main())