diff --git a/plugins/platforms/a2a/protocol.py b/plugins/platforms/a2a/protocol.py index f1522fccb1d5..99dd32096635 100644 --- a/plugins/platforms/a2a/protocol.py +++ b/plugins/platforms/a2a/protocol.py @@ -22,6 +22,7 @@ import json import copy +import hashlib import os import threading import time @@ -798,15 +799,26 @@ def _conv_dir() -> Path: def _safe_name(context_id: str) -> str: - return "".join(c for c in (context_id or "default") if c.isalnum() or c in "-_") or "default" + return hashlib.sha256(context_id.encode("utf-8")).hexdigest() + + +def _current_conv_dir() -> Path: + """Separate collision-resistant logs from ambiguous legacy filenames.""" + return _conv_dir() / "v2" def persist_message(context_id: str, role: str, text: str, task_id: str = "") -> None: """Append one message to the context's on-disk conversation log.""" try: - d = _conv_dir() + d = _current_conv_dir() d.mkdir(parents=True, exist_ok=True) - rec = {"ts": time.time(), "role": role, "text": text, "task_id": task_id} + rec = { + "ts": time.time(), + "context_id": context_id, + "role": role, + "text": text, + "task_id": task_id, + } with (d / f"{_safe_name(context_id)}.jsonl").open("a", encoding="utf-8") as fh: fh.write(json.dumps(rec, ensure_ascii=False) + "\n") except Exception: @@ -815,7 +827,7 @@ def persist_message(context_id: str, role: str, text: str, task_id: str = "") -> def load_conversation(context_id: str, limit: int = 50) -> list[dict]: """Load the last *limit* messages for a context (empty list if none).""" - path = _conv_dir() / f"{_safe_name(context_id)}.jsonl" + path = _current_conv_dir() / f"{_safe_name(context_id)}.jsonl" if not path.exists(): return [] out: list[dict] = [] @@ -836,7 +848,21 @@ def load_conversation(context_id: str, limit: int = 50) -> list[dict]: def list_conversations() -> list[str]: """Return known context-ids that have persisted conversations.""" - d = _conv_dir() + d = _current_conv_dir() if not d.exists(): return [] - return sorted(p.stem for p in d.glob("*.jsonl")) + context_ids: set[str] = set() + for path in d.glob("*.jsonl"): + try: + with path.open("r", encoding="utf-8") as fh: + for line in fh: + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + context_id = rec.get("context_id") + context_ids.add(context_id if isinstance(context_id, str) else path.stem) + break + except Exception: + continue + return sorted(context_ids) diff --git a/tests/plugins/test_a2a_plugin.py b/tests/plugins/test_a2a_plugin.py index 346284b27796..5abaf34f6871 100644 --- a/tests/plugins/test_a2a_plugin.py +++ b/tests/plugins/test_a2a_plugin.py @@ -431,6 +431,67 @@ def test_list_conversations(self, monkeypatch, tmp_path): protocol.persist_message("ctx-2", "user", "b", "t") assert set(protocol.list_conversations()) == {"ctx-1", "ctx-2"} + def test_distinct_context_ids_keep_separate_histories(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + protocol.persist_message("tenant/a", "user", "left-only", "task-left") + protocol.persist_message("tenanta", "user", "right-only", "task-right") + + assert [m["text"] for m in protocol.load_conversation("tenant/a")] == ["left-only"] + assert [m["text"] for m in protocol.load_conversation("tenanta")] == ["right-only"] + assert set(protocol.list_conversations()) == {"tenant/a", "tenanta"} + + def test_legacy_history_is_not_automatically_loaded(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + conversation_dir = tmp_path / "a2a_conversations" + conversation_dir.mkdir() + (conversation_dir / "ctx-legacy.jsonl").write_text( + '{"ts": 1, "role": "user", "text": "before-upgrade", "task_id": "old"}\n', + encoding="utf-8", + ) + + protocol.persist_message("ctx-legacy", "agent", "after-upgrade", "new") + + assert [m["text"] for m in protocol.load_conversation("ctx-legacy")] == [ + "after-upgrade" + ] + assert protocol.list_conversations() == ["ctx-legacy"] + + def test_ambiguous_legacy_history_is_not_loaded(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + conversation_dir = tmp_path / "a2a_conversations" + conversation_dir.mkdir() + (conversation_dir / "tenanta.jsonl").write_text( + '{"ts": 1, "role": "user", "text": "other-context", "task_id": "old"}\n', + encoding="utf-8", + ) + + assert protocol.load_conversation("tenant/a") == [] + assert protocol.load_conversation("tenanta") == [] + assert protocol.list_conversations() == [] + + def test_new_storage_does_not_collide_with_legacy_digest_name( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + conversation_dir = tmp_path / "a2a_conversations" + conversation_dir.mkdir() + legacy_context_id = ( + "584a11c77f870c594a006addb001b6af" + "9cfb03baca2d4081942d67565a14a245" + ) + (conversation_dir / f"{legacy_context_id}.jsonl").write_text( + '{"ts": 1, "role": "user", "text": "legacy-only", "task_id": "old"}\n', + encoding="utf-8", + ) + + protocol.persist_message("tenant/a", "agent", "new-only", "new") + + assert [m["text"] for m in protocol.load_conversation("tenant/a")] == [ + "new-only" + ] + assert protocol.load_conversation(legacy_context_id) == [] + assert protocol.list_conversations() == ["tenant/a"] + def test_load_missing_is_empty(self, monkeypatch, tmp_path): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) assert protocol.load_conversation("nope") == []