"""FsEvent dataclass + FsEventType enum + from_watchdog parser. The canonical event shape every consumer of the fs_watcher library sees. Frozen, JSON-serializable, parseable from a watchdog FileSystemEvent. """ from __future__ import annotations import json import os from dataclasses import asdict, dataclass from enum import Enum from pathlib import Path from typing import Optional _ASSET_CLASS_TO_TYPE = {"char": "character", "loc": "location", "prop": "prop"} class FsEventType(str, Enum): """The six canonical filesystem event types emitted by the broker.""" CREATED = "created" MODIFIED = "modified" DELETED = "deleted" MOVED = "moved" # rename within the tree INGESTED = "ingested" # synthetic — emitted after external Finder drop HEARTBEAT = "heartbeat" # every 15s, keeps clients alive _WATCHDOG_TO_FS_TYPE = { "created": FsEventType.CREATED, "modified": FsEventType.MODIFIED, "deleted": FsEventType.DELETED, "moved": FsEventType.MOVED, } @dataclass(frozen=True) class FsEvent: """Canonical filesystem event. Every consumer sees this shape. Fields: event_id: monotonic id assigned by the broker event_type: one of the six FsEventType values path: POSIX path relative to the repo root (parent of projects_root()). For MOVED events, this is the DESTINATION path. project: parsed from the path if under projects// asset_type: "character" | "location" | "prop" if under output/refs/_canonical/ asset_id: slug if resolvable from canonical refs path src_path: old path for MOVED events, None otherwise is_directory: whether the path refers to a directory size_bytes: file size in bytes (None for deleted or directories) sha256: always None at emit time — computed lazily by consumers mtime: file mtime as unix seconds (None for deleted) ts: broker assignment timestamp, unix seconds """ event_id: str event_type: FsEventType path: str project: Optional[str] asset_type: Optional[str] asset_id: Optional[str] src_path: Optional[str] is_directory: bool size_bytes: Optional[int] sha256: Optional[str] mtime: Optional[float] ts: float def to_json(self) -> str: """Serialize to JSON with the enum as its string value.""" d = asdict(self) d["event_type"] = self.event_type.value return json.dumps(d, separators=(",", ":")) @classmethod def from_watchdog(cls, wd_event, broker_ts: float, event_id: str) -> "FsEvent": """Parse a watchdog FileSystemEvent into an FsEvent. Extracts project, asset_type, and asset_id from the path if it matches the pattern projects//output/refs/_canonical/s//... For MOVED events, `path` is the destination and `src_path` is the source. For DELETED events, size_bytes and mtime are None because the file no longer exists. For directories, size_bytes is None. """ # Import lazily to avoid circular imports and honor monkeypatching in tests. # # IMPORTANT: try/except fallback is REQUIRED here. When review_server.py # runs the FsWatcher at module-init time, its sys.path has only # recoil/pipeline/ — NOT recoil/ — so the bare `from core.paths` import # raises ModuleNotFoundError on every real filesystem event. Tests work # because pytest auto-adds the rootdir to sys.path, but production # runtime does not. lib.constants is a pipeline-local proxy that # re-exports projects_root() from the same source of truth. try: from recoil.core.paths import projects_root except ImportError: from recoil.core.paths import projects_root repo_root = Path(projects_root()).parent # ~/CLAUDE_PROJECTS event_type = _WATCHDOG_TO_FS_TYPE.get( getattr(wd_event, "event_type", "modified"), FsEventType.MODIFIED ) # Path resolution — for MOVED events, dest_path is the new location primary_path_str = getattr(wd_event, "src_path", "") src_path_field: Optional[str] = None if event_type == FsEventType.MOVED and hasattr(wd_event, "dest_path"): # src_path holds the old location for moves try: src_rel = Path(wd_event.src_path).resolve().relative_to(repo_root).as_posix() except (ValueError, OSError): src_rel = str(wd_event.src_path) src_path_field = src_rel # path field holds the new location primary_path_str = wd_event.dest_path try: abs_primary = Path(primary_path_str).resolve() rel_path = abs_primary.relative_to(repo_root).as_posix() except (ValueError, OSError): # Path is outside repo_root — fall back to absolute (still a POSIX-ish string) rel_path = str(primary_path_str).replace(os.sep, "/") # Extract project from "projects//..." prefix project: Optional[str] = None parts = tuple(Path(rel_path).parts) if len(parts) >= 2 and parts[0] == "projects": project = parts[1] # Extract asset_type and asset_id from canonical refs path or assets/ path asset_type: Optional[str] = None asset_id: Optional[str] = None if "_canonical" in parts: canonical_idx = parts.index("_canonical") if len(parts) > canonical_idx + 2: type_plural = parts[canonical_idx + 1] if type_plural in ("characters", "locations", "props"): asset_type = type_plural[:-1] # strip 's' for singular asset_id = parts[canonical_idx + 2] if asset_type is None and "assets" in parts: assets_idx = parts.index("assets") if len(parts) > assets_idx + 2: asset_class = parts[assets_idx + 1] if asset_class in _ASSET_CLASS_TO_TYPE: asset_type = _ASSET_CLASS_TO_TYPE[asset_class] asset_id = parts[assets_idx + 2] # Stat for size and mtime (deleted files won't exist) size_bytes: Optional[int] = None mtime: Optional[float] = None if event_type != FsEventType.DELETED: try: st = os.stat(primary_path_str) if not getattr(wd_event, "is_directory", False): size_bytes = st.st_size mtime = st.st_mtime except (OSError, FileNotFoundError): pass return cls( event_id=event_id, event_type=event_type, path=rel_path, project=project, asset_type=asset_type, asset_id=asset_id, src_path=src_path_field, is_directory=bool(getattr(wd_event, "is_directory", False)), size_bytes=size_bytes, sha256=None, mtime=mtime, ts=broker_ts, )