From 7001260cb758d907d0f96b072b7165a8a59d874f Mon Sep 17 00:00:00 2001 From: Kuang Mi Date: Mon, 3 Aug 2026 17:08:27 +0800 Subject: [PATCH] feat(a2a): inject conversation history on context resume (multi-turn) When a caller reuses a contextId, prepend the persisted conversation so the agent sees the full thread instead of only the latest message. - protocol.format_history(): render prior messages as 'role: text' lines, bounded by A2A_HISTORY_INJECTION_LIMIT (default 20, max 200, 0 disables) - adapter._prepare_task(): inject history before dispatch; audit and on-disk persistence keep the original (un-augmented) message so injected prefixes never accumulate in the log - tests: 9 new cases covering empty history, role rendering, limit/env, injection on resume, original-only persistence, no-duplication over three turns Closes #64982 (superseded by this rebased implementation). --- plugins/platforms/a2a/adapter.py | 12 +- plugins/platforms/a2a/protocol.py | 41 +++++++ tests/plugins/test_a2a_multiturn.py | 170 ++++++++++++++++++++++++++++ 3 files changed, 221 insertions(+), 2 deletions(-) create mode 100644 tests/plugins/test_a2a_multiturn.py diff --git a/plugins/platforms/a2a/adapter.py b/plugins/platforms/a2a/adapter.py index e54fa659ab6c..5598a1a78c7f 100644 --- a/plugins/platforms/a2a/adapter.py +++ b/plugins/platforms/a2a/adapter.py @@ -729,9 +729,17 @@ def _prepare_task(self, params: dict, peer: str, agent: Optional[dict] = None) - "Empty task — nothing to do.", created_at=rec["created_iso"], ), None + # Multi-turn: when a caller reuses a contextId, prepend the prior + # conversation so the agent sees the full thread. The original + # message (not the augmented copy) is what gets audited and + # persisted, so history never accumulates injected prefixes. + original_text = text + history = protocol.format_history(context_id) + if history: + text = history + text framed = security.wrap_inbound(peer, text) - security.audit("inbound", peer, task_id, text) - protocol.persist_message(context_id, "user", text, task_id) + security.audit("inbound", peer, task_id, original_text) + protocol.persist_message(context_id, "user", original_text, task_id) protocol.metrics.inbound_total += 1 rec = self.tasks.create(task_id, context_id, peer, *self._scope_for_agent(agent)) diff --git a/plugins/platforms/a2a/protocol.py b/plugins/platforms/a2a/protocol.py index f1522fccb1d5..35c18ddcfb5a 100644 --- a/plugins/platforms/a2a/protocol.py +++ b/plugins/platforms/a2a/protocol.py @@ -74,6 +74,9 @@ _DEFAULT_MAX_PINGPONG = 5 _HARD_MAX_PINGPONG = 20 +_DEFAULT_HISTORY_LIMIT = 20 +_HARD_HISTORY_LIMIT = 200 + def max_pingpong_turns() -> int: try: @@ -83,6 +86,20 @@ def max_pingpong_turns() -> int: return _DEFAULT_MAX_PINGPONG +def history_injection_limit() -> int: + """Max prior messages prepended when a caller resumes a context_id. + + The agent sees the full thread on multi-turn continuations instead of + only the latest message. 0 disables injection. Bounded so a long-lived + context cannot balloon a single prompt. + """ + try: + v = int(os.getenv("A2A_HISTORY_INJECTION_LIMIT", str(_DEFAULT_HISTORY_LIMIT))) + return max(0, min(v, _HARD_HISTORY_LIMIT)) + except (ValueError, TypeError): + return _DEFAULT_HISTORY_LIMIT + + def now_iso() -> str: """ISO 8601 UTC timestamp with millisecond precision (A2A v1.0).""" return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" @@ -834,6 +851,30 @@ def load_conversation(context_id: str, limit: int = 50) -> list[dict]: return out[-limit:] +def format_history(context_id: str, limit: Optional[int] = None) -> str: + """Render prior messages of a resumed context as a plain-text prefix. + + Returns "" when the context has no history yet (or when injection is + disabled via ``limit <= 0``). The adapter prepends this to an inbound + message when a caller reuses a ``contextId``, so the agent sees the + full thread (multi-turn) instead of only the latest message. Lines are + ``role: text`` with the A2A roles ``user`` / ``assistant``; non-user + roles are treated as assistant output. + """ + limit = limit if limit is not None else history_injection_limit() + if limit <= 0: + return "" + recs = load_conversation(context_id, limit=limit) + if not recs: + return "" + lines = [] + for rec in recs: + role = rec.get("role") + label = "user" if role == "user" else "assistant" + lines.append(f"{label}: {rec.get('text', '')}") + return "\n".join(lines) + "\n\n" + + def list_conversations() -> list[str]: """Return known context-ids that have persisted conversations.""" d = _conv_dir() diff --git a/tests/plugins/test_a2a_multiturn.py b/tests/plugins/test_a2a_multiturn.py new file mode 100644 index 000000000000..0253fd90f5f6 --- /dev/null +++ b/tests/plugins/test_a2a_multiturn.py @@ -0,0 +1,170 @@ +"""Multi-turn history injection tests for the A2A platform plugin. + +When a caller reuses a ``contextId`` (multi-turn continuation), the adapter +prepends the persisted conversation so the agent sees the full thread. The +original message — not the augmented copy — is what gets audited and +persisted, so injected history never accumulates in the on-disk log. +""" + +from __future__ import annotations + +import asyncio +from concurrent.futures import Future +from types import SimpleNamespace + +import pytest + +from plugins.platforms.a2a import protocol, security + + +def _send_params(text: str, context_id: str) -> dict: + return { + "message": { + "role": "user", + "parts": [{"text": text, "mediaType": "text/plain"}], + }, + "contextId": context_id, + } + + +class TestFormatHistory: + def test_empty_when_no_history(self, monkeypatch, tmp_path): + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + assert protocol.format_history("ctx-none", limit=20) == "" + + def test_roles_rendered_user_assistant(self, monkeypatch, tmp_path): + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + protocol.persist_message("ctx-a", "user", "hello") + protocol.persist_message("ctx-a", "agent", "hi there") + assert protocol.format_history("ctx-a", limit=20) == ( + "user: hello\nassistant: hi there\n\n" + ) + + def test_limit_keeps_only_recent(self, monkeypatch, tmp_path): + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + for i in range(5): + protocol.persist_message("ctx-b", "user", f"msg-{i}") + out = protocol.format_history("ctx-b", limit=2) + assert "msg-0" not in out + assert "msg-4" in out + + def test_default_limit_from_env(self, monkeypatch, tmp_path): + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + monkeypatch.setenv("A2A_HISTORY_INJECTION_LIMIT", "3") + for i in range(5): + protocol.persist_message("ctx-c", "user", f"msg-{i}") + out = protocol.format_history("ctx-c") + assert "msg-0" not in out + assert "msg-4" in out + + def test_zero_limit_disables(self, monkeypatch, tmp_path): + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + protocol.persist_message("ctx-d", "user", "hello") + assert protocol.format_history("ctx-d", limit=0) == "" + + +def _bare_adapter(): + from plugins.platforms.a2a.adapter import A2AAdapter + from gateway.config import PlatformConfig + + return A2AAdapter(PlatformConfig(enabled=True)) + + +class TestInboundHistoryInjection: + def _drive(self, adapter, monkeypatch): + """Run _prepare_task with dispatch captured; return the event text list.""" + seen_events = [] + + async def fake_handle(event): + seen_events.append(event) + + adapter.handle_message = fake_handle + adapter._message_handler = lambda event: None + adapter._loop = asyncio.new_event_loop() + + captured_coros = [] + + def fake_schedule(coro, loop): + captured_coros.append(coro) + return SimpleNamespace(done=lambda: True) + + monkeypatch.setattr(asyncio, "run_coroutine_threadsafe", fake_schedule) + return seen_events, captured_coros + + def test_new_context_no_injection(self, monkeypatch, tmp_path): + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + adapter = _bare_adapter() + seen, coros = self._drive(adapter, monkeypatch) + + task, pending = adapter._prepare_task( + _send_params("first question", "ctx-new"), peer="peer-x" + ) + assert pending is not None + asyncio.run(coros[0]) + text = seen[0].text + assert "first question" in text + assert not text.startswith("user:") + + def test_resume_injects_history_into_agent_text(self, monkeypatch, tmp_path): + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + protocol.persist_message("ctx-mt", "user", "first question") + protocol.persist_message("ctx-mt", "agent", "first answer") + + adapter = _bare_adapter() + seen, coros = self._drive(adapter, monkeypatch) + + task, pending = adapter._prepare_task( + _send_params("second question", "ctx-mt"), peer="peer-x" + ) + assert pending is not None + asyncio.run(coros[0]) + text = seen[0].text + assert "first question" in text + assert "first answer" in text + assert "second question" in text + assert text.index("first question") < text.index("second question") + + def test_persist_and_audit_store_original_not_augmented( + self, monkeypatch, tmp_path + ): + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + protocol.persist_message("ctx-keep", "user", "first question") + protocol.persist_message("ctx-keep", "agent", "first answer") + + audited = [] + monkeypatch.setattr(security, "audit", lambda *a, **k: audited.append(a)) + adapter = _bare_adapter() + seen, coros = self._drive(adapter, monkeypatch) + + task, pending = adapter._prepare_task( + _send_params("second question", "ctx-keep"), peer="peer-x" + ) + assert pending is not None + asyncio.run(coros[0]) + + # Agent saw the augmented text (wrap_inbound adds its own prefix)… + assert "user: first question" in seen[0].text + assert "second question" in seen[0].text + # …but the persisted log and audit trail carry only the original. + recs = protocol.load_conversation("ctx-keep", limit=50) + assert recs[-1]["role"] == "user" + assert recs[-1]["text"] == "second question" + assert any("second question" in a[3] for a in audited) + assert all("first question" not in a[3] for a in audited) + + def test_resume_does_not_duplicate_injected_prefix(self, monkeypatch, tmp_path): + """A third turn must not re-inject the already-injected prefix.""" + monkeypatch.setattr(protocol, "_conv_dir", lambda: tmp_path) + adapter = _bare_adapter() + seen, coros = self._drive(adapter, monkeypatch) + + for i, text in enumerate(["q1", "q2", "q3"]): + task, pending = adapter._prepare_task( + _send_params(text, "ctx-loop"), peer="peer-x" + ) + assert pending is not None + asyncio.run(coros[-1]) + + # After three turns the on-disk log holds exactly the three originals. + recs = protocol.load_conversation("ctx-loop", limit=50) + assert [r["text"] for r in recs] == ["q1", "q2", "q3"]