fix(tools): keep memory tool available when fcntl is unavailable

This commit is contained in:
Dusk1e 2026-04-14 14:35:06 +03:00 committed by Teknium
parent 449c17e9a9
commit 420d27098f
2 changed files with 63 additions and 4 deletions

View file

@ -0,0 +1,31 @@
"""Regression tests for memory-tool import fallbacks."""
import builtins
import importlib
import sys
from tools.registry import registry
def test_memory_tool_imports_without_fcntl(monkeypatch, tmp_path):
original_import = builtins.__import__
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == "fcntl":
raise ImportError("simulated missing fcntl")
return original_import(name, globals, locals, fromlist, level)
registry.deregister("memory")
monkeypatch.delitem(sys.modules, "tools.memory_tool", raising=False)
monkeypatch.setattr(builtins, "__import__", fake_import)
memory_tool = importlib.import_module("tools.memory_tool")
monkeypatch.setattr(memory_tool, "get_memory_dir", lambda: tmp_path)
store = memory_tool.MemoryStore(memory_char_limit=200, user_char_limit=200)
store.load_from_disk()
result = store.add("memory", "fact learned during import fallback test")
assert memory_tool.fcntl is None
assert registry.get_entry("memory") is not None
assert result["success"] is True

View file

@ -23,7 +23,6 @@ Design:
- Frozen snapshot pattern: system prompt is stable, tool responses show live state
"""
import fcntl
import json
import logging
import os
@ -34,6 +33,16 @@ from pathlib import Path
from hermes_constants import get_hermes_home
from typing import Dict, Any, List, Optional
try:
import fcntl
except ImportError:
fcntl = None
try:
import msvcrt
except ImportError:
msvcrt = None
logger = logging.getLogger(__name__)
# Where memory files live — resolved dynamically so profile overrides
@ -139,12 +148,31 @@ class MemoryStore:
"""
lock_path = path.with_suffix(path.suffix + ".lock")
lock_path.parent.mkdir(parents=True, exist_ok=True)
fd = open(lock_path, "w")
if fcntl is None and msvcrt is None:
yield
return
if msvcrt and (not lock_path.exists() or lock_path.stat().st_size == 0):
lock_path.write_text(" ", encoding="utf-8")
fd = open(lock_path, "r+" if msvcrt else "a+")
try:
fcntl.flock(fd, fcntl.LOCK_EX)
if fcntl:
fcntl.flock(fd, fcntl.LOCK_EX)
else:
fd.seek(0)
msvcrt.locking(fd.fileno(), msvcrt.LK_LOCK, 1)
yield
finally:
fcntl.flock(fd, fcntl.LOCK_UN)
if fcntl:
fcntl.flock(fd, fcntl.LOCK_UN)
elif msvcrt:
try:
fd.seek(0)
msvcrt.locking(fd.fileno(), msvcrt.LK_UNLCK, 1)
except (OSError, IOError):
pass
fd.close()
@staticmethod