The upsert's inner create registered its callback after posting, so the reply could be dropped

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

@@ -505,7 +505,13 @@ If you find yourself using `.Take(N)` outside a `SelectMany` that immediately pr
 
 2. **No `*Async` extension shims on `IMeshService`.** Use `meshService.CreateNode(node)` / `UpdateNode(node)` / `DeleteNode(path)` — these return `IObservable<MeshNode>`. Never use `.CreateNodeAsync(...)` / `.UpdateNodeAsync(...)` / `.DeleteNodeAsync(...)` — those extensions bridge to Task via `.ToTask()` and deadlock every time they are reached from a hub handler.
 
-3. **Use `hub.Observe(...)` instead of `RegisterCallback` / `AwaitResponse`.** The Task-returning overloads are `[Obsolete]`. Production code MUST use `hub.Observe(delivery)` (already-posted) or `hub.Observe(request, options?)` (also posts) — both return `IObservable<IMessageDelivery[<TResponse>]>`. `DeliveryFailure` flows via `OnError`; no Task-await deadlock surface, no silently-skipped callback.
+3. **Use `hub.Observe(...)` instead of `RegisterCallback` / `AwaitResponse`.** The Task-returning overloads are `[Obsolete]`. Production code MUST use `hub.Observe(request, options?)` — it returns `IObservable<IMessageDelivery[<TResponse>]>`, `DeliveryFailure` flows via `OnError`, and there is no Task-await deadlock surface and no silently-skipped callback.
+
+   **🚨 3a. ALWAYS the pre-registering `hub.Observe(request, options)` — NEVER `hub.Post(...)` then `hub.Observe(delivery)`.** The two are not interchangeable. `Observe(request, options)` registers the response `AsyncSubject` **before** posting; `Observe(delivery)` registers it **after**, and `HandleCallbacks` **DROPS** any response whose correlation id has no registered subject yet ("No subject found for response … treating as processed"). A reply that lands in that window is consumed and gone — the caller's callback then sits pending forever (surfacing as a hang to its timeout, or as a leaked callback at teardown).
+
+   The window is real, not theoretical: `PostImplGeneric` runs `ScheduleNotify` **synchronously**, so by the time `Post` returns the delivery is already enqueued and its turn scheduled onto `turnScheduler` on another thread. Preempt the posting thread — routine under CI thread-pool contention — and the handler answers first. A warm owner answers in sub-millisecond time.
+
+   `Observe(delivery)` is safe ONLY when the response provably cannot have arrived yet, and "the delivery is already posted" is *not* that proof. If you already hold a posted delivery, that is the smell — restructure so the request is handed to `Observe(request, options)` instead. Prior occurrences: `MeshOperations` (four verbs), `MeshNodeStreamExtensions.GetMeshNode`, `HubStreamProviderFactory`, and `CreateOrUpdateNodeRequest`'s inner create (#981). Pinned by `WorkspaceCacheEvictionTest.GetMeshNode_WarmOwner_DropsResponse_WhenSubjectRegisteredAfterPost` and `UpsertInnerCreateObservationTest`.
 
 4. **Never `.QueryAsync<MeshNode>($"path:X").FirstOrDefaultAsync()` to read a known node.** Queries go through a lagged read-side index. For a known path: live = `workspace.GetMeshNodeStream(path).Subscribe(...)`; one-shot = the same stream with `.Where(n => n is not null).Take(1).Timeout(...)` (or the `hub.GetMeshNode(path, timeout?)` convenience, which wraps it null-on-absent).
 
@@ -894,13 +900,26 @@ return delivery.Processed();
 
 **Why `hub.Observe(...)` and not `RegisterCallback`:** the legacy `RegisterCallback` returns `Task<IMessageDelivery>` and the framework short-circuits the user callback for `DeliveryFailure` — the Task gets the exception, the callback never fires. Callers that drop the Task get silent infinite hangs. `hub.Observe(...)` exposes the same TCS-backed Task via `task.ToObservable()`, so `OnError` fires naturally.
 
-### When you already have a delivery
+### 🚨 Never post first and observe afterwards
 
 ```csharp
+// ❌ WRONG — the response subject is registered AFTER the post. HandleCallbacks DROPS a
+//    response whose id has no subject yet, so a reply that lands in this window is gone
+//    and the callback hangs forever. Post returns only after ScheduleNotify has already
+//    enqueued the delivery and scheduled its turn on another thread — the handler can and
+//    does win this race under load.
 var delivery = hub.Post(request, o => ...);
 hub.Observe(delivery).Subscribe(onNext, onError);
+
+// ✅ RIGHT — one call; the AsyncSubject is registered BEFORE the post, so however early the
+//    reply lands it is buffered and replayed to the subscriber.
+hub.Observe(request, o => ...).Subscribe(onNext, onError);
 ```
 
+See rule 3a. If some intermediate code needs the delivery id up front, pass an explicit
+`o.WithMessageId(id)` through the same `Observe(request, options)` call — do not split it
+into a post and a later observe.
+
 ### Inside an `IObservable<T>` chain
 
 ```csharp
@@ -994,7 +1013,7 @@ The view binds to `_suggestions` directly — no callback that returns `Task<T[]
 |---|---|
 | `var x = await mesh.QueryAsync(...).ToListAsync()` | `mesh.Query<T>(req).Subscribe(c => ApplyChange(c))` |
 | `await Hub.AwaitResponse<R>(req, ...)` | `Hub.Observe(req).Subscribe(r => UpdateState(r.Message), ex => …)` |
-| `Hub.RegisterCallback(delivery, r => { … })` | `Hub.Observe(delivery).Subscribe(r => …, ex => …)` |
+| `var d = Hub.Post(req, o); Hub.RegisterCallback(d, r => { … })` | `Hub.Observe(req, o).Subscribe(r => …, ex => …)` — pass the REQUEST, never a delivery you already posted (rule 3a) |
 | `var n = await mesh.GetMeshNodeStream(p).Take(1).ToTask()` | live = `mesh.GetMeshNodeStream(p).Subscribe(n => UpdateState(n))`; one-shot = `Hub.GetMeshNode(p).Subscribe(n => …)` |
 | `return Task.FromResult(_suggestions.ToArray())` | bind directly to `_suggestions`; let `Subscribe` push updates |
 | `_ = LoadAsync(); await ...` | sync method that fires `Subscribe` |
@@ -1400,7 +1419,7 @@ exist yet).
 |---|---|---|
 | `hub.Post(...)` | Yes | Fire-and-forget, safe from any thread |
 | `hub.Observe(request).Subscribe(onNext, onError)` | Yes | Reactive request/response; DeliveryFailure → onError |
-| `hub.Observe(delivery).Subscribe(...)` | Yes | Same, when the delivery was already posted |
+| `hub.Post(...)` then `hub.Observe(delivery).Subscribe(...)` | **NO** | Registers the response subject AFTER the post; a reply landing in that window is DROPPED and the callback hangs forever. Use `hub.Observe(request, options)` — see rule 3a |
 | `meshService.CreateNode(...).Subscribe()` | Yes | Fire-and-forget, no callback logic |
 | `workspace.GetMeshNodeStream(path).Update(...).Subscribe(...)` in handler body | Yes | Runs on grain scheduler |
 | `hub.RegisterCallback(...)` | **OBSOLETE** | Use `hub.Observe(...)` — RegisterCallback's Task short-circuits DeliveryFailure → callback silently never fires → caller hangs |

src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-10-upsert-always-answers.md

@@ -0,0 +1,18 @@
+---
+Name: Saving a node always gets an answer
+Category: Fix
+Description: A create-or-update could finish its work but never report back, leaving the caller waiting.
+Icon: Sparkle
+Order: -20260810
+---
+
+# Saving a node always gets an answer
+
+When something saved a node that did not exist yet — a repository sync, an import, or the
+compile status the platform records for you — the save occasionally completed but never
+reported back. Whatever was waiting for the confirmation simply waited, and the work looked
+stuck even though it had already succeeded.
+
+The reply was being discarded because the reply could arrive fractionally before the sender
+was ready to listen for it, and a busy machine made that far more likely. The sender now
+starts listening before it asks, so the confirmation can never be missed.

src/MeshWeaver.Mesh.Contract/MeshExtensions.cs

@@ -3219,12 +3219,42 @@ private static IMessageDelivery HandleCreateOrUpdateNodeRequest(
         void DispatchInnerCreate()
         {
             var inner = new CreateNodeRequest(node) { CreatedBy = requestedBy };
-            var forwarded = hub.Post(inner, o =>
-            {
-                var withTarget = o.WithTarget(hub.Address);
-                return inboundCtx is not null ? withTarget.WithAccessContext(inboundCtx) : withTarget;
-            })!;
-            hub.Observe(forwarded)
+            // 🚨 PRE-REGISTERING Observe(request, options) — never Post-then-Observe(delivery) (#981).
+            //
+            // This runs OFF the hub action block: `existingObs` is `persistence.Read(...)`, which for
+            // any adapter that is not a synchronous in-memory hit (partition-routed / embedded-resource
+            // / pooled backends) emits on an I/O thread. So `hub.Post` here is NOT reentrant into the
+            // turn loop — `KickDrain` finds `draining == false` and schedules the turn onto
+            // `turnScheduler` STRAIGHT AWAY, on another thread, while this thread walks on to the next
+            // statement. `PostImplGeneric` also runs `ScheduleNotify` synchronously, so POSTED /
+            // RECEIVED / ENQUEUED are already behind us by the time `Post` returns.
+            //
+            // With the old `Post(...)` + `Observe(forwarded)` pair, everything between those two
+            // statements was unprotected: preempt this thread (a saturated CI runner does exactly
+            // that) and the turn can run the whole create and post its `CreateNodeResponse` first.
+            // `HandleCallbacks` finds no subject for the correlation, treats the response as consumed
+            // and DROPS it — then this line registers a subject that nothing will ever answer. The
+            // upsert never replies, and the caller's callback sits pending until the teardown
+            // quiescing budget reports it as a leaked `CreateNodeRequest@mesh/<self>`.
+            //
+            // It also destroyed the evidence: `RequestFateLedger` starts a trail at REGISTRATION, so
+            // the post-hoc overload lost POSTED/RECEIVED/ENQUEUED/ROUTED and left a trail reading
+            // "AWAITING → REGISTERED_AFTER_POST" and nothing else — which is precisely the capture
+            // that could not name this mechanism.
+            //
+            // `Observe(request, options)` registers the AsyncSubject BEFORE posting, so the response
+            // is buffered no matter how early it lands, and the trail records every stage.
+            //
+            // Identity is unaffected by the swap. The overload re-seeds emissions from the AMBIENT
+            // context, which is exactly what is unreliable on this thread (MeshService.CreateNode's
+            // note) — but nothing here leans on it: `inner.CreatedBy` pins the identity as a request
+            // FIELD, the post stamps `inboundCtx` explicitly, `ApplyUpdateViaStream` opens its own
+            // `SwitchAccessContext(inboundCtx)`, and PostOk/PostFail are `ResponseFor` posts.
+            hub.Observe(inner, o =>
+                {
+                    var withTarget = o.WithTarget(hub.Address);
+                    return inboundCtx is not null ? withTarget.WithAccessContext(inboundCtx) : withTarget;
+                })
                 .Subscribe(
                     d =>
                     {

test/MeshWeaver.Hosting.Monolith.Test/UpsertInnerCreateObservationTest.cs

@@ -0,0 +1,132 @@
+using System;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
+using MeshWeaver.Data;
+using MeshWeaver.Hosting.Monolith.TestBase;
+using MeshWeaver.Mesh;
+using MeshWeaver.Messaging;
+using Xunit;
+
+namespace MeshWeaver.Hosting.Monolith.Test;
+
+/// <summary>
+/// Pins the correlation contract behind #981 — "callbacks still pending past the teardown quiescing
+/// budget", always reported as a <c>CreateNodeRequest@mesh/&lt;self&gt;</c> that the mesh hub never
+/// answered (FutuRe EuropeRe, Cession, PandasExplorer).
+///
+/// <para><b>The mechanism.</b> <c>MessageHub.HandleCallbacks</c> DROPS a response whose correlation id
+/// has no registered subject ("No subject found for response … treating as processed"). So the order of
+/// the two steps decides whether a reply can be lost:</para>
+/// <list type="bullet">
+///   <item><b>Post, then <c>Observe(delivery)</c></b> — the subject is registered AFTER the post.
+///     <c>PostImplGeneric</c> runs <c>ScheduleNotify</c> synchronously, so the delivery is already
+///     enqueued and its turn scheduled onto <c>turnScheduler</c> (another thread) by the time
+///     <c>Post</c> returns. Preempt the posting thread — routine on a saturated CI runner — and the
+///     turn answers first, the reply is dropped, and the caller's callback is pending FOREVER.</item>
+///   <item><b><c>Observe(request, options)</c></b> — registers the <c>AsyncSubject</c> BEFORE posting,
+///     so however early the reply lands it is buffered and replayed to a late subscriber.</item>
+/// </list>
+///
+/// <para>The upsert handler's <c>DispatchInnerCreate</c> ran the unsafe order on exactly this message
+/// shape — a self-targeted <c>CreateNodeRequest</c> on the mesh hub, dispatched OFF the action block
+/// from the <c>persistence.Read</c> continuation — which is what made <c>CreateOrUpdateNodeRequest</c>
+/// silently never answer.</para>
+///
+/// <para>The race is made DETERMINISTIC here the same way
+/// <c>WorkspaceCacheEvictionTest.GetMeshNode_WarmOwner_DropsResponse_WhenSubjectRegisteredAfterPost</c>
+/// does it, but without a fixed sleep: the fence is the node becoming readable plus two full
+/// request/response round-trips through the same mesh hub, which together prove the create's turn ran
+/// to completion and its reply was pumped.</para>
+/// </summary>
+public class UpsertInnerCreateObservationTest(ITestOutputHelper output) : MonolithMeshTestBase(output)
+{
+    /// <summary>Share Mesh/SP across [Fact]s — see MonolithMeshTestBase.ShareMeshAcrossTests.</summary>
+    protected override bool ShareMeshAcrossTests => true;
+
+    /// <summary>
+    /// Fences until the create's turn has demonstrably finished: the node is readable, and two
+    /// complete round-trips have since travelled through the mesh hub. Anything the create posted is
+    /// therefore long since pumped — which is the interleaving that loses a post-hoc registration.
+    /// </summary>
+    private async Task FenceCreateCompleted(string path)
+    {
+        // GetMeshNode is one-shot request/response — it answers null and COMPLETES for a node that
+        // is not there yet, so it has to be re-queried rather than awaited (WritingTests.md,
+        // "Polling loops around QueryAsync"). Never a fixed sleep: the wait is on the condition.
+        await Observable.Interval(TimeSpan.FromMilliseconds(50)).StartWith(0L)
+            .SelectMany(_ => Mesh.GetMeshNode(path))
+            .Where(n => n is not null)
+            .FirstAsync()
+            .Timeout(20.Seconds());
+
+        // Two further COMPLETE round-trips through the same mesh hub. Once these have answered, any
+        // reply the create posted is long since pumped — the interleaving that loses a post-hoc
+        // registration.
+        await Mesh.GetMeshNode(path).Should().Within(15.Seconds()).Match(n => n is not null);
+        await Mesh.GetMeshNode(path).Should().Within(15.Seconds()).Match(n => n is not null);
+    }
+
+    [Fact(Timeout = 60000)]
+    public async Task SelfTargetedCreate_OnTheMeshHub_DropsItsReply_WhenTheSubjectIsRegisteredAfterPost()
+    {
+        // ARM 1 — the #981 shape: post first, register the callback afterwards.
+        var doomedId = $"upsert-race-doomed-{Guid.NewGuid():N}";
+        var doomedPath = $"{TestPartition}/{doomedId}";
+        var doomed = Mesh.Post(
+            new CreateNodeRequest(new MeshNode(doomedId, TestPartition)
+            {
+                Name = doomedId,
+                NodeType = "Markdown"
+            }),
+            o => o.WithTarget(Mesh.Address));
+        doomed.Should().NotBeNull();
+
+        // The create really did run and answer — so its reply hit HandleCallbacks with no subject.
+        await FenceCreateCompleted(doomedPath);
+
+        // Registering now cannot recover a reply that was already consumed: the callback is orphaned.
+        // THIS is the leaked `CreateNodeRequest@mesh/<self>` the quiescing budget reports.
+        await Mesh.Observe(doomed!).Select(d => (object?)d.Message)
+            .Should().NotEmit(within: 3.Seconds(),
+                because: "a response with no registered subject is dropped by HandleCallbacks");
+
+        // ARM 2 — the fix shape: pre-register, then post. Same fence, reply still delivered.
+        var safeId = $"upsert-race-safe-{Guid.NewGuid():N}";
+        var safePath = $"{TestPartition}/{safeId}";
+        var safe = Mesh.Observe(
+            new CreateNodeRequest(new MeshNode(safeId, TestPartition)
+            {
+                Name = safeId,
+                NodeType = "Markdown"
+            }),
+            o => o.WithTarget(Mesh.Address));
+
+        await FenceCreateCompleted(safePath);
+
+        await safe.Select(d => d.Message)
+            .Should().Within(10.Seconds())
+            .Match(m => m is CreateNodeResponse { Success: true },
+                "the pre-registered AsyncSubject buffers the reply for an arbitrarily late subscriber");
+    }
+
+    /// <summary>
+    /// The production seam: <c>CreateOrUpdateNodeRequest</c> for a node that does NOT exist forwards an
+    /// inner self-targeted <c>CreateNodeRequest</c> and must ANSWER. Before the fix that inner create
+    /// used the post-then-observe order, so whenever the turn won the race the upsert produced no
+    /// response at all and its caller's callback leaked.
+    /// </summary>
+    [Fact(Timeout = 60000)]
+    public async Task Upsert_OfAMissingNode_Answers_ThroughTheInnerCreate()
+    {
+        var id = $"upsert-answers-{Guid.NewGuid():N}";
+        var node = new MeshNode(id, TestPartition) { Name = id, NodeType = "Markdown" };
+
+        var response = await Mesh
+            .Observe<CreateOrUpdateNodeResponse>(new CreateOrUpdateNodeRequest(node))
+            .Should().Within(30.Seconds()).Emit(
+                "the upsert's inner CreateNodeRequest must never lose its reply");
+
+        response.Message.Node.Should().NotBeNull();
+        response.Message.Node!.Path.Should().Be($"{TestPartition}/{id}");
+    }
+}