Add bulk entry updates as a job and the get_job tool

CONTRIBUTING.md

@@ -56,7 +56,7 @@ The browser tests drive the admin in headless Chromium through Playwright: keybo
 
 To run the host, `dotnet run --project src/FluentCMS.Host` starts the web host on `http://localhost:5000` with a SQLite database in `fluentcms.db`, and `dotnet run --project src/FluentCMS.Host -- mcp` starts the stdio mode. Configuration comes from environment variables such as `Database__ConnectionString`, `Database__Provider` (`sqlite` or `postgresql`, inferred from the connection string when omitted), `JobQueue__WorkerCount`, `Content__DeletedRetention` (how long a deleted entry can be restored, `30.00:00:00` by default), `Entities__VerifyWithin` (how long a fact stays verified before it is listed as stale, `180.00:00:00` by default), `Delivery__PreviewTokenLifetime` (how long a preview token works unless it is revoked or its change set ends, `7.00:00:00` by default), `Delivery__RateLimit__PermitLimit` and `Delivery__RateLimit__Window` (how many delivery API requests one client address may make per window, 600 a minute by default), `ReverseProxy__KnownProxies__0` and `ReverseProxy__KnownNetworks__0` (the reverse proxies, by address such as `10.0.0.5` or network such as `10.0.0.0/8`, whose `X-Forwarded-For` names the client that rate limits count, read one hop deep; with none, the header is ignored), `Media__Storage` (the storage new uploads go to, `local` by default), `Media__LocalDisk__Root` (where the local storage keeps bytes, `media` under the content root by default) and `Media__MaxBytes` (the largest upload, 50 MB by default; over MCP on HTTP, the server's 30 MB request limit holds an upload to about 22 MB, since base64 grows a file by a third), and from an optional `appsettings.json` next to the binary.
 
-Agents reach the MCP server (ADR-0007) through either mode. The web host serves streamable HTTP at `/mcp` in stateless mode: each request stands alone, no session id is issued, and a GET answers 405 because the server never opens a stream to the client. `fluentcms mcp` serves the same server over standard input and output to an agent that starts it as a local process, for example with `dotnet run --project src/FluentCMS.Host -- mcp` as its command. It reads the same settings as the web host, so pass it the web host's `Database__ConnectionString` and `DataProtection__KeysDirectory` to work on the same database and key ring. Relative paths resolve against the process's working directory, so give an agent absolute ones. Standard output carries only protocol messages, log lines go to standard error, and the process stops when its standard input closes. The stdio mode runs with the local user's database access, so it is meant for development and single-user setups. Clients on protocol revision 2026-07-28 connect with `server/discover`, earlier ones with `initialize`. Agents read content types, the schema document and change sets with the read-only tools, and propose changes with `open_change_set`, `create_content_type`, `revise_content_type` and `submit_change_set`. A write stores an unpublished version and its operation in the agent's change set together, and the version is not published until a person approves and publishes the set; no tool approves, rejects or publishes. Agents follow review with the read-only desk tools: `get_desk_counts`, `list_desk_queue` and `get_change_set_review`, which see only the change sets the agent opened.
+Agents reach the MCP server (ADR-0007) through either mode. The web host serves streamable HTTP at `/mcp` in stateless mode: each request stands alone, no session id is issued, and a GET answers 405 because the server never opens a stream to the client. `fluentcms mcp` serves the same server over standard input and output to an agent that starts it as a local process, for example with `dotnet run --project src/FluentCMS.Host -- mcp` as its command. It reads the same settings as the web host, so pass it the web host's `Database__ConnectionString` and `DataProtection__KeysDirectory` to work on the same database and key ring. Relative paths resolve against the process's working directory, so give an agent absolute ones. Standard output carries only protocol messages, log lines go to standard error, and the process stops when its standard input closes. The stdio mode runs with the local user's database access, so it is meant for development and single-user setups. Clients on protocol revision 2026-07-28 connect with `server/discover`, earlier ones with `initialize`. Agents read content types, the schema document and change sets with the read-only tools, and propose changes with `open_change_set`, `create_content_type`, `revise_content_type` and `submit_change_set`. A write stores an unpublished version and its operation in the agent's change set together, and the version is not published until a person approves and publishes the set; no tool approves, rejects or publishes. Agents follow review with the read-only desk tools: `get_desk_counts`, `list_desk_queue` and `get_change_set_review`, which see only the change sets the agent opened. `bulk_update_entries` opens a change set for many entry or entity updates and returns its id and a job id at once; the job applies the updates in the background (see Background jobs), and `get_job` shows its progress and any item that failed. The set can't be submitted while the job is still filling it.
 
 Every MCP call runs as an agent and needs that agent's key, issued on `/agents/new`. Over HTTP the key goes in the `Authorization: Bearer <key>` header of each request; `/mcp` reads no other credential, so a signed-in person's admin cookie is ignored there. Over stdio the key goes in the `FLUENTCMS_AGENT_KEY` environment variable of the process, which an agent's MCP configuration can set; it is never read from a command-line argument, which other users can see in process listings. Both transports check the key on every request or message, so an expired or revoked key is refused from its next call on, with a message that says why. Never put a key in a URL: the server ignores one there, but ASP.NET Core logs request URLs once `Logging__LogLevel__Microsoft.AspNetCore` is below its shipped `Warning`. Every tool call is recorded in the `ToolCalls` table with the agent, the tool, the outcome and how long it took. An agent can add the model it runs on and the task it works on as strings in the call's `_meta`, under `fluentcms/model` and `fluentcms/task`, and the record keeps both. Arguments and results are not recorded; the audit log records every change a tool makes.
 

src/FluentCMS.Core/Agents/AgentPermissions.cs

@@ -173,6 +173,34 @@ await RecordAsync(
             + $"{cap} (items per change set). Submit this change set and add the rest to a new one.");
     }
 
+    public async Task<Result> CheckNewChangeSetSizeAsync(int items, CancellationToken cancellationToken = default)
+    {
+        if (currentInitiator.Current is SystemInitiator system)
+        {
+            return SystemRefused(system, "open a change set");
+        }
+
+        var agent = await AgentAsync(cancellationToken);
+        if (!agent.IsSuccess)
+        {
+            return Result.Failure(agent.Errors);
+        }
+
+        if (agent.Value is not { } credential || items <= credential.Caps.ItemsPerChangeSet)
+        {
+            return Result.Success();
+        }
+
+        await RecordAsync(
+            new NewAuditEvent("agent.items_per_change_set_exceeded") { SubjectType = "agent_credential", SubjectId = credential.Id },
+            cancellationToken);
+
+        return Result.Failure(
+            AgentPermissionErrorCodes.CapExceeded,
+            $"{items} items are more than the credential '{credential.DisplayName}' allows in one change set: {credential.Caps.ItemsPerChangeSet} "
+            + "(items per change set). Split them over several calls of at most that many.");
+    }
+
     public async Task<Result> CheckContentAsync(
         string contentType,
         AgentOperation operation,

src/FluentCMS.Core/Agents/IAgentPermissions.cs

@@ -24,6 +24,9 @@ public interface IAgentPermissions
     // or ahead of a write with the number of operations it will add.
     Task<Result> CheckChangeSetSizeAsync(ChangeSet changeSet, int adding = 0, CancellationToken cancellationToken = default);
 
+    // Ahead of opening a set for that many items at once, as a bulk write does, before the set exists.
+    Task<Result> CheckNewChangeSetSizeAsync(int items, CancellationToken cancellationToken = default);
+
     // The operation on items of the content type (MCP-05). A write names the fields it changes, which a field list in the scope can
     // narrow; a read names none, so an agent reads whole entries of a type in its scope.
     Task<Result> CheckContentAsync(

src/FluentCMS.Core/Builtin/Jobs/BulkEntryUpdates.cs

@@ -0,0 +1,151 @@
+using System.Text.Json;
+using FluentCMS.Core.ChangeSets;
+using FluentCMS.Core.Content;
+using FluentCMS.Core.Jobs;
+using FluentCMS.Core.Persistence;
+using FluentCMS.Core.Results;
+using FluentCMS.Core.Schema;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace FluentCMS.Core.Builtin.Jobs;
+
+// The background half of bulk_update_entries (MCP-08): each item's new version and its operation in the change set commit together,
+// one item at a time, as the agent that asked, so its permissions, scopes and caps apply to every item. An item that fails is
+// recorded and the rest go on. An item already in the set is skipped, so a retried job repeats nothing.
+public sealed class BulkEntryUpdates : IJobHandler
+{
+    public const string Name = "bulk.entry_updates";
+
+    private const int MaxFailuresListed = 20;
+
+    // Short enough that the count and every listed item fit in the job's last error, which the queue keeps to 4000 characters.
+    private const int MaxReasonLength = 100;
+
+    public string Type => Name;
+
+    public static JsonElement Payload(Guid changeSetId, IReadOnlyList<BulkEntryUpdate> items)
+    {
+        return JsonSerializer.SerializeToElement(new Request(changeSetId, items), JsonSerializerOptions.Web);
+    }
+
+    private static Request? Read(JsonElement payload)
+    {
+        try
+        {
+            return payload.ValueKind == JsonValueKind.Object
+                && payload.Deserialize<Request>(JsonSerializerOptions.Web) is { ChangeSetId: var id, Items: not null } request
+                && id != Guid.Empty
+                    ? request
+                    : null;
+        }
+        catch (JsonException)
+        {
+            return null;
+        }
+    }
+
+    public async Task<Result> RunAsync(JobContext context, CancellationToken cancellationToken)
+    {
+        // Another attempt reads the same payload, so one this handler cannot read is a failure, not an exception to retry.
+        if (Read(context.Job.Payload) is not { } request)
+        {
+            return Result.Failure("payload_invalid", "A bulk update names its change set as changeSetId and its items as a list of id, baseVersion and fields.");
+        }
+
+        var services = context.Services;
+        var changeSets = services.GetRequiredService<IChangeSetDirectory>();
+
+        var set = await changeSets.GetAsync(request.ChangeSetId, cancellationToken);
+        if (!set.IsSuccess)
+        {
+            return Result.Failure(set.Errors);
+        }
+
+        var applied = set.Value.Operations.Select(operation => operation.ItemId).ToHashSet();
+        var failures = new List<Error>();
+        var total = request.Items.Count;
+
+        for (var index = 0; index < total; index++)
+        {
+            var item = request.Items[index];
+            if (applied.Add(item.Id))
+            {
+                var updated = await ApplyAsync(services, request.ChangeSetId, item, cancellationToken);
+                if (!updated.IsSuccess && !await AppliedMeanwhileAsync(changeSets, request.ChangeSetId, item.Id, cancellationToken))
+                {
+                    // One entry per item, however many problems it has, so the counts and the list are of items.
+                    var reason = string.Join(" ", updated.Errors.Select(error => error.Message));
+                    failures.Add(new Error(updated.Errors[0].Code, $"items[{index}] {item.Id}: {Shorten(reason)}"));
+                }
+            }
+
+            var message = $"Updated {index + 1 - failures.Count} of {total}" + (failures.Count > 0 ? $"; {failures.Count} failed." : ".");
+            if (!await context.ReportProgressAsync(index + 1, total, message, cancellationToken))
+            {
+                // Another worker has the job now and goes on from what is already in the set.
+                return Result.Success();
+            }
+        }
+
+        if (failures.Count == 0)
+        {
+            return Result.Success();
+        }
+
+        // The count first, so it is kept whatever follows.
+        var counted = $"{failures.Count} of {total} items failed" + (failures.Count > MaxFailuresListed ? $"; the first {MaxFailuresListed} are listed." : ".");
+        return Result.Failure([new Error("bulk_items_failed", counted), .. failures.Take(MaxFailuresListed)]);
+    }
+
+    private static string Shorten(string reason)
+    {
+        return reason.Length <= MaxReasonLength ? reason : string.Concat(reason.AsSpan(0, MaxReasonLength - 1), "…");
+    }
+
+    // A worker that lost its lease can commit an item after this run read the set; the item is in the set, so it has not failed.
+    private static async Task<bool> AppliedMeanwhileAsync(IChangeSetDirectory changeSets, Guid changeSetId, Guid itemId, CancellationToken cancellationToken)
+    {
+        var set = await changeSets.GetAsync(changeSetId, cancellationToken);
+        return set.IsSuccess && set.Value.Operations.Any(operation => operation.ItemId == itemId);
+    }
+
+    // The update and its operation commit together or not at all; leaving the unit of work uncommitted undoes both.
+    private static async Task<Result> ApplyAsync(IServiceProvider services, Guid changeSetId, BulkEntryUpdate item, CancellationToken cancellationToken)
+    {
+        await using var scope = await services.GetRequiredService<IUnitOfWork>().BeginAsync(cancellationToken);
+
+        var entry = await services.GetRequiredService<IEntryDirectory>().UpdateAsync(item.Id, item.BaseVersion, item.Fields, cancellationToken);
+        if (!entry.IsSuccess)
+        {
+            return Result.Failure(entry.Errors);
+        }
+
+        var type = await services.GetRequiredService<IContentTypeDirectory>().GetAsync(entry.Value.ContentType, cancellationToken);
+        if (!type.IsSuccess)
+        {
+            return Result.Failure(type.Errors);
+        }
+
+        var kind = type.Value.Definition.IsEntity ? ItemKind.Entity : ItemKind.Entry;
+        var operation = entry.Value.PublishedVersion is { } published
+            ? ChangeOperation.Create(OperationKind.Update, kind, entry.Value.Id, published, entry.Value.Latest.Version)
+            : ChangeOperation.Create(OperationKind.Create, kind, entry.Value.Id, null, entry.Value.Latest.Version);
+        if (!operation.IsSuccess)
+        {
+            return Result.Failure(operation.Errors);
+        }
+
+        var added = await services.GetRequiredService<IChangeSetDirectory>().AddOperationAsync(changeSetId, operation.Value, cancellationToken);
+        if (!added.IsSuccess)
+        {
+            return Result.Failure(added.Errors);
+        }
+
+        await scope.CommitAsync(cancellationToken);
+        return Result.Success();
+    }
+
+    private sealed record Request(Guid ChangeSetId, IReadOnlyList<BulkEntryUpdate> Items);
+}
+
+public sealed record BulkEntryUpdate(Guid Id, int BaseVersion, JsonElement Fields);