diff --git a/Endpoints/ChatEndpoint.cs b/Endpoints/ChatEndpoint.cs
index daa0151..a08dde1 100644
--- a/Endpoints/ChatEndpoint.cs
+++ b/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.
- // Etwaige nicht gedrainte Einwürfe verwerfen (kein Leak in die nächste Chain).
- steeringQueue.Clear(sessionId);
+ // Steering-Gate schließen. Close() ist atomar gegen TryEnqueue: ein Einwurf,
+ // der dem User als injected:true bestätigt wurde, kann ab hier nicht mehr
+ // 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
- // Einwurf nie gedraint (kein laufender Loop) und bis zur nächsten Chain liegen bleiben.
+ // Schneller Vorab-Check: läuft überhaupt eine Chain?
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 });
}
}
diff --git a/Services/SteeringQueue.cs b/Services/SteeringQueue.cs
index aa4f132..79cd192 100644
--- a/Services/SteeringQueue.cs
+++ b/Services/SteeringQueue.cs
@@ -1,12 +1,12 @@
-using System.Collections.Concurrent;
using System.Collections.Generic;
+using System.Collections.Concurrent;
namespace Gott.Services;
///
/// 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 – keyed by sessionId.
+/// Lebenszyklus 1:1 gespiegelt vom – 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): und 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
+/// gibt ihn als "leftover" zurück, damit der Aufrufer ihn persistieren kann.
///
public class SteeringQueue
{
- private readonly ConcurrentDictionary> _queues = new ConcurrentDictionary>();
-
- /// Reiht einen Einwurf für eine Session ein (FIFO).
- public void Enqueue(string sessionId, string text)
+ private sealed class SessionState
{
- if (string.IsNullOrEmpty(sessionId) || string.IsNullOrWhiteSpace(text)) return;
- var q = _queues.GetOrAdd(sessionId, _ => new ConcurrentQueue());
- q.Enqueue(text);
+ public readonly Queue Items = new Queue();
+ public bool Open;
+ }
+
+ private readonly ConcurrentDictionary _sessions = new ConcurrentDictionary();
+
+ /// Öffnet das Steering-Gate für eine Session (beim Chain-Start aufrufen).
+ 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;
+ }
+ }
+
+ ///
+ /// 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.
+ ///
+ 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;
+ }
}
///
/// 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.
///
public List Drain(string sessionId)
{
var result = new List();
- if (_queues.TryGetValue(sessionId, out var q))
+ if (_sessions.TryGetValue(sessionId, out var s))
{
- while (q.TryDequeue(out var text))
- result.Add(text);
+ lock (s)
+ {
+ while (s.Items.Count > 0) result.Add(s.Items.Dequeue());
+ }
}
return result;
}
/// Prüft ob für eine Session Einwürfe anstehen.
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; }
+ }
- /// Entfernt die Queue einer Session (beim Chain-Ende aufräumen).
- 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).
+ ///
+ public List Close(string sessionId)
+ {
+ var leftover = new List();
+ if (_sessions.TryRemove(sessionId, out var s))
+ {
+ lock (s)
+ {
+ s.Open = false;
+ while (s.Items.Count > 0) leftover.Add(s.Items.Dequeue());
+ }
+ }
+ return leftover;
+ }
}