Remove libby argument in LibbyDaemon

libby/daemon.py

@@ -250,14 +250,18 @@ def _ensure_last_error_handler(self) -> None:
 
     ### Optional hooks
 
-    def on_start(self, libby: Libby) -> None:
-        """Initialize hardware and register keywords."""
+    def on_start(self) -> None:
+        """Initialize hardware and register keywords. ``self.libby`` is set."""
 
-    def on_stop(self, libby: Optional[Libby] = None) -> None:
-        """Release hardware resources."""
+    def on_stop(self) -> None:
+        """Release hardware resources.
 
-    def on_hello(self, libby: Libby) -> None:
-        """Run after the initial discovery hello."""
+        Also runs after a start that failed partway, so ``self.libby`` may be
+        None; it is closed only after this returns.
+        """
+
+    def on_hello(self) -> None:
+        """Run after the initial discovery hello. ``self.libby`` is set."""
 
     def on_event(self, topic: str, msg: Any) -> None:
         self.logger.info("%s: %s", topic, msg.env.payload)
@@ -418,12 +422,12 @@ def start(self) -> None:
             if self.config_discovery_enabled():
                 try:
                     self.libby.hello()
-                    self.on_hello(self.libby)
+                    self.on_hello()
                 except Exception:
                     self.logger.exception("discovery hello failed")
 
             try:
-                self.on_start(self.libby)
+                self.on_start()
             except Exception:
                 self.logger.exception("daemon initialization failed")
                 if self.fail_fast_on_start:
@@ -454,7 +458,7 @@ def stop(self) -> None:
             return
 
         try:
-            self.on_stop(self.libby)
+            self.on_stop()
         except Exception:
             self.logger.exception("daemon shutdown hook failed")
         finally:

libby/how_to_create_a_daemon.md

@@ -74,9 +74,9 @@ class MyPeer(LibbyDaemon):
         "alerts.status": lambda payload: print("[X] status:", payload),
     }
 
-    def on_start(self, libby):
+    def on_start(self):
         # Manually teach peers about each other (discovery workaround)
-        libby.learn_peer_keys("peer-Y", ["service1", "service2"])
+        self.libby.learn_peer_keys("peer-Y", ["service1", "service2"])
 
 if __name__ == "__main__":
     MyPeer().serve()
@@ -112,9 +112,9 @@ class MyPeer(LibbyDaemon):
         "alerts.status": lambda payload: print("[X] status:", payload),
     }
 
-    def on_start(self, libby):
+    def on_start(self):
         # Manually teach peers about each other (discovery workaround)
-        libby.learn_peer_keys("peer-Y", ["service1", "service2"])
+        self.libby.learn_peer_keys("peer-Y", ["service1", "service2"])
 
 if __name__ == "__main__":
     MyPeer().serve()
@@ -196,18 +196,18 @@ class PeerA(LibbyDaemon):
     discovery_enabled = True
     discovery_interval_s = 2.0
 
-    def on_start(self, libby):
+    def on_start(self):
         try:
-            if not libby.wait_for_key("peer-B", "perf.echo", timeout_s=2.5):
-                libby.learn_peer_keys("peer-B", ["perf.echo", "ping.txt", "answer"])
+            if not self.libby.wait_for_key("peer-B", "perf.echo", timeout_s=2.5):
+                self.libby.learn_peer_keys("peer-B", ["perf.echo", "ping.txt", "answer"])
         except AttributeError:
             pass
 
         print("[PeerA] asking B: perf.echo …")
-        res = libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=8000)
+        res = self.libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=8000)
         print("[PeerA] result:", res)
 
-        libby.publish("alerts.status", {"source": "peer-A", "ok": True})
+        self.libby.publish("alerts.status", {"source": "peer-A", "ok": True})
         print("[PeerA] published alerts.status")
 
 if __name__ == "__main__":
@@ -253,18 +253,18 @@ class PeerC(LibbyDaemon):
         "math.add":     math_add,
     }
 
-    def on_start(self, libby):
+    def on_start(self):
         # Proxy service needing live libby handle
         def echo_proxy(_p: Dict[str, Any]):
-            res = libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=6000)
+            res = self.libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=6000)
             return {"ok": True, "forwarded_to": "peer-B", "result": res}
 
         self.add_service("perf.echo.proxy", echo_proxy)
 
         print("[PeerC] math.add(2,5) ->",
-              libby.rpc(self.peer_id, "math.add", {"a": 2, "b": 5}, ttl_ms=2000))
+              self.libby.rpc(self.peer_id, "math.add", {"a": 2, "b": 5}, ttl_ms=2000))
 
-        libby.publish("alerts.status", {"source": "peer-C", "ok": True})
+        self.libby.publish("alerts.status", {"source": "peer-C", "ok": True})
 
 if __name__ == "__main__":
     PeerC().serve()
@@ -312,5 +312,5 @@ class MyPeer(LibbyDaemon):
 - **Encapsulation goal**: application peers only implement business logic (`on_req`, `on_event`) and a few config methods.
 - **Handlers are payload-only.** They receive a Python dict and return anything JSON-serializable.
   If the return is not a dict, LibbyDaemon auto-wraps it as `{"data": <value>}`.
-- **Discovery workaround**: Currently, automatic peer discovery needs manual key learning via `libby.learn_peer_keys()` in `on_start()`.
+- **Discovery workaround**: Currently, automatic peer discovery needs manual key learning via `self.libby.learn_peer_keys()` in `on_start()`.
 

libby/keygrabber/daemon.py

@@ -13,7 +13,6 @@
 from ..config import ConfigError, DaemonConfigLoader, with_env_overrides
 from ..daemon import LibbyDaemon
 from ..errors import LibbyError
-from ..libby import Libby
 from .collection import FAILURES_BEFORE_BACKOFF, Collection
 from .config import TIMEOUT_HEADROOM, KeygrabberConfig, build_sink, parse_config
 from .scheduler import Scheduler
@@ -112,10 +111,10 @@ def __init__(self) -> None:
         self._reconnect_wanted = False
         self._clock: Callable[[], float] = time.monotonic
 
-    def on_start(self, libby: Libby) -> None:
+    def on_start(self) -> None:
         """Parse config, open the sink, and start the reader and writer threads."""
         self._settings = parse_config(self._config)
-        self._client = Client(libby)
+        self._client = Client(self.libby)
         self._writer = RetryingWriter(self.make_sink(),
                                       policy=self._settings.retry)
         self._writer.connect()
@@ -133,7 +132,7 @@ def on_start(self, libby: Libby) -> None:
         self.logger.info("collecting %d collections with %d workers",
                          len(self._collections), self._settings.workers)
 
-    def on_stop(self, libby: Optional[Libby] = None) -> None:
+    def on_stop(self) -> None:
         """Stop the threads and close the sink once the writer has drained it.
 
         The draining happens on the writer thread rather than here, so only one

peers/peer_a.py

@@ -17,20 +17,20 @@ class PeerA(LibbyDaemon):
     discovery_enabled = True
     discovery_interval_s = 2.0
 
-    def on_start(self, libby):
+    def on_start(self):
         try:
-            if not libby.wait_for_key(
+            if not self.libby.wait_for_key(
                 "peer-B",
                 "perf.echo",
                 timeout_s=2.5,
             ):
-                libby.learn_peer_keys(
+                self.libby.learn_peer_keys(
                     "peer-B",
                     ["perf.echo", "ping.txt", "answer"],
                 )
 
             print("[PeerA] asking B: perf.echo ...")
-            result = libby.rpc(
+            result = self.libby.rpc(
                 "peer-B",
                 "perf.echo",
                 {"t0": time.time()},
@@ -41,7 +41,7 @@ def on_start(self, libby):
         except Exception as exc:
             print(f"[PeerA] Peer B request failed: {exc}")
 
-        libby.publish(
+        self.libby.publish(
             "alerts.status",
             {
                 "source": self.peer_id,

peers/peer_a_rabbitmq.py

@@ -6,16 +6,16 @@ class PeerA(LibbyDaemon):
     transport = "rabbitmq"
     rabbitmq_url = "amqp://localhost"
 
-    def on_start(self, libby):
+    def on_start(self):
         # Give peer B a moment to register its services
         time.sleep(1.5)
 
         print("[PeerA] asking B: perf.echo …")
-        res = libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=8000)
+        res = self.libby.rpc("peer-B", "perf.echo", {"t0": time.time()}, ttl_ms=8000)
         print("[PeerA] result:", res)
 
         # publish a status
-        libby.publish("alerts.status", {"source": "peer-A", "ok": True})
+        self.libby.publish("alerts.status", {"source": "peer-A", "ok": True})
         print("[PeerA] published alerts.status")
 
 if __name__ == "__main__":

peers/peer_b.py

@@ -33,7 +33,7 @@ class PeerB(LibbyDaemon):
         },
     }
 
-    def on_start(self, libby):
+    def on_start(self):
         print("[PeerB] ready")
         print("[PeerB] services:", list(self.services))
 

peers/peer_c.py

@@ -32,7 +32,7 @@ def get_last_status(self, payload):
             "status": self.last_status,
         }
 
-    def on_start(self, libby):
+    def on_start(self):
         print("[PeerC] listening for alerts.status")
 
 

peers/peer_d.py

@@ -12,7 +12,7 @@ def ping(self, payload: Payload) -> Dict[str, Any]:
     def echo(self, payload: Payload) -> Dict[str, Any]:
         return self.payload(payload)
 
-    def on_start(self, libby):
+    def on_start(self):
         import time
         target = "peer-B"
         key = "perf.echo"
@@ -21,17 +21,17 @@ def on_start(self, libby):
         print(f"[{self.config_peer_id()}] routes: {ab}")
 
         # 1) Best-effort
-        libby.wait_for_peer(target, timeout_s=2.0)
+        self.libby.wait_for_peer(target, timeout_s=2.0)
 
         # 2) If the key isn't known yet (discovery not populated)
-        if not libby.wait_for_key(target, key, timeout_s=1.0):
+        if not self.libby.wait_for_key(target, key, timeout_s=1.0):
             try:
-                libby.learn_peer_keys(target, [key])
+                self.libby.learn_peer_keys(target, [key])
             except Exception:
                 pass
 
         print(f"[{self.config_peer_id()}] asking {target}: {key} …")
-        res = libby.rpc(target, key, {"t0": time.time()}, ttl_ms=8000)
+        res = self.libby.rpc(target, key, {"t0": time.time()}, ttl_ms=8000)
         print(f"[{self.config_peer_id()}] result: {res}")
 
 

peers/peer_group.py

@@ -20,7 +20,7 @@ class GroupTestPeer(LibbyDaemon):
         "echo": handle_echo,
     }
 
-    def on_start(self, libby):
+    def on_start(self):
         print(f"[GroupTestPeer] Started with group_id='{self.group_id}'")
         print(f"[GroupTestPeer] Expected queue name: libby.group.{self.group_id}.peer.{self.peer_id}")
 

tests/test_client_integration.py

@@ -52,7 +52,7 @@ class _FixtureDaemon(LibbyDaemon):
     group_id = GROUP_ID
     discovery_enabled = False
 
-    def on_start(self, libby) -> None:
+    def on_start(self) -> None:
         state = {"pos": INITIAL_POSITION}
         self.keyword_registry.float(
             "positionvalue",
@@ -72,7 +72,7 @@ class _LifecycleDaemon(LibbyDaemon):
     group_id = GROUP_ID
     discovery_enabled = False
 
-    def on_start(self, libby) -> None:
+    def on_start(self) -> None:
         self.keyword_registry.trigger(
             "shutdown",
             action=self.request_stop,

tests/test_keygrabber_daemon.py

@@ -47,7 +47,7 @@ class _FixtureDaemon(LibbyDaemon):
     group_id = "hsfei"
     discovery_enabled = False
 
-    def on_start(self, libby) -> None:
+    def on_start(self) -> None:
         self.keyword_registry.float("positionvalue", getter=lambda: 7.5,
                                     units="mm")
         self.keyword_registry.bool("isconnected", getter=lambda: True)