Verlorene Updates erkennen statt warten: optimistische Pruefung beim Ersetzen, Konflikte im Audit und im Lagebild (#720)
FileLedger merkt je Agent die gelesene Fassung einer Datei in gemeinsamen Bereichen. write, copy und Downloads (FTP, RocketChat) ersetzen eine Datei nur, wenn sie seit dem Lesen unveraendert ist; eine fremde Datei nicht ungelesen. append bleibt ungeprueft, macht aeltere Lesestaende aber ungueltig. Abgelehnte Schreibvorgaenge: eigener Audit-Status Conflict, Tabelle FileConflicts, Hotspots (>= 3 in 7 Tagen) in der Teamuebersicht. 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
8a7212c240
commit
79a3a7854f
@@ -37,7 +37,14 @@ public enum AuditStatus
|
||||
/// <see cref="Started"/>-Eintrag offen, und jeder gewöhnliche Abbruch sähe beim
|
||||
/// nächsten Start aus wie ein Absturz — bis niemand mehr hinsieht.</para>
|
||||
/// </summary>
|
||||
Cancelled
|
||||
Cancelled,
|
||||
|
||||
/// <summary>
|
||||
/// #720: The write was refused because the file changed since the agent read it, or
|
||||
/// another agent wrote it and this one never read it. Its own status, so the conflicts
|
||||
/// can be counted — frequent ones say the file should be stored differently.
|
||||
/// </summary>
|
||||
Conflict
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -60,6 +60,9 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
|
||||
/// </summary>
|
||||
public Budget.ISpecialBudgetRepository? SpecialBudgets { get; set; }
|
||||
|
||||
/// <summary>#720: Which version of a shared file each agent read — so a write cannot silently lose another.</summary>
|
||||
public Files.FileLedger? Files { get; set; }
|
||||
|
||||
/// <summary>#654: The sandbox tools run code in.</summary>
|
||||
public Sandbox.ISandboxRunner? Sandbox { get; set; }
|
||||
|
||||
@@ -1665,7 +1668,7 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
|
||||
|
||||
if (!result.Success)
|
||||
{
|
||||
await Audit(AuditStatus.Error, result.ErrorMessage ?? "");
|
||||
await Audit(result.IsConflict ? AuditStatus.Conflict : AuditStatus.Error, result.ErrorMessage ?? "");
|
||||
return JsonSerializer.Serialize(new { error = result.ErrorMessage });
|
||||
}
|
||||
|
||||
@@ -1856,6 +1859,7 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
|
||||
Cases = CaseDesk,
|
||||
SpecialBudgets = SpecialBudgets,
|
||||
Sandbox = Sandbox,
|
||||
Files = Files,
|
||||
// #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
|
||||
@@ -1890,7 +1894,7 @@ public sealed class AgentEngine : IAgentMessageRouter, Staging.IFrozenCallExecut
|
||||
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;
|
||||
var status = result.Success ? AuditStatus.Ok : result.IsConflict ? AuditStatus.Conflict : AuditStatus.Error;
|
||||
await RecordAuditAsync(runId, config, AuditSource.Approval, tool, argumentsJson,
|
||||
status, result.Success ? "Freigegeben ausgeführt" : (result.ErrorMessage ?? ""), sw.ElapsedMilliseconds);
|
||||
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Globalization;
|
||||
using System.Security.Cryptography;
|
||||
using ClawdDotNet.Core.Storage;
|
||||
|
||||
namespace ClawdDotNet.Core.Files;
|
||||
|
||||
/// <summary>#720: A write that was refused because it would have lost someone's change.</summary>
|
||||
public sealed record FileConflict(
|
||||
string Path,
|
||||
string Workspace,
|
||||
string AgentId,
|
||||
string? RunId,
|
||||
string? OtherAgentId,
|
||||
string? OtherRunId,
|
||||
string Kind,
|
||||
DateTime OccurredAtUtc)
|
||||
{
|
||||
/// <summary>The file changed after this agent read it.</summary>
|
||||
public const string ChangedSinceRead = "changed-since-read";
|
||||
|
||||
/// <summary>Another agent wrote the file, and this one never read it.</summary>
|
||||
public const string NotRead = "not-read";
|
||||
|
||||
/// <summary>What the agent is told — what happened, and what to do instead.</summary>
|
||||
public string Message(string displayPath)
|
||||
{
|
||||
var who = OtherAgentId is null ? "someone else" : $"agent '{OtherAgentId}'";
|
||||
return Kind == NotRead
|
||||
? $"Conflict: {displayPath} was last written by {who}, and you have not read it. " +
|
||||
"Read it first and make your change on its current content — or use 'append' if you only add to it."
|
||||
: $"Conflict: {displayPath} was changed by {who} since you read it. Your write would overwrite that change. " +
|
||||
"Read it again and make your change on the current content — or use 'append' if you only add to it.";
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>#720: A file that keeps running into conflicts — the candidate for a different kind of storage.</summary>
|
||||
public sealed record FileConflictHotspot(string Path, string Workspace, int Conflicts, DateTime LastAtUtc, IReadOnlyList<string> Agents);
|
||||
|
||||
/// <summary>
|
||||
/// #720: Lost updates in shared workspaces, found instead of waited for.
|
||||
///
|
||||
/// <para><b>Optimistic, not locked.</b> A lock held across model steps lasts minutes, is left
|
||||
/// behind when a run breaks off, and makes the waiting agents burn steps. Instead the
|
||||
/// instance remembers which version of a file each agent read. A write that replaces a file
|
||||
/// compares the file as it is now with that version; if someone changed it in between, the
|
||||
/// write is refused and the agent reads again. The agent needs no parameter for this, so it
|
||||
/// holds for a cheap model that would forget one.</para>
|
||||
///
|
||||
/// <para><b>What is remembered where.</b> Reads and last writers live in memory: after a
|
||||
/// restart there is no run that could build on an earlier read, and a file whose writer is
|
||||
/// unknown may be overwritten as before. Conflicts are kept in the database — they are what
|
||||
/// shows which files should become something else (split per agent, append-only, a table).</para>
|
||||
///
|
||||
/// <para>Appending is not checked: it loses nothing. It is noted, though — whoever read the
|
||||
/// file before the append does not know the appended part, and a later replace by them is a
|
||||
/// conflict. That is the point.</para>
|
||||
/// </summary>
|
||||
public sealed class FileLedger
|
||||
{
|
||||
private sealed record Mark(string Hash, string? RunId, DateTime AtUtc, string AgentId);
|
||||
|
||||
private readonly ConcurrentDictionary<(string Path, string Agent), Mark> _reads = new();
|
||||
private readonly ConcurrentDictionary<string, Mark> _writers = new();
|
||||
private readonly SqliteStorage? _storage;
|
||||
private readonly TimeProvider _clock;
|
||||
|
||||
public FileLedger(SqliteStorage? storage = null, TimeProvider? clock = null)
|
||||
{
|
||||
_storage = storage;
|
||||
_clock = clock ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <summary>The SHA-256 of a file's bytes; <c>null</c> if it does not exist.</summary>
|
||||
public static string? HashOf(string path)
|
||||
{
|
||||
if (!File.Exists(path))
|
||||
return null;
|
||||
using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete);
|
||||
return Convert.ToHexStringLower(SHA256.HashData(stream));
|
||||
}
|
||||
|
||||
/// <summary>An agent read the file in this version.</summary>
|
||||
public void NoteRead(string agentId, string path, string? hash)
|
||||
{
|
||||
if (hash is null)
|
||||
return;
|
||||
_reads[(Key(path), Agent(agentId))] = new Mark(hash, null, Now, agentId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Whether an agent may replace the file, given what it is now. <c>null</c>: yes. Pure
|
||||
/// apart from the lookups — the caller holds the file's lock while it checks and writes.
|
||||
/// </summary>
|
||||
public FileConflict? CheckReplace(string agentId, string? runId, string path, string workspace, string? currentHash)
|
||||
{
|
||||
if (currentHash is null)
|
||||
return null; // a new file replaces nothing
|
||||
|
||||
var key = Key(path);
|
||||
_writers.TryGetValue(key, out var writer);
|
||||
// The writer we know of is the one who made the file as it is now — otherwise
|
||||
// someone we do not know of (a human, a tool without the ledger) changed it since.
|
||||
var knownWriter = writer is not null && writer.Hash == currentHash ? writer : null;
|
||||
|
||||
if (_reads.TryGetValue((key, Agent(agentId)), out var read))
|
||||
{
|
||||
return read.Hash == currentHash
|
||||
? null
|
||||
: Conflict(FileConflict.ChangedSinceRead);
|
||||
}
|
||||
|
||||
return knownWriter is not null && !SameAgent(knownWriter.AgentId, agentId)
|
||||
? Conflict(FileConflict.NotRead)
|
||||
: null;
|
||||
|
||||
FileConflict Conflict(string kind) => new(
|
||||
path, workspace, agentId, runId,
|
||||
knownWriter is not null && !SameAgent(knownWriter.AgentId, agentId) ? knownWriter.AgentId : null,
|
||||
knownWriter is not null && !SameAgent(knownWriter.AgentId, agentId) ? knownWriter.RunId : null,
|
||||
kind, Now);
|
||||
}
|
||||
|
||||
/// <summary>The agent wrote the whole file: it is the last writer and knows what it wrote.</summary>
|
||||
public void NoteReplace(string agentId, string? runId, string path, string? newHash)
|
||||
{
|
||||
if (newHash is null)
|
||||
return;
|
||||
var key = Key(path);
|
||||
var mark = new Mark(newHash, runId, Now, agentId);
|
||||
_writers[key] = mark;
|
||||
_reads[(key, Agent(agentId))] = mark;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The agent appended. It is the last writer; its own read stays valid only if it knew
|
||||
/// the file as it was before — everyone else's read is now of an older version.
|
||||
/// </summary>
|
||||
public void NoteAppend(string agentId, string? runId, string path, string? hashBefore, string? hashAfter)
|
||||
{
|
||||
if (hashAfter is null)
|
||||
return;
|
||||
var key = Key(path);
|
||||
var mark = new Mark(hashAfter, runId, Now, agentId);
|
||||
_writers[key] = mark;
|
||||
var own = (key, Agent(agentId));
|
||||
if (hashBefore is null || (_reads.TryGetValue(own, out var read) && read.Hash == hashBefore))
|
||||
_reads[own] = mark;
|
||||
}
|
||||
|
||||
/// <summary>Keeps the conflict for the analysis. Best effort — the refusal itself does not depend on it.</summary>
|
||||
public async Task RecordAsync(FileConflict conflict, CancellationToken ct)
|
||||
{
|
||||
if (_storage is null)
|
||||
return;
|
||||
try
|
||||
{
|
||||
await _storage.WriteAsync(async conn =>
|
||||
{
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = """
|
||||
INSERT INTO FileConflicts (Path, Workspace, AgentId, RunId, OtherAgentId, OtherRunId, Kind, OccurredAt)
|
||||
VALUES (@path, @workspace, @agent, @run, @other, @otherRun, @kind, @at)
|
||||
""";
|
||||
cmd.Parameters.AddWithValue("@path", conflict.Path);
|
||||
cmd.Parameters.AddWithValue("@workspace", conflict.Workspace);
|
||||
cmd.Parameters.AddWithValue("@agent", conflict.AgentId);
|
||||
cmd.Parameters.AddWithValue("@run", (object?)conflict.RunId ?? DBNull.Value);
|
||||
cmd.Parameters.AddWithValue("@other", (object?)conflict.OtherAgentId ?? DBNull.Value);
|
||||
cmd.Parameters.AddWithValue("@otherRun", (object?)conflict.OtherRunId ?? DBNull.Value);
|
||||
cmd.Parameters.AddWithValue("@kind", conflict.Kind);
|
||||
cmd.Parameters.AddWithValue("@at", conflict.OccurredAtUtc.ToString("O", CultureInfo.InvariantCulture));
|
||||
await cmd.ExecuteNonQueryAsync(ct);
|
||||
}, ct);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
// The conflict was refused all the same; only the count is missing one.
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>The files with at least <paramref name="minimum"/> conflicts since <paramref name="sinceUtc"/>, most first.</summary>
|
||||
public async Task<IReadOnlyList<FileConflictHotspot>> HotspotsAsync(DateTime sinceUtc, int minimum, int limit, CancellationToken ct)
|
||||
{
|
||||
if (_storage is null)
|
||||
return [];
|
||||
|
||||
await using var conn = await _storage.OpenConnectionAsync(ct);
|
||||
using var cmd = conn.CreateCommand();
|
||||
cmd.CommandText = """
|
||||
SELECT Path, Workspace, AgentId, OtherAgentId, OccurredAt
|
||||
FROM FileConflicts
|
||||
WHERE OccurredAt >= @since
|
||||
""";
|
||||
cmd.Parameters.AddWithValue("@since", sinceUtc.ToString("O", CultureInfo.InvariantCulture));
|
||||
|
||||
var rows = new List<(string Path, string Workspace, string Agent, string? Other, DateTime At)>();
|
||||
await using (var reader = await cmd.ExecuteReaderAsync(ct))
|
||||
{
|
||||
while (await reader.ReadAsync(ct))
|
||||
{
|
||||
rows.Add((reader.GetString(0), reader.GetString(1), reader.GetString(2),
|
||||
reader.IsDBNull(3) ? null : reader.GetString(3),
|
||||
DateTime.Parse(reader.GetString(4), CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind)));
|
||||
}
|
||||
}
|
||||
|
||||
return rows
|
||||
.GroupBy(r => (r.Path, r.Workspace))
|
||||
.Select(g => new FileConflictHotspot(
|
||||
g.Key.Path, g.Key.Workspace, g.Count(), g.Max(r => r.At),
|
||||
g.SelectMany(r => r.Other is null ? new[] { r.Agent } : new[] { r.Agent, r.Other })
|
||||
.Distinct(StringComparer.OrdinalIgnoreCase).Order(StringComparer.OrdinalIgnoreCase).ToList()))
|
||||
.Where(h => h.Conflicts >= minimum)
|
||||
.OrderByDescending(h => h.Conflicts).ThenByDescending(h => h.LastAtUtc)
|
||||
.Take(limit)
|
||||
.ToList();
|
||||
}
|
||||
|
||||
private DateTime Now => _clock.GetUtcNow().UtcDateTime;
|
||||
|
||||
private static bool SameAgent(string a, string b) => string.Equals(a, b, StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static string Agent(string agentId) => agentId.ToLowerInvariant();
|
||||
|
||||
/// <summary>One key per file, however the path was written; case-insensitive where the file system is.</summary>
|
||||
private static string Key(string path)
|
||||
{
|
||||
var full = Path.GetFullPath(path);
|
||||
return OperatingSystem.IsWindows() || OperatingSystem.IsMacOS() ? full.ToLowerInvariant() : full;
|
||||
}
|
||||
}
|
||||
@@ -374,6 +374,20 @@ public sealed class SqliteStorage
|
||||
Json TEXT NOT NULL
|
||||
);
|
||||
|
||||
-- #720: Writes refused because they would have lost someone's change.
|
||||
CREATE TABLE IF NOT EXISTS FileConflicts (
|
||||
Id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
Path TEXT NOT NULL,
|
||||
Workspace TEXT NOT NULL,
|
||||
AgentId TEXT NOT NULL,
|
||||
RunId TEXT NULL,
|
||||
OtherAgentId TEXT NULL,
|
||||
OtherRunId TEXT NULL,
|
||||
Kind TEXT NOT NULL,
|
||||
OccurredAt TEXT NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS IX_FileConflicts_OccurredAt ON FileConflicts (OccurredAt);
|
||||
|
||||
-- #713: Personal tokens for the MCP server — only the hash is kept.
|
||||
CREATE TABLE IF NOT EXISTS McpTokens (
|
||||
Id TEXT PRIMARY KEY,
|
||||
|
||||
@@ -55,6 +55,12 @@ public sealed record AgentToolContext(
|
||||
/// <summary>#717: Special budgets — set up by an approved request_budget.</summary>
|
||||
public Budget.ISpecialBudgetRepository? SpecialBudgets { get; init; }
|
||||
|
||||
/// <summary>
|
||||
/// #720: Which version of a shared file each agent read. Tools that replace files in a
|
||||
/// shared workspace check it first. <c>null</c> in tests and outside the engine: no check.
|
||||
/// </summary>
|
||||
public Files.FileLedger? Files { get; init; }
|
||||
|
||||
/// <summary>#697: The run this call belongs to — for stamping where something came from.</summary>
|
||||
public string? RunId { get; init; }
|
||||
|
||||
|
||||
@@ -9,4 +9,9 @@ public sealed record ToolResult(
|
||||
public static ToolResult Ok(string content) => new(true, content);
|
||||
|
||||
public static ToolResult Fail(string error) => new(false, "", error);
|
||||
|
||||
/// <summary>#720: Refused because it would have lost another write — audited as a conflict, not as an error.</summary>
|
||||
public bool IsConflict { get; init; }
|
||||
|
||||
public static ToolResult Conflict(string error) => new(false, "", error) { IsConflict = true };
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user