RichardandClaude Opus 5.5
f812dbd856
- Werkzeugkontext kennt Lauf und Aufgabe des Laufs (RunId, RunTaskId); die Delegationskette traegt den Lauf des Aufrufers weiter (RunId, CallerRunId) - Aufgaben tragen origin_run / origin_task (Datei, DB, Taskboard "Arose from"), gestempelt, nie vom Agenten behauptet - Kommentare nennen den Lauf, in dem sie geschrieben wurden - Laufbeleg: CallerRunId; Laufansicht fuehrt zu "Delegiert von" und "Delegiert an" - #677: task_update ersetzt den Rumpf nicht mehr, Abnahmekriterien aendert nur der Ersteller; Kommentare der Fremdinhalt-Stufe auf fremden Aufgaben kommen gerahmt an - WebUI: Herkunft auf der Aufgabenkarte mit Link zum Lauf Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2023 lines
91 KiB
C#
2023 lines
91 KiB
C#
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.Organisation;
|
||
using ClawdDotNet.Core.Security;
|
||
using ClawdDotNet.Core.Accounting;
|
||
using ClawdDotNet.Core.Scheduling;
|
||
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;
|
||
private readonly TimeProvider _clock;
|
||
|
||
/// <summary>Tagesgrenzen der Instanz. Wird vom Host gesetzt.</summary>
|
||
public InstanceBudget InstanceBudget { get; set; } = InstanceBudget.Unlimited;
|
||
|
||
/// <summary>
|
||
/// #716: The guard that checks and reserves every run — the scanner's headroom
|
||
/// calculation has to see the same reservations. <c>null</c> without usage recording.
|
||
/// </summary>
|
||
public BudgetGuard? BudgetGuard => _budgetGuard;
|
||
|
||
/// <summary>#711: Who is working right now and who reached whom — for the team overview.</summary>
|
||
public Activity.AgentActivity Activity { get; } = new();
|
||
|
||
/// <summary>
|
||
/// #683: The organisation tree, built fresh from the configuration on each call. Without
|
||
/// it, the instance has no structure and every agent reaches every other, as before.
|
||
/// </summary>
|
||
public Func<Organisation.OrgChart>? OrganisationProvider { get; set; }
|
||
|
||
private Organisation.OrgChart CurrentOrganisation()
|
||
=> OrganisationProvider?.Invoke() ?? new Organisation.OrgChart([], _agentConfigProvider?.Invoke() ?? []);
|
||
|
||
/// <summary>The agent's system prompt with its organisation section (#683).</summary>
|
||
private string SystemPromptFor(AgentConfig agentConfig, bool hasTools)
|
||
=> Organisation.OrgBriefing.WithSection(
|
||
UntrustedContent.MitRegel(agentConfig.FullSystemPrompt, hasTools),
|
||
Organisation.OrgBriefing.For(CurrentOrganisation(), agentConfig.AgentId, _clock.GetUtcNow().UtcDateTime, TrustOf(agentConfig)));
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
public int MaxDelegationDepth { get; set; } = 3;
|
||
|
||
/// <summary>
|
||
/// Wie lange ein <b>delegierter</b> Lauf höchstens auf das Gate seines Zielagenten
|
||
/// wartet (#461). Danach kommt eine Absage statt weiterer Wartezeit.
|
||
///
|
||
/// <para>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.</para>
|
||
///
|
||
/// <para>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.</para>
|
||
/// </summary>
|
||
public TimeSpan DelegationGateTimeout { get; set; } = TimeSpan.FromSeconds(120);
|
||
|
||
private readonly ContextCompactor _compactor;
|
||
|
||
private readonly Dictionary<string, List<ChatEntry>> _chatHistories = new();
|
||
private readonly Dictionary<string, List<ChatMessage>> _chatContexts = new();
|
||
private readonly Lock _lock = new();
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
private readonly Dictionary<string, List<CancellationTokenSource>> _runningChats = new();
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
private readonly Dictionary<string, SemaphoreSlim> _agentGates = new();
|
||
|
||
private Func<IReadOnlyList<AgentConfig>>? _agentConfigProvider;
|
||
private Func<string, string?>? _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,
|
||
TimeProvider? clock = null)
|
||
{
|
||
_clock = clock ?? TimeProvider.System;
|
||
_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<string, string>? OnStepCompleted;
|
||
|
||
/// <summary>#660: Where prompt and response of every run are kept. <c>null</c> = not at all.</summary>
|
||
public Transcripts.IRunTranscriptStore? TranscriptStore { get; init; }
|
||
|
||
/// <summary>
|
||
/// #660: Instance-wide retention in days (0 = never, -1 = forever). A function, so a
|
||
/// changed setting applies to the next run without a restart.
|
||
/// </summary>
|
||
public Func<int> DefaultTranscriptRetention { get; init; } = () => Transcripts.PromptHistorySettings.DefaultRetentionDays;
|
||
|
||
/// <summary>
|
||
/// 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).
|
||
/// </summary>
|
||
public event Action<string, string, string, string?>? OnChatEntryAdded;
|
||
|
||
/// <summary>
|
||
/// Wird nach jedem abgeschlossenen Run/Chat ausgelöst (model, result).
|
||
/// Erlaubt der UI, Token-Verbrauch und Kosten für ALLE Runs zu tracken.
|
||
/// </summary>
|
||
public event Action<string, AgentRunResult>? OnRunCompleted;
|
||
|
||
public async Task<AgentRunResult> RunAsync(
|
||
AgentConfig agentConfig,
|
||
string userMessage,
|
||
string instanceId,
|
||
CancellationToken externalCt,
|
||
string? source = null,
|
||
string? taskId = null,
|
||
DelegationChain? delegation = null)
|
||
{
|
||
// Vor allem anderen: Endet die Instanz gerade, faengt hier nichts Neues mehr an.
|
||
if (RefuseWhileShuttingDown(agentConfig) is { } zu)
|
||
return zu;
|
||
|
||
// Vor der ersten Anfrage prüfen — ein erschöpftes Budget soll gar nichts kosten.
|
||
// #716: Die Reservierung gilt, bis der Verbrauch erfasst ist (Ende der Methode).
|
||
var (denied, reservation) = await CheckBudgetAsync(agentConfig, source ?? AuditSource.Direct, taskId, externalCt);
|
||
if (denied is not null)
|
||
return denied;
|
||
using var budgetReservation = reservation;
|
||
|
||
using var lauf = TrackRun();
|
||
|
||
var runId = Guid.NewGuid().ToString("N");
|
||
var chain = delegation ?? DelegationChain.Root(agentConfig.AgentId);
|
||
using var activity = Activity.Begin(runId, agentConfig.AgentId, source ?? AuditSource.Direct, taskId, chain);
|
||
var recorder = CreateRecorder(agentConfig);
|
||
var result = await RunCoreAsync(agentConfig, userMessage, instanceId, externalCt, runId, source, chain, recorder);
|
||
activity.Complete(result.Status.ToString());
|
||
await RecordUsageAsync(agentConfig, result);
|
||
await RecordReceiptAsync(runId, agentConfig, result, source ?? AuditSource.Direct, taskId, chain.CallerRunId);
|
||
await SaveTranscriptAsync(recorder, runId, agentConfig, result, source ?? AuditSource.Direct, taskId);
|
||
return result;
|
||
}
|
||
|
||
private async Task<AgentRunResult> RunCoreAsync(
|
||
AgentConfig agentConfig,
|
||
string userMessage,
|
||
string instanceId,
|
||
CancellationToken externalCt,
|
||
string runId,
|
||
string? source,
|
||
DelegationChain chain,
|
||
Transcripts.RunTranscriptRecorder recorder)
|
||
{
|
||
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);
|
||
|
||
// #578: Vor dem try, damit auch die Abbruchzweige wissen, was bis dahin bezahlt
|
||
// wurde. Vorher lag sie im try, und ein Lauf an der Schrittgrenze ging mit 0 Token
|
||
// und 0 USD in RunUsage - fuer das Tagesbudget unsichtbar.
|
||
var tally = new TokenTally();
|
||
|
||
try
|
||
{
|
||
var tools = _toolRegistry.GetForAgent(agentConfig);
|
||
var toolDefinitions = BuildToolDefinitions(tools);
|
||
|
||
var messages = new List<ChatMessage>();
|
||
|
||
// #193: Die Regel zu gerahmten Werkzeugergebnissen gehoert zum festen Teil.
|
||
// #683: dazu der Abschnitt zur Organisation.
|
||
var systemPrompt = SystemPromptFor(agentConfig, tools.Count > 0);
|
||
if (!string.IsNullOrWhiteSpace(systemPrompt))
|
||
messages.Add(ChatMessage.System(systemPrompt));
|
||
|
||
messages.Add(ChatMessage.User(MitZeit(userMessage, source)));
|
||
recorder.Begin(systemPrompt, messages[^1].Content ?? userMessage, userMessage);
|
||
|
||
string? finalMessage = null;
|
||
var truncated = false;
|
||
var cachingEnabled = PromptCache.IsEnabledFor(agentConfig.PromptCaching, agentConfig.Model);
|
||
|
||
// #577: Eine Antwort ohne Werkzeugaufruf ist nicht automatisch der Schluss.
|
||
var istAufgabe = source == ChatSource.Task;
|
||
var leerNachgefasst = false;
|
||
var ergebnisNachgefasst = false;
|
||
var abschluss = AgentRunStatus.Completed;
|
||
|
||
// #581: gleiche Fehlschlaege zaehlen, bevor sie die Schrittgrenze aufzehren.
|
||
var fehlschlaege = new FehlschlagWache();
|
||
|
||
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, tally.Add))
|
||
{
|
||
recorder.Event("Context compacted");
|
||
OnStepCompleted?.Invoke(agentConfig.AgentId, "Context kompaktiert (RunAsync)");
|
||
}
|
||
|
||
if (cachingEnabled)
|
||
PromptCache.ApplyBreakpoints(messages);
|
||
|
||
var request = BuildRequest(agentConfig, messages, toolDefinitions);
|
||
|
||
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 not null)
|
||
recorder.ModelStep(choice.Message, response.Usage, choice.FinishReason);
|
||
if (choice is null)
|
||
{
|
||
finalMessage = "[No response from model]";
|
||
break;
|
||
}
|
||
|
||
var assistantMessage = choice.Message;
|
||
|
||
if (assistantMessage.ToolCalls is { Count: > 0 })
|
||
{
|
||
// #547: Der Text neben den Aufrufen geht mit. Er ist der Plan, den
|
||
// das Modell sich selbst geschrieben hat — ohne ihn sieht es im
|
||
// nächsten Schritt nur noch, DASS es etwas getan hat.
|
||
// #549: Der Denkverlauf ebenso, sonst beginnt er jedes Mal von vorn.
|
||
messages.Add(ChatMessage.AssistantWithToolCalls(
|
||
assistantMessage.ToolCalls,
|
||
assistantMessage.Content,
|
||
assistantMessage.Reasoning,
|
||
assistantMessage.ReasoningDetails));
|
||
|
||
// #552: Die Antworten gehen im finally raus. Auf jedem Ausgangspfad
|
||
// — fertig, abgebrochen, gescheitert — bekommt jeder Aufruf seine
|
||
// Antwort, sonst bleibt eine Folge zurueck, die keine API annimmt.
|
||
var toolResults = new string?[assistantMessage.ToolCalls.Count];
|
||
string? abbruch = null;
|
||
try
|
||
{
|
||
await ExecuteToolCallsAsync(
|
||
assistantMessage.ToolCalls, toolResults, agentConfig, instanceId, tools,
|
||
logger, ct, runId, source, chain);
|
||
abbruch = fehlschlaege.Pruefe(assistantMessage.ToolCalls, toolResults);
|
||
}
|
||
finally
|
||
{
|
||
AppendToolResponses(messages, assistantMessage.ToolCalls, toolResults);
|
||
recorder.ToolResults(assistantMessage.ToolCalls, toolResults);
|
||
}
|
||
|
||
if (abbruch is not null)
|
||
throw new WiederholterFehlschlagException(abbruch);
|
||
|
||
OnStepCompleted?.Invoke(agentConfig.AgentId,
|
||
$"Step {loopGuard.Steps}: {assistantMessage.ToolCalls.Count} tool call(s) executed");
|
||
}
|
||
else
|
||
{
|
||
var text = assistantMessage.Content ?? "";
|
||
var abgeschnitten = IsTruncated(choice.FinishReason);
|
||
var urteil = LaufAbschluss.Beurteile(text, abgeschnitten, istAufgabe, leerNachgefasst, ergebnisNachgefasst);
|
||
|
||
if (urteil == AbschlussUrteil.LeerNachfassen)
|
||
{
|
||
leerNachgefasst = true;
|
||
logger.LogWarning("Leere Antwort von {AgentId} nach {Steps} Schritten - einmal nachgefasst",
|
||
agentConfig.AgentId, loopGuard.Steps);
|
||
messages.Add(ChatMessage.User(LaufAbschluss.LeerNachfrage));
|
||
recorder.EngineNote(LaufAbschluss.LeerNachfrage);
|
||
continue;
|
||
}
|
||
|
||
if (urteil == AbschlussUrteil.ErgebnisNachfassen)
|
||
{
|
||
ergebnisNachgefasst = true;
|
||
logger.LogInformation("Aufgabenlauf von {AgentId} ohne Ergebniszeile - einmal nachgefasst",
|
||
agentConfig.AgentId);
|
||
messages.Add(ChatMessage.Assistant(text));
|
||
messages.Add(ChatMessage.User(LaufAbschluss.ErgebnisNachfrage));
|
||
recorder.EngineNote(LaufAbschluss.ErgebnisNachfrage);
|
||
continue;
|
||
}
|
||
|
||
abschluss = StatusFuer(urteil, logger, agentConfig.AgentId);
|
||
finalMessage = text;
|
||
truncated = abgeschnitten;
|
||
break;
|
||
}
|
||
}
|
||
|
||
sw.Stop();
|
||
|
||
if (truncated)
|
||
{
|
||
logger.LogWarning(
|
||
"Agent run truncated: {AgentId} stieß nach {Steps} Schritten an die Ausgabegrenze " +
|
||
"von {Limit} Tokens. Das Ergebnis ist ein Teilergebnis.",
|
||
agentConfig.AgentId, loopGuard.Steps, agentConfig.LoopGuard.MaxResponseTokens);
|
||
}
|
||
|
||
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,
|
||
truncated ? AgentRunStatus.Truncated : abschluss,
|
||
finalMessage,
|
||
loopGuard.Steps,
|
||
tally.Total,
|
||
sw.Elapsed)
|
||
{
|
||
PromptTokens = tally.Prompt,
|
||
CompletionTokens = tally.Completion,
|
||
CachedTokens = tally.Cached,
|
||
ReportedCostUsd = tally.ReportedCost
|
||
};
|
||
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) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
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) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
OnRunCompleted?.Invoke(agentConfig.Model, result);
|
||
return result;
|
||
}
|
||
catch (WiederholterFehlschlagException ex)
|
||
{
|
||
sw.Stop();
|
||
logger.LogWarning("Agent run stopped: {AgentId}: {Message}", agentConfig.AgentId, ex.Message);
|
||
|
||
var result = new AgentRunResult(
|
||
agentConfig.AgentId, AgentRunStatus.RepeatedFailure, $"[Abgebrochen: {ex.Message}]",
|
||
loopGuard.Steps, tally.Total, sw.Elapsed, ex) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
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) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
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) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
OnRunCompleted?.Invoke(agentConfig.Model, result);
|
||
return result;
|
||
}
|
||
}
|
||
|
||
public async Task<AgentRunResult> ChatAsync(
|
||
AgentConfig agentConfig,
|
||
string userMessage,
|
||
string instanceId,
|
||
CancellationToken externalCt,
|
||
string? source = null,
|
||
string? taskId = null,
|
||
DelegationChain? delegation = null)
|
||
{
|
||
if (RefuseWhileShuttingDown(agentConfig) is { } zu)
|
||
return zu;
|
||
|
||
var (denied, reservation) = await CheckBudgetAsync(agentConfig, source, taskId, externalCt);
|
||
if (denied is not null)
|
||
return denied;
|
||
using var budgetReservation = reservation;
|
||
|
||
using var lauf = TrackRun();
|
||
|
||
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
|
||
{
|
||
// #711: "working" starts past the gate — waiting for it is not working.
|
||
using var activity = Activity.Begin(runId, agentConfig.AgentId, source ?? AuditSource.Direct, taskId, chain);
|
||
var recorder = CreateRecorder(agentConfig);
|
||
var result = await ChatCoreAsync(agentConfig, userMessage, instanceId, runCts.Token, source, runId, chain, recorder);
|
||
activity.Complete(result.Status.ToString());
|
||
await RecordUsageAsync(agentConfig, result);
|
||
await RecordReceiptAsync(runId, agentConfig, result, source, taskId, chain.CallerRunId);
|
||
await SaveTranscriptAsync(recorder, runId, agentConfig, result, source ?? AuditSource.Direct, taskId);
|
||
return result;
|
||
}
|
||
finally
|
||
{
|
||
UnregisterRun(agentConfig.AgentId, runCts);
|
||
gate.Release();
|
||
}
|
||
}
|
||
|
||
private async Task<AgentRunResult> ChatCoreAsync(
|
||
AgentConfig agentConfig,
|
||
string userMessage,
|
||
string instanceId,
|
||
CancellationToken runCt,
|
||
string? source,
|
||
string runId,
|
||
DelegationChain chain,
|
||
Transcripts.RunTranscriptRecorder recorder)
|
||
{
|
||
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;
|
||
|
||
// #578: siehe RunCoreAsync - auch Abbrueche tragen ihre bezahlten Token.
|
||
var tally = new TokenTally();
|
||
|
||
try
|
||
{
|
||
var tools = _toolRegistry.GetForAgent(agentConfig);
|
||
var toolDefinitions = BuildToolDefinitions(tools);
|
||
|
||
List<ChatMessage> messages;
|
||
lock (_lock)
|
||
{
|
||
if (!_chatContexts.TryGetValue(agentConfig.AgentId, out messages!))
|
||
{
|
||
messages = new List<ChatMessage>();
|
||
// #193 - siehe RunCoreAsync. Bestehende Verlaeufe behalten ihren
|
||
// Systemprompt; die Regel kommt mit dem naechsten neuen Kontext.
|
||
var systemPrompt = SystemPromptFor(agentConfig, tools.Count > 0);
|
||
if (!string.IsNullOrWhiteSpace(systemPrompt))
|
||
messages.Add(ChatMessage.System(systemPrompt));
|
||
_chatContexts[agentConfig.AgentId] = messages;
|
||
}
|
||
else if (messages.Count > 0 && messages[0].Role == "system")
|
||
{
|
||
// #683: Nur der Organisationsabschnitt folgt der Struktur — nach einem
|
||
// Umzug weiss der Agent sonst im laufenden Chat nichts davon. Der Rest
|
||
// des bestehenden Systemprompts bleibt, wie er war (#193).
|
||
var section = Organisation.OrgBriefing.For(
|
||
CurrentOrganisation(), agentConfig.AgentId, _clock.GetUtcNow().UtcDateTime, TrustOf(agentConfig));
|
||
var refreshed = Organisation.OrgBriefing.WithSection(messages[0].Content ?? "", section);
|
||
if (refreshed != messages[0].Content)
|
||
messages[0] = ChatMessage.System(refreshed);
|
||
}
|
||
}
|
||
|
||
// 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(MitZeit(routedMessage, source)));
|
||
AddChatEntry(agentConfig.AgentId, "user", userMessage, source);
|
||
recorder.Begin(
|
||
messages.Count > 0 && messages[0].Role == "system" ? messages[0].Content?.ToString() : null,
|
||
messages[^1].Content ?? userMessage,
|
||
userMessage);
|
||
|
||
string? finalMessage = null;
|
||
var truncated = false;
|
||
var cachingEnabled = PromptCache.IsEnabledFor(agentConfig.PromptCaching, agentConfig.Model);
|
||
|
||
// #577: Eine Antwort ohne Werkzeugaufruf ist nicht automatisch der Schluss.
|
||
var istAufgabe = source == ChatSource.Task;
|
||
var leerNachgefasst = false;
|
||
var ergebnisNachgefasst = false;
|
||
var abschluss = AgentRunStatus.Completed;
|
||
|
||
// #581: gleiche Fehlschlaege zaehlen, bevor sie die Schrittgrenze aufzehren.
|
||
var fehlschlaege = new FehlschlagWache();
|
||
|
||
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, tally.Add))
|
||
{
|
||
recorder.Event("Context compacted");
|
||
PersistChatState(agentConfig.AgentId);
|
||
OnStepCompleted?.Invoke(agentConfig.AgentId, "Context kompaktiert");
|
||
}
|
||
|
||
if (cachingEnabled)
|
||
PromptCache.ApplyBreakpoints(messages);
|
||
|
||
var request = BuildRequest(agentConfig, messages, toolDefinitions);
|
||
|
||
// 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 not null)
|
||
recorder.ModelStep(choice.Message, response.Usage, choice.FinishReason);
|
||
if (choice is null)
|
||
{
|
||
finalMessage = "[No response from model]";
|
||
break;
|
||
}
|
||
|
||
var assistantMessage = choice.Message;
|
||
|
||
if (assistantMessage.ToolCalls is { Count: > 0 })
|
||
{
|
||
// #547 / #549 — siehe RunCoreAsync: Begleittext und Denkverlauf
|
||
// gehören in den Verlauf, nicht in den Papierkorb.
|
||
messages.Add(ChatMessage.AssistantWithToolCalls(
|
||
assistantMessage.ToolCalls,
|
||
assistantMessage.Content,
|
||
assistantMessage.Reasoning,
|
||
assistantMessage.ReasoningDetails));
|
||
|
||
// #552: Die Antworten gehen im finally raus. Auf jedem Ausgangspfad
|
||
// — fertig, abgebrochen, gescheitert — bekommt jeder Aufruf seine
|
||
// Antwort, sonst bleibt eine Folge zurueck, die keine API annimmt.
|
||
var toolResults = new string?[assistantMessage.ToolCalls.Count];
|
||
string? abbruch = null;
|
||
try
|
||
{
|
||
await ExecuteToolCallsAsync(
|
||
assistantMessage.ToolCalls, toolResults, agentConfig, instanceId, tools,
|
||
logger, ct, runId, source, chain);
|
||
abbruch = fehlschlaege.Pruefe(assistantMessage.ToolCalls, toolResults);
|
||
}
|
||
finally
|
||
{
|
||
AppendToolResponses(messages, assistantMessage.ToolCalls, toolResults);
|
||
recorder.ToolResults(assistantMessage.ToolCalls, toolResults);
|
||
}
|
||
|
||
if (abbruch is not null)
|
||
throw new WiederholterFehlschlagException(abbruch);
|
||
|
||
OnStepCompleted?.Invoke(agentConfig.AgentId,
|
||
$"Chat step {loopGuard.Steps}: {assistantMessage.ToolCalls.Count} tool call(s)");
|
||
}
|
||
else
|
||
{
|
||
var text = assistantMessage.Content ?? "";
|
||
var abgeschnitten = IsTruncated(choice.FinishReason);
|
||
var urteil = LaufAbschluss.Beurteile(text, abgeschnitten, istAufgabe, leerNachgefasst, ergebnisNachgefasst);
|
||
|
||
// #577 - siehe RunCoreAsync. Die Nachfrage selbst steht nur im Kontext,
|
||
// nicht im sichtbaren Verlauf: Sie ist kein Beitrag des Menschen.
|
||
if (urteil == AbschlussUrteil.LeerNachfassen)
|
||
{
|
||
leerNachgefasst = true;
|
||
logger.LogWarning("Leere Antwort von {AgentId} nach {Steps} Schritten - einmal nachgefasst",
|
||
agentConfig.AgentId, loopGuard.Steps);
|
||
messages.Add(ChatMessage.User(LaufAbschluss.LeerNachfrage));
|
||
recorder.EngineNote(LaufAbschluss.LeerNachfrage);
|
||
continue;
|
||
}
|
||
|
||
if (urteil == AbschlussUrteil.ErgebnisNachfassen)
|
||
{
|
||
ergebnisNachgefasst = true;
|
||
logger.LogInformation("Aufgabenlauf von {AgentId} ohne Ergebniszeile - einmal nachgefasst",
|
||
agentConfig.AgentId);
|
||
messages.Add(ChatMessage.Assistant(text));
|
||
AddChatEntry(agentConfig.AgentId, "assistant", text, source);
|
||
messages.Add(ChatMessage.User(LaufAbschluss.ErgebnisNachfrage));
|
||
recorder.EngineNote(LaufAbschluss.ErgebnisNachfrage);
|
||
continue;
|
||
}
|
||
|
||
abschluss = StatusFuer(urteil, logger, agentConfig.AgentId);
|
||
finalMessage = text;
|
||
truncated = abgeschnitten;
|
||
var imVerlauf = string.IsNullOrWhiteSpace(text) ? LaufAbschluss.LeerePlatzhalter : text;
|
||
messages.Add(ChatMessage.Assistant(imVerlauf));
|
||
AddChatEntry(agentConfig.AgentId, "assistant", imVerlauf, source);
|
||
break;
|
||
}
|
||
}
|
||
|
||
sw.Stop();
|
||
|
||
if (truncated)
|
||
{
|
||
logger.LogWarning(
|
||
"Chat truncated: {AgentId} stieß an die Ausgabegrenze von {Limit} Tokens. " +
|
||
"Die Antwort ist ein Teilergebnis.",
|
||
agentConfig.AgentId, agentConfig.LoopGuard.MaxResponseTokens);
|
||
}
|
||
|
||
var result = new AgentRunResult(
|
||
agentConfig.AgentId,
|
||
truncated ? AgentRunStatus.Truncated : abschluss,
|
||
finalMessage,
|
||
loopGuard.Steps, tally.Total, sw.Elapsed)
|
||
{
|
||
PromptTokens = tally.Prompt,
|
||
CompletionTokens = tally.Completion,
|
||
CachedTokens = tally.Cached,
|
||
ReportedCostUsd = tally.ReportedCost
|
||
};
|
||
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) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
OnRunCompleted?.Invoke(agentConfig.Model, result);
|
||
return result;
|
||
}
|
||
catch (WiederholterFehlschlagException ex)
|
||
{
|
||
sw.Stop();
|
||
var stopMsg = $"[Abgebrochen: {ex.Message}]";
|
||
logger.LogWarning("Chat stopped: {AgentId}: {Message}", agentConfig.AgentId, ex.Message);
|
||
AddChatEntry(agentConfig.AgentId, "assistant", stopMsg, source);
|
||
var result = new AgentRunResult(
|
||
agentConfig.AgentId, AgentRunStatus.RepeatedFailure, stopMsg,
|
||
loopGuard.Steps, tally.Total, sw.Elapsed, ex) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
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) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
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) with { PromptTokens = tally.Prompt, CompletionTokens = tally.Completion, CachedTokens = tally.Cached, ReportedCostUsd = tally.ReportedCost };
|
||
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<CancellationTokenSource>();
|
||
_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<ChatEntry> 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);
|
||
}
|
||
|
||
/// <summary>Anzahl Agenten mit mindestens einem aktiven Chat-Lauf — für Diagnose/Heartbeat.</summary>
|
||
public int RunningChatCount
|
||
{
|
||
get { lock (_lock) return _runningChats.Count; }
|
||
}
|
||
|
||
// ─── Geordnetes Ende (#528) ───
|
||
|
||
private int _openRuns;
|
||
private int _accepting = 1;
|
||
|
||
/// <summary>
|
||
/// Wie viele Agentenläufe gerade laufen — Chats <em>und</em> angestoßene Läufe.
|
||
///
|
||
/// <para>Nicht dasselbe wie <see cref="RunningChatCount"/>: Der zählt <em>Agenten</em>
|
||
/// mit mindestens einem Chat, dieser zählt <em>Läufe</em>. Fürs Herunterfahren ist der
|
||
/// Lauf die Einheit — zwei Läufe desselben Agenten sind zwei Dinge, die zu Ende kommen
|
||
/// müssen.</para>
|
||
/// </summary>
|
||
public int OpenRunCount => Volatile.Read(ref _openRuns);
|
||
|
||
/// <summary>Nimmt die Engine noch neue Läufe an?</summary>
|
||
public bool IsAcceptingWork => Volatile.Read(ref _accepting) == 1;
|
||
|
||
/// <summary>
|
||
/// Schließt die Tür: ab jetzt kein neuer Lauf mehr.
|
||
///
|
||
/// <para>Laufende werden <b>nicht</b> abgebrochen — das ist der ganze Punkt. Wer sie
|
||
/// abbrechen will, hat dafür <c>AbortChat</c>.</para>
|
||
/// </summary>
|
||
public void StopAccepting() => Volatile.Write(ref _accepting, 0);
|
||
|
||
private AgentRunResult? RefuseWhileShuttingDown(AgentConfig agentConfig)
|
||
=> IsAcceptingWork
|
||
? null
|
||
: new AgentRunResult(
|
||
agentConfig.AgentId, AgentRunStatus.ShuttingDown,
|
||
"[Die Instanz endet gerade — es werden keine neuen Läufe mehr angenommen]",
|
||
0, 0, TimeSpan.Zero);
|
||
|
||
/// <summary>
|
||
/// Zählt einen laufenden Lauf, solange sie lebt.
|
||
///
|
||
/// <para>Als <c>using</c>-Wert statt als <c>try/finally</c> von Hand, weil beide
|
||
/// Einstiege mehrere Rückgabewege haben und ein vergessener Weg den Zähler dauerhaft
|
||
/// verschöbe — eine Instanz, die dann nie mehr enden könnte, weil ein Lauf zu Buche
|
||
/// steht, den es nicht gibt.</para>
|
||
/// </summary>
|
||
private RunScope TrackRun() => new(this);
|
||
|
||
private readonly struct RunScope : IDisposable
|
||
{
|
||
private readonly AgentEngine _engine;
|
||
|
||
public RunScope(AgentEngine engine)
|
||
{
|
||
_engine = engine;
|
||
Interlocked.Increment(ref engine._openRuns);
|
||
}
|
||
|
||
public void Dispose() => Interlocked.Decrement(ref _engine._openRuns);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
public IReadOnlyList<ChatMessage> GetChatContext(string agentId)
|
||
{
|
||
lock (_lock)
|
||
return _chatContexts.TryGetValue(agentId, out var ctx)
|
||
? ctx.ToList()
|
||
: [];
|
||
}
|
||
|
||
/// <summary>
|
||
/// Bricht ALLE Chat-Läufe des Agenten ab — laufende wie wartende.
|
||
/// </summary>
|
||
public void AbortChat(string agentId)
|
||
{
|
||
List<CancellationTokenSource> 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<IReadOnlyList<AgentConfig>> provider,
|
||
string instanceId,
|
||
Func<string, string?>? 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<List<ChatEntry>>(
|
||
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<ChatMessage>? context = null;
|
||
|
||
// Versuche zuerst als Array (direktes List<ChatMessage>)
|
||
// Dann als Wrapper-Objekt {"messages":[...]}
|
||
try
|
||
{
|
||
context = JsonSerializer.Deserialize<List<ChatMessage>>(raw, _jsonOpts);
|
||
}
|
||
catch (JsonException)
|
||
{
|
||
try
|
||
{
|
||
using var doc = JsonDocument.Parse(raw);
|
||
if (doc.RootElement.TryGetProperty("messages", out var msgs))
|
||
{
|
||
context = JsonSerializer.Deserialize<List<ChatMessage>>(
|
||
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 ───
|
||
|
||
/// <summary>
|
||
/// Darf dieser Aufruf weitergegeben werden (B8)? Gibt den Ablehnungsgrund zurück
|
||
/// oder <c>null</c>, wenn nichts dagegen spricht.
|
||
///
|
||
/// <para>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.</para>
|
||
/// </summary>
|
||
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.";
|
||
|
||
// #683: Die Grenze folgt der Tiefe des Baums; der eingestellte Wert bleibt Untergrenze.
|
||
var maxDepth = CurrentOrganisation().DelegationDepth(MaxDelegationDepth);
|
||
if (callerChain.Depth >= maxDepth)
|
||
return $"Delegationstiefe erschöpft: {callerChain} hat die Grenze von " +
|
||
$"{maxDepth} Weitergaben erreicht. Erledige den Rest selbst " +
|
||
"oder gib die Aufgabe an den Auftraggeber zurück.";
|
||
|
||
return null;
|
||
}
|
||
|
||
/// <summary>
|
||
/// #683: The address book, enforced — not merely described in the prompt. A refused
|
||
/// contact is an answer to the caller, naming the way that is open: its lead.
|
||
/// </summary>
|
||
private string? CheckAddressBook(string fromAgentId, string toAgentId)
|
||
{
|
||
var org = CurrentOrganisation();
|
||
if (org.CanAddress(fromAgentId, toAgentId))
|
||
return null;
|
||
|
||
var lead = org.EffectiveLeaderOf(fromAgentId);
|
||
return $"'{toAgentId}' is not in your address book: from your position in the organisation you " +
|
||
"cannot reach this agent directly. " +
|
||
(lead is not null
|
||
? $"Hand the matter to your lead '{lead.AgentId}', who can pass it on."
|
||
: "Create a task for @human if it needs someone outside your reach.");
|
||
}
|
||
|
||
/// <summary>
|
||
/// #715: The way up. An agent that reads foreign content writes nothing up the line, and
|
||
/// nobody writes past its lead — the refusal names the way that is open.
|
||
/// </summary>
|
||
private string? CheckReportingLine(AgentConfig? fromConfig, string fromAgentId, string toAgentId)
|
||
=> ReportingLine.CheckUpward(CurrentOrganisation(), TrustOf(fromConfig), fromAgentId, toAgentId, UpwardKind.Message);
|
||
|
||
private AgentTrust TrustOf(AgentConfig? agent)
|
||
=> agent is null ? AgentTrust.Standard : ReportingLine.TrustOf(agent, _toolRegistry.Get);
|
||
|
||
public IReadOnlyList<AgentInfo> ListAgents(string callerAgentId)
|
||
{
|
||
var configs = _agentConfigProvider?.Invoke() ?? [];
|
||
// #683: Only the agent's address book — the cut of the organisation.
|
||
var reachable = CurrentOrganisation().AddressBook(callerAgentId);
|
||
return configs
|
||
.Where(a => a.AgentId != callerAgentId && reachable.Contains(a.AgentId))
|
||
.Select(a => new AgentInfo(a.AgentId, a.DisplayName, a.Description, a.Model, IsRunning(a.AgentId)))
|
||
.ToList();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Die Instanz des Aufrufers, sonst die der Engine (B13).
|
||
///
|
||
/// <c>_instanceId</c> wird nur in <c>SetAgentConfigProvider</c> 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.
|
||
/// </summary>
|
||
private string ResolveInstanceId(string? fromCaller)
|
||
=> string.IsNullOrWhiteSpace(fromCaller) ? _instanceId : fromCaller;
|
||
|
||
public async Task<AgentMessageResult> 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 (CheckAddressBook(fromAgentId, toAgentId) is { } outOfReach)
|
||
return new AgentMessageResult(false, null, outOfReach);
|
||
|
||
if (CheckDelegation(chain, fromAgentId, toAgentId) is { } refused)
|
||
return new AgentMessageResult(false, null, refused);
|
||
|
||
var fromConfig = configs.FirstOrDefault(a => a.AgentId == fromAgentId);
|
||
if (CheckReportingLine(fromConfig, fromAgentId, toAgentId) is { } notUp)
|
||
return new AgentMessageResult(false, null, notUp);
|
||
|
||
var callerChain = chain ?? DelegationChain.Root(fromAgentId);
|
||
var fromName = fromConfig?.DisplayName ?? fromAgentId;
|
||
|
||
// #715: What an agent of the foreign tier writes arrives as data, not as an order.
|
||
var wrappedMessage = $"[Nachricht von Agent \"{fromName}\" ({fromAgentId})]\n\n"
|
||
+ ReportingLine.Frame(TrustOf(fromConfig), fromAgentId, 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<AgentSpawnResult> 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 (CheckAddressBook(fromAgentId, targetAgentId) is { } outOfReach)
|
||
return new AgentSpawnResult(false, targetAgentId, null, outOfReach);
|
||
|
||
if (CheckDelegation(chain, fromAgentId, targetAgentId) is { } refused)
|
||
return new AgentSpawnResult(false, targetAgentId, null, refused);
|
||
|
||
var fromConfig = configs.FirstOrDefault(a => a.AgentId == fromAgentId);
|
||
if (CheckReportingLine(fromConfig, fromAgentId, targetAgentId) is { } notUp)
|
||
return new AgentSpawnResult(false, targetAgentId, null, notUp);
|
||
|
||
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 fromName = fromConfig?.DisplayName ?? fromAgentId;
|
||
|
||
var wrappedMessage = $"[Spawn-Auftrag von Agent \"{fromName}\" ({fromAgentId})]\n\n"
|
||
+ ReportingLine.Frame(TrustOf(fromConfig), fromAgentId, 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<ChatEntry>();
|
||
_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<ChatEntry>? history;
|
||
List<ChatMessage>? 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);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Ob das Modell abgebrochen wurde, statt fertig zu werden (#548).
|
||
///
|
||
/// <para>Der Grund steht in <c>finish_reason</c>. Anbieter schreiben ihn verschieden:
|
||
/// <c>length</c> ist die verbreitete Form, <c>max_tokens</c> kommt bei einigen
|
||
/// vor. Beides meint dasselbe — die Ausgabegrenze war erreicht, bevor der Gedanke
|
||
/// zu Ende war.</para>
|
||
///
|
||
/// <para>Fehlt das Feld ganz, wird nicht von Abbruch ausgegangen. Ein Anbieter, der
|
||
/// nichts meldet, soll keine Warnung auslösen, die es nicht gibt.</para>
|
||
/// </summary>
|
||
internal static bool IsTruncated(string? finishReason) =>
|
||
finishReason is not null &&
|
||
(finishReason.Equals("length", StringComparison.OrdinalIgnoreCase) ||
|
||
finishReason.Equals("max_tokens", StringComparison.OrdinalIgnoreCase));
|
||
|
||
/// <summary>
|
||
/// Baut die Anfrage für einen Schritt. Eine Stelle für beide Schleifen — als das
|
||
/// hier noch zweimal dastand, war der Unterschied zwischen ihnen niemandes Absicht,
|
||
/// sondern das Ergebnis davon, dass eine Änderung nur an einer Stelle ankam.
|
||
/// </summary>
|
||
private static ChatRequest BuildRequest(
|
||
AgentConfig agentConfig,
|
||
List<ChatMessage> messages,
|
||
List<ToolDefinition> toolDefinitions) => new()
|
||
{
|
||
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,
|
||
|
||
// #550: Nur an Anbieter weiterreichen, die alles beherrschen, was mitgeht.
|
||
Provider = agentConfig.RequireProviderParameters
|
||
? new ProviderPreferences { RequireParameters = true }
|
||
: null,
|
||
|
||
// #550: Den abgerechneten Betrag mitliefern lassen, statt ihn zu schätzen.
|
||
Usage = new UsageRequest { Include = true },
|
||
|
||
Reasoning = agentConfig.BuildReasoningRequest()
|
||
};
|
||
|
||
/// <summary>
|
||
/// Führt die Werkzeugaufrufe eines Zuges aus und liefert die Ergebnisse
|
||
/// <b>in der Reihenfolge der Aufrufe</b> zurück — unabhängig davon, welcher zuerst
|
||
/// fertig wurde. Daran hängt zweierlei: die Zuordnung über die tool_call_id und der
|
||
/// Prompt-Cache, dessen Präfix bei wechselnder Reihenfolge jedes Mal neu wäre.
|
||
///
|
||
/// <para>Als nebenläufig gekennzeichnete Aufrufe laufen gemeinsam, alle übrigen
|
||
/// nacheinander. Gemischt heißt: erst die gemeinsame Gruppe, dann der Rest — ein
|
||
/// schreibendes Werkzeug sieht damit nie einen halbfertigen Zustand eines anderen.</para>
|
||
/// </summary>
|
||
/// <summary>
|
||
/// Ersatzantwort für einen Aufruf, der nie zu Ende kam (#552). Ohne sie bliebe der
|
||
/// Aufruf unbeantwortet, und eine solche Folge nimmt die Schnittstelle nicht an.
|
||
/// </summary>
|
||
internal const string AbgebrochenerAufruf =
|
||
"""{"error":"Aufruf abgebrochen, bevor ein Ergebnis vorlag."}""";
|
||
|
||
private async Task ExecuteToolCallsAsync(
|
||
IReadOnlyList<ToolCall> toolCalls,
|
||
string?[] results,
|
||
AgentConfig agentConfig,
|
||
string instanceId,
|
||
IReadOnlyList<IAgentTool> availableTools,
|
||
ILogger logger,
|
||
CancellationToken ct,
|
||
string runId,
|
||
string? source,
|
||
DelegationChain chain)
|
||
{
|
||
var limit = Math.Max(1, agentConfig.MaxParallelToolCalls);
|
||
|
||
// Nebenläufig ist ein Aufruf nur, wenn das Werkzeug sich dafür freigegeben hat.
|
||
// Ein unbekannter Name gilt als nicht freigegeben — ExecuteToolCallAsync macht
|
||
// daraus ohnehin einen Fehler, und der gehört in die serielle Reihe.
|
||
var parallel = new List<int>();
|
||
var serial = new List<int>();
|
||
|
||
for (var i = 0; i < toolCalls.Count; i++)
|
||
{
|
||
var tool = availableTools.FirstOrDefault(t => t.Name == toolCalls[i].Function.Name);
|
||
(tool?.IsParallelSafe == true && limit > 1 ? parallel : serial).Add(i);
|
||
}
|
||
|
||
if (parallel.Count == 1)
|
||
{
|
||
// Eine einzelne Aufgabe nebenläufig zu starten, kostet nur Umstand.
|
||
serial.Insert(0, parallel[0]);
|
||
parallel.Clear();
|
||
}
|
||
|
||
if (parallel.Count > 0)
|
||
{
|
||
logger.LogDebug("{Count} Werkzeuge laufen gemeinsam (Grenze {Limit})", parallel.Count, limit);
|
||
|
||
using var slots = new SemaphoreSlim(limit, limit);
|
||
|
||
// WhenAll statt einer Schleife mit Abbruch: Jede Aufgabe muss abgewartet
|
||
// werden, auch wenn eine andere bereits geworfen hat. Sonst läuft ein
|
||
// Werkzeug im Hintergrund weiter, während der Lauf sich schon beendet hat,
|
||
// und seine Ausnahme taucht später aus dem Nichts auf.
|
||
await Task.WhenAll(parallel.Select(async index =>
|
||
{
|
||
await slots.WaitAsync(ct);
|
||
try
|
||
{
|
||
results[index] = await ExecuteToolCallAsync(
|
||
toolCalls[index], agentConfig, instanceId, availableTools,
|
||
logger, ct, runId, source, chain);
|
||
}
|
||
finally
|
||
{
|
||
slots.Release();
|
||
}
|
||
}));
|
||
}
|
||
|
||
foreach (var index in serial)
|
||
{
|
||
results[index] = await ExecuteToolCallAsync(
|
||
toolCalls[index], agentConfig, instanceId, availableTools,
|
||
logger, ct, runId, source, chain);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Hängt für <b>jeden</b> Aufruf eine Antwort an — auch für den, der nie zu Ende kam
|
||
/// (#552).
|
||
///
|
||
/// <para>Wird aus einem <c>finally</c> gerufen. Der Kontext ist bei einem Chat die
|
||
/// geteilte, an Ort und Stelle veränderte Liste; bliebe hier eine Lücke, stünde
|
||
/// darin dauerhaft eine Folge, die die Schnittstelle mit HTTP 400 ablehnt — und
|
||
/// über <c>PersistChatState</c> wanderte sie zuverlässig auf die Platte. Der Agent
|
||
/// wäre danach auch nach einem Neustart nicht mehr ansprechbar.</para>
|
||
/// </summary>
|
||
private static void AppendToolResponses(
|
||
List<ChatMessage> messages,
|
||
IReadOnlyList<ToolCall> toolCalls,
|
||
string?[] results)
|
||
{
|
||
for (var i = 0; i < toolCalls.Count; i++)
|
||
messages.Add(ChatMessage.ToolResponse(toolCalls[i].Id, results[i] ?? AbgebrochenerAufruf));
|
||
}
|
||
|
||
private async Task<string> ExecuteToolCallAsync(
|
||
ToolCall toolCall,
|
||
AgentConfig agentConfig,
|
||
string instanceId,
|
||
IReadOnlyList<IAgentTool> 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, toolCall.Id);
|
||
|
||
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."));
|
||
}
|
||
|
||
// #581: Unvollstaendiges Argument-JSON gar nicht erst ans Tool reichen. Vorher
|
||
// bekam das Modell den Parsertext ("Expected depth to be zero...") und
|
||
// wiederholte denselben Aufruf bis zur Schrittgrenze. Beobachtet bei sehr
|
||
// grossen Argumenten: Das JSON brach genau vor dem grossen Wert ab.
|
||
if (!string.IsNullOrWhiteSpace(arguments) && !IstVollstaendigesJson(arguments))
|
||
{
|
||
var hinweis = UnvollstaendigeArgumente(toolName, arguments, agentConfig.LoopGuard.MaxResponseTokens);
|
||
logger.LogWarning("Unvollstaendige Argumente fuer {Tool} ({Laenge} Zeichen)", toolName, arguments.Length);
|
||
await Audit(AuditStatus.Error, hinweis);
|
||
return JsonSerializer.Serialize(new { error = hinweis });
|
||
}
|
||
|
||
var input = string.IsNullOrWhiteSpace(arguments)
|
||
? default
|
||
: JsonDocument.Parse(arguments).RootElement;
|
||
|
||
var context = BuildToolContext(agentConfig, instanceId, toolName, ct, chain, runId);
|
||
|
||
// #579: Was das Werkzeug ohnehin verbieten wuerde, kommt gar nicht erst in die
|
||
// Freigabeliste. Vorher lag dort z.B. ein Loeschvorschlag eines Agenten, der
|
||
// nur lesen darf - der Mensch entschied ueber etwas, das nie passieren konnte.
|
||
if (tool.Preflight(input, context) is { } verweigert)
|
||
{
|
||
logger.LogInformation("Vorpruefung von {Tool} lehnt ab: {Grund}", toolName, verweigert);
|
||
await Audit(AuditStatus.Denied, verweigert);
|
||
return WerkzeugFehler.Json(verweigert);
|
||
}
|
||
|
||
// Staging-Durchsetzung (A2): irreversible Aktionen werden vorgeschlagen statt
|
||
// ausgeführt. Eine Prompt-Injection kann so nur einen Vorschlag erzeugen.
|
||
if (_stagingGate is not null)
|
||
{
|
||
// #641: Die Rechte stehen in der Werkzeugkonfiguration des Agenten.
|
||
var operation = tool.StagingAktion(input) ?? Staging.StagingGate.ExtractAction(arguments);
|
||
var intercept = await _stagingGate.InterceptAsync(
|
||
agentConfig.AgentId, instanceId, runId, toolName, arguments, ct,
|
||
operation,
|
||
Staging.Werkzeugrechte.Aus(context.ToolConfig),
|
||
operation is null ? null : tool.Vorgabe(operation, context.ToolConfig));
|
||
|
||
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
|
||
}
|
||
|
||
context = context with { FreigabeGeprueft = true };
|
||
}
|
||
|
||
logger.LogDebug("Executing tool {Tool} for agent {AgentId}", toolName, agentConfig.AgentId);
|
||
|
||
// #554: Vor der Wirkung vermerken, nicht danach. Stirbt die Anwendung
|
||
// mitten im Aufruf, ist das hier die einzige Spur, die davon übrig bleibt —
|
||
// ein Eintrag nach der Ausführung wäre genau dann nie geschrieben worden.
|
||
// Steht bewusst nach Rechteprüfung und Staging: Was gar nicht erst läuft,
|
||
// hinterlässt auch nichts aufzuräumen.
|
||
await Audit(AuditStatus.Started, "");
|
||
|
||
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, "");
|
||
|
||
// #193: Was ein Werkzeug liefert, ist Datenmaterial - gerahmt, damit das Modell
|
||
// es von Anweisungen unterscheiden kann (Regel steht im Systemprompt).
|
||
return UntrustedContent.Rahmen(toolName, 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.
|
||
//
|
||
// Sehr wohl aber ins Protokoll (#554). Früher stand hier "auch kein Audit",
|
||
// und das war richtig, solange es keinen Eintrag vor dem Aufruf gab. Jetzt
|
||
// gibt es ihn: Bliebe er ohne Abschluss, sähe jeder gewöhnliche Abbruch beim
|
||
// nächsten Start aus wie ein Absturz. Die Wirkung bleibt in beiden Fällen
|
||
// ungewiss — nur ist sie hier wenigstens erklärt.
|
||
//
|
||
// CancellationToken.None, weil der reguläre Token gerade abgebrochen wurde
|
||
// und das Schreiben sonst selbst mit abbräche.
|
||
await RecordAuditAsync(runId, agentConfig, source, toolName, arguments,
|
||
AuditStatus.Cancelled, "Unterbrochen, während der Aufruf lief.",
|
||
sw.ElapsedMilliseconds, toolCall.Id);
|
||
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}" });
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Stellt einer Nachricht die aktuelle Zeit voran (#589, Teil 2). Aufgaben bringen sie
|
||
/// schon aus dem Dispatcher mit - dort samt Termin -, alle anderen Quellen bekommen sie
|
||
/// hier. Im Nachrichtentext, nicht im Systemprompt, damit der Prompt-Cache greift.
|
||
/// </summary>
|
||
private string MitZeit(string nachricht, string? source)
|
||
=> source == ChatSource.Task
|
||
? nachricht
|
||
: ZeitAngabe.Zeile(_clock.GetUtcNow()) + "\n\n" + nachricht;
|
||
|
||
/// <summary>Der Endstatus zu einem abschliessenden Urteil (#577), mit Warnung, wenn es kein Erfolg ist.</summary>
|
||
private static AgentRunStatus StatusFuer(AbschlussUrteil urteil, ILogger logger, string agentId)
|
||
{
|
||
switch (urteil)
|
||
{
|
||
case AbschlussUrteil.Leer:
|
||
logger.LogWarning("Lauf von {AgentId} endet ohne Antwort - auch nach dem Nachfassen leer", agentId);
|
||
return AgentRunStatus.EmptyResponse;
|
||
case AbschlussUrteil.Unvollstaendig:
|
||
logger.LogWarning("Aufgabenlauf von {AgentId} endet ohne Ergebniszeile - gilt als unvollstaendig", agentId);
|
||
return AgentRunStatus.Incomplete;
|
||
default:
|
||
return AgentRunStatus.Completed;
|
||
}
|
||
}
|
||
|
||
/// <summary>Ob die Argumente eines Werkzeugaufrufs vollstaendiges JSON sind (#581).</summary>
|
||
internal static bool IstVollstaendigesJson(string arguments)
|
||
{
|
||
try
|
||
{
|
||
using var _ = JsonDocument.Parse(arguments);
|
||
return true;
|
||
}
|
||
catch (JsonException)
|
||
{
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Die Rueckmeldung ans Modell, wenn Argumente abgeschnitten ankamen (#581): was los
|
||
/// ist und was stattdessen zu tun ist - nicht der Text eines JSON-Parsers.
|
||
/// </summary>
|
||
internal static string UnvollstaendigeArgumente(string toolName, string arguments, int maxResponseTokens)
|
||
{
|
||
var ende = arguments.Length <= 40 ? arguments : "..." + arguments[^40..];
|
||
return $"Die Argumente fuer '{toolName}' sind kein vollstaendiges JSON ({arguments.Length} Zeichen, "
|
||
+ $"enden mit: {ende}). Meist war ein Argument zu gross fuer eine einzelne Antwort "
|
||
+ $"(Ausgabegrenze {maxResponseTokens} Tokens) und wurde abgeschnitten. Denselben Aufruf "
|
||
+ "nicht wiederholen - den Inhalt auf mehrere kleinere Aufrufe verteilen.";
|
||
}
|
||
|
||
private AgentToolContext BuildToolContext(
|
||
AgentConfig agentConfig, string instanceId, string toolName, CancellationToken ct,
|
||
DelegationChain? chain = null, string? runId = null)
|
||
{
|
||
var toolConfig = agentConfig.Tools.TryGetValue(toolName, out var cfg)
|
||
? cfg.AsReadOnly()
|
||
: new Dictionary<string, object?>().AsReadOnly();
|
||
|
||
var toolLogger = _loggerFactory.CreateLogger($"ClawdDotNet.Tools.{toolName}.Execution");
|
||
var baseChain = chain ?? DelegationChain.Root(agentConfig.AgentId);
|
||
|
||
return new AgentToolContext(
|
||
agentConfig.AgentId,
|
||
instanceId,
|
||
toolConfig,
|
||
_stateStore,
|
||
toolLogger,
|
||
ct,
|
||
agentConfig.WorkspacePath,
|
||
agentConfig.SharedWorkspacePath,
|
||
this,
|
||
_memoryRepository,
|
||
_taskRepository,
|
||
// #697: The chain carries the caller's run, so a delegated run can name where it came from.
|
||
runId is null ? baseChain : baseChain.WithRun(runId))
|
||
{
|
||
Organisation = CurrentOrganisation(),
|
||
Trust = ReportingLine.TrustOf(agentConfig, _toolRegistry.Get),
|
||
// #697: Where something an agent hands on came from — known to the engine, not claimed by the agent.
|
||
RunId = runId,
|
||
RunTaskId = runId is null ? null : Activity.Find(runId)?.TaskId
|
||
};
|
||
}
|
||
|
||
/// <summary>
|
||
/// 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).
|
||
/// </summary>
|
||
public async Task<string> 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;
|
||
|
||
// #641: Ein Mensch hat genau diesen Aufruf freigegeben.
|
||
var context = BuildToolContext(config, _instanceId, tool, ct, runId: runId) with { FreigabeGeprueft = true };
|
||
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 ───
|
||
|
||
/// <summary>
|
||
/// Prüft das Tagesbudget. Gibt ein Ergebnis zurück, wenn der Lauf nicht stattfinden
|
||
/// darf — sonst null und, bei gesetzter Grenze, die Reservierung des Laufs (#716).
|
||
/// </summary>
|
||
private async Task<(AgentRunResult? Denied, IDisposable? Reservation)> CheckBudgetAsync(
|
||
AgentConfig agentConfig, string? source, string? taskId, CancellationToken ct)
|
||
{
|
||
if (_budgetGuard is null)
|
||
return (null, null);
|
||
|
||
var (status, reservation) = await _budgetGuard.ReserveAsync(agentConfig, InstanceBudget, ct);
|
||
if (status.IsAllowed)
|
||
return (null, reservation);
|
||
|
||
// #608: Ist ein Budget erschoepft, lehnt es jeden weiteren Lauf des Tages ab. Einmal je
|
||
// Agent, Grenze und Tag als Warnung genuegt - danach steht jede Ablehnung als Beleg da.
|
||
var logger = _loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Budget");
|
||
var heute = DateOnly.FromDateTime(_clock.GetLocalNow().DateTime);
|
||
var schluessel = agentConfig.AgentId + "|" + status.Limit;
|
||
if (_budgetGewarnt.TryGetValue(schluessel, out var gewarnt) && gewarnt == heute)
|
||
{
|
||
logger.LogInformation("Lauf abgelehnt für {AgentId} (Aufgabe {TaskId}): {Reason}",
|
||
agentConfig.AgentId, taskId ?? "-", status.Reason);
|
||
}
|
||
else
|
||
{
|
||
_budgetGewarnt[schluessel] = heute;
|
||
logger.LogWarning("Lauf abgelehnt für {AgentId} (Aufgabe {TaskId}): {Reason} Weitere Ablehnungen heute stehen als Beleg im Verlauf.",
|
||
agentConfig.AgentId, taskId ?? "-", status.Reason);
|
||
}
|
||
|
||
var result = new AgentRunResult(
|
||
agentConfig.AgentId, AgentRunStatus.BudgetExceeded,
|
||
$"[Budget erschöpft] {status.Reason}", 0, 0, TimeSpan.Zero);
|
||
|
||
// #608: Ohne Beleg blieb die Ablehnung unsichtbar - Aufgabenkarte und Verlauf zeigten
|
||
// weiter den letzten erfolgreichen Lauf, die Luecke fiel nur im Log auf.
|
||
await RecordReceiptAsync(Guid.NewGuid().ToString("N"), agentConfig, result, source, taskId);
|
||
|
||
OnRunCompleted?.Invoke(agentConfig.Model, result);
|
||
return (result, null);
|
||
}
|
||
|
||
/// <summary>Welche Budgetgrenze je Agent zuletzt an welchem Tag als Warnung gemeldet wurde.</summary>
|
||
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, DateOnly> _budgetGewarnt = new();
|
||
|
||
/// <summary>
|
||
/// Schreibt den Verbrauch eines Laufs fort. Fehler hierbei dürfen den Lauf nicht
|
||
/// nachträglich scheitern lassen — die eigentliche Arbeit ist bereits getan.
|
||
/// </summary>
|
||
private async Task RecordUsageAsync(AgentConfig agentConfig, AgentRunResult result)
|
||
{
|
||
if (_usageRepository is null)
|
||
return;
|
||
|
||
try
|
||
{
|
||
var estimate = _pricing?.Estimate(agentConfig.Model, result.PromptTokens, result.CompletionTokens);
|
||
|
||
// #550: Der gemeldete Betrag schlägt die Eigenrechnung. Er weiß von
|
||
// Cache-Aufschlägen und anbieterabhängigen Preisen, die eine Preisliste
|
||
// nicht abbildet. Fehlt er, bleibt die Schätzung als Rückfallebene —
|
||
// deshalb wird sie weiterhin unverändert berechnet.
|
||
var reported = result.ReportedCostUsd;
|
||
|
||
await _usageRepository.RecordAsync(new RunUsage
|
||
{
|
||
AgentId = agentConfig.AgentId,
|
||
Model = agentConfig.Model,
|
||
PromptTokens = result.PromptTokens,
|
||
CompletionTokens = result.CompletionTokens,
|
||
CachedTokens = result.CachedTokens,
|
||
CostUsd = reported ?? estimate?.Usd ?? 0m,
|
||
CostIsKnown = reported is not null || (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) ───
|
||
|
||
/// <summary>
|
||
/// Schreibt einen Tool-Aufruf ins Audit-Log. Best effort — ein Fehler hierbei darf den
|
||
/// Lauf nicht scheitern lassen; die eigentliche Arbeit ist bereits getan.
|
||
/// </summary>
|
||
private async Task RecordAuditAsync(
|
||
string runId, AgentConfig agentConfig, string? source, string tool,
|
||
string arguments, AuditStatus status, string summary, long durationMs,
|
||
string callId = "")
|
||
{
|
||
if (_auditRepository is null)
|
||
return;
|
||
|
||
var entry = new AuditEntry
|
||
{
|
||
RunId = runId,
|
||
CallId = callId,
|
||
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
|
||
};
|
||
|
||
// Der "angefangen"-Eintrag ist der einzige, dessen Fehlschlag zählt (#554):
|
||
// Auf ihn stützt sich die Wiederaufnahme nach einem Absturz. Kommt er nicht
|
||
// durch, liefe der Aufruf ohne jede Spur — dann lieber gar nicht erst laufen
|
||
// lassen. Alle übrigen Einträge bleiben "best effort": Ein Protokollfehler soll
|
||
// einen bereits erfolgreichen Aufruf nicht nachträglich zum Scheitern bringen.
|
||
if (status == AuditStatus.Started)
|
||
{
|
||
await _auditRepository.AppendAsync(entry, CancellationToken.None);
|
||
return;
|
||
}
|
||
|
||
try
|
||
{
|
||
await _auditRepository.AppendAsync(entry, CancellationToken.None);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Audit")
|
||
.LogWarning(ex, "Audit-Eintrag konnte nicht geschrieben werden ({Tool})", tool);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Hält den Abschluss-Beleg eines Laufs fest (Receipt). Best effort, wie beim Audit.
|
||
/// </summary>
|
||
// ─── Prompt history (#660) ───
|
||
|
||
/// <summary>Retention that applies to this agent: its own setting or the instance default.</summary>
|
||
public int TranscriptRetentionFor(AgentConfig agentConfig)
|
||
=> Transcripts.PromptHistorySettings.Resolve(agentConfig.PromptHistory?.RetentionDays, DefaultTranscriptRetention());
|
||
|
||
private Transcripts.RunTranscriptRecorder CreateRecorder(AgentConfig agentConfig)
|
||
=> TranscriptStore is not null && TranscriptRetentionFor(agentConfig) != Transcripts.PromptHistorySettings.Never
|
||
? new Transcripts.RunTranscriptRecorder(enabled: true, _clock)
|
||
: Transcripts.RunTranscriptRecorder.Disabled;
|
||
|
||
/// <summary>Best effort: a transcript that cannot be written must not fail the run.</summary>
|
||
private async Task SaveTranscriptAsync(
|
||
Transcripts.RunTranscriptRecorder recorder, string runId, AgentConfig agentConfig,
|
||
AgentRunResult result, string source, string? taskId)
|
||
{
|
||
if (!recorder.Enabled || TranscriptStore is null)
|
||
return;
|
||
|
||
try
|
||
{
|
||
var transcript = recorder.Build(runId, agentConfig.AgentId, agentConfig.Model, source, taskId, result.Status.ToString());
|
||
await TranscriptStore.SaveAsync(transcript, recorder.SystemPrompt, CancellationToken.None);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_loggerFactory.CreateLogger("ClawdDotNet.Core.Engine.Transcripts")
|
||
.LogWarning(ex, "Transcript of run {RunId} ({AgentId}) could not be stored", runId, agentConfig.AgentId);
|
||
}
|
||
}
|
||
|
||
private async Task RecordReceiptAsync(
|
||
string runId, AgentConfig agentConfig, AgentRunResult result, string? source, string? taskId,
|
||
string? callerRunId = null)
|
||
{
|
||
if (_auditRepository is null)
|
||
return;
|
||
|
||
try
|
||
{
|
||
var estimate = _pricing?.Estimate(agentConfig.Model, result.PromptTokens, result.CompletionTokens);
|
||
|
||
// #599: derselbe Betrag wie in RunUsage. Vorher stand hier nur die Schaetzung
|
||
// nach Listenpreis ohne Cache-Rabatt - im Testbetrieb ~33 % ueber dem, was
|
||
// OpenRouter abrechnete, waehrend RunUsage auf den Cent stimmte.
|
||
var reported = result.ReportedCostUsd;
|
||
|
||
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 = reported ?? estimate?.Usd ?? 0m,
|
||
CostIsKnown = reported is not null || (estimate?.IsKnown ?? false),
|
||
DurationMs = (long)result.Duration.TotalMilliseconds,
|
||
ResultRef = Cap(result.FinalMessage ?? "", 500),
|
||
OccurredAt = DateTime.Now,
|
||
CallerRunId = callerRunId
|
||
}, 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] + "…";
|
||
|
||
/// <summary>Sammelt die Token-Zahlen über alle Schritte eines Runs.</summary>
|
||
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; }
|
||
|
||
/// <summary>
|
||
/// Summe der vom Gateway gemeldeten Beträge (#550). Null heißt: keiner der
|
||
/// Schritte hat einen geliefert — dann bleibt es bei der Eigenrechnung.
|
||
///
|
||
/// <para>Bewusst nur dann gesetzt, wenn wirklich etwas kam. Ein stillschweigend
|
||
/// als 0 gezählter Schritt wäre schlimmer als gar keine Angabe: Er sähe aus wie
|
||
/// ein kostenloser Lauf.</para>
|
||
/// </summary>
|
||
public decimal? ReportedCost { 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;
|
||
|
||
if (usage.Cost is { } cost)
|
||
ReportedCost = (ReportedCost ?? 0m) + cost;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
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<ToolDefinition> BuildToolDefinitions(IReadOnlyList<IAgentTool> tools)
|
||
{
|
||
return tools.Select(t => new ToolDefinition
|
||
{
|
||
Function = new FunctionDefinition
|
||
{
|
||
Name = t.Name,
|
||
Description = t.Description,
|
||
Parameters = t.InputSchema
|
||
}
|
||
}).ToList();
|
||
}
|
||
}
|