fix(tests): wait on real signals in three flaky channel, headless, and SQLite recovery tests

src/Netclaw.Actors.Tests/Channels/ChannelBotFilterOrderTests.cs

@@ -116,25 +116,34 @@ public async Task Mattermost_person_denied_by_ACL_is_dropped()
         await AssertDroppedNotFiltered(ChannelType.Mattermost);
     }
 
+    // RecordEventDropped and RecordEventFiltered bump the snapshot total first. The meter listener then runs and
+    // records the reason. So a wait on the total does not prove that the reason exists. Each helper waits on the
+    // reason itself, and only then checks the negative side.
     private async Task AssertFilteredNotDropped(ChannelType channelType)
     {
         var metrics = ChannelTelemetry.For(channelType);
         await AwaitAssertAsync(
-            () => Assert.Equal(1, metrics.GetSnapshot().EventsFiltered),
+            () =>
+            {
+                Assert.Equal(1, metrics.GetSnapshot().EventsFiltered);
+                Assert.Equal(["bot_message"], Reasons("events.filtered"));
+            },
             cancellationToken: TestContext.Current.CancellationToken);
         Assert.Equal(0, metrics.GetSnapshot().EventsDropped);
-        Assert.Equal(["bot_message"], Reasons("events.filtered"));
         Assert.Empty(Reasons("events.dropped"));
     }
 
     private async Task AssertDroppedNotFiltered(ChannelType channelType)
     {
         var metrics = ChannelTelemetry.For(channelType);
         await AwaitAssertAsync(
-            () => Assert.Equal(1, metrics.GetSnapshot().EventsDropped),
+            () =>
+            {
+                Assert.Equal(1, metrics.GetSnapshot().EventsDropped);
+                Assert.Equal([AclDenyReasons.UserNotAllowed], Reasons("events.dropped"));
+            },
             cancellationToken: TestContext.Current.CancellationToken);
         Assert.Equal(0, metrics.GetSnapshot().EventsFiltered);
-        Assert.Equal([AclDenyReasons.UserNotAllowed], Reasons("events.dropped"));
         Assert.Empty(Reasons("events.filtered"));
     }
 

src/Netclaw.Actors.Tests/Sessions/ChildParkSqliteRecoveryTests.cs

@@ -5,7 +5,6 @@
 // -----------------------------------------------------------------------
 using Akka.Hosting;
 using Akka.Persistence.Sql.Hosting;
-using Microsoft.Data.Sqlite;
 using Netclaw.Actors.Protocol;
 using Netclaw.Actors.SubAgents;
 using Netclaw.Tests.Utilities;
@@ -32,9 +31,14 @@ protected override void ConfigureSessionPersistence(AkkaConfigurationBuilder bui
 
     protected override void DeleteOwnedDirectories()
     {
-        SqliteConnection.ClearAllPools();
-        _sqliteDirectory?.DisposeAsync().AsTask().GetAwaiter().GetResult();
-        base.DeleteOwnedDirectories();
+        try
+        {
+            SqlPersistencePlugins.WaitForStopThenDelete(Sys, _sqliteDirectory);
+        }
+        finally
+        {
+            base.DeleteOwnedDirectories();
+        }
     }
 
     /// <summary>
@@ -48,7 +52,8 @@ public async Task A_parked_run_survives_a_SQLite_journal_and_snapshot_restart()
     {
         // Arrange
         var id = new SessionId("park/sqlite");
-        await SqlPersistencePlugins.StartOneAtATimeAsync(this, ColdRecoveryCeiling);
+        await SqlPersistencePlugins.StartOneAtATimeAsync(this, ColdRecoveryCeiling,
+            Path.Combine(_sqliteDirectory!.Path, "journal.db"));
         ScriptParent();
         PlanChild("park-sqlite", [[Probe("p1")]]);
         await JoinRecorderAsync(id);

src/Netclaw.Actors.Tests/Sessions/LegacyCoordinationSqliteRecoveryTests.cs

@@ -61,7 +61,7 @@ await manager.Ask<CommandAck>(new SendUserMessage
         {
             SessionId = sessionId,
             Content = "Continue after the recovered child result."
-        }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+        }, ColdRecoveryCeiling, TestContext.Current.CancellationToken);
         await subscriber.FishForMessageAsync<TurnCompleted>(_ => true, TimeSpan.FromSeconds(10),
             cancellationToken: TestContext.Current.CancellationToken);
 
@@ -111,7 +111,7 @@ await manager.Ask<CommandAck>(new SendUserMessage
         {
             SessionId = sessionId,
             Content = "Continue after the quarantined legacy snapshot."
-        }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+        }, ColdRecoveryCeiling, TestContext.Current.CancellationToken);
         await subscriber.FishForMessageAsync<TurnCompleted>(_ => true, TimeSpan.FromSeconds(10),
             cancellationToken: TestContext.Current.CancellationToken);
 
@@ -157,7 +157,7 @@ await manager.Ask<CommandAck>(new SendUserMessage
         {
             SessionId = sessionId,
             Content = "Continue after the notice."
-        }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+        }, ColdRecoveryCeiling, TestContext.Current.CancellationToken);
         await subscriber.FishForMessageAsync<TurnCompleted>(_ => true, TimeSpan.FromSeconds(10),
             cancellationToken: TestContext.Current.CancellationToken);
         var later = _chatClient.ReceivedMessages.Last();

src/Netclaw.Actors.Tests/Sessions/RecordedJournalTestBase.cs

@@ -56,10 +56,7 @@ protected sealed override void ConfigureSessionSerialization(AkkaConfigurationBu
             .WithStrictSerialization();
 
     protected override void DeleteOwnedDirectories()
-    {
-        SqliteConnection.ClearAllPools();
-        _sqliteDirectory?.DisposeAsync().AsTask().GetAwaiter().GetResult();
-    }
+        => SqlPersistencePlugins.WaitForStopThenDelete(Sys, _sqliteDirectory);
 
     /// <summary>Writes the named corpus records into the journal of one persistence id, in order.</summary>
     protected async Task SeedJournalAsync(string persistenceId, string seederName, params string[] recordNames)
@@ -96,13 +93,18 @@ private async Task StopSeederAsync(IActorRef seeder)
     /// </summary>
     private async Task<IActorRef> StartSeederAsync(string persistenceId, string actorName)
     {
-        await SqlPersistencePlugins.StartOneAtATimeAsync(this, ColdRecoveryCeiling);
+        await SqlPersistencePlugins.StartOneAtATimeAsync(this, ColdRecoveryCeiling, SqliteDatabasePath());
         var recovered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         var seeder = Sys.ActorOf(Props.Create(() => new RecordedRowSeeder(persistenceId, recovered)), actorName);
         await recovered.Task.WaitAsync(ColdRecoveryCeiling, TestContext.Current.CancellationToken);
         return seeder;
     }
 
+    private string SqliteDatabasePath()
+        => Path.Combine(
+            _sqliteDirectory?.Path ?? throw new InvalidOperationException("The test SQLite directory was not configured."),
+            "journal.db");
+
     private void BackupAndRestoreSqliteDatabase()
     {
         var directory = _sqliteDirectory?.Path
@@ -258,17 +260,89 @@ internal static class SqlPersistencePlugins
     /// fight. That timeout is equal to the cold ceiling, so a long fight fails the start wait with no cause (#2599).
     /// One plugin at a time has no fight. A warm plugin answers at once.
     /// </summary>
-    public static async Task StartOneAtATimeAsync(Akka.TestKit.TestKitBase kit, TimeSpan ceiling)
+    public static async Task StartOneAtATimeAsync(Akka.TestKit.TestKitBase kit, TimeSpan ceiling, string databasePath)
     {
         const string unusedId = "recorded-journal-plugin-start";
         var persistence = AkkaPersistence.Instance.Apply(kit.Sys);
         var probe = kit.CreateTestProbe();
-        persistence.JournalFor(null).Tell(new ReplayMessages(0, long.MaxValue, long.MaxValue, unusedId, probe.Ref));
-        await probe.ExpectMsgAsync<RecoverySuccess>(ceiling,
-            cancellationToken: TestContext.Current.CancellationToken);
-        persistence.SnapshotStoreFor(null).Tell(
+        var journal = persistence.JournalFor(null);
+        journal.Tell(new ReplayMessages(0, long.MaxValue, long.MaxValue, unusedId, probe.Ref));
+        await ExpectStartReplyAsync<RecoverySuccess>(probe, ceiling, "journal", journal, databasePath);
+        var snapshotStore = persistence.SnapshotStoreFor(null);
+        snapshotStore.Tell(
             new LoadSnapshot(unusedId, SnapshotSelectionCriteria.Latest, long.MaxValue), probe.Ref);
-        await probe.ExpectMsgAsync<LoadSnapshotResult>(ceiling,
-            cancellationToken: TestContext.Current.CancellationToken);
+        await ExpectStartReplyAsync<LoadSnapshotResult>(probe, ceiling, "snapshot store", snapshotStore, databasePath);
+    }
+
+    // The plugin that does not reply is the one that failed to start. The message names it and the database file,
+    // so a stall shows where it happened. A plain timeout would not say which plugin blocked.
+    private static async Task ExpectStartReplyAsync<TReply>(
+        Akka.TestKit.TestProbe probe, TimeSpan ceiling, string plugin, IActorRef pluginActor, string databasePath)
+    {
+        try
+        {
+            await probe.ExpectMsgAsync<TReply>(ceiling, cancellationToken: TestContext.Current.CancellationToken);
+        }
+        catch (Exception ex) when (ex is not OperationCanceledException)
+        {
+            throw new InvalidOperationException(
+                $"The SQL {plugin} plugin at {pluginActor.Path} did not send {typeof(TReply).Name} within "
+                + $"{ceiling.TotalSeconds:0} s. The plugins start in this order: journal, snapshot store. "
+                + $"Database: {databasePath} (file exists: {File.Exists(databasePath)}).", ex);
+        }
+    }
+
+    // A fault ceiling for the stop of the actor system. It is not an orchestration delay.
+    private static readonly TimeSpan TerminationCeiling = TimeSpan.FromSeconds(15);
+
+    private const int DeleteAttempts = 3;
+
+    /// <summary>
+    /// Waits for the actor system to terminate, and then deletes the SQLite folder. The journal and the snapshot store
+    /// are actors of the system. Each one holds a connection until it stops. A plugin that blocks in a SQL call
+    /// cannot stop, and its connection stays open. Windows then refuses the delete with "journal.db in use". The wait is
+    /// bounded. If it ends first, the method still tries the delete, and then reports the stall. The seeders of the
+    /// base run in the same system, so this one wait covers them.
+    /// </summary>
+    public static void WaitForStopThenDelete(ActorSystem system, TestSessionTempDirectory? directory)
+    {
+        TimeoutException? stall = null;
+        if (!system.WhenTerminated.Wait(TerminationCeiling))
+        {
+            stall = new TimeoutException(
+                $"The actor system did not terminate within {TerminationCeiling.TotalSeconds:0} s. "
+                + "A SQL persistence plugin actor probably still holds its connection.");
+        }
+
+        try
+        {
+            DeleteWithPoolClear(directory);
+        }
+        catch (IOException ex) when (stall is not null)
+        {
+            throw new AggregateException(stall, ex);
+        }
+
+        if (stall is not null)
+            throw stall;
+    }
+
+    // Each attempt clears the pools first. A connection that a plugin returns late goes back to the pool, and only
+    // a new clear closes it. DisposableTempDir retries the delete between the attempts.
+    [SlopwatchSuppress("SW003", "A failed delete attempt is retried; the last attempt throws.")]
+    private static void DeleteWithPoolClear(TestSessionTempDirectory? directory)
+    {
+        for (var attempt = 1; attempt <= DeleteAttempts; attempt++)
+        {
+            SqliteConnection.ClearAllPools();
+            try
+            {
+                directory?.DisposeAsync().AsTask().GetAwaiter().GetResult();
+                return;
+            }
+            catch (IOException) when (attempt < DeleteAttempts)
+            {
+            }
+        }
     }
 }

src/Netclaw.Cli.MutationTests/HeadlessStopRuleMutationTests.cs

@@ -30,8 +30,13 @@ public sealed class HeadlessStopRuleMutationTests
     private const string Session = "fake/session";
 
     // A fault ceiling. A run that never stops fails here. No test orchestrates with this wait.
+    // It starts after the send. Stryker pays it in full for each mutant that never stops, so keep it short.
     private static readonly TimeSpan StopCeiling = TimeSpan.FromSeconds(10);
 
+    // A ceiling for the cold start of the first run in the process. It covers the session create, the first
+    // log file, and the first send. A loaded Windows runner needs more than 10 s for this work.
+    private static readonly TimeSpan ColdStartCeiling = TimeSpan.FromSeconds(30);
+
     [Fact]
     public async Task An_open_run_keeps_the_wait_until_a_turn_reports_none()
     {
@@ -103,10 +108,12 @@ private static async Task<Run> RunAsync(params SessionOutputDto[] outputs)
     {
         var directory = Path.Combine(Path.GetTempPath(), "netclaw-headless-stop-rule-" + Guid.NewGuid().ToString("N"));
         var transport = new FakeDaemonHubTransport();
+        var sent = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         transport.VoidInvokeHook = (method, _, _) =>
         {
             if (method == "SendMessage")
             {
+                sent.TrySetResult();
                 // The client pump delivers the outputs in order, so the channel sees every turn.
                 foreach (var output in outputs)
                     transport.PushOutput(output);
@@ -137,7 +144,28 @@ private static async Task<Run> RunAsync(params SessionOutputDto[] outputs)
         try
         {
             await channel.StartAsync(CancellationToken.None);
-            await stopped.Task.WaitAsync(StopCeiling);
+            // Phase one: the cold start. The ceiling of the stop rule does not cover it.
+            try
+            {
+                await sent.Task.WaitAsync(ColdStartCeiling);
+            }
+            catch (TimeoutException)
+            {
+                throw new TimeoutException(
+                    $"Cold start phase: the channel did not send the message within {ColdStartCeiling.TotalSeconds:0} s.");
+            }
+
+            // Phase two: the stop rule. The outputs are already queued, so the run stops soon after the send.
+            try
+            {
+                await stopped.Task.WaitAsync(StopCeiling);
+            }
+            catch (TimeoutException)
+            {
+                throw new TimeoutException(
+                    $"Stop phase: the run did not stop within {StopCeiling.TotalSeconds:0} s after the send.");
+            }
+
             await channel.StopAsync(CancellationToken.None);
             Console.SetOut(originalOut);
             Console.SetError(originalError);