MelekiandClaude Opus 4.8
f2fd493e5f
Let the user inject a message into a running chain instead of only being
able to abort it. While a chain streams, typing + Enter sends the text to
a new /api/chat/steer endpoint; it is drained at the next iteration
boundary, wrapped with a hint so the model actively classifies it
("complements the task -> keep working" vs. "changes course -> adjust"),
and stored as a Type="steer" user message so it survives reload.
Shared infrastructure (provider-agnostic):
- SteeringQueue: out-of-band queue keyed by sessionId, mirrored from
StreamingTracker (Enqueue/Drain/Clear).
- Program.cs registers it as a singleton and passes it to AnthropicService.
- ChatEndpoint: POST /api/chat/steer enqueues only while IsStreaming,
Clear() in the finally.
- ChatService.BuildHistoryFromDb groups Type="steer" so no two
consecutive user turns are produced on re-chat.
- Frontend api.js/chat.js/bubbles.js/history.js + chat.css render the
optimistic steer bubble and the steer_injected SSE event; Enter during
a stream sends a steer instead of being blocked.
Anthropic API path:
- Drain point B (tool boundary, merged into the tool_result user turn)
and C (turn end -> inject + continue instead of finalize).
Claude CLI path (kill-and-restart):
- The agentic loop runs inside the claude child, so true mid-turn inject
is impossible. On a pending steer the child is killed at the next tool
boundary, progress is rebuilt from the DB as a <work_in_progress>
transcript, and the CLI is restarted with the steer appended.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
55 lines
2.2 KiB
C#
55 lines
2.2 KiB
C#
using System.Collections.Concurrent;
|
||
using System.Collections.Generic;
|
||
|
||
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.
|
||
///
|
||
/// 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).
|
||
/// </summary>
|
||
public class SteeringQueue
|
||
{
|
||
private readonly ConcurrentDictionary<string, ConcurrentQueue<string>> _queues = new ConcurrentDictionary<string, ConcurrentQueue<string>>();
|
||
|
||
/// <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;
|
||
var q = _queues.GetOrAdd(sessionId, _ => new ConcurrentQueue<string>());
|
||
q.Enqueue(text);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Entnimmt ALLE pending Einwürfe einer Session (FIFO) und gibt sie zurück.
|
||
/// Leere Liste wenn nichts ansteht.
|
||
/// </summary>
|
||
public List<string> Drain(string sessionId)
|
||
{
|
||
var result = new List<string>();
|
||
if (_queues.TryGetValue(sessionId, out var q))
|
||
{
|
||
while (q.TryDequeue(out var text))
|
||
result.Add(text);
|
||
}
|
||
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;
|
||
|
||
/// <summary>Entfernt die Queue einer Session (beim Chain-Ende aufräumen).</summary>
|
||
public void Clear(string sessionId) => _queues.TryRemove(sessionId, out _);
|
||
}
|