Feat/stream loop

README.md

@@ -104,6 +104,77 @@ Inspect the agent's layered orientation and pending re-entries anytime with the
 
 ---
 
+## Real-Time Signal Processing
+
+LeapFlow continuously observes the operating environment through a unified signal pipeline — signals are never dropped, noise never reaches the agent, and resources are never wasted on redundant data.
+
+### End-to-End Pipeline
+
+```
+Signal Sources ─────────────────────────────────────────────────────────────
+  app_focus │ clipboard │ fs_watcher │ input_tap │ gateway │ perception
+       │          │           │           │          │          │
+       ▼──────────▼───────────▼───────────▼──────────▼──────────▼
+EventBus ─────────────────────────────────────────────────────────────────
+  normalize → privacy gate → dedup (2s) → memory ingest → subscriber fan-out
+       │                                              │
+       ▼                                              ▼
+  EventReorderBuffer (50ms settle, monotonic sort)    Monitor / Scheduler
+       │                                              │
+       ▼                                              ▼
+CausalFusionPipeline (<10ms SLA)                   EventTrigger → Finding
+  denoise → chain_build → infer → hotspots            │
+       │                                              ▼
+       ▼                                         NotificationBus → LeapBoard
+CausalGraph (ring_limit=100)
+       │
+       ▼
+Learning (Progressive Trust: DRAFT → CANDIDATE → VERIFIED → PRODUCTION)
+```
+
+### Signal Sources
+
+| Source | Transport | Rate / Throttle | Module |
+|--------|-----------|-----------------|--------|
+| `app_focus` | 0.5s polling | — | `platform/observers/` |
+| `clipboard` | 1.0s polling | — | `platform/observers/` |
+| `fs_watcher` | watchdog push | — | `platform/observers/` |
+| `input_tap` | CGEventTap / pynput | 50ms throttle | `platform/observers/` |
+| `gateway` | NDJSON / webhook / long-poll | per-platform | `gateway/` |
+| `perception` | information-gain gated | max 5 fps | `perception/` |
+
+### Core Mechanisms
+
+- **EventBus** (`platform/event_bus.py`) — single entry point for all signals. Applies normalization, a privacy gate (redaction before downstream), deduplication within a 2-second sliding window, and routes to subscribers in batch.
+- **EventReorderBuffer** — optional 50ms settling window that re-sorts out-of-order events by monotonic timestamp before processing, critical for fusing cross-channel signals.
+- **CausalFusionPipeline** (`causal/`) — T1 tier with <10ms SLA; constructs causal chains from temporally adjacent events, infers reliability, and maintains a ring-buffered `CausalGraph` (100 nodes max).
+- **Monitor / EventBridge** (`monitor/`) — pattern-matched subscribers (fnmatch globs) with 1s debounce; triggers wake the Scheduler (<2s), which drives Producers to emit scored Findings onto the NotificationBus.
+
+### LeapBoard Observation
+
+The `/board` web dashboard exposes real-time signal health:
+
+- **Signal Health** — subscriber count, active triggers, watches, drops, debounced events
+- **Live Signal Stream** — last 50 raw events, rolling
+- **Watch Portfolio** — active event-driven watches and their trigger state
+- **Recent Findings** — latest scored observations from Producers
+
+### Testing Signals Locally
+
+```bash
+# Emit mock signals for development / integration tests
+uv run pytest tests/mock_signals/ -q
+
+# Inject a synthetic event into the running EventBus
+from leapflow.platform.event_bus import EventBus
+bus = EventBus()
+bus.handle_event("clipboard_change", {"content": "test", "app": "Terminal"})
+```
+
+See `tests/mock_signals/` for ready-made fixtures covering all six source types.
+
+---
+
 ## Built-in Coding Tools
 
 LeapFlow ships a first-class coding toolset so the agent can *locate → read → edit → verify* code precisely instead of rewriting whole files or shelling out blindly. Every tool is registered with governance metadata (`x_leapflow`) so it flows through the existing idempotency, approval, redaction, path-sensitivity, and audit paths.

src/leapflow/cache/manager.py

@@ -3,6 +3,7 @@
 
 import hashlib
 import json
+import threading
 import time
 from dataclasses import dataclass, field
 from enum import Enum
@@ -51,6 +52,7 @@ class CacheManager:
     def __init__(self, layout: CacheLayout, *, profile_id: str) -> None:
         self._layout = layout
         self._profile_id = profile_id
+        self._connect_lock = threading.Lock()
         self._layout.ensure()
         self._init_schema()
 
@@ -322,7 +324,9 @@ def _build_entry(
         )
 
     def _connect(self):
-        return duckdb.connect(str(self._layout.index_path))
+        """Create a DuckDB connection, serialized to protect against concurrent writes."""
+        with self._connect_lock:
+            return duckdb.connect(str(self._layout.index_path))
 
     def _is_managed_path(self, path: Path) -> bool:
         try:

src/leapflow/causal/pipeline.py

@@ -57,6 +57,17 @@ class ReorderBuffer:
     Holds events for up to `window_s` seconds before releasing them in
     timestamp order. This handles platform-level delivery jitter (e.g.,
     app_switch arriving 200ms after the keyboard shortcut that caused it).
+
+    Timebase note:
+        This buffer sorts by ``CausalEvent.timestamp`` (wall-clock, ``time.time()``
+        origin) because causal events are domain objects that may be serialized,
+        persisted, and correlated across sessions — monotonic clocks are not
+        comparable across processes or restarts.
+
+        In contrast, :class:`~leapflow.platform.reorder_buffer.EventReorderBuffer`
+        sorts by ``payload["_mono_ts"]`` (``time.monotonic()``) because it operates
+        within a single process lifetime on raw observer events where monotonic
+        ordering is both available and more reliable than wall-clock.
     """
 
     __slots__ = ("_window_s", "_buffer")

src/leapflow/cli/commands/daemon.py

@@ -163,6 +163,12 @@ def _print_runtime_status(status: dict) -> None:
             print(f"host_capability: {host['capability_version']}")
         if host.get("last_error"):
             print(f"host_error: {host['last_error']}")
+    build = status.get("build")
+    if isinstance(build, dict) and build.get("commit"):
+        stale = build.get("stale")
+        stale_text = "UNKNOWN" if stale is None else ("STALE — restart with 'leap daemon restart'" if stale else "fresh")
+        dirty = " (dirty)" if build.get("dirty_digest") else ""
+        print(f"build: commit={build['commit']}{dirty} pid={build.get('pid')} status={stale_text}")
 
 
 def _start(settings: object, mock_host: bool) -> int:

src/leapflow/cli/commands/interactive.py

@@ -1573,6 +1573,9 @@ async def _refresh_watch_count() -> None:
                         app.invalidate()
                     elif event_type == "watch.state":
                         await _refresh_watch_count()
+                    elif event_type == "signal.stream":
+                        # Track signal flow activity for health visibility
+                        status.increment_signal_stream()
                     elif event_type == "monitor.error":
                         logger.debug("monitor error notification: %s", payload)
             except (DaemonUnavailableError, OSError, asyncio.IncompleteReadError):

src/leapflow/cli/commands/registry.py

@@ -137,6 +137,7 @@ def supports_runtime(self, runtime: CommandRuntime) -> bool:
     # Board & Monitors (LeapBoard) — one analysis target (current session),
     # rendered through a selectable template lens.
     CommandDef("board", "Analyze the current session; optionally pick a template lens", "Board", args_hint="[<template> | templates|refresh|pause|resume|stop|status]", effect=CommandEffect.SESSION, execution=CommandExecution.SHORT_OPERATION),
+    CommandDef("board signals", "View signal flow health and live event stream", "Board", effect=CommandEffect.READ_ONLY, execution=CommandExecution.SHORT_OPERATION),
     CommandDef("board templates", "List, add, remove, or show board templates", "Board", args_hint="[list|add <path.yaml> [--name id] [--force]|remove <id>|show <id>]", effect=CommandEffect.SESSION, execution=CommandExecution.SHORT_OPERATION),
     CommandDef("board refresh", "Re-analyze the current session (or a watch by id) now", "Board", args_hint="[<id>]", effect=CommandEffect.SESSION, execution=CommandExecution.SHORT_OPERATION),
     CommandDef("board pause", "Pause session analysis (or a watch by id)", "Board", args_hint="[<id>]", effect=CommandEffect.SESSION),

src/leapflow/cli/commands/slash_handlers.py

@@ -1233,7 +1233,7 @@ async def _execute_dashboard(
     if verb == "templates":
         return _execute_board_templates(ctx, rest_tokens)
     if verb == "status":
-        return _execute_board_status(ctx, monitors)
+        return await _execute_board_status(ctx, monitors)
     if verb in ("refresh", "pause", "resume", "stop"):
         return await _execute_board_control(
             ctx, monitors, verb, target=rest_tokens[0] if rest_tokens else "",
@@ -1356,18 +1356,22 @@ def _find_session_watch_id(monitors: Any) -> str:
     return ""
 
 
-def _execute_board_status(ctx: "Context", monitors: Any) -> dict[str, Any]:
+async def _execute_board_status(ctx: "Context", monitors: Any) -> dict[str, Any]:
     """Return session-observation status: watch detail + recent findings + lenses.
 
     Fields:
       - ``templates`` / ``default``: available lenses and the default.
       - ``watches``: every watch with state, run/finding counts, and last-run age.
       - ``findings``: recent findings (severity, title, summary, age) across watches.
+      - ``signal_flow``: real-time signal health metrics (if available).
+      - ``server``: the *separate* LeapBoard web server process's own build
+        fingerprint and staleness verdict, when one is currently running (see
+        ``_board_server_health``).
     """
     library = _template_library(ctx)
     data: dict[str, Any] = {
         "ok": True, "view": "dashboard", "mode": "status",
-        "templates": library.names(), "default": "generic",
+        "templates": library.visible_names(), "default": "generic",
         "watches": [], "findings": [],
     }
     if monitors is None:
@@ -1380,9 +1384,62 @@ def _execute_board_status(ctx: "Context", monitors: Any) -> dict[str, Any]:
         data["findings"] = [f.to_dict() for f in monitors.list_findings(limit=20)]
     except Exception:
         logger.debug("dashboard: finding list unavailable", exc_info=True)
+
+    # Signal flow health metrics (best-effort, non-blocking)
+    try:
+        from leapflow.monitor.signal_metrics import SignalMetricsCollector
+
+        collector = SignalMetricsCollector()
+        event_bus = getattr(ctx, "event_bus", None)
+        snapshot = collector.collect(
+            event_bus=event_bus,
+            monitor_manager=monitors,
+        )
+        data["signal_flow"] = {
+            "event_subscriber_count": snapshot.event_subscriber_count,
+            "active_trigger_count": snapshot.active_trigger_count,
+            "signal_buffer_dropped": snapshot.signal_buffer_dropped,
+            "signal_noise_suppressed": snapshot.signal_noise_suppressed,
+            "signal_noise_seen": snapshot.signal_noise_seen,
+            "active_watch_count": snapshot.active_watch_count,
+            "recent_findings_count": snapshot.recent_findings_count,
+        }
+    except Exception:
+        logger.debug("dashboard: signal metrics unavailable", exc_info=True)
+
+    data["server"] = await _board_server_health(ctx)
     return data
 
 
+async def _board_server_health(ctx: "Context") -> dict[str, Any] | None:
+    """Best-effort staleness check for the separately running LeapBoard web server.
+
+    The dashboard is a distinct long-lived process from leapd (spawned by
+    ``leap board`` and never restarted automatically); editing dashboard source
+    does not reach it until it is restarted. Returns None when no dashboard
+    server is currently running/discoverable, or the probe fails — this is a
+    diagnostic side-channel, never a reason to fail ``/board status``.
+    """
+    import asyncio
+
+    settings = getattr(ctx, "settings", None)
+    if settings is None:
+        return None
+    try:
+        from leapflow.dashboard import launcher
+
+        state = launcher.load_state(settings)
+        if not state:
+            return None
+        return await asyncio.to_thread(
+            launcher.fetch_server_info,
+            str(state.get("bind") or ""), int(state.get("port") or 0), str(state.get("token") or ""),
+        )
+    except Exception:
+        logger.debug("dashboard: board server health probe failed", exc_info=True)
+        return None
+
+
 def _execute_board_templates(ctx: "Context", rest_tokens: list[str]) -> dict[str, Any]:
     """Template hub: list / add / remove / show board templates."""
     from leapflow.dashboard.templates import sanitize_template_id
@@ -1392,7 +1449,7 @@ def _execute_board_templates(ctx: "Context", rest_tokens: list[str]) -> dict[str
     args = rest_tokens[1:]
 
     if op == "list":
-        items = [library.describe(name) or {"name": name} for name in library.names()]
+        items = [library.describe(name) or {"name": name} for name in library.visible_names()]
         return {"ok": True, "view": "dashboard", "mode": "templates",
                 "templates": items, "default": "generic"}
 
@@ -1899,6 +1956,31 @@ def _board_page_url() -> str:
         return ""
 
 
+def _render_board_server_health(console: "LeapConsole", server: Any) -> None:
+    """Render the LeapBoard web server's own staleness verdict, if known.
+
+    ``server`` is None when no dashboard server is currently running, or the
+    probe failed — both render nothing, since there is nothing actionable to
+    report. A definite ``stale`` verdict is the whole point of this warning:
+    the browser page a developer is looking at right now was built from code
+    that predates the current source tree.
+    """
+    if not isinstance(server, dict):
+        return
+    build = server.get("build") if isinstance(server.get("build"), dict) else {}
+    stale = server.get("stale")
+    if stale is None:
+        return  # unknown (not a git checkout) — nothing to warn about
+    if stale:
+        console.warning(
+            f"LeapBoard web server (pid={build.get('pid')}) predates the current source "
+            "tree — restart it to pick up recent changes: run /board again after killing "
+            f"pid {build.get('pid')}, or 'leap board --serve' manually."
+        )
+    else:
+        console.system(f"LeapBoard web server (pid={build.get('pid')}) is up to date.")
+
+
 def _render_dashboard_view(console: "LeapConsole", payload: dict[str, Any]) -> None:
     from rich.table import Table
 
@@ -1964,6 +2046,7 @@ def _render_dashboard_view(console: "LeapConsole", payload: dict[str, Any]) -> N
                 console.system(line)
         else:
             console.system("Findings: none yet.")
+        _render_board_server_health(console, payload.get("server"))
         return
 
     if mode == "templates":

src/leapflow/cli/tui_app/status.py

@@ -103,6 +103,7 @@ def __init__(self, theme: Optional[Theme | ResolvedTheme] = None) -> None:
         self.daemon_turn_waiting: int = 0
         self.watch_count: int = 0
         self.alert_count: int = 0
+        self.signal_stream_count: int = 0
         self.last_turn_elapsed: float = 0.0
         self._turn_start: float = 0.0
         self._session_start: float = time.monotonic()
@@ -302,4 +303,8 @@ def update_monitor_counts(
         if watches is not None:
             self.watch_count = max(0, watches)
         if alerts is not None:
-            self.alert_count = max(0, alerts)
+            self.alert_count = max(0, alerts)
+
+    def increment_signal_stream(self) -> None:
+        """Increment the signal stream event counter (called on each signal.stream push)."""
+        self.signal_stream_count += 1

src/leapflow/config.py

@@ -449,6 +449,32 @@ class Settings:
     # signal collection entirely (V0 baseline).
     signal_channels: frozenset = frozenset()
     signal_reactive_capture: bool = False
+    signal_noise_gate_enabled: bool = True
+    signal_noise_same_source_cooldown_s: float = 2.0
+    signal_noise_allow_fs_outside_workspace: bool = False
+    signal_noise_path_fragments: tuple = (
+        "/Library/Caches/",
+        "/Library/Logs/",
+        "/Library/Preferences/",
+        "/Application Support/Qoder/User/globalStorage/",
+        "/Application Support/Qoder/SharedClientCache/",
+        "/Application Support/Qoder/SharedCredentialCache/",
+        "/Application Support/Qoder/Partitions/native-browser/Cache/",
+        "/Application Support/Cursor/User/globalStorage/",
+        "/Application Support/DuetExpertCenter/",
+        "/.cache/",
+        "/.hermes/",
+        "/.r2c/logs/",
+    )
+    signal_noise_dir_names: tuple = (
+        ".git", ".hg", ".svn", ".venv", "__pycache__", ".pytest_cache", ".ruff_cache",
+        "node_modules", "Cache", "Caches", "SharedCredentialCache", "globalStorage",
+        "Code Cache", "GPUCache",
+    )
+    signal_noise_suffixes: tuple = (
+        ".tmp", ".temp", ".swp", ".lock", ".log", ".pyc", ".pyo",
+        "-journal", "-shm", "-wal", ".db-shm", ".db-wal", ".sqlite-journal",
+    )
 
     # ── RPC Transport ──
     # Default fallback timeout (seconds) used by CuaDriverClient when no
@@ -938,6 +964,25 @@ def _build_settings_from_env(
         ) & ALL_SIGNAL_CHANNELS
     signal_reactive_capture = _bool("LEAPFLOW_SIGNAL_REACTIVE_CAPTURE", "false")
 
+    def _tuple_env(key: str, default: tuple) -> tuple:
+        raw = os.getenv(key, ",".join(str(item) for item in default)).strip()
+        return tuple(item.strip() for item in raw.split(",") if item.strip())
+
+    signal_noise_gate_enabled = _bool("LEAPFLOW_SIGNAL_NOISE_GATE_ENABLED", "true")
+    signal_noise_same_source_cooldown_s = float(os.getenv("LEAPFLOW_SIGNAL_NOISE_SAME_SOURCE_COOLDOWN_S", "2.0"))
+    signal_noise_allow_fs_outside_workspace = _bool(
+        "LEAPFLOW_SIGNAL_NOISE_ALLOW_FS_OUTSIDE_WORKSPACE", "false",
+    )
+    signal_noise_path_fragments = _tuple_env(
+        "LEAPFLOW_SIGNAL_NOISE_PATH_FRAGMENTS", Settings.signal_noise_path_fragments,
+    )
+    signal_noise_dir_names = _tuple_env(
+        "LEAPFLOW_SIGNAL_NOISE_DIR_NAMES", Settings.signal_noise_dir_names,
+    )
+    signal_noise_suffixes = _tuple_env(
+        "LEAPFLOW_SIGNAL_NOISE_SUFFIXES", Settings.signal_noise_suffixes,
+    )
+
     # RPC Transport
     rpc_timeout_default = float(os.getenv("LEAPFLOW_RPC_TIMEOUT_DEFAULT", "30.0"))
 
@@ -1256,6 +1301,12 @@ def _build_settings_from_env(
         # Signal Fusion
         signal_channels=signal_channels,
         signal_reactive_capture=signal_reactive_capture,
+        signal_noise_gate_enabled=signal_noise_gate_enabled,
+        signal_noise_same_source_cooldown_s=signal_noise_same_source_cooldown_s,
+        signal_noise_allow_fs_outside_workspace=signal_noise_allow_fs_outside_workspace,
+        signal_noise_path_fragments=signal_noise_path_fragments,
+        signal_noise_dir_names=signal_noise_dir_names,
+        signal_noise_suffixes=signal_noise_suffixes,
         # RPC Transport
         rpc_timeout_default=rpc_timeout_default,
         # Cua Driver