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

processor-apps/processing/src/main/java/io/flightdeck/streams/FlightDeckStreamsApp.java

@@ -6,6 +6,7 @@
 import io.flightdeck.streams.processors.EnrichInputMessageProcessor;
 import io.flightdeck.streams.processors.ExtractToolUseItemsProcessor;
 import io.flightdeck.streams.processors.MemoirSessionEndProcessor;
+import io.flightdeck.streams.processors.SessionConcurrencyGuardProcessor;
 import io.flightdeck.streams.processors.SessionEndProcessor;
 import io.flightdeck.streams.processors.TransformToolUseDoneProcessor;
 import io.flightdeck.streams.model.ThinkResponse;
@@ -160,6 +161,10 @@ static Topology buildTopology(boolean memoirEnabled, boolean userSettingEnabled)
         );
 
         // ── Register each processor fragment ──────────────────────────────────
+        // Admission control runs first: it consumes message-input and produces
+        // admitted-message-input, dropping new user messages whose session has a
+        // turn still in progress (so the LLM never sees an unanswered tool_use).
+        SessionConcurrencyGuardProcessor.register(builder, thinkStream, replyToTable);
         EnrichInputMessageProcessor.register(builder, memoirTable, thinkTable, userSettingsTable);
         ExtractToolUseItemsProcessor.register(builder, thinkStream);
         EndTurnProcessor.register(builder, thinkStream, replyToTable);
@@ -194,6 +199,7 @@ private static void ensureTopicsExist(Properties streamsProps) {
 
         List<String> requiredTopics = new java.util.ArrayList<>(List.of(
                 Topics.MESSAGE_INPUT,
+                Topics.ADMITTED_MESSAGE_INPUT,
                 Topics.ENRICHED_MESSAGE_INPUT,
                 Topics.THINK_REQUEST_RESPONSE,
                 Topics.TOOL_USE,

processor-apps/processing/src/main/java/io/flightdeck/streams/config/Topics.java

@@ -28,6 +28,15 @@ private static String requireEnv(String key) {
     /** Per-session accumulated conversation history (KTable backing store) */
     public static final String SESSION_CONTEXT           = PREFIX + "session-context";
 
+    /**
+     * Admission-controlled user/tool messages: {@code message-input} records that
+     * have passed the {@code SessionConcurrencyGuardProcessor}. A new user message
+     * for a session whose previous turn is still in progress is rejected here
+     * (never reaching enrichment), so the LLM never receives a {@code tool_use}
+     * block that is not immediately followed by its {@code tool_result}.
+     */
+    public static final String ADMITTED_MESSAGE_INPUT   = PREFIX + "admitted-message-input";
+
     /** Merged: historical context + latest user message, ready for the LLM */
     public static final String ENRICHED_MESSAGE_INPUT   = PREFIX + "enriched-message-input";
 

processor-apps/processing/src/main/java/io/flightdeck/streams/model/SessionStatus.java

@@ -0,0 +1,32 @@
+package io.flightdeck.streams.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Per-session concurrency state owned by the
+ * {@code SessionConcurrencyGuardProcessor}'s {@code session-status-store}.
+ *
+ * <p>A session is {@code busy} from the moment a user message is admitted (a turn
+ * starts) until a <em>clean end-turn</em> response is observed
+ * ({@code endTurn == true} with no outstanding tool calls). While busy, a new
+ * incoming user message is rejected rather than appended to the conversation —
+ * appending it mid-turn would place a {@code tool_use} block immediately before a
+ * {@code user} message instead of its {@code tool_result}, which the LLM API
+ * rejects with a 400 and which kills the turn.
+ *
+ * <ul>
+ *   <li>{@code turnStartedMs} — wall-clock ms when the current turn was admitted.</li>
+ *   <li>{@code lastActivityMs} — wall-clock ms of the most recent event for this
+ *       session (admit, tool continuation, or think response). Drives the
+ *       idle-based TTL safety net that unwedges a session whose turn crashed or
+ *       whose end-turn signal was lost.</li>
+ * </ul>
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public record SessionStatus(
+        @JsonProperty("busy")             boolean busy,
+        @JsonProperty("turn_started_ms")  long turnStartedMs,
+        @JsonProperty("last_activity_ms") long lastActivityMs,
+        @JsonProperty("timestamp")        String timestamp
+) {}

processor-apps/processing/src/main/java/io/flightdeck/streams/model/UserResponse.java

@@ -32,5 +32,6 @@ public record UserResponse(
         @JsonProperty("cost")         Double cost,
         @JsonProperty("source_agent") String sourceAgent,  // which agent produced this
         @JsonProperty("reply_to")     Map<String, Object> replyTo,  // multi-agent reply route (nullable)
+        @JsonProperty("status")       String status,       // null for normal LLM output; "rejected" when the agent was busy
         @JsonProperty("timestamp")    String timestamp
 ) {}

processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EndTurnProcessor.java

@@ -158,6 +158,7 @@ static UserResponse toUserResponse(String sessionId, ThinkResponse response, Str
                 response.totalSessionCost(),
                 sourceAgent,
                 parseReply(replyJson),
+                null,  // status — normal end-turn output (not a busy rejection)
                 Instant.now().toString()
         );
     }

processor-apps/processing/src/main/java/io/flightdeck/streams/processors/EnrichInputMessageProcessor.java

@@ -53,9 +53,9 @@ public static void register(StreamsBuilder builder,
                                 KTable<String, ThinkResponse> thinkTable,
                                 KTable<String, UserSettings> userSettingsTable) {
 
-        // ── Left side: incoming user messages ────────────────────────────────
+        // ── Left side: incoming user messages, post-admission-control ────────
         KStream<String, MessageInput> inputStream = builder.stream(
-                Topics.MESSAGE_INPUT,
+                Topics.ADMITTED_MESSAGE_INPUT,
                 Consumed.with(Serdes.String(), JsonSerde.of(MessageInput.class))
         );