Activity Operations

Every activity state-transition in MeshWeaver — cancel, restart, or any RequestedStatus flip — goes through extension methods on IMessageHub defined in src/MeshWeaver.Mesh.Contract/HubActivityExtensions.cs. Tests, GUI click handlers, MCP agents, and plugins all call these methods. There is no other public entry point.

This page covers the client side: how callers request a transition. For the server side — the watcher that consumes the flip and drives the internal transition — see Activity Control Plane.

End-to-end activity cancellation: the caller writes RequestedStatus, the activity hub reacts, and the terminal state propagates back via the reactive stream.

Why a dedicated surface?

Before consolidation, every cancel button rolled its own five-line lambda — a different GetMeshNodeStream(path).Update(...) call per call site, with roughly half of them missing the no-op guard or the error logger. The IMessageHub extensions fix that in three ways:

The extension surface

using MeshWeaver.Mesh;   // HubActivityExtensions
using MeshWeaver.Data;   // ActivityStatus, ActivityLog

// Cancel a running activity.
// Patches RequestedStatus = Cancelled. The activity hub's WatchControlPlane
// handler trips the stored CTS and transitions Status → Cancelled.
hub.CancelActivity(activityPath);

// Generic status flip — use for restart (Running) or any other transition
// the activity hub's WatchControlPlane handler is wired to honour.
hub.RequestActivityStatus(activityPath, ActivityStatus.Running);
hub.RequestActivityStatus(activityPath, ActivityStatus.Cancelled);

// Both accept an optional onError callback for one-shot error signalling:
hub.CancelActivity(activityPath, onError: msg => ShowToast(msg));

CancelActivity is a thin alias — it calls RequestActivityStatus(path, ActivityStatus.Cancelled, onError). The single implementation guards twice inside the Update lambda and silently returns the node unchanged in both cases:

• the node's Content does not materialise as a typed ActivityLog, or
• RequestedStatus already equals the requested value (the request is in flight).

A silent no-op is the correct outcome for a duplicate request, but it also means a cancel against a node whose content is not an ActivityLog does nothing and reports nothing — onError fires only when the Update itself faults, never for either guard. If a cancel appears to be ignored, check the node's content type first.

hub can be any IMessageHub: a click context's ctx.Host.Hub, a test fixture's Mesh, an MCP plugin's captured hub, or even the activity hub itself patching its own status from within a worker. The extension routes the write through hub.GetWorkspace().GetMeshNodeStream(activityPath).Update(...), which auto-dispatches based on who is calling:

• Writer is the activity hub itself — write goes through its local data source.
• Writer is anywhere else — write routes as an RFC 7396 JSON-merge patch via the process-wide IMeshNodeStreamCache. The activity hub's single-threaded action block serialises every mirror's write, so there are no races.

Observing the result

The mutation methods are fire-and-forget. To observe the outcome, subscribe to the activity node's stream — the same shared handle the running-activities UI strip already binds to.

The flow is 100% reactive end-to-end. No FirstAsync().ToTask(ct), no await, no Task<T> boundary. The UI re-renders when the stream ticks; a worker waiting for a terminal state chains via SelectMany. See AsynchronousCalls → "Why await Deadlocks in Hub Handlers".

var sub = workspace.GetMeshNodeStream(activityPath)
    .Select(node => node?.Content as ActivityLog)
    .Where(log => log is { } l && l.Status != ActivityStatus.Running)
    .Take(1)
    .Subscribe(
        terminal => Logger.LogInformation(
            "Activity {Path} settled to {Status}", activityPath, terminal!.Status),
        ex => Logger.LogWarning(ex, "Activity stream errored for {Path}", activityPath));

// The caller owns `sub` and disposes it when the wait is no longer relevant
// (component dispose, parent scope dispose, etc.).

Tests bridge to Task exactly once at the assertion edge — see WritingTests. Application code stays observable throughout.

What the activity hub does in response

When RequestedStatus flips, the activity hub's WatchControlPlane subscription fires on the value change (it projects RequestedStatus off the hub's OWN node stream through DistinctUntilChanged, so the callback sees changes, not every emission). The subscription is installed from a MessageHubConfiguration.WithInitialization(...) callback in the activity NodeType's HubConfiguration:

var subscription = hub.WatchControlPlane(requested =>
{
    // requested is ActivityStatus? — null means "no pending request"
    // (never set, or cleared after a transition).
    if (requested == ActivityStatus.Cancelled)
    {
        cts.Cancel();   // trips the stored CancellationToken
    }
});
hub.RegisterForDisposal(subscription);   // the watcher's lifetime IS the hub's

Two things the signature makes non-optional:

• The callback parameter is ActivityStatus?. null is a real value — it means no request is pending — so never treat the callback as "a transition was requested".
• WatchControlPlane returns an IDisposable you must register with the hub's lifetime (hub.RegisterForDisposal(...)). Drop it and the subscription outlives the hub.

The handler runs on whatever scheduler the upstream stream emits on — in practice the hub's own action block. Treat it as hub-reachable code: no await, compose follow-up work as IObservable chains.

The subscription is self-healing but not infinitely retrying. It is established through SubscribeWithReEstablish, which re-establishes after ~1 s on a transient fault but stops permanently on two classes: own-node content that cannot be deserialized (re-subscribing would replay the same poisoned emission at 1 Hz), and a routing NotFound on its own node (the node is gone — re-subscribing is the resubscribe storm that took prod down on 2026-06-10). Both terminal cases are logged loudly rather than retried.

The running script receives the cancellation, throws OperationCanceledException, and the executor's normal terminal path writes Status = Cancelled back to the activity's MeshNode. The same stream the cancel button is bound to ticks one final time with the terminal state, and the UI re-renders — the cancel button disappears without any additional coordination.

WatchControlPlane — server side only

ActivityControlPlaneExtensions.WatchControlPlane is the server-side helper that an activity hub uses to install its own subscription inside WithInitialization. Application code never calls it directly.

Writing log messages: ActivityLogAppender

Every log line written onto a persisted activity node goes through ActivityLogAppender.Append (src/MeshWeaver.Mesh.Contract/ActivityLogAppender.cs). It is the append-side twin of the transition surface above: callers hand it messages plus an optional change to the log (terminal status, End, ReturnValue) and it performs one stream.Update carrying both — so a reader can never observe the terminal status before the lines that explain it.

ActivityLogAppender.Append(hub, activityPath, [new LogMessage(text, LogLevel.Information)])
    .Subscribe(_ => { }, ex => logger.LogDebug(ex, "activity append failed"));

// terminal status + its explanation, atomically:
ActivityLogAppender.Append(hub, activityPath, [new LogMessage(error, LogLevel.Error)],
        log => log with { Status = ActivityStatus.Failed, End = DateTime.UtcNow })
    .Subscribe(_ => { }, ex => logger.LogDebug(ex, "activity complete failed"));

Messages is a bounded window

ActivityLog.Messages holds at most ActivityLog.MessageWindowLimit (500) entries. Older lines are sealed into ActivityLogSegment satellites at {activityPath}/_Log/{index:D6} and drop off the head, leaving MessageWindowKeep (100) behind.

Why. Every stream.Update re-serialises the whole MeshNode.Content to compute its patch, so appending N lines to one growing list costs O(N²) — measured at ~719 MB of serialisation for a single memex-cloud import activity (5,239 writes over a 141 kB node), and the dominant term in that pod's CFS throttling. A delta field does not escape it: the cross-hub path ships an RFC 7396 merge patch, which clones a changed array whole, and the three-way merge's base extraction clones the previous array too — so one append to an N-element collection ships ~2N elements. Bounding the head is the only lever that changes the asymptotics; with the window fixed, each write is O(1) and the activity is O(N).

Below the window nothing changes. An activity that never reaches 500 messages takes exactly the single write per append it always did, with byte-identical content. Only long activities — the ones that actually hurt — take the new path.

Consequences for readers

🚨 Enumerate segments with a children query on {activityPath}/_Log, never a point-read of a segment path. A point-read of an absent satellite opens the shared stream cache's storm breaker on a path a concurrent write is about to use, and the breaker fast-fails writes too.

🚨 Never derive progress from Messages.Count. It stops growing once an activity passes the window, so anything keyed on it — a DistinctUntilChanged, a change detector — silently freezes for exactly the long-running activities it exists to follow. TotalMessageCount is the monotonic signal.

How the flush stays safe without a lock

A seal is claimed inside the head's update lambda (ActivityLog.ClaimSeal), which the owning hub serialises — so exactly one appender claims each slice and no two claims overlap. The claimed messages stay on the head until the segment write succeeds; only then are they trimmed (ActivityLog.CompleteSeal). A crash or a failed segment write therefore loses nothing and needs no watchdog: the claim is still standing and the messages are still there, so the next append retries the identical slice against the same (deterministic) segment index.

ActivityLogLogger (the kernel's script logger) is the one writer that does not use the appender: it re-asserts whole content on every 100 ms flush rather than patching, so it is the single writer of its node and seals its own overflow directly under the lock it already holds for the terminal settle. Same window, same segment shape, no claim protocol needed because there is no second appender to race.

See also

• Activity Control Plane — the Status / RequestedStatus pattern and how to wire your own NodeType to it
• Thread Operations — the matching IMessageHub surface for thread mutations (same shape)
• RequestViaStreamUpdate — the underlying stream.Update mechanism every method here is built on
• Owner Injection — the activity owner is the standing access context on the activity hub, injected everywhere and carried forward via CircuitContext