Persist steering messages that arrive during chain finalization

A steering message confirmed to the user as injected:true could still be
lost: after the loop's last HasPending check it runs out and the chain
finalizes (done event + final messages still streaming — the "few extra
messages" the user sees), then the finally block called Clear() and threw
the queue away. A message arriving in that window was accepted
(IsStreaming still true, Stop runs only in finally) but never drained and
never persisted, so it vanished after F5. Platform-independent; Windows
only widened the window via slower CLI teardown.

Replace the queue with a per-session gated state: Open() at chain start,
Close() at chain end. TryEnqueue() and Close() are mutually exclusive via
a per-session lock, so a confirmed injection can no longer be lost — the
loop either drains it or Close() returns it as leftover. The endpoint now
opens the gate at chain start, uses TryEnqueue as the authoritative check
(false → client resends as a normal message), and in finally closes the
gate and persists any leftover as Type="steer" instead of discarding it,
so it survives F5 and the model sees it on the next message.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Endpoints/ChatEndpoint.cs

@@ -204,6 +204,8 @@ public static class ChatEndpoint
         // - sseCt: Stop-Button + Client-Disconnect → für SSE-Writes an den Browser
         //   Bei Client-Disconnect werden SSE-Writes zu stillen No-Ops
         var tokens = streamingTracker.Start(sessionId, ctx.ChainId);
+        // Steering-Gate öffnen, damit Einwürfe während dieser Chain angenommen werden.
+        steeringQueue.Open(sessionId);
         var workCt = tokens.StopToken;
         var hardCt = tokens.HardToken;
         using var sseLinkCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, tokens.StopToken);
@@ -278,9 +280,27 @@ public static class ChatEndpoint
         finally
         {
             streamingTracker.Stop(sessionId);
-            // Nach Stop ist IsStreaming(sessionId) false → /api/chat/steer enqueued nichts mehr.
+            // Steering-Gate schließen. Close() ist atomar gegen TryEnqueue: ein Einwurf,
-            // Etwaige nicht gedrainte Einwürfe verwerfen (kein Leak in die nächste Chain).
+            // der dem User als injected:true bestätigt wurde, kann ab hier nicht mehr
-            steeringQueue.Clear(sessionId);
+            // verloren gehen. Im Race-Fenster zwischen dem letzten Drain des Loops und
+            // jetzt angekommene Einwürfe werden NICHT verworfen, sondern als Type="steer"
+            // persistiert (sonst Symptom: Einwurf nach F5 weg, da nie gedraint).
+            var leftoverSteers = steeringQueue.Close(sessionId);
+            foreach (var steer in leftoverSteers)
+            {
+                try
+                {
+                    await db.SaveMessage(new MessageRecord
+                    {
+                        ChainId = ctx.ChainId,
+                        Iteration = 1,
+                        Role = "user",
+                        Type = "steer",
+                        Content = steer
+                    });
+                }
+                catch { /* Persistenz best-effort – darf das Chain-Ende nicht stören */ }
+            }
             // Heartbeat-Task beenden (sseLinkCts wird disposed → Token cancelled)
         }
     }
@@ -322,15 +342,22 @@ public static class ChatEndpoint
             return;
         }
 
-        // Nur einreihen wenn tatsächlich gerade eine Chain läuft – sonst würde der
+        // Schneller Vorab-Check: läuft überhaupt eine Chain?
-        // Einwurf nie gedraint (kein laufender Loop) und bis zur nächsten Chain liegen bleiben.
         if (!streamingTracker.IsStreaming(sessionId))
         {
             await res.WriteAsJsonAsync(new { injected = false, reason = "not_streaming" });
             return;
         }
 
-        steeringQueue.Enqueue(sessionId, text.Trim());
+        // Maßgeblich ist das Steering-Gate: TryEnqueue ist atomar gegen Close() (Chain-Ende).
+        // Liefert es false, ist das Gate bereits geschlossen (Race am Chain-Ende) → der
+        // Client sendet die Nachricht dann als normale Nachricht statt sie zu verlieren.
+        if (!steeringQueue.TryEnqueue(sessionId, text.Trim()))
+        {
+            await res.WriteAsJsonAsync(new { injected = false, reason = "not_streaming" });
+            return;
+        }
+
         await res.WriteAsJsonAsync(new { injected = true });
     }
 }

Services/SteeringQueue.cs

@@ -1,12 +1,12 @@
-using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Collections.Concurrent;
 
 namespace Gott.Services;
 
 /// <summary>
 /// Out-of-band-Kanal für "Steering"-Einwürfe: Nachrichten, die der User schickt
 /// WÄHREND eine Chain läuft (vgl. /btw bzw. Queued-Steering in Claude Code).
-/// 1:1 gespiegelt vom <see cref="StreamingTracker"/> – keyed by sessionId.
+/// Lebenszyklus 1:1 gespiegelt vom <see cref="StreamingTracker"/> – keyed by sessionId.
 ///
 /// Der Streaming-Loop (AnthropicService.StreamChat) draint die Queue an den
 /// Iterations-Grenzen (nach einer Tool-Runde bzw. bevor die Chain finalisiert)
@@ -17,38 +17,95 @@ namespace Gott.Services;
 /// Im CLI-Pfad (AnthropicService.StreamChatViaCli) läuft der Agentic-Loop im
 /// claude-Child; dort draint ein Kill-and-Restart am nächsten Tool-Boundary
 /// (Child killen, Verlauf + Einwurf an den Prompt hängen, CLI neu starten).
+///
+/// WICHTIG (Race-Schutz): <see cref="TryEnqueue"/> und <see cref="Close"/> sind
+/// über ein per-Session-Lock gegenseitig ausgeschlossen. Damit kann ein Einwurf,
+/// der dem User bereits als injected:true bestätigt wurde, NICHT mehr still
+/// verloren gehen: Entweder der Loop draint ihn noch, oder <see cref="Close"/>
+/// gibt ihn als "leftover" zurück, damit der Aufrufer ihn persistieren kann.
 /// </summary>
 public class SteeringQueue
 {
-    private readonly ConcurrentDictionary<string, ConcurrentQueue<string>> _queues = new ConcurrentDictionary<string, ConcurrentQueue<string>>();
+    private sealed class SessionState
-
-    /// <summary>Reiht einen Einwurf für eine Session ein (FIFO).</summary>
-    public void Enqueue(string sessionId, string text)
     {
-        if (string.IsNullOrEmpty(sessionId) || string.IsNullOrWhiteSpace(text)) return;
+        public readonly Queue<string> Items = new Queue<string>();
-        var q = _queues.GetOrAdd(sessionId, _ => new ConcurrentQueue<string>());
+        public bool Open;
-        q.Enqueue(text);
+    }
+
+    private readonly ConcurrentDictionary<string, SessionState> _sessions = new ConcurrentDictionary<string, SessionState>();
+
+    /// <summary>Öffnet das Steering-Gate für eine Session (beim Chain-Start aufrufen).</summary>
+    public void Open(string sessionId)
+    {
+        if (string.IsNullOrEmpty(sessionId)) return;
+        var s = _sessions.GetOrAdd(sessionId, _ => new SessionState());
+        lock (s)
+        {
+            s.Items.Clear();
+            s.Open = true;
+        }
+    }
+
+    /// <summary>
+    /// Reiht einen Einwurf ein – aber NUR solange das Gate offen ist (Chain läuft
+    /// und ist noch nicht am Finalisieren). Gibt false zurück, wenn das Gate bereits
+    /// geschlossen ist (Race am Chain-Ende) → der Aufrufer soll die Nachricht dann
+    /// als normale Nachricht senden statt sie zu verlieren.
+    /// </summary>
+    public bool TryEnqueue(string sessionId, string text)
+    {
+        if (string.IsNullOrEmpty(sessionId) || string.IsNullOrWhiteSpace(text)) return false;
+        if (!_sessions.TryGetValue(sessionId, out var s)) return false;
+        lock (s)
+        {
+            if (!s.Open) return false;
+            s.Items.Enqueue(text);
+            return true;
+        }
     }
 
     /// <summary>
     /// Entnimmt ALLE pending Einwürfe einer Session (FIFO) und gibt sie zurück.
-    /// Leere Liste wenn nichts ansteht.
+    /// Leere Liste wenn nichts ansteht. Das Gate bleibt offen.
     /// </summary>
     public List<string> Drain(string sessionId)
     {
         var result = new List<string>();
-        if (_queues.TryGetValue(sessionId, out var q))
+        if (_sessions.TryGetValue(sessionId, out var s))
         {
-            while (q.TryDequeue(out var text))
+            lock (s)
-                result.Add(text);
+            {
+                while (s.Items.Count > 0) result.Add(s.Items.Dequeue());
+            }
         }
         return result;
     }
 
     /// <summary>Prüft ob für eine Session Einwürfe anstehen.</summary>
     public bool HasPending(string sessionId)
-        => _queues.TryGetValue(sessionId, out var q) && !q.IsEmpty;
+    {
+        if (!_sessions.TryGetValue(sessionId, out var s)) return false;
+        lock (s) { return s.Items.Count > 0; }
+    }
 
-    /// <summary>Entfernt die Queue einer Session (beim Chain-Ende aufräumen).</summary>
+    /// <summary>
-    public void Clear(string sessionId) => _queues.TryRemove(sessionId, out _);
+    /// Schließt das Gate (beim Chain-Ende) und gibt etwaige NICHT gedrainte
+    /// Einwürfe zurück. Diese sind im Race-Fenster zwischen dem letzten Drain des
+    /// Loops und dem Schließen angekommen, wurden dem User aber bereits als
+    /// injected:true bestätigt – der Aufrufer MUSS sie persistieren, sonst
+    /// verschwinden sie (Symptom: Einwurf nach F5 weg).
+    /// </summary>
+    public List<string> Close(string sessionId)
+    {
+        var leftover = new List<string>();
+        if (_sessions.TryRemove(sessionId, out var s))
+        {
+            lock (s)
+            {
+                s.Open = false;
+                while (s.Items.Count > 0) leftover.Add(s.Items.Dequeue());
+            }
+        }
+        return leftover;
+    }
 }