7bda83998e
The stage-transition payload {"state":"loading","payload":{"stage":"mmproj_model"}}
lacks the stages/current/value keys — accessing them raised KeyError which crashed
the LogWatcher task, freezing the UI at the last good state indefinitely.
- log_parser: skip loading payloads missing stages/current/value (next message
has the full payload within milliseconds)
- log_watcher: wrap on_line calls in try/except so any future parser error
keeps the tail loop alive instead of killing the task
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Callable, Awaitable, Optional
|
|
|
|
|
|
class LogWatcher:
|
|
POLL_INTERVAL = 0.1
|
|
FILE_RETRY_INTERVAL = 1.0
|
|
SEED_BYTES = 8192 # read last ~8 KB on open to seed initial state
|
|
|
|
def __init__(
|
|
self,
|
|
path: str,
|
|
on_line: Callable[[str], Awaitable[None]],
|
|
on_offline: Optional[Callable[[], Awaitable[None]]] = None,
|
|
offline_timeout: float = 60.0,
|
|
reopen: bool = False,
|
|
):
|
|
self._path = Path(path)
|
|
self._on_line = on_line
|
|
self._on_offline = on_offline
|
|
self._offline_timeout = offline_timeout
|
|
self._reopen = reopen
|
|
self._task: Optional[asyncio.Task] = None
|
|
|
|
def start(self) -> asyncio.Task:
|
|
self._task = asyncio.create_task(self._run())
|
|
return self._task
|
|
|
|
def stop(self):
|
|
if self._task:
|
|
self._task.cancel()
|
|
|
|
async def _run(self):
|
|
offline_since: Optional[float] = None
|
|
|
|
if self._reopen:
|
|
await self._run_reopen()
|
|
else:
|
|
await self._run_keepopen()
|
|
|
|
async def _run_keepopen(self):
|
|
"""Default: keep file open and poll readline(). Works on local filesystems."""
|
|
offline_since: Optional[float] = None
|
|
|
|
while True:
|
|
try:
|
|
with open(self._path, 'r', errors='replace') as f:
|
|
f.seek(0, 2)
|
|
end = f.tell()
|
|
f.seek(max(0, end - self.SEED_BYTES))
|
|
if f.tell() > 0:
|
|
f.readline() # discard partial first line
|
|
offline_since = None
|
|
while True:
|
|
line = f.readline()
|
|
if line:
|
|
try:
|
|
await self._on_line(line)
|
|
except Exception:
|
|
pass # keep tailing even if a line causes a parser error
|
|
else:
|
|
await asyncio.sleep(self.POLL_INTERVAL)
|
|
except OSError:
|
|
now = asyncio.get_event_loop().time()
|
|
if offline_since is None:
|
|
offline_since = now
|
|
elif now - offline_since >= self._offline_timeout and self._on_offline:
|
|
await self._on_offline()
|
|
offline_since = None
|
|
await asyncio.sleep(min(self.FILE_RETRY_INTERVAL, self._offline_timeout))
|
|
except asyncio.CancelledError:
|
|
return
|
|
|
|
async def _run_reopen(self):
|
|
"""Close and reopen each poll cycle. Use for network filesystems (e.g. sshfs)
|
|
that cache file content at open time and don't propagate remote appends to
|
|
held-open file descriptors."""
|
|
offline_since: Optional[float] = None
|
|
pos: int = -1 # -1 = not yet seeded
|
|
|
|
while True:
|
|
try:
|
|
with open(self._path, 'r', errors='replace') as f:
|
|
f.seek(0, 2)
|
|
end = f.tell()
|
|
|
|
if pos < 0:
|
|
f.seek(max(0, end - self.SEED_BYTES))
|
|
if f.tell() > 0:
|
|
f.readline() # discard partial first line
|
|
pos = f.tell()
|
|
elif end < pos:
|
|
pos = 0 # file truncated / rotated
|
|
|
|
f.seek(pos)
|
|
while True:
|
|
line = f.readline()
|
|
if line:
|
|
pos = f.tell()
|
|
try:
|
|
await self._on_line(line)
|
|
except Exception:
|
|
pass # keep tailing even if a line causes a parser error
|
|
else:
|
|
break
|
|
|
|
offline_since = None
|
|
await asyncio.sleep(self.POLL_INTERVAL)
|
|
except OSError:
|
|
now = asyncio.get_event_loop().time()
|
|
if offline_since is None:
|
|
offline_since = now
|
|
elif now - offline_since >= self._offline_timeout and self._on_offline:
|
|
await self._on_offline()
|
|
offline_since = None
|
|
await asyncio.sleep(min(self.FILE_RETRY_INTERVAL, self._offline_timeout))
|
|
except asyncio.CancelledError:
|
|
return
|