Redis cluster support Phase 4: two-phase XREADGROUP split for multi agent_type workers

[zhchxiao123]

What to build

Replace runner.py's current multi-stream blocking XREADGROUP call (which mixes multiple different agent_type ctrl streams in one call — not Cluster-safe, since they're different entities/slots) with a two-phase read loop per polling iteration in fetch_messages:

1. Phase one: concurrently issue one non-blocking XREADGROUP per active agent_type stream (no BLOCK argument at all — omitting BLOCK means "return immediately with whatever's available"; this is NOT the same as BLOCK 0, which blocks forever). If any stream returned messages, return them immediately.
2. Phase two: only if every stream came back empty, issue one real blocking XREADGROUP (BLOCK <ms>) against a single "primary" stream, chosen by round-robin across the declared agent_types so no agent_type is permanently starved of the blocking slot.

This eliminates CROSSSLOT (every XREADGROUP call now touches exactly one stream/slot) without changing the current worst-case delay characteristics: a message that arrives on a non-primary stream while phase two is mid-block still waits up to the full block timeout; a message arriving at any other time is picked up on the next loop iteration (near-zero added latency).

COUNT semantics are per-stream both before and after this change — a worker declaring N agent_types could already pull up to N × count messages per round pre-split; this doesn't change that ceiling. (If a hard global cap is wanted later, a count // N per-stream division is a documented option, not required here.)

Add multi-agent_type integration test coverage (can run against a local 3-master/3-replica Cluster docker-compose environment marked as a slow/optional CI job) verifying: no CROSSSLOT errors; phase-one non-blocking scan picks up messages within one loop iteration when not blocked in phase two; round-robin primary rotation actually rotates; explicit worst-case latency assertion (bounded by the block timeout, not by the loop interval).

Acceptance criteria

• Multi-stream blocking read replaced by two-phase (concurrent non-blocking scan, then single-primary blocking read)
• Round-robin primary rotation implemented and covered by a test asserting no agent_type is starved
• Worst-case latency test: message arriving during phase-two block on a non-primary stream is picked up only after that block ends, not sooner
• No CROSSSLOT errors in a local Cluster integration test with 3+ declared agent_types

Blocked by

#45