Teardown must quiesce activities before disposing

src/MeshWeaver.GitSync/ActivityRunner.cs

@@ -138,6 +138,13 @@ public static IObservable<string> RunActivity(
                 () => AccessContextScope.FromNode(created, accessService, logger),
                 _ =>
                 {
+                    // Register the run as in flight for the WHOLE of its life, released in the
+                    // Finally below — i.e. at TERMINAL (succeeded / failed / cancelled), not when
+                    // it was merely dispatched. Teardown quiesces on this: it stops starting new
+                    // work and waits for what is running, instead of walking away from it while
+                    // the command still executes against a tearing-down mesh.
+                    var runRegistration = hub.ServiceProvider.GetService<ActivityTracker>()?.Track();
+
                     var cts = new CancellationTokenSource();
                     // Cancel watcher: RequestedStatus = Cancelled trips the command's token
                     // (Activity Control Plane). Subscribed for the life of the run; disposed on
@@ -173,6 +180,8 @@ public static IObservable<string> RunActivity(
                         {
                             cancelWatch.Dispose();
                             cts.Dispose();
+                            // Terminal: the run is no longer in flight, so teardown may proceed.
+                            runRegistration?.Dispose();
                         })
                         .Select(_ => activityPath);
                 }));

src/MeshWeaver.Mesh.Contract/ActivityTracker.cs

@@ -0,0 +1,101 @@
+using System.Reactive;
+using System.Reactive.Disposables;
+using System.Reactive.Linq;
+using System.Reactive.Concurrency;
+using System.Reactive.Subjects;
+
+namespace MeshWeaver.Mesh;
+
+/// <summary>
+/// Counts the activity runs currently in flight on this mesh, so teardown can QUIESCE — stop
+/// starting new work, let what is running finish, and only then tear down.
+///
+/// <para><b>Why the existing drains do not cover this.</b> Mesh teardown already has three phases
+/// (<c>DisposalCompleted</c> → <c>IoPoolRegistry.DrainAll()</c> → <c>AsyncDisposeQueue</c>), but an
+/// activity falls through all of them:</para>
+/// <list type="bullet">
+///   <item><description>the trigger returns as soon as the activity exists (by design — it must not
+///     block the caller), so the action-block drain has nothing outstanding to wait on;</description></item>
+///   <item><description>the command runs off-turn via <c>ScheduleOffHubTurn</c>, so it holds no
+///     grain turn;</description></item>
+///   <item><description><c>SubscribeThroughPool</c> holds its pool permit for the SUBSCRIBE window
+///     only — deliberately, to protect the Autofac <c>BeginLifetimeScope</c>. Work that continues
+///     past the subscribe (a timer, an await continuation, a multi-second command) has already
+///     released the permit, so <c>DrainAll()</c> joins nothing.</description></item>
+/// </list>
+///
+/// <para>Measured before this existed: a 5-second activity, and <c>TeardownAsync</c> returned after
+/// 2028 ms with the command still running. Work that outlives the mesh keeps executing against a
+/// torn-down container — and when it touches a collectible ALC's metadata after unload, the process
+/// dies on a bare signal with no managed exception (FutuRe.Test exit=139, ~1 run in 5).</para>
+///
+/// <para><b>Quiesce, do not cancel.</b> The run is allowed to finish and emit its terminal
+/// ActivityLog status; teardown waits for that. Cancelling would abort work the caller believes is
+/// running and would leave the activity in a non-terminal state.</para>
+///
+/// <para><b>Must not deadlock.</b> The wait belongs to the teardown CALLER, never to a hub action
+/// block: a running activity's own Append/Finish writes go back through the hubs, so blocking a hub
+/// turn on this would block the very work it waits for. It is also bounded — a run that never
+/// settles must surface as a timeout, not hang teardown forever.</para>
+///
+/// <para>Instance state owned by the mesh (never <c>static</c>): the counter dies with the mesh
+/// rather than bleeding into the next test — see <c>Doc/Architecture/NoStaticState</c>.</para>
+/// </summary>
+public sealed class ActivityTracker : IDisposable
+{
+    // Deltas in, running count out. Scan accumulates — no lock, no Interlocked, no counter field.
+    //
+    // 🚨 No synchronisation primitive anywhere, including Observer.Synchronize: that takes a lock,
+    // and a lock reachable from hub-adjacent code is a deadlock waiting to happen. Concurrent
+    // Track()/dispose calls are serialised by QUEUEING them onto a scheduler (ObserveOn), which is
+    // how Rx serialises without anyone blocking. Scan therefore always runs on one thread and the
+    // running total cannot interleave.
+    private readonly Subject<int> deltas = new();
+    private readonly EventLoopScheduler scheduler = new();
+    private readonly IConnectableObservable<int> counts;
+    private readonly IDisposable connection;
+
+    /// <summary>Initializes a new instance of the <see cref="ActivityTracker"/> class.</summary>
+    public ActivityTracker()
+    {
+        // Replay(1) so a late subscriber (teardown) sees the CURRENT count immediately rather than
+        // waiting for the next change — a teardown on an idle mesh must complete at once.
+        counts = deltas
+            .ObserveOn(scheduler)
+            .Scan(0, (running, delta) => running + delta)
+            .StartWith(0)
+            .Replay(1);
+        connection = counts.Connect();
+    }
+
+    /// <summary>
+    /// Live count of in-flight runs, starting with the current value. Emits on every start and
+    /// every completion, so a consumer waits for zero without polling.
+    /// </summary>
+    public IObservable<int> InFlightChanges => counts;
+
+    /// <summary>
+    /// Completes once no run is in flight. Emits immediately when the mesh is already idle.
+    /// </summary>
+    public IObservable<Unit> WhenIdle =>
+        counts.Where(running => running == 0).Take(1).Select(_ => Unit.Default);
+
+    /// <summary>
+    /// Registers one run as in flight. Dispose the returned handle when the run reaches a TERMINAL
+    /// state (succeeded or failed) — not when it was merely dispatched. A double dispose does not
+    /// double-decrement.
+    /// </summary>
+    public IDisposable Track()
+    {
+        deltas.OnNext(1);
+        return Disposable.Create(() => deltas.OnNext(-1));
+    }
+
+    /// <inheritdoc />
+    public void Dispose()
+    {
+        connection.Dispose();
+        deltas.Dispose();
+        scheduler.Dispose();
+    }
+}

src/MeshWeaver.Mesh.Contract/MeshTeardownExtensions.cs

@@ -49,6 +49,24 @@ public static async Task TeardownAsync(this IMessageHub mesh, TimeSpan timeout)
         // never resolve DI once disposal has begun (the scope may already be tearing down).
         var ioPools = mesh.ServiceProvider.GetService<IoPoolRegistry>();
         var asyncDisposeQueue = mesh.ServiceProvider.GetService<AsyncDisposeQueue>();
+        var activities = mesh.ServiceProvider.GetService<ActivityTracker>();
+
+        // (0) QUIESCE ACTIVITIES FIRST — before anything is disposed.
+        //
+        // An activity falls through all three phases below: its trigger returned as soon as the
+        // activity existed, its command runs off-turn so it holds no grain turn, and
+        // SubscribeThroughPool holds a pool permit for the SUBSCRIBE window only — so DrainAll()
+        // joins nothing once the command continues past it. Measured: a 5s activity, and teardown
+        // returned after 2028ms with the command still running.
+        //
+        // This waits, it does not cancel: the run finishes and writes its terminal ActivityLog
+        // status. It must run BEFORE Dispose because those Append/Finish writes go back through
+        // hubs that are still alive. Bounded, so a run that never settles surfaces as a timeout
+        // rather than hanging teardown forever.
+        if (activities is not null)
+            await activities.WhenIdle.Timeout(timeout)
+                .Catch<Unit, Exception>(_ => Observable.Return(Unit.Default))
+                .ToTask();
 
         mesh.Dispose();
 

src/MeshWeaver.Mesh.Contract/Threading/IoPoolServiceCollectionExtensions.cs

@@ -36,6 +36,9 @@ public static IServiceCollection AddIoPools(
         // The async half of mesh teardown — resources enqueue async cleanup here from
         // their sync Dispose(); the mesh's DisposeAsync drains it before the scope dies.
         services.TryAddSingleton<AsyncDisposeQueue>();
+        // Mesh-scoped, so in-flight activity runs are counted per mesh and the counter dies with
+        // it (NoStaticState). Teardown quiesces on this — see ActivityTracker.
+        services.TryAddSingleton<ActivityTracker>();
         return services;
     }
 }

test/MeshWeaver.GitSync.Test/ActivityOutlivesDisposeTest.cs

@@ -0,0 +1,88 @@
+using System;
+using System.Reactive;
+using System.Reactive.Subjects;
+using System.Reactive.Linq;
+using System.Reactive.Threading.Tasks;
+using System.Threading;
+using System.Threading.Tasks;
+using MeshWeaver.Data;
+using MeshWeaver.Mesh;
+using Xunit;
+
+namespace MeshWeaver.GitSync.Test;
+
+/// <summary>
+/// An activity must never outlive the mesh that started it.
+///
+/// <para>The compile that crashed CI on 2026-08-04 (FutuRe.Test exit=139, ~1 run in 5) is one
+/// instance of this, but the invariant is general and does not need Roslyn to state: work running
+/// inside an activity is work the mesh owns, so disposal must not return while it is still in
+/// flight. If it does, the work keeps running against a torn-down mesh — and when that work is
+/// emitting or loading a collectible assembly, the ALC unloads underneath it and the process dies
+/// on a bare SIGSEGV with no managed exception.</para>
+///
+/// <para>Why an activity is the right lifecycle handle: it is a normal hub running inside the
+/// message-hub grain, and its ActivityLog reaching a terminal status (Succeeded OR Failed) is the
+/// documented signal that it has finished and the hub is free to deactivate
+/// (see CompileFinishAndDisposeTest). Nothing else covers it — the activity command deliberately
+/// runs OFF the hub turn (ScheduleOffHubTurn), so it holds no grain lock, and a compile
+/// additionally leaves the drainable I/O pool for a dedicated thread, so the pool's
+/// "teardown cancels + joins in-flight work" guarantee does not reach it either.</para>
+///
+/// <para>🚨 The fix this drives must be CANCEL-THEN-JOIN and must NOT block the action block: the
+/// activity's own Append/Finish writes go back through the hub, so a naive blocking join deadlocks
+/// against the work it is waiting for — the same class as holding the grain turn across
+/// <c>ctx.Log</c>.</para>
+/// </summary>
+public class ActivityOutlivesDisposeTest(ITestOutputHelper output) : GitHubSyncTestBase(output)
+{
+    /// <summary>
+    /// A deliberately slow activity — a plain delay, no compile, no I/O — started and NOT awaited.
+    /// Disposal must observe it to terminal before returning.
+    /// </summary>
+    [Fact(Timeout = 120_000)]
+    public async Task DisposingTheMesh_WaitsForARunningActivity()
+    {
+        var space = "GhDispose" + Guid.NewGuid().ToString("N")[..8];
+        await CreateSpace(space, "Dispose-race space");
+
+        // Reactive signals, not Task primitives: the command entering, and the command finishing.
+        var entered = new AsyncSubject<Unit>();
+        var completed = 0;
+
+        // Start the activity and DO NOT await it. The trigger returns as soon as the activity
+        // exists — by design — so this is precisely the window under test.
+        Mesh.RunActivity(space, ActivityCategory.Import, "Slow probe",
+                ctx =>
+                {
+                    entered.OnNext(Unit.Default);
+                    entered.OnCompleted();
+                    return Observable.Timer(TimeSpan.FromSeconds(5))
+                        .Do(_ => Interlocked.Exchange(ref completed, 1))
+                        .Select(_ => Unit.Default);
+                })
+            .Subscribe(_ => { }, ex => Output.WriteLine($"activity faulted: {ex.Message}"));
+
+        // Wait for the command to be genuinely running, or the test proves nothing. Reactive
+        // assertion — FirstAsync() blocks, so it has no place even here.
+        await entered.Should().Within(TimeSpan.FromSeconds(30)).Emit();
+
+        // The REAL teardown path. A bare Mesh.Dispose() runs only the synchronous half; the
+        // documented contract is TeardownAsync — await DisposalCompleted, DrainAll() the pools,
+        // then quiesce the AsyncDisposeQueue.
+        var started = DateTimeOffset.UtcNow;
+        await Mesh.TeardownAsync(TimeSpan.FromSeconds(60));
+        var elapsed = DateTimeOffset.UtcNow - started;
+        Output.WriteLine($"TeardownAsync returned after {elapsed.TotalMilliseconds:F0}ms; "
+                         + $"command completed={Volatile.Read(ref completed) == 1}");
+
+        // THE CONTRACT: teardown quiesces — it stops new work and waits for what is running to
+        // finish. It must not return while an activity command is still live: that work keeps
+        // executing against a torn-down mesh, and when it touches a collectible ALC's metadata
+        // after unload the process dies on a bare signal with no managed exception.
+        Volatile.Read(ref completed).Should().Be(1,
+            $"TeardownAsync returned after {elapsed.TotalMilliseconds:F0}ms while the activity "
+            + "command was still running. Activity work is mesh-owned: teardown must wait for it, "
+            + "never walk away from it.");
+    }
+}