using System.Diagnostics;
using System.Text.Json;
using ClawdDotNet.Core.Api;
using ClawdDotNet.Core.Api.Models;
using ClawdDotNet.Core.Audit;
using ClawdDotNet.Core.Budget;
using ClawdDotNet.Core.Config;
using ClawdDotNet.Core.Memory;
using ClawdDotNet.Core.Security;
using ClawdDotNet.Core.Accounting;
using ClawdDotNet.Core.Tools;
using ClawdDotNet.Core.State;
using ClawdDotNet.Core.Storage;
using Microsoft.Extensions.Logging;
namespace ClawdDotNet.Core.Engine;
public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecutor
{
private readonly IChatCompletionClient _client;
private readonly ToolRegistry _toolRegistry;
private readonly PermissionGate _permissionGate;
private readonly IStateStore _stateStore;
private readonly IMemoryRepository? _memoryRepository;
private readonly Tasks.ITaskRepository? _taskRepository;
private readonly IAuditRepository? _auditRepository;
private readonly Staging.StagingGate? _stagingGate;
private readonly IUsageRepository? _usageRepository;
private readonly BudgetGuard? _budgetGuard;
private readonly ModelPricingCatalog? _pricing;
private readonly ILoggerFactory _loggerFactory;
/// Tagesgrenzen der Instanz. Wird vom Host gesetzt.
public InstanceBudget InstanceBudget { get; set; } = InstanceBudget.Unlimited;
///
/// Wie viele Weitergaben eine Delegationskette höchstens haben darf (B8).
/// A→B ist die Tiefe 1, A→B→C die Tiefe 2.
///
/// Zyklen fängt schon die Kette selbst ab — dieser Wert begrenzt die Staffel, die
/// keiner ist: A beauftragt B, B beauftragt C, C beauftragt D. Jeder Schritt hält
/// das Gate seines Vorgängers und kostet einen vollen Lauf; drei Weitergaben sind
/// mehr, als eine sinnvolle Arbeitsteilung braucht.
///
/// Instanzweit und nicht je Agent: Eine Kette gehört keinem der Beteiligten.
///
public int MaxDelegationDepth { get; set; } = 3;
///
/// Wie lange ein delegierter Lauf höchstens auf das Gate seines Zielagenten
/// wartet (#461). Danach kommt eine Absage statt weiterer Wartezeit.
///
/// Die Delegationskette schließt Zyklen aus, aber nicht das Über-Kreuz: Lauf 1
/// startet bei A und beauftragt B, Lauf 2 startet zeitgleich bei B und beauftragt A.
/// Beide Ketten sind für sich zyklusfrei — trotzdem hält jeder das Gate, auf das der
/// andere wartet. Eine Kette kann das grundsätzlich nicht sehen; nur eine Frist
/// löst es auf.
///
/// Nur für delegierte Läufe. Ein vom Menschen angestoßener Chat reiht sich
/// weiterhin ein und wartet, so lange es dauert — dort gibt es niemanden, der
/// verklemmen könnte, und ein Abbruch wäre nur lästig.
///
public TimeSpan DelegationGateTimeout { get; set; } = TimeSpan.FromSeconds(120);
private readonly ContextCompactor _compactor;
private readonly Dictionary> _chatHistories = new();
private readonly Dictionary> _chatContexts = new();
private readonly Lock _lock = new();
///
/// Alle aktiven Chat-Läufe je Agent — laufende wie wartende. Mehrere Quellen können
/// denselben Agenten gleichzeitig ansprechen (WebView, ToolJob, AgentComm), deshalb
/// eine Liste: AbortChat muss jeden davon erreichen.
///
private readonly Dictionary> _runningChats = new();
///
/// Serialisiert ChatAsync pro Agent. Der Konversationskontext ist eine geteilte
/// Liste — liefen zwei Chats desselben Agenten gleichzeitig, verschränkten sich ihre
/// Nachrichten zu einer ungültigen Tool-Sequenz, die die API mit HTTP 400 ablehnt.
/// Verschiedene Agenten bleiben unabhängig voneinander.
///
private readonly Dictionary _agentGates = new();
private Func>? _agentConfigProvider;
private Func? _agentDirResolver;
private string _instanceId = "";
private static readonly JsonSerializerOptions _jsonOpts = new()
{
WriteIndented = true,
PropertyNamingPolicy = JsonNamingPolicy.CamelCase
};
public AgentEngine(
IChatCompletionClient client,
ToolRegistry toolRegistry,
PermissionGate permissionGate,
IStateStore stateStore,
ILoggerFactory loggerFactory,
IMemoryRepository? memoryRepository = null,
IUsageRepository? usageRepository = null,
ModelPricingCatalog? pricing = null,
Tasks.ITaskRepository? taskRepository = null,
IAuditRepository? auditRepository = null,
Staging.StagingGate? stagingGate = null)
{
_client = client;
_toolRegistry = toolRegistry;
_permissionGate = permissionGate;
_stateStore = stateStore;
_loggerFactory = loggerFactory;
_memoryRepository = memoryRepository;
_taskRepository = taskRepository;
_auditRepository = auditRepository;
_stagingGate = stagingGate;
_usageRepository = usageRepository;
_pricing = pricing;
_budgetGuard = usageRepository is null ? null : new BudgetGuard(usageRepository);
_compactor = new ContextCompactor(client, loggerFactory);
}
public event Action? OnStepCompleted;
///
/// Wird ausgelöst wenn ein neuer ChatEntry hinzugefügt wird (agentId, role, content, source).
/// Erlaubt der UI, Nachrichten aus Hintergrund-Runs (ToolJobs, AgentComm) live anzuzeigen.
/// Source gibt an, woher die Nachricht kam (webview, telegram, agentcomm, job, null).
///
public event Action? OnChatEntryAdded;
///
/// Wird nach jedem abgeschlossenen Run/Chat ausgelöst (model, result).
/// Erlaubt der UI, Token-Verbrauch und Kosten für ALLE Runs zu tracken.
///
public event Action? OnRunCompleted;
public async Task RunAsync(
AgentConfig agentConfig,
string userMessage,
string instanceId,
CancellationToken externalCt,
string? source = null,
string? taskId = null,
DelegationChain? delegation = null)
{
// Vor der ersten Anfrage prüfen — ein erschöpftes Budget soll gar nichts kosten.
if (await CheckBudgetAsync(agentConfig, externalCt) is { } denied)
return denied;
var runId = Guid.NewGuid().ToString("N");
var chain = delegation ?? DelegationChain.Root(agentConfig.AgentId);
var result = await RunCoreAsync(agentConfig, userMessage, instanceId, externalCt, runId, source, chain);
await RecordUsageAsync(agentConfig, result);
await RecordReceiptAsync(runId, agentConfig, result, source ?? AuditSource.Direct, taskId);
return result;
}
private async Task RunCoreAsync(
AgentConfig agentConfig,
string userMessage,
string instanceId,
CancellationToken externalCt,
string runId,
string? source,
DelegationChain chain)
{
var logger = _loggerFactory.CreateLogger($"ClawdDotNet.Core.Engine.{agentConfig.AgentId}");
var loopGuard = new LoopGuard(agentConfig.LoopGuard);
var sw = Stopwatch.StartNew();
using var timeoutCts = new CancellationTokenSource(agentConfig.LoopGuard.Timeout);
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(externalCt, timeoutCts.Token);
var ct = linkedCts.Token;
logger.LogInformation("Agent run started: {AgentId}, model={Model}",
agentConfig.AgentId, agentConfig.Model);
try
{
var tools = _toolRegistry.GetForAgent(agentConfig);
var toolDefinitions = BuildToolDefinitions(tools);
var messages = new List();
var systemPrompt = agentConfig.FullSystemPrompt;
if (!string.IsNullOrWhiteSpace(systemPrompt))
messages.Add(ChatMessage.System(systemPrompt));
messages.Add(ChatMessage.User(userMessage));
string? finalMessage = null;
var tally = new TokenTally();
var cachingEnabled = PromptCache.IsEnabledFor(agentConfig.PromptCaching, agentConfig.Model);
while (true)
{
ct.ThrowIfCancellationRequested();
loopGuard.RecordStep();
// Kompaktieren vor dem Aufruf — dieselbe Reihenfolge wie in ChatCoreAsync.
// Verhindert die Token-Explosion bei komplexen Analysen, und zwar bevor
// sie bezahlt ist statt danach.
if (await _compactor.CompactIfNeededAsync(messages, agentConfig.LoopGuard, agentConfig.Model, ct))
{
OnStepCompleted?.Invoke(agentConfig.AgentId, "Context kompaktiert (RunAsync)");
}
if (cachingEnabled)
PromptCache.ApplyBreakpoints(messages);
var request = new ChatRequest
{
Model = agentConfig.Model,
Messages = messages,
Tools = toolDefinitions.Count > 0 ? toolDefinitions : null,
// B11/T8: Ausgabe deckeln — die teuerste Token-Art gegen Ausreißer schützen.
MaxTokens = agentConfig.LoopGuard.MaxResponseTokens > 0
? agentConfig.LoopGuard.MaxResponseTokens
: null
};
var estimated = _compactor.EstimateFor(agentConfig.Model, messages);
var response = await _client.CompleteAsync(request, ct);
if (response.Usage is not null)
{
tally.Add(response.Usage);
loopGuard.RecordTokens(response.Usage.TotalTokens);
_compactor.Calibrate(agentConfig.Model, estimated, response.Usage.PromptTokens);
}
var choice = response.Choices.FirstOrDefault();
if (choice is null)
{
finalMessage = "[No response from model]";
break;
}
var assistantMessage = choice.Message;
if (assistantMessage.ToolCalls is { Count: > 0 })
{
messages.Add(ChatMessage.AssistantWithToolCalls(assistantMessage.ToolCalls));
foreach (var toolCall in assistantMessage.ToolCalls)
{
var toolResult = await ExecuteToolCallAsync(
toolCall, agentConfig, instanceId, tools, logger, ct, runId, source, chain);
messages.Add(ChatMessage.ToolResponse(toolCall.Id, toolResult));
}
OnStepCompleted?.Invoke(agentConfig.AgentId,
$"Step {loopGuard.Steps}: {assistantMessage.ToolCalls.Count} tool call(s) executed");
}
else
{
finalMessage = assistantMessage.Content ?? "";
break;
}
}
sw.Stop();
logger.LogInformation(
"Agent run completed: {AgentId}, steps={Steps}, tokens={Tokens} (davon {Cached} aus Cache), duration={Duration}ms",
agentConfig.AgentId, loopGuard.Steps, tally.Total, tally.Cached, sw.ElapsedMilliseconds);
var result = new AgentRunResult(
agentConfig.AgentId,
AgentRunStatus.Completed,
finalMessage,
loopGuard.Steps,
tally.Total,
sw.Elapsed)
{
PromptTokens = tally.Prompt,
CompletionTokens = tally.Completion,
CachedTokens = tally.Cached
};
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested)
{
sw.Stop();
logger.LogWarning("Agent run timed out: {AgentId} after {Duration}ms",
agentConfig.AgentId, sw.ElapsedMilliseconds);
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.Cancelled, null,
loopGuard.Steps, loopGuard.Tokens, sw.Elapsed);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
catch (OperationCanceledException)
{
sw.Stop();
logger.LogInformation("Agent run cancelled: {AgentId}", agentConfig.AgentId);
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.Cancelled, null,
loopGuard.Steps, loopGuard.Tokens, sw.Elapsed);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
catch (LoopLimitExceededException ex)
{
sw.Stop();
logger.LogWarning("Agent run loop limit exceeded: {AgentId}: {Message}",
agentConfig.AgentId, ex.Message);
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.LoopLimitExceeded, null,
loopGuard.Steps, loopGuard.Tokens, sw.Elapsed, ex);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
catch (Exception ex)
{
sw.Stop();
logger.LogError(ex, "Agent run failed: {AgentId}", agentConfig.AgentId);
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.Failed, null,
loopGuard.Steps, loopGuard.Tokens, sw.Elapsed, ex);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
}
public async Task ChatAsync(
AgentConfig agentConfig,
string userMessage,
string instanceId,
CancellationToken externalCt,
string? source = null,
string? taskId = null,
DelegationChain? delegation = null)
{
if (await CheckBudgetAsync(agentConfig, externalCt) is { } denied)
return denied;
var runId = Guid.NewGuid().ToString("N");
var chain = delegation ?? DelegationChain.Root(agentConfig.AgentId);
// Abbrechbar sein, schon bevor der Lauf an der Reihe ist — sonst hängt eine
// wartende Nachricht auch dann noch, wenn der Benutzer längst abgebrochen hat.
using var runCts = CancellationTokenSource.CreateLinkedTokenSource(externalCt);
RegisterRun(agentConfig.AgentId, runCts);
var gate = GetAgentGate(agentConfig.AgentId);
// Ein delegierter Lauf wartet nur begrenzt (#461): Zwei Ketten über Kreuz
// halten jeweils das Gate, auf das die andere wartet, und keine der beiden
// sieht die andere. Ein vom Menschen angestoßener Chat wartet dagegen weiter,
// so lange es dauert — dort gibt es niemanden, der verklemmen könnte.
var gateWait = chain.Depth > 0 ? DelegationGateTimeout : Timeout.InfiniteTimeSpan;
try
{
if (!await gate.WaitAsync(gateWait, runCts.Token))
{
UnregisterRun(agentConfig.AgentId, runCts);
return new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.TargetBusy,
$"[Agent belegt: '{agentConfig.AgentId}' hat innerhalb von " +
$"{gateWait.TotalSeconds:0} Sekunden nicht frei gehabt]",
0, 0, TimeSpan.Zero);
}
}
catch (OperationCanceledException)
{
UnregisterRun(agentConfig.AgentId, runCts);
return new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.Cancelled, "[Chat abgebrochen]",
0, 0, TimeSpan.Zero);
}
try
{
var result = await ChatCoreAsync(agentConfig, userMessage, instanceId, runCts.Token, source, runId, chain);
await RecordUsageAsync(agentConfig, result);
await RecordReceiptAsync(runId, agentConfig, result, source, taskId);
return result;
}
finally
{
UnregisterRun(agentConfig.AgentId, runCts);
gate.Release();
}
}
private async Task ChatCoreAsync(
AgentConfig agentConfig,
string userMessage,
string instanceId,
CancellationToken runCt,
string? source,
string runId,
DelegationChain chain)
{
var logger = _loggerFactory.CreateLogger($"ClawdDotNet.Core.Engine.Chat.{agentConfig.AgentId}");
var loopGuard = new LoopGuard(agentConfig.LoopGuard);
var sw = Stopwatch.StartNew();
// Die Timeout-Uhr läuft erst ab hier — Wartezeit in der Warteschlange
// darf den Lauf nicht aufzehren.
using var timeoutCts = new CancellationTokenSource(agentConfig.LoopGuard.Timeout);
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(runCt, timeoutCts.Token);
var ct = linkedCts.Token;
try
{
var tools = _toolRegistry.GetForAgent(agentConfig);
var toolDefinitions = BuildToolDefinitions(tools);
List messages;
lock (_lock)
{
if (!_chatContexts.TryGetValue(agentConfig.AgentId, out messages!))
{
messages = new List();
var systemPrompt = agentConfig.FullSystemPrompt;
if (!string.IsNullOrWhiteSpace(systemPrompt))
messages.Add(ChatMessage.System(systemPrompt));
_chatContexts[agentConfig.AgentId] = messages;
}
}
// Routing-Hinweis: Dem Agenten mitteilen, woher die Nachricht kommt
var routedMessage = source switch
{
ChatSource.WebView => $"[WebView Chat – antworte als normaler Text, NICHT über Telegram senden]\n{userMessage}",
ChatSource.Telegram => userMessage, // Telegram-Nachrichten kommen bereits mit [Telegram] Prefix vom Job
_ => userMessage
};
messages.Add(ChatMessage.User(routedMessage));
AddChatEntry(agentConfig.AgentId, "user", userMessage, source);
string? finalMessage = null;
var tally = new TokenTally();
var cachingEnabled = PromptCache.IsEnabledFor(agentConfig.PromptCaching, agentConfig.Model);
while (true)
{
ct.ThrowIfCancellationRequested();
loopGuard.RecordStep();
// Kompaktieren VOR dem Aufruf. Vorher lief die Prüfung danach und mit
// den promptTokens der gerade bezahlten Anfrage: Der überfüllte Prompt
// war zu dem Zeitpunkt längst abgerechnet, gekürzt wurde erst für den
// nächsten Schritt. Genau ein Schritt zu spät, und der teuerste.
if (await _compactor.CompactIfNeededAsync(messages, agentConfig.LoopGuard, agentConfig.Model, ct))
{
PersistChatState(agentConfig.AgentId);
OnStepCompleted?.Invoke(agentConfig.AgentId, "Context kompaktiert");
}
if (cachingEnabled)
PromptCache.ApplyBreakpoints(messages);
var request = new ChatRequest
{
Model = agentConfig.Model,
Messages = messages,
Tools = toolDefinitions.Count > 0 ? toolDefinitions : null,
// B11/T8: Ausgabe deckeln — die teuerste Token-Art gegen Ausreißer schützen.
MaxTokens = agentConfig.LoopGuard.MaxResponseTokens > 0
? agentConfig.LoopGuard.MaxResponseTokens
: null
};
// Was die Schätzung für genau diese Anfrage sagt — gleich neben dem, was
// die API dafür wirklich berechnet. Der Vergleich der beiden ist die
// einzige Rückmeldung, die die Schätzung je bekommt.
var estimated = _compactor.EstimateFor(agentConfig.Model, messages);
var response = await _client.CompleteAsync(request, ct);
if (response.Usage is not null)
{
tally.Add(response.Usage);
loopGuard.RecordTokens(response.Usage.TotalTokens);
_compactor.Calibrate(agentConfig.Model, estimated, response.Usage.PromptTokens);
}
var choice = response.Choices.FirstOrDefault();
if (choice is null)
{
finalMessage = "[No response from model]";
break;
}
var assistantMessage = choice.Message;
if (assistantMessage.ToolCalls is { Count: > 0 })
{
messages.Add(ChatMessage.AssistantWithToolCalls(assistantMessage.ToolCalls));
foreach (var toolCall in assistantMessage.ToolCalls)
{
var toolResult = await ExecuteToolCallAsync(
toolCall, agentConfig, instanceId, tools, logger, ct, runId, source, chain);
messages.Add(ChatMessage.ToolResponse(toolCall.Id, toolResult));
}
OnStepCompleted?.Invoke(agentConfig.AgentId,
$"Chat step {loopGuard.Steps}: {assistantMessage.ToolCalls.Count} tool call(s)");
}
else
{
finalMessage = assistantMessage.Content ?? "";
messages.Add(ChatMessage.Assistant(finalMessage));
AddChatEntry(agentConfig.AgentId, "assistant", finalMessage, source);
break;
}
}
sw.Stop();
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.Completed, finalMessage,
loopGuard.Steps, tally.Total, sw.Elapsed)
{
PromptTokens = tally.Prompt,
CompletionTokens = tally.Completion,
CachedTokens = tally.Cached
};
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
catch (OperationCanceledException)
{
sw.Stop();
var cancelMsg = "[Chat abgebrochen]";
AddChatEntry(agentConfig.AgentId, "assistant", cancelMsg, source);
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.Cancelled, cancelMsg,
loopGuard.Steps, loopGuard.Tokens, sw.Elapsed);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
catch (LoopLimitExceededException ex)
{
sw.Stop();
var limitMsg = $"[Loop-Limit erreicht: {ex.Message}]";
logger.LogWarning("Chat loop limit: {AgentId}: {Message}", agentConfig.AgentId, ex.Message);
AddChatEntry(agentConfig.AgentId, "assistant", limitMsg, source);
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.LoopLimitExceeded, limitMsg,
loopGuard.Steps, loopGuard.Tokens, sw.Elapsed, ex);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
catch (Exception ex)
{
sw.Stop();
var errorMsg = $"[Fehler: {ex.Message}]";
logger.LogError(ex, "Chat failed: {AgentId}", agentConfig.AgentId);
AddChatEntry(agentConfig.AgentId, "assistant", errorMsg, source);
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.Failed, errorMsg,
loopGuard.Steps, loopGuard.Tokens, sw.Elapsed, ex);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
}
// ─── Nebenläufigkeits-Helfer ───
private SemaphoreSlim GetAgentGate(string agentId)
{
lock (_lock)
{
if (!_agentGates.TryGetValue(agentId, out var gate))
{
gate = new SemaphoreSlim(1, 1);
_agentGates[agentId] = gate;
}
return gate;
}
}
private void RegisterRun(string agentId, CancellationTokenSource cts)
{
lock (_lock)
{
if (!_runningChats.TryGetValue(agentId, out var list))
{
list = new List();
_runningChats[agentId] = list;
}
list.Add(cts);
}
}
private void UnregisterRun(string agentId, CancellationTokenSource cts)
{
lock (_lock)
{
if (!_runningChats.TryGetValue(agentId, out var list))
return;
list.Remove(cts);
if (list.Count == 0)
_runningChats.Remove(agentId);
}
}
public IReadOnlyList GetChatHistory(string agentId)
{
lock (_lock)
return _chatHistories.TryGetValue(agentId, out var history)
? history.ToList().AsReadOnly()
: [];
}
public bool IsRunning(string agentId)
{
lock (_lock)
return _runningChats.ContainsKey(agentId);
}
/// Anzahl Agenten mit mindestens einem aktiven Chat-Lauf — für Diagnose/Heartbeat.
public int RunningChatCount
{
get { lock (_lock) return _runningChats.Count; }
}
///
/// Momentaufnahme des Konversationskontexts eines Agenten — also der Nachrichten,
/// die beim nächsten Schritt tatsächlich an das Modell gehen.
/// Nützlich für Diagnose und Kontextgrößen-Anzeige.
///
public IReadOnlyList GetChatContext(string agentId)
{
lock (_lock)
return _chatContexts.TryGetValue(agentId, out var ctx)
? ctx.ToList()
: [];
}
///
/// Bricht ALLE Chat-Läufe des Agenten ab — laufende wie wartende.
///
public void AbortChat(string agentId)
{
List toCancel;
lock (_lock)
{
if (!_runningChats.TryGetValue(agentId, out var list))
return;
toCancel = list.ToList();
}
// Außerhalb des Locks abbrechen: Cancel führt Continuations aus, die
// ihrerseits wieder auf _lock zugreifen können.
foreach (var cts in toCancel)
{
try { cts.Cancel(); }
catch (ObjectDisposedException) { /* Lauf war bereits fertig */ }
}
}
public void ClearChatHistory(string agentId)
{
lock (_lock)
{
_chatHistories.Remove(agentId);
_chatContexts.Remove(agentId);
}
var dir = _agentDirResolver?.Invoke(agentId);
if (dir is null) return;
var historyPath = Path.Combine(dir, "ChatHistory.json");
var contextPath = Path.Combine(dir, "ChatContext.json");
if (File.Exists(historyPath)) File.Delete(historyPath);
if (File.Exists(contextPath)) File.Delete(contextPath);
}
public void SetAgentConfigProvider(
Func> provider,
string instanceId,
Func? agentDirResolver = null)
{
_agentConfigProvider = provider;
_instanceId = instanceId;
_agentDirResolver = agentDirResolver;
}
public void LoadPersistedChats()
{
var configs = _agentConfigProvider?.Invoke() ?? [];
foreach (var agent in configs)
{
var dir = _agentDirResolver?.Invoke(agent.AgentId);
if (dir is null || !Directory.Exists(dir)) continue;
try
{
var historyPath = Path.Combine(dir, "ChatHistory.json");
if (File.Exists(historyPath))
{
var history = JsonSerializer.Deserialize>(
AtomicFile.ReadAllText(historyPath), _jsonOpts);
if (history is { Count: > 0 })
{
lock (_lock)
_chatHistories[agent.AgentId] = history;
}
}
var contextPath = Path.Combine(dir, "ChatContext.json");
if (File.Exists(contextPath))
{
var raw = AtomicFile.ReadAllText(contextPath);
List? context = null;
// Versuche zuerst als Array (direktes List)
// Dann als Wrapper-Objekt {"messages":[...]}
try
{
context = JsonSerializer.Deserialize>(raw, _jsonOpts);
}
catch (JsonException)
{
try
{
using var doc = JsonDocument.Parse(raw);
if (doc.RootElement.TryGetProperty("messages", out var msgs))
{
context = JsonSerializer.Deserialize>(
msgs.GetRawText(), _jsonOpts);
}
}
catch { /* Beide Formate fehlgeschlagen — ignorieren */ }
}
if (context is { Count: > 0 })
{
lock (_lock)
_chatContexts[agent.AgentId] = context;
}
}
}
catch (Exception ex)
{
_loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Persistence")
.LogWarning(ex, "Failed to load chat state for agent {AgentId}", agent.AgentId);
}
}
}
// ─── IAgentMessageRouter ───
///
/// Darf dieser Aufruf weitergegeben werden (B8)? Gibt den Ablehnungsgrund zurück
/// oder null, wenn nichts dagegen spricht.
///
/// Der Grund geht als Ergebnis an den aufrufenden Agenten zurück, nicht als
/// Ausnahme: Er soll lesen können, warum es nicht geht, und einen anderen Weg
/// wählen. Deshalb steht die ganze Kette in der Meldung — „Zyklus" allein sagt
/// niemandem, welcher.
///
private string? CheckDelegation(DelegationChain? chain, string fromAgentId, string toAgentId)
{
// Ohne Kette ist der Aufruf von aussen angestossen: Der Absender ist der erste
// in der Kette, weitergegeben wurde noch nichts.
var callerChain = chain ?? DelegationChain.Root(fromAgentId);
// Der eigentliche Deadlock. Der Zielagent hält sein Gate schon irgendwo weiter
// oben in dieser Kette und gibt es erst frei, wenn dieser Aufruf zurückkommt —
// was er nur könnte, wenn das Gate frei wäre.
if (callerChain.Contains(toAgentId))
return $"Zyklus in der Delegationskette: {callerChain} -> {toAgentId}. " +
$"Agent '{toAgentId}' wartet in dieser Kette bereits auf eine Antwort und " +
"kann keinen zweiten Auftrag annehmen, ohne sich selbst zu blockieren.";
if (callerChain.Depth >= MaxDelegationDepth)
return $"Delegationstiefe erschöpft: {callerChain} hat die Grenze von " +
$"{MaxDelegationDepth} Weitergaben erreicht. Erledige den Rest selbst " +
"oder gib die Aufgabe an den Auftraggeber zurück.";
return null;
}
public IReadOnlyList ListAgents(string callerAgentId)
{
var configs = _agentConfigProvider?.Invoke() ?? [];
return configs
.Where(a => a.AgentId != callerAgentId)
.Select(a => new AgentInfo(a.AgentId, a.DisplayName, a.Description, a.Model, IsRunning(a.AgentId)))
.ToList();
}
///
/// Die Instanz des Aufrufers, sonst die der Engine (B13).
///
/// _instanceId wird nur in SetAgentConfigProvider gesetzt und war
/// vorher die einzige Quelle — ein Lauf, der mit einer anderen Instanz-Id
/// angestoßen wurde, verlor sie beim Weiterreichen, und ohne den Aufruf war sie leer.
///
private string ResolveInstanceId(string? fromCaller)
=> string.IsNullOrWhiteSpace(fromCaller) ? _instanceId : fromCaller;
public async Task SendMessageAsync(
string fromAgentId, string toAgentId, string message,
string? instanceId, DelegationChain? chain, CancellationToken ct)
{
var configs = _agentConfigProvider?.Invoke();
if (configs is null)
return new AgentMessageResult(false, null, "Agent config provider not set.");
// Selbstadressierung würde am Agent-Gate hängen bleiben: Der laufende Chat
// hält es bereits und würde auf sich selbst warten.
if (fromAgentId == toAgentId)
return new AgentMessageResult(false, null,
"Ein Agent kann sich keine Nachricht an sich selbst schicken.");
var targetConfig = configs.FirstOrDefault(a => a.AgentId == toAgentId);
if (targetConfig is null)
return new AgentMessageResult(false, null, $"Agent '{toAgentId}' not found.");
if (CheckDelegation(chain, fromAgentId, toAgentId) is { } refused)
return new AgentMessageResult(false, null, refused);
var callerChain = chain ?? DelegationChain.Root(fromAgentId);
var fromConfig = configs.FirstOrDefault(a => a.AgentId == fromAgentId);
var fromName = fromConfig?.DisplayName ?? fromAgentId;
var wrappedMessage = $"[Nachricht von Agent \"{fromName}\" ({fromAgentId})]\n\n{message}";
try
{
var result = await ChatAsync(targetConfig, wrappedMessage, ResolveInstanceId(instanceId), ct,
source: ChatSource.AgentComm, delegation: callerChain.Append(toAgentId));
return result.Status switch
{
AgentRunStatus.Completed => new AgentMessageResult(true, result.FinalMessage),
AgentRunStatus.LoopLimitExceeded => new AgentMessageResult(true, result.FinalMessage,
"Agent hat das Step-Limit erreicht, die Nachricht wurde aber zugestellt."),
// #461: Das Ziel war die ganze Frist über belegt. Für den fragenden
// Agenten ist das kein Fehler, sondern eine Auskunft — er kann es
// später erneut versuchen oder ohne die Antwort weiterarbeiten.
AgentRunStatus.TargetBusy => new AgentMessageResult(false, null,
$"Agent '{targetConfig.DisplayName}' ({toAgentId}) ist belegt und hat innerhalb " +
$"von {DelegationGateTimeout.TotalSeconds:0} Sekunden nicht frei gehabt. " +
"Später erneut versuchen oder ohne diese Antwort weitermachen."),
_ => new AgentMessageResult(false, null, $"Agent run status: {result.Status}")
};
}
catch (Exception ex)
{
return new AgentMessageResult(false, null, $"Failed to reach agent: {ex.Message}");
}
}
public async Task SpawnAgentAsync(
string fromAgentId, string targetAgentId, string taskMessage,
string? instanceId, DelegationChain? chain, CancellationToken ct)
{
var configs = _agentConfigProvider?.Invoke();
if (configs is null)
return new AgentSpawnResult(false, targetAgentId, null, "Agent config provider not set.");
var targetConfig = configs.FirstOrDefault(a => a.AgentId == targetAgentId);
if (targetConfig is null)
return new AgentSpawnResult(false, targetAgentId, null, $"Agent '{targetAgentId}' not found.");
// Auch der Spawn zählt: Er startet einen vollen Lauf und kann seinerseits
// delegieren. Ohne diese Prüfung wäre die Kette über spawn zu umgehen.
if (CheckDelegation(chain, fromAgentId, targetAgentId) is { } refused)
return new AgentSpawnResult(false, targetAgentId, null, refused);
var callerChain = chain ?? DelegationChain.Root(fromAgentId);
if (IsRunning(targetAgentId))
return new AgentSpawnResult(false, targetAgentId, null,
$"Agent '{targetConfig.DisplayName}' läuft bereits. Verwende send_message statt spawn.");
var fromConfig = configs.FirstOrDefault(a => a.AgentId == fromAgentId);
var fromName = fromConfig?.DisplayName ?? fromAgentId;
var wrappedMessage = $"[Spawn-Auftrag von Agent \"{fromName}\" ({fromAgentId})]\n\n{taskMessage}";
try
{
var result = await RunAsync(targetConfig, wrappedMessage, ResolveInstanceId(instanceId), ct,
delegation: callerChain.Append(targetAgentId));
return result.Status == AgentRunStatus.Completed
? new AgentSpawnResult(true, targetAgentId, result.FinalMessage)
: new AgentSpawnResult(false, targetAgentId, null, $"Agent run status: {result.Status}");
}
catch (Exception ex)
{
return new AgentSpawnResult(false, targetAgentId, null, $"Spawn failed: {ex.Message}");
}
}
private void AddChatEntry(string agentId, string role, string content, string? source = null)
{
lock (_lock)
{
if (!_chatHistories.TryGetValue(agentId, out var history))
{
history = new List();
_chatHistories[agentId] = history;
}
history.Add(new ChatEntry(role, content, DateTime.Now, source));
}
PersistChatState(agentId);
OnChatEntryAdded?.Invoke(agentId, role, content, source);
}
private void PersistChatState(string agentId)
{
var dir = _agentDirResolver?.Invoke(agentId);
if (dir is null) return;
try
{
Directory.CreateDirectory(dir);
List? history;
List? context;
lock (_lock)
{
_chatHistories.TryGetValue(agentId, out history);
_chatContexts.TryGetValue(agentId, out context);
}
// Atomar schreiben: Ein Absturz mitten im Vorgang würde sonst den
// bisherigen Verlauf löschen und einen halben zurücklassen.
if (history is not null)
AtomicFile.WriteAllText(
Path.Combine(dir, "ChatHistory.json"),
JsonSerializer.Serialize(history, _jsonOpts));
if (context is not null)
AtomicFile.WriteAllText(
Path.Combine(dir, "ChatContext.json"),
JsonSerializer.Serialize(context, _jsonOpts));
}
catch (Exception ex)
{
_loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Persistence")
.LogWarning(ex, "Failed to persist chat state for agent {AgentId}", agentId);
}
}
private async Task ExecuteToolCallAsync(
ToolCall toolCall,
AgentConfig agentConfig,
string instanceId,
IReadOnlyList availableTools,
ILogger logger,
CancellationToken ct,
string runId,
string? source,
DelegationChain chain)
{
var toolName = toolCall.Function.Name;
var arguments = toolCall.Function.Arguments ?? "";
var sw = Stopwatch.StartNew();
// Das Audit wird von der Engine gestempelt (A3) — Herkunft aus dem Wissen der
// Engine, nie aus dem Tool-Ergebnis. Best effort: ein Audit-Fehler darf den Lauf
// nicht scheitern lassen.
Task Audit(AuditStatus status, string summary)
=> RecordAuditAsync(runId, agentConfig, source, toolName, arguments, status, summary, sw.ElapsedMilliseconds);
try
{
_permissionGate.Enforce(agentConfig.AgentId, toolName, agentConfig);
var tool = availableTools.FirstOrDefault(t => t.Name == toolName);
if (tool is null)
{
await Audit(AuditStatus.NotFound, $"Tool '{toolName}' nicht zugewiesen/unbekannt");
return JsonSerializer.Serialize(ToolResult.Fail($"Tool '{toolName}' not found."));
}
// Staging-Durchsetzung (A2): irreversible Aktionen werden vorgeschlagen statt
// ausgeführt. Eine Prompt-Injection kann so nur einen Vorschlag erzeugen.
if (_stagingGate is not null)
{
var intercept = await _stagingGate.InterceptAsync(
agentConfig.AgentId, instanceId, runId, toolName, arguments, ct);
if (intercept.Outcome == Staging.StagingOutcome.Denied)
{
await Audit(AuditStatus.Denied, intercept.Message);
return JsonSerializer.Serialize(new { error = intercept.Message });
}
if (intercept.Outcome == Staging.StagingOutcome.Staged)
{
await Audit(AuditStatus.Staged, intercept.Message);
return intercept.Message; // dem Agenten als reguläres Tool-Ergebnis
}
}
var input = string.IsNullOrWhiteSpace(toolCall.Function.Arguments)
? default
: JsonDocument.Parse(toolCall.Function.Arguments).RootElement;
var context = BuildToolContext(agentConfig, instanceId, toolName, ct, chain);
logger.LogDebug("Executing tool {Tool} for agent {AgentId}", toolName, agentConfig.AgentId);
var result = await tool.ExecuteAsync(input, context, ct);
logger.LogDebug("Tool {Tool} completed: success={Success}", toolName, result.Success);
if (!result.Success)
{
await Audit(AuditStatus.Error, result.ErrorMessage ?? "");
return JsonSerializer.Serialize(new { error = result.ErrorMessage });
}
var content = TruncateToolResult(result.Content, agentConfig.MaxToolResultChars);
if (content.Length != result.Content.Length)
{
logger.LogInformation(
"Tool-Ergebnis von {Tool} gekürzt: {Original} → {Limit} Zeichen",
toolName, result.Content.Length, agentConfig.MaxToolResultChars);
}
await Audit(AuditStatus.Ok, "");
return content;
}
catch (ToolAccessDeniedException ex)
{
logger.LogWarning("Tool access denied: {Message}", ex.Message);
await Audit(AuditStatus.Denied, ex.Message);
return JsonSerializer.Serialize(new { error = ex.Message });
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
// Nicht als Tool-Fehler zurückgeben: Sonst läuft die Schleife noch einen
// Schritt weiter und der Abbruch greift erst verzögert. Auch kein Audit —
// der Aufruf kam nicht zum Abschluss.
throw;
}
catch (Exception ex)
{
logger.LogError(ex, "Tool {Tool} threw an exception", toolName);
await Audit(AuditStatus.Error, ex.Message);
return JsonSerializer.Serialize(new { error = $"Tool execution failed: {ex.Message}" });
}
}
private AgentToolContext BuildToolContext(
AgentConfig agentConfig, string instanceId, string toolName, CancellationToken ct,
DelegationChain? chain = null)
{
var toolConfig = agentConfig.Tools.TryGetValue(toolName, out var cfg)
? cfg.AsReadOnly()
: new Dictionary().AsReadOnly();
var toolLogger = _loggerFactory.CreateLogger($"ClawdDotNet.Tools.{toolName}.Execution");
return new AgentToolContext(
agentConfig.AgentId,
instanceId,
toolConfig,
_stateStore,
toolLogger,
ct,
agentConfig.WorkspacePath,
agentConfig.SharedWorkspacePath,
this,
_memoryRepository,
_taskRepository,
chain ?? DelegationChain.Root(agentConfig.AgentId));
}
///
/// Führt einen freigegebenen, eingefrorenen Aufruf aus (A2) — mit gültigem Tool-Kontext,
/// aber ohne LLM-Schleife und ohne erneute Staging-Prüfung. Genau der übergebene
/// Argument-JSON wird ausgeführt (Plan-Freeze).
///
public async Task ExecuteApprovedCallAsync(
string agentId, string tool, string argumentsJson, string runId, CancellationToken ct)
{
var config = _agentConfigProvider?.Invoke().FirstOrDefault(a => a.AgentId == agentId);
if (config is null)
return JsonSerializer.Serialize(new { error = $"Agent '{agentId}' nicht gefunden." });
var agentTool = _toolRegistry.GetForAgent(config).FirstOrDefault(t => t.Name == tool)
?? _toolRegistry.Get(tool);
if (agentTool is null)
return JsonSerializer.Serialize(new { error = $"Tool '{tool}' nicht gefunden." });
var sw = Stopwatch.StartNew();
try
{
var input = string.IsNullOrWhiteSpace(argumentsJson)
? default
: JsonDocument.Parse(argumentsJson).RootElement;
var context = BuildToolContext(config, _instanceId, tool, ct);
var result = await agentTool.ExecuteAsync(input, context, ct);
var status = result.Success ? AuditStatus.Ok : AuditStatus.Error;
await RecordAuditAsync(runId, config, AuditSource.Approval, tool, argumentsJson,
status, result.Success ? "Freigegeben ausgeführt" : (result.ErrorMessage ?? ""), sw.ElapsedMilliseconds);
return result.Success
? TruncateToolResult(result.Content, config.MaxToolResultChars)
: JsonSerializer.Serialize(new { error = result.ErrorMessage });
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
await RecordAuditAsync(runId, config, AuditSource.Approval, tool, argumentsJson,
AuditStatus.Error, ex.Message, sw.ElapsedMilliseconds);
return JsonSerializer.Serialize(new { error = $"Tool execution failed: {ex.Message}" });
}
}
// ─── Budget und Verbrauchserfassung ───
///
/// Prüft das Tagesbudget. Gibt ein Ergebnis zurück, wenn der Lauf nicht stattfinden
/// darf — sonst null.
///
private async Task CheckBudgetAsync(AgentConfig agentConfig, CancellationToken ct)
{
if (_budgetGuard is null)
return null;
var status = await _budgetGuard.CheckAsync(agentConfig, InstanceBudget, ct);
if (status.IsAllowed)
return null;
var logger = _loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Budget");
logger.LogWarning("Lauf abgelehnt für {AgentId}: {Reason}", agentConfig.AgentId, status.Reason);
var result = new AgentRunResult(
agentConfig.AgentId, AgentRunStatus.BudgetExceeded,
$"[Budget erschöpft] {status.Reason}", 0, 0, TimeSpan.Zero);
OnRunCompleted?.Invoke(agentConfig.Model, result);
return result;
}
///
/// Schreibt den Verbrauch eines Laufs fort. Fehler hierbei dürfen den Lauf nicht
/// nachträglich scheitern lassen — die eigentliche Arbeit ist bereits getan.
///
private async Task RecordUsageAsync(AgentConfig agentConfig, AgentRunResult result)
{
if (_usageRepository is null)
return;
try
{
var estimate = _pricing?.Estimate(agentConfig.Model, result.PromptTokens, result.CompletionTokens);
await _usageRepository.RecordAsync(new RunUsage
{
AgentId = agentConfig.AgentId,
Model = agentConfig.Model,
PromptTokens = result.PromptTokens,
CompletionTokens = result.CompletionTokens,
CachedTokens = result.CachedTokens,
CostUsd = estimate?.Usd ?? 0m,
CostIsKnown = estimate?.IsKnown ?? false,
Status = result.Status.ToString(),
StepCount = result.StepCount,
DurationMs = (long)result.Duration.TotalMilliseconds,
OccurredAt = DateTime.Now
}, CancellationToken.None);
}
catch (Exception ex)
{
_loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Usage")
.LogWarning(ex, "Verbrauch konnte nicht festgehalten werden für {AgentId}",
agentConfig.AgentId);
}
}
// ─── Audit-Log und Receipts (A3) ───
///
/// Schreibt einen Tool-Aufruf ins Audit-Log. Best effort — ein Fehler hierbei darf den
/// Lauf nicht scheitern lassen; die eigentliche Arbeit ist bereits getan.
///
private async Task RecordAuditAsync(
string runId, AgentConfig agentConfig, string? source, string tool,
string arguments, AuditStatus status, string summary, long durationMs)
{
if (_auditRepository is null)
return;
try
{
await _auditRepository.AppendAsync(new AuditEntry
{
RunId = runId,
AgentId = agentConfig.AgentId,
Model = agentConfig.Model,
Source = AuditSource.Normalize(source),
Tool = tool,
Arguments = Cap(arguments, 4_000),
Status = status,
Summary = Cap(summary, 500),
DurationMs = durationMs,
OccurredAt = DateTime.Now
}, CancellationToken.None);
}
catch (Exception ex)
{
_loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Audit")
.LogWarning(ex, "Audit-Eintrag konnte nicht geschrieben werden ({Tool})", tool);
}
}
///
/// Hält den Abschluss-Beleg eines Laufs fest (Receipt). Best effort, wie beim Audit.
///
private async Task RecordReceiptAsync(
string runId, AgentConfig agentConfig, AgentRunResult result, string? source, string? taskId)
{
if (_auditRepository is null)
return;
try
{
var estimate = _pricing?.Estimate(agentConfig.Model, result.PromptTokens, result.CompletionTokens);
await _auditRepository.RecordReceiptAsync(new RunReceipt
{
RunId = runId,
AgentId = agentConfig.AgentId,
Model = agentConfig.Model,
Source = AuditSource.Normalize(source),
TaskId = taskId,
Status = result.Status.ToString(),
StepCount = result.StepCount,
PromptTokens = result.PromptTokens,
CompletionTokens = result.CompletionTokens,
CachedTokens = result.CachedTokens,
CostUsd = estimate?.Usd ?? 0m,
CostIsKnown = estimate?.IsKnown ?? false,
DurationMs = (long)result.Duration.TotalMilliseconds,
ResultRef = Cap(result.FinalMessage ?? "", 500),
OccurredAt = DateTime.Now
}, CancellationToken.None);
}
catch (Exception ex)
{
_loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Receipt")
.LogWarning(ex, "Receipt konnte nicht geschrieben werden für {AgentId}", agentConfig.AgentId);
}
}
private static string Cap(string value, int max)
=> value.Length <= max ? value : value[..max] + "…";
/// Sammelt die Token-Zahlen über alle Schritte eines Runs.
private sealed class TokenTally
{
public int Total { get; private set; }
public int Prompt { get; private set; }
public int Completion { get; private set; }
public int Cached { get; private set; }
// Voll qualifiziert: "Usage" ist auch ein Namespace (ClawdDotNet.Core.Accounting).
public void Add(Api.Models.Usage usage)
{
Total += usage.TotalTokens;
Prompt += usage.PromptTokens;
Completion += usage.CompletionTokens;
Cached += usage.CachedTokens;
}
}
///
/// Kürzt ein Tool-Ergebnis, bevor es in den Kontext wandert.
///
/// Ohne diese Grenze kann ein einzelner Aufruf den Kontext sprengen — ein WebFetch
/// mit dem Standardlimit von 512 KB entspricht rund 130.000 Tokens in EINER
/// Tool-Antwort. Die Compaction greift erst danach, der teure Request ist zu dem
/// Zeitpunkt längst bezahlt.
///
internal static string TruncateToolResult(string result, int maxChars)
{
if (maxChars <= 0 || result.Length <= maxChars)
return result;
var omitted = result.Length - maxChars;
return result[..maxChars] +
$"\n\n[… {omitted:N0} Zeichen gekürzt. Das Ergebnis war zu groß für den Kontext. " +
"Grenze die Abfrage ein, wenn du den Rest brauchst.]";
}
private static List BuildToolDefinitions(IReadOnlyList tools)
{
return tools.Select(t => new ToolDefinition
{
Function = new FunctionDefinition
{
Name = t.Name,
Description = t.Description,
Parameters = t.InputSchema
}
}).ToList();
}
}