Requesting Work via stream.Update()

🚨 DEFAULT PATTERN — required for every mesh-node mutation. Threads, thread messages, NodeType compile state, code editing, satellite annotations — all of them. If you are about to write class XxxRequest : IRequest<XxxResponse> to mutate a node's content, stop and read this page first.

The single rule: mutate the target node's state via workspace.GetMeshNodeStream(path).Update(current => modified). A server-side watcher on the owning hub picks up the change and dispatches any side-effect work. Results are published by writing back to the same node, and the synchronization protocol propagates them cluster-wide automatically.

Reads use the same stream. IMeshNodeStreamCache (src/MeshWeaver.Hosting/MeshNodeStreamCache.cs) hands out a single shared handle per node path — server-side hubs and Blazor views alike resolve it through GetMeshNodeStream(path) (see GUI Data Binding). Read and write share one stream — there is no separate read API to keep in sync.

Why this is mandatory, not merely preferred. Every recent "hub becomes unresponsive after the second operation" CI failure — CodeEditRecompile, NodeTypeRelease, LinkedInPullActions, ThreadAgentIntegration in run 26036857424 — traced back to bespoke request/response handlers racing the watcher: two concurrent activities, leaked callbacks, wedged hub. The stream-based pattern is race-free by construction. Stream-based mutation flow: caller writes via stream.Update, the owning hub's watcher dispatches work, the worker writes results back to the same node, and every subscriber (other silos, clients) receives the change automatically.

Sanctioned Exceptions

hub.Observe(request) is valid only for:

Everything else — state machines, "trigger work and observe progress" flows, every UI button that mutates a node — uses stream.Update().

Cross-Hub Patch Semantics

When you call workspace.GetMeshNodeStream(otherPath).Update(...) from a non-owner hub (UpdateRemote), the framework re-runs your lambda against the caller's snapshot and ships an RFC 7396 JSON-merge patch of the diff to the owner.

🔑 Identity across the hop. The outbound patch carries the caller's AccessContext. On the owner, applying it and propagating the result through the data-source sync stream happens on a deferred continuation where the live AsyncLocal is gone — so the owner falls back to its standing owner identity (the node's CreatedBy, carried via CircuitContext). On a cold start the owner must have that identity established before the first such write, or the deferred sync post carries a null context and fails closed. See Owner Injection.

This is safe only when the patch is idempotent under merge — applying it twice yields the same result as applying it once.

Merge-safe operations:

• Single-field assignment — { Foo: value }. Merging twice gives the same result.
• Dict SetItem by key — { Bag: { key: value } }. Merging twice leaves the same key-value pair.

Not merge-safe:

• List append / prepend — node with { Messages = Messages.Add(x) }. The patch becomes the whole list with x at the end. Two concurrent appends each compute a list ending in x; the owner merges them in order and the last write wins, silently dropping the first.
• Read-modify-write on a list — same root cause: the patch is the new list, unaware of concurrent writers.

Design rule: Cross-hub mutations should be a single stream.Update(...) on the target node. The owning hub's action-block serialisation guarantees race-free merge; RFC 7396 patch semantics ensure you touch only the fields you intend to change.

A worked consequence. Logon Actions builds its whole idempotency guarantee on the two rules above: the run-once ledger is a dictionary keyed by action id (merge-safe SetItem, never a list append), and the action's effect plus its ledger entry go into one patch on the user's profile — so two concurrent logons cannot apply a migration twice, and no restart can land between "it happened" and "we recorded it".

The thread refactor as the canonical example. The full resubmit/delete-from/record-failure flow once posted bespoke trigger messages, then briefly used intent-field payloads (RequestedResubmit, RequestedDeleteFromMessageId, PendingFailures) consumed by per-operation watchers. Today, the full mutation is inline inside the hub extension method's stream.Update lambda — truncate Messages, re-queue PendingUserMessages, etc., all in one patch. See ThreadOperations.md for the public API (hub.ResubmitMessage, hub.DeleteFromMessage, hub.RecordSubmissionFailure).

Canonical Helpers

Don't roll your own watcher. Two helpers in MeshWeaver.Mesh.Contract (ActivityControlPlaneExtensions.cs) implement this pattern as one-liners:

hub.WatchControlPlane(onRequestedStatus, logger) Use when the trigger is the single ActivityLog.RequestedStatus field. Drives Cancel / Start / Retry off a property patch.

hub.WatchSubmission(fingerprint, needsDispatch, dispatch, logger) Use when the trigger is an arbitrary "this state needs work now" predicate — for example, a thread has unprocessed user messages and isn't already executing. Internally it composes:

GetMeshNodeStream → DistinctUntilChanged(fingerprint) → Where(needsDispatch) → SelectMany(dispatch)

The canonical reference for both helpers is ActivityControlPlane.md (see § "Generalising: WatchSubmission" and § "Anti-patterns to remove on sight"). If you find yourself reaching for Throttle(...), Interlocked.CompareExchange, or a manual Subject + flag, that's the signal you should be calling one of these helpers instead.

🚨 The watcher runs on ACTIVATION — so a CREATE has to wake the owner

The watcher is installed by the owning per-node hub's WithInitialization, which means it runs if and only if that hub is activated. A create only writes the row. Nothing else in a create opens the node, so before #3153 a request filed by a create was never seen: it sat at Requested with no error, no failed state, no log line — and then ran, correctly and immediately, the next time anything happened to open the node (a page view, a stream subscription, an MCP get).

Measured on memex.meshweaver.cloud: a Store/Subscription created via MCP sat untouched for 4.5 hours and completed 20 s after the first read. The same node created from the admin UI activated in 31 s — because the open page held a stream handle, which activated the owner as a side effect.

That side effect is also why the pattern's own tests could not catch it. A control-plane test that subscribes to the node to await its terminal state is the thing making the watcher run, so it passes with the defect fully present. A test that means to pin this must file the request with IMeshService.CreateNode and observe the outcome through a channel that does not activate the owner — durable storage (IStorageAdapter.Read) is the one that qualifies. See ControlPlaneRunsWhenTheRequestArrivesByCreateTest.

MeshExtensions.ActivatePendingControlPlane now closes it: after the create response is posted, a node whose content carries a pending RequestedXxx has its owner opened once, so the watcher sees the request. Two properties matter and are deliberate:

• Strictly after the response, fire-and-forget. A cold per-node activation can take 5–45 s in CI (NodeType compile, dependency load, JIT), so it must never gate, delay or fail a create. An owner that cannot be woken is a warning naming the path; the node is created and unchanged.
• Gated on the content, not done for every create. A bulk install fans CreateOrUpdateNodeRequest out into an inner CreateNodeRequest per node, so activating unconditionally would wake a hub per imported file — 141 for Hosting, 425 for the core samples. Inert data needs no watcher; only a node that actually carries a pending request does. This is the load-bearing reason RequestedXxx is a naming convention and not a free choice: it is the only thing that tells the create path a request is present without this assembly knowing any domain type.

When to Use This Pattern

Use it when:

• The work is triggered by a state change (caller mutates input fields).
• The result belongs on the same node (caller observes output fields on the same MeshNode it mutated).
• You want automatic cluster-wide propagation — every silo that subscribes via a synced query (workspace.GetQuery) or the node's own shared stream sees the update without any explicit broadcast.
• You want no cross-silo round-trip during grain activation — the watcher runs in-grain on whichever silo owns the node.

Do not use it for:

• One-shot transient queries whose result doesn't belong on a node (use hub.Observe(request) instead).
• Cases requiring an immediate synchronous response — the watcher dispatches reactively off the node stream, so the round has multi-step latency compared to a direct hub.Observe round-trip.

The Canonical Example: Thread Execution

MeshWeaver.AI already implements this pattern end-to-end for thread execution dispatch.

Caller — writing the request into node state

// MeshWeaver.AI/ThreadInput.cs — AppendUserInput
workspace.GetMeshNodeStream(threadPath).Update(node =>
{
    // Bad-data tolerance: an existing node whose content can't be read is
    // left alone — NEVER clobbered with a fresh MeshThread.
    var thread = node.ContentAs<MeshThread>(workspace.Hub.JsonSerializerOptions, logger);
    if (node.Content is not null && thread is null)
        return node;
    thread ??= new MeshThread();
    return node with
    {
        Content = thread with
        {
            UserMessageIds = thread.UserMessageIds.Add(msgId),
            PendingUserMessages = thread.PendingUserMessages.SetItem(msgId, message),
            // ... other pending fields ...
        }
    };
}).Subscribe(
    _ => { },
    ex => logger?.LogWarning(ex, "AppendUserInput failed for {ThreadPath}", threadPath));

The caller writes the request into the thread node's state. PendingUserMessages is the request payload. No IRequest<TResponse> is posted. No callback is registered. The work is requested by the very act of mutating the node. Note the trailing .Subscribe(...) — Update returns a cold observable and the write only happens on subscribe; forgetting it means the submission silently never lands.

Server — the dispatch watcher

// The generic helper — use this for a NEW watcher:
hub.WatchSubmission(
    fingerprint:   node => /* the control-plane tuple that decides "actionable" */,
    needsDispatch: node => /* predicate over that state */,
    dispatch:      node => DoWork(node),          // IObservable<Unit>
    logger:        logger);

The watcher subscribes once at hub init. DistinctUntilChanged on the fingerprint guarantees the same actionable state cannot fire twice — there is no dispatching flag, no Throttle, no reentrancy guard; dispatch composes via SelectMany into one run per emission, and a failure is logged rather than killing the subscription (the next state change retries naturally).

The thread's own watcher (ThreadSubmissionServer.InstallServerWatcher in MeshWeaver.AI/ThreadSubmission.cs) is the same pattern hand-composed, because it needs two extras the helper does not carry: a fault re-establish (a dead watcher would park the thread forever) and an explicit owner-identity scope around its writes. Its shape is GetMeshNodeStream() → DistinctUntilChanged(SubmissionFingerprint) (status + pending/ingested/user-message ids — deliberately excluding every streaming field, so an executing round does not wake it) → an atomic claim Status: Idle → StartingExecution through stream.Update. Single-flight falls out of that claim: the owner's action block serialises the concurrent emissions, the first lambda that sees Idle flips it, and every other lambda re-reads Status != Idle and bails. Read that method before copying it.

Result publication

The dispatched work writes its progress and final result back onto the same node (or a satellite node it owns) via workspace.GetMeshNodeStream(path).Update(...). The result reaches every subscriber automatically:

• Local: other subscribers on the same hub see the next emission of the MeshNodeReference reducer.
• Cross-silo / clients: synced-query subscribers (workspace.GetQuery) and every GetMeshNodeStream(path) handle see the change via the synchronization protocol — no explicit broadcast needed.

Applied: Dynamic NodeType Compilation

NodeTypeEnrichmentHelpers.EnrichWithNodeType's slow path uses this pattern: the caller flips CompilationStatus to Pending on the NodeType node and waits for a terminal status on the same stream.

Caller — the trigger + observe pair

Note: this is the shape, not a copy of a live method — there is no NodeTypeService type in the tree. Split the chain into a one-shot trigger pipeline and a terminal-status observation pipeline, both reactive; .Take(1) is what makes the trigger one-shot, so no triggered flag and no Interlocked.CompareExchange is needed.

// Subscribe to the per-NodeType node via the shared handle.
// Trigger pipeline: takes the first emission whose CompilationStatus is null/Unknown
// and composes one stream.Update flipping it to Pending.
// .Take(1) makes the trigger inherently one-shot — no `triggered` flag, no CompareExchange.
// Observation pipeline: waits for terminal status (Ok / Error) on the same stream.
var stream = workspace.GetMeshNodeStream(nodeTypePath);

var trigger = stream
    .Where(node => node?.Content is NodeTypeDefinition def
        && (def.CompilationStatus is null || def.CompilationStatus == CompilationStatus.Unknown))
    .Take(1)
    .SelectMany(_ => stream.Update(current =>
        current?.Content is NodeTypeDefinition d
            && (d.CompilationStatus is null || d.CompilationStatus == CompilationStatus.Unknown)
            ? current with { Content = d with { CompilationStatus = CompilationStatus.Pending } }
            : current))      // no-op when someone else already flipped it
    .IgnoreElements();

var terminal = stream
    .Where(node => node?.Content is NodeTypeDefinition def
        && (def.CompilationStatus == CompilationStatus.Ok
            || def.CompilationStatus == CompilationStatus.Error))
    .Take(1);

return trigger.Merge(terminal!)
    .Take(1)
    .Timeout(TimeSpan.FromSeconds(30));

Server — NodeTypeCompilationHelpers.InstallCompileWatcher

Note: the live watcher is NodeTypeCompilationHelpers.InstallCompileWatcher (src/MeshWeaver.Graph/Configuration/) — it is installed per NodeType hub and carries a good deal more than the sketch below (kickoff, settle/registration waits, the Unavailable marker rules). The shape below is the minimal pattern it is an instance of: CompilationStatus as the fingerprint makes a Pending → Compiling throttle-guard unnecessary, because DistinctUntilChanged already prevents re-firing on our own write.

hub.WatchSubmission(
    fingerprint:   node => (node.Content as NodeTypeDefinition)?.CompilationStatus,
    needsDispatch: node => node.Content is NodeTypeDefinition d
                         && d.CompilationStatus == CompilationStatus.Pending,
    dispatch:      node => Compile(workspace, compilationService, node)
                              .Select(_ => Unit.Default),
    logger:        logger);

// Compile is an IObservable<Unit> that flips Pending → Compiling, runs
// compilationService.CompileAndGetConfigurations, then writes the terminal
// (Ok / Error) status + AssemblyLocation back via GetMeshNodeStream().Update.
// Because the fingerprint is CompilationStatus, the watcher's own writes
// are filtered out by DistinctUntilChanged — no Throttle needed, no reentrancy guard.

Cluster-wide cache propagation

Every silo subscribes to the NodeType nodes through a synced query (workspace.GetQuery — see Synced Mesh Node Queries). When the compiled result is written back, the synced query emits the new node state to every subscriber, so each silo's local NodeType-configuration cache picks up the new (AssemblyLocation, HubConfiguration) automatically — EnrichWithNodeType's fast path then hits for that NodeType, and grain activation stays synchronous.

Error Notification — Callers Must Observe Failure

The watcher dispatching the work must publish failure as well as success. A missing or invalid NodeType, a compilation error, an unreachable persistence layer — all must flip a status field on the node.

Silent timeout is not acceptable. The caller observes the same node's reducer. If the request never lands a response, the caller cannot distinguish "still working" from "broken".

Conventional status shape — the live one is MeshWeaver.Mesh.Contract/Services/CompilationStatus.cs:

enum CompilationStatus { Unknown, Pending, Compiling, Ok, Error, Unavailable }

Unavailable is deliberately distinct from Error: it means "the state could not be determined" (a settle wait or registration lookup timed out) — an availability problem, never "the source is broken".

Caller observation chain (GetMeshNodeStream, never GetRemoteStream<MeshNode, …> — that overload throws):

workspace.GetMeshNodeStream(nodeTypePath)
    .Where(node => node?.Content is NodeTypeDefinition def
             && def.CompilationStatus is CompilationStatus.Ok or CompilationStatus.Error)
    .Take(1)
    .Timeout(TimeSpan.FromSeconds(60))
    .Subscribe(node => { /* … */ }, ex => logger.LogWarning(ex, "compile wait failed"));

Pattern Comparison

Request/response is the exception, not a peer pattern. The work's result almost always belongs on a node (thread, message, NodeType, satellite). Making the node's own content the contract eliminates an entire class of race conditions, leaked callbacks, and hub wedges that bespoke handlers reintroduce every time.