From 31280c7abf218790696ffbd5abf9dbcafa9678ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Thu, 13 Aug 2026 13:09:01 +0200 Subject: [PATCH 1/2] perf: ship a string SPLICE in the cross-hub patch, not the whole value MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the chat half of the quadratic-write class (#1284, #1351). The activities half (#1398/#1399/#1400) moved growing log content to satellites; chat cannot take that treatment — a streaming cell must show all the text so far, and Thread.Messages already holds only ids (~10 kB of head at 200 messages, two orders of magnitude off being the problem). Chat's cost is the STREAMING WRITE. ThreadExecution streams through Sample(100 ms) into PushToResponseMessage, which rewrites the response cell's whole content per tick. Cross-hub that means ComputeMergePatchDiff emits the entire new text AND MeshNodePatchMerge.ExtractBaseValues emits the entire previous text, so a round costs O(ticks x final length) — measured 3.84 MB for a 20 kB answer, all of it LOH traffic on writer and owner both. A changed string leaf >= 1 kB whose change is smaller than the value now travels as patch { "$sd": [start, removedLength, "inserted"] } BaseValues { "$sdb": [baseLength, "fingerprint"] } Both halves are required: splicing only the patch leaves the base re-shipping the previous value every tick and the round stays quadratic. The owner applies a splice ONLY when the fingerprint (length + truncated SHA-256) proves its live text is the text the writer diffed against — then the result is byte-identical to what the whole-value patch would have written. On mismatch it REFUSES: keeps the newer live value and NACKs Conflict, which re-runs the writer's lambda against fresh state and re-diffs. So two mirrors splicing the same string concurrently cannot interleave into corrupted text; the loser re-diffs. The cost is that a DISJOINT concurrent edit to a big string, which the full-value three-way merge rebased and landed, now costs a round trip — a rare extra hop against a constant per-tick, per-viewer megabyte. Unchanged below 1 kB and for arrays/scalars: arrays in particular still need their full base (MergeArray consumes it), so that door stays shut as #1351 established. Measured, real runs, two sizes (StreamingCellWriteByteCountTest): 5,000 chars / 50 ticks 20,000 chars / 200 ticks before wire bytes 257,788 4,031,152 after wire bytes 23,794 69,881 before bytes/char 51.6 201.6 (grows 4x) after bytes/char 4.8 3.5 (falls) 57.7x less at 20 kB, and the per-character cost no longer grows with length. Correctness, since chat is the most user-visible surface: the final text is byte-identical (asserted at 20 kB / 200 splices and on a real streamed round), every intermediate state a viewer sees is a prefix of the final answer and never shrinks, and a late subscriber reads the whole text. Rolling deploys: the marker rides inside the patch, so a new writer reaching a not-yet-upgraded owner produces a merged node that fails to deserialize — NACKed and never committed, and self-healing because the mirror never advanced. The alternative (a sibling property an old owner ignores) would instead have it ACK a write it only partly applied — the acked-write-loss class of #648. A visibly stalled field beats a silently lost one. Not addressed here, and stated rather than implied: the owner->subscriber fan-out (JsonSynchronizationStream.CreateSingleObjectPatch, RFC 6902) still replaces the whole string per tick per subscriber, and #1351's MessageDelivery.Package double materialisation is a ~2x transport constant. Both are separate wire formats with separate consumers. Co-Authored-By: Claude Opus 5 (1M context) --- src/MeshWeaver.Data/DataExtensions.cs | 17 ++ .../Serialization/MeshNodePatchMerge.cs | 79 +++++ .../Serialization/PatchStringSplice.cs | 162 ++++++++++ .../Data/Architecture/MeshNodeStreamCache.md | 51 ++++ ...nswer-no-longer-costs-more-than-it-says.md | 39 +++ .../MeshNodeStreamExtensions.cs | 24 +- .../PatchStringSpliceTest.cs | 288 ++++++++++++++++++ .../StreamedTextIntegrityTest.cs | 228 ++++++++++++++ .../StreamingCellWriteByteCountTest.cs | 214 +++++++++++++ 9 files changed, 1100 insertions(+), 2 deletions(-) create mode 100644 src/MeshWeaver.Data/Serialization/PatchStringSplice.cs create mode 100644 src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-a-long-answer-no-longer-costs-more-than-it-says.md create mode 100644 test/MeshWeaver.Data.Test/PatchStringSpliceTest.cs create mode 100644 test/MeshWeaver.Threading.Test/StreamedTextIntegrityTest.cs create mode 100644 test/MeshWeaver.Threading.Test/StreamingCellWriteByteCountTest.cs diff --git a/src/MeshWeaver.Data/DataExtensions.cs b/src/MeshWeaver.Data/DataExtensions.cs index fb60fb8965..a6a53b3201 100644 --- a/src/MeshWeaver.Data/DataExtensions.cs +++ b/src/MeshWeaver.Data/DataExtensions.cs @@ -883,6 +883,7 @@ InvalidOperationException ioe when ioe.Message.Contains("Patch must be a JSON ob /// Patch semantics: /// /// null in patch → remove the key from current. + /// A marker → splice onto current's text. /// Object in patch AND object in current at same key → recurse (deep merge). /// Anything else → replace current's value with patch's value (deep clone). /// @@ -890,6 +891,9 @@ InvalidOperationException ioe when ioe.Message.Contains("Patch must be a JSON ob /// (e.g. { "Content": { "RequestedCancellationAt": "..." } }) leaves every /// other nested field on the owner intact, instead of being overwritten by a /// full-object replacement. + /// 🚨 The splice case is not optional here. This is the base-less last-write-wins path, + /// and a marker object written verbatim into a string field would corrupt it — so the marker is + /// decoded wherever a patch is applied, not only on the three-way path that produced it. /// internal static void MergePatchRecursive( System.Text.Json.Nodes.JsonObject current, @@ -902,6 +906,19 @@ internal static void MergePatchRecursive( current.Remove(kvp.Key); continue; } + // Spliced string leaf. No base is carried on this path, so — exactly as + // StringDeltaPatch.Apply / EntityDelta.Apply do — the splice replays onto the + // target's CURRENT text, which is also what last-write-wins means here. + if (Serialization.PatchStringSplice.TryDecode(kvp.Value, out var splice)) + { + if (current[kvp.Key] is System.Text.Json.Nodes.JsonValue liveValue + && liveValue.TryGetValue(out var liveText)) + current[kvp.Key] = System.Text.Json.Nodes.JsonValue.Create(splice.Apply(liveText)); + // No string to splice onto (absent, or the shape changed) → leave the live value + // alone. Falling through would write the MARKER OBJECT into the field, which is the + // one outcome this whole branch exists to make impossible. + continue; + } if (kvp.Value is System.Text.Json.Nodes.JsonObject patchObj && current[kvp.Key] is System.Text.Json.Nodes.JsonObject currentObj) { diff --git a/src/MeshWeaver.Data/Serialization/MeshNodePatchMerge.cs b/src/MeshWeaver.Data/Serialization/MeshNodePatchMerge.cs index 3a0dc0fe00..247ba9203c 100644 --- a/src/MeshWeaver.Data/Serialization/MeshNodePatchMerge.cs +++ b/src/MeshWeaver.Data/Serialization/MeshNodePatchMerge.cs @@ -49,6 +49,14 @@ public static void Apply( && baseValues.TryGetPropertyValue(key, out var baseRaw) && baseRaw is not null; var baseVal = hasBase ? baseValues![key] : null; + // SPLICED string leaf — resolved before every other branch: its patch value is a marker + // object, not the field's value, so no generic rule below may touch it. + if (PatchStringSplice.TryDecode(patchVal, out var splice)) + { + ApplySplice(live, key, splice, liveVal, baseVal, onRefuse); + continue; + } + // Nested object on patch AND live → recurse (deep three-way merge). if (patchVal is JsonObject patchObj && liveVal is JsonObject liveObj) { @@ -109,6 +117,68 @@ public static void Apply( } } + /// + /// Resolves a leaf the writer encoded as a . + /// + /// The splice's offsets were computed against the writer's base, so they only address the + /// right characters while the owner's live text IS that base. The base fingerprint decides: + /// + /// Fingerprint matches live → apply the splice. The result is provably the exact + /// string a full-value patch would have written — the splice is a pure encoding win. + /// Fingerprint does not match → the owner changed the field since the writer's + /// base. REFUSE: keep the newer live value and report, which NACKs Conflict and + /// re-runs the writer's update lambda against the fresh state (the same self-healing route a + /// refused scalar takes). A splice is never applied at an offset the fingerprint did not + /// vouch for, so concurrent splices from two mirrors cannot corrupt the text — the loser is + /// re-diffed, never interleaved blind. + /// A FULL base string carried alongside the splice (a sender that did not + /// fingerprint) → the full three-way string rebase, identical to the whole-value path. + /// + /// + private static void ApplySplice( + JsonObject live, string key, StringDelta splice, + JsonNode? liveVal, JsonNode? baseVal, Action? onRefuse) + { + if (!TryGetString(liveVal, out var liveStr)) + { + // Live is absent or no longer a string — the splice has nothing to anchor to. + onRefuse?.Invoke(key); + return; + } + + if (PatchStringSplice.TryDecodeBase(baseVal, out var baseLength, out var fingerprint)) + { + if (PatchStringSplice.BaseMatches(liveStr, baseLength, fingerprint)) + { + live[key] = JsonValue.Create(splice.Apply(liveStr)); + return; + } + onRefuse?.Invoke(key); + return; + } + + if (TryGetString(baseVal, out var baseStr)) + { + if (JsonNode.DeepEquals(liveVal, baseVal)) + { + live[key] = JsonValue.Create(splice.Apply(liveStr)); + return; + } + var intervening = StringDelta.Compute(baseStr, liveStr); + if (StringDelta.Overlaps(splice, intervening)) + { + onRefuse?.Invoke(key); + return; + } + live[key] = JsonValue.Create(StringDelta.ApplyAll(baseStr, new[] { intervening, splice })); + return; + } + + // No base signal at all: nothing vouches for the offsets. Refusing costs a re-diff; + // splicing blind costs the user's text. + onRefuse?.Invoke(key); + } + /// /// TWO-WAY sibling of , for the one caller that has no base to rebase against: a /// write that lost a version race AT THE STORE. There the write-integrity chain holds only the two @@ -285,6 +355,10 @@ private static JsonArray MergeArray(JsonArray live, JsonArray patch, JsonArray b /// . A leaf absent at base (a writer-added field) is omitted — there is no /// old value to conflict with. Returns null when nothing comparable is carried (an empty base /// object would just be wire overhead). + /// 🚨 A leaf the patch encoded as a SPLICE carries a compact FINGERPRINT of the base + /// () instead of the whole old string. Without this the + /// splice would be pointless: the base half alone would still re-ship the entire previous value on + /// every write, leaving the round O(ticks × length). /// public static JsonObject? ExtractBaseValues(JsonObject baseNode, JsonObject patch) { @@ -293,6 +367,11 @@ private static JsonArray MergeArray(JsonArray live, JsonArray patch, JsonArray b { if (!baseNode.TryGetPropertyValue(key, out var baseVal) || baseVal is null) continue; // base lacked this field → no conflict signal + if (PatchStringSplice.TryDecode(patchVal, out _) && TryGetString(baseVal, out var spliceBase)) + { + result[key] = PatchStringSplice.EncodeBase(spliceBase); + continue; + } if (patchVal is JsonObject patchObj && baseVal is JsonObject baseObj) { var sub = ExtractBaseValues(baseObj, patchObj); diff --git a/src/MeshWeaver.Data/Serialization/PatchStringSplice.cs b/src/MeshWeaver.Data/Serialization/PatchStringSplice.cs new file mode 100644 index 0000000000..88bb1efb10 --- /dev/null +++ b/src/MeshWeaver.Data/Serialization/PatchStringSplice.cs @@ -0,0 +1,162 @@ +using System.Runtime.InteropServices; +using System.Security.Cryptography; +using System.Text.Json.Nodes; + +namespace MeshWeaver.Data.Serialization; + +/// +/// The SPLICE encoding for a changed string leaf inside a cross-hub RFC 7396 merge patch +/// (PatchDataRequest), plus the compact BASE FINGERPRINT that keeps the writer's +/// three-way-merge signal for that leaf O(1) instead of O(length). +/// +/// Why. A cross-hub stream.Update ships the changed leaf's WHOLE new value and +/// — via — its WHOLE old value. For a field that +/// GROWS one chunk at a time (an agent response cell streaming tokens at +/// Sample(100 ms)) that makes the round's wire cost +/// O(ticks × final length): a 20 kB answer re-ships ~20 kB twice per tick, megabytes +/// for one reply, every byte of it a fresh Large-Object-Heap allocation on the owner and on the +/// writer. Satellites cannot fix it — a streaming cell must show all the text so far — but the +/// PATCH does not have to carry text the owner already has. Encoding the leaf as a splice makes the +/// write O(chunk), and it does so for EVERY large string field, not just chat. +/// +/// Wire shape. Two objects, each a single reserved $-prefixed key so they can +/// never be confused with domain content: +/// +/// in the patch — { "$sd": [start, removedLength, "inserted"] }; +/// in BaseValues at the same leaf — +/// { "$sdb": [baseLength, "fingerprint"] }. +/// +/// +/// Correctness. The owner applies the splice at its literal offsets ONLY when the +/// fingerprint proves its live text is byte-identical to the text the writer diffed against — then +/// the result is provably the same string the full-value patch would have written. When the +/// fingerprint does NOT match, the owner has moved since the writer's base and the splice's offsets +/// no longer address the text they were computed for, so the leaf is REFUSED (the owner keeps its +/// newer value and NACKs Conflict, which re-runs the writer's update lambda against the fresh +/// state and re-diffs). Refusing is the conservative half of the trade: a splice is never applied at +/// an unverified offset, so two mirrors splicing the same string concurrently can never corrupt it. +/// +/// +/// Distinct from , which encodes the same splice for the +/// EntityDeltaUpdate sync-stream path and replays it onto the target's CURRENT text with no +/// base signal at all. Both use the same $sd marker on purpose — one shape, one meaning. +/// +public static class PatchStringSplice +{ + /// Marker key of a spliced string leaf. Shared with . + public const string Marker = StringDeltaPatch.Marker; + + /// Marker key of the compact base fingerprint carried in BaseValues. + public const string BaseMarker = "$sdb"; + + /// + /// Below this length the whole value is cheaper than a splice plus a fingerprint, and the + /// three-way merge keeps its full-fidelity string rebase. Only big string fields — markdown + /// bodies, prerendered html, a streaming agent response — take the splice path. Matches + /// , the same threshold the sync-stream delta uses. + /// + public const int MinSpliceLength = EntityDelta.MinDeltaSize; + + // Wire cost of the wrapper itself: {"$sd":[NNNNN,NNNNN,""]} plus JSON escaping headroom. + // Used only to reject a "splice" that would be no smaller than the value it replaces. + private const int MarkerOverhead = 48; + + // 96 bits of SHA-256, base64. Paired with an exact length check, so a false "unchanged since + // base" verdict needs a same-length preimage collision — far below every other failure mode + // on this path, and unlike a non-cryptographic hash it is not craftable from user content. + private const int FingerprintBytes = 12; + + /// + /// Encodes → as a splice, when that is + /// worth doing: the new value is at least long, the strings really + /// differ, and the encoded splice is smaller than the value it replaces. Returns false + /// (and leaves null) otherwise, so the caller ships the full value + /// with unchanged semantics. + /// + public static bool TryEncode(string? oldValue, string? newValue, out JsonObject? encoded) + { + encoded = null; + if (newValue is null || newValue.Length < MinSpliceLength) + return false; + var delta = StringDelta.Compute(oldValue, newValue); + if (delta.IsEmpty) + return false; + if (delta.Inserted.Length + MarkerOverhead >= newValue.Length) + return false; // a near-total rewrite — the full value is no bigger and keeps full fidelity + encoded = new JsonObject + { + [Marker] = new JsonArray(delta.Start, delta.RemovedLength, delta.Inserted) + }; + return true; + } + + /// Builds the compact base-fingerprint node for a spliced leaf. + public static JsonObject EncodeBase(string? baseValue) + { + baseValue ??= string.Empty; + return new JsonObject + { + [BaseMarker] = new JsonArray(baseValue.Length, Fingerprint(baseValue)) + }; + } + + /// + /// Recognises a spliced string leaf. Deliberately strict — exactly one property, exactly the + /// marker, exactly [int, int, string] with non-negative offsets — so an ordinary content + /// object can never be mistaken for one. + /// + public static bool TryDecode(JsonNode? node, out StringDelta delta) + { + delta = default; + if (node is not JsonObject obj || obj.Count != 1) + return false; + if (!obj.TryGetPropertyValue(Marker, out var value) || value is not JsonArray { Count: 3 } arr) + return false; + if (arr[0] is not JsonValue s || !s.TryGetValue(out var start) || start < 0) + return false; + if (arr[1] is not JsonValue r || !r.TryGetValue(out var removed) || removed < 0) + return false; + if (arr[2] is not JsonValue i || !i.TryGetValue(out var inserted)) + return false; + delta = new StringDelta(start, removed, inserted); + return true; + } + + /// Recognises the compact base fingerprint written by . + public static bool TryDecodeBase(JsonNode? node, out int length, out string fingerprint) + { + length = 0; + fingerprint = string.Empty; + if (node is not JsonObject obj || obj.Count != 1) + return false; + if (!obj.TryGetPropertyValue(BaseMarker, out var value) || value is not JsonArray { Count: 2 } arr) + return false; + if (arr[0] is not JsonValue l || !l.TryGetValue(out var len) || len < 0) + return false; + if (arr[1] is not JsonValue f || !f.TryGetValue(out var fp)) + return false; + length = len; + fingerprint = fp; + return true; + } + + /// + /// True when is the same text the writer diffed against — the ONLY + /// condition under which a splice may be applied at its literal offsets. + /// + public static bool BaseMatches(string? live, int length, string fingerprint) + { + live ??= string.Empty; + return live.Length == length + && string.Equals(Fingerprint(live), fingerprint, StringComparison.Ordinal); + } + + /// Stable content fingerprint of a string (truncated SHA-256 over its UTF-16 bytes). + public static string Fingerprint(string? value) + { + value ??= string.Empty; + Span hash = stackalloc byte[32]; + SHA256.HashData(MemoryMarshal.AsBytes(value.AsSpan()), hash); + return Convert.ToBase64String(hash[..FingerprintBytes]); + } +} diff --git a/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md b/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md index 68c89f490a..4216c6a57f 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/MeshNodeStreamCache.md @@ -128,6 +128,57 @@ hub.SubmitMessage(threadPath, "second"); hub.SubmitMessage(threadPath, "third"); ``` +## Big strings ship as a SPLICE, not as the whole value + +Step 2 above — "ships only the JSON-merge patch" — is per *leaf*, and a changed leaf normally travels +whole. For a field that **grows one chunk at a time** that is quadratic: an agent response cell +re-sent its entire text on every `Sample(100 ms)` tick, and `ExtractBaseValues` re-sent the entire +*previous* text alongside it, so one reply cost `O(ticks × final length)` — **3.8 MB measured for a +20 kB answer**, all of it Large-Object-Heap traffic on the writer and the owner both. + +So a changed **string** leaf at least `PatchStringSplice.MinSpliceLength` (1 kB) long, whose change +is smaller than the value, is encoded as a splice instead: + +| where | shape | +|---|---| +| in the patch | `{ "$sd": [start, removedLength, "inserted"] }` | +| in `BaseValues`, same leaf | `{ "$sdb": [baseLength, "fingerprint"] }` | + +Both halves matter. Splicing only the patch would still leave the base half re-shipping the previous +value every tick, and the round would stay `O(ticks × length)`. + +**The owner applies a splice only when it can prove the offsets still address the right characters.** +The fingerprint (length + truncated SHA-256) is compared against the owner's live text: + +- **match** → the owner's text *is* the text the writer diffed against, so applying the splice yields + byte-for-byte the string a whole-value patch would have written. This is the normal path. +- **mismatch** → the owner moved on since the writer's base. The leaf is **refused**: the owner keeps + its newer value and NACKs `Conflict`, which re-runs the writer's update lambda against the fresh + state and re-diffs — the same self-healing route a refused scalar takes. + +That refusal is the deliberate half of the trade. A splice is never applied at an offset the +fingerprint did not vouch for, so two mirrors splicing the same string concurrently cannot interleave +into corrupted text; the loser re-diffs instead. The cost is that a *disjoint* concurrent edit to a +big string, which the full-value three-way merge would have rebased and landed, now costs a round +trip. For every field this applies to — streaming cells, markdown bodies, prerendered html — that is +a trade of a rare extra round trip against a constant, per-tick, per-viewer megabyte. + +Below 1 kB, and for arrays and scalars, nothing changes: the whole value and the whole base still +travel, and `MeshNodePatchMerge` resolves them exactly as before (arrays in particular *need* their +full base — `MergeArray` consumes it to tell a writer's removal from an element the owner dropped). + +**Rolling deploys.** The marker lives *inside* the patch, so during the minute or two a rollout has +old and new pods coexisting, a new writer can send a splice to a per-node hub still running the old +image. That owner does not decode `$sd`, merges the marker object into a string-typed field, and the +merged node then **fails to deserialize** — the write is NACKed `Deserialization` and never commits. +Loud and non-destructive, and it self-heals: the mirror never advanced, so the next splice is +computed against the same base and lands the whole missing span once the owner is upgraded. + +That failure mode is deliberate. The obvious alternative — carrying splices in a new sibling property +the old owner would simply ignore — degrades instead into an owner that **acks a write it only +partly applied**, which is the acked-write-loss class of [#648](https://github.com/Systemorph/MeshWeaver/issues/648). +A visibly stalled field beats a silently lost one. + ## Idle release — quiet paths give their upstream back A read entry does **not** live for the process lifetime. Like the write-side serial diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-a-long-answer-no-longer-costs-more-than-it-says.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-a-long-answer-no-longer-costs-more-than-it-says.md new file mode 100644 index 0000000000..025f6872d0 --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-13-a-long-answer-no-longer-costs-more-than-it-says.md @@ -0,0 +1,39 @@ +--- +Name: A long answer no longer costs the portal more than it says +Category: Fix +Description: While an agent typed out a reply, the portal re-sent the entire answer-so-far on every refresh — so a long answer cost megabytes and slowed everything sharing the machine. It now sends only the new words. +Icon: Sparkle +Order: -20260813 +--- + +# A long answer no longer costs the portal more than it says + +When an agent writes you a reply, the text appears a few words at a time. Behind that, roughly ten +times a second, the portal saved the answer so far so that the page you are watching — and anyone +else watching the same conversation — stays up to date. + +The problem was *what* it saved. Every one of those saves carried the **whole answer from the +beginning**, not just the words that had arrived since the last one. Early in a reply that costs +nothing. By the end of a long one it means shipping the entire text again, and again, and again. A +page-length answer ended up moving several megabytes through the portal to deliver twenty kilobytes +of writing — and the cost grew with the *square* of the answer's length, so an answer twice as long +was four times as expensive. + +Nobody saw a wrong word on screen; the text was always correct. What people saw was everything +*else* getting slower while a long reply was in progress. Handling those repeated copies is real +work for the machine, and it competes with every other thing the portal is doing at that moment — +other people's pages, background jobs, the health checks that decide whether the service is +answering at all. + +Now a save carries only the part that changed. The portal already has the earlier text; it does not +need to be told again. In a measured run, a twenty-thousand-character answer went from **3.8 MB of +saved updates to 0.07 MB** — about fifty-five times less — and the cost now grows in step with the +answer's length instead of racing ahead of it. + +Two things are deliberately unchanged. The finished text is identical to the character — this is a +change in how the update travels, never in what it says. And if two things somehow write to the same +text at the same moment, the portal refuses the second one and has it start again from the current +text rather than guessing where the new words belong; correctness wins over saving a round trip. + +The same saving applies anywhere the portal stores a large piece of text that grows or is edited in +place — long documents, generated pages, notes — not just agent replies. diff --git a/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs b/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs index 6a529ba5b3..97490a39b4 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs @@ -1416,6 +1416,12 @@ or MeshNodeErrorCode.Conflict /// (e.g. via HandlePatchDataRequest's recursive merge), reproduces /// . Keys in current and missing in updated /// emit null (RFC 7396 remove). Equal values emit nothing. + /// 🚨 A changed large string leaf is emitted as a SPLICE + /// () rather than the whole new value, so a field that + /// grows one chunk at a time — an agent response cell streaming tokens — costs + /// O(chunk) per write instead of O(length). Below + /// , or when the splice would not + /// actually be smaller, the full value is emitted exactly as before. /// private static System.Text.Json.Nodes.JsonObject ComputeMergePatchDiff( System.Text.Json.Nodes.JsonObject current, @@ -1433,8 +1439,22 @@ private static System.Text.Json.Nodes.JsonObject ComputeMergePatchDiff( patch[key] = sub; continue; } - if (!System.Text.Json.Nodes.JsonNode.DeepEquals(currentValue, updatedValue)) - patch[key] = updatedValue?.DeepClone(); + if (System.Text.Json.Nodes.JsonNode.DeepEquals(currentValue, updatedValue)) + continue; + // Big string that changed on one side → ship only the splice. The matching base + // fingerprint is written by MeshNodePatchMerge.ExtractBaseValues, which recognises + // this shape; the owner applies the splice only once that fingerprint proves its + // live text is the text this diff was computed against. + if (currentValue is System.Text.Json.Nodes.JsonValue cv + && cv.TryGetValue(out var oldStr) + && updatedValue is System.Text.Json.Nodes.JsonValue uv + && uv.TryGetValue(out var newStr) + && PatchStringSplice.TryEncode(oldStr, newStr, out var spliced)) + { + patch[key] = spliced; + continue; + } + patch[key] = updatedValue?.DeepClone(); } // Keys removed in updated → emit null per RFC 7396. foreach (var (key, _) in current) diff --git a/test/MeshWeaver.Data.Test/PatchStringSpliceTest.cs b/test/MeshWeaver.Data.Test/PatchStringSpliceTest.cs new file mode 100644 index 0000000000..8d614e828b --- /dev/null +++ b/test/MeshWeaver.Data.Test/PatchStringSpliceTest.cs @@ -0,0 +1,288 @@ +using System.Text.Json.Nodes; +using MeshWeaver.Data.Serialization; +using Xunit; + +namespace MeshWeaver.Data.Test; + +/// +/// Pins and its resolution inside +/// — the encoding that makes a cross-hub write to a GROWING +/// string field cost O(chunk) instead of O(length). +/// +/// Two properties carry the whole design and both are asserted here: +/// +/// Exactness. When the owner's live text is the text the writer diffed against, +/// applying the splice yields the byte-identical string a full-value patch would have +/// written. Anything less would be visible in chat, the most user-visible surface there is. +/// No blind splicing. When the owner has moved on, the splice's offsets no longer +/// address the text they were computed for — so the leaf is REFUSED rather than applied. +/// Two mirrors splicing the same string concurrently therefore cannot corrupt it. +/// +/// +public class PatchStringSpliceTest +{ + // Long enough to clear PatchStringSplice.MinSpliceLength so the splice path engages. + private static string Big(string tail = "") => + string.Concat(Enumerable.Repeat("The quick brown fox jumps over the lazy dog. ", 40)) + tail; + + private static JsonObject Str(string key, string value) => new() { [key] = JsonValue.Create(value) }; + + // ---- encoding --------------------------------------------------------------------------- + + [Fact] + public void AnAppendToABigString_EncodesAsASplice_NotTheWholeValue() + { + var baseText = Big(); + var newText = baseText + "one more chunk"; + + PatchStringSplice.TryEncode(baseText, newText, out var encoded).Should().BeTrue(); + + var wire = encoded!.ToJsonString(); + wire.Should().Contain("one more chunk"); + wire.Length.Should().BeLessThan(newText.Length / 4, + "the splice must carry the appended chunk, not the whole grown string"); + } + + [Fact] + public void ASmallString_StaysOnTheWholeValuePath() + { + PatchStringSplice.TryEncode("hello", "hello world", out var encoded).Should().BeFalse( + "below MinSpliceLength the whole value is cheaper AND keeps the full-fidelity rebase"); + Assert.Null(encoded); + } + + [Fact] + public void ANearTotalRewrite_StaysOnTheWholeValuePath() + { + var a = Big("aaaa"); + var b = string.Concat(Enumerable.Repeat("Something else entirely, word by word. ", 40)); + + PatchStringSplice.TryEncode(a, b, out _).Should().BeFalse( + "a splice that is no smaller than the value buys nothing and loses fidelity"); + } + + [Fact] + public void TheDecoderIsStrict_SoOrdinaryContentIsNeverMistakenForASplice() + { + PatchStringSplice.TryDecode(JsonValue.Create("plain text"), out _).Should().BeFalse(); + PatchStringSplice.TryDecode(new JsonObject { ["text"] = "x" }, out _).Should().BeFalse(); + // Right marker, wrong arity / wrong element types / extra properties. + PatchStringSplice.TryDecode(new JsonObject { [PatchStringSplice.Marker] = new JsonArray(1, 2) }, out _) + .Should().BeFalse(); + PatchStringSplice.TryDecode(new JsonObject { [PatchStringSplice.Marker] = new JsonArray(1, 2, 3) }, out _) + .Should().BeFalse(); + PatchStringSplice.TryDecode( + new JsonObject { [PatchStringSplice.Marker] = new JsonArray(1, 2, "x"), ["other"] = 1 }, out _) + .Should().BeFalse(); + PatchStringSplice.TryDecode(new JsonObject { [PatchStringSplice.Marker] = new JsonArray(-1, 0, "x") }, out _) + .Should().BeFalse(); + } + + // ---- owner-side resolution -------------------------------------------------------------- + + [Fact] + public void AgainstTheBaseItWasComputedFrom_TheSpliceReproducesTheFullValueExactly() + { + var baseText = Big(); + var newText = baseText + "…and the tail the writer added"; + + PatchStringSplice.TryEncode(baseText, newText, out var encoded).Should().BeTrue(); + var live = Str("body", baseText); + var patch = new JsonObject { ["body"] = encoded!.DeepClone() }; + var baseValues = new JsonObject { ["body"] = PatchStringSplice.EncodeBase(baseText) }; + + MeshNodePatchMerge.Apply(live, patch, baseValues); + + live["body"]!.GetValue().Should().Be(newText, + "the spliced write must land byte-for-byte the value the whole-string patch would have"); + } + + [Fact] + public void ManySequentialSplices_ReproduceTheWholeStreamedText() + { + // The chat shape: one cell, many appends, each diffed against the state the previous + // append produced. The end state must equal the plain concatenation. + var chunks = Enumerable.Range(0, 200).Select(i => $"chunk-{i:D4} of the streamed answer. ").ToArray(); + var live = Str("text", string.Empty); + var expected = string.Empty; + + foreach (var chunk in chunks) + { + var previous = live["text"]!.GetValue(); + var next = previous + chunk; + expected = next; + + JsonObject patch; + JsonObject baseValues; + if (PatchStringSplice.TryEncode(previous, next, out var encoded)) + { + patch = new JsonObject { ["text"] = encoded!.DeepClone() }; + baseValues = new JsonObject { ["text"] = PatchStringSplice.EncodeBase(previous) }; + } + else + { + patch = Str("text", next); + baseValues = Str("text", previous); + } + MeshNodePatchMerge.Apply(live, patch, baseValues); + } + + live["text"]!.GetValue().Should().Be(expected); + expected.Length.Should().BeGreaterThan(PatchStringSplice.MinSpliceLength * 4, + "the run must actually have crossed into the splice regime"); + } + + [Fact] + public void WhenTheOwnerMovedSinceTheWritersBase_TheSpliceIsRefused_NotAppliedBlind() + { + var writerBase = Big(); + var writerNew = writerBase + "WRITER-TAIL"; + PatchStringSplice.TryEncode(writerBase, writerNew, out var encoded).Should().BeTrue(); + + // The owner's text moved on between the writer's read and the patch's arrival. + var ownerLive = "PREFIX-THE-WRITER-NEVER-SAW " + writerBase + "OWNER-TAIL"; + var live = Str("body", ownerLive); + var patch = new JsonObject { ["body"] = encoded!.DeepClone() }; + var baseValues = new JsonObject { ["body"] = PatchStringSplice.EncodeBase(writerBase) }; + + var refused = new List(); + MeshNodePatchMerge.Apply(live, patch, baseValues, refused.Add); + + // The fingerprint did not vouch for the offsets — the owner must report a conflict. + refused.Should().Equal("body"); + live["body"]!.GetValue().Should().Be(ownerLive, + "the owner keeps its newer value; the writer is NACKed Conflict and re-diffs"); + } + + [Fact] + public void TwoMirrorsSplicingTheSameStringConcurrently_CannotCorruptIt() + { + // Both writers read the same base, then each appends its own tail. Whichever lands first + // wins outright; the second is refused because the live text no longer matches its base. + var shared = Big(); + PatchStringSplice.TryEncode(shared, shared + "AAA", out var first).Should().BeTrue(); + PatchStringSplice.TryEncode(shared, shared + "BBB", out var second).Should().BeTrue(); + var baseValues = new JsonObject { ["body"] = PatchStringSplice.EncodeBase(shared) }; + + var live = Str("body", shared); + MeshNodePatchMerge.Apply(live, new JsonObject { ["body"] = first!.DeepClone() }, baseValues); + live["body"]!.GetValue().Should().Be(shared + "AAA"); + + var refused = new List(); + MeshNodePatchMerge.Apply(live, new JsonObject { ["body"] = second!.DeepClone() }, baseValues, refused.Add); + + refused.Should().Equal("body"); + live["body"]!.GetValue().Should().Be(shared + "AAA", + "no interleaving at unverified offsets — the loser is refused whole and re-diffs"); + } + + [Fact] + public void ALiveValueThatIsNoLongerAString_RefusesRatherThanWritingTheMarker() + { + PatchStringSplice.TryEncode(Big(), Big("tail"), out var encoded).Should().BeTrue(); + var live = new JsonObject { ["body"] = 42 }; + + var refused = new List(); + MeshNodePatchMerge.Apply(live, new JsonObject { ["body"] = encoded!.DeepClone() }, + new JsonObject { ["body"] = PatchStringSplice.EncodeBase(Big()) }, refused.Add); + + refused.Should().Equal("body"); + live["body"]!.GetValue().Should().Be(42, "the marker object must never be written as a value"); + } + + [Fact] + public void ASpliceWithNoBaseSignal_IsRefused_NeverAppliedAtAnUnverifiedOffset() + { + PatchStringSplice.TryEncode(Big(), Big("tail"), out var encoded).Should().BeTrue(); + var live = Str("body", Big()); + + var refused = new List(); + MeshNodePatchMerge.Apply(live, new JsonObject { ["body"] = encoded!.DeepClone() }, + baseValues: null, refused.Add); + + refused.Should().Equal("body"); + live["body"]!.GetValue().Should().Be(Big()); + } + + [Fact] + public void AFullBaseStringCarriedWithASplice_StillGetsTheThreeWayRebase() + { + // Defensive path: a sender that splices but ships the whole base keeps today's semantics — + // a DISJOINT concurrent edit on the owner merges rather than being refused. + var baseText = Big(); + PatchStringSplice.TryEncode(baseText, baseText + "WRITER", out var encoded).Should().BeTrue(); + var ownerLive = "OWNER-PREFIX " + baseText; + + var live = Str("body", ownerLive); + var refused = new List(); + MeshNodePatchMerge.Apply(live, new JsonObject { ["body"] = encoded!.DeepClone() }, + Str("body", baseText), refused.Add); + + refused.Should().BeEmpty(); + live["body"]!.GetValue().Should().Be("OWNER-PREFIX " + baseText + "WRITER"); + } + + // ---- the base-less RFC 7396 fallback ----------------------------------------------------- + + [Fact] + public void OnTheBaseLessMergePath_ASpliceIsDecoded_NeverWrittenAsAMarker() + { + // DataExtensions.MergePatchRecursive is reached when no BaseValues were carried (an MCP + // one-off, a legacy sender). It has no conflict signal at all, so the splice replays onto + // CURRENT — the same semantics as StringDeltaPatch.Apply / EntityDelta.Apply. + var baseText = Big(); + PatchStringSplice.TryEncode(baseText, baseText + "tail", out var encoded).Should().BeTrue(); + + var current = Str("body", baseText); + MergePatchRecursiveViaOwner(current, new JsonObject { ["body"] = encoded!.DeepClone() }); + + current["body"]!.GetValue().Should().Be(baseText + "tail"); + current["body"]!.ToJsonString().Should().NotContain(PatchStringSplice.Marker); + } + + [Fact] + public void OnTheBaseLessMergePath_ASpliceOntoANonStringLeavesItAlone() + { + PatchStringSplice.TryEncode(Big(), Big("tail"), out var encoded).Should().BeTrue(); + var current = new JsonObject { ["body"] = 42 }; + + MergePatchRecursiveViaOwner(current, new JsonObject { ["body"] = encoded!.DeepClone() }); + + current["body"]!.GetValue().Should().Be(42, + "with nothing to splice onto, the live value stands — the marker is never written"); + } + + // MergePatchRecursive is internal to MeshWeaver.Data; this test assembly sees it via + // InternalsVisibleTo, so call it directly rather than reimplementing the semantics here. + private static void MergePatchRecursiveViaOwner(JsonObject current, JsonObject patch) => + DataExtensions.MergePatchRecursive(current, patch); + + // ---- base-value extraction --------------------------------------------------------------- + + [Fact] + public void ExtractBaseValues_ShipsAFingerprintForASplicedLeaf_NotTheWholeOldString() + { + var baseText = Big(); + PatchStringSplice.TryEncode(baseText, baseText + "tail", out var encoded).Should().BeTrue(); + var patch = new JsonObject { ["body"] = encoded!.DeepClone() }; + + var baseValues = MeshNodePatchMerge.ExtractBaseValues(Str("body", baseText), patch); + + Assert.NotNull(baseValues); + var wire = baseValues!.ToJsonString(); + wire.Should().NotContain("quick brown fox", "the base half must not re-ship the previous text"); + wire.Length.Should().BeLessThan(64); + PatchStringSplice.TryDecodeBase(baseValues["body"], out var length, out _).Should().BeTrue(); + length.Should().Be(baseText.Length); + } + + [Fact] + public void ExtractBaseValues_StillShipsTheWholeOldValueForANonSplicedLeaf() + { + var patch = Str("name", "after"); + var baseValues = MeshNodePatchMerge.ExtractBaseValues(Str("name", "before"), patch); + + baseValues!["name"]!.GetValue().Should().Be("before", + "arrays and small scalars keep the full base — MergeArray genuinely consumes it"); + } +} diff --git a/test/MeshWeaver.Threading.Test/StreamedTextIntegrityTest.cs b/test/MeshWeaver.Threading.Test/StreamedTextIntegrityTest.cs new file mode 100644 index 0000000000..9703e7f806 --- /dev/null +++ b/test/MeshWeaver.Threading.Test/StreamedTextIntegrityTest.cs @@ -0,0 +1,228 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reactive.Linq; +using System.Reactive.Threading.Tasks; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.AI; +using MeshWeaver.Data; +using MeshWeaver.Graph; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Messaging; +using Microsoft.Agents.AI; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.DependencyInjection; +using Xunit; +using MeshThread = MeshWeaver.AI.Thread; + +namespace MeshWeaver.Threading.Test; + +/// +/// Chat is the most user-visible surface in the product, and the streaming write now ships a SPLICE +/// rather than the cell's whole text (see PatchStringSplice). This test runs a REAL round +/// through ThreadExecution — long enough that Sample(100 ms) produces many ticks and +/// the text crosses PatchStringSplice.MinSpliceLength — and pins the three things a user +/// would notice if a splice ever landed at the wrong offset: +/// +/// the finished text is EXACTLY what the model streamed, byte for byte; +/// every intermediate state a subscriber sees is a PREFIX of that text — never a scrambled, +/// duplicated or truncated middle; +/// a subscriber that joins LATE sees the whole text so far, not just the last chunk. +/// +/// The wire-cost measurement that motivates the change is +/// ; it is kept separate because +/// Sample(100 ms) makes the tick count here a function of wall-clock timing. +/// +public class StreamedTextIntegrityTest(ITestOutputHelper output) : MonolithMeshTestBase(output) +{ + private const string ContextPath = "User/TestUser"; + + /// + /// Longest state exempt from the prefix check. The framework writes short status placeholders + /// before the first token ("Allocating agent…" = 19 chars, "Generating response…" = 22); the + /// model's own chunks are an order of magnitude longer. Bounding the exemption by length rather + /// than by matching those strings keeps it from ever excusing a corrupted model state — and the + /// strings themselves are localizable, so matching them would be brittle as well as unsafe. + /// + private const int PlaceholderMaxLength = 64; + + protected override MeshBuilder ConfigureMesh(MeshBuilder builder) + => base.ConfigureMesh(builder) + .ConfigureServices(services => + { + services.AddSingleton(new LongAnswerChatClientFactory()); + return services; + }) + // A streaming round leaves in-flight DataChangeRequest callbacks past the default + // 500 ms quiesce budget; the round is the thing under test, so give teardown room. + .ConfigureHub(c => c.WithQuiesceTimeout(TimeSpan.FromSeconds(15))) + .AddAI() + .AddSampleUsers(); + + protected override MessageHubConfiguration ConfigureClient(MessageHubConfiguration configuration) + { + configuration.TypeRegistry.AddAITypes(); + return base.ConfigureClient(configuration).AddData(); + } + + [Fact] + public async Task ARealStreamedRound_LandsTheExactText_AndEveryStateSeenIsAPrefixOfIt() + { + var client = GetClient(); + var workspace = client.GetWorkspace(); + var threadNode = ThreadNodeType.BuildThreadNode(ContextPath, "Streaming integrity", "TestUser"); + var created = await client.Observe(new CreateNodeRequest(threadNode), o => o.WithTarget(Mesh.Address)) + .Should().Within(60.Seconds()).Emit(); + created.Message.Success.Should().BeTrue(created.Message.Error ?? ""); + var threadPath = created.Message.Node!.Path!; + + // Resolve the response cell as soon as the round allocates it, so we can watch the text grow + // from (nearly) the start rather than only seeing the finished value. + var cellPathTask = workspace.GetMeshNodeStream(threadPath) + .Select(n => n.ContentAs(client.JsonSerializerOptions)) + .Where(t => !string.IsNullOrEmpty(t?.ActiveMessageId)) + .Select(t => $"{threadPath}/{t!.ActiveMessageId}") + .FirstAsync().Timeout(30.Seconds()).ToTask(); + + var roundTask = ThreadFlow + .SubmitAndWait(client, threadPath, "go", contextPath: ContextPath, timeout: 60.Seconds()) + .FirstAsync().ToTask(); + + var cellPath = await cellPathTask; + Output.WriteLine($"response cell: {cellPath}"); + + var observed = new List(); + using (workspace.GetMeshNodeStream(cellPath) + .Select(n => n.ContentAs(client.JsonSerializerOptions)) + .Where(m => m is not null) + .Subscribe(m => { lock (observed) observed.Add(m!.Text); })) + { + await roundTask; + } + + var final = await ThreadFlow.ReadMessage(client, threadPath, cellPath.Split('/')[^1], + m => m.CompletedAt != null && !string.IsNullOrEmpty(m.Text), + timeout: 60.Seconds()).Should().Within(60.Seconds()).Emit(); + + // 1. Byte-identical to what the model actually streamed. + final.Text.Should().Be(LongAnswerChatClient.Answer, + "the spliced streaming write must land exactly the text the model produced"); + LongAnswerChatClient.Answer.Length.Should() + .BeGreaterThan(MeshWeaver.Data.Serialization.PatchStringSplice.MinSpliceLength * 3, + "the answer must be long enough that the splice path actually engaged"); + + // 2. Every state carrying model text was a PREFIX of the answer, and they grew monotonically. + // A splice applied at a wrong offset shows up here immediately — as a state that is not a + // prefix (scrambled / duplicated middle) or one that shrinks. + // + // Exempt: the framework's own pre-stream status text ("Allocating agent…", + // "Generating response…"), which is not model output at all. The exemption is bounded by + // LENGTH rather than by matching those strings — they are localizable, and a length rule + // cannot accidentally excuse a long corrupted state. + List snapshot; + lock (observed) snapshot = [.. observed]; + var nonEmpty = snapshot.Where(t => !string.IsNullOrEmpty(t)).ToList(); + var modelText = nonEmpty.Where(t => t.Length > PlaceholderMaxLength).ToList(); + var placeholders = nonEmpty.Where(t => t.Length <= PlaceholderMaxLength).ToList(); + + Output.WriteLine($"observed {snapshot.Count} emissions; {modelText.Count} carrying model text; " + + $"lengths {string.Join(",", modelText.Select(t => t.Length).Distinct().Take(12))}…"); + foreach (var p in placeholders.Distinct()) + Output.WriteLine($"pre-stream placeholder (len {p.Length}): {p}"); + + modelText.Should().NotBeEmpty("the round must have been observed while streaming"); + var offenders = modelText + .Where(t => !LongAnswerChatClient.Answer.StartsWith(t, StringComparison.Ordinal)) + .ToList(); + foreach (var o in offenders.Take(5)) + Output.WriteLine($"NOT A PREFIX (len {o.Length}): >>>{o[..Math.Min(200, o.Length)]}<<<"); + offenders.Should().BeEmpty("every state a viewer sees must be a prefix of the final answer"); + modelText.Select(t => t.Length).Should().BeInAscendingOrder( + "the streamed text only ever grows — a splice can never shorten or rewrite it"); + modelText.Any(t => t.Length > MeshWeaver.Data.Serialization.PatchStringSplice.MinSpliceLength) + .Should().BeTrue("states past MinSpliceLength are the ones written as splices"); + + // 3. A subscriber that joins after the fact sees everything, not just the last splice. + var lateClient = GetClient(); + var late = await lateClient.GetWorkspace().GetMeshNodeStream(cellPath) + .Select(n => n.ContentAs(lateClient.JsonSerializerOptions)) + .Where(m => m is not null && m.CompletedAt != null) + .FirstAsync().Timeout(60.Seconds()).ToTask(); + late!.Text.Should().Be(LongAnswerChatClient.Answer, + "a late subscriber reads the owner's committed state — the whole text, not a fragment"); + } + + #region Long-answer chat client — streams past MinSpliceLength over many Sample(100ms) ticks + + private sealed class LongAnswerChatClient : IChatClient + { + private const string Paragraph = + "The cedent retains the first band of loss and the reinsurer attaches above it, so the " + + "treaty's economics turn on where that attachment point sits relative to the modelled " + + "severity curve; move it down and the premium rises faster than the expected recovery. "; + + private const int ChunkCount = 24; + + // Declared before Answer: static field initializers run in textual order. + private static readonly string[] Chunks = Enumerable.Range(0, ChunkCount) + .Select(i => $"[{i:D2}] {Paragraph}") + .ToArray(); + + /// ~6 kB over 24 chunks — several times MinSpliceLength, ~3 s of streaming. + internal static readonly string Answer = string.Concat(Chunks); + + public ChatClientMetadata Metadata => new("LongAnswerProvider"); + + public Task GetResponseAsync( + IEnumerable messages, ChatOptions? options = null, + CancellationToken cancellationToken = default) + => Task.FromResult(new ChatResponse(new ChatMessage(ChatRole.Assistant, Answer))); + + public async IAsyncEnumerable GetStreamingResponseAsync( + IEnumerable messages, ChatOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + foreach (var chunk in Chunks) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return new ChatResponseUpdate(ChatRole.Assistant, chunk); + // > Sample(100 ms) so most chunks produce their own write rather than coalescing. + await Task.Delay(130, cancellationToken); + } + } + + public object? GetService(Type serviceType, object? serviceKey = null) => + serviceType == typeof(IChatClient) ? this : null; + + public void Dispose() { } + } + + private sealed class LongAnswerChatClientFactory : IChatClientFactory + { + public string Name => "LongAnswerFactory"; + public IReadOnlyList Models => ["long-answer-model"]; + public int Order => 0; + + public ChatClientAgent CreateAgent( + AgentConfiguration config, IAgentChat chat, + IReadOnlyDictionary existingAgents, + IReadOnlyList hierarchyAgents, + string? modelName = null) + => new(chatClient: new LongAnswerChatClient(), + instructions: config.Instructions ?? "You are a verbose test assistant.", + name: config.Id, description: config.Description ?? config.Id, + tools: [], loggerFactory: null, services: null); + + public Task CreateAgentAsync( + AgentConfiguration config, IAgentChat chat, + IReadOnlyDictionary existingAgents, + IReadOnlyList hierarchyAgents, + string? modelName = null) + => Task.FromResult(CreateAgent(config, chat, existingAgents, hierarchyAgents, modelName)); + } + + #endregion +} diff --git a/test/MeshWeaver.Threading.Test/StreamingCellWriteByteCountTest.cs b/test/MeshWeaver.Threading.Test/StreamingCellWriteByteCountTest.cs new file mode 100644 index 0000000000..616aaa4c62 --- /dev/null +++ b/test/MeshWeaver.Threading.Test/StreamingCellWriteByteCountTest.cs @@ -0,0 +1,214 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Reactive.Linq; +using System.Reactive.Threading.Tasks; +using System.Threading.Tasks; +using MeshWeaver.AI; +using MeshWeaver.Data; +using MeshWeaver.Graph; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Messaging; +using Xunit; + +namespace MeshWeaver.Threading.Test; + +/// +/// The chat half of the quadratic-write class (#1284 / #1351): a streaming agent response rewrites +/// its cell's WHOLE text on every Sample(100 ms) tick, so one reply costs +/// O(ticks × final length) on the wire — megabytes for a 20 kB answer, every byte of it +/// a fresh Large-Object-Heap allocation on both the writer and the owner. +/// +/// Satellites cannot fix this the way they fixed the activity log: a streaming cell must show +/// all the text so far, so the cell's own value legitimately grows. What does NOT have to grow is +/// the PATCH — the owner already has everything but the newest chunk. This test measures the wire +/// cost of the real cross-hub write path (GetMeshNodeStream(path).Update → the shared stream +/// cache → PatchDataRequest → the owner's three-way merge) by counting the actual +/// PatchDataRequest payloads as the owning node hub receives them. +/// +/// The measurement is deliberately NOT driven through a model: Sample(100 ms) makes the +/// tick count a function of wall-clock timing, which would make the two-size comparison noise. Here +/// the tick count is exact, so the two columns differ only in the thing under test. The end-to-end +/// correctness of a real streamed round — final text byte-identical, late subscriber sees +/// everything — is . +/// +public class StreamingCellWriteByteCountTest(ITestOutputHelper output) : MonolithMeshTestBase(output) +{ + /// Characters flushed per tick — roughly what 100 ms of token streaming carries. + private const int ChunkChars = 100; + + private const int SmallAnswerChars = 5_000; + private const int LargeAnswerChars = 20_000; // the benchmark quoted in #1284 / #1351 + + /// + /// Wire budget per character of the finished answer. A linear write path pays the answer ONCE + /// (plus a per-tick constant: the cell's other fields, the base fingerprint, JSON framing, and + /// the sub-MinSpliceLength warm-up ticks that still ship the whole value). Under the + /// quadratic shape a 20 kB answer costs ~200 bytes per character, so this fails by two + /// orders of magnitude before the fix and passes with room after it. + /// + private const int WireBytesPerAnswerCharBudget = 40; + + private readonly PatchWireRecorder recorder = new(); + + protected override MeshBuilder ConfigureMesh(MeshBuilder builder) + => base.ConfigureMesh(builder) + // Passive recorder on EVERY node hub: it measures the request and hands the delivery + // back UNPROCESSED, so the owner's real PatchDataRequest handler still runs. Registered + // through ConfigureDefaultNodeHub, which composes BEFORE the node's own config (and thus + // before AddData), so it sees each request first in the rule chain. + .ConfigureDefaultNodeHub(config => config.WithHandler((_, delivery) => + { + recorder.Record(delivery.Target?.ToString(), delivery.Message); + return delivery; + })) + .AddAI() + .AddSampleUsers(); + + protected override MessageHubConfiguration ConfigureClient(MessageHubConfiguration configuration) + { + configuration.TypeRegistry.AddAITypes(); + return base.ConfigureClient(configuration); + } + + [Fact] + public async Task StreamingACellsText_CostsTheAnswerAboutOnce_NotOncePerTick() + { + var small = await StreamAnswerIntoCell("bytes-small", SmallAnswerChars); + var large = await StreamAnswerIntoCell("bytes-large", LargeAnswerChars); + + Output.WriteLine( + $"ticks: {SmallAnswerChars:N0} chars -> {small.Writes}; {LargeAnswerChars:N0} chars -> {large.Writes}"); + Output.WriteLine( + $"patch bytes: {SmallAnswerChars:N0} chars -> {small.PatchBytes:N0}; {LargeAnswerChars:N0} chars -> {large.PatchBytes:N0}"); + Output.WriteLine( + $"base bytes: {SmallAnswerChars:N0} chars -> {small.BaseBytes:N0}; {LargeAnswerChars:N0} chars -> {large.BaseBytes:N0}"); + Output.WriteLine( + $"TOTAL wire bytes: {SmallAnswerChars:N0} chars -> {small.TotalBytes:N0} ({small.TotalBytes / 1024.0 / 1024.0:F3} MB);" + + $" {LargeAnswerChars:N0} chars -> {large.TotalBytes:N0} ({large.TotalBytes / 1024.0 / 1024.0:F3} MB)"); + Output.WriteLine( + $"bytes per answer char: {(double)small.TotalBytes / SmallAnswerChars:F1} -> {(double)large.TotalBytes / LargeAnswerChars:F1}" + + $" (4x the answer length; a quadratic path multiplies this by 4 too)"); + + // The tick count is exact by construction — this guards the measurement itself. + small.Writes.Should().Be(SmallAnswerChars / ChunkChars); + large.Writes.Should().Be(LargeAnswerChars / ChunkChars); + large.FinalTextLength.Should().Be(LargeAnswerChars); + + // 1. Absolute: the round must cost the answer a small constant number of times, not once + // per tick. 20 kB × 40 = 800 kB, against ~4 MB for the whole-value shape. + large.TotalBytes.Should().BeLessThan((long)LargeAnswerChars * WireBytesPerAnswerCharBudget, + "a streaming round must ship the answer about once, not once per tick"); + + // 2. Relative: quadrupling the answer must not quadruple the cost PER CHARACTER. This is + // the shape assertion — it holds whatever the per-tick constant happens to be. + var smallPerChar = (double)small.TotalBytes / SmallAnswerChars; + var largePerChar = (double)large.TotalBytes / LargeAnswerChars; + largePerChar.Should().BeLessThan(smallPerChar * 1.5, + "wire cost per character of the answer must not grow with the length of the answer"); + } + + private sealed record StreamMeasurement(int Writes, long PatchBytes, long BaseBytes, int FinalTextLength) + { + public long TotalBytes => PatchBytes + BaseBytes; + } + + /// + /// Drives the production streaming write: a response cell whose Text grows one chunk at a time + /// through the cross-hub stream.Update — exactly what + /// ThreadExecution.PushToResponseMessage does on every sampled tick, minus the model. + /// Each write is awaited so the tick count is exact and the per-path serial write queue is not + /// coalescing anything behind our back. + /// + private async Task StreamAnswerIntoCell(string cellId, int answerChars) + { + var threadPath = $"{TestPartition}/_Thread/{cellId}"; + var cellPath = $"{threadPath}/resp"; + + await NodeFactory.CreateNode(new MeshNode("resp", threadPath) + { + NodeType = ThreadMessageNodeType.NodeType, + MainNode = TestPartition, + Content = new ThreadMessage + { + Role = "assistant", + Text = string.Empty, + Type = ThreadMessageType.AgentResponse, + Status = ThreadMessageStatus.Streaming + } + }).Should().Emit(); + + recorder.Reset(); + var workspace = Mesh.GetWorkspace(); + var answer = Answer(answerChars); + var ticks = answerChars / ChunkChars; + + for (var tick = 1; tick <= ticks; tick++) + { + var soFar = answer[..(tick * ChunkChars)]; + await workspace.GetMeshNodeStream(cellPath) + .Update(current => current with { Text = soFar }) + .FirstAsync().Timeout(30.Seconds()).ToTask(); + } + + // Read the settled cell back off the shared handle: the per-path write queue is serial, so + // once the last write's text is visible every earlier one has landed. + var settled = await workspace.GetMeshNodeStream(cellPath) + .Select(n => n.ContentAs(Mesh.JsonSerializerOptions)) + .Where(m => m is not null && m.Text.Length == answerChars) + .FirstAsync().Timeout(60.Seconds()).ToTask(); + + settled!.Text.Should().Be(answer, + "a spliced write must reproduce the whole-value result byte for byte"); + + var wire = recorder.For(cellPath); + return new StreamMeasurement(wire.Writes, wire.PatchBytes, wire.BaseBytes, settled.Text.Length); + } + + /// Deterministic prose-shaped filler — realistic JSON escaping, no random noise. + private static string Answer(int chars) + { + const string sentence = + "The reinsurance treaty allocates losses across layers, and the cedent retains the first band. "; + var text = string.Concat(Enumerable.Repeat(sentence, chars / sentence.Length + 1)); + return text[..chars]; + } + + /// + /// Sums the bytes of the PatchDataRequests each owner hub actually received, keyed by + /// target. Instance state owned by the test — never static (a process-wide counter would bleed + /// across tests exactly like the caches AGENTS.md bans). + /// + private sealed class PatchWireRecorder + { + private readonly ConcurrentDictionary byTarget = new(); + + public void Record(string? target, PatchDataRequest request) + { + if (target is null) + return; + var patch = request.Patch.Content?.Length ?? 0; + var baseValues = request.BaseValues?.Content?.Length ?? 0; + byTarget.AddOrUpdate(target, + _ => new Wire(1, patch, baseValues), + (_, w) => new Wire(w.Writes + 1, w.PatchBytes + patch, w.BaseBytes + baseValues)); + } + + public void Reset() => byTarget.Clear(); + + /// Everything recorded for the node at , however addressed. + public Wire For(string path) + { + var matches = byTarget + .Where(kv => kv.Key.EndsWith(path, StringComparison.OrdinalIgnoreCase)) + .Select(kv => kv.Value) + .ToList(); + return matches.Aggregate(new Wire(0, 0, 0), + (a, b) => new Wire(a.Writes + b.Writes, a.PatchBytes + b.PatchBytes, a.BaseBytes + b.BaseBytes)); + } + + public sealed record Wire(int Writes, long PatchBytes, long BaseBytes); + } +} From bdc218cb7704067b37b7ec3bd6cb9d6860a69dda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Thu, 13 Aug 2026 13:24:40 +0200 Subject: [PATCH 2/2] review: count UTF-8 bytes, and create the thread through its own namespace MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both from the Copilot review, both worth taking. PatchWireRecorder called its metric "bytes" while summing string.Length — a UTF-16 char count. For this test's ASCII prose the two coincide, so the conclusion never moved, but the payload travels as UTF-8 and a char count silently understates anything non-ASCII, which real chat is full of. Now Encoding.UTF8.GetByteCount, and both columns were re-measured under it: 5,000 chars / 50 ticks 20,000 chars / 200 ticks before wire bytes 257,788 4,031,144 after wire bytes 23,795 69,893 before bytes/char 51.6 201.6 (grows 4x) after bytes/char 4.8 3.5 (falls) StreamedTextIntegrityTest created its thread against Mesh.Address. That is the root mesh hub; the majority of the thread tests (ThreadCreationTest, the Orleans thread suite, DelegationCompletionTest) target the context namespace, which is also what a real client does and what keeps the test representative. Switched to new Address(ContextPath). Co-Authored-By: Claude Opus 5 (1M context) --- .../StreamedTextIntegrityTest.cs | 5 ++++- .../StreamingCellWriteByteCountTest.cs | 9 +++++++-- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/test/MeshWeaver.Threading.Test/StreamedTextIntegrityTest.cs b/test/MeshWeaver.Threading.Test/StreamedTextIntegrityTest.cs index 9703e7f806..3341ba6fb5 100644 --- a/test/MeshWeaver.Threading.Test/StreamedTextIntegrityTest.cs +++ b/test/MeshWeaver.Threading.Test/StreamedTextIntegrityTest.cs @@ -74,7 +74,10 @@ public async Task ARealStreamedRound_LandsTheExactText_AndEveryStateSeenIsAPrefi var client = GetClient(); var workspace = client.GetWorkspace(); var threadNode = ThreadNodeType.BuildThreadNode(ContextPath, "Streaming integrity", "TestUser"); - var created = await client.Observe(new CreateNodeRequest(threadNode), o => o.WithTarget(Mesh.Address)) + // Target the CONTEXT namespace, the way a real client creates a thread — not the root mesh + // hub. Routing creation through the owning namespace is both the majority convention in the + // thread tests and what keeps this representative of the production path. + var created = await client.Observe(new CreateNodeRequest(threadNode), o => o.WithTarget(new Address(ContextPath))) .Should().Within(60.Seconds()).Emit(); created.Message.Success.Should().BeTrue(created.Message.Error ?? ""); var threadPath = created.Message.Node!.Path!; diff --git a/test/MeshWeaver.Threading.Test/StreamingCellWriteByteCountTest.cs b/test/MeshWeaver.Threading.Test/StreamingCellWriteByteCountTest.cs index 616aaa4c62..e459a6874b 100644 --- a/test/MeshWeaver.Threading.Test/StreamingCellWriteByteCountTest.cs +++ b/test/MeshWeaver.Threading.Test/StreamingCellWriteByteCountTest.cs @@ -189,13 +189,18 @@ public void Record(string? target, PatchDataRequest request) { if (target is null) return; - var patch = request.Patch.Content?.Length ?? 0; - var baseValues = request.BaseValues?.Content?.Length ?? 0; + // UTF-8 byte counts, not string.Length: the payload travels as UTF-8, and a char count + // silently understates anything non-ASCII — which real chat is full of (German, emoji). + var patch = Utf8Bytes(request.Patch.Content); + var baseValues = Utf8Bytes(request.BaseValues?.Content); byTarget.AddOrUpdate(target, _ => new Wire(1, patch, baseValues), (_, w) => new Wire(w.Writes + 1, w.PatchBytes + patch, w.BaseBytes + baseValues)); } + private static long Utf8Bytes(string? s) => + string.IsNullOrEmpty(s) ? 0 : System.Text.Encoding.UTF8.GetByteCount(s); + public void Reset() => byTarget.Clear(); /// Everything recorded for the node at , however addressed.