fix(runtime): reduce noisy warnings and harden startup

flocks/agent/agent_factory.py

@@ -223,8 +223,9 @@ def scan_and_load(dirs: Optional[List[Path]] = None) -> Dict[str, AgentInfo]:
     """
     Scan agent directories and load all valid agents.
 
-    Scans built-in agents first, then user plugin agents, then project plugin
-    agents.  Name conflicts are skipped with a warning (first wins).
+    Scans built-in agents first, then user plugin agents, then project-bundled
+    agents. The first match wins, so user customizations take precedence over
+    project bundles while core built-ins remain protected.
 
     ``native`` is determined by the source directory, not by agent.yaml:
     - ``_BUILTIN_AGENTS_DIR``           → native=True
@@ -238,34 +239,43 @@ def scan_and_load(dirs: Optional[List[Path]] = None) -> Dict[str, AgentInfo]:
     Returns:
         Dict mapping agent name → AgentInfo.
     """
-    # (directory, is_native) pairs — order determines first-wins conflict resolution
-    search_dirs: List[tuple[Path, bool]] = [(_BUILTIN_AGENTS_DIR, True)]
+    # (directory, is_native, source) tuples — order determines first-wins priority.
+    search_dirs: List[tuple[Path, bool, str]] = [(_BUILTIN_AGENTS_DIR, True, "builtin")]
     if _PLUGIN_AGENTS_DIR.exists():
-        search_dirs.append((_PLUGIN_AGENTS_DIR, False))
+        search_dirs.append((_PLUGIN_AGENTS_DIR, False, "user"))
     # Project-level plugin agents: resolved at call time because cwd is dynamic
     project_agents_dir = Path.cwd() / ".flocks" / "plugins" / "agents"
     if project_agents_dir.exists() and project_agents_dir != _PLUGIN_AGENTS_DIR:
-        search_dirs.append((project_agents_dir, True))
+        search_dirs.append((project_agents_dir, True, "project"))
     if dirs:
-        search_dirs.extend((d, False) for d in dirs)
+        search_dirs.extend((d, False, "extra") for d in dirs)
 
     result: Dict[str, AgentInfo] = {}
+    selected_sources: Dict[str, tuple[str, Path]] = {}
 
-    for scan_dir, is_native in search_dirs:
+    for scan_dir, is_native, source in search_dirs:
         if not scan_dir.is_dir():
             continue
         for agent_dir in _iter_agent_dirs(scan_dir):
             agent = load_agent(agent_dir, native=is_native)
             if agent is None:
                 continue
             if agent.name in result:
-                log.warn("agent.factory.name_conflict", {
+                selected_source, selected_dir = selected_sources[agent.name]
+                details = {
                     "name": agent.name,
-                    "existing_source": "previous scan",
-                    "skipped_source": str(agent_dir),
-                })
+                    "selected_source": selected_source,
+                    "selected_path": str(selected_dir),
+                    "skipped_source": source,
+                    "skipped_path": str(agent_dir),
+                }
+                if source != selected_source and source != "extra":
+                    log.info("agent.factory.lower_priority_skipped", details)
+                else:
+                    log.warn("agent.factory.name_conflict", details)
                 continue
             result[agent.name] = agent
+            selected_sources[agent.name] = (source, agent_dir)
             log.debug("agent.factory.loaded", {
                 "name": agent.name,
                 "dir": str(agent_dir),

flocks/channel/builtin/weixin/channel.py

@@ -237,7 +237,7 @@ async def start(
             "base_url": self._base_url,
         })
         if self._group_policy != "disabled":
-            log.warning("weixin.group_policy.note", {
+            log.info("weixin.group_policy.note", {
                 "group_policy": self._group_policy,
                 "note": (
                     "QR-login connects an iLink bot identity (e.g. ...@im.bot), not a "

flocks/ingest/kafka/manager.py

@@ -297,7 +297,14 @@ async def start_all(self) -> None:
             if not workflow_id:
                 continue
             if isinstance(data, dict) and data.get("enabled"):
-                await self.restart_workflow(workflow_id)
+                try:
+                    await self.restart_workflow(workflow_id, startup=True)
+                except Exception as exc:
+                    self._status[workflow_id] = {"state": "failed", "error": str(exc)}
+                    log.warning(
+                        "kafka.start_failed",
+                        {"workflow_id": workflow_id, "error": str(exc)},
+                    )
 
     async def stop_all(self) -> None:
         for workflow_id in list(self._tasks.keys()):
@@ -360,7 +367,12 @@ async def stop_workflow(self, workflow_id: str) -> None:
         if workflow_id in self._status:
             self._status[workflow_id] = {"state": "stopped", "error": None}
 
-    async def restart_workflow(self, workflow_id: str) -> Dict[str, Any]:
+    async def restart_workflow(
+        self,
+        workflow_id: str,
+        *,
+        startup: bool = False,
+    ) -> Dict[str, Any]:
         """Restart the consumer and return its post-connect runtime status.
 
         Blocks until the consumer connects, the connection fails, or
@@ -377,6 +389,21 @@ async def restart_workflow(self, workflow_id: str) -> Dict[str, Any]:
             self._status[workflow_id] = {"state": "stopped", "error": None}
             return {"state": "stopped", "error": None}
 
+        # Load and cache the workflow JSON once; avoids a disk read per message.
+        wf_data = read_workflow_from_fs(workflow_id)
+        if not wf_data:
+            err = "workflow_not_found"
+            if startup:
+                self._status[workflow_id] = {"state": "stopped", "error": err}
+                log.info("kafka.workflow_not_found_on_start", {
+                    "workflow_id": workflow_id,
+                    "action": "stale_config_skipped",
+                })
+                return {"state": "stopped", "error": err}
+            self._status[workflow_id] = {"state": "failed", "error": err}
+            log.warning("kafka.workflow_not_found", {"workflow_id": workflow_id})
+            return {"state": "failed", "error": err}
+
         input_broker = str(data.get("inputBroker") or "").strip()
         input_topic = str(data.get("inputTopic") or "").strip()
         if not input_broker or not input_topic:
@@ -385,13 +412,6 @@ async def restart_workflow(self, workflow_id: str) -> Dict[str, Any]:
             log.warning("kafka.config_incomplete", {"workflow_id": workflow_id})
             return {"state": "failed", "error": err}
 
-        # Load and cache the workflow JSON once; avoids a disk read per message.
-        wf_data = read_workflow_from_fs(workflow_id)
-        if not wf_data:
-            err = "workflow_not_found"
-            self._status[workflow_id] = {"state": "failed", "error": err}
-            log.warning("kafka.workflow_not_found_on_start", {"workflow_id": workflow_id})
-            return {"state": "failed", "error": err}
         workflow_json = wf_data.get("workflowJson")
         if not workflow_json:
             err = "workflow_json_missing"

flocks/ingest/syslog/manager.py

@@ -186,7 +186,17 @@ async def start_all(self) -> None:
             if not workflow_id:
                 continue
             if isinstance(data, dict) and data.get("enabled"):
-                await self.restart_workflow(workflow_id)
+                try:
+                    await self.restart_workflow(workflow_id, startup=True)
+                except Exception as exc:
+                    self._listener_status[workflow_id] = {
+                        "state": "failed",
+                        "error": str(exc),
+                    }
+                    log.warning(
+                        "syslog.start_failed",
+                        {"workflow_id": workflow_id, "error": str(exc)},
+                    )
 
     async def stop_all(self) -> None:
         for workflow_id in list(self._tasks.keys()):
@@ -245,7 +255,12 @@ async def stop_workflow(self, workflow_id: str) -> None:
         if workflow_id in self._listener_status:
             self._listener_status[workflow_id] = {"state": "stopped", "error": None}
 
-    async def restart_workflow(self, workflow_id: str) -> Dict[str, Any]:
+    async def restart_workflow(
+        self,
+        workflow_id: str,
+        *,
+        startup: bool = False,
+    ) -> Dict[str, Any]:
         """Restart the listener and return its post-bind runtime status.
 
         This call blocks until the underlying socket either binds successfully,
@@ -268,8 +283,15 @@ async def restart_workflow(self, workflow_id: str) -> Dict[str, Any]:
         wf_data = read_workflow_from_fs(workflow_id)
         if not wf_data:
             err = "workflow_not_found"
+            if startup:
+                self._listener_status[workflow_id] = {"state": "stopped", "error": err}
+                log.info("syslog.workflow_not_found_on_start", {
+                    "workflow_id": workflow_id,
+                    "action": "stale_config_skipped",
+                })
+                return {"state": "stopped", "error": err}
             self._listener_status[workflow_id] = {"state": "failed", "error": err}
-            log.warning("syslog.workflow_not_found_on_start", {"workflow_id": workflow_id})
+            log.warning("syslog.workflow_not_found", {"workflow_id": workflow_id})
             return {"state": "failed", "error": err}
         workflow_json = wf_data.get("workflowJson")
         if not workflow_json:
@@ -286,6 +308,10 @@ async def restart_workflow(self, workflow_id: str) -> Dict[str, Any]:
             self._listener_status[workflow_id] = {"state": "failed", "error": err}
             log.warning("syslog.workflow_plan_failed", {"workflow_id": workflow_id, "error": str(exc)})
             return self.get_listener_status(workflow_id)
+
+        host = str(data.get("host") or "0.0.0.0")
+        port = int(data.get("port") or 5140)
+        protocol = str(data.get("protocol") or "udp").lower()
         queue: asyncio.Queue = asyncio.Queue(maxsize=_MAX_QUEUE_SIZE)
         self._queues[workflow_id] = queue
 
@@ -295,9 +321,6 @@ async def restart_workflow(self, workflow_id: str) -> Dict[str, Any]:
         ready = asyncio.Event()
         self._listener_ready[workflow_id] = ready
 
-        host = str(data.get("host") or "0.0.0.0")
-        port = int(data.get("port") or 5140)
-        protocol = str(data.get("protocol") or "udp").lower()
         self._listener_status[workflow_id] = {
             "state": "binding",
             "error": None,

flocks/mcp/client.py

@@ -703,15 +703,7 @@ async def list_resources(self) -> List[McpResource]:
         Raises:
             RuntimeError: If not connected
         """
-        try:
-            result = await self._submit_command("list_resources")
-            return result
-        except Exception as exc:
-            log.error("mcp.client.list_resources_error", {
-                "server": self.name,
-                "error": str(exc),
-            })
-            raise
+        return await self._submit_command("list_resources")
 
     async def read_resource(self, uri: str) -> Any:
         """

flocks/mcp/errors.py

@@ -0,0 +1,35 @@
+"""Helpers for classifying MCP protocol errors."""
+
+from __future__ import annotations
+
+from mcp.types import METHOD_NOT_FOUND
+
+
+def is_method_not_found_error(exc: BaseException) -> bool:
+    """Return whether an exception chain contains JSON-RPC method-not-found."""
+    return _contains_method_not_found(exc, seen=set())
+
+
+def _contains_method_not_found(exc: BaseException, *, seen: set[int]) -> bool:
+    exc_id = id(exc)
+    if exc_id in seen:
+        return False
+    seen.add(exc_id)
+
+    error = getattr(exc, "error", None)
+    if getattr(error, "code", None) == METHOD_NOT_FOUND:
+        return True
+
+    if isinstance(exc, BaseExceptionGroup):
+        if any(_contains_method_not_found(child, seen=seen) for child in exc.exceptions):
+            return True
+
+    cause = exc.__cause__
+    if cause is not None and _contains_method_not_found(cause, seen=seen):
+        return True
+    context = exc.__context__
+    if context is not None and _contains_method_not_found(context, seen=seen):
+        return True
+
+    message = str(exc).strip().lower()
+    return message == "method not found" or message.endswith(": method not found")

flocks/mcp/server.py

@@ -8,6 +8,7 @@
 import time
 from typing import Dict, Optional, Any, List
 from flocks.mcp.client import McpClient
+from flocks.mcp.errors import is_method_not_found_error
 from flocks.mcp.types import (
     McpStatus,
     McpStatusInfo,
@@ -226,10 +227,13 @@ async def _connect_and_register(
                 resources = await client.list_resources()
                 self._resources_cache[name] = resources
             except Exception as e:
-                log.warn("mcp.resources_unavailable", {
-                    "server": name,
-                    "error": str(e)
-                })
+                if is_method_not_found_error(e):
+                    log.info("mcp.resources_unsupported", {"server": name})
+                else:
+                    log.warn("mcp.resources_unavailable", {
+                        "server": name,
+                        "error": str(e)
+                    })
                 self._resources_cache[name] = []
 
             # 5. Register tools

flocks/notifications/service.py

@@ -206,7 +206,6 @@ async def list_active(
         *,
         user_id: str,
         locale: str | None = None,
-        current_version: str | None = None,
     ) -> list[NotificationResponse]:
         target_locale = _normalize_locale(locale)
         config_notifications = await cls._load_config_notifications()

flocks/server/app.py

@@ -1101,7 +1101,8 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
 @app.exception_handler(StarletteHTTPException)
 async def http_exception_handler(request: Request, exc: StarletteHTTPException):
     """Handle HTTP exceptions"""
-    log.error("http.error", {
+    http_log = log.warn if 400 <= exc.status_code < 500 else log.error
+    http_log("http.error", {
         "path": request.url.path,
         "status": exc.status_code,
         "detail": exc.detail,

flocks/server/routes/notifications.py

@@ -26,13 +26,11 @@
 async def list_active_notifications(
     request: Request,
     locale: str | None = None,
-    current_version: str | None = None,
 ) -> list[NotificationResponse]:
     user = require_user(request)
     return await NotificationService.list_active(
         user_id=user.id,
         locale=locale,
-        current_version=current_version,
     )