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). /// 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) /// und injiziert die Texte als User-Message in die laufende Konversation. /// Ein laufender messages.create-Request ist unveränderlich – Injektion geht /// nur ZWISCHEN den API-Calls, nicht mitten im Stream. /// /// 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 sealed class SessionState { 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. Das Gate bleibt offen. /// public List Drain(string sessionId) { var result = new List(); if (_sessions.TryGetValue(sessionId, out var s)) { 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) { if (!_sessions.TryGetValue(sessionId, out var s)) return false; lock (s) { return s.Items.Count > 0; } } /// /// 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; } }