fix(desktop): skip inaccessible logs during foreign session discovery
One unreadable or rotated log must not hide the rest of the list; discovery tolerates OSError from resolve()/stat() per entry and reuses the stat result.
This commit is contained in:
@@ -4,6 +4,7 @@ import hashlib
|
||||
import re
|
||||
import socket
|
||||
from pathlib import Path
|
||||
from stat import S_ISREG
|
||||
|
||||
from hermes_cli.foreign_sessions import _SOURCES, _SOURCE_LABELS, _SOURCE_DB_NAMES
|
||||
|
||||
@@ -30,14 +31,14 @@ def _candidates(source=None):
|
||||
if not root.is_dir():
|
||||
continue
|
||||
for path in root.rglob(pattern) if recursive else root.glob(pattern):
|
||||
resolved = path.resolve()
|
||||
if not resolved.is_relative_to(root):
|
||||
continue
|
||||
try:
|
||||
resolved = path.resolve()
|
||||
if not resolved.is_relative_to(root):
|
||||
continue
|
||||
stat = resolved.stat()
|
||||
except FileNotFoundError:
|
||||
continue # A running source may rotate a log during discovery.
|
||||
if not resolved.is_file():
|
||||
except OSError:
|
||||
continue # One inaccessible or rotated log must not hide the rest.
|
||||
if not S_ISREG(stat.st_mode):
|
||||
continue
|
||||
handle = hashlib.sha256(f"{name}:{resolved}".encode()).hexdigest()
|
||||
rows.append((stat.st_mtime, handle, name, resolved, stat.st_size))
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
"""Discovery keeps readable sessions available when a neighboring log is inaccessible."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.foreign_session_browser import list_foreign_sessions
|
||||
|
||||
|
||||
@pytest.mark.parametrize("operation", ["resolve", "stat"])
|
||||
def test_discovery_skips_inaccessible_log(tmp_path, monkeypatch, operation):
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
folder = tmp_path / ".codex" / "sessions"
|
||||
folder.mkdir(parents=True)
|
||||
for name in ("readable", "inaccessible"):
|
||||
(folder / f"rollout-{name}.jsonl").write_text(json.dumps({
|
||||
"type": "response_item", "payload": {"type": "message", "role": "user",
|
||||
"content": [{"type": "input_text", "text": name}]},
|
||||
}), encoding="utf-8")
|
||||
|
||||
original = getattr(Path, operation)
|
||||
|
||||
def access(path, *args, **kwargs):
|
||||
if path.name == "rollout-inaccessible.jsonl":
|
||||
raise PermissionError("Log access denied")
|
||||
return original(path, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(Path, operation, access)
|
||||
page = list_foreign_sessions("codex")
|
||||
assert [session["title"] for session in page["sessions"]] == ["readable"]
|
||||
Reference in New Issue
Block a user