mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-04-25 00:51:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge b18725fd76 into fd10463069

Browse source
This commit is contained in:
Es1la 2026-04-24 17:29:19 -05:00 • committed by GitHub
commit 609be198c1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 64 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

9
gateway/platforms/feishu.py
View file

@ -3784,7 +3784,14 @@ class FeishuAdapter(BasePlatformAdapter):
if isinstance(seen_data, list):
entries: Dict[str, float] = {str(item).strip(): 0.0 for item in seen_data if str(item).strip()}
elif isinstance(seen_data, dict):
entries = {k: float(v) for k, v in seen_data.items() if isinstance(k, str) and k.strip()}
entries = {}
for key, value in seen_data.items():
if not isinstance(key, str) or not key.strip():
continue
try:
entries[key] = float(value)
except (TypeError, ValueError):
continue
else:
return
# Filter out TTL-expired entries (entries saved with ts=0.0 are treated as immortal

84
tests/gateway/test_feishu.py
View file

@ -3068,60 +3068,88 @@ class TestWebhookSecurity(unittest.TestCase):
class TestDedupTTL(unittest.TestCase):
"""Tests for TTL-aware deduplication."""
@patch.dict(os.environ, {}, clear=True)
def test_duplicate_within_ttl_is_rejected(self):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter
adapter = FeishuAdapter(PlatformConfig())
with patch.object(adapter, "_persist_seen_message_ids"):
adapter._seen_message_ids = {"om_dup": time.time()}
adapter._seen_message_order = ["om_dup"]
self.assertTrue(adapter._is_duplicate("om_dup"))
with tempfile.TemporaryDirectory() as temp_home, \
patch.dict(os.environ, {"HERMES_HOME": temp_home}, clear=True):
adapter = FeishuAdapter(PlatformConfig())
with patch.object(adapter, "_persist_seen_message_ids"):
adapter._seen_message_ids = {"om_dup": time.time()}
adapter._seen_message_order = ["om_dup"]
self.assertTrue(adapter._is_duplicate("om_dup"))
@patch.dict(os.environ, {}, clear=True)
def test_expired_entry_is_not_considered_duplicate(self):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter, _FEISHU_DEDUP_TTL_SECONDS
adapter = FeishuAdapter(PlatformConfig())
# Plant an entry that expired well past the TTL.
stale_ts = time.time() - _FEISHU_DEDUP_TTL_SECONDS - 60
adapter._seen_message_ids = {"om_old": stale_ts}
adapter._seen_message_order = ["om_old"]
with patch.object(adapter, "_persist_seen_message_ids"):
self.assertFalse(adapter._is_duplicate("om_old"))
with tempfile.TemporaryDirectory() as temp_home, \
patch.dict(os.environ, {"HERMES_HOME": temp_home}, clear=True):
adapter = FeishuAdapter(PlatformConfig())
# Plant an entry that expired well past the TTL.
stale_ts = time.time() - _FEISHU_DEDUP_TTL_SECONDS - 60
adapter._seen_message_ids = {"om_old": stale_ts}
adapter._seen_message_order = ["om_old"]
with patch.object(adapter, "_persist_seen_message_ids"):
self.assertFalse(adapter._is_duplicate("om_old"))
@patch.dict(os.environ, {}, clear=True)
def test_persist_saves_timestamps_as_dict(self):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter
adapter = FeishuAdapter(PlatformConfig())
ts = time.time()
adapter._seen_message_ids = {"om_ts1": ts}
adapter._seen_message_order = ["om_ts1"]
with tempfile.TemporaryDirectory() as tmpdir:
adapter._dedup_state_path = Path(tmpdir) / "dedup.json"
adapter._persist_seen_message_ids()
saved = json.loads(adapter._dedup_state_path.read_text())
with patch.dict(os.environ, {"HERMES_HOME": tmpdir}, clear=True):
adapter = FeishuAdapter(PlatformConfig())
ts = time.time()
adapter._seen_message_ids = {"om_ts1": ts}
adapter._seen_message_order = ["om_ts1"]
adapter._dedup_state_path = Path(tmpdir) / "dedup.json"
adapter._persist_seen_message_ids()
saved = json.loads(adapter._dedup_state_path.read_text())
self.assertIsInstance(saved["message_ids"], dict)
self.assertAlmostEqual(saved["message_ids"]["om_ts1"], ts, places=1)
@patch.dict(os.environ, {}, clear=True)
def test_load_backward_compat_list_format(self):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter
adapter = FeishuAdapter(PlatformConfig())
with tempfile.TemporaryDirectory() as tmpdir:
path = Path(tmpdir) / "dedup.json"
path.write_text(json.dumps({"message_ids": ["om_a", "om_b"]}), encoding="utf-8")
adapter._dedup_state_path = path
adapter._load_seen_message_ids()
with patch.dict(os.environ, {"HERMES_HOME": tmpdir}, clear=True):
adapter = FeishuAdapter(PlatformConfig())
path = Path(tmpdir) / "dedup.json"
path.write_text(json.dumps({"message_ids": ["om_a", "om_b"]}), encoding="utf-8")
adapter._dedup_state_path = path
adapter._load_seen_message_ids()
self.assertIn("om_a", adapter._seen_message_ids)
self.assertIn("om_b", adapter._seen_message_ids)
def test_load_skips_invalid_timestamp_entries(self):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter
with tempfile.TemporaryDirectory() as tmpdir:
with patch.dict(os.environ, {"HERMES_HOME": tmpdir}, clear=True):
adapter = FeishuAdapter(PlatformConfig())
valid_ts = time.time()
path = Path(tmpdir) / "dedup.json"
path.write_text(
json.dumps({
"message_ids": {
"om_good": valid_ts,
"om_bad": "not-a-number",
"om_null": None,
}
}),
encoding="utf-8",
)
adapter._dedup_state_path = path
adapter._load_seen_message_ids()
self.assertIn("om_good", adapter._seen_message_ids)
self.assertNotIn("om_bad", adapter._seen_message_ids)
self.assertNotIn("om_null", adapter._seen_message_ids)
class TestGroupMentionAtAll(unittest.TestCase):
"""Tests for @_all (Feishu @everyone) group mention routing."""