diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbPhase2ConvergenceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbPhase2ConvergenceTests.cs
new file mode 100644
index 00000000..610a9e4c
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/LocalDbPhase2ConvergenceTests.cs
@@ -0,0 +1,193 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.LocalDb;
+using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
+
+namespace ZB.MOM.WW.ScadaBridge.IntegrationTests;
+
+/// <summary>
+/// Phase 2 convergence driven through the REAL <see cref="SiteStorageService"/>, over a real
+/// site pair, rather than through hand-written SQL.
+/// </summary>
+/// <remarks>
+/// <para>
+/// The sibling suites (<see cref="LocalDbConfigConvergenceTests"/>,
+/// <see cref="LocalDbStoreAndForwardConvergenceTests"/>) were written as specifications
+/// BEFORE the cutover, so they necessarily use hand-written SQL — the production writers were
+/// still going through the bespoke replicator at the time. This suite runs after the cutover
+/// and therefore drives the production writer, which is what makes the cascade scenario below
+/// meaningful: it is the shipped multi-statement transaction under test, not a re-creation
+/// of it.
+/// </para>
+/// <para>
+/// <see cref="SiteStorageService"/> is constructed directly on the fixture's
+/// <see cref="ILocalDb"/> — the same instance the host registers — so its writes flow through
+/// the same CDC triggers.
+/// </para>
+/// </remarks>
+[Collection("LocalDbSitePairConvergence")]
+public sealed class LocalDbPhase2ConvergenceTests : LocalDbSitePairHarness
+{
+    private SiteStorageService Storage(ILocalDb db) =>
+        new(db, NullLogger<SiteStorageService>.Instance);
+
+    // ---- read helpers -----------------------------------------------------------------
+
+    private static Task<long> CountAsync(ILocalDb db, string table, string instance) =>
+        ScalarAsync(db, $"SELECT COUNT(*) FROM {table} WHERE instance_unique_name = @instance",
+            new { instance });
+
+    private static async Task<long> ScalarAsync(ILocalDb db, string sql, object? args = null)
+    {
+        var rows = await db.QueryAsync(sql, static r => r.GetInt64(0), args);
+        return rows[0];
+    }
+
+    /// <summary>
+    /// The whole <c>deployed_configurations</c> row as one comparable string, so a scenario
+    /// can assert every column replicated rather than just the payload it happened to check.
+    /// </summary>
+    private static async Task<string?> ReadWholeConfigRowAsync(ILocalDb db, string instance)
+    {
+        var rows = await db.QueryAsync(
+            """
+            SELECT instance_unique_name || '|' || config_json || '|' || deployment_id || '|' ||
+                   revision_hash || '|' || is_enabled || '|' || deployed_at
+            FROM deployed_configurations WHERE instance_unique_name = @instance
+            """,
+            static r => r.GetString(0), new { instance });
+        return rows.Count == 0 ? null : rows[0];
+    }
+
+    /// <summary>Unacked oplog backlog. Drains to zero once the peer acknowledges.</summary>
+    private static Task<long> OplogDepthAsync(ILocalDb db) =>
+        ScalarAsync(db,
+            "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > " +
+            "(SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)");
+
+    // ---- scenarios --------------------------------------------------------------------
+
+    [Fact]
+    public async Task DeployedConfigRow_ConvergesToB_ColumnForColumn()
+    {
+        // LocalDbConfigConvergenceTests asserts config_json converges. That is the payload
+        // only: a capture that dropped, defaulted or reordered any other column would still
+        // pass it. deployed_at is the one that matters most — SiteReconciliationActor's
+        // guarded write compares it, so a node whose replicated copy carried a different
+        // timestamp would make different staleness decisions from its peer.
+        await Storage(A).StoreDeployedConfigAsync(
+            "inst-columns", """{"setpoint":7}""", "dep-col-1", "hash-col-1", isEnabled: true);
+
+        var onA = await ReadWholeConfigRowAsync(A, "inst-columns");
+        Assert.NotNull(onA);
+
+        await WaitUntilAsync(
+            async () => await ReadWholeConfigRowAsync(B, "inst-columns") == onA,
+            "node B's deployed_configurations row to match node A's in EVERY column");
+    }
+
+    [Fact]
+    public async Task RemovingAnInstance_ConvergesAllThreeCascadeTables()
+    {
+        // The plan flagged this as the most likely real defect in Phase 2, and the reasoning
+        // is sound: RemoveDeployedConfigAsync deletes from three tables in one transaction,
+        // the schema has NO foreign keys, and CDC captures the three deletes as independent
+        // per-table streams that LWW may apply to the peer in any order. Nothing in the
+        // system re-derives an orphan, so a dropped delete leaves a permanently stale
+        // static_attribute_overrides or native_alarm_state row on the standby — invisible
+        // until that instance name is redeployed and picks up ghost overrides.
+        var storageA = Storage(A);
+
+        // A second instance that is never removed. Without it this test cannot distinguish
+        // "the cascade converged" from "node B lost these tables entirely" — a registration
+        // bug that wiped B would satisfy the absence assertions on its own.
+        foreach (var instance in new[] { "inst-cascade", "inst-survivor" })
+        {
+            await storageA.StoreDeployedConfigAsync(
+                instance, """{"v":1}""", $"dep-{instance}", $"hash-{instance}", isEnabled: true);
+            await storageA.SetStaticOverrideAsync(instance, "Setpoint", "42");
+            await storageA.SetStaticOverrideAsync(instance, "Mode", "Auto");
+            await storageA.UpsertNativeAlarmAsync(
+                instance, "Area1.Line1.Alarm", "ref-1", """{"active":true}""", DateTimeOffset.UtcNow);
+        }
+
+        await WaitUntilAsync(
+            async () =>
+                await CountAsync(B, "deployed_configurations", "inst-cascade") == 1 &&
+                await CountAsync(B, "static_attribute_overrides", "inst-cascade") == 2 &&
+                await CountAsync(B, "native_alarm_state", "inst-cascade") == 1,
+            "all three tables to reach node B before the instance is removed");
+
+        await storageA.RemoveDeployedConfigAsync("inst-cascade");
+
+        await WaitUntilAsync(
+            async () =>
+                await CountAsync(B, "deployed_configurations", "inst-cascade") == 0 &&
+                await CountAsync(B, "static_attribute_overrides", "inst-cascade") == 0 &&
+                await CountAsync(B, "native_alarm_state", "inst-cascade") == 0,
+            "the 3-table cascade delete to converge on node B with NO orphans left behind");
+
+        // The control instance is untouched on both nodes: the cascade is scoped, not a wipe.
+        Assert.Equal(1, await CountAsync(B, "deployed_configurations", "inst-survivor"));
+        Assert.Equal(2, await CountAsync(B, "static_attribute_overrides", "inst-survivor"));
+        Assert.Equal(1, await CountAsync(B, "native_alarm_state", "inst-survivor"));
+    }
+
+    [Fact]
+    public async Task ANativeAlarmBurst_Converges_AndTheOplogDrains()
+    {
+        // native_alarm_state is the highest-rate replicated table (Task 1 found it bounded by
+        // per-SourceReference coalescing on a 100 ms flush, correcting the plan's D4). This
+        // asserts the burst converges AND that the oplog goes back to empty afterwards —
+        // convergence alone would still pass if entries replicated but were never acked, and
+        // an oplog that only grows eventually trips the caps and forces a snapshot resync.
+        const int burst = 200;
+        var storageA = Storage(A);
+        var at = DateTimeOffset.UtcNow;
+
+        for (var i = 0; i < burst; i++)
+        {
+            await storageA.UpsertNativeAlarmAsync(
+                "inst-burst", "Area1.Line1.Alarm", $"ref-{i}", $$"""{"seq":{{i}}}""", at);
+        }
+
+        await WaitUntilAsync(
+            async () => await CountAsync(B, "native_alarm_state", "inst-burst") == burst,
+            $"all {burst} alarm rows to reach node B");
+
+        await WaitUntilAsync(
+            async () => await OplogDepthAsync(A) == 0,
+            "node A's oplog to drain once node B acknowledges the burst");
+    }
+
+    [Fact]
+    public async Task RowsWrittenOnBWhileItsListenerIsDown_SurviveTheRejoin()
+    {
+        // Companion to LocalDbConfigConvergenceTests' N1 scenario, which covers the same
+        // rejoin for sf_messages and deployed_configurations. This one exercises the tables
+        // that scenario does not touch, because "the union survives" is a per-table property:
+        // it holds only if each table is actually registered, and an unregistered table is
+        // silently local-only rather than an error.
+        await Storage(A).StoreSharedScriptAsync("script-before", "return 1;", null, null);
+        await WaitUntilAsync(
+            async () => await ScalarAsync(B,
+                "SELECT COUNT(*) FROM shared_scripts WHERE name = 'script-before'") == 1,
+            "the pre-outage script to reach node B");
+
+        await StopPassiveAsync();
+
+        // Each node writes to a different Phase 2 table while they cannot see each other.
+        await Storage(A).StoreExternalSystemAsync(
+            "sys-from-a", "http://a.invalid", "None", null, null);
+        await Storage(B).SetStaticOverrideAsync("inst-offline", "WrittenWhileDown", "yes");
+
+        await RestartPairAsync();
+
+        await WaitUntilAsync(
+            async () =>
+                await ScalarAsync(A,
+                    "SELECT COUNT(*) FROM static_attribute_overrides WHERE attribute_name = 'WrittenWhileDown'") == 1 &&
+                await ScalarAsync(B,
+                    "SELECT COUNT(*) FROM external_systems WHERE name = 'sys-from-a'") == 1,
+            "both nodes to hold the union of what each wrote while partitioned");
+    }
+}