diff --git a/README.md b/README.md index 2aabc1f..59f0a39 100644 --- a/README.md +++ b/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. diff --git a/src/leapflow/cache/manager.py b/src/leapflow/cache/manager.py index 29c3549..aa65306 100644 --- a/src/leapflow/cache/manager.py +++ b/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: diff --git a/src/leapflow/causal/pipeline.py b/src/leapflow/causal/pipeline.py index bba0bc5..6064d07 100644 --- a/src/leapflow/causal/pipeline.py +++ b/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") diff --git a/src/leapflow/cli/commands/daemon.py b/src/leapflow/cli/commands/daemon.py index c1d9bbb..4b7fd32 100644 --- a/src/leapflow/cli/commands/daemon.py +++ b/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: diff --git a/src/leapflow/cli/commands/interactive.py b/src/leapflow/cli/commands/interactive.py index 63da6d8..cb68dac 100644 --- a/src/leapflow/cli/commands/interactive.py +++ b/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): diff --git a/src/leapflow/cli/commands/registry.py b/src/leapflow/cli/commands/registry.py index ca1dcda..f4ffdac 100644 --- a/src/leapflow/cli/commands/registry.py +++ b/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="[