feat(streams): reject concurrent messages that break the tool-use loop

[tsuz]

The problem

Nothing stopped a user from sending a second message while the agent was mid-turn, and doing so could kill the turn outright.

EnrichInputMessageProcessor rebuilds the history for the next LLM call as:

previousMessages + [lastInputMessage] + lastInputResponse

When the previous turn issued tools, lastInputResponse ends with an assistant message carrying tool_use blocks. The API requires the matching tool_result blocks to come next. Those arrive as a separate message-input record with role:"tool" (written back by TransformToolUseDoneProcessor).

If a role:"user" message gets appended in that window instead, the LLM sees a tool_use that is not immediately followed by its tool_result and returns 400 — the turn dies.

There was no rate limiting or in-flight guard anywhere else, either: no debounce in the frontend, nothing in chat-api. So the race was fully reachable by typing fast.

The fix

SessionConcurrencyGuardProcessor — admission control placed before enrichment.

message-input ─(re-key by session_id)─┐
                                       ├─►[guard]─┬─► admitted-message-input ─►[enrich]─►…
think-request-response (KStream) ──────┘  │       │
                 owns: session-status-store       └─(⋈ reply-to)─► message-output (rejection)

The guard consumes message-input and republishes only the admitted subset to a new admitted-message-input topic. EnrichInputMessageProcessor now sources from that topic instead of message-input — so a rejected message never reaches enrichment at all.

Busy lifecycle, per session

Event	Behavior
role:"user", session idle	Admitted; turn opens, session marked busy
role:"user", session busy	Rejected — status:"rejected" UserResponse to message-output
role:"tool" continuation	Always admitted, mid-turn included
Clean end-turn (endTurn, no tool calls)	Busy cleared
Think response still carrying tool calls	Busy kept, activity refreshed
Idle > SESSION_BUSY_TTL_MS (default 600s)	Wall-clock punctuator force-clears

Two details worth calling out:

The role:"tool" path is never rejected. That record is precisely the one that completes the pending tool_use blocks — rejecting it would cause the exact failure this PR prevents. TransformToolUseDoneProcessor writes it back to message-input, so it re-enters through the guard and is waved past.

Rejections carry the reply-to route. They're left-joined with the per-session reply-to table, mirroring EndTurnProcessor, so a multi-agent caller whose message is rejected fails fast through its HTTP callback rather than hanging until its own timeout — which would also leave a stale route that the in-flight turn's reply could be mis-delivered to.

The TTL sweep is the anti-wedge net. Without it, a turn that crashed without emitting a clean end-turn would leave its session rejecting every message forever. 600s comfortably exceeds the longest legitimate quiet stretch inside a turn (ASYNC_TOOL_TIMEOUT_MS, default 300s).

API surface

UserResponse gains a nullable status field — null for normal output, "rejected" for a busy rejection. Additive, and the record is @JsonIgnoreProperties(ignoreUnknown = true), so existing consumers are unaffected. It's the only definition in the repo (3 constructor call sites, all in this module).

New topic admitted-message-input is added to requiredTopics so it's auto-created on startup.

Configuration

Env var	Default	Purpose
SESSION_BUSY_TTL_MS	600000	Idle timeout before a busy session is force-cleared
SESSION_GUARD_PUNCTUATE_INTERVAL_MS	15000	How often the sweep scans for wedged sessions

Verification

• 119/119 tests pass in processor-apps/processing, including 10 new guard tests covering admit/reject, the role:"tool" exemption, busy clear on clean end-turn, busy retention while tools pend, TTL force-clear, and cross-session isolation.
• Dumped the real buildTopology() to confirm the wiring holds end-to-end, not just per-fragment:

  Sub-topology: 1
    Source: (topics: [message-input])
    Processor: (stores: [session-status-store])
    Sink: (topic: admitted-message-input)
  Sub-topology: 2
    Source: (topics: [admitted-message-input])   ← enrich
  

• EnrichInputMessageProcessorTest now feeds admitted-message-input directly; the guard is exercised separately in its own test.

Follow-ups, not in this PR

• No frontend feedback. The rejection reaches the client as a normal message ("The agent is still processing your previous request…"). Disabling the send control while a turn is in flight would beat letting users fire messages that bounce.
• The 600s TTL is the only unwedge path. If turns crash with any regularity, a session stays locked for up to 10 minutes. Worth revisiting if that shows up in practice.

🤖 Generated with Claude Code