mirror of
https://github.com/nesquena/hermes-webui.git
synced 2026-07-16 12:40:18 +00:00
fix: handle Windows media and turn journal edge cases
This commit is contained in:
@@ -3,6 +3,11 @@
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Fixed
|
||||
|
||||
- Windows native runs now skip the POSIX-only turn-journal directory fsync instead of raising `AttributeError` for missing `os.O_DIRECTORY` on every submitted turn (#3170).
|
||||
- `/api/media` now treats Windows cross-drive `commonpath()` comparisons as non-matches instead of 500ing when media paths and allowed roots live on different drives (#3171).
|
||||
|
||||
## [v0.51.171] — 2026-05-30 — Release EQ (stage-batch53 — tool-output card badge + Neon opt-in skin)
|
||||
|
||||
### Added
|
||||
|
||||
+9
-1
@@ -7887,6 +7887,14 @@ def _session_media_token_allows_image_path(sid: str, target: Path, image_mimes:
|
||||
return False
|
||||
|
||||
|
||||
def _path_is_within_root(child: Path, root: Path) -> bool:
|
||||
"""Return True when ``child`` is inside ``root`` without crashing on Windows drives."""
|
||||
try:
|
||||
return os.path.commonpath([str(child), str(root)]) == str(root)
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def _handle_media(handler, parsed):
|
||||
"""Serve a local file by absolute path for inline display in the chat.
|
||||
|
||||
@@ -7963,7 +7971,7 @@ def _handle_media(handler, parsed):
|
||||
"image/x-icon", "image/bmp",
|
||||
}
|
||||
within_allowed = any(
|
||||
_os.path.commonpath([str(target), str(root)]) == str(root)
|
||||
_path_is_within_root(target, root)
|
||||
for root in allowed_roots
|
||||
if root.exists()
|
||||
)
|
||||
|
||||
+9
-7
@@ -94,14 +94,16 @@ def append_turn_journal_event(
|
||||
fh.write(line)
|
||||
fh.flush()
|
||||
os.fsync(fh.fileno())
|
||||
try:
|
||||
dir_fd = os.open(path.parent, os.O_DIRECTORY)
|
||||
o_directory = getattr(os, "O_DIRECTORY", None)
|
||||
if o_directory is not None:
|
||||
try:
|
||||
os.fsync(dir_fd)
|
||||
finally:
|
||||
os.close(dir_fd)
|
||||
except OSError:
|
||||
pass
|
||||
dir_fd = os.open(path.parent, o_directory)
|
||||
try:
|
||||
os.fsync(dir_fd)
|
||||
finally:
|
||||
os.close(dir_fd)
|
||||
except OSError:
|
||||
pass
|
||||
return payload
|
||||
|
||||
|
||||
|
||||
@@ -255,6 +255,32 @@ class TestMediaEndpointUnit(unittest.TestCase):
|
||||
self.assertIn(".split(_os.pathsep)", block)
|
||||
self.assertNotIn('.split(":")', block)
|
||||
|
||||
def test_path_is_within_root_treats_commonpath_valueerror_as_not_within(self):
|
||||
"""Windows cross-drive commonpath() errors must not crash /api/media."""
|
||||
from api import routes
|
||||
|
||||
with mock.patch.object(
|
||||
routes.os.path,
|
||||
"commonpath",
|
||||
side_effect=ValueError("Paths don't have the same drive"),
|
||||
):
|
||||
self.assertFalse(
|
||||
routes._path_is_within_root(
|
||||
pathlib.Path("D:/outputs/card.png"),
|
||||
pathlib.Path("C:/Users/agent/.hermes"),
|
||||
)
|
||||
)
|
||||
|
||||
def test_path_is_within_root_accepts_child_path(self):
|
||||
from api import routes
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpd:
|
||||
root = pathlib.Path(tmpd).resolve()
|
||||
child = root / "media" / "card.png"
|
||||
child.parent.mkdir()
|
||||
child.write_bytes(b"png")
|
||||
self.assertTrue(routes._path_is_within_root(child.resolve(), root))
|
||||
|
||||
def test_media_endpoints_advertise_byte_range_support(self):
|
||||
routes_src = (REPO_ROOT / "api" / "routes.py").read_text(encoding="utf-8")
|
||||
self.assertIn("Accept-Ranges", routes_src)
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import os
|
||||
|
||||
from api.turn_journal import (
|
||||
append_turn_journal_event,
|
||||
append_turn_journal_event_for_stream,
|
||||
@@ -36,3 +38,16 @@ def test_append_turn_journal_event_for_stream_falls_back_to_new_turn_for_missing
|
||||
assert event["stream_id"] == "stream-missing"
|
||||
assert event["turn_id"]
|
||||
assert event["event"] == "interrupted"
|
||||
|
||||
|
||||
def test_append_turn_journal_event_skips_directory_fsync_without_o_directory(tmp_path, monkeypatch):
|
||||
monkeypatch.delattr(os, "O_DIRECTORY", raising=False)
|
||||
|
||||
event = append_turn_journal_event(
|
||||
"sid-windows",
|
||||
{"event": "submitted", "content": "hello"},
|
||||
session_dir=tmp_path,
|
||||
)
|
||||
|
||||
assert event["event"] == "submitted"
|
||||
assert (tmp_path / "_turn_journal" / "sid-windows.jsonl").is_file()
|
||||
|
||||
Reference in New Issue
Block a user