From 4dd0c22bd71ef6ff602c26c377eee145c98d7ca5 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:36:31 -0700 Subject: [PATCH] refactor(gateway): compact kanban notifier collector and dispatcher log call sites --- gateway/kanban_watchers_dispatcher.py | 44 +++++------------ gateway/kanban_watchers_notifier.py | 70 ++++++++++----------------- 2 files changed, 38 insertions(+), 76 deletions(-) diff --git a/gateway/kanban_watchers_dispatcher.py b/gateway/kanban_watchers_dispatcher.py index 9dda4cf3c0..28688b6068 100644 --- a/gateway/kanban_watchers_dispatcher.py +++ b/gateway/kanban_watchers_dispatcher.py @@ -38,10 +38,8 @@ def _resolve_dispatcher_settings(kanban_cfg: dict, kb: Any) -> _DispatcherSettin try: interval = float(kanban_cfg.get("dispatch_interval_seconds", 60) or 60) except (ValueError, TypeError): - logger.warning( - "kanban dispatcher: invalid dispatch_interval_seconds=%r, using default 60", - kanban_cfg.get("dispatch_interval_seconds"), - ) + logger.warning("kanban dispatcher: invalid dispatch_interval_seconds=%r, using default 60", + kanban_cfg.get("dispatch_interval_seconds")) interval = 60.0 interval = max(interval, 1.0) # sanity floor ā€” tighter than this is a footgun @@ -66,16 +64,12 @@ def _resolve_dispatcher_settings(kanban_cfg: dict, kb: Any) -> _DispatcherSettin try: failure_limit = int(raw_failure_limit) except (TypeError, ValueError): - logger.warning( - "kanban dispatcher: invalid kanban.failure_limit=%r; using default %d", - raw_failure_limit, kb.DEFAULT_FAILURE_LIMIT, - ) + logger.warning("kanban dispatcher: invalid kanban.failure_limit=%r; using default %d", + raw_failure_limit, kb.DEFAULT_FAILURE_LIMIT) failure_limit = kb.DEFAULT_FAILURE_LIMIT if failure_limit < 1: - logger.warning( - "kanban dispatcher: kanban.failure_limit=%r is below 1; using default %d", - raw_failure_limit, kb.DEFAULT_FAILURE_LIMIT, - ) + logger.warning("kanban dispatcher: kanban.failure_limit=%r is below 1; using default %d", + raw_failure_limit, kb.DEFAULT_FAILURE_LIMIT) failure_limit = kb.DEFAULT_FAILURE_LIMIT # 0 disables stale detection. @@ -83,22 +77,16 @@ def _resolve_dispatcher_settings(kanban_cfg: dict, kb: Any) -> _DispatcherSettin try: stale_timeout_seconds = int(raw_stale or 0) except (TypeError, ValueError): - logger.warning( - "kanban dispatcher: invalid kanban.dispatch_stale_timeout_seconds=%r; " - "disabling stale detection", - raw_stale, - ) + logger.warning("kanban dispatcher: invalid kanban.dispatch_stale_timeout_seconds=%r; " + "disabling stale detection", raw_stale) stale_timeout_seconds = 0 # Fallback profile for tasks created without an assignee (e.g. via the # dashboard). Empty (the schema default) keeps skipping them. default_assignee = (kanban_cfg.get("default_assignee") or "").strip() or None if default_assignee: - logger.info( - "kanban dispatcher: default_assignee=%r (unassigned ready tasks " - "will route to this profile)", - default_assignee, - ) + logger.info("kanban dispatcher: default_assignee=%r (unassigned ready tasks " + "will route to this profile)", default_assignee) return _DispatcherSettings( interval=interval, @@ -151,10 +139,7 @@ class _KanbanDispatcher: corrupt_guard_error = getattr(self.kb, "KanbanDbCorruptError", None) if corrupt_guard_error is not None and isinstance(exc, corrupt_guard_error): return True - if not isinstance(exc, sqlite3.DatabaseError): - return False - msg = str(exc).lower() - return any(marker in msg for marker in _CORRUPT_DB_MARKERS) + return isinstance(exc, sqlite3.DatabaseError) and any(m in str(exc).lower() for m in _CORRUPT_DB_MARKERS) def _quarantine_lifted(self, slug: str, fingerprint: tuple) -> bool: """Return False while *slug* stays quarantined; lift (and log) otherwise.""" @@ -166,11 +151,8 @@ class _KanbanDispatcher: if disabled_fingerprint == fingerprint and age < self.CORRUPT_BOARD_RETRY_AFTER_SECONDS: return False if disabled_fingerprint == fingerprint: - logger.info( - "kanban dispatcher: board %s database fingerprint unchanged " - "after %.0fs quarantine; retrying dispatch", - slug, age, - ) + logger.info("kanban dispatcher: board %s database fingerprint unchanged " + "after %.0fs quarantine; retrying dispatch", slug, age) else: logger.info("kanban dispatcher: board %s database changed; retrying dispatch", slug) self.disabled_corrupt_boards.pop(slug, None) diff --git a/gateway/kanban_watchers_notifier.py b/gateway/kanban_watchers_notifier.py index 2ae4f2f7fd..128b7ec037 100644 --- a/gateway/kanban_watchers_notifier.py +++ b/gateway/kanban_watchers_notifier.py @@ -89,14 +89,13 @@ class _Collector: self.profile_adapters = getattr(runner, "_profile_adapters", {}) self.notifier_profiles = {notifier_profile} self.notifier_profiles.update(str(p).strip() for p in self.profile_adapters if str(p).strip()) - self.active_platforms = _platform_names(runner.adapters) # Include every platform any secondary profile has live. This is only a # coarse pre-filter; the precise per-profile check (_authorization_adapter, # no default fallback) runs at delivery and rewinds the claim if it # resolves to None. An unclaimed event never retries, so dropping a # secondary-profile sub here would lose it. - for _profile_adapter_map in self.profile_adapters.values(): - self.active_platforms.update(_platform_names(_profile_adapter_map)) + self.active_platforms = _platform_names(runner.adapters).union( + *(_platform_names(m) for m in self.profile_adapters.values())) def collect(self) -> list[dict]: if not self.active_platforms: @@ -124,31 +123,24 @@ class _Collector: """Cheap read-only probe before the writable connect() (schema init, WAL sidecars, checkpoints); a probe failure falls back to the writable open.""" try: - if self.kb.count_notify_subs( - board=slug, notifier_profiles=self.notifier_profiles, include_unowned=self.include_unowned, - ) == 0: - logger.debug( - "kanban notifier: board %s has no subscriptions owned by %s; skipping open", - slug, sorted(self.notifier_profiles), - ) - return False + count = self.kb.count_notify_subs( + board=slug, notifier_profiles=self.notifier_profiles, include_unowned=self.include_unowned) except Exception as exc: - logger.debug( - "kanban notifier: read-only subscription probe failed " - "for board %s (%s); falling back to writable open", - slug, exc, - ) - return True + logger.debug("kanban notifier: read-only subscription probe failed " + "for board %s (%s); falling back to writable open", slug, exc) + return True + if count == 0: + logger.debug("kanban notifier: board %s has no subscriptions owned by %s; skipping open", + slug, sorted(self.notifier_profiles)) + return count != 0 def _gc_stale_subs(self, conn: Any, slug: str) -> None: """Best-effort stale-sub sweep: a failed sweep never blocks delivery; the next hourly gate retries.""" try: _purged = self.kb.purge_stale_done_notify_subs(conn, max_age_days=self.gc_retention_days) if _purged: - logger.info( - "kanban notifier: purged %d stale done/blocked-task subscription(s) on board %s (retention %dd)", - _purged, slug, self.gc_retention_days, - ) + logger.info("kanban notifier: purged %d stale done/blocked-task subscription(s) on board %s (retention %dd)", + _purged, slug, self.gc_retention_days) except Exception as _gc_exc: logger.debug("kanban notifier: stale-sub GC failed for board %s: %s", slug, _gc_exc) @@ -156,17 +148,13 @@ class _Collector: """Claim one subscription's unseen events; None when skipped or nothing new.""" owner_profile = sub.get("notifier_profile") or None if owner_profile and owner_profile != self.notifier_profile and not self.profile_adapters.get(owner_profile): - logger.debug( - "kanban notifier: subscription for %s owned by profile %s; current profile %s has no adapter for it, skipping", - sub.get("task_id"), owner_profile, self.notifier_profile, - ) + logger.debug("kanban notifier: subscription for %s owned by profile %s; current profile %s has no adapter for it, skipping", + sub.get("task_id"), owner_profile, self.notifier_profile) return None platform = (sub.get("platform") or "").lower() if platform not in self.active_platforms: - logger.debug( - "kanban notifier: subscription for %s on %s skipped; adapter not connected", - sub.get("task_id"), platform or "", - ) + logger.debug("kanban notifier: subscription for %s on %s skipped; adapter not connected", + sub.get("task_id"), platform or "") return None old_cursor, cursor, events = self.kb.claim_unseen_events_for_sub( conn, task_id=sub["task_id"], platform=sub["platform"], chat_id=sub["chat_id"], @@ -175,10 +163,8 @@ class _Collector: if not events: return None task = self.kb.get_task(conn, sub["task_id"]) - logger.debug( - "kanban notifier: claimed %d event(s) for %s on board %s cursor %sā†’%s", - len(events), sub["task_id"], slug, old_cursor, cursor, - ) + logger.debug("kanban notifier: claimed %d event(s) for %s on board %s cursor %sā†’%s", + len(events), sub["task_id"], slug, old_cursor, cursor) return {"sub": sub, "old_cursor": old_cursor, "cursor": cursor, "events": events, "task": task, "board": slug} def collect_board(self, slug: str) -> None: @@ -207,10 +193,8 @@ class _Collector: self.deliveries.append(claimed) except Exception as sub_exc: # One bad subscription must not block the rest of the tick. - logger.warning( - "kanban notifier: subscription for %s on board %s failed: %s", - sub.get("task_id"), slug, sub_exc, - ) + logger.warning("kanban notifier: subscription for %s on board %s failed: %s", + sub.get("task_id"), slug, sub_exc) finally: conn.close() @@ -433,10 +417,8 @@ class _KanbanNotification: self.synth = synth + "\n\n" + t("gateway.kanban.wake.guidance") def _log_woke(self) -> None: - logger.info( - "kanban notifier: woke agent for %s on %s/%s profile=%s events=%s", - self.task_id, self.platform_str, self.sub["chat_id"], self.sub_profile or "default", self.wake_kinds, - ) + logger.info("kanban notifier: woke agent for %s on %s/%s profile=%s events=%s", + self.task_id, self.platform_str, self.sub["chat_id"], self.sub_profile or "default", self.wake_kinds) async def wake(self) -> None: """Wake the creator session (raises on failure): push adapters get a full SessionSource, non-push a raw self-post.""" @@ -478,10 +460,8 @@ class _KanbanNotification: # "no exception == delivered" contract. if getattr(_send_res, "success", True) is False: raise RuntimeError(f"adapter send() reported failure: {getattr(_send_res, 'error', None) or 'unknown error'}") - logger.debug( - "kanban notifier: delivered %s event for %s to %s/%s on board %s", - ev.kind, self.task_id, self.platform_str, sub["chat_id"], self.board_slug, - ) + logger.debug("kanban notifier: delivered %s event for %s to %s/%s on board %s", + ev.kind, self.task_id, self.platform_str, sub["chat_id"], self.board_slug) # Upload artifact paths from the completion payload / legacy result as # native files. Only on ``completed`` so retries never spam attachments. if ev.kind == "completed":