mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-21 03:39:54 +00:00
vanthinh6886
62573f44cf
1. trajectory_compressor.py: yaml.safe_load() returns None on empty
files, crashing with TypeError on `if 'tokenizer' in data`. Fix by
adding `or {}` fallback. (HIGH — blocks startup with empty config)
2. 6 files with fcntl.flock(LOCK_UN) in finally blocks without
try/except: cron/scheduler.py, hermes_cli/auth.py,
agent/shell_hooks.py, tools/skill_usage.py,
tools/environments/file_sync.py, tools/memory_tool.py. If unlock
raises OSError, fd.close() is skipped and the lock is held forever.
The msvcrt branches already had try/except; the fcntl branches did
not. Fix by wrapping in try/except (OSError, IOError): pass.
3. agent/copilot_acp_client.py line 639: TOCTOU race — path.exists()
followed by path.read_text() with no try/except. If file is deleted
between the check and the read, FileNotFoundError propagates. Fix
by using try/except FileNotFoundError.
4. gateway/sticker_cache.py: non-atomic write via Path.write_text()
can leave truncated JSON on crash, causing JSONDecodeError on next
load. Fix by writing to tempfile + fsync + os.replace (atomic).
125 lines
3.4 KiB
Python
125 lines
3.4 KiB
Python
"""
|
|
Sticker description cache for Telegram.
|
|
|
|
When users send stickers, we describe them via the vision tool and cache
|
|
the descriptions keyed by file_unique_id so we don't re-analyze the same
|
|
sticker image on every send. Descriptions are concise (1-2 sentences).
|
|
|
|
Cache location: ~/.hermes/sticker_cache.json
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from typing import Optional
|
|
|
|
from hermes_cli.config import get_hermes_home
|
|
|
|
|
|
CACHE_PATH = get_hermes_home() / "sticker_cache.json"
|
|
|
|
# Vision prompt for describing stickers -- kept concise to save tokens
|
|
STICKER_VISION_PROMPT = (
|
|
"Describe this sticker in 1-2 sentences. Focus on what it depicts -- "
|
|
"character, action, emotion. Be concise and objective."
|
|
)
|
|
|
|
|
|
def _load_cache() -> dict:
|
|
"""Load the sticker cache from disk."""
|
|
if CACHE_PATH.exists():
|
|
try:
|
|
return json.loads(CACHE_PATH.read_text(encoding="utf-8"))
|
|
except (json.JSONDecodeError, OSError):
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def _save_cache(cache: dict) -> None:
|
|
"""Save the sticker cache to disk atomically."""
|
|
CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp_path = tempfile.mkstemp(
|
|
dir=str(CACHE_PATH.parent), suffix=".tmp"
|
|
)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
json.dump(cache, f, indent=2, ensure_ascii=False)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp_path, str(CACHE_PATH))
|
|
except BaseException:
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
def get_cached_description(file_unique_id: str) -> Optional[dict]:
|
|
"""
|
|
Look up a cached sticker description.
|
|
|
|
Returns:
|
|
dict with keys {description, emoji, set_name, cached_at} or None.
|
|
"""
|
|
cache = _load_cache()
|
|
return cache.get(file_unique_id)
|
|
|
|
|
|
def cache_sticker_description(
|
|
file_unique_id: str,
|
|
description: str,
|
|
emoji: str = "",
|
|
set_name: str = "",
|
|
) -> None:
|
|
"""
|
|
Store a sticker description in the cache.
|
|
|
|
Args:
|
|
file_unique_id: Telegram's stable sticker identifier.
|
|
description: Vision-generated description text.
|
|
emoji: Associated emoji (e.g. "😀").
|
|
set_name: Sticker set name if available.
|
|
"""
|
|
cache = _load_cache()
|
|
cache[file_unique_id] = {
|
|
"description": description,
|
|
"emoji": emoji,
|
|
"set_name": set_name,
|
|
"cached_at": time.time(),
|
|
}
|
|
_save_cache(cache)
|
|
|
|
|
|
def build_sticker_injection(
|
|
description: str,
|
|
emoji: str = "",
|
|
set_name: str = "",
|
|
) -> str:
|
|
"""
|
|
Build the warm-style injection text for a sticker description.
|
|
|
|
Returns a string like:
|
|
[The user sent a sticker 😀 from "MyPack"~ It shows: "A cat waving" (=^.w.^=)]
|
|
"""
|
|
context = ""
|
|
if set_name and emoji:
|
|
context = f" {emoji} from \"{set_name}\""
|
|
elif emoji:
|
|
context = f" {emoji}"
|
|
|
|
return f"[The user sent a sticker{context}~ It shows: \"{description}\" (=^.w.^=)]"
|
|
|
|
|
|
def build_animated_sticker_injection(emoji: str = "") -> str:
|
|
"""
|
|
Build injection text for animated/video stickers we can't analyze.
|
|
"""
|
|
if emoji:
|
|
return (
|
|
f"[The user sent an animated sticker {emoji}~ "
|
|
f"I can't see animated ones yet, but the emoji suggests: {emoji}]"
|
|
)
|
|
return "[The user sent an animated sticker~ I can't see animated ones yet]"
|