refac

open_terminal/main.py

@@ -39,6 +39,7 @@ from open_terminal.utils.fs import UserFS, FullFileResponse as FileResponse, Tem
 from open_terminal.utils.limits import (CHUNK_SIZE, NormalizeNullQueryMiddleware, run_io, run_worker)
 from open_terminal.utils.file_compare import CompareRequest, run_comparison
 from open_terminal.utils.git import GitRepository
+from open_terminal.utils.events import filesystem_events
 from open_terminal.utils.search import search_entries, matching_files, ranked_page
 from open_terminal.utils.skills import join_skill_path, list_skill_resources, parse_skill_frontmatter
 
@@ -671,6 +672,31 @@ async def list_files(
     return {"dir": target, "writable": await fs.is_writable(target), "entries": entries}
 
 
+@app.websocket("/files/events")
+async def ws_filesystem_events(ws: WebSocket):
+    await ws.accept()
+    try:
+        payload = await asyncio.wait_for(ws.receive_json(), timeout=10)
+        if not isinstance(payload, dict) or payload.get("type") != "auth" or not matches_api_key(payload.get("token")):
+            await ws.close(code=4001, reason="Invalid API key")
+            return
+    except WebSocketDisconnect:
+        return
+    except (TimeoutError, ValueError, KeyError):
+        await ws.close(code=4001, reason="Auth timeout or invalid payload")
+        return
+    request = Request({**ws.scope, "type": "http", "method": "GET"})
+    try:
+        fs = await asyncio.to_thread(get_filesystem, request)
+        session_id = ws.headers.get("x-session-id", payload.get("chat_id", ""))
+        if not isinstance(session_id, str):
+            await ws.close(code=4000, reason="Invalid session")
+            return
+        await filesystem_events(ws, fs, _get_session_cwd(session_id, fs))
+    except HTTPException:
+        await ws.close(code=4003, reason="Filesystem access denied")
+
+
 @app.get("/git/status", dependencies=[Depends(verify_api_key)], include_in_schema=False)
 async def git_status(http_request: Request, root: str = ".", fs: UserFS = Depends(get_filesystem)):
     cwd = _get_session_cwd(http_request.headers.get("x-session-id"), fs)

open_terminal/utils/events.py

@@ -0,0 +1,78 @@
+"""Native filesystem notifications for the file browser and Git status."""
+
+import asyncio
+import os
+import threading
+
+import anyio
+from fastapi import HTTPException, WebSocket, WebSocketDisconnect
+from watchfiles import awatch
+
+from open_terminal.utils.git import GitRepository
+
+
+async def watch_paths(fs, path, cwd):
+    repo = GitRepository(fs, path, cwd)
+    paths = {repo.root}
+    if await repo.detect():
+        paths = {repo.root}
+        # A linked worktree keeps its index and shared refs outside its working tree.
+        for flag in ("--absolute-git-dir", "--git-common-dir"):
+            _, value, _ = await repo.run("rev-parse", flag)
+            paths.add(repo.checked_path(os.path.join(repo.root, value.removesuffix("\n"))))
+    return sorted(paths)
+
+
+async def filesystem_events(ws: WebSocket, fs, cwd):
+    task = None
+    stop = threading.Event()
+
+    async def stop_watcher():
+        if task:
+            stop.set()
+            # ASGI servers can cancel the handler while its native watcher is exiting.
+            with anyio.CancelScope(shield=True):
+                task.cancel()
+                await asyncio.gather(task, return_exceptions=True)
+
+    async def send_changes(paths):
+        ready = False
+        try:
+            # The first yield happens after the native watcher is installed. Refreshing
+            # then closes the race between the initial status request and subscribing.
+            async for changes in awatch(
+                *paths, watch_filter=None, debounce=250, step=50,
+                force_polling=False, yield_on_timeout=True, rust_timeout=1000,
+                stop_event=stop,
+            ):
+                if not ready:
+                    ready = True
+                    await ws.send_json({"type": "ready"})
+                elif changes:
+                    # No filenames are exposed; clients only need to invalidate status.
+                    await ws.send_json({"type": "change"})
+        except (OSError, RuntimeError):
+            await ws.close(code=1011, reason="Filesystem watcher unavailable")
+
+    try:
+        while True:
+            payload = await asyncio.wait_for(ws.receive_json(), timeout=10 if task is None else None)
+            if not isinstance(payload, dict) or payload.get("type") != "subscribe":
+                await ws.close(code=4000, reason="Expected subscribe message")
+                return
+            path = payload.get("path")
+            if not isinstance(path, str) or not path or len(path) > 4096:
+                await ws.close(code=4000, reason="Invalid watch path")
+                return
+            paths = await watch_paths(fs, path, cwd)
+            await stop_watcher()
+            stop = threading.Event()
+            task = asyncio.create_task(send_changes(paths))
+    except WebSocketDisconnect:
+        pass
+    except (HTTPException, PermissionError):
+        await ws.close(code=4003, reason="Watch path unavailable")
+    except (TimeoutError, ValueError, KeyError):
+        await ws.close(code=4000, reason="Invalid watch request")
+    finally:
+        await stop_watcher()

pyproject.toml

@@ -11,6 +11,7 @@ requires-python = ">=3.11"
 dependencies = [
     "fastapi>=0.115.0",
     "uvicorn[standard]>=0.34.0",
+    "watchfiles>=1.1.1",
     "click>=8.1.0",
     "httpx>=0.27.0",
     "python-multipart>=0.0.22",