Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions .claude/skills/sigsegv/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<UnauthorizedAccessException>()
.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<UnauthorizedAccessException>();
error.Exception!.Message.Should().Contain("Access denied");
}
}
```
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<UnauthorizedAccessException>();
// 🚨 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<UnauthorizedAccessException>();
}
}
```
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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<byte[]> — the work runs on subscription.
var bytes = await Mesh.ServiceProvider.GetRequiredService<PdfDocumentRenderer>()
.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";
Expand Down
4 changes: 2 additions & 2 deletions src/MeshWeaver.Documentation/Data/Architecture/ActorModel.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<T>` end-to-end — no `await`, no `Task<T>`, 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<T>` end-to-end — no `await`, no `Task<T>`, 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<T>`. Subscribe in the handler; the framework serialises the write through the owning hub's action block without blocking anything.

Expand Down Expand Up @@ -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<TResponse>(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` |

---

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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`.
Expand Down Expand Up @@ -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<T>()` 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.
Original file line number Diff line number Diff line change
Expand Up @@ -483,9 +483,18 @@ await mesh.TeardownAsync(TimeSpan.FromSeconds(15)); // MeshWeaver.Mesh.MeshTea
var ioPools = mesh.ServiceProvider.GetService<IoPoolRegistry>(); // capture BEFORE Dispose()
var disposeQueue = mesh.ServiceProvider.GetService<AsyncDisposeQueue>();
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<Unit, Exception>(_ => 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
Expand Down
25 changes: 19 additions & 6 deletions src/MeshWeaver.Documentation/Data/Architecture/HubDisposalModel.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<Unit, Exception>(_ => 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.

---

Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<string>` 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<string>` 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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 "
Expand All @@ -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…");
```

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ IObservable<bool> canTheyEdit = hub.CheckPermission(nodePath, "alice", Permissio
IObservable<Permission> theirPerms = hub.GetEffectivePermissions(nodePath, "alice");
```

All overloads return `IObservable<T>`. 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<T>`. 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/`.
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 760 280" style="width:100%;max-width:760px;height:auto;display:block;margin:20px auto;">
<defs>
<marker id="perm-arr" markerWidth="8" markerHeight="8" refX="7" refY="3" orient="auto">
Expand Down
Loading
Loading