Add Slopwatch & fix flaky tests

Author: st0o0

.claude/settings.json

@@ -0,0 +1,16 @@
+{
+  "hooks": {
+    "PostToolUse": [
+      {
+        "matcher": "Write|Edit|MultiEdit",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "slopwatch analyze -d $(git rev-parse --show-toplevel) --hook",
+            "timeout": 60000
+          }
+        ]
+      }
+    ]
+  }
+}

.config/dotnet-tools.json

@@ -0,0 +1,11 @@
+{
+  "version": 1,
+  "isRoot": true,
+  "tools": {
+    "slopwatch.cmd": {
+      "version": "0.2.0",
+      "commands": ["slopwatch"],
+      "rollForward": false
+    }
+  }
+}

.slopwatch/baseline.json

@@ -0,0 +1,16 @@
+{
+  "minSeverity": "warning",
+  "rules": {
+    "SW001": { "enabled": true, "severity": "error" },
+    "SW002": { "enabled": true, "severity": "error" },
+    "SW003": { "enabled": true, "severity": "error" },
+    "SW004": { "enabled": true, "severity": "error" },
+    "SW005": { "enabled": true, "severity": "error" },
+    "SW006": { "enabled": true, "severity": "error" }
+  },
+  "exclude": [
+    "**/Generated/**",
+    "**/obj/**",
+    "**/bin/**"
+  ]
+}

.slopwatch/slopwatch.json

@@ -222,7 +222,7 @@ public async Task Cache_should_produce_different_cache_entries_per_vary_header_v
         Assert.Equal(body1, body3);
     }
 
-    [Fact(Timeout = 5000)]
+    [Fact(Timeout = 10_000)]
     [Trait("RFC", "RFC9111-5.2.2.2")]
     public async Task Cache_should_force_revalidation_with_must_revalidate()
     {

src/TurboHTTP.AcceptanceTests/H11/CacheSpec.cs

@@ -222,7 +222,7 @@ public async Task Cache_should_produce_different_cache_entries_per_vary_header_v
         Assert.Equal(body1, body3);
     }
 
-    [Fact(Timeout = 5000)]
+    [Fact(Timeout = 10_000)]
     [Trait("RFC", "RFC9111-5.2.2.2")]
     public async Task Cache_should_force_revalidation_with_must_revalidate_over_https()
     {

src/TurboHTTP.AcceptanceTests/TLS/CacheSpec.cs

@@ -96,8 +96,10 @@ public async Task CacheBidiStage_should_push_response_immediately_while_body_rea
         var result = Assert.Single(results);
         Assert.Equal(HttpStatusCode.OK, result.StatusCode);
 
-        // Allow the PipeTo callback to fire
-        await Task.Delay(200, TestContext.Current.CancellationToken);
+        // Poll until the PipeTo callback fires and cache is populated
+        await AwaitAssertAsync(
+            () => Assert.NotNull(store.Get(new HttpRequestMessage(HttpMethod.Get, "http://example.com/slow"))),
+            TimeSpan.FromSeconds(2), cancellationToken: TestContext.Current.CancellationToken);
 
         var entry = store.Get(new HttpRequestMessage(HttpMethod.Get, "http://example.com/slow"));
         Assert.NotNull(entry);
@@ -129,8 +131,10 @@ public async Task CacheBidiStage_should_store_in_cache_after_async_body_complete
         var result = Assert.Single(results);
         Assert.Equal(HttpStatusCode.OK, result.StatusCode);
 
-        // Allow the PipeTo callback to fire
-        await Task.Delay(200, TestContext.Current.CancellationToken);
+        // Poll until the PipeTo callback fires and cache is populated
+        await AwaitAssertAsync(
+            () => Assert.NotNull(store.Get(new HttpRequestMessage(HttpMethod.Get, "http://example.com/pending"))),
+            TimeSpan.FromSeconds(2), cancellationToken: TestContext.Current.CancellationToken);
 
         var entry = store.Get(new HttpRequestMessage(HttpMethod.Get, "http://example.com/pending"));
         Assert.NotNull(entry);

src/TurboHTTP.StreamTests/Caching/CacheBidiAsyncBodySpec.cs

@@ -4,6 +4,7 @@
 using TurboHTTP.Diagnostics;
 using TurboHTTP.Streams.Stages.Features;
 using TurboHTTP.Tests.Shared;
+using Activity = System.Diagnostics.Activity;
 using ActivityListener = System.Diagnostics.ActivityListener;
 using ActivitySamplingResult = System.Diagnostics.ActivitySamplingResult;
 using ActivitySource = System.Diagnostics.ActivitySource;
@@ -17,10 +18,20 @@ public sealed class TracingActivityLeakSpec : StreamTestBase
     public async Task TracingBidiStage_should_stop_activity_when_stage_tears_down_without_response()
     {
         var stoppedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        Activity? capturedActivity = null;
 
         using var listener = new ActivityListener();
         listener.ShouldListenTo = source => source.Name == TurboHttpInstrumentation.SourceName;
         listener.Sample = (ref _) => ActivitySamplingResult.AllData;
+        // Wire ActivityStopped before AddActivityListener so the callback is always
+        // registered before the Akka dispatch thread can call PostStop.
+        listener.ActivityStopped = stopped =>
+        {
+            if (capturedActivity is { } a && ReferenceEquals(stopped, a))
+            {
+                stoppedTcs.TrySetResult();
+            }
+        };
         ActivitySource.AddActivityListener(listener);
 
         var stage = new TracingBidiStage();
@@ -55,13 +66,7 @@ public async Task TracingBidiStage_should_stop_activity_when_stage_tears_down_wi
         Assert.True(forwarded.Options.TryGetValue(TurboHttpInstrumentation.RequestActivityKey, out var activity));
         Assert.NotNull(activity);
 
-        listener.ActivityStopped = stopped =>
-        {
-            if (ReferenceEquals(stopped, activity))
-            {
-                stoppedTcs.TrySetResult();
-            }
-        };
+        capturedActivity = activity;
 
         // Tear down the stage — complete both inlets and cancel downstream
         reqInSub.SendComplete();

src/TurboHTTP.StreamTests/Semantics/TracingActivityLeakSpec.cs

@@ -251,7 +251,7 @@ public async Task ConnectionStage_should_release_with_no_reuse_when_connection_r
         var decision = ConnectionReuseDecision.Close("Connection: close");
         var reuseItem = new ConnectionReuseItem(decision) { Key = TestKey };
         await inputQueue.OfferAsync(reuseItem);
-        await Task.Delay(300, TestContext.Current.CancellationToken);
+        AwaitCondition(() => tracker.Released, TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken);
 
         Assert.False(lease.Reusable);
         Assert.True(tracker.Released);
@@ -366,11 +366,9 @@ public async Task ConnectionStage_should_survive_and_continue_when_data_item_arr
 
         var data = MakeData(0xFF);
         await inputQueue.OfferAsync(data);
-        await Task.Delay(200, TestContext.Current.CancellationToken);
 
         var data2 = MakeData(0xEE);
         await inputQueue.OfferAsync(data2);
-        await Task.Delay(200, TestContext.Current.CancellationToken);
 
         inputQueue.Complete();
         var results = await outputTask.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
@@ -409,7 +407,7 @@ public async Task
 
         var data = MakeData(0xBB);
         await inputQueue.OfferAsync(data);
-        await Task.Delay(300, TestContext.Current.CancellationToken);
+        AwaitCondition(() => tracker.Released, TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken);
 
         Assert.True(tracker.Released);
         Assert.False(tracker.ReleasedCanReuse);

src/TurboHTTP.StreamTests/Streams/ConnectionStageSpec.cs

@@ -136,8 +136,11 @@ public async Task GroupByEndpointFanOut_should_route_to_same_slot_when_request_h
         await queue.OfferAsync(firstRequest)
             .WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
 
-        // Give the stage time to process the push and stamp the affinity tag.
-        await Task.Delay(TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
+        // Poll until the stage stamps the affinity tag.
+        AwaitCondition(
+            () => firstRequest.Options.TryGetValue(
+                GroupByRequestEndpointStage<HttpRequestMessage>.ConnectionAffinitySlot, out _), TimeSpan.FromSeconds(2),
+            TestContext.Current.CancellationToken);
 
         // Read back the slot ID stamped by the stage.
         var hasTag = firstRequest.Options.TryGetValue(