What to build
Python and TS already keep the per-agent_type ctrl-stream reads and the worker's direct control-stream read as two separate code paths. Java's WorkerRunner.runLoop() currently combines them: the streams map passed to a single xreadGroup() call contains both every ctrlStream(agentType) and workerCtrlStream(workerId) at once. One of these belongs to the "agent_type" hash-tag group, the other to the "worker" hash-tag group — they don't share a tag and never will (they're different entities), so this single combined call is fundamentally CROSSSLOT-prone under Cluster no matter what's done to the agent_type side alone.
This is prefactoring work needed before the two-phase XREADGROUP plan (Phase 4 main issue) can be implemented in Java at all: split runLoop() into two independent read paths — one for the agent_type streams (which the main Phase 4 issue will turn into the two-phase scan+block plan), and one for workerCtrlStream(workerId) on its own (blocking or non-blocking, running alongside, not mixed in).
To support the split, add a minimal Redis Stream operations interface (RedisStreamOps) with just enough surface for this: group-create-if-not-exists (idempotent, swallowing BUSYGROUP), a single-stream read supporting both non-blocking (blockMs=null) and blocking (blockMs>0) modes, and ack. WorkerRunner should call this interface instead of Jedis/JedisCluster directly for these operations. The remaining stream operations (xadd, xpendingRange, xclaim) are out of scope here — those land in the Phase 5 issue.
Acceptance criteria
Blocked by
#8
What to build
Python and TS already keep the per-
agent_typectrl-stream reads and the worker's direct control-stream read as two separate code paths. Java'sWorkerRunner.runLoop()currently combines them: thestreamsmap passed to a singlexreadGroup()call contains both everyctrlStream(agentType)andworkerCtrlStream(workerId)at once. One of these belongs to the "agent_type" hash-tag group, the other to the "worker" hash-tag group — they don't share a tag and never will (they're different entities), so this single combined call is fundamentally CROSSSLOT-prone under Cluster no matter what's done to theagent_typeside alone.This is prefactoring work needed before the two-phase
XREADGROUPplan (Phase 4 main issue) can be implemented in Java at all: splitrunLoop()into two independent read paths — one for theagent_typestreams (which the main Phase 4 issue will turn into the two-phase scan+block plan), and one forworkerCtrlStream(workerId)on its own (blocking or non-blocking, running alongside, not mixed in).To support the split, add a minimal Redis Stream operations interface (
RedisStreamOps) with just enough surface for this: group-create-if-not-exists (idempotent, swallowing BUSYGROUP), a single-stream read supporting both non-blocking (blockMs=null) and blocking (blockMs>0) modes, and ack.WorkerRunnershould call this interface instead ofJedis/JedisClusterdirectly for these operations. The remaining stream operations (xadd,xpendingRange,xclaim) are out of scope here — those land in the Phase 5 issue.Acceptance criteria
runLoop()split into two independent read loops:agent_typestreams vsworkerCtrlStream(workerId), no longer combined into onexreadGroup()callRedisStreamOpsinterface added:xgroupCreateIfNotExists(streamName, groupName), a single-stream read method takingcountand an optionalblockMs(null = non-blocking, >0 = blocked for that many ms),xack(streamName, groupName, id)StandaloneRedisStreamOps/ClusterRedisStreamOpsimplementations added (both just directly transpose toJedis/JedisClustercalls — these are all single-key operations, natively supported by both)WorkerRunnerupdated to depend onRedisStreamOpsfor these operations instead of importingJedis/JedisClusterdirectly for themBlocked by
#8