feat(sql): opt-in writes — allowWrites + per-tag writable, single-row parameterised UPDATE (lmxopcua#753)

allowWrites is no longer inert: SqlDriver implements IWritable, default off.

- Gates (outermost first): node Writable + WriteOperate (server), driver
  allowWrites, the tag's own TagConfig "writable": true (strict JSON boolean;
  anything else refuses the tag, and the deploy gate says why), and a model that
  names exactly one row — KeyValue, or WideRow by where-pair. topByTimestamp and
  Query are BadNotWritable.
- SqlWritePlanner: the only statement a write can become is
  UPDATE <table> SET <value col> = @v WHERE <key col> = @k, built from the
  catalog-gate-canonical identifiers (dialect-quoted) with value and key bound;
  value-before-key marker order so positional ODBC binds correctly. No authored
  write SQL, INSERT, UPSERT or procedure.
- Executed in a transaction that commits only when exactly one row was touched:
  0 rows -> BadNoData (never inserts), >1 -> rolled back, BadConfigurationError,
  unknown count -> rolled back, BadNotSupported. Bounded by operationTimeout;
  provider errors -> BadCommunicationError; health untouched.
- Discovery advertises Operate only where a write could succeed.
- AdminUI: driver form's Allow writes switch is live; tag editor gains a
  per-tag write switch; model validates writable strictly (parity corpus).
- Factory no longer warns "allowWrites ignored"; it logs whether writes are on.
- Tests: SqlWriteTests (planner SQL/bind order/injection/coercion, every gate,
  readback, absent key, two-row rollback, gate-rejected tag, discovery);
  MySqlReadTests write legs against InnoDB (keyed row only, two-row rollback,
  server-side grant refusal, default-off). Seed adds a setpoints table and the
  one UPDATE grant.

(lmxopcua#753)

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/mysql/seed-mysql.sql

@@ -6,8 +6,8 @@
 -- backends. Two things are load-bearing:
 --
 -- 1. The suite connects as `otopcua`, which holds SELECT on `sqlpoll` and nothing else (plus EXECUTE on the
---    side-effect function, below). A MySQL "schema" is a database, and information_schema shows a principal
---    only the objects it holds a privilege on — so MySqlDialect.ListSchemasSql returning `sqlpoll` for this
+--    side-effect function and UPDATE on the one write-target table, below). A MySQL "schema" is a database,
+--    and information_schema shows a principal only the objects it holds a privilege on — so MySqlDialect.ListSchemasSql returning `sqlpoll` for this
 --    login is the real provisioning check, not a superuser's view.
 -- 2. Every identifier is lower case. The suite authors the table name in UPPER case and expects the catalog
 --    gate to resolve it through a unique case-insensitive match: on Linux, MySQL table names are case-
@@ -65,9 +65,23 @@ BEGIN
 END//
 DELIMITER ;
 
+-- Write target (lmxopcua#753): the ONLY table the suite login may UPDATE. `dup` deliberately holds two rows
+-- for one key, so a write keyed on it matches two rows and must be rolled back. tag_values stays SELECT-only,
+-- so a write tag pointed there proves the server's own grant refuses it.
+CREATE TABLE setpoints (
+    tag_name varchar(128) NOT NULL,
+    sp_value double       NULL
+) ENGINE = InnoDB;
+INSERT INTO setpoints (tag_name, sp_value) VALUES
+    ('Line1.SpeedSp', 10.0),
+    ('Line1.TempSp',  20.0),
+    ('dup',           1.0),
+    ('dup',           2.0);
+
 -- ── the read-only login ──────────────────────────────────────────────────────────────────────
 DROP USER IF EXISTS 'otopcua'@'%';
 CREATE USER 'otopcua'@'%' IDENTIFIED BY '__APP_PASSWORD__';
 GRANT SELECT ON sqlpoll.* TO 'otopcua'@'%';
 GRANT EXECUTE ON FUNCTION sqlpoll.bump_counter TO 'otopcua'@'%';
+GRANT UPDATE ON sqlpoll.setpoints TO 'otopcua'@'%';
 FLUSH PRIVILEGES;

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/MySqlReadTests.cs

@@ -265,6 +265,88 @@ public sealed class MySqlReadTests
         (await _my.ScalarAsync("SELECT COUNT(*) FROM sqlpoll.side_effects")).ShouldBe(0L);
     }
 
+    // ---- writes (design §3.4) against a real server ----
+
+    /// <summary>All gates open: the UPDATE lands on exactly the keyed row, and the driver reads it back.</summary>
+    [Fact]
+    public async Task Write_realMySql_updatesTheKeyedRowOnly()
+    {
+        SkipUnlessLive();
+
+        await using var driver = await StartAsync(null, allowWrites: true,
+            Setpoint("Sql/Sp/Speed", "Line1.SpeedSp", writable: true));
+        var target = 10.0 + Random.Shared.Next(1, 1000);
+
+        (await WriteAsync(driver, "Sql/Sp/Speed", target)).ShouldBe(SqlStatusCodes.Good);
+
+        Convert.ToDouble(await _my.ScalarRawAsync("SELECT sp_value FROM sqlpoll.setpoints WHERE tag_name = 'Line1.SpeedSp'"),
+            CultureInfo.InvariantCulture).ShouldBe(target);
+        Convert.ToDouble(await _my.ScalarRawAsync("SELECT sp_value FROM sqlpoll.setpoints WHERE tag_name = 'Line1.TempSp'"),
+            CultureInfo.InvariantCulture).ShouldBe(20.0);
+        Convert.ToDouble((await driver.ReadAsync(["Sql/Sp/Speed"], TestContext.Current.CancellationToken))[0].Value,
+            CultureInfo.InvariantCulture).ShouldBe(target);
+    }
+
+    /// <summary>A key matching two rows is rolled back on InnoDB — neither row changes.</summary>
+    [Fact]
+    public async Task Write_realMySql_aKeyMatchingTwoRowsIsRolledBack()
+    {
+        SkipUnlessLive();
+
+        await using var driver = await StartAsync(null, allowWrites: true, Setpoint("Sql/Sp/Dup", "dup", writable: true));
+
+        (await WriteAsync(driver, "Sql/Sp/Dup", 99.0)).ShouldBe(SqlStatusCodes.BadConfigurationError);
+        (await _my.ScalarAsync("SELECT COUNT(*) FROM sqlpoll.setpoints WHERE tag_name = 'dup' AND sp_value = 99")).ShouldBe(0L);
+    }
+
+    /// <summary>
+    ///     Defence in depth: with every driver gate open, a table the login holds no UPDATE on is refused by the
+    ///     server itself (the provider error surfaces as BadCommunicationError) and the row is unchanged.
+    /// </summary>
+    [Fact]
+    public async Task Write_realMySql_aTableWithoutAnUpdateGrantIsRefusedByTheServer()
+    {
+        SkipUnlessLive();
+
+        await using var driver = await StartAsync(null, allowWrites: true, Tag("Sql/Kv/Speed", new JsonObject
+        {
+            ["model"] = "KeyValue", ["table"] = MySqlFixture.Qualified("tag_values"), ["keyColumn"] = "tag_name",
+            ["keyValue"] = MySqlFixture.PresentKey, ["valueColumn"] = "num_value", ["writable"] = true,
+        }));
+
+        (await WriteAsync(driver, "Sql/Kv/Speed", 1.0)).ShouldBe(SqlStatusCodes.BadCommunicationError);
+        Convert.ToDouble(await _my.ScalarRawAsync("SELECT num_value FROM sqlpoll.tag_values WHERE tag_name = 'Line1.Speed'"),
+            CultureInfo.InvariantCulture).ShouldBe(MySqlFixture.PresentValue);
+    }
+
+    /// <summary>The default: no allowWrites ⇒ refused before any connection, whatever the tag says.</summary>
+    [Fact]
+    public async Task Write_realMySql_withoutAllowWritesIsRefused()
+    {
+        SkipUnlessLive();
+
+        await using var driver = await StartAsync(null, allowWrites: false, Setpoint("Sql/Sp/Temp", "Line1.TempSp", writable: true));
+
+        (await WriteAsync(driver, "Sql/Sp/Temp", 77.0)).ShouldBe(SqlStatusCodes.BadNotWritable);
+        Convert.ToDouble(await _my.ScalarRawAsync("SELECT sp_value FROM sqlpoll.setpoints WHERE tag_name = 'Line1.TempSp'"),
+            CultureInfo.InvariantCulture).ShouldBe(20.0);
+    }
+
+    private static RawTagEntry Setpoint(string rawPath, string key, bool writable) => Tag(rawPath, new JsonObject
+    {
+        ["model"] = "KeyValue",
+        ["table"] = MySqlFixture.Qualified("setpoints"),
+        ["keyColumn"] = "tag_name",
+        ["keyValue"] = key,
+        ["valueColumn"] = "sp_value",
+        ["type"] = "Float64",
+        ["writable"] = writable,
+    });
+
+    private static async Task<uint> WriteAsync(SqlDriver driver, string reference, object? value)
+        => (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], TestContext.Current.CancellationToken))
+            .ShouldHaveSingleItem().StatusCode;
+
     // ---- helpers ----
 
     private void SkipUnlessLive()
@@ -274,13 +356,16 @@ public sealed class MySqlReadTests
 
     private Task<SqlDriver> StartAsync(params RawTagEntry[] tags) => StartAsync(queriesJson: null, tags);
 
+    private Task<SqlDriver> StartAsync(string? queriesJson, params RawTagEntry[] tags)
+        => StartAsync(queriesJson, allowWrites: false, tags);
+
     /// <summary>
     ///     Builds and initializes a driver through the PRODUCTION factory (<c>provider: MySql</c>, the connection
     ///     string resolved from the fixture's <c>connectionStringRef</c> environment variable), so the suite
     ///     exercises <c>MySqlConnectorFactory.Instance</c> and the factory's own named-query validation exactly as a
     ///     deployed config would.
     /// </summary>
-    private async Task<SqlDriver> StartAsync(string? queriesJson, params RawTagEntry[] tags)
+    private async Task<SqlDriver> StartAsync(string? queriesJson, bool allowWrites, params RawTagEntry[] tags)
     {
         var rawTags = new JsonArray();
         foreach (var tag in tags)
@@ -302,6 +387,7 @@ public sealed class MySqlReadTests
             ["rawTags"] = rawTags,
         };
         if (queriesJson is not null) config["queries"] = JsonNode.Parse(queriesJson);
+        if (allowWrites) config["allowWrites"] = true;
         var json = config.ToJsonString();
 
         var driver = SqlDriverFactoryExtensions.CreateInstance("sql-mysql-integration", json);
@@ -433,6 +519,15 @@ public sealed class MySqlFixture : IAsyncLifetime
     /// <param name="sql">The statement.</param>
     /// <returns>The scalar, as a 64-bit integer when numeric.</returns>
     public async Task<object?> ScalarAsync(string sql)
+    {
+        var value = await ScalarRawAsync(sql).ConfigureAwait(false);
+        return value is IConvertible and not string ? Convert.ToInt64(value, CultureInfo.InvariantCulture) : value;
+    }
+
+    /// <summary>Runs a scalar query as the fixture login and returns the provider's value unconverted.</summary>
+    /// <param name="sql">The statement.</param>
+    /// <returns>The scalar.</returns>
+    public async Task<object?> ScalarRawAsync(string sql)
     {
         var connection = new MySqlDialect().Factory.CreateConnection()
             ?? throw new InvalidOperationException("MySqlConnector returned no connection object.");
@@ -444,8 +539,7 @@ public sealed class MySqlFixture : IAsyncLifetime
             await using (command.ConfigureAwait(false))
             {
                 command.CommandText = sql;
-                var value = await command.ExecuteScalarAsync().ConfigureAwait(false);
-                return value is IConvertible and not string ? Convert.ToInt64(value, CultureInfo.InvariantCulture) : value;
+                return await command.ExecuteScalarAsync().ConfigureAwait(false);
             }
         }
     }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverFactoryTests.cs

@@ -337,34 +337,43 @@ public sealed class SqlDriverFactoryTests
         options.RawTags.ShouldBeEmpty();
     }
 
-    // ---- v1 is read-only, structurally ----
+    // ---- allowWrites (lmxopcua#753): honoured now, default off ----
+
+    [Theory]
+    [InlineData(",\"allowWrites\":true", true)]
+    [InlineData(",\"allowWrites\":false", false)]
+    [InlineData("", false)]                         // absent ⇒ read-only
+    public void ParseBinding_carriesAllowWritesOntoTheOptions_defaultOff(string fragment, bool expected)
+    {
+        var name = NewRef();
+        using var _ = new EnvironmentVariableScope(
+            SqlConnectionStringResolver.EnvironmentVariableFor(name), "Server=srv;Database=db");
+
+        SqlDriverFactoryExtensions.ParseBinding("d1", $$"""{"connectionStringRef":"{{name}}"{{fragment}}}""")
+            .Options.AllowWrites.ShouldBe(expected);
+    }
 
     [Fact]
-    public void CreateInstance_anAuthoredAllowWrites_warnsLoudly_andTheDriverStaysReadOnly()
+    public void CreateInstance_withAllowWrites_logsNoWarning_andSaysWritesAreOn()
     {
-        // allowWrites is inert by construction (IWritable is not implemented), so honouring it is impossible
-        // and rejecting it would brick a deployment over a flag that cannot do harm. The remaining failure
-        // mode is an operator who believes writes are enabled — which only a loud warning closes.
+        // allowWrites used to be inert and drew a "this build ignores it" warning; it is honoured now, so the
+        // flag is an ordinary setting — reported in the configured line, never as a warning.
         var name = NewRef();
         using var _ = new EnvironmentVariableScope(
             SqlConnectionStringResolver.EnvironmentVariableFor(name), "Server=srv;Database=db");
         var loggerFactory = new RecordingLoggerFactory();
 
-        var driver = SqlDriverFactoryExtensions.CreateInstance(
-            "d1", $$"""{"connectionStringRef":"{{name}}","allowWrites":true}""", loggerFactory);
+        SqlDriverFactoryExtensions.CreateInstance(
+            "d1", $$"""{"connectionStringRef":"{{name}}","allowWrites":true}""", loggerFactory)
+            .ShouldBeAssignableTo<IWritable>();
 
-        driver.ShouldNotBeAssignableTo<IWritable>();
-        var warning = loggerFactory.Entries
-            .Where(e => e.Level >= LogLevel.Warning)
-            .ShouldHaveSingleItem();
-        warning.Message.ShouldContain("allowWrites");
-        warning.Message.ShouldContain("read-only");
+        loggerFactory.Entries.Where(e => e.Level >= LogLevel.Warning).ShouldBeEmpty();
+        loggerFactory.Entries.ShouldContain(e => e.Message.Contains("allowed for opted-in tags"));
     }
 
     [Fact]
     public void CreateInstance_withoutAllowWrites_logsNoWarning()
     {
-        // The falsifiability control for the warning above — a warning that always fires proves nothing.
         var name = NewRef();
         using var _ = new EnvironmentVariableScope(
             SqlConnectionStringResolver.EnvironmentVariableFor(name), "Server=srv;Database=db");

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs

@@ -46,8 +46,9 @@ public sealed class SqlDriverTests
 
         capture.Variables.ShouldContain(v =>
             v.Info.FullName == "Speed" && v.Info.SecurityClass == SecurityClassification.ViewOnly);
-        // v1 is read-only by construction, not by configuration.
-        driver.ShouldNotBeAssignableTo<IWritable>();
+        // Writes are off by default (no allowWrites): the driver is IWritable, but refuses.
+        (await ((IWritable)driver).WriteAsync([new WriteRequest("Speed", 1.0)], CancellationToken.None))
+            .ShouldHaveSingleItem().StatusCode.ShouldBe(SqlStatusCodes.BadNotWritable);
     }
 
     [Fact]

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlWriteTests.cs

@@ -0,0 +1,307 @@
+using System.Globalization;
+using System.Text.Json.Nodes;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests;
+
+/// <summary>
+///     The opt-in write path (design §3.4, lmxopcua#753): <c>allowWrites</c> + the tag's <c>"writable": true</c>
+///     + a single-row model, executed as a parameterised <c>UPDATE</c> over catalog-validated identifiers that
+///     commits only when exactly one row matched. Over SQLite, end to end through <see cref="SqlDriver"/>.
+/// </summary>
+public sealed class SqlWriteTests
+{
+    private const string DriverInstanceId = "sql-write";
+
+    // ---- tag opt-in parse ----
+
+    [Fact]
+    public void TryParse_readsWritable_andDefaultsItOff()
+    {
+        SqlEquipmentTagParser.TryParse(Kv(writable: true), "a", out var on).ShouldBeTrue();
+        on.Writable.ShouldBeTrue();
+        SqlEquipmentTagParser.TryParse(Kv(writable: null), "b", out var absent).ShouldBeTrue();
+        absent.Writable.ShouldBeFalse();
+    }
+
+    [Theory]
+    [InlineData("\"true\"")]
+    [InlineData("1")]
+    [InlineData("null")]
+    public void TryParse_aNonBooleanWritable_refusesTheTag_andTheGateSaysWhy(string writable)
+    {
+        var json = $$"""{"model":"KeyValue","table":"T","keyColumn":"k","keyValue":"x","valueColumn":"v","writable":{{writable}}}""";
+        SqlEquipmentTagParser.TryParse(json, "a", out _).ShouldBeFalse();
+        SqlTagDefinitionFactory.Inspect(json).ShouldContain(w => w.Contains("writable", StringComparison.Ordinal));
+    }
+
+    // ---- planner: the only statement a write can become ----
+
+    [Fact]
+    public void Plan_keyValue_isASingleRowUpdate_withTheValueAndKeyBound()
+    {
+        var tag = new SqlTagDefinition("A", SqlTagModel.KeyValue, "dbo.TagValues",
+            KeyColumn: "tag_name", KeyValue: "Line1.Speed", ValueColumn: "num_value", Writable: true);
+
+        SqlWritePlanner.TryPlan(tag, 12.5, new SqlServerDialect(), out var plan).ShouldBe(SqlStatusCodes.Good);
+
+        plan!.SqlText.ShouldBe("UPDATE [dbo].[TagValues] SET [num_value] = @v WHERE [tag_name] = @k");
+        plan.ParameterBindNames.ShouldBe(["@v", "@k"]);
+        plan.Parameters.ShouldBe([12.5, "Line1.Speed"]);
+    }
+
+    [Fact]
+    public void Plan_wideRowWherePair_updatesTheSelectedRowsColumn_valueBeforeKeyForPositionalProviders()
+    {
+        var tag = new SqlTagDefinition("A", SqlTagModel.WideRow, "plant.status", ColumnName: "setpoint",
+            RowSelectorColumn: "station_id", RowSelectorValue: "7", Writable: true);
+
+        SqlWritePlanner.TryPlan(tag, 3, new OdbcDialect(), out var plan).ShouldBe(SqlStatusCodes.Good);
+
+        plan!.SqlText.ShouldBe("UPDATE \"plant\".\"status\" SET \"setpoint\" = ? WHERE \"station_id\" = ?");
+        plan.Parameters.ShouldBe([3, "7"]);   // marker order: SET value, then WHERE key
+    }
+
+    [Fact]
+    public void Plan_aHostileValueIsBound_neverConcatenated()
+    {
+        const string hostile = "x'; DROP TABLE TagValues; --";
+        var tag = new SqlTagDefinition("A", SqlTagModel.KeyValue, "T", KeyColumn: "k", KeyValue: "x", ValueColumn: "v");
+
+        SqlWritePlanner.TryPlan(tag, hostile, new SqlServerDialect(), out var plan).ShouldBe(SqlStatusCodes.Good);
+
+        plan!.SqlText.ShouldNotContain("DROP");
+        plan.Parameters[0].ShouldBe(hostile);
+    }
+
+    [Fact]
+    public void Plan_refusesModelsWithNoSingleRowTarget()
+    {
+        var newest = new SqlTagDefinition("A", SqlTagModel.WideRow, "T", ColumnName: "c", RowSelectorTopByTimestamp: "ts");
+        var query = new SqlTagDefinition("B", SqlTagModel.Query, string.Empty, ColumnName: "c", QueryName: "q");
+
+        SqlWritePlanner.TryPlan(newest, 1, new SqlServerDialect(), out _).ShouldBe(SqlStatusCodes.BadNotWritable);
+        SqlWritePlanner.TryPlan(query, 1, new SqlServerDialect(), out _).ShouldBe(SqlStatusCodes.BadNotWritable);
+    }
+
+    [Fact]
+    public void Plan_coercesToTheDeclaredType_andRefusesWhatWillNotCoerce()
+    {
+        var tag = new SqlTagDefinition("A", SqlTagModel.KeyValue, "T", KeyColumn: "k", KeyValue: "x", ValueColumn: "v",
+            DeclaredType: DriverDataType.Int32);
+
+        SqlWritePlanner.TryPlan(tag, "41", new SqlServerDialect(), out var plan).ShouldBe(SqlStatusCodes.Good);
+        plan!.Parameters[0].ShouldBe(41);
+        SqlWritePlanner.TryPlan(tag, "not a number", new SqlServerDialect(), out _).ShouldBe(SqlStatusCodes.BadTypeMismatch);
+        SqlWritePlanner.TryPlan(tag with { DeclaredType = null }, new object(), new SqlServerDialect(), out _)
+            .ShouldBe(SqlStatusCodes.BadTypeMismatch);
+    }
+
+    // ---- the gates, end to end ----
+
+    [Fact]
+    public async Task Write_withAllowWritesOff_isRefused_andTheTableIsUntouched()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture, allowWrites: false, Entry("Speed", Kv(writable: true)));
+
+        (await WriteAsync(driver, "Speed", 99.0)).ShouldBe(SqlStatusCodes.BadNotWritable);
+        ValueOf(fixture, SqlitePollFixture.PresentKey).ShouldBe(SqlitePollFixture.PresentValue);
+    }
+
+    [Fact]
+    public async Task Write_toATagThatDidNotOptIn_isRefused_evenWithAllowWritesOn()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture, allowWrites: true, Entry("Speed", Kv(writable: null)));
+
+        (await WriteAsync(driver, "Speed", 99.0)).ShouldBe(SqlStatusCodes.BadNotWritable);
+        ValueOf(fixture, SqlitePollFixture.PresentKey).ShouldBe(SqlitePollFixture.PresentValue);
+    }
+
+    [Fact]
+    public async Task Write_allGatesOpen_updatesTheRow_andAReadSeesIt()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture, allowWrites: true,
+            Entry("Speed", Kv(writable: true)), Entry("Pressure", Kv(writable: true, key: SqlitePollFixture.SecondPresentKey)));
+
+        (await WriteAsync(driver, "Speed", 99.5)).ShouldBe(SqlStatusCodes.Good);
+
+        ValueOf(fixture, SqlitePollFixture.PresentKey).ShouldBe(99.5);
+        ValueOf(fixture, SqlitePollFixture.SecondPresentKey).ShouldBe(SqlitePollFixture.SecondPresentValue);   // only one row
+        var read = (await driver.ReadAsync(["Speed"], CancellationToken.None))[0];
+        Convert.ToDouble(read.Value, CultureInfo.InvariantCulture).ShouldBe(99.5);
+    }
+
+    [Fact]
+    public async Task Write_aWideRowWherePair_updatesThatStationsColumnOnly()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture, allowWrites: true, Entry("Oven7", $$"""
+            {"model":"WideRow","table":"{{SqlitePollFixture.WideRowTable}}","columnName":"oven_temp",
+             "rowSelector":{"whereColumn":"station_id","whereValue":7},"writable":true}
+            """));
+
+        (await WriteAsync(driver, "Oven7", 190.0)).ShouldBe(SqlStatusCodes.Good);
+
+        Scalar(fixture, "SELECT oven_temp FROM LatestStatus WHERE station_id = 7").ShouldBe(190.0);
+        Scalar(fixture, "SELECT oven_temp FROM LatestStatus WHERE station_id = 8").ShouldBe(SqlitePollFixture.NewestStationOvenTemp);
+    }
+
+    [Fact]
+    public async Task Write_toAnAbsentKey_isBadNoData_andInsertsNothing()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture, allowWrites: true,
+            Entry("Missing", Kv(writable: true, key: SqlitePollFixture.AbsentKey)));
+
+        (await WriteAsync(driver, "Missing", 1.0)).ShouldBe(SqlStatusCodes.BadNoData);
+        Scalar(fixture, "SELECT COUNT(*) FROM TagValues").ShouldBe(3.0);
+    }
+
+    /// <summary>
+    ///     A key that matches two rows would update both. The write is rolled back and refused, so neither row
+    ///     changes — the write-side twin of the read path's duplicate-key refusal.
+    /// </summary>
+    [Fact]
+    public async Task Write_matchingMoreThanOneRow_isRolledBack_andRefused()
+    {
+        using var fixture = new SqlitePollFixture();
+        fixture.Execute($"INSERT INTO TagValues (tag_name, num_value, sample_ts) VALUES ('{SqlitePollFixture.PresentKey}', 7.0, '2026-07-24T11:00:00Z')");
+        await using var driver = await StartAsync(fixture, allowWrites: true, Entry("Speed", Kv(writable: true)));
+
+        (await WriteAsync(driver, "Speed", 55.0)).ShouldBe(SqlStatusCodes.BadConfigurationError);
+
+        Scalar(fixture, $"SELECT COUNT(*) FROM TagValues WHERE num_value = 55.0").ShouldBe(0.0);
+    }
+
+    [Fact]
+    public async Task Write_toAnUnknownOrGateRejectedTag_isBadNodeIdUnknown()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture, allowWrites: true, Entry("Ghost", """
+            {"model":"KeyValue","table":"NoSuchTable","keyColumn":"k","keyValue":"x","valueColumn":"v","writable":true}
+            """));
+
+        (await WriteAsync(driver, "Ghost", 1.0)).ShouldBe(SqlStatusCodes.BadNodeIdUnknown);   // the catalog gate rejected it
+        (await WriteAsync(driver, "Nobody", 1.0)).ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
+    }
+
+    [Fact]
+    public async Task Write_aHostileString_isStoredVerbatim()
+    {
+        using var fixture = new SqlitePollFixture();
+        fixture.Execute("CREATE TABLE Notes (k TEXT NOT NULL, note TEXT NULL)");
+        fixture.Execute("INSERT INTO Notes VALUES ('a', 'old')");
+        const string hostile = "x'); DROP TABLE Notes; --";
+        await using var driver = await StartAsync(fixture, allowWrites: true, Entry("Note", """
+            {"model":"KeyValue","table":"Notes","keyColumn":"k","keyValue":"a","valueColumn":"note","writable":true,"type":"String"}
+            """));
+
+        (await WriteAsync(driver, "Note", hostile)).ShouldBe(SqlStatusCodes.Good);
+
+        using var command = fixture.Connection.CreateCommand();
+        command.CommandText = "SELECT note FROM Notes WHERE k = 'a'";
+        command.ExecuteScalar().ShouldBe(hostile);
+    }
+
+    [Fact]
+    public async Task Discover_advertisesOperate_onlyWhereAWriteCouldSucceed()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture, allowWrites: true,
+            Entry("Writable", Kv(writable: true)),
+            Entry("ReadOnly", Kv(writable: null)),
+            Entry("Newest", $$"""
+                {"model":"WideRow","table":"{{SqlitePollFixture.WideRowTable}}","columnName":"oven_temp",
+                 "rowSelector":{"topByTimestamp":"sample_ts"},"writable":true}
+                """));
+
+        var builder = new ClassCapturingBuilder();
+        await driver.DiscoverAsync(builder, CancellationToken.None);
+
+        builder.Classes["Writable"].ShouldBe(SecurityClassification.Operate);
+        builder.Classes["ReadOnly"].ShouldBe(SecurityClassification.ViewOnly);
+        builder.Classes["Newest"].ShouldBe(SecurityClassification.ViewOnly);   // no single-row target
+    }
+
+    // ---- helpers ----
+
+    private static string Kv(bool? writable, string key = SqlitePollFixture.PresentKey)
+    {
+        var o = new JsonObject
+        {
+            ["model"] = "KeyValue",
+            ["table"] = SqlitePollFixture.KeyValueTable,
+            ["keyColumn"] = SqlitePollFixture.KeyColumn,
+            ["keyValue"] = key,
+            ["valueColumn"] = SqlitePollFixture.ValueColumn,
+            ["type"] = "Float64",
+        };
+        if (writable is { } w) o["writable"] = w;
+        return o.ToJsonString();
+    }
+
+    private static RawTagEntry Entry(string rawPath, string json) => new(rawPath, json, WriteIdempotent: false);
+
+    private static async Task<uint> WriteAsync(SqlDriver driver, string reference, object? value)
+        => (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], CancellationToken.None))
+            .ShouldHaveSingleItem().StatusCode;
+
+    private static double ValueOf(SqlitePollFixture fixture, string key)
+        => Scalar(fixture, $"SELECT num_value FROM TagValues WHERE tag_name = '{key}'");
+
+    private static double Scalar(SqlitePollFixture fixture, string sql)
+    {
+        using var command = fixture.Connection.CreateCommand();
+        command.CommandText = sql;
+        return Convert.ToDouble(command.ExecuteScalar(), CultureInfo.InvariantCulture);
+    }
+
+    private static async Task<SqlDriver> StartAsync(SqlitePollFixture fixture, bool allowWrites, params RawTagEntry[] tags)
+    {
+        var driver = new SqlDriver(
+            new SqlDriverOptions
+            {
+                RawTags = tags,
+                OperationTimeout = TimeSpan.FromSeconds(15),
+                CommandTimeout = TimeSpan.FromSeconds(10),
+                AllowWrites = allowWrites,
+            },
+            DriverInstanceId, new SqliteDialect(), fixture.ConnectionString, fixture.Factory);
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        return driver;
+    }
+
+    /// <summary>Records each discovered variable's security classification.</summary>
+    private sealed class ClassCapturingBuilder : IAddressSpaceBuilder
+    {
+        public Dictionary<string, SecurityClassification> Classes { get; } = [];
+
+        public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;
+
+        public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
+        {
+            Classes[attributeInfo.FullName] = attributeInfo.SecurityClass;
+            return new Handle(attributeInfo.FullName);
+        }
+
+        public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
+
+        private sealed class Handle(string fullReference) : IVariableHandle
+        {
+            public string FullReference => fullReference;
+
+            public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => new Sink();
+
+            private sealed class Sink : IAlarmConditionSink
+            {
+                public void OnTransition(AlarmEventArgs args) { }
+            }
+        }
+    }
+}