refac

open_terminal/main.py

@@ -39,7 +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.events import system_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
 
@@ -354,9 +354,23 @@ class BackgroundProcess:
     log_path: Optional[str] = field(default=None, repr=False)
     user_id: str = ""
     session_id: str = ""
+    changed: asyncio.Event = field(default_factory=asyncio.Event, repr=False)
+
+    def notify_output(self):
+        changed, self.changed = self.changed, asyncio.Event()
+        changed.set()
 
 
 _processes: dict[str, BackgroundProcess] = {}
+_process_list_changed = asyncio.Event()
+
+
+def _notify_process_list():
+    global _process_list_changed
+    changed, _process_list_changed = _process_list_changed, asyncio.Event()
+    changed.set()
+
+
 _EXPIRY_SECONDS = 300  # auto-clean finished processes after 5 min
 
 
@@ -396,7 +410,7 @@ def _set_session_cwd(session_id: str | None, path: str, fs: "UserFS"):
         _session_cwds[(fs.home, session_id)] = (path, time.time())
 
 
-from open_terminal.utils.log import log_process, log_response
+from open_terminal.utils.log import log_process, log_response, read_log
 
 
 
@@ -407,6 +421,7 @@ def _cleanup_expired():
     for process in finished:
         if now - process.finished_at > _EXPIRY_SECONDS:
             _processes.pop(process.id, None)
+            _notify_process_list()
 
 
 def _cleanup_log_files(active_paths):
@@ -484,6 +499,9 @@ async def get_config():
             "terminal": ENABLE_TERMINAL,
             "notebooks": ENABLE_NOTEBOOKS,
             "system": ENABLE_SYSTEM_PROMPT,
+            "git": shutil.which("git") is not None,
+            "events": True,
+            "execute_stream": True,
         },
     }
 
@@ -672,8 +690,30 @@ 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):
+@app.websocket("/events")
+async def ws_events(ws: WebSocket):
+    payload = await _authenticate_websocket(ws)
+    if payload is None:
+        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", ""))
+        async def ports():
+            previous = None
+            async for snapshot in watch_listening_ports():
+                visible = await _visible_ports(request, snapshot)
+                visible = [{k: v for k, v in port.items() if k != "uid"} for port in visible]
+                if visible != previous:
+                    yield visible
+                    previous = visible
+
+        await system_events(ws, fs, _get_session_cwd(session_id, fs), ports)
+    except HTTPException:
+        await ws.close(code=4003, reason="Filesystem access denied")
+
+
+async def _authenticate_websocket(ws: WebSocket):
     await ws.accept()
     try:
         payload = await asyncio.wait_for(ws.receive_json(), timeout=10)
@@ -685,16 +725,10 @@ async def ws_filesystem_events(ws: WebSocket):
     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")
+    if not isinstance(ws.headers.get("x-session-id", payload.get("chat_id", "")), str):
+        await ws.close(code=4000, reason="Invalid session")
+        return
+    return payload
 
 
 @app.get("/git/status", dependencies=[Depends(verify_api_key)], include_in_schema=False)
@@ -1616,6 +1650,96 @@ async def list_processes(http_request: Request):
     ]
 
 
+async def _stream_websocket(ws, send):
+    """Stop delivery on disconnect without stopping the underlying command."""
+    async def receive():
+        while (await ws.receive())["type"] != "websocket.disconnect":
+            pass
+
+    tasks = [asyncio.create_task(send()), asyncio.create_task(receive())]
+    close_code = 1000
+    try:
+        done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+        for task in done:
+            task.result()
+    except WebSocketDisconnect:
+        pass
+    except Exception:
+        close_code = 1011
+        log.exception("Command stream failed")
+    finally:
+        import anyio
+        with anyio.CancelScope(shield=True):
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+        try:
+            await ws.close(code=close_code)
+        except (WebSocketDisconnect, RuntimeError):
+            pass
+
+
+async def _command_socket_request(ws):
+    payload = await _authenticate_websocket(ws)
+    if payload is None:
+        return None
+    headers = list(ws.scope["headers"])
+    if "x-session-id" not in ws.headers:
+        headers.append((b"x-session-id", payload.get("chat_id", "").encode()))
+    return Request({**ws.scope, "type": "http", "method": "GET", "headers": headers})
+
+
+@app.websocket("/execute")
+async def stream_processes(ws: WebSocket):
+    request = await _command_socket_request(ws)
+    if request is None:
+        return
+
+    async def send():
+        previous = None
+        while True:
+            changed = _process_list_changed
+            processes = await list_processes(request)
+            if processes != previous:
+                await ws.send_json(processes)
+                previous = processes
+            await changed.wait()
+
+    await _stream_websocket(ws, send)
+
+
+@app.websocket("/execute/{process_id}/status")
+async def stream_process_output(ws: WebSocket, process_id: str,
+                                offset: int = Query(0, ge=0),
+                                tail: int = Query(1000, ge=1, le=1000)):
+    request = await _command_socket_request(ws)
+    if request is None:
+        return
+    try:
+        process = _get_process(process_id, request)
+    except HTTPException:
+        await ws.close(code=4004, reason="Process not found")
+        return
+
+    async def send():
+        cursor = offset
+        while True:
+            changed = process.changed
+            finished = process.finished_at is not None
+            output, next_offset, truncated = await read_log(process.log_path, cursor, tail)
+            await ws.send_json({"id": process.id, "command": process.command,
+                                "status": process.status if finished else "running",
+                                "exit_code": process.exit_code if finished else None,
+                                "output": output, "next_offset": next_offset,
+                                "truncated": truncated})
+            cursor = next_offset
+            if finished:
+                return
+            await changed.wait()
+
+    await _stream_websocket(ws, send)
+
+
 @app.post(
     "/execute",
     operation_id="run_command",
@@ -1656,8 +1780,9 @@ async def execute(
         user_id=http_request.headers.get("x-user-id", ""),
         session_id=http_request.headers.get("x-session-id", ""),
     )
-    background_process.log_task = asyncio.create_task(log_process(background_process))
+    background_process.log_task = asyncio.create_task(log_process(background_process, on_finish=_notify_process_list))
     _processes[process_id] = background_process
+    _notify_process_list()
 
     if wait is None and EXECUTE_TIMEOUT:
         wait = EXECUTE_TIMEOUT
@@ -1796,11 +1921,12 @@ async def kill_process(
 # Port detection & proxy
 # ---------------------------------------------------------------------------
 
-from open_terminal.utils.port import detect_listening_ports, get_descendant_pids
+from open_terminal.utils.port import detect_listening_ports, get_descendant_pids, watch_listening_ports
 
 
-async def _visible_ports(request: Request) -> list[dict]:
-    all_ports = await run_io(detect_listening_ports)
+async def _visible_ports(request: Request, all_ports=None) -> list[dict]:
+    if all_ports is None:
+        all_ports = await run_io(detect_listening_ports)
 
     try:
         fs = get_filesystem(request)
@@ -2241,17 +2367,8 @@ if ENABLE_TERMINAL:
             await ws.close(code=4004, reason="Session has ended")
             return
 
-        await ws.accept()
-
-        # First-message authentication
-        try:
-            msg = await asyncio.wait_for(ws.receive_text(), timeout=10.0)
-            payload = json.loads(msg)
-            if not isinstance(payload, dict) or payload.get("type") != "auth" or (API_KEY and not matches_api_key(payload.get("token"))):
-                await ws.close(code=4001, reason="Invalid API key")
-                return
-        except (asyncio.TimeoutError, ValueError, AttributeError, KeyError, WebSocketDisconnect):
-            await ws.close(code=4001, reason="Auth timeout or invalid payload")
+        payload = await _authenticate_websocket(ws)
+        if payload is None:
             return
         if (
             session["user_id"] != ws.headers.get("x-user-id", "")

open_terminal/utils/events.py

@@ -1,7 +1,8 @@
-"""Native filesystem notifications for the file browser and Git status."""
+"""Filesystem and port subscriptions for the shared events endpoint."""
 
 import asyncio
 import os
+import shutil
 import threading
 
 import anyio
@@ -14,7 +15,7 @@ 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():
+    if shutil.which("git") is not None and 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"):
@@ -23,8 +24,9 @@ async def watch_paths(fs, path, cwd):
     return sorted(paths)
 
 
-async def filesystem_events(ws: WebSocket, fs, cwd):
+async def system_events(ws: WebSocket, fs, cwd, ports):
     task = None
+    port_task = None
     stop = threading.Event()
 
     async def stop_watcher():
@@ -35,6 +37,13 @@ async def filesystem_events(ws: WebSocket, fs, cwd):
                 task.cancel()
                 await asyncio.gather(task, return_exceptions=True)
 
+    async def send_ports():
+        try:
+            async for snapshot in ports():
+                await ws.send_json({"type": "ports:change", "ports": snapshot})
+        except Exception:
+            await ws.close(code=1011, reason="Port discovery unavailable")
+
     async def send_changes(paths):
         ready = False
         try:
@@ -47,27 +56,37 @@ async def filesystem_events(ws: WebSocket, fs, cwd):
             ):
                 if not ready:
                     ready = True
-                    await ws.send_json({"type": "ready"})
+                    await ws.send_json({"type": "fs:ready"})
                 elif changes:
                     # No filenames are exposed; clients only need to invalidate status.
-                    await ws.send_json({"type": "change"})
+                    await ws.send_json({"type": "fs: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)
+            payload = await asyncio.wait_for(ws.receive_json(), timeout=10 if task is None and port_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")
+            include_ports = payload.get("ports", False)
+            if (type(include_ports) is not bool or
+                path is not None and (not isinstance(path, str) or not path or len(path) > 4096)):
+                await ws.close(code=4000, reason="Invalid subscription")
                 return
-            paths = await watch_paths(fs, path, cwd)
+            paths = await watch_paths(fs, path, cwd) if path is not None else None
             await stop_watcher()
+            task = None
             stop = threading.Event()
-            task = asyncio.create_task(send_changes(paths))
+            if paths is not None:
+                task = asyncio.create_task(send_changes(paths))
+            if port_task:
+                with anyio.CancelScope(shield=True):
+                    port_task.cancel()
+                    await asyncio.gather(port_task, return_exceptions=True)
+            port_task = asyncio.create_task(send_ports()) if include_ports else None
+
     except WebSocketDisconnect:
         pass
     except (HTTPException, PermissionError):
@@ -76,3 +95,7 @@ async def filesystem_events(ws: WebSocket, fs, cwd):
         await ws.close(code=4000, reason="Invalid watch request")
     finally:
         await stop_watcher()
+        if port_task:
+            with anyio.CancelScope(shield=True):
+                port_task.cancel()
+                await asyncio.gather(port_task, return_exceptions=True)

open_terminal/utils/log.py

@@ -56,7 +56,7 @@ def _rotate_log(path: str) -> int:
 class BoundedLogWriter:
     """Rotate on disk; disable logging on I/O failure while draining the child."""
 
-    def __init__(self, file, log_path: str, *, flush_interval: float = 0, flush_buffer: int = 0):
+    def __init__(self, file, log_path: str, *, flush_interval: float = 0, flush_buffer: int = 0, on_flush=None):
         self._file = file
         self._log_path = log_path
         self._bytes_written = 0
@@ -66,6 +66,8 @@ class BoundedLogWriter:
         self._unflushed = 0
         self._last_flush = time.monotonic()
         self._lock = asyncio.Lock()
+        self._on_flush = on_flush
+        self._flush_task = None
 
     async def _disable(self, exc):
         log.warning("Disabling process log %s: %s", self._log_path, exc)
@@ -86,19 +88,29 @@ class BoundedLogWriter:
                     or time.monotonic() - self._last_flush >= self._flush_interval
                     or self._flush_buffer > 0 and self._unflushed >= self._flush_buffer):
                     await self.flush()
+                elif self._flush_task is None:
+                    self._flush_task = asyncio.create_task(self._flush_pending())
             except (OSError, ValueError) as exc:
                 await self._disable(exc)
 
     async def flush(self) -> None:
+        task, self._flush_task = self._flush_task, None
+        if task and task is not asyncio.current_task():
+            task.cancel()
         if self._file is not None:
             try:
                 await self._file.flush()
                 self._unflushed = 0
                 self._last_flush = time.monotonic()
+                if self._on_flush:
+                    self._on_flush()
             except (OSError, ValueError) as exc:
                 await self._disable(exc)
 
     async def close(self):
+        if self._flush_task:
+            self._flush_task.cancel()
+            self._flush_task = None
         file, self._file = self._file, None
         if file is not None:
             try:
@@ -106,6 +118,12 @@ class BoundedLogWriter:
             except (OSError, ValueError):
                 pass
 
+    async def _flush_pending(self):
+        # A quiet command still needs to publish its last buffered chunk.
+        await asyncio.sleep(self._flush_interval)
+        async with self._lock:
+            await self.flush()
+
     async def _rotate(self):
         await self._file.flush()
         await self._file.close()
@@ -115,7 +133,7 @@ class BoundedLogWriter:
         self.rotated = True
 
 
-async def log_process(background_process) -> None:
+async def log_process(background_process, on_finish=None) -> None:
     writer = None
     runner = background_process.runner
     try:
@@ -124,7 +142,8 @@ async def log_process(background_process) -> None:
                 await aiofiles.os.makedirs(os.path.dirname(background_process.log_path), exist_ok=True)
                 file = await aiofiles.open(background_process.log_path, "a", encoding="utf-8")
                 writer = BoundedLogWriter(file, background_process.log_path,
-                    flush_interval=LOG_FLUSH_INTERVAL, flush_buffer=LOG_FLUSH_BUFFER)
+                    flush_interval=LOG_FLUSH_INTERVAL, flush_buffer=LOG_FLUSH_BUFFER,
+                    on_flush=background_process.notify_output)
                 await file.write(json.dumps({"type": "start", "command": background_process.command, "pid": runner.pid, "ts": time.time()}) + "\n")
                 await writer.flush()
             except OSError as exc:
@@ -145,13 +164,16 @@ async def log_process(background_process) -> None:
             runner.close()
             if background_process.status == "running":
                 background_process.status = "done"
-            background_process.finished_at = time.time()
             if writer:
                 try:
                     await writer.write(json.dumps({"type": "end", "exit_code": background_process.exit_code, "log_rotated": writer.rotated, "ts": time.time()}) + "\n")
                     await writer.flush()
                 finally:
                     await writer.close()
+            background_process.finished_at = time.time()
+            background_process.notify_output()
+            if on_finish:
+                on_finish()
 
 
 def _read_log(path, offset, tail):

open_terminal/utils/port.py

@@ -2,6 +2,60 @@
 
 import os
 import platform
+import asyncio
+
+import anyio
+
+from open_terminal.utils.limits import run_io
+
+
+_subscribers: set[asyncio.Queue] = set()
+_scanner: asyncio.Task | None = None
+_snapshot: list[dict] | None = None
+
+
+async def watch_listening_ports():
+    """Share one OS scan across viewers; publish only changed snapshots."""
+    global _scanner, _snapshot
+    queue = asyncio.Queue(maxsize=1)
+    _subscribers.add(queue)
+    if _snapshot is not None:
+        queue.put_nowait(_snapshot)
+
+    async def scan():
+        global _snapshot
+        try:
+            while True:
+                ports = await run_io(detect_listening_ports)
+                if ports != _snapshot:
+                    _snapshot = ports
+                    for subscriber in _subscribers:
+                        if subscriber.full():
+                            subscriber.get_nowait()
+                        subscriber.put_nowait(ports)
+                await asyncio.sleep(5)
+        except Exception as exc:
+            for subscriber in _subscribers:
+                if subscriber.full():
+                    subscriber.get_nowait()
+                subscriber.put_nowait(exc)
+
+    if _scanner is None:
+        _scanner = asyncio.create_task(scan())
+    try:
+        while True:
+            snapshot = await queue.get()
+            if isinstance(snapshot, Exception):
+                raise snapshot
+            yield snapshot
+    finally:
+        _subscribers.discard(queue)
+        if not _subscribers:
+            scanner, _scanner = _scanner, None
+            _snapshot = None
+            with anyio.CancelScope(shield=True):
+                scanner.cancel()
+                await asyncio.gather(scanner, return_exceptions=True)
 
 
 def detect_listening_ports() -> list[dict]: