From de07b49b6709a4e022687e66c89e8eee16e0e13d Mon Sep 17 00:00:00 2001 From: Meleki Date: Wed, 10 Jun 2026 14:50:31 +0200 Subject: [PATCH] Persist steering messages that arrive during chain finalization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- Endpoints/ChatEndpoint.cs | 39 ++++++++++++++--- Services/SteeringQueue.cs | 89 ++++++++++++++++++++++++++++++++------- 2 files changed, 106 insertions(+), 22 deletions(-) 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; + } }