fix(watcher): graceful type-flip degradation + regression tests

_apply_raw now catches TypeMismatchError, logs a warning, and falls back
to the field's default value (matching Go SDK behavior). Previously a
type mismatch from the server would propagate uncaught and crash the
background subscribe thread/task.

Adds regression tests for both sync (WatchedField, ConfigWatcher) and
async (AsyncWatchedField, AsyncConfigWatcher) paths: _load_initial
type-flip falls back to default; _update type-flip falls back to
default; _process_change type-flip continues the stream.

Refs #563

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: zeevdr

sdk/src/opendecree/_watcher_base.py

@@ -7,6 +7,7 @@
 from typing import Generic, TypeVar
 
 from opendecree._convert import convert_value
+from opendecree.errors import TypeMismatchError
 
 T = TypeVar("T")
 
@@ -57,8 +58,18 @@ def _apply_raw(self, raw_value: str | None) -> tuple[T, T]:
         """Set _value/_is_set from a raw string. Returns (old, new). Caller must lock if needed."""
         old = self._value
         if raw_value is not None:
-            self._value = convert_value(raw_value, self._type)  # type: ignore[assignment]
-            self._is_set = True
+            try:
+                self._value = convert_value(raw_value, self._type)  # type: ignore[assignment]
+                self._is_set = True
+            except TypeMismatchError:
+                _logger.warning(
+                    "Type mismatch for field %r: cannot convert %r to %s, using default",
+                    self._path,
+                    raw_value,
+                    self._type.__name__,
+                )
+                self._value = self._default
+                self._is_set = False
         else:
             self._value = self._default
             self._is_set = False

sdk/tests/test_async_watcher.py

@@ -91,6 +91,26 @@ def test_repr(self):
         f = AsyncWatchedField("payments.fee", float, 0.01)
         assert "payments.fee" in repr(f)
 
+    def test_load_initial_type_flip_falls_back_to_default(self):
+        f = AsyncWatchedField("payments.fee", float, 0.01)
+        f._load_initial("not-a-number")
+        assert f.value == pytest.approx(0.01)
+        assert not f._is_set
+
+    def test_update_type_flip_falls_back_to_default(self):
+        f = AsyncWatchedField("payments.fee", float, 0.01)
+        f._load_initial("0.025")
+        assert f.value == pytest.approx(0.025)
+
+        change = Change(
+            field_path="payments.fee", old_value="0.025", new_value="not-a-number", version=2
+        )
+        f._update("not-a-number", change)
+
+        # Falls back to default on type mismatch, does not crash.
+        assert f.value == pytest.approx(0.01)
+        assert not f._is_set
+
     def test_callback_exception_is_logged(self):
         f = AsyncWatchedField("x", int, 0)
         f._load_initial("1")
@@ -348,6 +368,36 @@ def test_process_change(self):
         w._process_change(change)
         assert fee.value == 2.0
 
+    def test_type_flip_mid_stream_uses_default_and_continues(self):
+        """Type-flip mid-stream falls back to default and keeps the stream alive."""
+        from opendecree._generated.centralconfig.v1 import types_pb2
+
+        w = self._make_watcher()
+        fee = w.field("fee", float, default=0.01)
+        fee._load_initial("0.025")
+
+        bad_change = MagicMock()
+        bad_change.field_path = "fee"
+        bad_change.HasField.side_effect = lambda name: name in ("old_value", "new_value")
+        bad_change.old_value = types_pb2.TypedValue(string_value="0.025")
+        bad_change.new_value = types_pb2.TypedValue(string_value="not-a-number")
+        bad_change.version = 2
+        bad_change.changed_by = ""
+
+        w._process_change(bad_change)
+        assert fee.value == pytest.approx(0.01)
+
+        good_change = MagicMock()
+        good_change.field_path = "fee"
+        good_change.HasField.side_effect = lambda name: name in ("old_value", "new_value")
+        good_change.old_value = types_pb2.TypedValue(string_value="not-a-number")
+        good_change.new_value = types_pb2.TypedValue(string_value="0.1")
+        good_change.version = 3
+        good_change.changed_by = ""
+
+        w._process_change(good_change)
+        assert fee.value == pytest.approx(0.1)
+
     def test_process_change_unknown_field_ignored(self):
         w = self._make_watcher()
         w.field("known", str, default="")

sdk/tests/test_watcher.py

@@ -107,6 +107,28 @@ def test_repr(self):
         assert "payments.fee" in repr(f)
         assert "0.01" in repr(f)
 
+    def test_load_initial_type_flip_falls_back_to_default(self):
+        f = WatchedField("payments.fee", float, 0.01)
+        f._load_initial("not-a-number")
+        assert f.value == pytest.approx(0.01)
+        assert not f._is_set
+
+    def test_update_type_flip_falls_back_to_default(self):
+        f = WatchedField("payments.fee", float, 0.01)
+        f._load_initial("0.025")
+        assert f.value == pytest.approx(0.025)
+
+        from opendecree.types import Change
+
+        change = Change(
+            field_path="payments.fee", old_value="0.025", new_value="not-a-number", version=2
+        )
+        f._update("not-a-number", change)
+
+        # Falls back to default on type mismatch, does not crash.
+        assert f.value == pytest.approx(0.01)
+        assert not f._is_set
+
     def test_callback_exception_is_logged(self):
         f = WatchedField("x", int, 0)
         f._load_initial("1")
@@ -373,6 +395,38 @@ def test_process_change(self):
         w._process_change(change)
         assert fee.value == 2.0
 
+    def test_type_flip_mid_stream_uses_default_and_continues(self):
+        """Type-flip mid-stream falls back to default and keeps the stream alive."""
+        from opendecree._generated.centralconfig.v1 import types_pb2
+
+        w = self._make_watcher()
+        fee = w.field("fee", float, default=0.01)
+        fee._load_initial("0.025")
+
+        # Type-flip: string value for a float field.
+        bad_change = MagicMock()
+        bad_change.field_path = "fee"
+        bad_change.HasField.side_effect = lambda name: name in ("old_value", "new_value")
+        bad_change.old_value = types_pb2.TypedValue(string_value="0.025")
+        bad_change.new_value = types_pb2.TypedValue(string_value="not-a-number")
+        bad_change.version = 2
+        bad_change.changed_by = ""
+
+        w._process_change(bad_change)
+        assert fee.value == pytest.approx(0.01)
+
+        # Subsequent valid change must still apply.
+        good_change = MagicMock()
+        good_change.field_path = "fee"
+        good_change.HasField.side_effect = lambda name: name in ("old_value", "new_value")
+        good_change.old_value = types_pb2.TypedValue(string_value="not-a-number")
+        good_change.new_value = types_pb2.TypedValue(string_value="0.1")
+        good_change.version = 3
+        good_change.changed_by = ""
+
+        w._process_change(good_change)
+        assert fee.value == pytest.approx(0.1)
+
     def test_process_change_unknown_field_ignored(self):
         w = self._make_watcher()
         w.field("known", str, default="")