fix(store-and-forward): resolve S&F delivery + replication wiring (3 Critical findings)

Resolves StoreAndForward-001, ExternalSystemGateway-001, NotificationService-001 — one systemic gap where buffered messages were persisted but never delivered, and the active node never replicated its buffer to the standby. Delivery handlers (ExternalSystemGateway-001 / NotificationService-001): - AkkaHostedService registers delivery handlers for the ExternalSystem, CachedDbWrite and Notification categories after StoreAndForwardService starts; each resolves its scoped consumer in a fresh DI scope. - ExternalSystemClient, DatabaseGateway and NotificationDeliveryService each gain a DeliverBufferedAsync method: re-resolve the target and re-attempt delivery, returning true/false/throwing per the transient-vs-permanent contract. - EnqueueAsync gains an attemptImmediateDelivery flag; CachedCallAsync and NotificationDeliveryService.SendAsync pass false (they already attempted delivery themselves) so registering a handler does not dispatch twice. Replication (StoreAndForward-001): - ReplicationService is injected into StoreAndForwardService; a new BufferAsync helper replicates every enqueue, and successful-retry removes and parks are replicated too. Fire-and-forget, no-op when replication is disabled. Tests: StoreAndForwardReplicationTests (Add/Remove/Park observed), attemptImmediateDelivery behaviour, and DeliverBufferedAsync paths for each consumer. Full solution builds; StoreAndForward/ExternalSystemGateway/ NotificationService suites green.
fix(central-ui): resolve CentralUI-001 — enforce script trust model before sandbox execution
2026-05-16 18:58:11 -04:00 · 2026-05-16 18:41:12 -04:00 · 2026-05-16 18:32:52 -04:00 · 2026-05-16 18:26:43 -04:00
24 changed files with 1068 additions and 143 deletions
@@ -8,7 +8,7 @@
 | Last reviewed | 2026-05-16 |
 | Reviewer | claude-agent |
 | Commit reviewed | `9c60592` |
-| Open findings | 19 |
+| Open findings | 18 |

 ## Summary

@@ -55,7 +55,7 @@ pages and the auth bridge are untested.
 |--|--|
 | Severity | Critical |
 | Category | Security |
-| Status | Open |
+| Status | Resolved |
 | Location | `src/ScadaLink.CentralUI/ScriptAnalysis/ScriptAnalysisService.cs:171-424` |

 **Description**
@@ -85,7 +85,22 @@ an editor hint.

 **Resolution**

-_Unresolved._
+Resolved 2026-05-16. A Roslyn semantic trust-model gate was added. `RunInSandboxAsync`
+now calls `EnforceTrustModel` after compilation and before `script.RunAsync`; if the
+script references any forbidden API the run is rejected (`SandboxErrorKind.CompileError`)
+with the offending markers, and the same gate is applied to nested shared scripts in
+`callSharedFunc`. `FindForbiddenApiUsages` was reworked so it resolves every identifier
+(not just the leftmost) against the semantic model and checks types **and** members —
+so a fully-qualified call such as `System.IO.File.WriteAllText(...)` is now caught, not
+only `using`-directive or bare-type forms. This is a static semantic gate consistent
+with the documented trust model; it is not a process sandbox — reflection-based
+indirection remains out of its reach, and full isolation would require running scripts
+in a separate constrained process (a larger change deliberately not taken here).
+Regression tests `RunInSandbox_FullyQualifiedForbiddenApi_IsBlockedBeforeExecution`,
+`RunInSandbox_ForbiddenUsingDirective_IsBlockedBeforeExecution` and
+`Diagnose_FullyQualifiedForbiddenCall_RaisesSCADA002` fail against the pre-fix code and
+pass after; `RunInSandbox_CleanScript_StillRuns` guards against over-blocking. Fixed by
+the commit whose message references `CentralUI-001`.

 ### CentralUI-002 — Site-scoped Deployment permissions are issued but never enforced

@@ -8,7 +8,7 @@
 | Last reviewed | 2026-05-16 |
 | Reviewer | claude-agent |
 | Commit reviewed | `9c60592` |
-| Open findings | 11 |
+| Open findings | 10 |

 ## Summary

@@ -16,9 +16,7 @@ The Communication module is generally well-structured and matches the design doc
 two-transport model (ClusterClient for command/control, gRPC server-streaming for
 real-time data). The actors keep mutable state on the actor thread, use `PipeTo` for
 async work, and the gRPC server/client lifecycle is mostly disciplined. However the
-review found one Critical issue (a `TimeoutException` from `DebugStreamService` leaves
-an orphaned bridge actor and an active site-side subscription, leaking resources on
-every snapshot timeout) and several High/Medium issues clustered around two themes:
+review found several High and Medium issues clustered around two themes:
 **(a) gRPC subscription bookkeeping races** — `SiteStreamGrpcClient` overwrites and
 removes subscription entries by correlation ID without disposal or ownership checks,
 so reconnect cycles leak `CancellationTokenSource`es and can cancel the wrong stream;
@@ -44,43 +42,55 @@ mutation races, and the snapshot-timeout cleanup path.

 ## Findings

-### Communication-001 — Snapshot timeout leaves orphaned bridge actor and site subscription
+### Communication-001 — Early stream termination escapes StartStreamAsync's narrow exception handling

 | | |
 |--|--|
-| Severity | Critical |
-| Category | Performance & resource management |
-| Status | Open |
-| Location | `src/ScadaLink.Communication/DebugStreamService.cs:139`, `src/ScadaLink.Communication/DebugStreamService.cs:149` |
+| Severity | Medium |
+| Category | Error handling & resilience |
+| Status | Resolved |
+| Location | `src/ScadaLink.Communication/DebugStreamService.cs:130-143` |
+
+**Re-triaged 2026-05-16:** originally filed Critical, claiming an orphaned bridge actor
+and a multi-minute site-side resource leak on every snapshot timeout. On verification
+that impact does **not** occur: `DebugStreamBridgeActor` calls `CleanupGrpc()` and
+`Context.Stop(Self)` on every path that invokes `onTerminated` (site disconnect, gRPC
+max-retries, `ReceiveTimeout`), so it always self-terminates and releases its gRPC
+subscription; and the pure-timeout path does reach `StopStream`, which also stops it.
+The genuine defect described below is an error-handling gap, not a leak — severity
+corrected to Medium.

 **Description**

-When `StartStreamAsync` times out waiting for the initial snapshot it calls
-`StopStream(sessionId)` and throws. `StopStream` only sends `StopDebugStream` to the
-bridge actor **if the session is still in `_sessions`**. But the bridge actor was added
-to `_sessions` at line 124 and is only removed by `onTerminatedWrapper`. The serious
-case is the race where `onTerminatedWrapper` fires first (e.g. site disconnect arrives
-during the wait): `snapshotTcs.TrySetException` completes the await with an
-`InvalidOperationException` rather than `OperationCanceledException`, which is **not**
-caught by the `catch (OperationCanceledException)` block. The exception propagates
-uncaught, `StopStream` is never reached, and if the bridge actor is instead orphaned
-(snapshot never arrives, site silent, no terminate) the only cleanup is the 5-minute
-`ReceiveTimeout` in the actor — meaning a site-side `StreamRelayActor` and gRPC stream
-can stay alive for up to 5 minutes after the central caller has given up. Combined with
-the 30s timeout, every transient snapshot delay leaks site resources for minutes.
+`StartStreamAsync` awaits the initial snapshot inside a `try` whose only handler is
+`catch (OperationCanceledException)`. When the stream terminates before the snapshot
+arrives, `onTerminatedWrapper` completes the await via
+`snapshotTcs.TrySetException(new InvalidOperationException(...))`. That
+`InvalidOperationException` is not an `OperationCanceledException`, so it escapes the
+catch entirely: the caller (Blazor debug view / SignalR hub) receives a raw,
+untranslated exception, and `StartStreamAsync` performs no teardown of its own on that
+path — it relies implicitly on the bridge actor self-terminating. Cleanup from the
+service side is therefore not deterministic, and the failure surfaced to the caller is
+not a meaningful, documented result.

 **Recommendation**

-In `StartStreamAsync`, wrap the `await` so that *any* failure or cancellation
-deterministically calls `StopStream(sessionId)` (e.g. `try/catch (Exception)` or a
-`finally` that stops the session when the result was not returned). Ensure
-`StopStream` is idempotent and always sends `StopDebugStream` even if the session was
-already removed, so the bridge actor (and its site-side subscription) is torn down
-promptly rather than waiting for the orphan `ReceiveTimeout`.
+In `StartStreamAsync`, catch any exception from the snapshot await, deterministically
+tear down the bridge actor (`Tell(StopDebugStream)` via the local actor reference, since
+a racing `onTerminatedWrapper` may already have removed the session entry), and translate
+the failure into a meaningful exception for the caller.

 **Resolution**

-_Unresolved._
+Resolved 2026-05-16. The `catch (OperationCanceledException)`-only block in
+`StartStreamAsync` was replaced with `catch (Exception)`: it removes the session entry,
+sends `StopDebugStream` to the bridge actor via the local reference (idempotent — the
+actor may already be stopping itself), and throws a descriptive exception —
+`TimeoutException` for the 30s timeout, otherwise an `InvalidOperationException` that
+names the instance/site and wraps the underlying cause. Regression test
+`DebugStreamServiceTests.StartStreamAsync_StreamTerminatesBeforeSnapshot_ThrowsMeaningfulException`
+fails against the pre-fix code and passes after. Fixed by the commit whose message
+references `Communication-001`.

 ### Communication-002 — gRPC reconnect does not unsubscribe the previous stream, leaking site-side relay actors

@@ -8,7 +8,7 @@
 | Last reviewed | 2026-05-16 |
 | Reviewer | claude-agent |
 | Commit reviewed | `9c60592` |
-| Open findings | 13 |
+| Open findings | 12 |

 ## Summary

@@ -53,7 +53,7 @@ tag-resolution retry, disconnect/re-subscribe, and concurrency around `HandleSub
 |--|--|
 | Severity | Critical |
 | Category | Concurrency & thread safety |
-| Status | Open |
+| Status | Resolved |
 | Location | `src/ScadaLink.DataConnectionLayer/Actors/DataConnectionActor.cs:473-538` |

 **Description**
@@ -82,7 +82,18 @@ handler too.

 **Resolution**

-_Unresolved._
+Resolved 2026-05-16. `HandleSubscribe` was restructured to follow the actor's own
+`PipeTo(Self)` pattern (the one already used by `HandleRetryTagResolution`): the
+background `Task.Run` now performs only adapter I/O (`SubscribeAsync`/`ReadAsync`),
+collects per-tag outcomes into an immutable `SubscribeCompleted` message, and pipes
+that to `Self`. All mutation of `_subscriptionIds`, `_subscriptionsByInstance`,
+`_totalSubscribed`, `_resolvedTags` and `_unresolvedTags` now happens in the new
+`HandleSubscribeCompleted` handler on the actor thread; it is wired into the
+Connected, Connecting and Reconnecting states so an in-flight subscribe is applied
+regardless of state transitions. Regression test
+`DCL001_ConcurrentSubscribes_DoNotCorruptSubscriptionCounters` (30×30 concurrent
+subscribes) fails against the pre-fix code and passes after. Fixed by the commit
+whose message references `DataConnectionLayer-001`.

 ### DataConnectionLayer-002 — `Restart` supervision discards all subscription state on connection-actor crash

@@ -8,7 +8,7 @@
 | Last reviewed | 2026-05-16 |
 | Reviewer | claude-agent |
 | Commit reviewed | `9c60592` |
-| Open findings | 14 |
+| Open findings | 13 |

 ## Summary

@@ -53,7 +53,7 @@ requirements (timeout, retry settings) that are declared but not implemented.
 |--|--|
 | Severity | Critical |
 | Category | Error handling & resilience |
-| Status | Open |
+| Status | Resolved |
 | Location | `src/ScadaLink.ExternalSystemGateway/ExternalSystemClient.cs:109`, `src/ScadaLink.ExternalSystemGateway/DatabaseGateway.cs:81` |

 **Description**
@@ -89,7 +89,19 @@ verifies it is delivered by a retry sweep.

 **Resolution**

-_Unresolved._
+Resolved 2026-05-16. Delivery handlers for `StoreAndForwardCategory.ExternalSystem` and
+`CachedDbWrite` are now registered at site startup in `AkkaHostedService`, after
+`StoreAndForwardService.StartAsync()`. Each handler resolves its consumer in a fresh DI
+scope and calls a new `DeliverBufferedAsync`: `ExternalSystemClient.DeliverBufferedAsync`
+re-resolves the system/method and re-invokes `InvokeHttpAsync`, and
+`DatabaseGateway.DeliverBufferedAsync` executes the buffered SQL — each returning `true`
+on success, `false` (park) when the target no longer exists or fails permanently, and
+throwing on transient failure so the engine retries. `EnqueueAsync` gained an
+`attemptImmediateDelivery` parameter; `CachedCallAsync` passes `false` so registering the
+handler does not dispatch the request twice (the double-dispatch noted in
+`ExternalSystemGateway-003`). Regression tests cover the success, target-removed and
+transient-retry paths. Fixed by the commit whose message references
+`ExternalSystemGateway-001`.

 ### ExternalSystemGateway-002 — Per-system call timeout is never applied to HTTP requests

@@ -8,7 +8,7 @@
 | Last reviewed | 2026-05-16 |
 | Reviewer | claude-agent |
 | Commit reviewed | `9c60592` |
-| Open findings | 12 |
+| Open findings | 11 |

 ## Summary

@@ -53,7 +53,7 @@ fallback in `DeliverAsync`, and concurrency on the token cache.
 |--|--|
 | Severity | Critical |
 | Category | Error handling & resilience |
-| Status | Open |
+| Status | Resolved |
 | Location | `src/ScadaLink.NotificationService/NotificationDeliveryService.cs:96`, `src/ScadaLink.NotificationService/ServiceCollectionExtensions.cs:8` |

 **Description**
@@ -66,7 +66,15 @@ Register a delivery handler for `StoreAndForwardCategory.Notification` during st

 **Resolution**

-_Unresolved._
+Resolved 2026-05-16. A delivery handler for `StoreAndForwardCategory.Notification` is now
+registered at site startup in `AkkaHostedService`. The handler resolves
+`NotificationDeliveryService` in a fresh DI scope and calls the new `DeliverBufferedAsync`,
+which re-resolves the list, recipients and SMTP config and re-attempts delivery —
+returning `true` on success, `false` (park) on permanent failure or missing
+configuration, and throwing on transient failure so the engine retries. `SendAsync` now
+buffers with `attemptImmediateDelivery: false` so registering the handler does not send
+the notification twice. Regression tests cover the happy path and the list-removed park
+path. Fixed by the commit whose message references `NotificationService-001`.

 ### NotificationService-002 — `TimeoutException`/`OperationCanceledException` misclassified as transient

@@ -28,56 +28,52 @@ code-reviews/

 ## Baseline review — 2026-05-16

-All 19 modules were reviewed at commit `9c60592`. This established the baseline below.
+All 19 modules were reviewed at commit `9c60592` (241 findings: 6 Critical, 46 High,
+100 Medium, 89 Low). The tables below track what remains **open** as findings are
+resolved and re-triaged.

 | Severity | Open findings |
 |----------|---------------|
-| Critical | 6 |
+| Critical | 0 |
 | High | 46 |
 | Medium | 100 |
 | Low | 89 |
-| **Total** | **241** |
+| **Total** | **235** |

 ## Module Status

-| Module | Review status | Last reviewed | Commit | Open (C/H/M/L) | Total |
-|--------|---------------|---------------|--------|----------------|-------|
-| [CentralUI](CentralUI/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 1/3/10/5 | 19 |
-| [CLI](CLI/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/1/6/6 | 13 |
-| [ClusterInfrastructure](ClusterInfrastructure/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/1/4/3 | 8 |
-| [Commons](Commons/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/0/4/8 | 12 |
-| [Communication](Communication/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 1/2/5/3 | 11 |
-| [ConfigurationDatabase](ConfigurationDatabase/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/1/4/6 | 11 |
-| [DataConnectionLayer](DataConnectionLayer/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 1/4/6/2 | 13 |
-| [DeploymentManager](DeploymentManager/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/3/6/5 | 14 |
-| [ExternalSystemGateway](ExternalSystemGateway/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 1/2/7/4 | 14 |
-| [HealthMonitoring](HealthMonitoring/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/2/5/5 | 12 |
-| [Host](Host/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/1/3/7 | 11 |
-| [InboundAPI](InboundAPI/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/3/5/5 | 13 |
-| [ManagementService](ManagementService/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/3/5/5 | 13 |
-| [NotificationService](NotificationService/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 1/3/5/3 | 12 |
-| [Security](Security/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/3/4/4 | 11 |
-| [SiteEventLogging](SiteEventLogging/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/4/4/3 | 11 |
-| [SiteRuntime](SiteRuntime/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/3/8/5 | 16 |
-| [StoreAndForward](StoreAndForward/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 1/2/4/6 | 13 |
-| [TemplateEngine](TemplateEngine/findings.md) | Reviewed | 2026-05-16 | `9c60592` | 0/5/5/4 | 14 |
+| Module | Last reviewed | Commit | Open (C/H/M/L) | Open | Total |
+|--------|---------------|--------|----------------|------|-------|
+| [CentralUI](CentralUI/findings.md) | 2026-05-16 | `9c60592` | 0/3/10/5 | 18 | 19 |
+| [CLI](CLI/findings.md) | 2026-05-16 | `9c60592` | 0/1/6/6 | 13 | 13 |
+| [ClusterInfrastructure](ClusterInfrastructure/findings.md) | 2026-05-16 | `9c60592` | 0/1/4/3 | 8 | 8 |
+| [Commons](Commons/findings.md) | 2026-05-16 | `9c60592` | 0/0/4/8 | 12 | 12 |
+| [Communication](Communication/findings.md) | 2026-05-16 | `9c60592` | 0/2/5/3 | 10 | 11 |
+| [ConfigurationDatabase](ConfigurationDatabase/findings.md) | 2026-05-16 | `9c60592` | 0/1/4/6 | 11 | 11 |
+| [DataConnectionLayer](DataConnectionLayer/findings.md) | 2026-05-16 | `9c60592` | 0/4/6/2 | 12 | 13 |
+| [DeploymentManager](DeploymentManager/findings.md) | 2026-05-16 | `9c60592` | 0/3/6/5 | 14 | 14 |
+| [ExternalSystemGateway](ExternalSystemGateway/findings.md) | 2026-05-16 | `9c60592` | 0/2/7/4 | 13 | 14 |
+| [HealthMonitoring](HealthMonitoring/findings.md) | 2026-05-16 | `9c60592` | 0/2/5/5 | 12 | 12 |
+| [Host](Host/findings.md) | 2026-05-16 | `9c60592` | 0/1/3/7 | 11 | 11 |
+| [InboundAPI](InboundAPI/findings.md) | 2026-05-16 | `9c60592` | 0/3/5/5 | 13 | 13 |
+| [ManagementService](ManagementService/findings.md) | 2026-05-16 | `9c60592` | 0/3/5/5 | 13 | 13 |
+| [NotificationService](NotificationService/findings.md) | 2026-05-16 | `9c60592` | 0/3/5/3 | 11 | 12 |
+| [Security](Security/findings.md) | 2026-05-16 | `9c60592` | 0/3/4/4 | 11 | 11 |
+| [SiteEventLogging](SiteEventLogging/findings.md) | 2026-05-16 | `9c60592` | 0/4/4/3 | 11 | 11 |
+| [SiteRuntime](SiteRuntime/findings.md) | 2026-05-16 | `9c60592` | 0/3/8/5 | 16 | 16 |
+| [StoreAndForward](StoreAndForward/findings.md) | 2026-05-16 | `9c60592` | 0/2/4/6 | 12 | 13 |
+| [TemplateEngine](TemplateEngine/findings.md) | 2026-05-16 | `9c60592` | 0/5/5/4 | 14 | 14 |

 ## Pending Findings

-All findings are currently `Open`. As findings are resolved, remove them from the
-tables below (see [REVIEW-PROCESS.md](REVIEW-PROCESS.md) §5). Full detail for each
-finding — description, location, recommendation — lives in the module's `findings.md`.
+Every `Open` / `In Progress` finding across all modules, highest severity first.
+Resolved findings drop off this list but remain recorded in their module's
+`findings.md` (see [REVIEW-PROCESS.md](REVIEW-PROCESS.md) §4–§5). Full detail —
+description, location, recommendation — lives in the module's `findings.md`.

-### Critical (6)
+### Critical (0)

-| ID | Module | Title |
-|----|--------|-------|
-| CentralUI-001 | [CentralUI](CentralUI/findings.md) | Test Run sandbox executes arbitrary C# with no trust-model enforcement |
-| Communication-001 | [Communication](Communication/findings.md) | Snapshot timeout leaves orphaned bridge actor and site subscription |
-| DataConnectionLayer-001 | [DataConnectionLayer](DataConnectionLayer/findings.md) | `Task.Run` in `HandleSubscribe` mutates actor state off the actor thread |
-| ExternalSystemGateway-001 | [ExternalSystemGateway](ExternalSystemGateway/findings.md) | No S&F delivery handler registered; cached calls and writes can never be delivered |
-| NotificationService-001 | [NotificationService](NotificationService/findings.md) | Buffered notifications are never retried (no S&F delivery handler) |
-| StoreAndForward-001 | [StoreAndForward](StoreAndForward/findings.md) | Replication to standby is never triggered by the active node |
+_None open._

 ### High (46)

@@ -8,7 +8,7 @@
 | Last reviewed | 2026-05-16 |
 | Reviewer | claude-agent |
 | Commit reviewed | `9c60592` |
-| Open findings | 13 |
+| Open findings | 12 |

 ## Summary

@@ -53,7 +53,7 @@ replication and retry-count issues are functional defects against the design.
 |--|--|
 | Severity | Critical |
 | Category | Error handling & resilience |
-| Status | Open |
+| Status | Resolved |
 | Location | `src/ScadaLink.StoreAndForward/ReplicationService.cs:40`, `:53`, `:66`; `src/ScadaLink.StoreAndForward/StoreAndForwardService.cs:155`, `:212`, `:222`, `:236` |

 **Description**
@@ -81,7 +81,14 @@ asserts the replication handler observes each operation type.

 **Resolution**

-_Unresolved._
+Resolved 2026-05-16. `ReplicationService` is now injected into `StoreAndForwardService`
+(wired in `AddStoreAndForward`), and every buffer operation is forwarded to the standby:
+a new `BufferAsync` helper calls `ReplicateEnqueue` after each persist, `ReplicateRemove`
+runs after a successful retry removes a message, and `ReplicatePark` runs on both park
+paths. Replication stays fire-and-forget and is a no-op when `ReplicationEnabled` is
+false or no handler is wired. Regression tests `StoreAndForwardReplicationTests` assert
+the replication handler observes the Add, Remove and Park operations. Fixed by the
+commit whose message references `StoreAndForward-001`.

 ### StoreAndForward-002 — Messages enqueued with no registered handler are buffered but never deliverable

@@ -220,6 +220,20 @@ public class ScriptAnalysisService
                SandboxErrorKind.CompileError, 0, markers);
        }

+        // Trust-model gate (CentralUI-001): the documented forbidden-API set is
+        // enforced HERE, before execution — not merely surfaced as an editor hint.
+        // Without this, a Design-role user could run arbitrary file/process/
+        // reflection/network code in the central host process.
+        var trustViolations = EnforceTrustModel(script.GetCompilation());
+        if (trustViolations.Count > 0)
+        {
+            return new SandboxRunResult(false, null, null, "",
+                "Script blocked by the trust model — it references forbidden APIs "
+                + "(System.IO, System.Diagnostics, System.Reflection, System.Net, threading). "
+                + "See the highlighted diagnostics.",
+                SandboxErrorKind.CompileError, 0, trustViolations);
+        }
+
        var parameters = ConvertJsonParameters(request.Parameters);

        using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(timeoutSeconds));
@@ -311,6 +325,13 @@ public class ScriptAnalysisService
                    throw new ScriptSandboxException(
                        $"Scripts.CallShared(\"{name}\") compile failed: {string.Join("; ", nestedErrors.Select(d => d.GetMessage()))}");

+                // Trust-model gate (CentralUI-001) — a nested shared script runs
+                // arbitrary code too, so it must clear the same forbidden-API gate.
+                if (EnforceTrustModel(built.GetCompilation()).Count > 0)
+                    throw new ScriptSandboxException(
+                        $"Scripts.CallShared(\"{name}\") is blocked by the script trust model — "
+                        + "the shared script references forbidden APIs.");
+
                lock (compileCacheLock)
                {
                    if (!compileCache.TryGetValue(name, out compiled))
@@ -1086,15 +1107,25 @@ public class ScriptAnalysisService
        return new(AttributeContextKind.None, null);
    }

+    /// <summary>
+    /// Finds every reference to a forbidden API — the documented script trust model,
+    /// see <see cref="ForbiddenNamespacePrefixes"/>. Identifiers are resolved against
+    /// the semantic model, so a forbidden type or member is caught however it is
+    /// written: bare (<c>File</c>), fully qualified
+    /// (<c>System.IO.File.WriteAllText</c>), or via an alias — while a user identifier
+    /// that merely shares a name with a forbidden type (<c>var File = …</c>) does not
+    /// false-positive. Used both for editor diagnostics and as the pre-execution
+    /// trust-model gate (see <see cref="EnforceTrustModel"/>).
+    /// </summary>
    private static IEnumerable<DiagnosticMarker> FindForbiddenApiUsages(SyntaxTree tree, SemanticModel model)
    {
        var root = tree.GetRoot();

-        // Banned using directives — pure namespace string match is fine here.
+        // Banned using directives.
        foreach (var u in root.DescendantNodes().OfType<UsingDirectiveSyntax>())
        {
            var name = u.Name?.ToString() ?? "";
-            if (ForbiddenNamespacePrefixes.Any(p => name == p || name.StartsWith(p + ".")))
+            if (IsForbiddenName(name))
            {
                var span = u.GetLocation().GetLineSpan().Span;
                yield return new DiagnosticMarker(
@@ -1108,20 +1139,14 @@ public class ScriptAnalysisService
            }
        }

-        // Banned type usages — resolved via the semantic model so a user
-        // identifier named "File" or "Thread" does NOT trigger the diagnostic
-        // unless it actually resolves to a forbidden type.
+        // Banned type / member references, resolved via the semantic model. Every
+        // identifier is checked — including the right-hand side of a member access —
+        // so a fully-qualified forbidden call (System.IO.File.WriteAllText) cannot
+        // slip past by avoiding a `using` directive or a bare type name.
        foreach (var ident in root.DescendantNodes().OfType<IdentifierNameSyntax>())
        {
-            // Skip the identifier on the right side of a member access — only
-            // the leftmost (the type or qualifier) is what we want to check.
-            if (ident.Parent is MemberAccessExpressionSyntax m && m.Name == ident) continue;
-
-            var symbol = model.GetSymbolInfo(ident).Symbol;
-            if (symbol is not INamedTypeSymbol type) continue;
-
-            var ns = type.ContainingNamespace?.ToDisplayString() ?? "";
-            if (!ForbiddenNamespacePrefixes.Any(p => ns == p || ns.StartsWith(p + "."))) continue;
+            var forbidden = ForbiddenNameFor(model.GetSymbolInfo(ident).Symbol);
+            if (forbidden == null) continue;

            var span = ident.GetLocation().GetLineSpan().Span;
            yield return new DiagnosticMarker(
@@ -1130,11 +1155,75 @@ public class ScriptAnalysisService
                StartColumn: span.Start.Character + 1,
                EndLineNumber: span.End.Line + 1,
                EndColumn: span.End.Character + 1,
-                Message: $"Type '{type.Name}' from forbidden namespace '{ns}' is not allowed in scripts.",
+                Message: $"'{ident.Identifier.ValueText}' resolves to forbidden API '{forbidden}', " +
+                         "which is not allowed in scripts (script trust model).",
                Code: "SCADA002");
        }
    }

+    /// <summary>
+    /// The forbidden namespace/type a symbol implicates, or null if it is allowed.
+    /// Checks the symbol's namespace and — for a type or member — the type's full
+    /// name, so an entry like <c>System.Threading.Thread</c> bans that exact type
+    /// while <c>System.Threading</c> (e.g. <c>CancellationToken</c>) stays allowed.
+    /// </summary>
+    private static string? ForbiddenNameFor(ISymbol? symbol)
+    {
+        if (symbol == null) return null;
+        foreach (var name in QualifiedNamesOf(symbol))
+            if (IsForbiddenName(name))
+                return name;
+        return null;
+    }
+
+    /// <summary>Fully-qualified names a symbol reference implicates for trust-model checking.</summary>
+    private static IEnumerable<string> QualifiedNamesOf(ISymbol symbol)
+    {
+        switch (symbol)
+        {
+            case INamespaceSymbol { IsGlobalNamespace: false } ns:
+                yield return ns.ToDisplayString();
+                break;
+            case ITypeSymbol type:
+                if (type.ContainingNamespace is { IsGlobalNamespace: false } tn)
+                    yield return tn.ToDisplayString();
+                yield return FullTypeName(type);
+                break;
+            default:
+                if (symbol.ContainingType is { } ct)
+                {
+                    if (ct.ContainingNamespace is { IsGlobalNamespace: false } cn)
+                        yield return cn.ToDisplayString();
+                    yield return FullTypeName(ct);
+                }
+                break;
+        }
+    }
+
+    private static string FullTypeName(ITypeSymbol type) =>
+        type.ContainingNamespace is { IsGlobalNamespace: false } ns
+            ? ns.ToDisplayString() + "." + type.Name
+            : type.Name;
+
+    private static bool IsForbiddenName(string qualifiedName) =>
+        ForbiddenNamespacePrefixes.Any(p =>
+            qualifiedName == p || qualifiedName.StartsWith(p + ".", StringComparison.Ordinal));
+
+    /// <summary>
+    /// Pre-execution trust-model gate (CentralUI-001). Returns the forbidden-API
+    /// markers (SCADA001/SCADA002) for a compiled script; an empty list means the
+    /// script is clear to run. This is a static semantic gate, not a process
+    /// sandbox — reflection-based indirection is still out of its reach; full
+    /// isolation would require running scripts in a separate constrained process.
+    /// </summary>
+    private static IReadOnlyList<DiagnosticMarker> EnforceTrustModel(Compilation compilation)
+    {
+        var tree = compilation.SyntaxTrees.FirstOrDefault();
+        if (tree == null) return Array.Empty<DiagnosticMarker>();
+        var model = compilation.GetSemanticModel(tree);
+        return FindForbiddenApiUsages(tree, model).ToList();
+    }
+
    private static CompletionItem ToCompletionItem(ISymbol symbol)
    {
        var kind = symbol.Kind switch
@@ -127,21 +127,37 @@ public class DebugStreamService
        using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        timeoutCts.CancelAfter(TimeSpan.FromSeconds(30));

+        DebugViewSnapshot snapshot;
        try
        {
-            var snapshot = await snapshotTcs.Task.WaitAsync(timeoutCts.Token);
+            snapshot = await snapshotTcs.Task.WaitAsync(timeoutCts.Token);
+        }
+        catch (Exception ex)
+        {
+            // Any failure before the snapshot arrives — the 30s timeout, or the stream
+            // terminating early (site disconnect / gRPC failure, surfaced by
+            // onTerminatedWrapper as an InvalidOperationException) — must deterministically
+            // tear down the bridge actor and its site-side subscription. Use the local
+            // actor reference: a racing onTerminatedWrapper may already have removed the
+            // session, which would make StopStream a no-op. StopDebugStream is idempotent
+            // (the actor may already be stopping itself).
+            _sessions.TryRemove(sessionId, out _);
+            bridgeActor.Tell(new StopDebugStream());
+
+            if (ex is OperationCanceledException)
+                throw new TimeoutException(
+                    $"Timed out waiting for debug snapshot from {instanceUniqueName} on site {siteIdentifier}.");
+
+            throw new InvalidOperationException(
+                $"Debug stream for {instanceUniqueName} on site {siteIdentifier} terminated before a snapshot was received.",
+                ex);
+        }

        _logger.LogInformation("Debug stream {SessionId} started for {Instance} on site {Site}",
            sessionId, instanceUniqueName, siteIdentifier);

        return new DebugStreamSession(sessionId, snapshot);
    }
-        catch (OperationCanceledException)
-        {
-            StopStream(sessionId);
-            throw new TimeoutException($"Timed out waiting for debug snapshot from {instanceUniqueName} on site {siteIdentifier}.");
-        }
-    }

    /// <summary>
    /// Stops an active debug stream session.
@@ -171,6 +171,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
            case UnsubscribeTagsRequest:
                Stash.Stash();
                break;
+            case SubscribeCompleted sc:
+                // A subscribe started while Connected can complete after a transition;
+                // apply it so its state survives into the next ReSubscribeAll.
+                HandleSubscribeCompleted(sc);
+                break;
            case GetHealthReport:
                ReplyWithHealthReport();
                break;
@@ -207,6 +212,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
            case SubscribeTagsRequest req:
                HandleSubscribe(req);
                break;
+            case SubscribeCompleted sc:
+                HandleSubscribeCompleted(sc);
+                break;
            case UnsubscribeTagsRequest req:
                HandleUnsubscribe(req);
                break;
@@ -338,6 +346,11 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
            case TagResolutionFailed:
                // Ignore — stale results from previous connection; ReSubscribeAll runs after reconnect
                break;
+            case SubscribeCompleted sc:
+                // A subscribe started while Connected can complete after a transition;
+                // apply it so its state survives into the next ReSubscribeAll.
+                HandleSubscribeCompleted(sc);
+                break;
            case GetHealthReport:
                ReplyWithHealthReport();
                break;
@@ -466,18 +479,27 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
        if (!_subscriptionsByInstance.ContainsKey(request.InstanceUniqueName))
            _subscriptionsByInstance[request.InstanceUniqueName] = new HashSet<string>();

-        var instanceTags = _subscriptionsByInstance[request.InstanceUniqueName];
        var self = Self;
        var sender = Sender;

+        // Snapshot the already-subscribed tag set on the actor thread. The background
+        // task below must NOT read or mutate actor state — it performs only adapter
+        // I/O and reports results back via a SubscribeCompleted message, which is
+        // applied to actor state on the actor thread (see HandleSubscribeCompleted).
+        var alreadySubscribed = new HashSet<string>(_subscriptionIds.Keys);
+
        Task.Run(async () =>
        {
+            var results = new List<SubscribeTagResult>(request.TagPaths.Count);
+            var tagsToSeed = new List<string>();
+
            foreach (var tagPath in request.TagPaths)
            {
-                if (_subscriptionIds.ContainsKey(tagPath))
+                if (alreadySubscribed.Contains(tagPath))
                {
-                    // Already subscribed — just track for this instance
-                    instanceTags.Add(tagPath);
+                    // Already subscribed by another instance — just track for this one.
+                    results.Add(new SubscribeTagResult(tagPath, AlreadySubscribed: true, Success: true, null, null));
+                    tagsToSeed.Add(tagPath);
                    continue;
                }

@@ -487,27 +509,21 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
                    {
                        self.Tell(new TagValueReceived(path, value));
                    });
-                    _subscriptionIds[tagPath] = subId;
-                    instanceTags.Add(tagPath);
-                    _totalSubscribed++;
-                    _resolvedTags++;
+                    results.Add(new SubscribeTagResult(tagPath, AlreadySubscribed: false, Success: true, subId, null));
+                    tagsToSeed.Add(tagPath);
                }
                catch (Exception ex)
                {
-                    // WP-12: Tag path resolution failure — mark as unresolved, retry later
-                    _unresolvedTags.Add(tagPath);
-                    instanceTags.Add(tagPath);
-                    _totalSubscribed++;
-
-                    self.Tell(new TagResolutionFailed(tagPath, ex.Message));
+                    // WP-12: Tag path resolution failure — reported back as unresolved.
+                    results.Add(new SubscribeTagResult(tagPath, AlreadySubscribed: false, Success: false, null, ex.Message));
                }
            }

-            // Initial read — seed current values for all resolved tags so the Instance Actor
-            // doesn't stay Uncertain until the next OPC UA data change notification
-            foreach (var tagPath in instanceTags)
+            // Initial read — seed current values for resolved tags so the Instance Actor
+            // doesn't stay Uncertain until the next OPC UA data change notification.
+            // Tell is thread-safe, so seeded values are delivered directly as messages.
+            foreach (var tagPath in tagsToSeed)
            {
-                if (_unresolvedTags.Contains(tagPath)) continue;
                try
                {
                    var readResult = await _adapter.ReadAsync(tagPath);
@@ -522,11 +538,51 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
                }
            }

-            return new SubscribeTagsResponse(
-                request.CorrelationId, request.InstanceUniqueName, true, null, DateTimeOffset.UtcNow);
-        }).PipeTo(sender);
+            return new SubscribeCompleted(request, sender, results);
+        }).PipeTo(self);
+    }

-        // Start tag resolution retry timer if we have unresolved tags
+    /// <summary>
+    /// Applies the result of an asynchronous subscribe on the actor thread. ALL mutation
+    /// of subscription state and counters happens here — never on the background task —
+    /// so the actor model's single-threaded state guarantee holds.
+    /// </summary>
+    private void HandleSubscribeCompleted(SubscribeCompleted msg)
+    {
+        var instanceName = msg.Request.InstanceUniqueName;
+        if (!_subscriptionsByInstance.TryGetValue(instanceName, out var instanceTags))
+        {
+            // The instance was unsubscribed while the subscribe I/O was in flight.
+            instanceTags = new HashSet<string>();
+            _subscriptionsByInstance[instanceName] = instanceTags;
+        }
+
+        foreach (var result in msg.Results)
+        {
+            instanceTags.Add(result.TagPath);
+
+            // Re-check against current state: another subscribe may have resolved the
+            // same tag while this request's I/O was in flight.
+            if (result.AlreadySubscribed || _subscriptionIds.ContainsKey(result.TagPath))
+                continue;
+
+            if (result.Success)
+            {
+                _subscriptionIds[result.TagPath] = result.SubscriptionId!;
+                _totalSubscribed++;
+                _resolvedTags++;
+            }
+            else
+            {
+                // WP-12: mark unresolved so the periodic retry timer picks it up.
+                _unresolvedTags.Add(result.TagPath);
+                _totalSubscribed++;
+                _log.Debug("[{0}] Tag resolution failed for {1}: {2}",
+                    _connectionName, result.TagPath, result.Error);
+            }
+        }
+
+        // Start the tag-resolution retry timer if any tags are unresolved.
        if (_unresolvedTags.Count > 0)
        {
            Timers.StartPeriodicTimer(
@@ -535,6 +591,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
                _options.TagResolutionRetryInterval,
                _options.TagResolutionRetryInterval);
        }
+
+        msg.ReplyTo.Tell(new SubscribeTagsResponse(
+            msg.Request.CorrelationId, instanceName, true, null, DateTimeOffset.UtcNow));
    }

    private void HandleUnsubscribe(UnsubscribeTagsRequest request)
@@ -764,5 +823,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
    internal record TagResolutionFailed(string TagPath, string Error);
    internal record TagResolutionSucceeded(string TagPath, string SubscriptionId);
    internal record RetryTagResolution;
+    internal record SubscribeTagResult(
+        string TagPath, bool AlreadySubscribed, bool Success, string? SubscriptionId, string? Error);
+    internal record SubscribeCompleted(
+        SubscribeTagsRequest Request, IActorRef ReplyTo, IReadOnlyList<SubscribeTagResult> Results);
    public record GetHealthReport;
 }
@@ -87,6 +87,64 @@ public class DatabaseGateway : IDatabaseGateway
            definition.RetryDelay > TimeSpan.Zero ? definition.RetryDelay : null);
    }

+    /// <summary>
+    /// WP-9/10: Delivers a buffered CachedDbWrite during a store-and-forward retry
+    /// sweep — executes the SQL against the named connection. Returns true on
+    /// success, false if the connection no longer exists (the message is parked);
+    /// throws on any execution error so the engine retries.
+    /// </summary>
+    public async Task<bool> DeliverBufferedAsync(
+        StoreAndForwardMessage message, CancellationToken cancellationToken = default)
+    {
+        var payload = JsonSerializer.Deserialize<CachedWritePayload>(message.PayloadJson);
+        if (payload == null || string.IsNullOrEmpty(payload.ConnectionName) || string.IsNullOrEmpty(payload.Sql))
+        {
+            _logger.LogError("Buffered CachedDbWrite message {Id} has an unreadable payload; parking.", message.Id);
+            return false;
+        }
+
+        var definition = await ResolveConnectionAsync(payload.ConnectionName, cancellationToken);
+        if (definition == null)
+        {
+            _logger.LogError(
+                "Buffered DB write to '{Connection}' cannot be delivered — the connection no longer exists; parking.",
+                payload.ConnectionName);
+            return false;
+        }
+
+        await using var connection = new SqlConnection(definition.ConnectionString);
+        await connection.OpenAsync(cancellationToken);
+        using var command = connection.CreateCommand();
+        command.CommandText = payload.Sql;
+        if (payload.Parameters != null)
+        {
+            foreach (var (key, value) in payload.Parameters)
+            {
+                var parameter = command.CreateParameter();
+                parameter.ParameterName = key.StartsWith('@') ? key : "@" + key;
+                parameter.Value = JsonElementToParameterValue(value);
+                command.Parameters.Add(parameter);
+            }
+        }
+        await command.ExecuteNonQueryAsync(cancellationToken);
+        return true;
+    }
+
+    private static object JsonElementToParameterValue(JsonElement element) => element.ValueKind switch
+    {
+        JsonValueKind.String => (object?)element.GetString() ?? DBNull.Value,
+        JsonValueKind.Number => element.TryGetInt64(out var l) ? l : element.GetDouble(),
+        JsonValueKind.True => true,
+        JsonValueKind.False => false,
+        JsonValueKind.Null or JsonValueKind.Undefined => DBNull.Value,
+        _ => element.GetRawText()
+    };
+
+    private sealed record CachedWritePayload(
+        string ConnectionName,
+        string Sql,
+        Dictionary<string, JsonElement>? Parameters);
+
    private async Task<DatabaseConnectionDefinition?> ResolveConnectionAsync(
        string connectionName,
        CancellationToken cancellationToken)
@@ -106,18 +106,67 @@ public class ExternalSystemClient : IExternalSystemClient
                Parameters = parameters
            });

-            var sfResult = await _storeAndForward.EnqueueAsync(
+            // attemptImmediateDelivery: false — this method already made the HTTP
+            // attempt above; letting EnqueueAsync re-invoke the handler would
+            // dispatch the same request a second time.
+            await _storeAndForward.EnqueueAsync(
                StoreAndForwardCategory.ExternalSystem,
                systemName,
                payload,
                originInstanceName,
                system.MaxRetries > 0 ? system.MaxRetries : null,
-                system.RetryDelay > TimeSpan.Zero ? system.RetryDelay : null);
+                system.RetryDelay > TimeSpan.Zero ? system.RetryDelay : null,
+                attemptImmediateDelivery: false);

            return new ExternalCallResult(true, null, null, WasBuffered: true);
        }
    }

+    /// <summary>
+    /// WP-7/10: Delivers a buffered ExternalSystem call during a store-and-forward
+    /// retry sweep. Returns true on success, false on permanent failure (the message
+    /// is parked); throws <see cref="TransientExternalSystemException"/> on a
+    /// transient failure so the engine retries.
+    /// </summary>
+    public async Task<bool> DeliverBufferedAsync(
+        StoreAndForwardMessage message, CancellationToken cancellationToken = default)
+    {
+        var payload = JsonSerializer.Deserialize<CachedCallPayload>(message.PayloadJson);
+        if (payload == null || string.IsNullOrEmpty(payload.SystemName) || string.IsNullOrEmpty(payload.MethodName))
+        {
+            _logger.LogError("Buffered ExternalSystem message {Id} has an unreadable payload; parking.", message.Id);
+            return false;
+        }
+
+        var (system, method) = await ResolveSystemAndMethodAsync(
+            payload.SystemName, payload.MethodName, cancellationToken);
+        if (system == null || method == null)
+        {
+            _logger.LogError(
+                "Buffered call to '{System}'/'{Method}' cannot be delivered — the system or method no longer exists; parking.",
+                payload.SystemName, payload.MethodName);
+            return false;
+        }
+
+        var parameters = payload.Parameters?.ToDictionary(kv => kv.Key, kv => (object?)kv.Value);
+        try
+        {
+            await InvokeHttpAsync(system, method, parameters, cancellationToken);
+            return true;
+        }
+        catch (PermanentExternalSystemException ex)
+        {
+            _logger.LogError(ex, "Buffered call to '{System}' failed permanently; parking.", payload.SystemName);
+            return false;
+        }
+        // TransientExternalSystemException propagates — the S&F engine retries.
+    }
+
+    private sealed record CachedCallPayload(
+        string SystemName,
+        string MethodName,
+        Dictionary<string, JsonElement>? Parameters);
+
    /// <summary>
    /// WP-6: Executes the HTTP request against the external system.
    /// </summary>
@@ -344,6 +344,42 @@ akka {{
            // any actor or HTTP handler touches the service.
            storeAndForwardService.StartAsync().GetAwaiter().GetResult();

+            // Register the store-and-forward delivery handlers so buffered
+            // ExternalSystem calls, cached DB writes and notifications are actually
+            // delivered by the retry sweep. Without this, every buffered message is
+            // persisted but never delivered. Each handler resolves its scoped consumer
+            // service in a fresh DI scope — the sweep runs on a timer thread, outside
+            // any request scope.
+            storeAndForwardService.RegisterDeliveryHandler(
+                ScadaLink.Commons.Types.Enums.StoreAndForwardCategory.ExternalSystem,
+                async msg =>
+                {
+                    using var scope = _serviceProvider.CreateScope();
+                    return await scope.ServiceProvider
+                        .GetRequiredService<ScadaLink.ExternalSystemGateway.ExternalSystemClient>()
+                        .DeliverBufferedAsync(msg);
+                });
+            storeAndForwardService.RegisterDeliveryHandler(
+                ScadaLink.Commons.Types.Enums.StoreAndForwardCategory.CachedDbWrite,
+                async msg =>
+                {
+                    using var scope = _serviceProvider.CreateScope();
+                    return await scope.ServiceProvider
+                        .GetRequiredService<ScadaLink.ExternalSystemGateway.DatabaseGateway>()
+                        .DeliverBufferedAsync(msg);
+                });
+            storeAndForwardService.RegisterDeliveryHandler(
+                ScadaLink.Commons.Types.Enums.StoreAndForwardCategory.Notification,
+                async msg =>
+                {
+                    using var scope = _serviceProvider.CreateScope();
+                    return await scope.ServiceProvider
+                        .GetRequiredService<ScadaLink.NotificationService.NotificationDeliveryService>()
+                        .DeliverBufferedAsync(msg);
+                });
+            _logger.LogInformation(
+                "Store-and-forward delivery handlers registered (ExternalSystem, CachedDbWrite, Notification)");
+
            var parkedMessageHandler = _actorSystem.ActorOf(
                Props.Create(() => new ParkedMessageHandlerActor(
                    storeAndForwardService, _nodeOptions.SiteId!)),
@@ -93,18 +93,75 @@ public class NotificationDeliveryService : INotificationDeliveryService
                Message = message
            });

+            // attemptImmediateDelivery: false — DeliverAsync was already attempted
+            // above; letting EnqueueAsync re-invoke the handler would send twice.
            await _storeAndForward.EnqueueAsync(
                StoreAndForwardCategory.Notification,
                listName,
                payload,
                originInstanceName,
                smtpConfig.MaxRetries > 0 ? smtpConfig.MaxRetries : null,
-                smtpConfig.RetryDelay > TimeSpan.Zero ? smtpConfig.RetryDelay : null);
+                smtpConfig.RetryDelay > TimeSpan.Zero ? smtpConfig.RetryDelay : null,
+                attemptImmediateDelivery: false);

            return new NotificationResult(true, null, WasBuffered: true);
        }
    }

+    /// <summary>
+    /// WP-11/12: Delivers a buffered notification during a store-and-forward retry
+    /// sweep — re-resolves the list, recipients and SMTP config and re-attempts
+    /// delivery. Returns true on success, false on permanent failure (the message
+    /// is parked); throws on a transient failure so the engine retries.
+    /// </summary>
+    public async Task<bool> DeliverBufferedAsync(
+        StoreAndForwardMessage message, CancellationToken cancellationToken = default)
+    {
+        var payload = JsonSerializer.Deserialize<BufferedNotification>(message.PayloadJson);
+        if (payload == null || string.IsNullOrEmpty(payload.ListName))
+        {
+            _logger.LogError("Buffered notification message {Id} has an unreadable payload; parking.", message.Id);
+            return false;
+        }
+
+        var list = await _repository.GetListByNameAsync(payload.ListName, cancellationToken);
+        if (list == null)
+        {
+            _logger.LogError(
+                "Buffered notification to list '{List}' cannot be delivered — the list no longer exists; parking.",
+                payload.ListName);
+            return false;
+        }
+
+        var recipients = await _repository.GetRecipientsByListIdAsync(list.Id, cancellationToken);
+        if (recipients.Count == 0)
+        {
+            _logger.LogError("Buffered notification to list '{List}' has no recipients; parking.", payload.ListName);
+            return false;
+        }
+
+        var smtpConfig = (await _repository.GetAllSmtpConfigurationsAsync(cancellationToken)).FirstOrDefault();
+        if (smtpConfig == null)
+        {
+            _logger.LogError("Buffered notification cannot be delivered — no SMTP configuration available; parking.");
+            return false;
+        }
+
+        try
+        {
+            await DeliverAsync(smtpConfig, recipients, payload.Subject, payload.Message, cancellationToken);
+            return true;
+        }
+        catch (SmtpPermanentException ex)
+        {
+            _logger.LogError(ex, "Buffered notification to list '{List}' failed permanently; parking.", payload.ListName);
+            return false;
+        }
+        // Transient SMTP errors propagate out of DeliverAsync — the S&F engine retries.
+    }
+
+    private sealed record BufferedNotification(string ListName, string Subject, string Message);
+
    /// <summary>
    /// Delivers an email via SMTP. Throws on failure.
    /// </summary>
@@ -22,7 +22,8 @@ public static class ServiceCollectionExtensions
            var storage = sp.GetRequiredService<StoreAndForwardStorage>();
            var options = sp.GetRequiredService<IOptions<StoreAndForwardOptions>>().Value;
            var logger = sp.GetRequiredService<ILogger<StoreAndForwardService>>();
-            return new StoreAndForwardService(storage, options, logger);
+            var replication = sp.GetRequiredService<ReplicationService>();
+            return new StoreAndForwardService(storage, options, logger, replication);
        });

        services.AddSingleton<ReplicationService>(sp =>
@@ -30,6 +30,7 @@ public class StoreAndForwardService
 {
    private readonly StoreAndForwardStorage _storage;
    private readonly StoreAndForwardOptions _options;
+    private readonly ReplicationService? _replication;
    private readonly ILogger<StoreAndForwardService> _logger;
    private Timer? _retryTimer;
    private int _retryInProgress;
@@ -48,11 +49,13 @@ public class StoreAndForwardService
    public StoreAndForwardService(
        StoreAndForwardStorage storage,
        StoreAndForwardOptions options,
-        ILogger<StoreAndForwardService> logger)
+        ILogger<StoreAndForwardService> logger,
+        ReplicationService? replication = null)
    {
        _storage = storage;
        _options = options;
        _logger = logger;
+        _replication = replication;
    }

    /// <summary>
@@ -109,7 +112,8 @@ public class StoreAndForwardService
        string payloadJson,
        string? originInstanceName = null,
        int? maxRetries = null,
-        TimeSpan? retryInterval = null)
+        TimeSpan? retryInterval = null,
+        bool attemptImmediateDelivery = true)
    {
        var message = new StoreAndForwardMessage
        {
@@ -125,8 +129,10 @@ public class StoreAndForwardService
            OriginInstanceName = originInstanceName
        };

-        // Attempt immediate delivery
-        if (_deliveryHandlers.TryGetValue(category, out var handler))
+        // Attempt immediate delivery — unless the caller has already made a
+        // delivery attempt of its own (attemptImmediateDelivery: false). In that
+        // case re-invoking the handler here would dispatch the request twice.
+        if (attemptImmediateDelivery && _deliveryHandlers.TryGetValue(category, out var handler))
        {
            try
            {
@@ -136,12 +142,10 @@ public class StoreAndForwardService
                    RaiseActivity("Delivered", category, $"Immediate delivery to {target}");
                    return new StoreAndForwardResult(true, message.Id, false);
                }
-                else
-                {
+
                // Permanent failure — do not buffer
                return new StoreAndForwardResult(false, message.Id, false);
            }
-            }
            catch (Exception ex)
            {
                // Transient failure — buffer for retry
@@ -152,19 +156,39 @@ public class StoreAndForwardService
                message.LastAttemptAt = DateTimeOffset.UtcNow;
                message.RetryCount = 1;
                message.LastError = ex.Message;
-                await _storage.EnqueueAsync(message);
+                await BufferAsync(message);

                RaiseActivity("Queued", category, $"Buffered for retry: {target} ({ex.Message})");
                return new StoreAndForwardResult(true, message.Id, true);
            }
        }

-        // No handler registered — buffer for later
-        await _storage.EnqueueAsync(message);
-        RaiseActivity("Queued", category, $"No handler registered, buffered: {target}");
+        // Either no handler is registered yet, or the caller already attempted
+        // delivery itself — buffer for the background retry sweep to deliver.
+        if (!attemptImmediateDelivery)
+        {
+            // The caller made (and failed) one attempt before handing the
+            // message over, so it counts as the first retry.
+            message.RetryCount = 1;
+            message.LastAttemptAt = DateTimeOffset.UtcNow;
+        }
+        await BufferAsync(message);
+        RaiseActivity("Queued", category, attemptImmediateDelivery
+            ? $"No handler registered, buffered: {target}"
+            : $"Buffered for retry: {target}");
        return new StoreAndForwardResult(true, message.Id, true);
    }

+    /// <summary>
+    /// Persists a message to the local SQLite buffer and (WP-11) replicates the
+    /// add to the standby node so a failover does not lose the buffered message.
+    /// </summary>
+    private async Task BufferAsync(StoreAndForwardMessage message)
+    {
+        await _storage.EnqueueAsync(message);
+        _replication?.ReplicateEnqueue(message);
+    }
+
    /// <summary>
    /// WP-10: Background retry sweep. Processes all pending messages that are due for retry.
    /// </summary>
@@ -210,6 +234,7 @@ public class StoreAndForwardService
            if (success)
            {
                await _storage.RemoveMessageAsync(message.Id);
+                _replication?.ReplicateRemove(message.Id);
                RaiseActivity("Delivered", message.Category,
                    $"Delivered to {message.Target} after {message.RetryCount} retries");
                return;
@@ -220,6 +245,7 @@ public class StoreAndForwardService
            message.LastAttemptAt = DateTimeOffset.UtcNow;
            message.LastError = "Permanent failure (handler returned false)";
            await _storage.UpdateMessageAsync(message);
+            _replication?.ReplicatePark(message);
            RaiseActivity("Parked", message.Category,
                $"Permanent failure for {message.Target}: handler returned false");
        }
@@ -234,6 +260,7 @@ public class StoreAndForwardService
            {
                message.Status = StoreAndForwardMessageStatus.Parked;
                await _storage.UpdateMessageAsync(message);
+                _replication?.ReplicatePark(message);
                RaiseActivity("Parked", message.Category,
                    $"Max retries ({message.MaxRetries}) reached for {message.Target}");
                _logger.LogWarning(
@@ -405,4 +405,64 @@ public class ScriptAnalysisServiceTests
        Assert.Contains("name", resp.Markdown);
        Assert.Contains("String", resp.Markdown);
    }
+
+    // ── CentralUI-001: trust-model gate before sandbox execution ──────────
+
+    [Fact]
+    public void Diagnose_FullyQualifiedForbiddenCall_RaisesSCADA002()
+    {
+        // A forbidden API reached by fully-qualified name (no `using`, no bare
+        // type identifier) must still be flagged — the pre-fix semantic check
+        // only inspected the leftmost identifier and missed this shape.
+        var resp = _svc.Diagnose(new DiagnoseRequest(
+            "var d = System.IO.Directory.GetCurrentDirectory(); return d;"));
+        Assert.Contains(resp.Markers, m => m.Code == "SCADA002");
+    }
+
+    [Fact]
+    public async Task RunInSandbox_FullyQualifiedForbiddenApi_IsBlockedBeforeExecution()
+    {
+        // Regression test for CentralUI-001. RunInSandboxAsync used to execute any
+        // script that compiled, with no trust-model enforcement — so fully-qualified
+        // forbidden API code ran in the central host process. The fix gates execution
+        // on the forbidden-API analysis.
+        var result = await _svc.RunInSandboxAsync(
+            new SandboxRunRequest(
+                "var d = System.IO.Directory.GetCurrentDirectory(); return d;",
+                Parameters: null,
+                TimeoutSeconds: null),
+            CancellationToken.None);
+
+        Assert.False(result.Success);
+        Assert.Equal(SandboxErrorKind.CompileError, result.ErrorKind);
+        Assert.Contains("trust model", result.Error);
+        Assert.NotNull(result.Markers);
+        Assert.Contains(result.Markers!, m => m.Code is "SCADA001" or "SCADA002");
+    }
+
+    [Fact]
+    public async Task RunInSandbox_ForbiddenUsingDirective_IsBlockedBeforeExecution()
+    {
+        var result = await _svc.RunInSandboxAsync(
+            new SandboxRunRequest(
+                "using System.Diagnostics; var p = Process.GetCurrentProcess().Id; return p;",
+                Parameters: null,
+                TimeoutSeconds: null),
+            CancellationToken.None);
+
+        Assert.False(result.Success);
+        Assert.Equal(SandboxErrorKind.CompileError, result.ErrorKind);
+    }
+
+    [Fact]
+    public async Task RunInSandbox_CleanScript_StillRuns()
+    {
+        // The gate must not block a script that stays within the allowed surface.
+        var result = await _svc.RunInSandboxAsync(
+            new SandboxRunRequest("return 21 * 2;", Parameters: null, TimeoutSeconds: null),
+            CancellationToken.None);
+
+        Assert.True(result.Success);
+        Assert.Equal("42", result.ReturnValueJson);
+    }
 }
@@ -0,0 +1,77 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using NSubstitute;
+using ScadaLink.Commons.Entities.Instances;
+using ScadaLink.Commons.Entities.Sites;
+using ScadaLink.Commons.Interfaces.Repositories;
+using ScadaLink.Communication;
+using ScadaLink.Communication.Actors;
+using ScadaLink.Communication.Grpc;
+
+namespace ScadaLink.Communication.Tests;
+
+/// <summary>
+/// Tests for DebugStreamService session lifecycle.
+/// </summary>
+public class DebugStreamServiceTests : TestKit
+{
+    [Fact]
+    public async Task StartStreamAsync_StreamTerminatesBeforeSnapshot_ThrowsMeaningfulException()
+    {
+        // Regression test for Communication-001. When the debug stream terminates before
+        // the initial snapshot arrives, StartStreamAsync used to let the raw
+        // InvalidOperationException from onTerminatedWrapper escape its
+        // OperationCanceledException-only catch — the caller saw an untranslated exception
+        // and the failure path did not deterministically tear the bridge actor down.
+        // The fix catches any failure, tells the bridge actor StopDebugStream, and throws
+        // a descriptive exception that names the instance and wraps the underlying cause.
+        var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 };
+        var site = new Site("Site One", "site-1")
+        {
+            Id = 3,
+            GrpcNodeAAddress = "http://localhost:5100",
+            GrpcNodeBAddress = "http://localhost:5200"
+        };
+
+        var instanceRepo = Substitute.For<ITemplateEngineRepository>();
+        instanceRepo.GetInstanceByIdAsync(7, Arg.Any<CancellationToken>()).Returns(instance);
+        var siteRepo = Substitute.For<ISiteRepository>();
+        siteRepo.GetSiteByIdAsync(3, Arg.Any<CancellationToken>()).Returns(site);
+
+        var services = new ServiceCollection();
+        services.AddScoped(_ => instanceRepo);
+        services.AddScoped(_ => siteRepo);
+        using var provider = services.BuildServiceProvider();
+
+        var commProbe = CreateTestProbe();
+        var commService = new CommunicationService(
+            Options.Create(new CommunicationOptions()),
+            NullLogger<CommunicationService>.Instance);
+        commService.SetCommunicationActor(commProbe.Ref);
+
+        using var grpcFactory = new SiteStreamGrpcClientFactory(NullLoggerFactory.Instance);
+        var service = new DebugStreamService(
+            commService, provider, grpcFactory, NullLogger<DebugStreamService>.Instance);
+        service.SetActorSystem(Sys);
+
+        // Act — start the stream; it blocks awaiting the initial snapshot.
+        var startTask = service.StartStreamAsync(instanceId: 7, onEvent: _ => { }, onTerminated: () => { });
+
+        // The bridge actor's PreStart sends SubscribeDebugViewRequest to the comm actor;
+        // the envelope's sender is the bridge actor itself.
+        commProbe.ExpectMsg<SiteEnvelope>(TimeSpan.FromSeconds(5));
+        var bridgeActor = commProbe.LastSender;
+
+        // Simulate the site terminating the stream before any snapshot is delivered.
+        bridgeActor.Tell(new DebugStreamTerminated("site-1", "corr"));
+
+        // Assert — a descriptive exception that names the instance and wraps the cause,
+        // not the raw "terminated before snapshot received" InvalidOperationException.
+        var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => startTask);
+        Assert.Contains("Site1.Pump01", ex.Message);
+        Assert.NotNull(ex.InnerException);
+    }
+}
@@ -458,4 +458,87 @@ public class DataConnectionActorTests : TestKit
        await backupAdapter.Received().SubscribeAsync(
            "sensor/temp", Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>());
    }
+
+    // ── DataConnectionLayer-001: subscribe must not mutate actor state off-thread ──
+
+    private static async Task<string> DelayedSubscribeAsync()
+    {
+        // A short delay so concurrent subscribe background tasks pile up and their
+        // post-await state mutations would race under the pre-fix implementation.
+        await Task.Delay(1);
+        return "sub-" + Guid.NewGuid().ToString("N");
+    }
+
+    [Fact]
+    public async Task DCL001_ConcurrentSubscribes_DoNotCorruptSubscriptionCounters()
+    {
+        // Regression test for DataConnectionLayer-001. HandleSubscribe used to mutate
+        // actor state (_subscriptionIds, _totalSubscribed, _resolvedTags, the per-instance
+        // HashSet) from a Task.Run background thread. Many concurrent subscribes then race
+        // on non-thread-safe Dictionary/HashSet and on non-atomic int++ — losing increments
+        // or throwing. After the fix every mutation is applied on the actor thread via a
+        // SubscribeCompleted message, so the final counts are exact.
+        _mockAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask);
+        _mockAdapter.Status.Returns(ConnectionHealth.Connected);
+        _mockAdapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(_ => DelayedSubscribeAsync());
+        _mockAdapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
+            .Returns(new ReadResult(false, null, null));
+
+        var actor = CreateConnectionActor("dcl001-concurrent");
+        await Task.Delay(300); // reach Connected state
+
+        const int instances = 30;
+        const int tagsPerInstance = 30;
+        for (var i = 0; i < instances; i++)
+        {
+            var tags = Enumerable.Range(0, tagsPerInstance)
+                .Select(j => $"inst{i}/tag{j}")
+                .ToArray();
+            actor.Tell(new SubscribeTagsRequest(
+                $"corr{i}", $"inst{i}", "dcl001-concurrent", tags, DateTimeOffset.UtcNow));
+        }
+
+        // Every subscribe must be acknowledged.
+        for (var i = 0; i < instances; i++)
+            ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(15));
+
+        actor.Tell(new DataConnectionActor.GetHealthReport());
+        var report = ExpectMsg<DataConnectionHealthReport>(TimeSpan.FromSeconds(5));
+
+        // Every tag is distinct, so each is a fresh, resolved subscription.
+        Assert.Equal(instances * tagsPerInstance, report.TotalSubscribedTags);
+        Assert.Equal(instances * tagsPerInstance, report.ResolvedTags);
+    }
+
+    [Fact]
+    public async Task DCL001_SubscribeWithFailedTags_CountsResolvedAndUnresolvedSeparately()
+    {
+        // Behavioural guard: the restructured subscribe must preserve the original
+        // accounting — failed tags count toward TotalSubscribed but not ResolvedTags.
+        _mockAdapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask);
+        _mockAdapter.Status.Returns(ConnectionHealth.Connected);
+        _mockAdapter.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
+            .Returns(ci => ((string)ci[0]).StartsWith("bad")
+                ? Task.FromException<string>(new Exception("tag not found"))
+                : Task.FromResult("sub-ok"));
+        _mockAdapter.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
+            .Returns(new ReadResult(false, null, null));
+
+        var actor = CreateConnectionActor("dcl001-failed-tags");
+        await Task.Delay(300);
+
+        actor.Tell(new SubscribeTagsRequest(
+            "c1", "inst1", "dcl001-failed-tags",
+            ["good/a", "good/b", "good/c", "bad/x", "bad/y"], DateTimeOffset.UtcNow));
+        ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
+
+        actor.Tell(new DataConnectionActor.GetHealthReport());
+        var report = ExpectMsg<DataConnectionHealthReport>(TimeSpan.FromSeconds(3));
+
+        Assert.Equal(5, report.TotalSubscribedTags); // all 5 tags tracked
+        Assert.Equal(3, report.ResolvedTags);        // only the 3 good ones resolved
+    }
 }
@@ -53,4 +53,26 @@ public class DatabaseGatewayTests
        await Assert.ThrowsAsync<InvalidOperationException>(
            () => gateway.CachedWriteAsync("nonexistent", "INSERT INTO t VALUES (1)"));
    }
+
+    // ── ExternalSystemGateway-001: buffered CachedDbWrite delivery handler ──
+
+    [Fact]
+    public async Task DeliverBuffered_ConnectionNoLongerExists_ReturnsFalseSoMessageParks()
+    {
+        _repository.GetAllDatabaseConnectionsAsync().Returns(new List<DatabaseConnectionDefinition>());
+        var gateway = new DatabaseGateway(_repository, NullLogger<DatabaseGateway>.Instance);
+
+        var message = new ScadaLink.StoreAndForward.StoreAndForwardMessage
+        {
+            Id = Guid.NewGuid().ToString("N"),
+            Category = ScadaLink.Commons.Types.Enums.StoreAndForwardCategory.CachedDbWrite,
+            Target = "gone-db",
+            PayloadJson =
+                """{"ConnectionName":"gone-db","Sql":"INSERT INTO t VALUES (1)","Parameters":null}""",
+        };
+
+        var delivered = await gateway.DeliverBufferedAsync(message);
+
+        Assert.False(delivered); // permanent — the S&F engine parks the message
+    }
 }
@@ -153,6 +153,69 @@ public class ExternalSystemClientTests
        Assert.False(result.WasBuffered);
    }

+    // ── ExternalSystemGateway-001: buffered-call delivery handler ──
+
+    private static ScadaLink.StoreAndForward.StoreAndForwardMessage BufferedCall(
+        string systemName, string methodName) =>
+        new()
+        {
+            Id = Guid.NewGuid().ToString("N"),
+            Category = ScadaLink.Commons.Types.Enums.StoreAndForwardCategory.ExternalSystem,
+            Target = systemName,
+            PayloadJson =
+                $$"""{"SystemName":"{{systemName}}","MethodName":"{{methodName}}","Parameters":null}""",
+        };
+
+    [Fact]
+    public async Task DeliverBuffered_SuccessfulHttp_ReturnsTrue()
+    {
+        var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com", "none") { Id = 1 };
+        var method = new ExternalSystemMethod("getData", "GET", "/data") { Id = 1, ExternalSystemDefinitionId = 1 };
+        _repository.GetAllExternalSystemsAsync().Returns(new List<ExternalSystemDefinition> { system });
+        _repository.GetMethodsByExternalSystemIdAsync(1).Returns(new List<ExternalSystemMethod> { method });
+
+        var httpClient = new HttpClient(new MockHttpMessageHandler(HttpStatusCode.OK, "{\"ok\":true}"));
+        _httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
+
+        var client = new ExternalSystemClient(
+            _httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance);
+
+        var delivered = await client.DeliverBufferedAsync(BufferedCall("TestAPI", "getData"));
+
+        Assert.True(delivered);
+    }
+
+    [Fact]
+    public async Task DeliverBuffered_SystemNoLongerExists_ReturnsFalseSoMessageParks()
+    {
+        _repository.GetAllExternalSystemsAsync().Returns(new List<ExternalSystemDefinition>());
+
+        var client = new ExternalSystemClient(
+            _httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance);
+
+        var delivered = await client.DeliverBufferedAsync(BufferedCall("GoneAPI", "method"));
+
+        Assert.False(delivered); // permanent — the S&F engine parks the message
+    }
+
+    [Fact]
+    public async Task DeliverBuffered_Transient500_ThrowsSoEngineRetries()
+    {
+        var system = new ExternalSystemDefinition("TestAPI", "https://api.example.com", "none") { Id = 1 };
+        var method = new ExternalSystemMethod("failMethod", "POST", "/fail") { Id = 1, ExternalSystemDefinitionId = 1 };
+        _repository.GetAllExternalSystemsAsync().Returns(new List<ExternalSystemDefinition> { system });
+        _repository.GetMethodsByExternalSystemIdAsync(1).Returns(new List<ExternalSystemMethod> { method });
+
+        var httpClient = new HttpClient(new MockHttpMessageHandler(HttpStatusCode.InternalServerError, "boom"));
+        _httpClientFactory.CreateClient(Arg.Any<string>()).Returns(httpClient);
+
+        var client = new ExternalSystemClient(
+            _httpClientFactory, _repository, NullLogger<ExternalSystemClient>.Instance);
+
+        await Assert.ThrowsAsync<TransientExternalSystemException>(
+            () => client.DeliverBufferedAsync(BufferedCall("TestAPI", "failMethod")));
+    }
+
    /// <summary>
    /// Test helper: mock HTTP message handler.
    /// </summary>
@@ -192,4 +192,37 @@ public class NotificationDeliveryServiceTests
        Assert.True(result.Success);
        Assert.True(result.WasBuffered);
    }
+
+    // ── NotificationService-001: buffered-notification delivery handler ──
+
+    private static StoreAndForward.StoreAndForwardMessage BufferedNotification(string listName) =>
+        new()
+        {
+            Id = Guid.NewGuid().ToString("N"),
+            Category = ScadaLink.Commons.Types.Enums.StoreAndForwardCategory.Notification,
+            Target = listName,
+            PayloadJson = $$"""{"ListName":"{{listName}}","Subject":"Alert","Message":"Body"}""",
+        };
+
+    [Fact]
+    public async Task DeliverBuffered_HappyPath_ReturnsTrue()
+    {
+        SetupHappyPath();
+        var service = CreateService();
+
+        var delivered = await service.DeliverBufferedAsync(BufferedNotification("ops-team"));
+
+        Assert.True(delivered);
+    }
+
+    [Fact]
+    public async Task DeliverBuffered_ListNoLongerExists_ReturnsFalseSoMessageParks()
+    {
+        _repository.GetListByNameAsync("gone-list").Returns((NotificationList?)null);
+        var service = CreateService();
+
+        var delivered = await service.DeliverBufferedAsync(BufferedNotification("gone-list"));
+
+        Assert.False(delivered); // permanent — the S&F engine parks the message
+    }
 }
@@ -0,0 +1,108 @@
+using Microsoft.Data.Sqlite;
+using Microsoft.Extensions.Logging.Abstractions;
+using ScadaLink.Commons.Types.Enums;
+
+namespace ScadaLink.StoreAndForward.Tests;
+
+/// <summary>
+/// StoreAndForward-001: the active node must forward every buffer operation
+/// (add / remove / park) to the standby via the ReplicationService, so a
+/// failover does not lose the buffer.
+/// </summary>
+public class StoreAndForwardReplicationTests : IAsyncLifetime, IDisposable
+{
+    private readonly SqliteConnection _keepAlive;
+    private readonly StoreAndForwardStorage _storage;
+    private readonly StoreAndForwardService _service;
+    private readonly List<ReplicationOperation> _replicated = new();
+
+    public StoreAndForwardReplicationTests()
+    {
+        var connStr = $"Data Source=ReplTests_{Guid.NewGuid():N};Mode=Memory;Cache=Shared";
+        _keepAlive = new SqliteConnection(connStr);
+        _keepAlive.Open();
+
+        _storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
+
+        var options = new StoreAndForwardOptions
+        {
+            DefaultRetryInterval = TimeSpan.Zero,
+            DefaultMaxRetries = 1,
+            RetryTimerInterval = TimeSpan.FromMinutes(10),
+            ReplicationEnabled = true,
+        };
+
+        var replication = new ReplicationService(options, NullLogger<ReplicationService>.Instance);
+        replication.SetReplicationHandler(op =>
+        {
+            lock (_replicated) _replicated.Add(op);
+            return Task.CompletedTask;
+        });
+
+        _service = new StoreAndForwardService(
+            _storage, options, NullLogger<StoreAndForwardService>.Instance, replication);
+    }
+
+    public async Task InitializeAsync() => await _storage.InitializeAsync();
+    public Task DisposeAsync() => Task.CompletedTask;
+    public void Dispose() => _keepAlive.Dispose();
+
+    /// <summary>Replication is fire-and-forget (Task.Run); poll until the expected ops arrive.</summary>
+    private async Task<List<ReplicationOperation>> WaitForReplicationAsync(int count)
+    {
+        for (var i = 0; i < 100; i++)
+        {
+            lock (_replicated)
+                if (_replicated.Count >= count) return _replicated.ToList();
+            await Task.Delay(20);
+        }
+        lock (_replicated) return _replicated.ToList();
+    }
+
+    [Fact]
+    public async Task BufferingAMessage_ReplicatesAnAddOperation()
+    {
+        // No handler registered → message is buffered → an Add is replicated.
+        var result = await _service.EnqueueAsync(
+            StoreAndForwardCategory.ExternalSystem, "api", """{}""");
+        Assert.True(result.WasBuffered);
+
+        var ops = await WaitForReplicationAsync(1);
+        Assert.Contains(ops, o =>
+            o.OperationType == ReplicationOperationType.Add && o.MessageId == result.MessageId);
+    }
+
+    [Fact]
+    public async Task SuccessfulRetry_ReplicatesARemoveOperation()
+    {
+        var calls = 0;
+        _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
+            _ => ++calls == 1
+                ? throw new HttpRequestException("transient")
+                : Task.FromResult(true));
+
+        var result = await _service.EnqueueAsync(
+            StoreAndForwardCategory.ExternalSystem, "api", """{}""");
+        await _service.RetryPendingMessagesAsync();
+
+        var ops = await WaitForReplicationAsync(2);
+        Assert.Contains(ops, o => o.OperationType == ReplicationOperationType.Add);
+        Assert.Contains(ops, o =>
+            o.OperationType == ReplicationOperationType.Remove && o.MessageId == result.MessageId);
+    }
+
+    [Fact]
+    public async Task ParkedMessage_ReplicatesAParkOperation()
+    {
+        _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
+            _ => throw new HttpRequestException("always fails"));
+
+        var result = await _service.EnqueueAsync(
+            StoreAndForwardCategory.ExternalSystem, "api", """{}""", maxRetries: 1);
+        await _service.RetryPendingMessagesAsync();
+
+        var ops = await WaitForReplicationAsync(2);
+        Assert.Contains(ops, o =>
+            o.OperationType == ReplicationOperationType.Park && o.MessageId == result.MessageId);
+    }
+}
@@ -310,4 +310,28 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable
        Assert.Equal(100, msg!.MaxRetries);
        Assert.Equal(60000, msg.RetryIntervalMs);
    }
+
+    // ── attemptImmediateDelivery: false — caller already attempted delivery ──
+
+    [Fact]
+    public async Task EnqueueAsync_AttemptImmediateDeliveryFalse_BuffersWithoutInvokingHandler()
+    {
+        // A caller that has already made its own delivery attempt passes
+        // attemptImmediateDelivery: false so the request is not dispatched twice.
+        var handlerCalls = 0;
+        _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
+            _ => { Interlocked.Increment(ref handlerCalls); return Task.FromResult(true); });
+
+        var result = await _service.EnqueueAsync(
+            StoreAndForwardCategory.ExternalSystem, "api", """{}""",
+            attemptImmediateDelivery: false);
+
+        Assert.Equal(0, handlerCalls);   // handler NOT invoked at enqueue time
+        Assert.True(result.WasBuffered);
+
+        var msg = await _storage.GetMessageByIdAsync(result.MessageId);
+        Assert.NotNull(msg);
+        Assert.Equal(StoreAndForwardMessageStatus.Pending, msg!.Status);
+        Assert.Equal(1, msg.RetryCount);  // counts as the caller's first attempt
+    }
 }
Author	SHA1	Message	Date
Joseph Doherty	61253e3269	fix(store-and-forward): resolve S&F delivery + replication wiring (3 Critical findings) Resolves StoreAndForward-001, ExternalSystemGateway-001, NotificationService-001 — one systemic gap where buffered messages were persisted but never delivered, and the active node never replicated its buffer to the standby. Delivery handlers (ExternalSystemGateway-001 / NotificationService-001): - AkkaHostedService registers delivery handlers for the ExternalSystem, CachedDbWrite and Notification categories after StoreAndForwardService starts; each resolves its scoped consumer in a fresh DI scope. - ExternalSystemClient, DatabaseGateway and NotificationDeliveryService each gain a DeliverBufferedAsync method: re-resolve the target and re-attempt delivery, returning true/false/throwing per the transient-vs-permanent contract. - EnqueueAsync gains an attemptImmediateDelivery flag; CachedCallAsync and NotificationDeliveryService.SendAsync pass false (they already attempted delivery themselves) so registering a handler does not dispatch twice. Replication (StoreAndForward-001): - ReplicationService is injected into StoreAndForwardService; a new BufferAsync helper replicates every enqueue, and successful-retry removes and parks are replicated too. Fire-and-forget, no-op when replication is disabled. Tests: StoreAndForwardReplicationTests (Add/Remove/Park observed), attemptImmediateDelivery behaviour, and DeliverBufferedAsync paths for each consumer. Full solution builds; StoreAndForward/ExternalSystemGateway/ NotificationService suites green.	2026-05-16 18:58:11 -04:00
Joseph Doherty	a9bd7ee37c	fix(central-ui): resolve CentralUI-001 — enforce script trust model before sandbox execution ScriptAnalysisService.RunInSandboxAsync compiled and executed arbitrary user C# in the central host process with no trust-model enforcement — the forbidden-API set was only a Monaco editor diagnostic. A Design-role user could run System.IO/Process/Reflection/network code on the central node. Added a Roslyn semantic gate (EnforceTrustModel) invoked after compilation and before script.RunAsync, and on nested shared scripts in callSharedFunc; a script referencing any forbidden API is rejected before it runs. Reworked FindForbiddenApiUsages: it now resolves every identifier against the semantic model and checks types and members, so a fully-qualified call (System.IO.File.WriteAllText) is caught — the pre-fix check only inspected the leftmost identifier and missed that shape. This is a static semantic gate, not a process sandbox. Adds gate regression tests that fail against the pre-fix code, plus a clean-script test guarding against over-blocking.	2026-05-16 18:41:12 -04:00
Joseph Doherty	a9ceba00d0	fix(communication): resolve Communication-001 — early stream termination handling DebugStreamService.StartStreamAsync awaited the initial debug snapshot inside a try whose only handler was catch (OperationCanceledException). When the stream terminated before the snapshot arrived, onTerminatedWrapper completed the await with an InvalidOperationException that escaped the catch — the caller got a raw, untranslated exception and the service did no teardown of its own on that path. Replaced with catch (Exception): it removes the session entry, sends StopDebugStream to the bridge actor via the local reference (deterministic teardown, idempotent), and throws a descriptive exception — TimeoutException for the 30s timeout, otherwise an InvalidOperationException naming the instance/site and wrapping the cause. Re-triaged Critical -> Medium: the originally-claimed multi-minute site-side resource leak does not occur (the bridge actor self-terminates on every onTerminated path). Adds the first DebugStreamService test, which fails against the pre-fix code.	2026-05-16 18:32:52 -04:00
Joseph Doherty	239bee3bc4	fix(data-connection): resolve DataConnectionLayer-001 — off-thread actor state mutation HandleSubscribe spawned a Task.Run that mutated DataConnectionActor private state (_subscriptionIds, _subscriptionsByInstance, _totalSubscribed, _resolvedTags, _unresolvedTags) from a thread-pool thread, racing the actor's own message loop — a data race on non-thread-safe Dictionary/HashSet and non-atomic counters. Restructured HandleSubscribe to follow the actor's existing PipeTo(Self) pattern: the background task now performs only adapter I/O and pipes a SubscribeCompleted message to Self; all subscription-state mutation happens in the new HandleSubscribeCompleted handler on the actor thread (wired into the Connected, Connecting and Reconnecting states). Adds DCL001_ConcurrentSubscribes_DoNotCorruptSubscriptionCounters (30x30 concurrent subscribes) which fails against the pre-fix code and passes after.	2026-05-16 18:26:43 -04:00