diff --git a/open_terminal/main.py b/open_terminal/main.py index 9986dbd..d6c1714 100644 --- a/open_terminal/main.py +++ b/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) diff --git a/open_terminal/utils/events.py b/open_terminal/utils/events.py new file mode 100644 index 0000000..21bb4a6 --- /dev/null +++ b/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() diff --git a/pyproject.toml b/pyproject.toml index 7d52dd7..0f91dc5 100644 --- a/pyproject.toml +++ b/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",