Activity log: bound the message list, flush overflow to segment satellites (2/3)

clients/react/src/i18n/strings.de.json

@@ -412,6 +412,7 @@
   "ui.sourceChanged": "Quelle geändert — muss kompiliert werden",
   "ui.upToDate": "Aktuell",
   "ui.notYetRun": "Noch nicht ausgeführt.",
+  "ui.activityEarlierMessages": "… {0} frühere Meldungen archiviert",
   "ui.activityNoOutput": "Dieser Lauf hat keine Ausgabe erzeugt.",
   "ui.running": "Wird ausgeführt…",
   "ui.statusDone": "Fertig",

clients/react/src/i18n/strings.en.json

@@ -412,6 +412,7 @@
   "ui.sourceChanged": "Source changed — needs compile",
   "ui.upToDate": "Up to date",
   "ui.notYetRun": "Not yet run.",
+  "ui.activityEarlierMessages": "… {0} earlier messages archived",
   "ui.activityNoOutput": "This run produced no output.",
   "ui.running": "Running…",
   "ui.statusDone": "Done",

src/MeshWeaver.ContentCollections.Indexing.Graph/ContentIndexingActivity.cs

@@ -210,17 +210,12 @@ private static void Append(
         // context-null → the partition RLS denies → the progress line never lands.
         using (owner is not null && accessService is not null ? accessService.SwitchAccessContext(owner) : null)
         {
-            workspace.GetMeshNodeStream(activityPath).Update(node =>
-            {
-                // ContentAs<T>, NOT `Content as ActivityLog`: the activity node is owned by the partition's
-                // _Activity hub, so this cross-hub Update lambda diffs against a LOCAL mirror whose Content
-                // can be a JsonElement (not strongly-typed). A plain cast would be null → the update no-ops →
-                // the RFC 7396 patch is never sent. ContentAs deserializes the JsonElement so the write lands.
-                // (project_baddata_contentas_pattern.)
-                var log = node.ContentAs<ActivityLog>(workspace.Hub.JsonSerializerOptions, logger);
-                if (log is null) return node;
-                return node with { Content = log.Append(messages) };
-            }).Subscribe(_ => { }, ex => logger?.LogWarning(ex, "Indexing activity log append failed for {Path}", activityPath));
+            // ActivityLogAppender, not a hand-rolled Update: it keeps Messages bounded and flushes the
+            // overflow into _Log segment satellites, so a full reindex's per-batch progress costs
+            // O(batches) instead of O(batches²) in re-serialisation.
+            ActivityLogAppender
+                .Append(workspace.Hub, activityPath, messages, mutate: null, logger)
+                .Subscribe(_ => { }, ex => logger?.LogWarning(ex, "Indexing activity log append failed for {Path}", activityPath));
         }
     }
 
@@ -233,23 +228,15 @@ private static IObservable<Unit> Finish(
         // RLS lets the write land — otherwise the activity hangs Running forever.
         using (owner is not null && accessService is not null ? accessService.SwitchAccessContext(owner) : null)
         {
-            return workspace.GetMeshNodeStream(activityPath).Update(node =>
-            {
-                // ContentAs<T>, NOT `Content as ActivityLog`: the cross-hub Update lambda diffs against a
-                // local mirror whose Content may be a JsonElement; a plain cast would no-op and the terminal
-                // Status would never be written → the activity hangs Running forever (the ReindexAll / Upload
-                // indexing-activity timeout). See project_baddata_contentas_pattern.
-                var log = node.ContentAs<ActivityLog>(workspace.Hub.JsonSerializerOptions, logger);
-                if (log is null) return node;
-                var withFinal = string.IsNullOrEmpty(finalMessage)
-                    ? log
-                    : log.Append(new LogMessage(finalMessage,
-                        status == ActivityStatus.Failed ? LogLevel.Error : LogLevel.Information));
-                return node with
-                {
-                    Content = withFinal with { Status = status, End = DateTime.UtcNow, RequestedStatus = null }
-                };
-            }).Select(_ => Unit.Default);
+            return ActivityLogAppender.Append(
+                    workspace.Hub, activityPath,
+                    string.IsNullOrEmpty(finalMessage)
+                        ? []
+                        : [new LogMessage(finalMessage,
+                            status == ActivityStatus.Failed ? LogLevel.Error : LogLevel.Information)],
+                    log => log with { Status = status, End = DateTime.UtcNow, RequestedStatus = null },
+                    logger)
+                .Select(_ => Unit.Default);
         }
     }
 

src/MeshWeaver.Data.Contract/ActivityLog.cs

@@ -1,6 +1,7 @@
 using System.Collections.Immutable;
 using System.ComponentModel.DataAnnotations;
 using System.Text.Json;
+using System.Text.Json.Serialization;
 using Microsoft.Extensions.Logging;
 using MeshWeaver.ShortGuid;
 
@@ -20,12 +21,58 @@ public record ActivityLog(string Category)
     /// <summary>The workspace version recorded when the activity finished.</summary>
     public int Version { get; init; }
 
+    /// <summary>
+    /// How many messages the head keeps before sealing the oldest ones into an
+    /// <see cref="ActivityLogSegment"/> satellite. An activity that never exceeds this behaves
+    /// exactly as it did before segments existed — same single write per append, same bytes.
+    /// </summary>
+    public const int MessageWindowLimit = 500;
+
+    /// <summary>
+    /// How many messages stay on the head after a seal. The gap to
+    /// <see cref="MessageWindowLimit"/> is what amortises the seal: one extra write and one segment
+    /// create per <c>Limit - Keep</c> appends. Every live reader wants far less than this — the
+    /// widest is a 12-line preview — so nothing on screen notices the trim.
+    /// </summary>
+    public const int MessageWindowKeep = 100;
+
     /// <summary>Unique identifier of this activity.</summary>
     [property: Key]
     public string Id { get; init; } = Guid.NewGuid().AsString();
-    /// <summary>The log messages accumulated by this activity, in order.</summary>
+
+    /// <summary>
+    /// The activity's log messages, most recent last — a bounded WINDOW, not necessarily the whole
+    /// transcript. Once an activity exceeds <see cref="MessageWindowLimit"/> the oldest lines are
+    /// sealed into <see cref="ActivityLogSegment"/> satellites under <c>{activityPath}/_Log</c> and
+    /// drop out of here, so a write stays O(1) instead of O(transcript).
+    ///
+    /// <para>🚨 This is the reason nothing may derive a total, a severity, or a progress signal from
+    /// <c>Messages.Count</c> — use <see cref="TotalMessageCount"/> and <see cref="MaxSeverity"/>.
+    /// The window still holds the LAST messages, so tail readers (<c>[^1]</c>, <c>TakeLast(n)</c>)
+    /// are unaffected.</para>
+    /// </summary>
     public ImmutableList<LogMessage> Messages { get; init; } = ImmutableList<LogMessage>.Empty;
 
+    /// <summary>
+    /// How many <see cref="ActivityLogSegment"/> satellites this activity has sealed — also the index
+    /// of the next one. A reader enumerates the segments with a children query rather than trusting
+    /// this count, so a seal whose satellite write failed leaves a harmless gap, never a dangling
+    /// point-read.
+    /// </summary>
+    public int SegmentCount { get; init; }
+
+    /// <summary>
+    /// How many LEADING entries of <see cref="Messages"/> a seal has claimed but not yet trimmed —
+    /// the flush's claim marker, written atomically with the append that made it.
+    ///
+    /// <para>🚨 This is what makes a flush safe without a lock. Writes to one node are serialised by
+    /// the owning hub, so exactly one appender can move this 0 → n; every other appender sees it
+    /// non-zero and does not claim a second, overlapping slice. The claimed messages stay in
+    /// <see cref="Messages"/> until the segment is durable, so a crash or a failed segment write
+    /// loses nothing — the next append simply retries the same slice.</para>
+    /// </summary>
+    public int SealingCount { get; init; }
+
     /// <summary>
     /// How many messages this activity has appended in total — <see cref="Messages"/> plus anything
     /// no longer held there. Maintained incrementally by <see cref="Append(IReadOnlyList{LogMessage})"/>.
@@ -93,7 +140,12 @@ public record ActivityLog(string Category)
     /// How many messages this activity has recorded, counting any no longer held in
     /// <see cref="Messages"/>. Back-compatible with logs persisted before <see cref="MessageCount"/>
     /// existed (those carry 0, so the in-list count wins).
+    /// <para>🚨 <c>[JsonIgnore]</c>, like every computed property here. System.Text.Json serialises
+    /// get-only properties by default, and anything that reaches the JSON reaches the RFC 7396 merge
+    /// patch — where a derived field the writer never set becomes a phantom conflict the merge guard
+    /// refuses ("changed since the writer's base"), taking the real fields in that patch down with it.</para>
     /// </summary>
+    [JsonIgnore]
     public int TotalMessageCount => Math.Max(MessageCount, Messages.Count);
 
     /// <summary>
@@ -126,6 +178,58 @@ public ActivityLog Append(IReadOnlyList<LogMessage> messages)
     /// <returns>The activity log with the message appended and the roll-ups advanced.</returns>
     public ActivityLog Append(LogMessage message) => Append([message]);
 
+    /// <summary>
+    /// Claims the oldest messages for sealing into an <see cref="ActivityLogSegment"/> when the window
+    /// has overflowed — a no-op below <see cref="MessageWindowLimit"/> and while another seal is still
+    /// in flight. Nothing is removed here: the claim only marks how many LEADING entries of
+    /// <see cref="Messages"/> the next segment will carry, so the messages remain readable on the head
+    /// until the satellite is durable.
+    /// <para>Called inside the head's update lambda, which the owning hub serialises — that is what
+    /// makes "exactly one appender claims each slice" true without a lock.</para>
+    /// </summary>
+    /// <returns>The activity log with the claim recorded, or the same instance when nothing to claim.</returns>
+    public ActivityLog ClaimSeal() =>
+        SealingCount == 0 && Messages.Count > MessageWindowLimit
+            ? this with { SealingCount = Messages.Count - MessageWindowKeep }
+            : this;
+
+    /// <summary>
+    /// The messages a claimed seal must write into its segment — the leading
+    /// <see cref="SealingCount"/> entries of <see cref="Messages"/>. Empty when no seal is claimed.
+    /// </summary>
+    /// <remarks>🚨 <c>[JsonIgnore]</c> — without it this list would be serialised into every patch
+    /// alongside <see cref="Messages"/>, doubling the payload of exactly the write the window exists
+    /// to shrink.</remarks>
+    [JsonIgnore]
+    public ImmutableList<LogMessage> SealedMessages =>
+        SealingCount <= 0 ? ImmutableList<LogMessage>.Empty : Messages.GetRange(0, Math.Min(SealingCount, Messages.Count));
+
+    /// <summary>
+    /// The activity-wide ordinal of the first message in the currently claimed seal — what the segment
+    /// records as its <see cref="ActivityLogSegment.FirstOrdinal"/>.
+    /// </summary>
+    [JsonIgnore]
+    public int SealedFirstOrdinal => TotalMessageCount - Messages.Count;
+
+    /// <summary>
+    /// Drops the claimed messages from the window now that their segment is durable, and advances
+    /// <see cref="SegmentCount"/>. Idempotent: a no-op when no seal is claimed, so a retried completion
+    /// cannot trim a second, unrelated slice.
+    /// <para>Trimming from the FRONT is what makes this order-independent: messages only ever arrive at
+    /// the end, so the leading <see cref="SealingCount"/> entries are still exactly the ones the claim
+    /// captured, however many appends landed in between.</para>
+    /// </summary>
+    /// <returns>The activity log with the sealed slice removed.</returns>
+    public ActivityLog CompleteSeal() =>
+        SealingCount <= 0
+            ? this
+            : this with
+            {
+                Messages = Messages.RemoveRange(0, Math.Min(SealingCount, Messages.Count)),
+                SegmentCount = SegmentCount + 1,
+                SealingCount = 0,
+            };
+
     /// <summary>
     /// Returns a copy of this activity marked <see cref="ActivityStatus.Failed"/> with the
     /// given error appended as an error message and <see cref="End"/> set to now.

src/MeshWeaver.Data.Contract/ActivityLogSegment.cs

@@ -0,0 +1,44 @@
+using System.Collections.Immutable;
+using System.ComponentModel.DataAnnotations;
+using MeshWeaver.ShortGuid;
+
+namespace MeshWeaver.Data;
+
+/// <summary>
+/// A sealed slice of an activity's transcript, stored as a satellite MeshNode at
+/// <c>{activityPath}/_Log/{index:D6}</c> once the messages have scrolled out of
+/// <see cref="ActivityLog.Messages"/> (the head's bounded window).
+///
+/// <para><b>Why the transcript moves off the head.</b> Every <c>stream.Update</c> re-serialises the
+/// WHOLE <c>MeshNode.Content</c> to compute its patch, so appending N lines to one growing list costs
+/// O(N²) — on memex-cloud a single import activity spent ~719 MB of serialisation across 5,239 writes
+/// to a 141 kB node. No delta field fixes that: the cross-hub path ships an RFC 7396 merge patch, which
+/// clones a changed array whole, and the three-way merge's base extraction clones the previous array
+/// too. Bounding the head is what changes the asymptotics — with the window fixed, each write is O(1)
+/// and the whole activity is O(N).</para>
+///
+/// <para><b>Segments are append-once and immutable.</b> A segment is written exactly once, by the
+/// append that sealed it, and never updated afterwards — so nothing ever reads-then-writes a segment
+/// path, and no reader may point-read one that might be absent (that opens the shared stream cache's
+/// storm breaker on a path a concurrent write is about to use). Enumerate them with a children query
+/// on <c>{activityPath}/_Log</c>; the ordering key is <see cref="FirstOrdinal"/>.</para>
+/// </summary>
+public record ActivityLogSegment
+{
+    /// <summary>Unique identifier of this segment — mirrors the node id.</summary>
+    [property: Key]
+    public string Id { get; init; } = Guid.NewGuid().AsString();
+
+    /// <summary>
+    /// The activity-wide ordinal of this segment's FIRST message: how many messages the activity had
+    /// already recorded before this slice. Segments are ordered by it, and it is what lets a reader
+    /// stitch segments and the head window back into the original sequence.
+    /// </summary>
+    public int FirstOrdinal { get; init; }
+
+    /// <summary>The messages in this slice, in order.</summary>
+    public ImmutableList<LogMessage> Messages { get; init; } = ImmutableList<LogMessage>.Empty;
+
+    /// <summary>The path of the activity this slice belongs to.</summary>
+    public string? ActivityPath { get; init; }
+}

src/MeshWeaver.Data/DataExtensions.cs

@@ -199,6 +199,9 @@ private static MessageHubConfiguration GetDefaultConfiguration(MessageHubConfigu
             )
             .WithType(typeof(Address), nameof(Address))
             .WithType(typeof(ActivityLog), nameof(ActivityLog))
+            // A sealed slice of an activity's transcript, stored as its own satellite node. Registered
+            // beside ActivityLog because the segment crosses the wire on exactly the same paths.
+            .WithType(typeof(ActivityLogSegment), nameof(ActivityLogSegment))
             // ActivityLog content children — serialised inside log entries / user attribution.
             .WithType(typeof(LogMessage), nameof(LogMessage))
             .WithType(typeof(UserInfo), nameof(UserInfo))

src/MeshWeaver.Documentation/Data/Architecture/ActivityOperations.md

@@ -159,6 +159,49 @@ The running script receives the cancellation, throws `OperationCanceledException
 
 ---
 
+## Writing log messages: `ActivityLogAppender`
+
+Every log line written onto a **persisted activity node** goes through `ActivityLogAppender.Append` (`src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs`). It is the append-side twin of the transition surface above: callers hand it messages plus an optional change to the log (terminal status, `End`, `ReturnValue`) and it performs **one** `stream.Update` carrying both — so a reader can never observe the terminal status before the lines that explain it.
+
+```csharp
+ActivityLogAppender.Append(hub, activityPath, [new LogMessage(text, LogLevel.Information)])
+    .Subscribe(_ => { }, ex => logger.LogDebug(ex, "activity append failed"));
+
+// terminal status + its explanation, atomically:
+ActivityLogAppender.Append(hub, activityPath, [new LogMessage(error, LogLevel.Error)],
+        log => log with { Status = ActivityStatus.Failed, End = DateTime.UtcNow })
+    .Subscribe(_ => { }, ex => logger.LogDebug(ex, "activity complete failed"));
+```
+
+### `Messages` is a bounded window
+
+`ActivityLog.Messages` holds at most `ActivityLog.MessageWindowLimit` (500) entries. Older lines are sealed into `ActivityLogSegment` satellites at `{activityPath}/_Log/{index:D6}` and drop off the head, leaving `MessageWindowKeep` (100) behind.
+
+**Why.** Every `stream.Update` re-serialises the whole `MeshNode.Content` to compute its patch, so appending N lines to one growing list costs **O(N²)** — measured at ~719 MB of serialisation for a single memex-cloud import activity (5,239 writes over a 141 kB node), and the dominant term in that pod's CFS throttling. A delta field does not escape it: the cross-hub path ships an RFC 7396 merge patch, which clones a changed array whole, and the three-way merge's base extraction clones the previous array too — so one append to an N-element collection ships ~2N elements. Bounding the head is the only lever that changes the asymptotics; with the window fixed, each write is O(1) and the activity is O(N).
+
+**Below the window nothing changes.** An activity that never reaches 500 messages takes exactly the single write per append it always did, with byte-identical content. Only long activities — the ones that actually hurt — take the new path.
+
+### Consequences for readers
+
+| You want… | Read… |
+|---|---|
+| How many lines the activity produced | `log.TotalMessageCount` — **never** `log.Messages.Count` |
+| Whether it errored / its terminal status | `log.HasErrors()`, `log.Finish(...)` — both answer from the `MaxSeverity` counter |
+| The latest line, or the last few | `log.Messages` — the window keeps the **most recent** entries, so `[^1]` and `TakeLast(n)` are unaffected |
+| The full transcript | the window plus the `_Log` segments, ordered by `ActivityLogSegment.FirstOrdinal` |
+
+🚨 **Enumerate segments with a children query on `{activityPath}/_Log`, never a point-read of a segment path.** A point-read of an absent satellite opens the shared stream cache's storm breaker on a path a concurrent write is about to use, and the breaker fast-fails writes too.
+
+🚨 **Never derive progress from `Messages.Count`.** It stops growing once an activity passes the window, so anything keyed on it — a `DistinctUntilChanged`, a change detector — silently freezes for exactly the long-running activities it exists to follow. `TotalMessageCount` is the monotonic signal.
+
+### How the flush stays safe without a lock
+
+A seal is **claimed inside the head's update lambda** (`ActivityLog.ClaimSeal`), which the owning hub serialises — so exactly one appender claims each slice and no two claims overlap. The claimed messages **stay on the head** until the segment write succeeds; only then are they trimmed (`ActivityLog.CompleteSeal`). A crash or a failed segment write therefore loses nothing and needs no watchdog: the claim is still standing and the messages are still there, so the next append retries the identical slice against the same (deterministic) segment index.
+
+`ActivityLogLogger` (the kernel's script logger) is the one writer that does **not** use the appender: it re-asserts whole content on every 100 ms flush rather than patching, so it is the single writer of its node and seals its own overflow directly under the lock it already holds for the terminal settle. Same window, same segment shape, no claim protocol needed because there is no second appender to race.
+
+---
+
 ## See also
 
 - [Activity Control Plane](/Doc/Architecture/ActivityControlPlane) — the `Status` / `RequestedStatus` pattern and how to wire your own NodeType to it