diff --git a/.claude/skills/sigsegv/SKILL.md b/.claude/skills/sigsegv/SKILL.md index d6eb4fdef5..dd04d3488a 100644 --- a/.claude/skills/sigsegv/SKILL.md +++ b/.claude/skills/sigsegv/SKILL.md @@ -189,8 +189,10 @@ The ordering traps that produce this shape: - **The Orleans silo is itself a hosted service**, so it stops before the mesh drains: `silo stops → grains deactivate → every Unload() → only THEN the mesh drains`. `IoPoolSiloTeardown` subscribes at `ServiceLifecycleStage.First` ⇒ **stops LAST**, and composes - `registry.Disposed` into **one** `.ToTask()` at the `ILifecycleObserver` boundary — the only - sanctioned place a Task appears. + `registry.Disposed` into **one** `ReactiveCompletion.ObserveCompletion(reportLateFault, ct)` at + the `ILifecycleObserver` boundary — the only sanctioned place a Task appears, and never Rx's + `.ToTask()` (forbidden repo-wide, 2026-08-30): that bridge resumes its caller INLINE on the + signalling thread — here, the very lifecycle thread the unload is waiting on. - **A blocking `Dispose()` is not the fix.** Making `IoPool.Dispose()` join first *hangs* — `using var pool = …` in an async method runs Dispose on a ThreadPool thread and blocks it for the whole `DrainTimeout`. An 18-core dev box hides it completely. The shape that works: diff --git a/src/MeshWeaver.Documentation/Data/Architecture/AccessContextPropagation.md b/src/MeshWeaver.Documentation/Data/Architecture/AccessContextPropagation.md index d380d2bba8..4b0eda128d 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/AccessContextPropagation.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/AccessContextPropagation.md @@ -456,9 +456,14 @@ public async Task MeshNodeCache_Identity_CannotWrite() { using (accessService.SwitchAccessContext(new AccessContext { ObjectId = "cache/mesh-node-cache" })) { - var act = () => meshService.CreateNode(someNode).ToTask(); - await act.Should().ThrowAsync() - .Where(ex => ex.Message.Contains("Access denied")); + // 🚨 Materialize, not act.Should().ThrowAsync() over an Rx ToTask bridge: + // ToTask is forbidden repo-wide (2026-08-30), and folding OnError into a + // value keeps the ORIGINAL exception type assertable. The write is cold, so + // the assertion's Subscribe IS the write. + var error = await meshService.CreateNode(someNode).Take(1).Materialize() + .Should().Match(n => n.Kind == NotificationKind.OnError); + error.Exception.Should().BeOfType(); + error.Exception!.Message.Should().Contain("Access denied"); } } ``` diff --git a/src/MeshWeaver.Documentation/Data/Architecture/AccessControl.md b/src/MeshWeaver.Documentation/Data/Architecture/AccessControl.md index 86c7647b87..45af915e6b 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/AccessControl.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/AccessControl.md @@ -1050,8 +1050,13 @@ public async Task MeshNodeCacheIdentity_CannotWrite() { using (accessService.SwitchAccessContext(new AccessContext { ObjectId = "cache/mesh-node-cache" })) { - var act = () => meshService.CreateNode(someNode).ToTask(); - await act.Should().ThrowAsync(); + // 🚨 Materialize, not act.Should().ThrowAsync() over an Rx ToTask bridge: + // ToTask is forbidden repo-wide (2026-08-30), and folding OnError into a + // value keeps the ORIGINAL exception type assertable. The write is cold, so + // the assertion's Subscribe IS the write. + var error = await meshService.CreateNode(someNode).Take(1).Materialize() + .Should().Match(n => n.Kind == NotificationKind.OnError); + error.Exception.Should().BeOfType(); } } ``` diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ActivityControlPlane.md b/src/MeshWeaver.Documentation/Data/Architecture/ActivityControlPlane.md index 9c63eb2844..69f3b63f9f 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ActivityControlPlane.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ActivityControlPlane.md @@ -465,10 +465,18 @@ The pattern inside scripts (and inside any worker hub doing long-running work): await Mesh.GetWorkspace() .GetMeshNodeStream("rbuergi/downstream-job") .Where(n => (n?.Content as JobContent)?.Status == JobStatus.Succeeded) - .Take(1) - .ToTask(Ct); // ← cancels via Activity Control Plane + .FirstAsync() // ← not .Take(1): an empty completion must FAULT + .ObserveCompletion( // rather than settle silently + ex => Log.LogWarning(ex, "downstream watch faulted AFTER the wait settled"), + Ct); // ← cancels via Activity Control Plane ``` + 🚨 `ObserveCompletion` (`MeshWeaver.Messaging`, imported by default in scripts), never + `.ToTask(Ct)` — forbidden repo-wide as of 2026-08-30 — and never a bare `await someObservable` + either. Rx's bridge and Rx's own awaiter both resume the rest of the script INLINE on the mesh + thread that signalled; `ObserveCompletion` completes with `RunContinuationsAsynchronously`, so the + hub thread is released the moment it signals. + - For compute-heavy steps with no natural log lines, an `Observable.Interval` heartbeat keeps the user informed: ```csharp @@ -548,16 +556,24 @@ var md = (src?.Content as MarkdownContent)?.Content ?? ""; Log.LogInformation("Resolving branding"); var branding = await Mesh.GetWorkspace() .GetMeshNodeStream(inputs.BrandNodePath) - .Where(n => n is not null).Take(1) + .Where(n => n is not null) .Select(n => (n!.Content as CorporateIdentity).ToOptions()) - .ToTask(Ct); // top-level await is OK in a kernel Script + .FirstAsync() + // A top-level await is fine in a kernel Script; the BRIDGE is what is constrained — + // ObserveCompletion, never Rx's ToTask and never a bare `await someObservable`. + .ObserveCompletion( + ex => Log.LogWarning(ex, "branding read faulted AFTER the wait settled"), + Ct); Log.LogInformation("Rendering"); var doc = new DocumentBuilder().Build(src!.Name, [(src.Name, md)], inputs.Options, branding); // The PDF renderer composes the document into print HTML and prints it with the headless // browser, so it returns a COLD IObservable — the work runs on subscription. var bytes = await Mesh.ServiceProvider.GetRequiredService() - .Render(doc).FirstAsync().ToTask(Ct); + .Render(doc).FirstAsync() + .ObserveCompletion( + ex => Log.LogWarning(ex, "PDF render faulted AFTER the wait settled"), + Ct); Log.LogInformation("Writing {Bytes} bytes to content collection", bytes.Length); var outputPath = $"{inputs.TargetCollection}/{Sanitize(src.Name)}.pdf"; diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ActorModel.md b/src/MeshWeaver.Documentation/Data/Architecture/ActorModel.md index fdac2a44e1..61dd820963 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ActorModel.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ActorModel.md @@ -143,7 +143,7 @@ This is not a timing edge case — it is a structural certainty whenever a handl ## The Fix: Reactive Streams (the Modern API) -> **The `AwaitResponse` and `RegisterCallback` APIs are gone** — they no longer exist anywhere in `src/`; `hub.Observe(...)` is the only request/response surface. All hub-reachable code must use `IObservable` end-to-end — no `await`, no `Task`, no `TaskCompletionSource`. Tests may use `.FirstAsync().ToTask()` at the boundary; production code never does. +> **The `AwaitResponse` and `RegisterCallback` APIs are gone** — they no longer exist anywhere in `src/`; `hub.Observe(...)` is the only request/response surface. All hub-reachable code must use `IObservable` end-to-end — no `await`, no `Task`, no `TaskCompletionSource`. **Tests are not exempt either**: `.FirstAsync().ToTask()` is forbidden everywhere as of 2026-08-30 (*"no ToTask ever"*). Tests assert on the observable through `MeshWeaver.Reactive.Assertions` (`await obs.Should().Match(...)`), which owns the wait; a foreign `Task`-shaped signature that genuinely needs the value uses `ReactiveCompletion.ObserveCompletion(reportLateFault, ct)`. The correct pattern for any write that produces a side effect is `stream.Update(...)`, which returns a cold `IObservable`. Subscribe in the handler; the framework serialises the write through the owning hub's action block without blocking anything. @@ -210,7 +210,7 @@ The response arrives as an observable emission, and the subscriber runs on the * | Mutate a node on a different hub | Same API — `GetMeshNodeStream` auto-dispatches cross-hub | | Wait for a state change | `GetMeshNodeStream(path).Where(predicate).Take(1).Timeout(...).Subscribe(...)` | | Call a different hub and handle the reply | `hub.Observe(request, o => o.WithTarget(addr)).Subscribe(...)` | -| Test boundary only | `await stream.FirstAsync().ToTask()` is acceptable | +| Test boundary | `await stream.Should().Match(predicate)` — the assertion owns the wait. **Never** `.FirstAsync().ToTask()`, and never a bare `await stream` | --- diff --git a/src/MeshWeaver.Documentation/Data/Architecture/AggregatingProviders.md b/src/MeshWeaver.Documentation/Data/Architecture/AggregatingProviders.md index 09b6c288a6..c15f27fc11 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/AggregatingProviders.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/AggregatingProviders.md @@ -148,16 +148,16 @@ For a request/response consumer (cross-hub `AutocompleteRequest`), take the sett ### Testing progressive-snapshot providers -Tests may bridge to `await` at the test edge — but bound it, and wait for the *shape* you expect rather than the first emission (the first is the empty seed): +Tests assert on the stream and let the assertion own the wait — but bound it, and wait for the *shape* you expect rather than the first emission (the first is the empty seed): ```csharp var items = await provider.GetItems("@Sys", null) - .Where(snapshot => snapshot.Count > 0) - .Timeout(10.Seconds()) - .FirstAsync() - .ToTask(ct); + .Should().Within(10.Seconds()) + .Match(snapshot => snapshot.Count > 0); ``` +> **🚨 Not `.FirstAsync().ToTask(ct)`.** That bridge is forbidden repo-wide as of 2026-08-30, `test/**` included, and a bare `await observable` is the same defect: both resume the test inline on the mesh thread that emitted. `.Within(t)` is the bound, which is why there is no `Timeout` operator and no `CancellationToken` left in the chain. + --- ## Reactive snapshot-set providers (node menus, permission-gated panels) @@ -222,10 +222,8 @@ Because the menu re-emits as permissions enrich, a test must **not** grab the fi ```csharp var items = await MenuStream(client, nodeAddress, NodeMenuContext) .CombineLatest(MenuStream(client, nodeAddress, MeshMenuContext), Merge) - .Where(set => set.Select(i => i.Label).ToHashSet().SetEquals(expectedLabels)) - .Timeout(20.Seconds()) - .FirstAsync() - .ToTask(ct); + .Should().Within(20.Seconds()) + .Match(set => set.Select(i => i.Label).ToHashSet().SetEquals(expectedLabels)); ``` `SetEquals` waiting catches both *missing* items (role not yet propagated) and *extra* items (wrong gating) — either way the menu never reaches the expected set and the `Timeout` fails the test. See `MenuAccessControlTest`. @@ -312,4 +310,4 @@ Any new aggregator that gathers items from multiple providers should look like o **Both:** - [ ] Providers are resolved via `hub.ServiceProvider.GetServices()` after `TryAddEnumerable` (or, for menu delegate providers, registered via `AddNodeMenuItems`). -- [ ] Tests bridge back to `await` via `.Where(predicate).Timeout(...).FirstAsync().ToTask(ct)` — waiting for the expected shape, never the first emission, and never `.ToTask()` on a raw hub-touching observable without a bounding `Timeout`. +- [ ] Tests wait via `await stream.Should().Within(t).Match(predicate)` — for the expected shape, never the first emission. **No `.ToTask()` anywhere** (forbidden repo-wide, 2026-08-30) and no bare `await observable`; `.Within(t)` is the bound. diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ControlledIoPooling.md b/src/MeshWeaver.Documentation/Data/Architecture/ControlledIoPooling.md index d1623e3171..b0f3a088fe 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ControlledIoPooling.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ControlledIoPooling.md @@ -483,9 +483,18 @@ await mesh.TeardownAsync(TimeSpan.FromSeconds(15)); // MeshWeaver.Mesh.MeshTea var ioPools = mesh.ServiceProvider.GetService(); // capture BEFORE Dispose() var disposeQueue = mesh.ServiceProvider.GetService(); mesh.Dispose(); +// 🚨 ObserveCompletion, never Rx's ToTask bridge (forbidden repo-wide, 2026-08-30) and +// never a bare `await disposalCompleted` either: both resume this method INLINE on the +// hub's own disposal thread, which then has to run phases 2 and 3 while the mesh is +// trying to finish tearing itself down. ObserveCompletion completes with +// RunContinuationsAsynchronously, so the disposing thread is released immediately. +using var phase1Deadline = new CancellationTokenSource(TimeSpan.FromSeconds(15)); await mesh.DisposalCompleted .Catch(_ => Observable.Return(Unit.Default)) - .FirstOrDefaultAsync().ToTask().WaitAsync(TimeSpan.FromSeconds(15)); // phase 1 + .FirstOrDefaultAsync() + .ObserveCompletion( + ex => logger.LogWarning(ex, "disposal faulted AFTER the wait settled"), + phase1Deadline.Token); // phase 1 var leakedIoLeaves = ioPools?.DrainAll() ?? 0; // phase 2 — cancel + join, NOT a polled wait if (disposeQueue is not null) await disposeQueue.DrainAsync(TimeSpan.FromSeconds(15)); // phase 3 diff --git a/src/MeshWeaver.Documentation/Data/Architecture/HubDisposalModel.md b/src/MeshWeaver.Documentation/Data/Architecture/HubDisposalModel.md index 8dab49adaf..d58b2b0f13 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/HubDisposalModel.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/HubDisposalModel.md @@ -113,18 +113,30 @@ awaits. The old `Task? Disposal` property is **gone** — a `Disposal is not nul check becomes `IsDisposing`, and an `await hub.Disposal` becomes a subscription. **At a genuine async edge** — xUnit teardown, `MessageHubGrain.OnDeactivateAsync` — -where a `Task` is legitimately the calling convention, bridge the observable *once*: +where a `Task` is legitimately the calling convention, bridge the observable *once*, +through `ReactiveCompletion.ObserveCompletion`: ```csharp // Grain deactivation / test teardown — the ONLY place a Task appears, at the edge: await hub.DisposalCompleted .Catch(_ => Observable.Return(Unit.Default)) // fault → "done" .FirstOrDefaultAsync() - .ToTask(ct); + .ObserveCompletion( + ex => logger.LogWarning(ex, "disposal faulted AFTER the wait settled"), + ct); ``` -That `.ToTask()` is the framework-lifecycle Task boundary (the same place tests -`await`). Nowhere inside `src/` hub-reachable code does it appear. +🚨 **Not `.ToTask(ct)`, and not a bare `await` on the observable either** — both are +forbidden repo-wide as of 2026-08-30 (*"no ToTask ever"*), and for this signal in +particular they are the deadlock: Rx completes its `TaskCompletionSource` without +`RunContinuationsAsynchronously`, so the awaiter resumes INLINE on the thread that +signalled — the hub's own disposal thread, or the grain's turn scheduler — and the +rest of the teardown then runs there. `ObserveCompletion` queues the continuation +instead, and keeps its error arm attached so a fault arriving after the wait settled +is reported rather than orphaned as an `UnobservedTaskException`. + +That `ObserveCompletion` is the framework-lifecycle Task boundary (the same place tests +`await`). Nowhere inside `src/` hub-reachable code does a Task appear at all. --- @@ -421,8 +433,9 @@ teardown; silent on a direct `Dispose()`). its leaf must run on a mesh-scoped resource (the persistence layer / `IIoPool`) so the mesh-level teardown drain can join it. - **Need to wait for the hub to finish disposing?** Subscribe to - `hub.DisposalCompleted`. Only at the test / grain edge may you bridge it once with - `.FirstOrDefaultAsync().ToTask()`. To ask "is it shutting down?", read `IsDisposing`. + `hub.DisposalCompleted`. Only at the test / grain edge may you bridge it once, and + only with `.FirstOrDefaultAsync().ObserveCompletion(reportLateFault, ct)` — never + `.ToTask()`. To ask "is it shutting down?", read `IsDisposing`. - **Tempted to `await` something during disposal?** Don't — it deadlocks the action block. Express the wait as an `Observable` (`Interval` poll, `Timer`/`Amb` deadline, subscribe to a child's `DisposalCompleted`) and post the next phase from its diff --git a/src/MeshWeaver.Documentation/Data/Architecture/LanguageServices.md b/src/MeshWeaver.Documentation/Data/Architecture/LanguageServices.md index d2e4332cec..2d7be259af 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/LanguageServices.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/LanguageServices.md @@ -155,7 +155,7 @@ Speculative compiles do **not** cache — every `LspCheckNode` call rebuilds the > 🚨 **Never `Observable.FromAsync` here — or anywhere in `src/`.** A bare `FromAsync` runs the function's synchronous prologue on the *subscribing* thread and applies no concurrency bound; under a blocking subscriber it deadlocks, because `SubscribeOn` moves only the subscribe, not the `await` continuation. The pool runs the leaf with `ConfigureAwait(false)` behind a gate, which is what makes it safe. See [Controlled I/O Pooling](/Doc/Architecture/ControlledIoPooling). -MCP and agent tool surfaces must return `Task` because their SDK signature demands it — they compose reactively and settle through `ObserveCompletion`, never `.ToTask()` (forbidden everywhere since 2026-08-30). Historically this read `.FirstAsync().ToTask()`, which was the sanctioned exception for external-protocol adapters (see [AsynchronousCalls](/Doc/Architecture/AsynchronousCalls)). No `await` anywhere in hub-reachable code. +MCP and agent tool surfaces have a `Task` signature that is not ours to change, so they bridge once at that edge — with `.FirstAsync().ObserveCompletion(reportLateFault, ct)`, never `.ToTask()` (forbidden repo-wide as of 2026-08-30) and never a bare `await someObservable`, both of which resume the adapter inline on the mesh thread that answered. Historically this read `.FirstAsync().ToTask()`, which was the sanctioned exception for external-protocol adapters (see [AsynchronousCalls](/Doc/Architecture/AsynchronousCalls)). No `await` anywhere in hub-reachable code. ## What's Deferred diff --git a/src/MeshWeaver.Documentation/Data/Architecture/NegativeControls.md b/src/MeshWeaver.Documentation/Data/Architecture/NegativeControls.md index 4fde61c015..4fb9ad7cea 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/NegativeControls.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/NegativeControls.md @@ -101,7 +101,7 @@ The rebuilt suite makes each test single-variable, and demands a leak before bel ```csharp // ── Control: without the policy the slide MUST reach the listener. ── probe.Reset(); -await renderer.Render(WithoutPolicy(html)).FirstAsync().ToTask(); +await renderer.Render(WithoutPolicy(html)).Should().Emit(); var leaked = probe.Connections; leaked.Should().BeGreaterThan(0, "the control must demonstrate the vector is real — a slide CAN otherwise make the " @@ -110,7 +110,7 @@ leaked.Should().BeGreaterThan(0, // ── Protected: the composed document must reach nothing. ── probe.Reset(); -var pdf = await renderer.Render(html).FirstAsync().ToTask(); +var pdf = await renderer.Render(html).Should().Emit(); probe.Connections.Should().Be(0, "…this is the SSRF surface…"); ``` diff --git a/src/MeshWeaver.Documentation/Data/Architecture/PermissionApi.md b/src/MeshWeaver.Documentation/Data/Architecture/PermissionApi.md index 0f0985abb6..9ccb06be18 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/PermissionApi.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/PermissionApi.md @@ -23,7 +23,7 @@ IObservable canTheyEdit = hub.CheckPermission(nodePath, "alice", Permissio IObservable theirPerms = hub.GetEffectivePermissions(nodePath, "alice"); ``` -All overloads return `IObservable`. Compose them with `CombineLatest`, `Select`, and `Where` as you would any other stream. In tests, bridge to `Task` at the outermost edge with `.FirstAsync().ToTask()`. Never use `await` inside `src/`. +All overloads return `IObservable`. Compose them with `CombineLatest`, `Select`, and `Where` as you would any other stream. In tests, assert on the stream and let the assertion own the wait — `await hub.GetEffectivePermissions(path).Should().Be(Permission.Read);` — never `.FirstAsync().ToTask()`, which is forbidden in `test/` as well as `src/` (2026-08-30). Never use `await` inside `src/`. diff --git a/src/MeshWeaver.Documentation/Data/Architecture/ReactiveTestAssertions.md b/src/MeshWeaver.Documentation/Data/Architecture/ReactiveTestAssertions.md index a64cf25b68..ff62f4e919 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/ReactiveTestAssertions.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/ReactiveTestAssertions.md @@ -1,15 +1,17 @@ --- Name: Reactive Test Assertions Category: Architecture -Description: The MeshWeaver.Reactive.Assertions surface — assert on observables directly, awaiting the terminal assertion instead of hand-rolling an Rx→Task bridge. Covers the full API, why the assertion subscribes on the thread pool, error assertions via Materialize, and the equivalency/JSON helpers. +Description: The MeshWeaver.Reactive.Assertions surface — assert on observables directly, awaiting the terminal assertion instead of hand-rolling an Rx→Task bridge (`.ToTask()` is forbidden everywhere, tests included). Covers the full API, why the assertion subscribes off xUnit's sync context and settles through its own ReactiveWait, error assertions via Materialize, and the equivalency/JSON helpers. Icon: --- # Reactive Test Assertions -MeshWeaver is reactive end-to-end: services, handlers, layout areas, and activities return `IObservable` and never `await` (see [Asynchronous Calls](/Doc/Architecture/AsynchronousCalls)). Its tests follow the same principle — **you assert on the stream directly, and the Rx→Task bridge lives inside the assertion instead of in your test body**. +MeshWeaver is reactive end-to-end: services, handlers, layout areas, and activities return `IObservable` and never `await` (see [Asynchronous Calls](/Doc/Architecture/AsynchronousCalls)). Its tests follow the same principle — **you assert on the stream directly, and the assertion OWNS the wait instead of your test body**. -All test assertions flow through `MeshWeaver.Reactive.Assertions`, an in-house library (it *replaced* FluentAssertions — that package is no longer referenced) wired in via a global `using` and project reference in `test/Directory.Build.props`. The names and chaining patterns (`.And` / `.Which`, trailing `because` args) are familiar. What is different is that the **observable assertions own the wait**: each terminal method returns a `Task` that the test body `await`s, so a test method is `async Task` and contains no hand-rolled `.FirstAsync().ToTask()`, `.Result`, `.Wait()` or `Task.Delay`. +All test assertions flow through `MeshWeaver.Reactive.Assertions`, an in-house library (it *replaced* FluentAssertions — that package is no longer referenced) wired in via a global `using` and project reference in `test/Directory.Build.props`. The names and chaining patterns (`.And` / `.Which`, trailing `because` args) are familiar. What is different is that the **observable assertions own the wait**: each terminal method returns a `Task` that the test body `await`s, so a test method is `async Task` and contains no `.FirstAsync().ToTask()` — none at all, anywhere — no `.Result`, no `.Wait()` and no `Task.Delay`. + +> **🚨 `.ToTask()` is forbidden repo-wide, `test/**` included** (maintainer ruling, 2026-08-30: *"totask is forbidden"*, *"no totask ever"* — the earlier "tests are the one sanctioned place" carve-out is RETRACTED). Rx's bridge completes its `TaskCompletionSource` **without** `TaskCreationOptions.RunContinuationsAsynchronously`, so it resumes the awaiter *inline on the signalling thread, still inside Rx's trampoline*. In a test that thread then carries the rest of the test, its mesh teardown, and — under xUnit — the runner starting the next test class, so the green proves the wrong thing. **`await source.FirstAsync()` is not a lighter alternative**: Rx's own awaiter is an `AsyncSubject` that completes its continuation from inside `OnCompleted`, and measures the same (`INLINE=True` for both spellings; only the queued completion below measures `INLINE=False`). Where an assertion genuinely cannot express the wait, the bridge is `MeshWeaver.Messaging.ReactiveCompletion.ObserveCompletion(reportLateFault, ct)`. > **🚨 The terminal assertions are `Task`-returning, not blocking.** `Emit()` / `Match()` return `Task`; `Be()` / `Complete()` / `NotEmit()` return `Task>`. **Every one must be `await`ed.** An un-awaited `obs.Should().Emit();` statement compiles without a warning, subscribes, and then lets the test race on — the classic green-but-lying test. `ObservableAssertions.cs` says so in its own doc comment: the wait is *"never a thread-blocking `ManualResetEventSlim` + `Wait`"*. @@ -36,7 +38,7 @@ For the surrounding test-writing rules, see [Writing Tests](/Doc/Architecture/Wr await .Should().Match(…) - SubscribeOn(TaskPool) + ToTask + off sync-context + queued resume Hub delivers emission @@ -70,7 +72,7 @@ For the surrounding test-writing rules, see [Writing Tests](/Doc/Architecture/Wr maxParallelThreads: 1 — everything serialises behind the held thread -*The assertion `SubscribeOn`s the source onto `TaskPoolScheduler` and bridges with `.ToTask()`, so every mesh round-trip runs on the thread pool and only the awaited result hops back to the test. Blocking that thread yourself (`.Result` / `.Wait()`) funnels those continuations back onto xUnit's single-threaded `MaxConcurrencySyncContext` — measured at 8 s versus 3 ms on `AddressResolutionTest` (2026-06-15). Do not remove the `SubscribeOn`.* +*The assertion subscribes **synchronously on the calling thread with the ambient `SynchronizationContext` suppressed** (`SubscribeHereOffSyncContext`), so the observer is attached in time to see a hot subject's emission while the mesh's continuations still capture `null` and land on the pool. It then waits through the package's own `ReactiveWait.First` — a `TaskCompletionSource` created with `TaskCreationOptions.RunContinuationsAsynchronously` — **not** `.ToTask()`, so the test's resume is queued instead of running on whichever mesh thread signalled. Blocking that thread yourself (`.Result` / `.Wait()`) funnels those continuations back onto xUnit's single-threaded `MaxConcurrencySyncContext` — measured at 8 s versus 3 ms on `AddressResolutionTest` (2026-06-15). Do not reintroduce `SubscribeOn`, and do not drop the suppression.* --- @@ -106,11 +108,11 @@ public async Task ObserveQuery_EmitsInitialResults() // ← async Tas (That is `ObservableQueryTests.ObserveQuery_EmitsInitialResults` in `test/MeshWeaver.Query.Test` — the role model for this shape.) -`.Emit()` and `.Match()` **return** the matched value, so `var x = await obs.Should().Match(...)` replaces a hand-rolled `var x = await obs.FirstAsync().ToTask()` one-for-one. +`.Emit()` and `.Match()` **return** the matched value, so `var x = await obs.Should().Match(...)` replaces a hand-rolled `var x = await obs.FirstAsync().ToTask()` one-for-one — and that hand-rolled shape is now a defect wherever it appears, not merely a verbose alternative. -> 🚨 **There is no sanctioned Rx→Task bridge — not even at the test edge** (2026-08-30). The wait goes through `ObservableAwait.Await(ct)` / `ReactiveCompletion.ObserveCompletion`, which queue the continuation instead of resuming it inline on the signalling thread. +> 🚨 **There is no sanctioned Rx→Task bridge — not even at the test edge** (2026-08-30). Nothing in the assertion path calls `.ToTask()`. > -> **How the wait is composed.** The source is `SubscribeOn`'d onto `TaskPoolScheduler`, filtered, `Take(1)`'d, bounded with `Timeout` (throwing a private sentinel so the assertion's own timeout stays distinguishable from a `TimeoutException` raised by the *source*), and bridged via `.ToTask()`. Nothing blocks a thread. A timed-out assertion reports what the stream actually emitted — "emitted nothing at all" versus "last of N emissions was …" — because those two failures have opposite fixes. +> **The assertion owns the wait.** The source is subscribed, filtered, `Take(1)`'d, collected with `ToList()` (so an empty completion arrives as data rather than as an exception indistinguishable from a source fault), bounded with `Timeout` (throwing a private sentinel so the assertion's own timeout stays distinguishable from a `TimeoutException` raised by the *source*), and settled through `ReactiveWait` — a `TaskCompletionSource` created with `RunContinuationsAsynchronously`, which is the line that keeps the test's continuation off the signalling thread. **The subscription is disposed when the wait settles**, so a settled assertion stops consuming a stream its siblings still need. Nothing blocks a thread. A timed-out assertion reports what the stream actually emitted — "emitted nothing at all" versus "last of N emissions was …" — because those two failures have opposite fixes. --- @@ -260,7 +262,7 @@ This exercises the real control plane: the owning hub's watcher reacts to `Reque The library lives in `src/MeshWeaver.Reactive.Assertions` (`System.Reactive` only) — the observable surface in `ObservableAssertions.cs`, the value surface in `ObjectAssertions.cs` / `CollectionAssertions.cs` / `MoreAssertions.cs`, and the equivalency + JSON helpers in `Equivalency.cs` / `JsonAssertions.cs`. -If a genuinely missing assertion is blocking a test, **add it to the library with a unit test in `test/MeshWeaver.Reactive.Assertions.Test` that exercises both the pass and the fail path** — do not work around it with a hand-rolled `.FirstAsync().ToTask()` in the test body. +If a genuinely missing assertion is blocking a test, **add it to the library with a unit test in `test/MeshWeaver.Reactive.Assertions.Test` that exercises both the pass and the fail path** — do not work around it with a hand-rolled `.FirstAsync().ToTask()` in the test body. That workaround is not merely discouraged; it is the forbidden shape, and the library's `ReactiveWait.First` exists precisely so no call site ever needs it. --- diff --git a/src/MeshWeaver.Documentation/Data/Architecture/SatelliteEntityPatterns.md b/src/MeshWeaver.Documentation/Data/Architecture/SatelliteEntityPatterns.md index 3eb86c58c6..dc08bf3332 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/SatelliteEntityPatterns.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/SatelliteEntityPatterns.md @@ -327,11 +327,13 @@ Subscribe to the node's stream **before** triggering the operation, then reactiv // GetMeshNodeStream(path) is the shared per-path handle; do NOT hand-roll // GetRemoteStream(address) + FirstOrDefault(n => n.Path == …), // which pulls the whole collection and re-opens its own subscription. +// The assertion subscribes SYNCHRONOUSLY on the calling thread, so arming it here +// really does attach the observer before step 1 sends the request. It also owns the +// wait: no FirstAsync + ToTask bridge, which is forbidden repo-wide (2026-08-30). var markersAppeared = workspace.GetMeshNodeStream(docPath) .Select(node => (node?.Content as MarkdownContent)?.Content ?? "") - .Where(content => content.Contains($"