using System.Globalization;
using ClawdDotNet.Core.Storage;
using Microsoft.Data.Sqlite;
namespace ClawdDotNet.Core.Audit;
///
/// Das Audit-Log in der Instanz-Datenbank. Bewusst nur Einfügen und Lesen — es gibt keine
/// Update-/Delete-Methoden, weil die Unveränderlichkeit die eigentliche Zusage ist.
///
public sealed class SqliteAuditRepository : IAuditRepository
{
private readonly SqliteStorage _storage;
public SqliteAuditRepository(SqliteStorage storage) => _storage = storage;
public Task AppendAsync(AuditEntry entry, CancellationToken ct)
=> _storage.WriteAsync(async conn =>
{
using var cmd = conn.CreateCommand();
cmd.CommandText = """
INSERT INTO AuditLog
(RunId, CallId, AgentId, Model, Source, Tool, Arguments, Status, Summary, DurationMs, OccurredAt)
VALUES
(@runId, @callId, @agentId, @model, @source, @tool, @arguments, @status, @summary, @durationMs, @occurredAt)
""";
cmd.Parameters.AddWithValue("@runId", entry.RunId);
cmd.Parameters.AddWithValue("@callId", entry.CallId);
cmd.Parameters.AddWithValue("@agentId", entry.AgentId);
cmd.Parameters.AddWithValue("@model", entry.Model);
cmd.Parameters.AddWithValue("@source", AuditSource.Normalize(entry.Source));
cmd.Parameters.AddWithValue("@tool", entry.Tool);
cmd.Parameters.AddWithValue("@arguments", entry.Arguments);
cmd.Parameters.AddWithValue("@status", entry.Status.ToString());
cmd.Parameters.AddWithValue("@summary", entry.Summary);
cmd.Parameters.AddWithValue("@durationMs", entry.DurationMs);
cmd.Parameters.AddWithValue("@occurredAt", Format(entry.OccurredAt));
await cmd.ExecuteNonQueryAsync(ct);
}, ct);
public Task RecordReceiptAsync(RunReceipt receipt, CancellationToken ct)
=> _storage.WriteAsync(async conn =>
{
using var cmd = conn.CreateCommand();
cmd.CommandText = """
INSERT INTO RunReceipts
(RunId, AgentId, Model, Source, TaskId, Status, StepCount,
PromptTokens, CompletionTokens, CachedTokens, CostUsd, CostIsKnown,
DurationMs, ResultRef, OccurredAt, CallerRunId, BudgetPot)
VALUES
(@runId, @agentId, @model, @source, @taskId, @status, @stepCount,
@prompt, @completion, @cached, @cost, @costKnown,
@durationMs, @resultRef, @occurredAt, @callerRunId, @budgetPot)
""";
cmd.Parameters.AddWithValue("@runId", receipt.RunId);
cmd.Parameters.AddWithValue("@agentId", receipt.AgentId);
cmd.Parameters.AddWithValue("@model", receipt.Model);
cmd.Parameters.AddWithValue("@source", AuditSource.Normalize(receipt.Source));
cmd.Parameters.AddWithValue("@taskId", (object?)receipt.TaskId ?? DBNull.Value);
cmd.Parameters.AddWithValue("@status", receipt.Status);
cmd.Parameters.AddWithValue("@stepCount", receipt.StepCount);
cmd.Parameters.AddWithValue("@prompt", receipt.PromptTokens);
cmd.Parameters.AddWithValue("@completion", receipt.CompletionTokens);
cmd.Parameters.AddWithValue("@cached", receipt.CachedTokens);
cmd.Parameters.AddWithValue("@cost", receipt.CostUsd.ToString(CultureInfo.InvariantCulture));
cmd.Parameters.AddWithValue("@costKnown", receipt.CostIsKnown ? 1 : 0);
cmd.Parameters.AddWithValue("@durationMs", receipt.DurationMs);
cmd.Parameters.AddWithValue("@resultRef", receipt.ResultRef);
cmd.Parameters.AddWithValue("@occurredAt", Format(receipt.OccurredAt));
cmd.Parameters.AddWithValue("@callerRunId", (object?)receipt.CallerRunId ?? DBNull.Value);
cmd.Parameters.AddWithValue("@budgetPot", (object?)receipt.BudgetPot ?? DBNull.Value);
await cmd.ExecuteNonQueryAsync(ct);
}, ct);
public async Task> ListRecentAsync(int limit, CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT " + AuditColumns + " FROM AuditLog ORDER BY Id DESC LIMIT @limit";
cmd.Parameters.AddWithValue("@limit", Math.Clamp(limit, 1, 1000));
return await ReadEntriesAsync(cmd, ct);
}
public async Task> ListForRunAsync(string runId, CancellationToken ct)
{
await using (var conn = await _storage.OpenConnectionAsync(ct))
{
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT " + AuditColumns + " FROM AuditLog WHERE RunId = @runId ORDER BY Id";
cmd.Parameters.AddWithValue("@runId", runId);
var hot = await ReadEntriesAsync(cmd, ct);
if (hot.Count > 0)
return hot;
}
// #700: an older run moved to the archive — it is still evidence and still found.
foreach (var archive in ArchiveConnections())
await using (archive)
{
using var cmd = archive.CreateCommand();
cmd.CommandText = "SELECT " + AuditColumns + " FROM AuditLog WHERE RunId = @runId ORDER BY Id";
cmd.Parameters.AddWithValue("@runId", runId);
var found = await ReadEntriesAsync(cmd, ct);
if (found.Count > 0)
return found;
}
return [];
}
public async Task GetReceiptForRunAsync(string runId, CancellationToken ct)
{
await using (var conn = await _storage.OpenConnectionAsync(ct))
{
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT " + ReceiptColumns + " FROM RunReceipts WHERE RunId = @runId ORDER BY Id DESC LIMIT 1";
cmd.Parameters.AddWithValue("@runId", runId);
await using var reader = await cmd.ExecuteReaderAsync(ct);
if (await reader.ReadAsync(ct))
return ReadReceipt(reader);
}
foreach (var archive in ArchiveConnections())
await using (archive)
{
using var cmd = archive.CreateCommand();
cmd.CommandText = "SELECT " + ReceiptColumns + " FROM RunReceipts WHERE RunId = @runId ORDER BY Id DESC LIMIT 1";
cmd.Parameters.AddWithValue("@runId", runId);
await using var reader = await cmd.ExecuteReaderAsync(ct);
if (await reader.ReadAsync(ct))
return ReadReceipt(reader);
}
return null;
}
///
/// #700: The month archives, newest first, opened read-only one at a time. A lookup by run
/// is rare and asks for one run — opening a few small files for it is cheap.
///
private IEnumerable ArchiveConnections()
{
foreach (var file in AuditArchive.FilesIn(AuditArchive.ArchiveDirectoryOf(_storage)))
{
var connection = new SqliteConnection(new SqliteConnectionStringBuilder
{
DataSource = Path.Combine(AuditArchive.ArchiveDirectoryOf(_storage), file.FileName),
Mode = SqliteOpenMode.ReadOnly,
Pooling = false
}.ToString());
connection.Open();
yield return connection;
}
}
public async Task> ListReceiptsAsync(DateTime fromUtc, DateTime toUtc, int limit, CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
// OccurredAt is ISO-8601 UTC ("O"), so the text comparison is a time comparison.
cmd.CommandText = "SELECT " + ReceiptColumns +
" FROM RunReceipts WHERE OccurredAt >= @from AND OccurredAt < @to ORDER BY OccurredAt LIMIT @limit";
cmd.Parameters.AddWithValue("@from", Format(fromUtc));
cmd.Parameters.AddWithValue("@to", Format(toUtc));
cmd.Parameters.AddWithValue("@limit", limit);
var results = new List();
await using var reader = await cmd.ExecuteReaderAsync(ct);
while (await reader.ReadAsync(ct))
results.Add(ReadReceipt(reader));
return results;
}
public async Task> ListReceiptsCalledFromAsync(string runId, CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT " + ReceiptColumns + " FROM RunReceipts WHERE CallerRunId = @runId ORDER BY Id";
cmd.Parameters.AddWithValue("@runId", runId);
var results = new List();
await using var reader = await cmd.ExecuteReaderAsync(ct);
while (await reader.ReadAsync(ct))
results.Add(ReadReceipt(reader));
return results;
}
public async Task> ListReceiptsForTaskAsync(string taskId, CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT " + ReceiptColumns + " FROM RunReceipts WHERE TaskId = @taskId ORDER BY Id";
cmd.Parameters.AddWithValue("@taskId", taskId);
var results = new List();
await using var reader = await cmd.ExecuteReaderAsync(ct);
while (await reader.ReadAsync(ct))
results.Add(ReadReceipt(reader));
return results;
}
public async Task CountAsync(CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT COUNT(*) FROM AuditLog";
return Convert.ToInt32(await cmd.ExecuteScalarAsync(ct));
}
///
/// Offene Aufrufe (#554): ein Started, zu dem kein späterer Eintrag mit
/// derselben CallId existiert.
///
/// Einträge ohne CallId bleiben außen vor — sie stammen aus der Zeit vor #554
/// und ließen sich nicht paaren. Sie alle als offen zu melden, wäre schlimmer als
/// sie zu übergehen: Der erste Start nach der Aktualisierung legte sonst das ganze
/// Altprotokoll zur Prüfung vor.
///
public async Task> ListOpenCallsAsync(CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = $"""
SELECT {AuditColumns} FROM AuditLog a
WHERE a.Status = 'Started'
AND a.CallId <> ''
AND NOT EXISTS (
SELECT 1 FROM AuditLog b
WHERE b.CallId = a.CallId
AND b.Id > a.Id
AND b.Status <> 'Started'
)
ORDER BY a.Id
""";
return await ReadEntriesAsync(cmd, ct);
}
public async Task> ListForAgentAsync(
string agentId, DateTime? seit, int limit, CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = $"""
SELECT {AuditColumns} FROM AuditLog
WHERE AgentId = @agentId
AND (@seit IS NULL OR OccurredAt >= @seit)
ORDER BY Id DESC
LIMIT @limit
""";
cmd.Parameters.AddWithValue("@agentId", agentId);
cmd.Parameters.AddWithValue("@seit", seit is { } s ? Format(s) : DBNull.Value);
cmd.Parameters.AddWithValue("@limit", limit);
return await ReadEntriesAsync(cmd, ct);
}
///
/// Kennzahlen je Agent und Werkzeug (#556).
///
/// Die Dauer wird nur über abgeschlossene Aufrufe gemittelt. Ein
/// Started-Eintrag trägt die Dauer 0, weil zum Zeitpunkt des Schreibens noch
/// nichts gelaufen war — würde er mitgezählt, sähe jedes Werkzeug doppelt so schnell
/// aus, wie es ist.
///
public async Task> SummarizeToolUsageAsync(
string? agentId, DateTime? seit, CancellationToken ct)
{
await using var conn = await _storage.OpenConnectionAsync(ct);
using var cmd = conn.CreateCommand();
cmd.CommandText = """
SELECT AgentId, Tool,
SUM(CASE WHEN Status <> 'Started' THEN 1 ELSE 0 END) AS Calls,
SUM(CASE WHEN Status = 'Error' THEN 1 ELSE 0 END) AS Failed,
SUM(CASE WHEN Status = 'Denied' THEN 1 ELSE 0 END) AS Denied,
SUM(CASE WHEN Status = 'Started' THEN 1 ELSE 0 END) AS Begonnen,
SUM(CASE WHEN Status <> 'Started' THEN DurationMs ELSE 0 END) AS GesamtMs,
MAX(DurationMs) AS MaxMs
FROM AuditLog
WHERE (@agentId IS NULL OR AgentId = @agentId)
AND (@seit IS NULL OR OccurredAt >= @seit)
GROUP BY AgentId, Tool
ORDER BY Calls DESC
""";
cmd.Parameters.AddWithValue("@agentId", (object?)agentId ?? DBNull.Value);
cmd.Parameters.AddWithValue("@seit", seit is { } s ? Format(s) : DBNull.Value);
var results = new List();
await using var reader = await cmd.ExecuteReaderAsync(ct);
while (await reader.ReadAsync(ct))
{
var abgeschlossen = reader.GetInt32(2);
var begonnen = reader.GetInt32(5);
results.Add(new ToolUsageStat
{
AgentId = reader.GetString(0),
Tool = reader.GetString(1),
Calls = abgeschlossen,
Failed = reader.GetInt32(3),
Denied = reader.GetInt32(4),
// Begonnene minus abgeschlossene: Was übrig bleibt, hat nie geendet.
// Nie negativ — ein Abschluss ohne Start kann es nicht geben, aber
// Alteinträge ohne Started-Zeile gibt es sehr wohl.
Unfinished = Math.Max(0, begonnen - abgeschlossen),
TotalDurationMs = reader.GetInt64(6),
MaxDurationMs = reader.GetInt64(7)
});
}
return results;
}
// ─── Hilfsfunktionen ───
private const string AuditColumns =
"Id, RunId, AgentId, Model, Source, Tool, Arguments, Status, Summary, DurationMs, OccurredAt, CallId";
private const string ReceiptColumns =
"Id, RunId, AgentId, Model, Source, TaskId, Status, StepCount, PromptTokens, " +
"CompletionTokens, CachedTokens, CostUsd, CostIsKnown, DurationMs, ResultRef, OccurredAt, CallerRunId, BudgetPot";
private static async Task> ReadEntriesAsync(SqliteCommand cmd, CancellationToken ct)
{
var results = new List();
await using var reader = await cmd.ExecuteReaderAsync(ct);
while (await reader.ReadAsync(ct))
results.Add(ReadEntry(reader));
return results;
}
private static AuditEntry ReadEntry(SqliteDataReader r) => new()
{
Id = r.GetInt64(0),
RunId = r.GetString(1),
AgentId = r.GetString(2),
Model = r.GetString(3),
Source = r.GetString(4),
Tool = r.GetString(5),
Arguments = r.GetString(6),
Status = Enum.TryParse(r.GetString(7), out var s) ? s : AuditStatus.Ok,
Summary = r.GetString(8),
DurationMs = r.GetInt64(9),
OccurredAt = Parse(r.GetString(10)),
CallId = r.GetString(11)
};
private static RunReceipt ReadReceipt(SqliteDataReader r) => new()
{
Id = r.GetInt64(0),
RunId = r.GetString(1),
AgentId = r.GetString(2),
Model = r.GetString(3),
Source = r.GetString(4),
TaskId = r.IsDBNull(5) ? null : r.GetString(5),
Status = r.GetString(6),
StepCount = r.GetInt32(7),
PromptTokens = r.GetInt32(8),
CompletionTokens = r.GetInt32(9),
CachedTokens = r.GetInt32(10),
CostUsd = decimal.TryParse(r.GetString(11), NumberStyles.Any, CultureInfo.InvariantCulture, out var c) ? c : 0m,
CostIsKnown = r.GetInt32(12) != 0,
DurationMs = r.GetInt64(13),
ResultRef = r.GetString(14),
OccurredAt = Parse(r.GetString(15)),
CallerRunId = r.IsDBNull(16) ? null : r.GetString(16),
BudgetPot = r.IsDBNull(17) ? null : r.GetString(17)
};
private static string Format(DateTime value) => value.ToUniversalTime().ToString("O");
private static DateTime Parse(string value)
=> DateTime.TryParse(value, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var dt)
? dt
: DateTime.MinValue;
}