Audit und Laufbelege: Tageswerte je Agent und Einheit, Monatsarchiv statt Loeschen (#700)
Nach der Vorgabe vom 2026-09-22 (so wenig loeschen wie moeglich): Laeufe aelter als hotDays (Vorgabe 180, 0 = alles bleibt) wandern aus AuditLog und RunReceipts in Archive/audit-JJJJ-MM.db - kopiert und entfernt in einer Transaktion, Zeilen-IDs bleiben, spaeter dazugekommene Spalten werden im Archiv ergaenzt. Vorher wird jeder Tag in RunDaily verdichtet (Laeufe, Fehlschlaege, Schrittgrenze, Tokens, Kosten, Werkzeugaufrufe, Fehler, Verweigert, Freigaben) mit der Einheit des Agenten; die Tageswerte bleiben. Nachschlagen eines Laufs findet ihn auch im Archiv. API /usage/days, /usage/archive; Karte 'Tageswerte' in der Verbrauchsansicht. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Richard
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
0dd8a761b8
commit
095722ae9c
@@ -0,0 +1,388 @@
|
||||
using System.Globalization;
|
||||
using System.Text.Json.Serialization;
|
||||
using ClawdDotNet.Core.Storage;
|
||||
using Microsoft.Data.Sqlite;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace ClawdDotNet.Core.Audit;
|
||||
|
||||
/// <summary>#700: How long audit and receipts stay in the working database (<c>auditArchive</c> in the instance settings).</summary>
|
||||
public sealed class AuditArchiveSettings
|
||||
{
|
||||
/// <summary>
|
||||
/// Days a run stays in the working database; older ones move to the monthly archive.
|
||||
/// 0 keeps everything in the working database. Nothing is ever deleted either way.
|
||||
/// </summary>
|
||||
[JsonPropertyName("hotDays")]
|
||||
public int HotDays { get; set; } = 180;
|
||||
}
|
||||
|
||||
/// <summary>#700: One day of one agent, condensed — stays in the working database for good.</summary>
|
||||
public sealed record RunDay(
|
||||
string Day, string AgentId, string UnitId, int Runs, int Failed, int LoopLimit,
|
||||
long PromptTokens, long CompletionTokens, decimal CostUsd, int ToolCalls, int ToolErrors, int Denied, int Staged);
|
||||
|
||||
/// <summary>An archive file as the administration shows it.</summary>
|
||||
public sealed record AuditArchiveFile(string Month, string FileName, long Bytes);
|
||||
|
||||
/// <summary>
|
||||
/// #700: The audit is evidence, and evidence is kept — the user's decision of 2026-09-22: as
|
||||
/// little deleted as possible. Five years after a failed project, how it came about can be read:
|
||||
/// which task was worded how, what the agent read, who approved what. That is the one real
|
||||
/// advantage an organisation of agents has over one of people.
|
||||
///
|
||||
/// <para><b>What grows is the working table, not the disk.</b> A hundred agents write some
|
||||
/// 7,400 audit rows a day; every supervision and every view reads those tables. So rows older
|
||||
/// than <see cref="AuditArchiveSettings.HotDays"/> move — never get deleted — into one SQLite
|
||||
/// file per month under <c>Archive/</c>, which the backup takes along like any other file.
|
||||
/// Looking up a run finds them there.</para>
|
||||
///
|
||||
/// <para><b>Day values stay.</b> Before rows move, each day is condensed per agent and unit
|
||||
/// (<see cref="RunDay"/>): runs, failures, tokens, costs, tool calls. Most questions over months
|
||||
/// are answered from those, without opening an archive. The unit is taken when the day is
|
||||
/// condensed — later "what did marketing cost in March" cannot be answered otherwise.</para>
|
||||
/// </summary>
|
||||
public sealed class AuditArchive(
|
||||
SqliteStorage storage,
|
||||
Func<AuditArchiveSettings> settings,
|
||||
Func<string, string?> unitOf,
|
||||
ILogger logger,
|
||||
TimeProvider? clock = null) : IAsyncDisposable
|
||||
{
|
||||
public const string FolderName = "Archive";
|
||||
|
||||
private readonly TimeProvider _clock = clock ?? TimeProvider.System;
|
||||
|
||||
public string Directory => ArchiveDirectoryOf(storage);
|
||||
|
||||
public static string ArchiveDirectoryOf(SqliteStorage storage)
|
||||
=> Path.Combine(Path.GetDirectoryName(Path.GetFullPath(storage.DatabasePath))!, FolderName);
|
||||
|
||||
public static string FileNameOf(string month) => $"audit-{month}.db";
|
||||
|
||||
private DateTime Now => _clock.GetUtcNow().UtcDateTime;
|
||||
|
||||
// ─── Day values ───
|
||||
|
||||
/// <summary>
|
||||
/// Condenses every day before today that is not condensed yet — and the last two again,
|
||||
/// because a run can end after midnight. Days are UTC days.
|
||||
/// </summary>
|
||||
public async Task<int> CondenseAsync(CancellationToken ct)
|
||||
{
|
||||
var today = Now.ToString("yyyy-MM-dd", CultureInfo.InvariantCulture);
|
||||
return await storage.WriteAsync(async conn =>
|
||||
{
|
||||
string? last;
|
||||
using (var max = conn.CreateCommand())
|
||||
{
|
||||
max.CommandText = "SELECT MAX(Day) FROM RunDaily";
|
||||
last = await max.ExecuteScalarAsync(ct) as string;
|
||||
}
|
||||
var from = last is null ? "" : DateOnly.ParseExact(last, "yyyy-MM-dd", CultureInfo.InvariantCulture).AddDays(-1).ToString("yyyy-MM-dd", CultureInfo.InvariantCulture);
|
||||
|
||||
var days = new Dictionary<(string Day, string Agent), RunDayBuilder>();
|
||||
using (var receipts = conn.CreateCommand())
|
||||
{
|
||||
receipts.CommandText = """
|
||||
SELECT substr(OccurredAt, 1, 10) AS Day, AgentId,
|
||||
SUM(CASE WHEN Status IN ('BudgetExceeded', 'TargetBusy', 'ShuttingDown') THEN 0 ELSE 1 END),
|
||||
SUM(CASE WHEN Status IN ('Failed', 'Cancelled', 'Incomplete') THEN 1 ELSE 0 END),
|
||||
SUM(CASE WHEN Status = 'LoopLimitExceeded' THEN 1 ELSE 0 END),
|
||||
SUM(PromptTokens), SUM(CompletionTokens), GROUP_CONCAT(CostUsd, ';')
|
||||
FROM RunReceipts
|
||||
WHERE substr(OccurredAt, 1, 10) >= @from AND substr(OccurredAt, 1, 10) < @today
|
||||
GROUP BY Day, AgentId
|
||||
""";
|
||||
receipts.Parameters.AddWithValue("@from", from);
|
||||
receipts.Parameters.AddWithValue("@today", today);
|
||||
await using var r = await receipts.ExecuteReaderAsync(ct);
|
||||
while (await r.ReadAsync(ct))
|
||||
{
|
||||
var b = Builder(days, r.GetString(0), r.GetString(1));
|
||||
b.Runs = r.GetInt32(2);
|
||||
b.Failed = r.GetInt32(3);
|
||||
b.LoopLimit = r.GetInt32(4);
|
||||
b.PromptTokens = r.GetInt64(5);
|
||||
b.CompletionTokens = r.GetInt64(6);
|
||||
b.CostUsd = r.IsDBNull(7) ? 0m : r.GetString(7).Split(';').Sum(c =>
|
||||
decimal.TryParse(c, NumberStyles.Any, CultureInfo.InvariantCulture, out var v) ? v : 0m);
|
||||
}
|
||||
}
|
||||
using (var audit = conn.CreateCommand())
|
||||
{
|
||||
audit.CommandText = """
|
||||
SELECT substr(OccurredAt, 1, 10) AS Day, AgentId,
|
||||
SUM(CASE WHEN Status NOT IN ('Started', 'Staged', 'Denied') THEN 1 ELSE 0 END),
|
||||
SUM(CASE WHEN Status = 'Error' THEN 1 ELSE 0 END),
|
||||
SUM(CASE WHEN Status = 'Denied' THEN 1 ELSE 0 END),
|
||||
SUM(CASE WHEN Status = 'Staged' THEN 1 ELSE 0 END)
|
||||
FROM AuditLog
|
||||
WHERE substr(OccurredAt, 1, 10) >= @from AND substr(OccurredAt, 1, 10) < @today AND AgentId <> ''
|
||||
GROUP BY Day, AgentId
|
||||
""";
|
||||
audit.Parameters.AddWithValue("@from", from);
|
||||
audit.Parameters.AddWithValue("@today", today);
|
||||
await using var r = await audit.ExecuteReaderAsync(ct);
|
||||
while (await r.ReadAsync(ct))
|
||||
{
|
||||
var b = Builder(days, r.GetString(0), r.GetString(1));
|
||||
b.ToolCalls = r.GetInt32(2);
|
||||
b.ToolErrors = r.GetInt32(3);
|
||||
b.Denied = r.GetInt32(4);
|
||||
b.Staged = r.GetInt32(5);
|
||||
}
|
||||
}
|
||||
|
||||
foreach (var ((day, agent), b) in days)
|
||||
{
|
||||
using var upsert = conn.CreateCommand();
|
||||
upsert.CommandText = """
|
||||
INSERT INTO RunDaily (Day, AgentId, UnitId, Runs, Failed, LoopLimit, PromptTokens, CompletionTokens,
|
||||
CostUsd, ToolCalls, ToolErrors, Denied, Staged)
|
||||
VALUES (@day, @agent, @unit, @runs, @failed, @loop, @prompt, @completion, @cost, @calls, @errors, @denied, @staged)
|
||||
ON CONFLICT (Day, AgentId) DO UPDATE SET
|
||||
Runs = excluded.Runs, Failed = excluded.Failed, LoopLimit = excluded.LoopLimit,
|
||||
PromptTokens = excluded.PromptTokens, CompletionTokens = excluded.CompletionTokens,
|
||||
CostUsd = excluded.CostUsd, ToolCalls = excluded.ToolCalls, ToolErrors = excluded.ToolErrors,
|
||||
Denied = excluded.Denied, Staged = excluded.Staged
|
||||
""";
|
||||
// The unit stays what it was when the day was first condensed.
|
||||
upsert.Parameters.AddWithValue("@day", day);
|
||||
upsert.Parameters.AddWithValue("@agent", agent);
|
||||
upsert.Parameters.AddWithValue("@unit", unitOf(agent) ?? "");
|
||||
upsert.Parameters.AddWithValue("@runs", b.Runs);
|
||||
upsert.Parameters.AddWithValue("@failed", b.Failed);
|
||||
upsert.Parameters.AddWithValue("@loop", b.LoopLimit);
|
||||
upsert.Parameters.AddWithValue("@prompt", b.PromptTokens);
|
||||
upsert.Parameters.AddWithValue("@completion", b.CompletionTokens);
|
||||
upsert.Parameters.AddWithValue("@cost", b.CostUsd.ToString(CultureInfo.InvariantCulture));
|
||||
upsert.Parameters.AddWithValue("@calls", b.ToolCalls);
|
||||
upsert.Parameters.AddWithValue("@errors", b.ToolErrors);
|
||||
upsert.Parameters.AddWithValue("@denied", b.Denied);
|
||||
upsert.Parameters.AddWithValue("@staged", b.Staged);
|
||||
await upsert.ExecuteNonQueryAsync(ct);
|
||||
}
|
||||
return days.Count;
|
||||
}, ct);
|
||||
}
|
||||
|
||||
private sealed class RunDayBuilder
|
||||
{
|
||||
public int Runs, Failed, LoopLimit, ToolCalls, ToolErrors, Denied, Staged;
|
||||
public long PromptTokens, CompletionTokens;
|
||||
public decimal CostUsd;
|
||||
}
|
||||
|
||||
private static RunDayBuilder Builder(Dictionary<(string, string), RunDayBuilder> days, string day, string agent)
|
||||
{
|
||||
if (!days.TryGetValue((day, agent), out var b))
|
||||
days[(day, agent)] = b = new RunDayBuilder();
|
||||
return b;
|
||||
}
|
||||
|
||||
/// <summary>Day values in a range (inclusive), optionally for one agent or one unit.</summary>
|
||||
public async Task<IReadOnlyList<RunDay>> DaysAsync(DateOnly from, DateOnly to, string? agentId, string? unitId, CancellationToken ct)
|
||||
{
|
||||
await using var conn = await storage.OpenConnectionAsync(ct);
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = """
|
||||
SELECT Day, AgentId, UnitId, Runs, Failed, LoopLimit, PromptTokens, CompletionTokens, CostUsd,
|
||||
ToolCalls, ToolErrors, Denied, Staged
|
||||
FROM RunDaily
|
||||
WHERE Day >= @from AND Day <= @to
|
||||
AND (@agent IS NULL OR AgentId = @agent COLLATE NOCASE)
|
||||
AND (@unit IS NULL OR UnitId = @unit COLLATE NOCASE)
|
||||
ORDER BY Day, AgentId
|
||||
""";
|
||||
cmd.Parameters.AddWithValue("@from", from.ToString("yyyy-MM-dd", CultureInfo.InvariantCulture));
|
||||
cmd.Parameters.AddWithValue("@to", to.ToString("yyyy-MM-dd", CultureInfo.InvariantCulture));
|
||||
cmd.Parameters.AddWithValue("@agent", (object?)agentId ?? DBNull.Value);
|
||||
cmd.Parameters.AddWithValue("@unit", (object?)unitId ?? DBNull.Value);
|
||||
var result = new List<RunDay>();
|
||||
await using var r = await cmd.ExecuteReaderAsync(ct);
|
||||
while (await r.ReadAsync(ct))
|
||||
result.Add(new RunDay(r.GetString(0), r.GetString(1), r.GetString(2), r.GetInt32(3), r.GetInt32(4), r.GetInt32(5),
|
||||
r.GetInt64(6), r.GetInt64(7),
|
||||
decimal.TryParse(r.GetString(8), NumberStyles.Any, CultureInfo.InvariantCulture, out var c) ? c : 0m,
|
||||
r.GetInt32(9), r.GetInt32(10), r.GetInt32(11), r.GetInt32(12)));
|
||||
return result;
|
||||
}
|
||||
|
||||
// ─── Moving to the archive ───
|
||||
|
||||
/// <summary>
|
||||
/// Moves audit rows and receipts older than the hot window into their month's archive file.
|
||||
/// Days are condensed first, so no day value is lost with the move. Copy and delete happen
|
||||
/// in one transaction: a row is in one place or the other, never in none.
|
||||
/// </summary>
|
||||
public async Task<int> ArchiveAsync(CancellationToken ct)
|
||||
{
|
||||
var hotDays = settings().HotDays;
|
||||
if (hotDays <= 0)
|
||||
return 0;
|
||||
await CondenseAsync(ct);
|
||||
|
||||
// Whole days only, and never today's or yesterday's: those are not condensed for good yet.
|
||||
var cutoff = Now.Date.AddDays(-Math.Max(2, hotDays)).ToString("O");
|
||||
System.IO.Directory.CreateDirectory(Directory);
|
||||
|
||||
return await storage.WriteAsync(async conn =>
|
||||
{
|
||||
var months = new SortedSet<string>(StringComparer.Ordinal);
|
||||
foreach (var table in new[] { "AuditLog", "RunReceipts" })
|
||||
{
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = $"SELECT DISTINCT substr(OccurredAt, 1, 7) FROM {table} WHERE OccurredAt < @cutoff";
|
||||
cmd.Parameters.AddWithValue("@cutoff", cutoff);
|
||||
await using var r = await cmd.ExecuteReaderAsync(ct);
|
||||
while (await r.ReadAsync(ct))
|
||||
months.Add(r.GetString(0));
|
||||
}
|
||||
|
||||
var moved = 0;
|
||||
foreach (var month in months)
|
||||
moved += await MoveMonthAsync(conn, month, cutoff, ct);
|
||||
if (moved > 0)
|
||||
logger.LogInformation("Audit archive: {Count} row(s) older than {Days} days moved to {Folder}", moved, hotDays, Directory);
|
||||
return moved;
|
||||
}, ct);
|
||||
}
|
||||
|
||||
private async Task<int> MoveMonthAsync(SqliteConnection conn, string month, string cutoff, CancellationToken ct)
|
||||
{
|
||||
var file = Path.Combine(Directory, FileNameOf(month));
|
||||
using (var attach = conn.CreateCommand())
|
||||
{
|
||||
attach.CommandText = "ATTACH DATABASE @file AS arch";
|
||||
attach.Parameters.AddWithValue("@file", file);
|
||||
await attach.ExecuteNonQueryAsync(ct);
|
||||
}
|
||||
try
|
||||
{
|
||||
using (var create = conn.CreateCommand())
|
||||
{
|
||||
// Same columns as the working tables, row ids kept — an id in a report still finds its row.
|
||||
create.CommandText = """
|
||||
CREATE TABLE IF NOT EXISTS arch.AuditLog AS SELECT * FROM main.AuditLog WHERE 0;
|
||||
CREATE TABLE IF NOT EXISTS arch.RunReceipts AS SELECT * FROM main.RunReceipts WHERE 0;
|
||||
CREATE INDEX IF NOT EXISTS arch.IX_Archive_AuditLog_Run ON AuditLog (RunId);
|
||||
CREATE INDEX IF NOT EXISTS arch.IX_Archive_RunReceipts_Run ON RunReceipts (RunId);
|
||||
CREATE INDEX IF NOT EXISTS arch.IX_Archive_RunReceipts_Task ON RunReceipts (TaskId);
|
||||
""";
|
||||
await create.ExecuteNonQueryAsync(ct);
|
||||
}
|
||||
|
||||
// A column the working table gained after the archive file was made is added to it —
|
||||
// otherwise the month's next move would fail on a column count.
|
||||
var columns = new Dictionary<string, IReadOnlyList<string>>();
|
||||
foreach (var table in new[] { "AuditLog", "RunReceipts" })
|
||||
columns[table] = await AlignColumnsAsync(conn, table, ct);
|
||||
|
||||
await using var tx = await conn.BeginTransactionAsync(ct);
|
||||
var moved = 0;
|
||||
foreach (var table in new[] { "AuditLog", "RunReceipts" })
|
||||
{
|
||||
var list = string.Join(", ", columns[table]);
|
||||
using var copy = conn.CreateCommand();
|
||||
copy.Transaction = (SqliteTransaction)tx;
|
||||
copy.CommandText = $"""
|
||||
INSERT INTO arch.{table} ({list}) SELECT {list} FROM main.{table}
|
||||
WHERE OccurredAt < @cutoff AND substr(OccurredAt, 1, 7) = @month
|
||||
AND Id NOT IN (SELECT Id FROM arch.{table});
|
||||
DELETE FROM main.{table} WHERE OccurredAt < @cutoff AND substr(OccurredAt, 1, 7) = @month
|
||||
AND Id IN (SELECT Id FROM arch.{table});
|
||||
""";
|
||||
copy.Parameters.AddWithValue("@cutoff", cutoff);
|
||||
copy.Parameters.AddWithValue("@month", month);
|
||||
moved += await copy.ExecuteNonQueryAsync(ct) / 2;
|
||||
}
|
||||
await tx.CommitAsync(ct);
|
||||
return moved;
|
||||
}
|
||||
finally
|
||||
{
|
||||
using var detach = conn.CreateCommand();
|
||||
detach.CommandText = "DETACH DATABASE arch";
|
||||
await detach.ExecuteNonQueryAsync(CancellationToken.None);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The working table's columns, each present in the archive table too.</summary>
|
||||
private static async Task<IReadOnlyList<string>> AlignColumnsAsync(SqliteConnection conn, string table, CancellationToken ct)
|
||||
{
|
||||
async Task<List<(string Name, string Type)>> Columns(string schema)
|
||||
{
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = $"SELECT name, type FROM pragma_table_info('{table}', '{schema}')";
|
||||
var list = new List<(string, string)>();
|
||||
await using var r = await cmd.ExecuteReaderAsync(ct);
|
||||
while (await r.ReadAsync(ct))
|
||||
list.Add((r.GetString(0), r.GetString(1)));
|
||||
return list;
|
||||
}
|
||||
|
||||
var main = await Columns("main");
|
||||
var archived = (await Columns("arch")).Select(c => c.Name).ToHashSet(StringComparer.OrdinalIgnoreCase);
|
||||
foreach (var (name, type) in main.Where(c => !archived.Contains(c.Name)))
|
||||
{
|
||||
using var add = conn.CreateCommand();
|
||||
add.CommandText = $"ALTER TABLE arch.{table} ADD COLUMN {name} {type}";
|
||||
await add.ExecuteNonQueryAsync(ct);
|
||||
}
|
||||
return main.Select(c => c.Name).ToList();
|
||||
}
|
||||
|
||||
/// <summary>The archive files, newest month first.</summary>
|
||||
public IReadOnlyList<AuditArchiveFile> Files() => FilesIn(Directory);
|
||||
|
||||
private readonly CancellationTokenSource _cts = new();
|
||||
private Task? _loop;
|
||||
|
||||
/// <summary>Condenses and archives once after the start, then every six hours.</summary>
|
||||
public void Start() => _loop ??= LoopAsync(_cts.Token);
|
||||
|
||||
private async Task LoopAsync(CancellationToken ct)
|
||||
{
|
||||
try { await Task.Delay(TimeSpan.FromMinutes(2), ct); }
|
||||
catch (OperationCanceledException) { return; }
|
||||
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await ArchiveAsync(ct);
|
||||
if (settings().HotDays <= 0)
|
||||
await CondenseAsync(ct);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
logger.LogWarning(ex, "Audit archive: condensing or moving failed");
|
||||
}
|
||||
|
||||
try { await Task.Delay(TimeSpan.FromHours(6), ct); }
|
||||
catch (OperationCanceledException) { return; }
|
||||
}
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _cts.CancelAsync();
|
||||
if (_loop is not null)
|
||||
{
|
||||
try { await _loop; }
|
||||
catch (OperationCanceledException) { }
|
||||
}
|
||||
_cts.Dispose();
|
||||
}
|
||||
|
||||
public static IReadOnlyList<AuditArchiveFile> FilesIn(string directory)
|
||||
{
|
||||
if (!System.IO.Directory.Exists(directory))
|
||||
return [];
|
||||
return System.IO.Directory.EnumerateFiles(directory, "audit-????-??.db")
|
||||
.Select(f => new AuditArchiveFile(Path.GetFileNameWithoutExtension(f)["audit-".Length..], Path.GetFileName(f), new FileInfo(f).Length))
|
||||
.OrderByDescending(f => f.Month, StringComparer.Ordinal)
|
||||
.ToList();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user