fix(a2a): preserve authoritative text replacements

ksadk/a2a/executor.py

@@ -342,8 +342,9 @@ async def _run_streaming(
                 if self.include_reasoning:
                     await artifacts.push("thinking", text)
                 continue
-            output_text += text
-            await artifacts.push("text", text)
+            replace = bool(isinstance(chunk, dict) and chunk.get("replace"))
+            output_text = text if replace else output_text + text
+            await artifacts.push("text", text, replace_snapshot=replace)
 
         await artifacts.close()
         return output_text
@@ -444,7 +445,8 @@ async def _run_runtime(
                 reasoning_text += text
                 await artifacts.push("thinking", text)
                 continue
-            # TEXT_COMPLETED 是累计全文,去重只发新增 suffix;TEXT_DELTA 是增量直接透传。
+            # TEXT_COMPLETED 是累计全文,去重只发新增 suffix;TEXT_DELTA 默认是增量,
+            # 但 runner 显式标记 replace 时是权威快照。
             if event.event_type == EventType.TEXT_COMPLETED:
                 if not output_text:
                     delta = text
@@ -460,8 +462,8 @@ async def _run_runtime(
                     replace_snapshot = True
             else:
                 delta = text
-                output_text += text
-                replace_snapshot = False
+                replace_snapshot = bool(event.payload.get("replace"))
+                output_text = text if replace_snapshot else output_text + text
             if not delta:
                 continue
             await artifacts.push("text", delta, replace_snapshot=replace_snapshot)

ksadk/runtime/runner_adapter.py

@@ -597,7 +597,7 @@ async def _map_runner_stream(
             else nullcontext()
         )
         runner_name = _runner_name(self._runner)
-        accumulated_output: list[str] = []
+        accumulated_output = ""
         usage: dict[str, Any] = {}
         runner_gen: Optional[AsyncIterator[Any]] = None
         async with _conversation_span_scope(runner_name) as span:
@@ -664,7 +664,10 @@ async def _map_runner_stream(
                                     chunk.get("delta") or chunk.get("output") or chunk.get("data")
                                 )
                                 if text:
-                                    accumulated_output.append(text)
+                                    if chunk_type == "final" or chunk.get("replace"):
+                                        accumulated_output = text
+                                    else:
+                                        accumulated_output += text
                             raw_usage = chunk.get("usage")
                             if isinstance(raw_usage, dict):
                                 usage.update(raw_usage)
@@ -677,7 +680,7 @@ async def _map_runner_stream(
                             yield self._event(handle, event_type, payload)
             finally:
                 if accumulated_output:
-                    _set_conversation_output_attributes(span, "".join(accumulated_output))
+                    _set_conversation_output_attributes(span, accumulated_output)
                 _set_conversation_usage_attributes(span, usage)
                 if runner_gen is not None:
                     aclose = getattr(runner_gen, "aclose", None)
@@ -840,7 +843,10 @@ def _chunk_to_event(
         text = self._coerce(chunk.get("delta") or chunk.get("output") or chunk.get("data"))
         if not text:
             return None
-        return self._event(handle, EventType.TEXT_DELTA, {"text": text}, phase="commentary")
+        payload: dict[str, Any] = {"text": text}
+        if chunk.get("replace"):
+            payload["replace"] = True
+        return self._event(handle, EventType.TEXT_DELTA, payload, phase="commentary")
 
     def _event(
         self,

tests/test_a2a_text_replacement.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from types import SimpleNamespace
+from typing import Any
+
+import httpx
+import pytest
+from fastapi import FastAPI
+
+from ksadk.a2a import A2ARuntimeTaskAdapter, add_a2a_protocol_routes
+from ksadk.a2a.executor import A2ARuntimeExecutor
+from ksadk.a2a.langgraph import stream_a2a_agent_to_writer
+from ksadk.a2a.routes import A2AConfig
+from ksadk.runtime.runner_adapter import RunnerRuntimeAdapter
+
+
+class _RecordingUpdater:
+    def __init__(self) -> None:
+        self.artifacts: list[dict[str, Any]] = []
+
+    async def add_artifact(self, **kwargs: Any) -> None:
+        self.artifacts.append(kwargs)
+
+
+class _ReplacingRunner:
+    async def stream(self, _runner_input: dict[str, Any]) -> AsyncIterator[dict[str, Any]]:
+        yield {"type": "text", "delta": "旧答"}
+        yield {"type": "text", "delta": "新答", "replace": True}
+        yield {"type": "final", "output": "新答"}
+
+
+@pytest.mark.asyncio
+async def test_runner_text_replacement_is_marked_as_authoritative_snapshot() -> None:
+    runner = _ReplacingRunner()
+    executor = A2ARuntimeExecutor(runner=runner)
+    updater = _RecordingUpdater()
+
+    output = await executor._run_streaming(  # noqa: SLF001
+        SimpleNamespace(task_id="task-1"),
+        updater,  # type: ignore[arg-type]
+        runner.stream,
+        {},
+    )
+
+    assert output == "新答"
+    assert [item["parts"][0].text for item in updater.artifacts] == ["旧答", "新答"]
+    assert updater.artifacts[-1]["append"] is False
+    assert updater.artifacts[-1]["parts"][0].metadata["ksadk_output_snapshot"] is True
+
+
+@pytest.mark.asyncio
+async def test_a2a_server_round_trip_preserves_text_replacement(tmp_path: Any) -> None:
+    runner = _ReplacingRunner()
+    app = FastAPI()
+    add_a2a_protocol_routes(
+        app,
+        runner,
+        A2AConfig(
+            enabled=True,
+            base_url="http://testserver",
+            agent_name="replacing-agent",
+            task_store_dsn=f"sqlite+aiosqlite:///{tmp_path}/tasks.db",
+        ),
+        task_adapter=A2ARuntimeTaskAdapter(
+            RunnerRuntimeAdapter(runner, runtime_type="test"),  # type: ignore[arg-type]
+            runtime_type="test",
+        ),
+    )
+    written: list[dict[str, Any]] = []
+    transport = httpx.ASGITransport(app=app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+        output = await stream_a2a_agent_to_writer(
+            "http://testserver",
+            "问题",
+            writer=written.append,
+            httpx_client=client,
+        )
+
+    assert output == "新答"
+    assert written == [
+        {"type": "text", "delta": "旧答", "replace": False},
+        {"type": "text", "delta": "新答", "replace": True},
+    ]