diff --git a/Endpoints/ChatEndpoint.cs b/Endpoints/ChatEndpoint.cs index 7fd8f1f..daa0151 100644 --- a/Endpoints/ChatEndpoint.cs +++ b/Endpoints/ChatEndpoint.cs @@ -19,9 +19,10 @@ public static class ChatEndpoint { app.MapPost("/api/chat", HandleChat); app.MapPost("/api/chat/stop", HandleStop); + app.MapPost("/api/chat/steer", HandleSteer); } - private static async Task HandleChat(HttpRequest req, HttpResponse res, ServerSettingsService settings, ChatService chatService, DatabaseService db, StreamingTracker streamingTracker, AnthropicService anthropic, OpenAiService openAi, RemoteConnectionManager remoteMgr, EmbeddingBackgroundService embeddingBg, CancellationToken cancellationToken) + private static async Task HandleChat(HttpRequest req, HttpResponse res, ServerSettingsService settings, ChatService chatService, DatabaseService db, StreamingTracker streamingTracker, SteeringQueue steeringQueue, AnthropicService anthropic, OpenAiService openAi, RemoteConnectionManager remoteMgr, EmbeddingBackgroundService embeddingBg, CancellationToken cancellationToken) { var body = await JsonSerializer.DeserializeAsync(req.Body); if (body == null) @@ -277,6 +278,9 @@ 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); // Heartbeat-Task beenden (sseLinkCts wird disposed → Token cancelled) } } @@ -297,4 +301,36 @@ public static class ChatEndpoint var stopLevel = streamingTracker.Cancel(sessionId); await res.WriteAsJsonAsync(new { stopped = stopLevel != null, level = stopLevel ?? "none" }); } + + /// + /// "Steering": Reiht eine Nachricht out-of-band in die laufende Chain ein. + /// Sie wird an der nächsten Iterations-Grenze als User-Message injiziert. + /// Anthropic-API-Pfad: Injektion zwischen den messages.create-Calls. + /// Claude-CLI-Pfad: Kill-and-Restart am nächsten Tool-Boundary (der Agentic-Loop + /// läuft im claude-Child, daher kein echtes Mid-Turn-Inject möglich). + /// + private static async Task HandleSteer(HttpRequest req, HttpResponse res, StreamingTracker streamingTracker, SteeringQueue steeringQueue) + { + var body = await JsonSerializer.DeserializeAsync(req.Body); + var sessionId = body.TryGetProperty("sessionId", out var sid) ? sid.GetString() : null; + var text = body.TryGetProperty("text", out var txt) ? txt.GetString() : null; + + if (string.IsNullOrEmpty(sessionId) || string.IsNullOrWhiteSpace(text)) + { + res.StatusCode = 400; + await res.WriteAsJsonAsync(new { error = "sessionId and text required" }); + 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. + if (!streamingTracker.IsStreaming(sessionId)) + { + await res.WriteAsJsonAsync(new { injected = false, reason = "not_streaming" }); + return; + } + + steeringQueue.Enqueue(sessionId, text.Trim()); + await res.WriteAsJsonAsync(new { injected = true }); + } } diff --git a/Program.cs b/Program.cs index a01247b..5b5b132 100644 --- a/Program.cs +++ b/Program.cs @@ -52,6 +52,8 @@ builder.Services.AddSingleton(authService); builder.Services.AddSingleton(dbService); builder.Services.AddSingleton(new ImageService()); builder.Services.AddSingleton(new StreamingTracker()); +var steeringQueue = new SteeringQueue(); +builder.Services.AddSingleton(steeringQueue); builder.Services.AddSingleton(new RemoteConnectionManager()); var memoryService = new MemoryService(dataDir); var skillService = new SkillService(dataDir); @@ -88,7 +90,8 @@ builder.Services.AddSingleton(sp => toolService, dbService, memoryService, serverSettings, vectorSearch: vectorSearchService, - embeddingBg: embeddingBgService); + embeddingBg: embeddingBgService, + steeringQueue: steeringQueue); svc.SetChatService(chatService); return svc; }); diff --git a/Services/AnthropicService.Cli.cs b/Services/AnthropicService.Cli.cs index f7f865e..35afedb 100644 --- a/Services/AnthropicService.Cli.cs +++ b/Services/AnthropicService.Cli.cs @@ -54,6 +54,7 @@ public partial class AnthropicService var tempImagePaths = new List(); string mcpConfigPath = null; bool isUserChoice = false; + CancellationTokenSource segCts = null; // aktuelles CLI-Segment (Kill-and-Restart-Steering) var chain = new ChainRecord { @@ -133,130 +134,238 @@ public partial class AnthropicService string choiceToolCallId = null; - var result = await _cliService.ChatStreamingAsync( - prompt, - async evt => + // ── Kill-and-Restart-Steering ── + // Der Agentic-Loop läuft INNERHALB von `claude -p` – gott sieht nur Events, + // echtes Mid-Turn-Inject ist über die CLI nicht möglich (Einwürfe würden nur + // gequeued). Stattdessen: kommt während eines Laufs ein Einwurf rein, killen wir + // den Child am nächsten Tool-Boundary, hängen den bisherigen Verlauf + den Einwurf + // an den Prompt und starten die CLI neu (Abbruch-Semantik – mehr geht über die CLI nicht). + var basePrompt = prompt; + var currentPrompt = prompt; + int totalTurns = 0; + double totalCost = 0; + int steerRestartCount = 0; + CliChatResult result = null; + bool steerRestart; + bool segSteerTriggered = false; + + // Event-Handler als lokale Funktion (damit der Restart-Loop flach bleibt) + async Task HandleCliEvent(CliStreamEvent evt) + { + switch (evt.Type) { - switch (evt.Type) + case CliEventType.TextStart: + await SseHelper.TryWriteSse(response, "text_start", new { index = 0 }, sseCt); + break; + case CliEventType.TextDelta: + fullText.Append(evt.Text); + await SseHelper.TryWriteSse(response, "text_delta", new { text = evt.Text }, sseCt); + break; + case CliEventType.TextEnd: + await SseHelper.TryWriteSse(response, "text_end", new { }, sseCt); + break; + case CliEventType.ToolStart: { - case CliEventType.TextStart: - await SseHelper.TryWriteSse(response, "text_start", new { index = 0 }, sseCt); - break; - case CliEventType.TextDelta: - fullText.Append(evt.Text); - await SseHelper.TryWriteSse(response, "text_delta", new { text = evt.Text }, sseCt); - break; - case CliEventType.TextEnd: - await SseHelper.TryWriteSse(response, "text_end", new { }, sseCt); - break; - case CliEventType.ToolStart: + // Text-Puffer flushen bevor Tool-Use gespeichert wird (korrekte Reihenfolge in DB) + if (fullText.Length > 0) { - // Text-Puffer flushen bevor Tool-Use gespeichert wird (korrekte Reihenfolge in DB) - if (fullText.Length > 0) - { - var textForEmbed = fullText.ToString(); - var msgId = await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = 1, - Role = "assistant", - Type = "text", - Content = textForEmbed - }); - _embeddingBg?.Enqueue(msgId, chainId, sessionId, agentId, "assistant", textForEmbed); - fullText.Clear(); - } - var mappedName = MapCliToolName(evt.ToolName); - await SseHelper.TryWriteSse(response, "tool_start", new { name = mappedName, id = evt.ToolId }, sseCt); - break; - } - case CliEventType.ToolEnd: - { - var mappedName = MapCliToolName(evt.ToolName); - var mappedInput = MapCliToolInput(evt.ToolName, evt.ToolInput); - // Tool-Use in DB speichern (damit Poll die Tools anzeigt) - await _db.SaveMessage(new MessageRecord + var textForEmbed = fullText.ToString(); + var msgId = await _db.SaveMessage(new MessageRecord { ChainId = chainId, Iteration = 1, Role = "assistant", - Type = "tool_use", - ToolName = mappedName, - ToolCallId = evt.ToolId, - Content = mappedInput + Type = "text", + Content = textForEmbed }); - await SseHelper.TryWriteSse(response, "tool_end", new { name = mappedName, id = evt.ToolId, input = mappedInput }, sseCt); - - // ask_user_choice via MCP: TCS erstellen + SSE user_choice Event - if (mappedName == "ask_user_choice") - { - try - { - var choiceInput = JsonSerializer.Deserialize(mappedInput); - var cQuestion = choiceInput.TryGetProperty("question", out var cq) ? cq.GetString() ?? "" : ""; - var cChoices = new List(); - if (choiceInput.TryGetProperty("choices", out var cc) && cc.ValueKind == JsonValueKind.Array) - foreach (var cOpt in cc.EnumerateArray()) - cChoices.Add(cOpt.GetString() ?? ""); - - // Alte Antwort entfernen (falls vorhanden) - InternalEndpoint.ChoiceAnswers.TryRemove(sessionId, out _); - - // SSE Event mit viaCli-Flag → Frontend zeigt Buttons + nutzt choice-respond Endpoint - choiceToolCallId = evt.ToolId; - await SseHelper.TryWriteSse(response, "user_choice", new - { - toolCallId = evt.ToolId, - question = cQuestion, - choices = cChoices, - viaCli = true - }, sseCt); - - isUserChoice = true; - } - catch (JsonException) { /* Input nicht parsbar – MCP-Server gibt Fehler */ } - } - break; - } - case CliEventType.ToolResult: - { - // Tool-Result in DB speichern (damit es beim Neuladen sichtbar ist) - var resultContent = SseHelper.TruncateToolResult(evt.ToolResultContent ?? ""); - await _db.SaveMessage(new MessageRecord - { - ChainId = chainId, - Iteration = 1, - Role = "user", - Type = "tool_result", - ToolCallId = evt.ToolId, - Content = resultContent, - IsError = evt.ToolResultIsError - }); - // SSE Event an Client senden - await SseHelper.TryWriteSse(response, "tool_result", new - { - id = evt.ToolId, - result = resultContent, - isError = evt.ToolResultIsError - }, sseCt); - break; + _embeddingBg?.Enqueue(msgId, chainId, sessionId, agentId, "assistant", textForEmbed); + fullText.Clear(); } + var mappedName = MapCliToolName(evt.ToolName); + await SseHelper.TryWriteSse(response, "tool_start", new { name = mappedName, id = evt.ToolId }, sseCt); + break; } - }, - modelId: modelId, - systemPrompt: systemText.Length > 0 ? systemText.ToString() : null, - workspace: workspace, - hasWorkspace: !string.IsNullOrEmpty(workspace), - planMode: planMode, - webSearch: webSearch, - mcpConfigPath: mcpConfigPath, - hasUnityMcp: !string.IsNullOrEmpty(unityMcpUrl), - effort: effort, - cancellationToken: workCt - ); + case CliEventType.ToolEnd: + { + var mappedName = MapCliToolName(evt.ToolName); + var mappedInput = MapCliToolInput(evt.ToolName, evt.ToolInput); + // Tool-Use in DB speichern (damit Poll die Tools anzeigt) + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, + Iteration = 1, + Role = "assistant", + Type = "tool_use", + ToolName = mappedName, + ToolCallId = evt.ToolId, + Content = mappedInput + }); + await SseHelper.TryWriteSse(response, "tool_end", new { name = mappedName, id = evt.ToolId, input = mappedInput }, sseCt); + + // ask_user_choice via MCP: TCS erstellen + SSE user_choice Event + if (mappedName == "ask_user_choice") + { + try + { + var choiceInput = JsonSerializer.Deserialize(mappedInput); + var cQuestion = choiceInput.TryGetProperty("question", out var cq) ? cq.GetString() ?? "" : ""; + var cChoices = new List(); + if (choiceInput.TryGetProperty("choices", out var cc) && cc.ValueKind == JsonValueKind.Array) + foreach (var cOpt in cc.EnumerateArray()) + cChoices.Add(cOpt.GetString() ?? ""); + + // Alte Antwort entfernen (falls vorhanden) + InternalEndpoint.ChoiceAnswers.TryRemove(sessionId, out _); + + // SSE Event mit viaCli-Flag → Frontend zeigt Buttons + nutzt choice-respond Endpoint + choiceToolCallId = evt.ToolId; + await SseHelper.TryWriteSse(response, "user_choice", new + { + toolCallId = evt.ToolId, + question = cQuestion, + choices = cChoices, + viaCli = true + }, sseCt); + + isUserChoice = true; + } + catch (JsonException) { /* Input nicht parsbar – MCP-Server gibt Fehler */ } + } + break; + } + case CliEventType.ToolResult: + { + // Tool-Result in DB speichern (damit es beim Neuladen sichtbar ist) + var resultContent = SseHelper.TruncateToolResult(evt.ToolResultContent ?? ""); + await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, + Iteration = 1, + Role = "user", + Type = "tool_result", + ToolCallId = evt.ToolId, + Content = resultContent, + IsError = evt.ToolResultIsError + }); + // SSE Event an Client senden + await SseHelper.TryWriteSse(response, "tool_result", new + { + id = evt.ToolId, + result = resultContent, + isError = evt.ToolResultIsError + }, sseCt); + + // Steering: Tool-Boundary erreicht – steht ein Einwurf an, Child killen + // (löst OperationCanceledException aus → Restart unten). + if (!segSteerTriggered && !isUserChoice + && _steeringQueue != null && !string.IsNullOrEmpty(sessionId) + && _steeringQueue.HasPending(sessionId)) + { + segSteerTriggered = true; + try { segCts?.Cancel(); } catch { } + } + break; + } + } + } + + do + { + steerRestart = false; + segSteerTriggered = false; + segCts?.Dispose(); + segCts = CancellationTokenSource.CreateLinkedTokenSource(workCt); + + try + { + result = await _cliService.ChatStreamingAsync( + currentPrompt, + HandleCliEvent, + modelId: modelId, + systemPrompt: systemText.Length > 0 ? systemText.ToString() : null, + workspace: workspace, + hasWorkspace: !string.IsNullOrEmpty(workspace), + planMode: planMode, + webSearch: webSearch, + mcpConfigPath: mcpConfigPath, + hasUnityMcp: !string.IsNullOrEmpty(unityMcpUrl), + effort: effort, + cancellationToken: segCts.Token + ); + } + catch (OperationCanceledException) when (segSteerTriggered && !workCt.IsCancellationRequested) + { + // Steering-Kill (kein echter Abbruch) → Restart unten vorbereiten + result = null; + } + + // Tokens des abgeschlossenen Segments akkumulieren (bei Kill: result == null) + if (result != null) + { + usage.InputTokens += result.InputTokens; + usage.OutputTokens += result.OutputTokens; + usage.CacheReadTokens += result.CacheReadTokens; + usage.CacheCreationTokens += result.CacheCreationTokens; + totalCost += result.CostUsd; + totalTurns += result.NumTurns > 0 ? result.NumTurns : 1; + } + + // Restart nötig? (a) Kill am Tool-Boundary, (b) Einwurf kam erst am Turn-Ende rein + bool steerPending = !isUserChoice && _steeringQueue != null + && !string.IsNullOrEmpty(sessionId) && _steeringQueue.HasPending(sessionId); + if (segSteerTriggered || steerPending) + { + steerRestart = true; + Console.WriteLine($"[CLI-Steering] Restart #{++steerRestartCount} (chain {chainId})"); + + // 1. Bisher gepufferten Assistant-Text dieses Segments persistieren + if (fullText.Length > 0) + { + var partial = fullText.ToString(); + var pid = await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = 1, + Role = "assistant", Type = "text", Content = partial + }); + _embeddingBg?.Enqueue(pid, chainId, sessionId, agentId, "assistant", partial); + fullText.Clear(); + } + + // 2. Verlauf dieses Durchlaufs aus der DB als Transkript (vor dem Persistieren der neuen Einwürfe) + var progressLog = await BuildCliProgressLog(chainId); + + // 3. Einwürfe entnehmen, RAW persistieren (Type=steer), via SSE melden, gewrappt sammeln + var steers = _steeringQueue.Drain(sessionId); + var wrapped = new StringBuilder(); + foreach (var steer in steers) + { + var sid = await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = 1, + Role = "user", Type = "steer", Content = steer + }); + _embeddingBg?.Enqueue(sid, chainId, sessionId, agentId, "user", steer); + await SseHelper.TryWriteSse(response, "steer_injected", new { text = steer }, sseCt); + wrapped.AppendLine(WrapSteeringMessage(steer)); + } + + // 4. Restart-Prompt: Basis-Prompt + bisheriger Verlauf + Einwürfe + var sb = new StringBuilder(basePrompt); + if (!string.IsNullOrEmpty(progressLog)) + { + sb.Append("\n\n\nDu arbeitest bereits an obiger Aufgabe. Bisheriger Verlauf in diesem Durchlauf (Text, Tool-Aufrufe, Ergebnisse):\n"); + sb.Append(progressLog); + sb.Append("\n"); + } + sb.Append("\n\n"); + sb.Append(wrapped.ToString().Trim()); + currentPrompt = sb.ToString(); + } + } + while (steerRestart && !workCt.IsCancellationRequested); // Bei CLI-Fehler → Fehlermeldung an Client senden und in DB speichern - if (!result.Success && !isUserChoice) + if (result != null && !result.Success && !isUserChoice) { var errorMsg = !string.IsNullOrEmpty(result.Error) ? result.Error : "CLI hat keine Antwort geliefert"; // Error in DB speichern (damit er beim Neuladen sichtbar ist) @@ -271,7 +380,7 @@ public partial class AnthropicService await SseHelper.TryWriteSse(response, "error", new { error = $"CLI-Fehler: {errorMsg}" }, sseCt); } - // Antwort in DB speichern + // Antwort in DB speichern (finaler Text des letzten Segments) if (fullText.Length > 0) { var finalText = fullText.ToString(); @@ -286,24 +395,19 @@ public partial class AnthropicService _embeddingBg?.Enqueue(msgId, chainId, sessionId, agentId, "assistant", finalText); } - // Token-Usage vom CLI-Result übernehmen - usage.InputTokens = result.InputTokens; - usage.OutputTokens = result.OutputTokens; - usage.CacheReadTokens = result.CacheReadTokens; - usage.CacheCreationTokens = result.CacheCreationTokens; - - var stopReason = isUserChoice ? "awaiting_user_choice" : (result.Success ? "end_turn" : "error"); - await FinalizeChain(chain, usage, result.NumTurns > 0 ? result.NumTurns : 1, stopReason, sw); + // Token-Usage wurde segmentweise in usage/totalTurns/totalCost akkumuliert (siehe Restart-Loop) + var stopReason = isUserChoice ? "awaiting_user_choice" : (result != null && result.Success ? "end_turn" : "error"); + await FinalizeChain(chain, usage, totalTurns > 0 ? totalTurns : 1, stopReason, sw); await SseHelper.TryWriteSse(response, "done", new { model = modelId, - inputTokens = result.InputTokens, - outputTokens = result.OutputTokens, - cacheCreationTokens = result.CacheCreationTokens, - cacheReadTokens = result.CacheReadTokens, - costUsd = result.CostUsd, - iterations = result.NumTurns > 0 ? result.NumTurns : 1, + inputTokens = usage.InputTokens, + outputTokens = usage.OutputTokens, + cacheCreationTokens = usage.CacheCreationTokens, + cacheReadTokens = usage.CacheReadTokens, + costUsd = totalCost, + iterations = totalTurns > 0 ? totalTurns : 1, chainId, durationMs = sw.ElapsedMilliseconds, stopReason, @@ -360,6 +464,8 @@ public partial class AnthropicService } finally { + segCts?.Dispose(); + // Temp-Bilddateien aufräumen CleanupTempImages(tempImagePaths); @@ -371,6 +477,43 @@ public partial class AnthropicService } } + /// + /// Baut aus den bereits gespeicherten Messages einer Chain ein kompaktes Verlauf-Transkript + /// für den Kill-and-Restart-Steering-Prompt (was hat das Modell in diesem Durchlauf bereits getan). + /// + private async Task BuildCliProgressLog(string chainId) + { + var msgs = await _db.GetChainMessages(chainId); + var sb = new StringBuilder(); + foreach (var m in msgs) + { + switch (m.Type) + { + case "text": + if (m.Role == "assistant" && !string.IsNullOrWhiteSpace(m.Content)) + sb.AppendLine("[Du]: " + m.Content.Trim()); + break; + case "tool_use": + sb.AppendLine($"[Du hast aufgerufen: {m.ToolName}] {TruncateForLog(m.Content, 600)}"); + break; + case "tool_result": + sb.AppendLine($"[Ergebnis{(m.IsError ? " (Fehler)" : "")}]: {TruncateForLog(m.Content, 600)}"); + break; + case "steer": + sb.AppendLine("[Vorheriger Live-Einwurf des Users]: " + (m.Content ?? "").Trim()); + break; + } + } + return sb.ToString().Trim(); + } + + private static string TruncateForLog(string s, int max) + { + if (string.IsNullOrEmpty(s)) return ""; + s = s.Replace("\r", " ").Replace("\n", " ").Trim(); + return s.Length <= max ? s : s.Substring(0, max) + " …[gekürzt]"; + } + /// /// Non-Streaming Chat über die Claude CLI für Telegram (und andere Caller, die kein SSE brauchen). /// Sammelt den finalen Text und gibt ihn zurück. Persistiert Text, Tool-Use und Tool-Result in der DB, diff --git a/Services/AnthropicService.Streaming.cs b/Services/AnthropicService.Streaming.cs index 2e4a929..3a9cdf5 100644 --- a/Services/AnthropicService.Streaming.cs +++ b/Services/AnthropicService.Streaming.cs @@ -14,6 +14,45 @@ namespace Gott.Services; public partial class AnthropicService { + /// + /// Verpackt einen Live-Einwurf (Steering) des Users mit einem Hinweis, damit das + /// Modell ihn aktiv einordnet statt ihn nur passiv anzuhängen: laufende Aufgabe + /// ergänzen ODER Kurs ändern. + /// + private static string WrapSteeringMessage(string text) + { + return "[Live-Einwurf vom User – kam herein, während du an der laufenden Aufgabe arbeitest]:\n" + + text + + "\n\nBerücksichtige diesen Einwurf ab sofort. Ergänzt er nur die laufende Aufgabe → arbeite weiter und beziehe ihn ein. Ändert er den Kurs → passe dein Vorgehen an und sag in einem Satz, was du jetzt anders machst."; + } + + /// + /// Draint die Steering-Queue der Session und hängt etwaige Live-Einwürfe als + /// Text-Blöcke (mit Hinweis-Wrapper) an die übergebene Content-Liste eines + /// user-Turns. Persistiert jeden Einwurf RAW (Role=user, Type=steer) für History + + /// Re-Chat und meldet ihn via SSE (steer_injected) ans Frontend. + /// Gibt die Anzahl injizierter Einwürfe zurück. + /// + private async Task InjectSteeringAsync(HttpResponse response, List content, string chainId, string sessionId, string agentId, int iteration, CancellationToken sseCt) + { + if (_steeringQueue == null || string.IsNullOrEmpty(sessionId)) return 0; + var steers = _steeringQueue.Drain(sessionId); + foreach (var steer in steers) + { + // API: gewrappter Text-Block (mit Hinweis). DB: RAW-Text (sauber für History/Re-Chat). + content.Add(new { type = "text", text = WrapSteeringMessage(steer) }); + var msgId = await _db.SaveMessage(new MessageRecord + { + ChainId = chainId, Iteration = iteration, + Role = "user", Type = "steer", + Content = steer + }); + _embeddingBg?.Enqueue(msgId, chainId, sessionId, agentId, "user", steer); + await SseHelper.TryWriteSse(response, "steer_injected", new { text = steer }, sseCt); + } + return steers.Count; + } + public async Task StreamChat( HttpResponse response, string modelId, @@ -288,6 +327,23 @@ public partial class AnthropicService iterations += finalizerIterations; } + // ── Steering-Injektion (Drain-Punkt C) ── + // Das Modell würde jetzt seinen Turn beenden. Hat der User einen Einwurf + // geschickt (während des Turns oder der Review/Finalizer-Phase), injizieren + // wir ihn als neuen user-Turn und lassen die Chain WEITERlaufen statt zu + // finalisieren. NICHT bei awaiting_user_choice – dort wartet die Chain + // bewusst auf die Choice-Antwort des Users. + if (stopReason != "awaiting_user_choice") + { + var steerContent = new List(); + var injectedCount = await InjectSteeringAsync(response, steerContent, chainId, sessionId, agentId, iterations, sseCt); + if (injectedCount > 0) + { + messages.Add(new { role = "user", content = steerContent }); + continue; // → Chain läuft weiter + } + } + var finalStopReason = string.IsNullOrEmpty(stopReason) ? "unexpected_end" : stopReason; await FinalizeChain(chain, usage, iterations, finalStopReason, sw, streamErrorInfo); await SseHelper.TryWriteSse(response, "done", new @@ -320,6 +376,12 @@ public partial class AnthropicService }); } + // ── Steering-Injektion (Drain-Punkt B) ── + // Live-Einwürfe, die während dieser Tool-Runde eingingen, an DENSELBEN + // user-Turn anhängen (tool_result-Blöcke zuerst, dann Text – API-konform). + // So sieht das Modell den Einwurf in der nächsten Iteration, mitten in der Arbeit. + await InjectSteeringAsync(response, toolResults, chainId, sessionId, agentId, iterations, sseCt); + messages.Add(new { role = "user", content = toolResults }); // Hard-Stop Check nach Tool-Ausführung: Sofort raus statt nächste Iteration diff --git a/Services/AnthropicService.cs b/Services/AnthropicService.cs index dc84d61..3eb8c41 100644 --- a/Services/AnthropicService.cs +++ b/Services/AnthropicService.cs @@ -43,11 +43,12 @@ public partial class AnthropicService private readonly ServerSettingsService _settings; private readonly VectorSearchService _vectorSearch; private readonly EmbeddingBackgroundService _embeddingBg; + private readonly SteeringQueue _steeringQueue; private ChatService _chatService; private TelegramService _telegramService; private ClaudeCliService _cliService; - public AnthropicService(IHttpClientFactory httpFactory, ToolService toolService, DatabaseService db, MemoryService memoryService = null, ServerSettingsService settings = null, VectorSearchService vectorSearch = null, EmbeddingBackgroundService embeddingBg = null) + public AnthropicService(IHttpClientFactory httpFactory, ToolService toolService, DatabaseService db, MemoryService memoryService = null, ServerSettingsService settings = null, VectorSearchService vectorSearch = null, EmbeddingBackgroundService embeddingBg = null, SteeringQueue steeringQueue = null) { _httpFactory = httpFactory; _toolService = toolService; @@ -56,6 +57,7 @@ public partial class AnthropicService _settings = settings; _vectorSearch = vectorSearch; _embeddingBg = embeddingBg; + _steeringQueue = steeringQueue; } public void SetCliService(ClaudeCliService cliService) diff --git a/Services/ChatService.cs b/Services/ChatService.cs index cb7887e..7caa08c 100644 --- a/Services/ChatService.cs +++ b/Services/ChatService.cs @@ -294,7 +294,7 @@ Für welche Plattform soll die App entwickelt werden? Web, Mobile oder Desktop? { foreach (var msg in msgs) { - if (msg.Role == "user" && msg.Type == "text" && !string.IsNullOrEmpty(msg.Content)) + if (msg.Role == "user" && (msg.Type == "text" || msg.Type == "steer") && !string.IsNullOrEmpty(msg.Content)) { history.Add(new Dictionary { @@ -362,6 +362,15 @@ Für welche Plattform soll die App entwickelt werden? Web, Mobile oder Desktop? if (content.Count > 0) history.Add(new Dictionary { ["role"] = "user", ["content"] = content }); } + else if (msg.Role == "user" && msg.Type == "steer") + { + // Live-Einwurf (Steering) des Users während die Chain lief. + // AppendToLastMessage mergt ihn in den vorausgehenden user-Turn + // (tool_result-Runde bei Drain-B) bzw. erzeugt einen neuen user-Turn + // nach dem Assistant (Drain-C) → niemals zwei aufeinanderfolgende user-Turns. + if (!string.IsNullOrEmpty(msg.Content)) + AppendToLastMessage(history, "user", new { type = "text", text = msg.Content }); + } else if (msg.Role == "assistant" && msg.Type == "thinking" && includeThinking) { // Thinking-Inhalt für OpenAI-kompatible Provider mitführen. diff --git a/Services/SteeringQueue.cs b/Services/SteeringQueue.cs new file mode 100644 index 0000000..aa4f132 --- /dev/null +++ b/Services/SteeringQueue.cs @@ -0,0 +1,54 @@ +using System.Collections.Concurrent; +using System.Collections.Generic; + +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. +/// +/// 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). +/// +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) + { + if (string.IsNullOrEmpty(sessionId) || string.IsNullOrWhiteSpace(text)) return; + var q = _queues.GetOrAdd(sessionId, _ => new ConcurrentQueue()); + q.Enqueue(text); + } + + /// + /// Entnimmt ALLE pending Einwürfe einer Session (FIFO) und gibt sie zurück. + /// Leere Liste wenn nichts ansteht. + /// + public List Drain(string sessionId) + { + var result = new List(); + if (_queues.TryGetValue(sessionId, out var q)) + { + while (q.TryDequeue(out var text)) + result.Add(text); + } + 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; + + /// Entfernt die Queue einer Session (beim Chain-Ende aufräumen). + public void Clear(string sessionId) => _queues.TryRemove(sessionId, out _); +} diff --git a/wwwroot/css/chat.css b/wwwroot/css/chat.css index 9462a92..3f04beb 100644 --- a/wwwroot/css/chat.css +++ b/wwwroot/css/chat.css @@ -159,6 +159,29 @@ background: var(--f2-bg); } +/* Steering: Live-Einwurf während eine Chain läuft */ +.msg.user.steer { + border-color: var(--warning); + border-left-width: 3px; +} + +.steer-label { + font-size: .72em; + font-weight: 600; + color: var(--warning); + text-transform: uppercase; + letter-spacing: .04em; + margin-bottom: 4px; + opacity: .9; +} + +.steer-label .fa { margin-right: 3px; } + +/* Eingabefeld signalisiert "Einwurf-Modus" solange ein Stream läuft */ +#input.steer-mode { + box-shadow: inset 0 0 0 1px var(--warning); +} + .msg.assistant { align-self: flex-start; background: var(--f1-bg); diff --git a/wwwroot/js/api.js b/wwwroot/js/api.js index 467de8c..7f7da7e 100644 --- a/wwwroot/js/api.js +++ b/wwwroot/js/api.js @@ -78,6 +78,9 @@ function apiChoiceRespond(sessionId, choice) { function apiStopChat(sessionId) { return _post('/api/chat/stop', { sessionId: sessionId }); } +/** Steering: Nachricht in die LAUFENDE Chain einwerfen (wird an nächster Iterations-Grenze injiziert). */ +function apiChatSteer(sessionId, text) { return _post('/api/chat/steer', { sessionId: sessionId, text: text }); } + // ── Agent API ── function apiGetAgents() { return _get('/api/agents'); } diff --git a/wwwroot/js/bubbles.js b/wwwroot/js/bubbles.js index deb0383..fdabcb8 100644 --- a/wwwroot/js/bubbles.js +++ b/wwwroot/js/bubbles.js @@ -157,6 +157,27 @@ function addUserBubble(text, streamed = false, timestamp = '', messageId = null, scrollDown(); } +// ── Steering Bubble (Live-Einwurf während eine Chain läuft) ── + +/** Baut eine Steering-Bubble (User-Stil + Einwurf-Label). Returns den div – für Live + History. */ +function _buildSteerBubble(opts) { + const div = _createMessageBubble({ role: 'user', extraClass: 'user steer', timestamp: opts.timestamp, messageId: opts.messageId, isMarked: opts.isMarked, deletedOn: opts.deletedOn, archived: opts.archived, streamed: opts.streamed }); + const label = document.createElement('div'); + label.className = 'steer-label'; + label.innerHTML = ' Einwurf'; + div.appendChild(label); + renderUserContent(div, opts.text); + _addBubbleActions(div, { timestamp: opts.timestamp, messageId: opts.messageId, deletedOn: opts.deletedOn, isMarked: opts.isMarked }); + return div; +} + +function addSteerBubble(text, streamed = false) { + const div = _buildSteerBubble({ text, streamed }); + chatEl().appendChild(div); + scrollDown(); + return div; +} + function _addSystemBubble(text, icon, color, streamed) { const div = document.createElement('div'); div.className = 'msg system-info'; diff --git a/wwwroot/js/chat.js b/wwwroot/js/chat.js index 89f599c..a5f4d69 100644 --- a/wwwroot/js/chat.js +++ b/wwwroot/js/chat.js @@ -136,6 +136,12 @@ function handleStreamEvent(eventType, data) { addWarningBubble(`Auto-Continue (${data.attempt}/${data.maxAttempts}): Modell wird fortgesetzt…`, true); break; + case 'steer_injected': + // Das Modell hat den Einwurf an dieser Stelle erhalten. Die Anzeige übernimmt + // bereits die optimistische Bubble aus sendSteer(); am Stream-Ende ersetzt + // pollHistory() sie durch die persistierte DB-Version (Type=steer). + break; + case 'warning': case 'info': addWarningBubble(data.message, true); @@ -278,6 +284,14 @@ async function _runStream(fetchPromise) { _stopClickCount = 0; startTypingIndicator(); + // Während des Streams wird die Eingabe zum Steering-Einwurf (Enter wirft ein). + const _steerInputEl = $('#input'); + const _prevPlaceholder = _steerInputEl ? _steerInputEl.placeholder : ''; + if (_steerInputEl) { + _steerInputEl.placeholder = '💬 Einwurf an die laufende Arbeit… (Enter)'; + _steerInputEl.classList.add('steer-mode'); + } + try { const resp = await fetchPromise(_abortController.signal); const reader = resp.body.getReader(); @@ -331,6 +345,11 @@ async function _runStream(fetchPromise) { stopBtn.title = 'Antwort abbrechen'; isStreaming = false; _abortController = null; + // Steering-Hinweis zurücksetzen (Eingabe ist wieder normales Senden) + if (_steerInputEl) { + _steerInputEl.placeholder = _prevPlaceholder; + _steerInputEl.classList.remove('steer-mode'); + } // Meta zurücksetzen falls kein done-Event kam (z.B. nach Stop) var metaEl = $('#meta'); if (metaEl && (metaEl.textContent.includes('Streaming') || metaEl.textContent.includes('...'))) { @@ -350,8 +369,14 @@ async function _runStream(fetchPromise) { // ── Send Message ── async function sendMessage() { - if (isStreaming) return; const input = $('#input'); + if (isStreaming) { + // Steering: Während eine Chain läuft, wird die Nachricht als Einwurf in die + // laufende Arbeit eingespeist (statt wie früher hart blockiert zu werden). + const steerText = input.value.trim(); + if (steerText) sendSteer(steerText); + return; + } const text = input.value.trim(); if (!text && pendingFiles.length === 0 && pendingRefs.length === 0) return; @@ -424,6 +449,34 @@ async function sendMessage() { await _runStream((signal) => apiChat(content, getCurrentSessionId(), signal, _planModeActive)); } +// ── Steering (Einwurf in die laufende Chain) ── +async function sendSteer(text) { + const input = $('#input'); + // Eingabe leeren + optimistische Bubble (wird am Stream-Ende durch die DB-Version ersetzt) + input.value = ''; + input.dispatchEvent(new Event('input')); + const bubble = addSteerBubble(text, true); + scrollDownForce(); + + let res; + try { + res = await apiChatSteer(getCurrentSessionId(), text); + } catch (e) { + if (bubble) bubble.remove(); + input.value = text; + input.dispatchEvent(new Event('input')); + addWarningBubble('Einwurf fehlgeschlagen: ' + (e.message || e)); + return; + } + if (!res || !res.injected) { + // Race: die laufende Antwort ist gerade fertig geworden → Text zurücklegen, normal erneut senden + if (bubble) bubble.remove(); + input.value = text; + input.dispatchEvent(new Event('input')); + addWarningBubble('Die laufende Antwort ist bereits beendet – bitte erneut senden.'); + } +} + // ── Direct Message (für Inline-Choice-Buttons) ── async function sendMessageDirect(text) { if (isStreaming || !text) return; diff --git a/wwwroot/js/history.js b/wwwroot/js/history.js index 6d4d116..749fb74 100644 --- a/wwwroot/js/history.js +++ b/wwwroot/js/history.js @@ -261,6 +261,20 @@ function _renderMessagesIntoFragment(fragment, msgs, imageMessageIds, opts) { } break; } + case 'steer': { + // Live-Einwurf des Users während die Chain lief (User-Stil + Einwurf-Label) + if (msg.role === 'user' && msg.content) { + fragment.appendChild(_buildSteerBubble({ + text: msg.content, + timestamp: msg.createdAt, + messageId: msg.id, + isMarked: msg.isMarked, + deletedOn: msg.deletedOn, + archived: archived + })); + } + break; + } case 'auto_context': { if (msg.content) { const acDiv = document.createElement('div');