fix(a2a): isolate persisted context histories

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)

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") == []