feat(#80): broker editor service write ops (save handshake, clear, re-GET)

The write half of the broker editor, against RedCreek's confirmed field-at-a-time
contract. The existing 0xC0 key/value command already carries the broker keys --
no new wire codes -- so the codec comment at observer_config_client.dart:166 is
stale, not a gap.

- saveBroker(): disable a live slot first, write each changed field, write
  `enabled` LAST. Stops at the first ERR/timeout and reports the failed field --
  a partial save leaves the slot disabled, never live-corrupt (activation guard).
  TDD: ordering + partial-failure tests.
- clearBroker(): mqtt.broker.<slot>.clear.
- getBroker(): single-slot re-GET via 14 per-field GETs (recovery / post-save
  read), not the 84-frame pool dump. Password is write-only -> best-effort.

Key spellings (mqtt.broker.N.enabled / .clear, scalar field GET) are inferred
from the existing scheme; they're the one thing the joint firmware test confirms.
12/12 service tests green. UI (brokers screen + editor) next.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

lib/services/observer_config_service.dart

@@ -132,6 +132,73 @@ class ObserverConfigService extends ChangeNotifier {
   Future<bool> setBrokerField(int slot, String field, String value) =>
       setFlat('mqtt.broker.$slot.$field', value);
 
+  /// Save a broker [slot] field-at-a-time per the firmware contract: a live slot
+  /// is disabled first, each changed [fields] entry is written, then `enabled`
+  /// is written LAST (when [enable]). Stops at the first ERR/timeout and reports
+  /// the failed field — a partial save therefore leaves the slot disabled, never
+  /// live-corrupt (#80). The caller re-GETs the slot to show true state + offer
+  /// retry/clear.
+  Future<BrokerSaveResult> saveBroker(
+    int slot, {
+    required Map<String, String> fields,
+    required bool enable,
+    required bool wasLive,
+  }) async {
+    if (wasLive) {
+      if (!await setBrokerField(slot, 'enabled', '0')) {
+        return const BrokerSaveResult.failed('enabled');
+      }
+    }
+    for (final e in fields.entries) {
+      if (!await setBrokerField(slot, e.key, e.value)) {
+        return BrokerSaveResult.failed(e.key);
+      }
+    }
+    if (enable) {
+      if (!await setBrokerField(slot, 'enabled', '1')) {
+        return const BrokerSaveResult.failed('enabled');
+      }
+    }
+    return const BrokerSaveResult.ok();
+  }
+
+  /// Definitively wipe a broker slot — `mqtt.broker.<slot>.clear` (#80).
+  Future<bool> clearBroker(int slot) => setBrokerField(slot, 'clear', '1');
+
+  /// Re-read a single broker [slot] field-by-field — the post-save / recovery
+  /// re-GET. Cheaper than the whole pool: 14 scalar GETs for one slot, not the
+  /// 84-frame OCFG_BROKERS dump (#80). Returns null if any field GET fails.
+  Future<BrokerConfig?> getBroker(int slot) async {
+    const fields = [
+      'enabled',
+      'url',
+      'port',
+      'transport',
+      'auth_type',
+      'username',
+      'password',
+      'topic_prefix',
+      'iata_override',
+      'jwt_audience',
+      'jwt_refresh',
+      'jwt_owner',
+      'jwt_email',
+      'ca_cert',
+    ];
+    final kv = <String, String>{};
+    for (final f in fields) {
+      final v = await getFlat('mqtt.broker.$slot.$f');
+      if (v == null) {
+        // password is write-only — a missing read is its presence, not a slot
+        // failure; any other missing field means the re-GET is incomplete.
+        if (f == 'password') continue;
+        return null;
+      }
+      kv[f] = v;
+    }
+    return BrokerConfig.fromWireFields(slot, kv);
+  }
+
   /// Read the broker pool (paginated START → KV → END). Null on timeout.
   Future<List<BrokerConfig>?> getBrokers() {
     return _serialized(() async {
@@ -225,3 +292,12 @@ class ObserverConfigService extends ChangeNotifier {
     );
   }
 }
+
+/// Outcome of [ObserverConfigService.saveBroker]. On failure [failedField] names
+/// the field whose SET failed (or `enabled`); the slot is left disabled, safe.
+class BrokerSaveResult {
+  const BrokerSaveResult.ok() : ok = true, failedField = null;
+  const BrokerSaveResult.failed(this.failedField) : ok = false;
+  final bool ok;
+  final String? failedField;
+}

test/services/observer_config_service_test.dart

@@ -80,6 +80,14 @@ class _AutoConnector extends MeshCoreConnector {
           ? _resp(ObserverConfigClient.rErr, 'ERROR not available')
           : _resp(ObserverConfigClient.rValue, '$key = ${_valueFor(key)}');
       Future.microtask(() => _frames.add(frame));
+    } else if (op == ObserverConfigClient.opSet) {
+      // SET payload is "key value"; the key is everything before the 1st space.
+      final payload = utf8.decode(data.sublist(2, data.indexOf(0, 2)));
+      final key = payload.split(' ').first;
+      final frame = failKeys.contains(key)
+          ? _resp(ObserverConfigClient.rErr, 'ERROR not available')
+          : _resp(ObserverConfigClient.rAck, '$key = ok');
+      Future.microtask(() => _frames.add(frame));
     } else if (op == ObserverConfigClient.opBrokers) {
       if (failBrokers) return; // no response -> getBrokers times out
       Future.microtask(() {
@@ -272,4 +280,97 @@ void main() {
       auto.closeStream();
     },
   );
+
+  // ---- #80: broker save handshake (enabled-last, per-field ACK, recovery) ----
+  List<String> setKeys(_AutoConnector c) => c.sent
+      .where((f) => f.length > 2 && f[1] == ObserverConfigClient.opSet)
+      .map((f) => utf8.decode(f.sublist(2, f.indexOf(0, 2))).split(' ').first)
+      .toList();
+
+  test(
+    'saveBroker on a live slot disables first, writes fields, enables LAST',
+    () async {
+      final auto = _AutoConnector();
+      final s = ObserverConfigService(auto);
+      final r = await s.saveBroker(
+        2,
+        fields: {'url': 'mqtt://h', 'port': '1883'},
+        enable: true,
+        wasLive: true,
+      );
+      expect(r.ok, isTrue);
+      final keys = setKeys(auto);
+      expect(
+        keys.first,
+        'mqtt.broker.2.enabled',
+        reason: 'a live slot is disabled first',
+      );
+      expect(
+        keys.last,
+        'mqtt.broker.2.enabled',
+        reason: 'enabled is written LAST (activation guard)',
+      );
+      expect(keys.where((k) => k == 'mqtt.broker.2.enabled').length, 2);
+      expect(
+        keys.indexOf('mqtt.broker.2.url'),
+        greaterThan(keys.indexOf('mqtt.broker.2.enabled')),
+      );
+      expect(
+        keys.indexOf('mqtt.broker.2.port'),
+        lessThan(keys.lastIndexOf('mqtt.broker.2.enabled')),
+      );
+      auto.closeStream();
+    },
+  );
+
+  test('saveBroker stops on a field ERR and never reaches the enabled-last '
+      'activation', () async {
+    final auto = _AutoConnector(failKeys: {'mqtt.broker.2.port'});
+    final s = ObserverConfigService(auto);
+    final r = await s.saveBroker(
+      2,
+      fields: {'url': 'mqtt://h', 'port': '1883'},
+      enable: true,
+      wasLive: false,
+    );
+    expect(r.ok, isFalse);
+    expect(r.failedField, 'port');
+    expect(
+      setKeys(auto),
+      isNot(contains('mqtt.broker.2.enabled')),
+      reason: 'a failed field must never reach enabled=1 — slot stays safe',
+    );
+    auto.closeStream();
+  });
+
+  test('clearBroker sends the slot clear op', () async {
+    final auto = _AutoConnector();
+    final s = ObserverConfigService(auto);
+    expect(await s.clearBroker(3), isTrue);
+    expect(setKeys(auto), contains('mqtt.broker.3.clear'));
+    auto.closeStream();
+  });
+
+  test(
+    'getBroker re-reads one slot field-by-field, never the pool dump',
+    () async {
+      final auto = _AutoConnector();
+      final s = ObserverConfigService(auto);
+      final b = await s.getBroker(4);
+      expect(b, isNotNull);
+      expect(b!.slot, 4);
+      final getKeys = auto.sent
+          .where((f) => f.length > 2 && f[1] == ObserverConfigClient.opGet)
+          .map((f) => utf8.decode(f.sublist(2, f.indexOf(0, 2))))
+          .toList();
+      expect(getKeys, contains('mqtt.broker.4.url'));
+      expect(getKeys, contains('mqtt.broker.4.enabled'));
+      expect(
+        auto.sent.any((f) => f[1] == ObserverConfigClient.opBrokers),
+        isFalse,
+        reason: 'single-slot re-GET must not trigger the 84-frame pool dump',
+      );
+      auto.closeStream();
+    },
+  );
 }