feat/tdp alert pull workflow

flocks/agent/registry.py

@@ -579,6 +579,27 @@ def unregister(cls, name: str) -> bool:
 # ---------------------------------------------------------------------------
 
 
+def _agent_event_should_reload(event: object) -> bool:
+    """Return True if a watchdog event should invalidate the agent cache.
+
+    Mirrors ``flocks.tool.registry._tool_event_should_reload``: atomic-save
+    editors surface the real target via ``dest_path``, so we inspect both
+    endpoints before deciding to skip.
+    """
+    candidates = []
+    src = getattr(event, "src_path", "") or ""
+    if src:
+        candidates.append(src)
+    dest = getattr(event, "dest_path", "") or ""
+    if dest:
+        candidates.append(dest)
+    for path in candidates:
+        fname = os.path.basename(path)
+        if fname == "agent.yaml" or path.endswith(".md"):
+            return True
+    return False
+
+
 class AgentFileWatcher:
     """Watch plugin agent directories and auto-invalidate the Agent cache on change.
 
@@ -621,13 +642,20 @@ def start(self) -> None:
 
         watcher = self
 
+        # Only react to actual content-mutation events.  Without this guard the
+        # ``opened``/``closed``/``closed_no_write`` events that watchdog emits
+        # whenever any code (including the agent loader itself) reads
+        # ``agent.yaml`` / ``*.md`` would re-trigger cache invalidation on every
+        # access, causing a self-sustaining reload loop.
+        _RELOAD_EVENT_TYPES = frozenset({"modified", "created", "deleted", "moved"})
+
         class _Handler(FileSystemEventHandler):
             def on_any_event(self, event: FileSystemEvent) -> None:
                 if event.is_directory:
                     return
-                src = getattr(event, "src_path", "") or ""
-                fname = os.path.basename(src)
-                if fname == "agent.yaml" or src.endswith(".md"):
+                if getattr(event, "event_type", "") not in _RELOAD_EVENT_TYPES:
+                    return
+                if _agent_event_should_reload(event):
                     watcher._schedule_invalidate()
 
         handler = _Handler()

flocks/agent/toolset.py

@@ -47,7 +47,11 @@ def normalize_declared_tool_names(
             matches = [raw_name] if raw_name in available else []
 
         if not matches:
-            log.warn("agent.toolset.tool_missing", {"tool": raw_name})
+            # Built-in agent definitions (librarian, metis, …) declare optional
+            # tools such as ``lsp_*`` / ``ast_grep_search`` that ship in separate
+            # binaries; they are gracefully skipped when not installed.  Treat
+            # this as informational only to avoid flooding operational logs.
+            log.debug("agent.toolset.tool_missing", {"tool": raw_name})
             continue
 
         for match in matches:

flocks/command/command_loader.py

@@ -103,7 +103,7 @@ def discover_commands() -> Dict[str, CommandInfo]:
 
     try:
         from flocks.utils.compat import get_flocks_config_dir
-        opencode_global_dir = str(get_flocks_config_dir(binary="opencode") / "command")
+        flocks_global_dir = str(get_flocks_config_dir(binary="opencode") / "command")
         sources.append(("flocks-global", flocks_global_dir, "**/*.md"))
     except Exception as e:
         log.warn("command.flocks_dir.error", {"error": str(e)})

flocks/ingest/syslog/__init__.py

@@ -0,0 +1,6 @@
+"""Syslog ingestion for workflow triggers (UDP/TCP listeners)."""
+
+from flocks.ingest.syslog.constants import WORKFLOW_SYSLOG_CONFIG_PREFIX
+from flocks.ingest.syslog.manager import SyslogManager, default_manager
+
+__all__ = ["SyslogManager", "default_manager", "WORKFLOW_SYSLOG_CONFIG_PREFIX"]

flocks/ingest/syslog/constants.py

@@ -0,0 +1,3 @@
+"""Storage key prefix for per-workflow syslog config (must match server routes)."""
+
+WORKFLOW_SYSLOG_CONFIG_PREFIX = "workflow_syslog_config/"

flocks/ingest/syslog/listener.py

@@ -0,0 +1,128 @@
+"""Asyncio UDP/TCP syslog listeners."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Awaitable, Callable, Union
+
+from flocks.ingest.syslog.parser import parse_syslog
+
+OnSyslogMessage = Callable[[dict], Union[None, Awaitable[None]]]
+
+
+class SyslogUDPProtocol(asyncio.DatagramProtocol):
+    """Receive syslog datagrams and invoke async callback with parsed dict.
+
+    The *on_message* callback is expected to be non-blocking (e.g. a queue
+    put_nowait).  This protocol deliberately does NOT create unbounded asyncio
+    tasks on every datagram — the caller owns concurrency control.
+    """
+
+    def __init__(
+        self,
+        on_message: OnSyslogMessage,
+        format_hint: str,
+    ) -> None:
+        self._on_message = on_message
+        self._format_hint = format_hint
+
+    def datagram_received(self, data: bytes, _addr) -> None:  # noqa: ANN001
+        text = data.decode("utf-8", errors="replace")
+        parsed = parse_syslog(text, self._format_hint)
+        try:
+            res = self._on_message(parsed)
+            # If the callback returns a coroutine (legacy path), schedule it
+            # but only once — do NOT create_task without bound.
+            if asyncio.iscoroutine(res):
+                try:
+                    loop = asyncio.get_running_loop()
+                except RuntimeError:
+                    return
+                loop.create_task(self._safe_await(res))
+        except Exception:
+            pass
+
+    @staticmethod
+    async def _safe_await(coro) -> None:  # noqa: ANN001
+        try:
+            await coro
+        except Exception:
+            pass
+
+
+async def run_udp_syslog_server(
+    host: str,
+    port: int,
+    format_hint: str,
+    on_message: OnSyslogMessage,
+    *,
+    abort_event: asyncio.Event,
+) -> None:
+    loop = asyncio.get_running_loop()
+    transport, _protocol = await loop.create_datagram_endpoint(
+        lambda: SyslogUDPProtocol(on_message, format_hint),
+        local_addr=(host, port),
+    )
+    try:
+        await abort_event.wait()
+    finally:
+        transport.close()
+
+
+async def _handle_tcp_client(
+    reader: asyncio.StreamReader,
+    writer: asyncio.StreamWriter,
+    format_hint: str,
+    on_message: OnSyslogMessage,
+) -> None:
+    try:
+        while True:
+            line = await reader.readline()
+            if not line:
+                break
+            text = line.decode("utf-8", errors="replace").strip()
+            if not text:
+                continue
+            parsed = parse_syslog(text, format_hint)
+            try:
+                res = on_message(parsed)
+                if asyncio.iscoroutine(res):
+                    await res
+            except Exception:
+                pass
+    finally:
+        try:
+            writer.close()
+            await writer.wait_closed()
+        except Exception:
+            pass
+
+
+async def run_tcp_syslog_server(
+    host: str,
+    port: int,
+    format_hint: str,
+    on_message: OnSyslogMessage,
+    *,
+    abort_event: asyncio.Event,
+) -> None:
+    async def handle_client(
+        reader: asyncio.StreamReader,
+        writer: asyncio.StreamWriter,
+    ) -> None:
+        await _handle_tcp_client(reader, writer, format_hint, on_message)
+
+    server = await asyncio.start_server(handle_client, host, port)
+    serve_task: asyncio.Task[None] | None = None
+    try:
+        serve_task = asyncio.create_task(server.serve_forever())
+        await abort_event.wait()
+    finally:
+        if serve_task and not serve_task.done():
+            serve_task.cancel()
+            try:
+                await serve_task
+            except asyncio.CancelledError:
+                pass
+        server.close()
+        await server.wait_closed()