test: close service boundary gaps

obliteratus/models_client.py

@@ -107,6 +107,8 @@ def models(channel: str | None = None, vendor: str | None = None,
     channel = _normalize_filter(channel, "channel")
     vendor = _normalize_filter(vendor, "vendor")
     capability = _normalize_filter(capability, "capability")
+    if open_weight is not None and not isinstance(open_weight, bool):
+        raise TypeError("open_weight must be a boolean")
     if channel in ("open-weight", "openweight", "weights"):
         open_weight, channel = True, None
     if channel:
@@ -118,11 +120,14 @@ def models(channel: str | None = None, vendor: str | None = None,
         record_vendor = m.get("vendor")
         if record_vendor is not None and not isinstance(record_vendor, str):
             raise ValueError("BESTIARY catalog vendor must be a string")
+        record_open_weight = m.get("open_weight", False)
+        if not isinstance(record_open_weight, bool):
+            raise ValueError("BESTIARY catalog open_weight must be a boolean")
         if channel and channel not in channels:
             continue
         if vendor and (record_vendor or "").strip().casefold() != vendor:
             continue
-        if open_weight is not None and bool(m.get("open_weight")) != open_weight:
+        if open_weight is not None and record_open_weight != open_weight:
             continue
         if capability and capability not in capabilities:
             continue

obliteratus/watchtower.py

@@ -524,6 +524,12 @@ class Watchtower:
                     SchedulerEvent.REQUEST_STOP,
                 )
             self._scheduler_stop.set()
+            if thread is threading.current_thread():
+                logger.info(
+                    "Watchtower: scheduler stop requested from its own thread; "
+                    "confirmation deferred",
+                )
+                return False
             thread.join(timeout=self._scheduler_join_timeout)
             if thread.is_alive():
                 self._scheduler_state = scheduler_transition(

tests/test_models_client_contracts.py

@@ -142,6 +142,8 @@ def test_path_resolution_is_size_bounded(
 def test_filters_reject_non_string_arguments(isolated_sources: None) -> None:
     with pytest.raises(TypeError, match="channel must be a string"):
         models_client.models(channel=7)  # type: ignore[arg-type]
+    with pytest.raises(TypeError, match="open_weight must be a boolean"):
+        models_client.models(open_weight=1)  # type: ignore[arg-type]
 
 
 @pytest.mark.parametrize(
@@ -206,6 +208,7 @@ def test_filters_normalize_aliases_and_catalog_values(
         ({"id": "bad", "channels": "api"}, "channels must be a list"),
         ({"id": "bad", "capabilities": [7]}, "capabilities must be a list"),
         ({"id": "bad", "vendor": 7}, "vendor must be a string"),
+        ({"id": "bad", "open_weight": "false"}, "open_weight must be a boolean"),
     ],
 )
 def test_filters_reject_malformed_catalog_fields(

tests/test_watchtower_contracts.py

@@ -466,6 +466,32 @@ def test_scheduler_runs_immediately_and_stops_without_sleep_races(tmp_path: Path
     assert not watchtower.is_scanning
 
 
+def test_scheduler_callback_can_request_its_own_stop_without_self_join(
+    tmp_path: Path,
+) -> None:
+    callback_finished = threading.Event()
+    stop_results: list[bool] = []
+    watchtower = Watchtower(
+        tmp_path / "watchtower.json",
+        clock=lambda: NOW,
+        fetch_models=lambda: [_model("community/self-stop-instruct-7B")],
+    )
+
+    def request_stop(_model_info: DiscoveredModel) -> None:
+        stop_results.append(watchtower.stop_scheduler())
+        callback_finished.set()
+
+    watchtower.on_new_model(request_stop)
+    watchtower.start_scheduler(interval=3_600)
+
+    assert callback_finished.wait(timeout=1)
+    assert stop_results == [False]
+    assert watchtower._scheduler_state is SchedulerState.STOPPING
+    assert watchtower.stop_scheduler() is True
+    assert watchtower._scheduler_thread is None
+    assert watchtower._scheduler_state is SchedulerState.STOPPED
+
+
 def test_scheduler_restart_and_start_failure_are_transactional(tmp_path: Path) -> None:
     threads = []