feat: kafka listen capture daemon

[HumanBean17]

What

Adds a kafka listen command group — a long-lived Kafka capture daemon that lets an agent register a listener at the start of a runbook/debug session and verify results at the end, eliminating the false negatives that today's windowed kafka assert produces on long or high-volume runs.

Why

kafka assert/consume are stateless, one-shot, windowed scans ([now−lookback, now+timeout], single-threaded poll). They false-negative under three conditions that long debug/runbook sessions hit:

1. Scan-window miss — message produced early, asserted late, falls before the seek point.
2. Volume-induced timeout truncation — high-volume backlog can't be drained within the timeout → (None, scanned).
3. Broker retention cleanup — long sessions outlive topic retention; the message is physically gone, so a live-topic scan fails permanently (and agents misattribute it to "integration broken").

The listener drains live from start (positioned at the head before the runbook triggers, via seek-to-OFFSET_END on assignment) and persists captures to disk, so assertion reads a retention-independent buffer with no wall-clock deadline — none of the three failure modes apply.

Commands (agctl kafka listen …)

Command	Behavior
run	Foreground streaming (NDJSON); the daemon spawn target + cross-platform fallback (6th streaming exception).
start / status / stop	Managed daemon (POSIX/WSL-gated, like mock). stop deletes the run dir; raises on fatal events.
assert	Attach an expectation to asserts.jsonl (listener stays alive).
results	Evaluate all expectations over the captured files; exit 1 on any failure.
messages	Debug tap — dump captured (filtered) messages for a topic.

Runbook shape: start (before the trigger) → assert → results → stop. Note: stop does not auto-run results (an uncollected expectation is dropped on cleanup).

How

Mirrors the agctl mock managed-daemon pattern and reuses its lifecycle machinery (extracted into a shared agctl/daemon.py — inward-only, no listen→mock coupling). New agctl/listen/ subpackage (daemon.py, capture_file.py, assert_eval.py, capture.py::CaptureLoop, engine.py::ListenEngine). Reuses KafkaClient.consume_loop (seek-to-end via on_assign), _build_assert_predicate, resolve_cluster_name/new_kafka_client. Deliberate second stateless-invariant carve-out alongside mock (state confined to --state-dir).

Design spec & implementation plan: docs/superpowers/specs|plans/active/2026-07-15-kafka-listen-capture-*.md.

Testing

• 1221 unit tests green (no regressions); the load-bearing seek-to-latest invariant is unit-pinned (seek every partition to OFFSET_END, forwarded via consume_loop; ready_event set once/rebalance-safe) and asserted live in integration.
• 2 integration tests self-skip without a broker; CI runs them live under AGCTL_TEST_LIVE=1.
• Final whole-branch review (opus): Ready to merge — Yes, 0 Critical. All three load-bearing invariants verified against the real wiring.

Accepted tradeoffs (per final-review triage)

• No per-line fsync — would regress high-volume throughput 100–1000×; flush() gives read-your-writes for live readers and clean SIGTERM shutdown. Loss on SIGKILL/OOM is the accepted trade.
• Eager file reads in capture_file (bounded by the 256 MiB --max-bytes-per-topic valve) — a streaming line-iterator is a cheap future win; no deadline means slow-not-broken.
• A handful of cosmetic Minors deferred (see the SDD ledger).

🤖 Generated with Claude Code