diff --git a/flex/ai/skills/dev/flex-dev-core/SKILL.md b/flex/ai/skills/dev/flex-dev-core/SKILL.md new file mode 100644 index 0000000..fb4e48b --- /dev/null +++ b/flex/ai/skills/dev/flex-dev-core/SKILL.md @@ -0,0 +1,198 @@ +--- +name: flex:dev:core +description: Flex daemon architecture, event-driven watcher, registry, engine internals. Load when modifying the daemon loop, watcher, cell lifecycle, registry, or refresh system. Not for module-level work — use flex:dev:modules for that. +user-invocable: false +--- + +# flex:dev:core + +Reference for the flex daemon runtime, registry, and engine internals. +Load this before touching daemon.py, watcher.py, registry.py, engine.py, +refresh.py, or the systemd integration. + +## Daemon Architecture + +The flex daemon (`flex/daemon.py`) is the local capture/watch process. Entry +points: + +| Path | Command | What runs | +|------|---------|-----------| +| Unified daemon | `python -m flex.daemon` | All loops (main + background + refresh) | +| Worker-only | `python -m flex.daemon --no-refresh --no-background` | Main loop only | +| Direct worker | `python -m flex.modules.claude_code.compile.worker --daemon` | Polling-only fallback | +| Systemd | `flex-worker.service` | Runs worker-only | +| Refresh timer | `flex-refresh.timer` → `python -m flex.refresh` | 30min refresh cycle | + +### Thread Model + +`daemon.py:main()` starts up to 3 threads: + +| Thread | Function | Default interval | Purpose | +|--------|----------|-----------------|---------| +| Main | `daemon_loop()` from worker.py | 0.25s (inotify) / 2s (polling) | File change detection + sync | +| Background | `_background_tick_loop()` | 60s | Hook-driven tasks (`daemon_tick` hook) | +| Refresh | `_refresh_loop()` | 30min (30s startup delay) | Remote cell refresh via `run_due_refreshes()` | + +CLI flags: `--interval`, `--remote-interval`, `--refresh-interval`, +`--no-background`, `--no-refresh`, `--no-inotify`. + +### Instance Locking + +`daemon.py:main()` acquires `fcntl.LOCK_EX` on `~/.flex/daemon.lock`. +Only one daemon instance runs at a time. + +## Event-Driven Watcher + +`flex/watcher.py` provides `FlexWatcher` — a watchdog-based inotify wrapper +with debounced drain. + +### How It Works + +``` +Observer thread (inotify) Main thread (daemon_loop) +───────────────────────── ───────────────────────── +file modified/created ──► _pending dict ◄── drain() every 0.25s + {path: timestamp} returns paths where + (lock-protected) now - ts >= debounce +``` + +- Observer thread does NO SQLite work. It only records `(path, monotonic_time)`. +- Main thread calls `drain()` → gets debounced paths → calls `sync_session_messages()`. +- Debounce window: 500ms default (`FLEX_DEBOUNCE_MS`). Starts from first event per path (prevents starvation during sustained writes). + +### FlexWatcher API + +| Method | Purpose | +|--------|---------| +| `watch(path, pattern, recursive)` | Register directory for inotify | +| `start() -> bool` | Start observer thread | +| `stop()` | Join observer thread | +| `drain() -> list[Path]` | Return debounced paths, clear buffer | +| `active: bool` | Whether observer is running | +| `status() -> dict` | Diagnostics: watches, pending count, debounce | + +### Watcher Setup in daemon.py + +`_build_watcher()` handles all setup: + +1. Check `FLEX_DISABLE_INOTIFY` env → skip if set +2. Import `FlexWatcher` → graceful ImportError if watchdog missing +3. Register `~/.claude/projects/` for `*.jsonl` (Claude Code sessions) +4. Register `discover_watched()` cells (markdown vaults, coding-agent dirs) +5. `start()` → returns watcher or None on failure + +### Fallback Behavior + +If watcher is None (watchdog missing, inotify failed, `--no-inotify`): +- `daemon_loop` runs in polling mode — identical to pre-watcher behavior +- 2-second tick, full `rglob("*.jsonl")` + stat() scan every tick +- Zero behavioral change from the legacy path + +## The Daemon Loop + +`flex/modules/claude_code/compile/worker.py:daemon_loop(interval, watcher)` + +### Two Modes + +**Event-driven (watcher active):** +- Tick: 0.25s +- Each tick: `watcher.drain()` → sync only changed files +- Every 60s (`FLEX_INTEGRITY_INTERVAL`): full `scan_sessions()` as safety net +- Embed sweep: only on changes or integrity tick + +**Polling (watcher is None):** +- Tick: 2s (configurable via `--interval`) +- Each tick: full `scan_sessions()` — rglob + stat() of every JSONL +- Embed sweep: every tick + +### Per-Tick Phases (both modes) + +| Phase | What | Cadence | +|-------|------|---------| +| 0 | Session sync (inotify drain OR polling scan) | Every tick | +| 1 | Embed orphan chunks (`_batch_embed_chunks`, batch=64) | Every tick | +| 2 | Corpus document indexing (`_corpus_drainer`) | Every tick | +| 3 | Secondary cell drain (`_secondary_cell_drainer`) | Every tick | +| 4 | Markdown vault scan (`_markdown_scanner`) | Every tick | +| 5 | Coding-agent watch scan (`_coding_agent_scanner`) | 30s throttle | +| 6 | SOMA heal + eternity backup | 24h | +| 7 | Enrichment cycle + corpus graph refresh | 30min | + +Phases 2-5 are guarded by `try/except ImportError` — missing modules are skipped. + +## scan_sessions (The Filebeat Pattern) + +`worker.py:scan_sessions(conn, size_cache, error_cache)` + +Pure stat()-based polling. The inotify watcher replaces this as the primary +detection mechanism but it remains as the integrity scanner. + +- Walks `~/.claude/projects/**/*.jsonl` via `rglob()` +- Compares `stat().st_size` to `size_cache[session_id]` +- If file grew → `sync_session_messages(session_id, conn)` +- On error → exponential backoff (5s, 10s, 20s... up to 300s) per session +- Rejects symlinks + +## Registry + +`flex/registry.py` — SQLite database at `~/.flex/registry.db`. + +### Key Functions + +| Function | Purpose | +|----------|---------| +| `register_cell(name, path, cell_type, ...)` | UPSERT cell into registry | +| `resolve_cell(name_or_type)` | Find cell path by name or cell_type | +| `discover_watched()` | Cells with `lifecycle='watch'` + `watch_path` | +| `discover_install_modules()` | Scan for module `install.py` files | +| `discover_module_specs()` | Load MODULE dicts from install modules | +| `module_spec_for(cell_type)` | Resolve spec by cell type | +| `get_hook(name)` | Get registered hook function | +| `load_plugins()` | Load plugin modules from `~/.flex/plugins/` | + +### Cell Lifecycle Values + +| Value | Meaning | +|-------|---------| +| `static` | No refresh (default) | +| `refresh` | Periodic refresh via `refresh_module` or `refresh_script` | +| `watch` | Daemon monitors `watch_path` for file changes | + +### Registry Schema (cells table) + +Key columns: `name`, `path`, `cell_type`, `lifecycle`, `refresh_module`, +`refresh_script`, `refresh_interval`, `watch_path`, `watch_pattern`, +`last_refresh_at`, `refresh_status`, `active`, `unlisted`, `substrate`. + +## Environment Variables + +| Variable | Default | Purpose | +|----------|---------|---------| +| `FLEX_HOME` | `~/.flex` | Flex home directory | +| `FLEX_DISABLE_INOTIFY` | unset | Force polling mode | +| `FLEX_DEBOUNCE_MS` | `500` | inotify debounce window (ms) | +| `FLEX_INTEGRITY_INTERVAL` | `60` | Seconds between full stat() scans | +| `FLEX_CODING_AGENT_WATCH_INTERVAL_SEC` | `30` | Coding-agent scan throttle | +| `FLEX_MODULE_PATH` | unset | Additional module search paths | + +## File Map + +``` +flex/ +├── daemon.py ← unified daemon entry point, watcher setup +├── watcher.py ← FlexWatcher (inotify + debounce) +├── registry.py ← cell registry, module discovery +├── engine.py ← query engine (run_from_spec substrate dispatch) +├── refresh.py ← remote cell refresh orchestrator +├── health.py ← daemon health checks +├── cli.py ← CLI entry point (flex command) +├── core.py ← shared utilities +├── serve.py ← MCP server launcher +├── mcp_server.py ← MCP tool implementations +├── mcp_core.py ← MCP transport core +├── sdk.py ← public SDK +├── views.py ← view installation +├── secrets.py ← ~/.flex/secrets loader +├── instructions.py ← cell instruction generator +└── modules/ ← see flex:dev:modules +``` diff --git a/flex/ai/skills/dev/flex-dev-modules/SKILL.md b/flex/ai/skills/dev/flex-dev-modules/SKILL.md new file mode 100644 index 0000000..d1bd98e --- /dev/null +++ b/flex/ai/skills/dev/flex-dev-modules/SKILL.md @@ -0,0 +1,256 @@ +--- +name: flex:dev:modules +description: Flex module system — how modules declare themselves, the MODULE spec dict, worker patterns, transpilers, install flow, substrate types, and the watch/refresh lifecycle. Load when creating a new module, modifying an existing module's worker or install, or working with the substrate extraction layer. +user-invocable: false +--- + +# flex:dev:modules + +Reference for the flex module system. Load this before creating or modifying +modules under `flex/modules/`, working with MODULE specs, transpilers, +install flows, or the substrate layer. + +## Module Directory Structure + +Every module lives under `flex/modules//` with a standard layout: + +``` +modules// +├── __init__.py # Usually empty or exports constants +├── install.py # MODULE dict + register_args() + run() +├── compile/ +│ ├── __init__.py +│ └── worker.py # transpile() and/or scan functions +└── stock/ + ├── instructions.md # Cell-bundled instructions for agents + ├── views/*.sql # SQL view definitions + └── presets/*.sql # Named query presets +``` + +Optional files: `contract.py` (validation), `refresh.py` (remote refresh), +`README.md` (module docs). + +## Current Modules (v0.40.0+) + +| Module | cell_type | substrate | Pattern | Purpose | +|--------|-----------|-----------|---------|---------| +| `claude_code` | `claude_code` | — | Scanner + daemon | CC session ingestion, master daemon loop | +| `codex` | `codex` | `claude_code` | Transpiler | Codex rollout JSONL → CC schema | +| `goose` | `goose` | `claude_code` | Transpiler | Goose SQLite → CC schema | +| `markdown` | `markdown` | — | Scanner | Obsidian/vault markdown files | +| `arxiv` | `arxiv` | — | Transpiler | arXiv papers via API | +| `github` | `github` | — | Transpiler | GitHub issues/discussions | +| `hn` | `hn` | — | Transpiler | Hacker News threads | +| `reddit` | `reddit` | — | Transpiler | Reddit threads via Arctic Shift | +| `skills` | `skills` | — | Transpiler | Public AI skill/tool catalog | +| `soma` | — | — | Library | Identity module (no install) | + +## The MODULE Spec Dict + +Each module's `install.py` exports a `MODULE` dict. This is the module's +self-declaration — discovered by `registry.py:discover_module_specs()`. + +### Required Fields + +```python +MODULE = { + "cell_type": "mymodule", # unique identifier, used in registry +} +``` + +### Common Fields + +```python +MODULE = { + "cell_type": "codex", + "substrate": "claude_code", # schema inheritance + "default_cell_name": "codex", # registry name + "transpile": "flex.modules.codex.compile.worker:transpile", # dotted ref + "signature": "flex.modules.codex.compile.worker:compute_dir_signature", + "refresh_module": "flex.modules.codex.refresh", # for lifecycle=refresh + "watch_path": "/path/to/watch", # for lifecycle=watch + "watch_pattern": "**/*.jsonl", # glob pattern + "views_from": ("claude_code",), # reuse views + "presets_from": ("claude_code", "soma"), # reuse presets + "enrichment_stubs_from": "claude_code", # reuse enrichment + "default_source": {"path": "...", "type": "dir"}, # install source +} +``` + +### Substrate Field + +The `substrate` field controls which schema tables are created at install time. + +| Value | Effect | +|-------|--------| +| `"base"` | Base tables only (chunks, sources, metadata, cells, fts) | +| `"claude_code"` | Base + CC tables (messages, sessions, files, types, tools) + SOMA | +| _(absent)_ | Module handles its own schema | + +Substrate extraction (`contract.py`) defines: +- `REQUIRED_BASE_TABLES`: chunks, sources, source_graph, chunk_metadata, cells +- `REQUIRED_TABLES`: base + messages, sessions, files (coding-agent cells) +- `validate_base_cell(conn)` / `validate_coding_agent_cell(conn)` → `ContractReport` + +## install.py Pattern + +Every module's `install.py` exports three things: + +```python +MODULE = { ... } # spec dict + +def register_args(sub): + """Register CLI subcommand arguments.""" + sub.add_argument("source", nargs="?", help="Source path or URL") + sub.add_argument("--name", help="Cell name override") + # ... module-specific args + +def run(args, console): + """Install entry point. Called by `flex install `.""" + from flex.modules.claude_code.coding_agent_install import run_from_spec + run_from_spec(args, console, MODULE) +``` + +Most modules delegate to `run_from_spec()` which handles: +1. Cell creation with correct substrate schema +2. Transpiler loading via `_load_ref(spec["transpile"])` +3. Transpile → embed → enrichment pipeline +4. Registry registration with lifecycle kwargs + +## Worker Patterns + +### Pattern A: Transpiler (codex, goose, arxiv, github, hn, reddit, skills) + +`compile/worker.py` exports a `transpile(source_path, conn)` function. +Called on-demand by `run_from_spec()` or `refresh_cell()`. One-shot processing. + +```python +def transpile(source_path: str | Path, conn: sqlite3.Connection, + progress=None, encode_fn=None) -> dict: + """Transpile source data into cell chunks. + + Returns: + dict with stats like {'sessions': N, 'chunks': N, 'skipped': N} + """ +``` + +Transpilers read external data (JSONLs, SQLite, APIs), map it to the cell's +chunk schema, and insert via `insert_base_chunk()` or `insert_chunk_atom()`. + +### Pattern B: Scanner (claude_code, markdown) + +`compile/worker.py` exports a daemon-callable scan function that maintains +in-memory caches and is called every tick from the daemon loop. + +```python +def scan_sessions(conn, size_cache, error_cache=None) -> dict: + """Stat-based polling scan. Returns {'synced': N, 'chunks': N}.""" + +def scan_markdown_cells() -> dict: + """Walk vault directories, sync changed files. Returns {'indexed': N}.""" +``` + +Scanners own their caches (`size_cache`, `_hash_cache`) and are designed +for repeated invocation. + +### claude_code Is Both + +The `claude_code` module is unique — it contains the master `daemon_loop()` +AND is a scanner for its own cell type. All other scanners and transpilers +are called from within `claude_code`'s daemon loop. + +## Cell Lifecycle + +Modules participate in the daemon via lifecycle registration: + +| Lifecycle | How the daemon handles it | +|-----------|--------------------------| +| `static` | Ignored by daemon — install-time only | +| `watch` | `discover_watched()` returns the cell; daemon monitors `watch_path` for changes via inotify or polling | +| `refresh` | `run_due_refreshes()` checks `refresh_interval` and calls `refresh_module` when due | + +### Watch Lifecycle (Event-Driven) + +When a cell has `lifecycle='watch'`: +1. `discover_watched()` finds it in the registry +2. `daemon.py:_build_watcher()` registers its `watch_path` + `watch_pattern` with `FlexWatcher` +3. inotify detects file changes → `drain()` returns paths → daemon syncs +4. Integrity scan (60s) catches anything inotify missed + +### Refresh Lifecycle (Timer-Driven) + +When a cell has `lifecycle='refresh'`: +1. `flex/refresh.py:run_due_refreshes()` checks if `refresh_interval` has elapsed +2. Loads `refresh_module` → calls its `refresh(cell_name)` function +3. Updates `last_refresh_at` and `refresh_status` in registry + +## Key Functions for Module Authors + +### Schema Bootstrap + +| Function | File | Purpose | +|----------|------|---------| +| `_ensure_core_tables(conn)` | worker.py | CC-specific tables | +| `_ensure_content_tables(conn)` | worker.py | Content/FTS tables | +| `bootstrap_cell(name, cell_type, substrate)` | coding_agent_install.py | Create cell with substrate-appropriate schema | + +### Chunk Insertion + +| Function | File | Purpose | +|----------|------|---------| +| `insert_base_chunk(conn, ...)` | worker.py | Insert into base schema (any substrate) | +| `insert_chunk_atom(conn, ...)` | worker.py | Insert into full CC schema (messages, sessions, files) | +| `sync_session_messages(session_id, conn)` | worker.py | Sync a CC JSONL file into chunks | + +### Discovery + +| Function | File | Purpose | +|----------|------|---------| +| `discover_module_specs()` | specs.py | Load all MODULE dicts | +| `module_spec_for(cell_type)` | specs.py | Get spec for a cell type | +| `discover_install_modules()` | specs.py | Find all installable modules | +| `discover_watched()` | registry.py | Cells with watch lifecycle | + +## Module Search Paths + +Modules are discovered from three locations (in order): + +1. **Packaged:** `flex/modules/*/install.py` (in the repo) +2. **User-installed:** `~/.flex/modules/*/install.py` +3. **FLEX_MODULE_PATH:** Additional directories (colon-separated env var) + +## Creating a New Module + +1. Create `flex/modules//` with the standard layout +2. Define `MODULE` dict in `install.py` with at minimum `cell_type` +3. Set `substrate` if inheriting schema (e.g. `"claude_code"` or `"base"`) +4. Implement `compile/worker.py` with `transpile()` function +5. Add `stock/views/*.sql` for queryable surfaces +6. Add `stock/instructions.md` for agent-facing documentation +7. Delegate `run()` to `run_from_spec(args, console, MODULE)` +8. For watch cells: set `watch_path`, `watch_pattern` in MODULE dict +9. For refresh cells: implement `refresh.py` with `refresh(cell_name)` function + +## File Map + +``` +flex/modules/ +├── claude_code/ +│ ├── __init__.py # BASE_ENRICHMENT_STUBS, ENRICHMENT_STUBS +│ ├── contract.py # Schema validation (base vs full) +│ ├── coding_agent_install.py # run_from_spec() — generic install pipeline +│ ├── coding_agent_watch.py # scan_coding_agent_cells() for watch lifecycle +│ └── compile/ +│ ├── worker.py # daemon_loop(), scan_sessions(), sync, embed +│ └── soft_detect.py # Session type detection heuristics +├── codex/ # Codex rollout ingestion +├── goose/ # Goose session ingestion +├── markdown/ # Vault/markdown ingestion +├── arxiv/ # arXiv paper ingestion +├── github/ # GitHub issue/discussion ingestion +├── hn/ # Hacker News thread ingestion +├── reddit/ # Reddit thread ingestion +├── skills/ # Public skill/tool catalog +├── soma/ # Identity module (library, no install) +└── specs.py # Module discovery + spec resolution +``` diff --git a/flex/daemon.py b/flex/daemon.py index a3396f3..eecf34b 100644 --- a/flex/daemon.py +++ b/flex/daemon.py @@ -40,6 +40,47 @@ def _load_secrets(): os.environ.setdefault(key.strip(), val.strip()) +def _build_watcher(): + """Create and configure the inotify watcher. Returns None on failure.""" + from pathlib import Path + + if os.environ.get("FLEX_DISABLE_INOTIFY"): + print(" inotify watcher: disabled (FLEX_DISABLE_INOTIFY)", file=sys.stderr) + return None + + try: + from flex.watcher import FlexWatcher + except ImportError: + print(" inotify watcher: unavailable (watchdog not installed)", file=sys.stderr) + return None + + debounce_ms = int(os.environ.get("FLEX_DEBOUNCE_MS", "500")) + watcher = FlexWatcher(debounce_ms=debounce_ms) + + # Primary: Claude Code JSONLs + cc_projects = Path.home() / ".claude" / "projects" + if cc_projects.exists(): + watcher.watch(cc_projects, pattern="*.jsonl", recursive=True) + + # Secondary: watched cells from registry (markdown vaults, coding-agent dirs) + try: + from flex.registry import discover_watched + for cell in discover_watched(): + wp = cell.get("watch_pattern", "*") + watcher.watch(cell["watch_path"], pattern=wp, recursive=True) + except Exception as e: + print(f" inotify watcher: registry scan failed ({e})", file=sys.stderr) + + if watcher.start(): + print(f" inotify watcher: active ({watcher.watch_count} watches, " + f"{debounce_ms}ms debounce)", file=sys.stderr) + return watcher + else: + print(" inotify watcher: failed to start, falling back to polling", + file=sys.stderr) + return None + + def _background_tick_loop(interval: int = 60): """Background task loop. Hook-driven — no-op if no hook registered.""" while True: @@ -122,8 +163,13 @@ def main(): help="Disable background tasks") parser.add_argument("--no-refresh", action="store_true", help="Disable refresh cycle") + parser.add_argument("--no-inotify", action="store_true", + help="Disable inotify watcher, use polling only") args = parser.parse_args() + if args.no_inotify: + os.environ["FLEX_DISABLE_INOTIFY"] = "1" + print("[flex-daemon] Starting unified daemon", file=sys.stderr) print(f" Local scan: {args.interval}s", file=sys.stderr) print(f" Background: {'disabled' if args.no_background else f'{args.remote_interval}s'}", @@ -131,6 +177,9 @@ def main(): print(f" Refresh: {'disabled' if args.no_refresh else f'{args.refresh_interval}s'}", file=sys.stderr) + # inotify watcher (best-effort, falls back to polling) + watcher = _build_watcher() + # Thread 1: background tasks (plugin-driven) if not args.no_background: t = threading.Thread( @@ -152,7 +201,11 @@ def main(): # Main thread: local cell scan (blocks if module available) try: from flex.modules.claude_code.compile.worker import daemon_loop - daemon_loop(interval=args.interval) + try: + daemon_loop(interval=args.interval, watcher=watcher) + finally: + if watcher: + watcher.stop() except ImportError: print("[flex-daemon] claude_code module not installed — running background services only", file=sys.stderr) diff --git a/flex/modules/claude_code/__init__.py b/flex/modules/claude_code/__init__.py index 78c0667..b8c47ab 100644 --- a/flex/modules/claude_code/__init__.py +++ b/flex/modules/claude_code/__init__.py @@ -9,8 +9,20 @@ (formerly install._run_enrichment_quiet). """ -# Single source of truth for coding-agent enrichment stub tables. -ENRICHMENT_STUBS: list[str] = [ +# Base enrichment stubs -- generic tables any flex module may need. +BASE_ENRICHMENT_STUBS: list[str] = [ + """CREATE TABLE IF NOT EXISTS _ops ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp INTEGER DEFAULT (strftime('%s','now')), + operation TEXT, target TEXT, sql TEXT, params TEXT, + rows_affected INTEGER, source TEXT)""", + """CREATE TABLE IF NOT EXISTS _views ( + name TEXT PRIMARY KEY, sql TEXT NOT NULL, + description TEXT, created_at INTEGER)""", +] + +# CC-specific enrichment stubs -- coding-agent graph intelligence tables. +_CC_ENRICHMENT_STUBS: list[str] = [ """CREATE TABLE IF NOT EXISTS _enrich_source_graph ( source_id TEXT PRIMARY KEY, centrality REAL, is_hub INTEGER DEFAULT 0, is_bridge INTEGER DEFAULT 0, community_id INTEGER, community_label TEXT)""", @@ -27,16 +39,11 @@ source_id TEXT PRIMARY KEY, agents_spawned INTEGER, is_orchestrator INTEGER DEFAULT 0, delegation_depth INTEGER, parent_session TEXT)""", - """CREATE TABLE IF NOT EXISTS _ops ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - timestamp INTEGER DEFAULT (strftime('%s','now')), - operation TEXT, target TEXT, sql TEXT, params TEXT, - rows_affected INTEGER, source TEXT)""", - """CREATE TABLE IF NOT EXISTS _views ( - name TEXT PRIMARY KEY, sql TEXT NOT NULL, - description TEXT, created_at INTEGER)""", ] +# Full coding-agent enrichment stubs (base + CC). Backward-compatible. +ENRICHMENT_STUBS: list[str] = BASE_ENRICHMENT_STUBS + _CC_ENRICHMENT_STUBS + def __getattr__(name): """Lazy import of run_enrichment — avoids heavy module load at package init.""" diff --git a/flex/modules/claude_code/coding_agent_install.py b/flex/modules/claude_code/coding_agent_install.py index 8fe7da0..479a8d6 100644 --- a/flex/modules/claude_code/coding_agent_install.py +++ b/flex/modules/claude_code/coding_agent_install.py @@ -70,16 +70,17 @@ def run_from_spec(args, console, spec: dict[str, Any]) -> None: from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn from rich.text import Text - from flex.modules.claude_code import ENRICHMENT_STUBS, run_enrichment + from flex.modules.claude_code import BASE_ENRICHMENT_STUBS, ENRICHMENT_STUBS, run_enrichment from flex.modules.claude_code.compile.worker import ( _batch_embed_chunks, - bootstrap_claude_code_cell, + bootstrap_cell, ) - from flex.modules.claude_code.contract import validate_coding_agent_cell + from flex.modules.claude_code.contract import validate_base_cell, validate_coding_agent_cell from flex.registry import register_cell from flex.cli import _install_claude_assets cell_type = spec["cell_type"] + substrate = spec.get("substrate", "claude_code") name = getattr(args, "name", None) or spec.get("default_cell_name") or cell_type description = spec.get("description") or f"{cell_type} coding-agent session provenance." source_attr = spec["source_arg"].lstrip("-").replace("-", "_") @@ -94,13 +95,14 @@ def run_from_spec(args, console, spec: dict[str, Any]) -> None: console.print(f" [yellow]not found[/yellow] — {spec.get('missing_hint', 'run the source agent at least once.')}") return - db_path = bootstrap_claude_code_cell(name=name, cell_type=cell_type) + db_path = bootstrap_cell(name=name, cell_type=cell_type, substrate=substrate) conn = sqlite3.connect(str(db_path), timeout=30.0) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=30000") - for ddl in ENRICHMENT_STUBS: + stubs = ENRICHMENT_STUBS if substrate == "claude_code" else BASE_ENRICHMENT_STUBS + for ddl in stubs: conn.execute(ddl) conn.execute( "INSERT OR REPLACE INTO _meta (key, value) VALUES ('description', ?)", @@ -161,7 +163,10 @@ def _e_cb(done, total): def _g_cb(step): progress.update(t_graph, info=step) - n_comm, failed = run_enrichment(conn, cell_type=cell_type, progress_cb=_g_cb) + if substrate == "claude_code": + n_comm, failed = run_enrichment(conn, cell_type=cell_type, progress_cb=_g_cb) + else: + n_comm, failed = 0, [] progress.update( t_graph, visible=True, @@ -181,7 +186,10 @@ def _g_cb(step): except OSError: pass - report = validate_coding_agent_cell(conn, cell_type=cell_type) + if substrate == "claude_code": + report = validate_coding_agent_cell(conn, cell_type=cell_type) + else: + report = validate_base_cell(conn, cell_type=cell_type) if not report.ok or report.warnings: console.print() console.print(f" [yellow]{report.summary()}[/yellow]") diff --git a/flex/modules/claude_code/compile/worker.py b/flex/modules/claude_code/compile/worker.py index ed69d6d..c507dcb 100644 --- a/flex/modules/claude_code/compile/worker.py +++ b/flex/modules/claude_code/compile/worker.py @@ -289,8 +289,12 @@ def update_source_stats(conn: sqlite3.Connection, session_id: str, chunk: dict): """, (clean[:250], session_id)) -def _ensure_core_tables(conn: sqlite3.Connection): - """Create all chunk-atom tables for a fresh cell. Idempotent.""" +def _ensure_base_tables(conn: sqlite3.Connection): + """Create generic flex chunk-atom tables. Idempotent. + + These tables are the shared storage contract for all flex modules -- + not specific to any particular source type (Claude Code, Hermes, Matrix, etc.). + """ conn.executescript(""" CREATE TABLE IF NOT EXISTS _raw_chunks ( id TEXT PRIMARY KEY, @@ -327,6 +331,50 @@ def _ensure_core_tables(conn: sqlite3.Connection): CREATE INDEX IF NOT EXISTS idx_es_chunk ON _edges_source(chunk_id); CREATE INDEX IF NOT EXISTS idx_es_source ON _edges_source(source_id); + CREATE TABLE IF NOT EXISTS _meta ( + key TEXT PRIMARY KEY, + value TEXT + ); + + CREATE TABLE IF NOT EXISTS _presets ( + name TEXT PRIMARY KEY, + description TEXT, + params TEXT DEFAULT '', + sql TEXT + ); + + CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5( + content, + content='_raw_chunks', + content_rowid='rowid' + ); + """) + # FTS triggers -- can't use IF NOT EXISTS, so check first + has_trigger = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='trigger' AND name='raw_chunks_ai'" + ).fetchone() + if not has_trigger: + conn.executescript(""" + CREATE TRIGGER raw_chunks_ai AFTER INSERT ON _raw_chunks BEGIN + INSERT INTO chunks_fts(rowid, content) VALUES (new.rowid, new.content); + END; + CREATE TRIGGER raw_chunks_ad AFTER DELETE ON _raw_chunks BEGIN + INSERT INTO chunks_fts(chunks_fts, rowid, content) VALUES('delete', old.rowid, old.content); + END; + CREATE TRIGGER raw_chunks_au AFTER UPDATE ON _raw_chunks BEGIN + INSERT INTO chunks_fts(chunks_fts, rowid, content) VALUES('delete', old.rowid, old.content); + INSERT INTO chunks_fts(rowid, content) VALUES (new.rowid, new.content); + END; + """) + + +def _ensure_cc_tables(conn: sqlite3.Connection): + """Create Claude Code specific extension tables. Idempotent. + + These tables capture coding-agent concepts: tool operations, message + threading, agent delegation, soft file-op detection, and file bodies. + """ + conn.executescript(""" CREATE TABLE IF NOT EXISTS _edges_tool_ops ( chunk_id TEXT PRIMARY KEY, tool_name TEXT, @@ -382,43 +430,16 @@ def _ensure_core_tables(conn: sqlite3.Connection): position INTEGER ); CREATE INDEX IF NOT EXISTS idx_tfb_file ON _types_file_body(target_file); + """) - CREATE TABLE IF NOT EXISTS _meta ( - key TEXT PRIMARY KEY, - value TEXT - ); - - CREATE TABLE IF NOT EXISTS _presets ( - name TEXT PRIMARY KEY, - description TEXT, - params TEXT DEFAULT '', - sql TEXT - ); - CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5( - content, - content='_raw_chunks', - content_rowid='rowid' - ); - """) - # FTS triggers — can't use IF NOT EXISTS, so check first - has_trigger = conn.execute( - "SELECT 1 FROM sqlite_master WHERE type='trigger' AND name='raw_chunks_ai'" - ).fetchone() - if not has_trigger: - conn.executescript(""" - CREATE TRIGGER raw_chunks_ai AFTER INSERT ON _raw_chunks BEGIN - INSERT INTO chunks_fts(rowid, content) VALUES (new.rowid, new.content); - END; - CREATE TRIGGER raw_chunks_ad AFTER DELETE ON _raw_chunks BEGIN - INSERT INTO chunks_fts(chunks_fts, rowid, content) VALUES('delete', old.rowid, old.content); - END; - CREATE TRIGGER raw_chunks_au AFTER UPDATE ON _raw_chunks BEGIN - INSERT INTO chunks_fts(chunks_fts, rowid, content) VALUES('delete', old.rowid, old.content); - INSERT INTO chunks_fts(rowid, content) VALUES (new.rowid, new.content); - END; - """) +def _ensure_core_tables(conn: sqlite3.Connection): + """Create all chunk-atom tables for a fresh cell. Idempotent. + Backward-compatible wrapper -- calls base + CC tables. + """ + _ensure_base_tables(conn) + _ensure_cc_tables(conn) def _ensure_content_tables(conn: sqlite3.Connection): """Create content store tables if they don't exist.""" @@ -582,22 +603,37 @@ def _normalize_tool_result(content) -> str | None: return None -def insert_chunk_atom(conn: sqlite3.Connection, chunk: dict): - """Insert a chunk into all chunk-atom tables.""" +def insert_base_chunk(conn: sqlite3.Connection, chunk: dict, + source_type: str = 'claude-code'): + """Insert a chunk into base tables only. Module-agnostic. + + Writes to _raw_chunks and _edges_source. Any flex module can call + this without requiring CC-specific extension tables. + """ cur = conn.cursor() chunk_id = chunk['id'] - # _raw_chunks cur.execute(""" INSERT OR IGNORE INTO _raw_chunks (id, content, embedding, timestamp) VALUES (?, ?, ?, ?) """, (chunk_id, chunk['content'], chunk.get('embedding'), chunk['timestamp'])) - # _edges_source cur.execute(""" INSERT OR IGNORE INTO _edges_source (chunk_id, source_id, source_type, position) - VALUES (?, ?, 'claude-code', ?) - """, (chunk_id, chunk['doc_id'], chunk['chunk_number'])) + VALUES (?, ?, ?, ?) + """, (chunk_id, chunk['doc_id'], source_type, chunk['chunk_number'])) + + +def insert_chunk_atom(conn: sqlite3.Connection, chunk: dict): + """Insert a chunk into all CC chunk-atom tables. + + Calls insert_base_chunk() for generic storage, then writes + CC-specific extension tables (message types, tool ops, etc.). + """ + insert_base_chunk(conn, chunk, source_type='claude-code') + + cur = conn.cursor() + chunk_id = chunk['id'] # _types_message cur.execute(""" @@ -1289,17 +1325,19 @@ def process_queue(conn: sqlite3.Connection) -> dict: ) -def bootstrap_claude_code_cell( +def bootstrap_cell( name: str = 'claude_code', cell_type: str = 'claude-code', description: str | None = None, + substrate: str = 'claude_code', ) -> Path: - """Create a coding-agent cell with the CC canonical schema. Idempotent. + """Create a flex cell with the appropriate schema. Idempotent. + + substrate='claude_code' -- full coding-agent schema (base + CC + content + SOMA) + substrate='base' -- generic chunk schema only (base + content) Defaults preserve the original behavior — existing CC callers pass nothing and get a cell named 'claude_code' / cell_type='claude-code'. - Compatible coding-agent modules pass their own name/cell_type to reuse - the same substrate. """ desc = description or _DEFAULT_CC_DESCRIPTION existing = resolve_cell(name) @@ -1314,10 +1352,14 @@ def bootstrap_claude_code_cell( conn = sqlite3.connect(str(db_path), timeout=30.0) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=30000") - _ensure_core_tables(conn) + + _ensure_base_tables(conn) _ensure_content_tables(conn) - if soma_ensure_tables: - soma_ensure_tables(conn) + + if substrate == 'claude_code': + _ensure_cc_tables(conn) + if soma_ensure_tables: + soma_ensure_tables(conn) conn.execute("INSERT OR IGNORE INTO _meta VALUES ('description', ?)", (desc,)) conn.execute("INSERT OR IGNORE INTO _meta VALUES ('cell_type', ?)", (cell_type,)) @@ -1328,6 +1370,10 @@ def bootstrap_claude_code_cell( return db_path +# Backward-compatible alias +bootstrap_claude_code_cell = bootstrap_cell + + def _batch_embed_chunks(conn, batch_size: int = 500, quiet: bool = False, progress_cb=None, embedder=None) -> int: """Phase 2 of decoupled backfill: batch embed all NULL-embedding chunks. @@ -1802,8 +1848,15 @@ def _run_enrichment_cycle(conn, graph_threshold=50): print(f"[enrich] Cycle done in {elapsed:.1f}s", file=sys.stderr) -def daemon_loop(interval=2): - """Main daemon loop.""" +def daemon_loop(interval=2, watcher=None): + """Main daemon loop. + + Args: + interval: Polling interval in seconds (default 2). When watcher is + active, the tick interval drops to 0.25s for responsiveness. + watcher: Optional FlexWatcher instance for inotify-based change detection. + When provided, polling becomes a slow integrity scan. + """ # Resolve cell cell_path = resolve_cell('claude_code') if not cell_path: @@ -1834,6 +1887,11 @@ def daemon_loop(interval=2): if _corpus_drainer: print(" Corpus indexing: enabled", file=sys.stderr) + if watcher and watcher.active: + print(" Mode: event-driven (inotify + integrity scan)", file=sys.stderr) + else: + print(" Mode: polling", file=sys.stderr) + ENRICHMENT_INTERVAL = 30 * 60 # 30 minutes — graph, fingerprints, repo_project GRAPH_STALENESS_THRESHOLD = 50 # sessions since last graph build @@ -1842,16 +1900,61 @@ def daemon_loop(interval=2): # Size caches — empty dict triggers full initial scan on first tick size_cache: dict = {} + error_cache: dict = {} + + # inotify mode: integrity scan cadence (full rglob as safety net) + INTEGRITY_INTERVAL = int(os.environ.get("FLEX_INTEGRITY_INTERVAL", "60")) + last_integrity = 0.0 + inotify_active = watcher is not None and watcher.active + tick_interval = 0.25 if inotify_active else interval while True: - # Phase 0: stat() scan — sync any JSONLs that grew (replaces queue + backfill) - try: - stats = scan_sessions(conn, size_cache) - if stats['synced'] > 0: - print(f"[worker] synced={stats['synced']} chunks={stats['chunks']}", + now_mono = time.monotonic() + + if inotify_active: + # Event-driven path: sync only files that inotify flagged as changed + changed_paths = watcher.drain() + ev_synced = 0 + ev_chunks = 0 + for p in changed_paths: + session_id = p.stem + try: + count = sync_session_messages(session_id, conn) + _update_warmup(conn, session_id) + try: + size_cache[session_id] = p.stat().st_size + except OSError: + pass + if count > 0: + ev_synced += 1 + ev_chunks += count + except Exception as e: + print(f"[worker] inotify sync error {session_id[:12]}: {e}", + file=sys.stderr) + if ev_chunks > 0: + conn.commit() + print(f"[worker] inotify synced={ev_synced} chunks={ev_chunks}", file=sys.stderr) - except Exception as e: - print(f"[worker] Scan error: {e}", file=sys.stderr) + + # Integrity scan on slow cadence — catches anything inotify missed + if now_mono - last_integrity > INTEGRITY_INTERVAL: + try: + stats = scan_sessions(conn, size_cache, error_cache) + if stats['synced'] > 0: + print(f"[worker] integrity synced={stats['synced']} " + f"chunks={stats['chunks']}", file=sys.stderr) + except Exception as e: + print(f"[worker] Integrity scan error: {e}", file=sys.stderr) + last_integrity = now_mono + else: + # Polling fallback — original behavior, unchanged + try: + stats = scan_sessions(conn, size_cache, error_cache) + if stats['synced'] > 0: + print(f"[worker] synced={stats['synced']} chunks={stats['chunks']}", + file=sys.stderr) + except Exception as e: + print(f"[worker] Scan error: {e}", file=sys.stderr) # Sweep NULL embeddings every tick — catches interrupted flex init, # failed embeds, or any other orphaned chunks. Small batch (64) so @@ -1935,7 +2038,7 @@ def daemon_loop(interval=2): except Exception as e: print(f"[worker] Corpus graph refresh error: {e}", file=sys.stderr) - time.sleep(interval) + time.sleep(tick_interval) if __name__ == "__main__": diff --git a/flex/modules/claude_code/contract.py b/flex/modules/claude_code/contract.py index c86a09b..83ce04b 100644 --- a/flex/modules/claude_code/contract.py +++ b/flex/modules/claude_code/contract.py @@ -18,6 +18,16 @@ from dataclasses import dataclass, field +# Tables that MUST exist after ingest for any flex cell (base contract). +REQUIRED_BASE_TABLES: tuple[str, ...] = ( + "_raw_sources", + "_raw_chunks", + "_raw_content", + "_edges_source", + "_edges_raw_content", +) + + # Tables that MUST exist after ingest for any coding-agent cell. # A missing table = schema-level violation (probably a transpiler bug). REQUIRED_TABLES: tuple[str, ...] = ( @@ -157,3 +167,35 @@ def validate_coding_agent_cell( )) return report + + +def validate_base_cell( + conn: sqlite3.Connection, + cell_type: str = "unknown", +) -> ContractReport: + """ + Validate any flex cell against the base (non-coding-agent) contract. + + Checks only the generic tables that all flex modules produce. + Use this for modules with substrate='base'. + """ + n_sources = conn.execute("SELECT COUNT(*) FROM _raw_sources").fetchone()[0] + report = ContractReport(cell_type=cell_type, n_sources=n_sources) + + existing = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master " + "WHERE type = 'table' AND name NOT LIKE 'sqlite_%'" + ) + } + for tbl in REQUIRED_BASE_TABLES: + if tbl not in existing: + report.violations.append(ContractViolation( + severity="error", + table=tbl, + message="required base table missing", + )) + + return report + diff --git a/flex/watcher.py b/flex/watcher.py new file mode 100644 index 0000000..a6e05b8 --- /dev/null +++ b/flex/watcher.py @@ -0,0 +1,161 @@ +""" +Event-driven file watcher for flex daemon. + +Wraps watchdog.Observer to provide inotify-based file change detection +with debouncing. The observer thread records which files changed; the +main daemon thread calls drain() to collect debounced paths and do +the actual sync work (SQLite, embedding). + +Graceful degradation: if watchdog is not installed or inotify fails, +the daemon falls back to its existing polling loop. +""" + +import fnmatch +import os +import sys +import threading +import time +from pathlib import Path + +try: + from watchdog.observers import Observer + from watchdog.events import FileSystemEventHandler, FileModifiedEvent, FileCreatedEvent + + _WATCHDOG_AVAILABLE = True +except ImportError: + _WATCHDOG_AVAILABLE = False + + +class _FlexHandler(FileSystemEventHandler): + """Filters file events by pattern and records them for debounced drain.""" + + def __init__(self, pattern: str, pending: dict, lock: threading.Lock): + super().__init__() + self._pattern = pattern + self._pending = pending + self._lock = lock + + def _handle(self, event): + if event.is_directory: + return + src = event.src_path + if Path(src).is_symlink(): + return + if not fnmatch.fnmatch(Path(src).name, self._pattern): + return + with self._lock: + # Only record the earliest event time per path (debounce window + # starts from first event, not last — prevents starvation during + # sustained writes). + if src not in self._pending: + self._pending[src] = time.monotonic() + + def on_modified(self, event): + self._handle(event) + + def on_created(self, event): + self._handle(event) + + +class FlexWatcher: + """inotify-based file watcher with debounced drain. + + Usage: + watcher = FlexWatcher(debounce_ms=500) + watcher.watch("~/.claude/projects", pattern="*.jsonl") + watcher.start() + + # In daemon loop: + for path in watcher.drain(): + sync_session(path) + + watcher.stop() + """ + + def __init__(self, debounce_ms: int = 500): + if not _WATCHDOG_AVAILABLE: + raise ImportError("watchdog is not installed") + self._observer = Observer() + self._pending: dict[str, float] = {} + self._lock = threading.Lock() + self._debounce = debounce_ms / 1000.0 + self._active = False + self._watches: list[str] = [] + + def watch(self, path: str | Path, pattern: str = "*", recursive: bool = True): + """Register a directory for inotify monitoring. + + Args: + path: Directory to watch. + pattern: Filename glob pattern (e.g. "*.jsonl"). + recursive: Watch subdirectories. + """ + resolved = str(Path(path).expanduser().resolve()) + if not Path(resolved).is_dir(): + print(f"[watcher] Skipping non-directory: {resolved}", file=sys.stderr) + return + handler = _FlexHandler(pattern, self._pending, self._lock) + self._observer.schedule(handler, resolved, recursive=recursive) + self._watches.append(f"{resolved} ({pattern})") + + def start(self) -> bool: + """Start the observer thread. Returns True on success, False on failure.""" + if not self._watches: + return False + try: + self._observer.start() + self._active = True + return True + except Exception as e: + print(f"[watcher] Failed to start: {e}", file=sys.stderr) + self._active = False + return False + + def stop(self): + """Stop and join the observer thread.""" + if self._active: + try: + self._observer.stop() + self._observer.join(timeout=5) + except Exception: + pass + self._active = False + + def drain(self) -> list[Path]: + """Return paths whose debounce window has elapsed, clear them from buffer. + + Thread-safe. Called from the main daemon thread every tick. + """ + if not self._pending: + return [] + + now = time.monotonic() + ready = [] + with self._lock: + expired = [ + path for path, ts in self._pending.items() + if now - ts >= self._debounce + ] + for path in expired: + del self._pending[path] + ready.append(Path(path)) + return ready + + @property + def active(self) -> bool: + return self._active + + @property + def watch_count(self) -> int: + return len(self._watches) + + def status(self) -> dict: + """Return watcher status for diagnostics.""" + with self._lock: + pending_count = len(self._pending) + return { + "active": self._active, + "watches": self._watches, + "pending": pending_count, + "debounce_ms": int(self._debounce * 1000), + } diff --git a/pyproject.toml b/pyproject.toml index 5c5ba35..dd4bafb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ ] [project.optional-dependencies] +watch = ["watchdog>=4.0,<7"] [project.urls] Homepage = "https://getflex.dev"