fix(monitor): bound pending tasks, not just concurrent scans

[Ap6pack]

Fixes #57.

Confirmed

The report is accurate. build_snapshot admitted one Task per skill to asyncio.gather and bounded only how many ran, via a semaphore inside the worker. Live Tasks therefore scaled with the registry rather than with the concurrency limit.

Reproduced independently — without running the attached PoC — at the reporter's parameters and at the sweep's real ceiling of --max-scans 12000:

n=512    gather-all   tasks_alive=513    peak_active=8      707 KiB
n=512    worker-pool  tasks_alive=9      peak_active=8       26 KiB
n=12000  gather-all   tasks_alive=12001  peak_active=8   16,579 KiB
n=12000  worker-pool  tasks_alive=9      peak_active=8      477 KiB

~16 MiB of Tasks to keep 8 of them busy. The reporter's counts match ours exactly.

Severity

Low. The concurrency limit worked as documented and no extra load ever reached the registry — the report doesn't claim otherwise. But it is waste that grows with the registry, and the registry is the thing that grows: 74k skills today, up from 66k when that code was written.

Fix

malwar.core.concurrency.run_bounded inverts the structure — a fixed pool pulls from a queue, so live Tasks equal the limit instead of the input size.

Applied to all three sites, not just the one reported. The escalation and enrichment phases had the identical shape:

site	was
Phase 1 scan	gather(*(_worker(m) for m in to_scan))
Phase 2 escalate	gather(*(_escalate(s) for s in candidates))
Phase 3 enrich	gather(*(_enrich(s) for s in flagged))

On failure the pool is cancelled rather than left running, so a sweep that is already failing stops hitting the registry.

On the tests

They were run against the old implementation and fail there:

OLD code  n=12000  peak=8 (ok)  live=12001  -> FAILS
NEW code  n=12000  peak=8 (ok)  live=9      -> PASSES

That check matters here. Peak concurrency was already correct before this change, so a test asserting only the limit would have passed against the bug — which is the failure mode that has bitten this repo twice recently, where a test encoded the same assumption as the code it guarded.

Also covered: every item runs exactly once, empty input, fewer items than workers, concurrency=0 making progress rather than deadlocking, and no workers surviving a failure.

1727 passed, 16 skipped   (9 new)

Thanks @Nievesjyl — clear report with reproducible numbers.

🤖 Generated with Claude Code

https://claude.ai/code/session_01DNoTXU8k3pfSBzR7aJubqL

---

Generated by Claude Code