meleki/gott
1
0 Fork 0
Code Activity

Add live steering injection into running chains

Browse Source 
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>
This commit is contained in:
MelekiandClaude Opus 4.8 committed 2026-06-02 06:49:58 +02:00
1 parent a58f279eaa
commit f2fd493e5f
12 files changed
+555 -132
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Services/AnthropicService.Cli.cs
+270 -127
View File
@@ -54,6 +54,7 @@ public partial class AnthropicService
var tempImagePaths = new List<string>();
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<JsonElement>(mappedInput);
var cQuestion = choiceInput.TryGetProperty("question", out var cq) ? cq.GetString() ?? "" : "";
var cChoices = new List<string>();
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<JsonElement>(mappedInput);
var cQuestion = choiceInput.TryGetProperty("question", out var cq) ? cq.GetString() ?? "" : "";
var cChoices = new List<string>();
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<work_in_progress>\nDu arbeitest bereits an obiger Aufgabe. Bisheriger Verlauf in diesem Durchlauf (Text, Tool-Aufrufe, Ergebnisse):\n");
sb.Append(progressLog);
sb.Append("\n</work_in_progress>");
}
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
}
}
/// <summary>
/// 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).
/// </summary>
private async Task<string> 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]";
}
/// <summary>
/// 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,