From 81d5dba9253c9eefe5612e0d36dc0fc4102949f8 Mon Sep 17 00:00:00 2001 From: Eitan Yarmush Date: Wed, 26 Aug 2026 20:56:01 +0000 Subject: [PATCH 1/2] fix: preserve A2A transcript order Signed-off-by: Eitan Yarmush --- go/adk/pkg/a2a/executor.go | 29 ++++++- go/adk/pkg/a2a/executor_test.go | 25 ++++++ ui/src/api/chat/a2aGrpcChatClient.test.ts | 21 +++++ ui/src/api/chat/a2aGrpcChatClient.ts | 36 +-------- ui/src/api/chat/transcriptOrder.test.ts | 89 ---------------------- ui/src/api/chat/transcriptOrder.ts | 93 ----------------------- 6 files changed, 76 insertions(+), 217 deletions(-) delete mode 100644 ui/src/api/chat/transcriptOrder.test.ts delete mode 100644 ui/src/api/chat/transcriptOrder.ts diff --git a/go/adk/pkg/a2a/executor.go b/go/adk/pkg/a2a/executor.go index 57d312007..23f26c833 100644 --- a/go/adk/pkg/a2a/executor.go +++ b/go/adk/pkg/a2a/executor.go @@ -34,7 +34,7 @@ type KAgentExecutorConfig struct { } // KAgentExecutor keeps kagent's request/session glue around the upstream ADK -// A2A executor. Event conversion and artifact streaming are delegated to ADK. +// A2A executor. type KAgentExecutor struct { builtin a2asrv.AgentExecutor sessionService adksession.Service @@ -105,7 +105,7 @@ func (u *userIDInterceptor) Before(ctx context.Context, callCtx *a2asrv.CallCont } // Execute applies kagent-specific request setup and delegates event generation -// to the upstream ADK executor, which streams output as artifact updates. +// to the upstream ADK executor. func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.ExecutorContext) iter.Seq2[a2atype.Event, error] { return func(yield func(a2atype.Event, error) bool) { if reqCtx.Message == nil { @@ -175,6 +175,15 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.ExecutorCon update.Status.Message.TaskID = update.TaskID update.Status.Message.ContextID = update.ContextID } + if update, ok := event.(*a2atype.TaskArtifactUpdateEvent); ok && artifactContainsToolEvent(update.Artifact) { + message := a2atype.NewMessageForTask(a2atype.MessageRoleAgent, update, update.Artifact.Parts...) + message.ID = string(update.Artifact.ID) + message.Extensions = update.Artifact.Extensions + message.Metadata = update.Artifact.Metadata + status := a2atype.NewStatusUpdateEvent(update, a2atype.TaskStateWorking, message) + status.Metadata = update.Metadata + event = status + } if !yield(event, err) { return } @@ -182,6 +191,22 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.ExecutorCon } } +func artifactContainsToolEvent(artifact *a2atype.Artifact) bool { + if artifact == nil { + return false + } + for _, part := range artifact.Parts { + if part == nil { + continue + } + partType, _ := ReadMetadataValue(part.Metadata, A2ADataPartMetadataTypeKey) + if partType == A2ADataPartMetadataTypeFunctionCall || partType == A2ADataPartMetadataTypeFunctionResponse { + return true + } + } + return false +} + // ensureSession ensures that a session exists for the given user and session ID. // If a session does not exist, it creates a new session with the given user and session ID. func (e *KAgentExecutor) ensureSession(ctx context.Context, message *a2atype.Message, userID, sessionID string) error { diff --git a/go/adk/pkg/a2a/executor_test.go b/go/adk/pkg/a2a/executor_test.go index 4de2dba6f..1bec175e5 100644 --- a/go/adk/pkg/a2a/executor_test.go +++ b/go/adk/pkg/a2a/executor_test.go @@ -198,6 +198,31 @@ func TestKAgentExecutor_PreservesContentBearingLastChunk(t *testing.T) { } } +func TestKAgentExecutor_EmitsToolEventsAsStatusMessages(t *testing.T) { + reqCtx := &a2asrv.ExecutorContext{TaskID: "task-1", ContextID: "ctx-1"} + reqCtx.Message = a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("hi")) + toolPart := a2atype.NewDataPart(map[string]any{PartKeyName: "search"}) + toolPart.Metadata = map[string]any{GetKAgentMetadataKey(A2ADataPartMetadataTypeKey): A2ADataPartMetadataTypeFunctionCall} + tool := a2atype.NewArtifactEvent(reqCtx, toolPart) + text := a2atype.NewArtifactEvent(reqCtx, a2atype.NewTextPart("done")) + executor := &KAgentExecutor{builtin: &recordingExecutor{events: []a2atype.Event{tool, text}}, logger: logr.Discard()} + + var got []a2atype.Event + for event, err := range executor.Execute(t.Context(), reqCtx) { + if err != nil { + t.Fatal(err) + } + got = append(got, event) + } + status, ok := got[0].(*a2atype.TaskStatusUpdateEvent) + if !ok || status.Status.State != a2atype.TaskStateWorking || status.Status.Message.ID != string(tool.Artifact.ID) || status.Status.Message.Parts[0] != toolPart { + t.Fatalf("tool event = %#v, want working status message", got[0]) + } + if got[1] != text { + t.Fatalf("text event = %#v, want original artifact", got[1]) + } +} + func TestKAgentExecutor_StreamsArtifactsThroughUpstreamExecutor(t *testing.T) { const ( appName = "test-app" diff --git a/ui/src/api/chat/a2aGrpcChatClient.test.ts b/ui/src/api/chat/a2aGrpcChatClient.test.ts index efef8ddd1..4fcbc8e32 100644 --- a/ui/src/api/chat/a2aGrpcChatClient.test.ts +++ b/ui/src/api/chat/a2aGrpcChatClient.test.ts @@ -554,6 +554,27 @@ describe("A2AGrpcChatClient.history", () => { }); } + it("replays protocol history before deliverable artifacts", async () => { + serveTasks([ + { + id: "task-1", + contextId: CONVERSATION.id, + status: { state: TaskState.COMPLETED, timestamp: { seconds: 1767225600n } }, + history: [ + { messageId: "u0", role: Role.USER, parts: [text("ask me a question")] }, + { messageId: "a0", role: Role.AGENT, parts: [data({ name: "ask_user" })] }, + { messageId: "a1", role: Role.AGENT, parts: [text("Which topic?")] }, + { messageId: "u1", role: Role.USER, parts: [text("Personal development")] }, + { messageId: "a2", role: Role.AGENT, parts: [data({ name: "ask_user" })] }, + ], + artifacts: [{ artifactId: "result", parts: [text("Thank you for sharing.")] }], + }, + ]); + + const { messages } = await new A2AGrpcChatClient().history(CONVERSATION); + expect(messages.map((message) => message.id)).toEqual(["u0", "a0", "a1", "u1", "a2", "result"]); + }); + it("keeps a tool call and its result apart when replaying", async () => { // Consecutive agent messages carrying data parts and no text at all: comparing // text made them look identical and dropped the result, so a replayed diff --git a/ui/src/api/chat/a2aGrpcChatClient.ts b/ui/src/api/chat/a2aGrpcChatClient.ts index 1b4aa7fb1..97b7a86c0 100644 --- a/ui/src/api/chat/a2aGrpcChatClient.ts +++ b/ui/src/api/chat/a2aGrpcChatClient.ts @@ -76,7 +76,6 @@ import { type PendingRequest, } from "./hitl"; import { agentInstanceShareToken } from "../shareToken"; -import { interleaveTaskMessages } from "./transcriptOrder"; import { serviceClient } from "../transport"; import type { ChatClient, @@ -765,43 +764,22 @@ export function messagesFromTask(task: A2ATask): ChatMessage[] { }); }; - /* - * Sorted into three, because the gateway hands back two lists with no way to - * interleave them — see `interleaveTaskMessages`, which does the inferring and - * carries the reasoning. - * - * The split has to happen here rather than after conversion: what marks a reader - * turn as an answer is the HITL metadata on the A2A message, and a `ChatMessage` - * does not carry it. - */ - const openingCount = messages.length; - const answerAt: number[] = []; for (const message of task.history) { - if (isAskUserResponse(message)) answerAt.push(messages.length); push(message); } if (task.status?.message) push(task.status.message); - const fromHistory = messages.slice(openingCount); - const answered = new Set(answerAt.map((index) => index - openingCount)); - const answers = fromHistory.filter((_, index) => answered.has(index)); - const opening = [ - ...messages.slice(0, openingCount), - ...fromHistory.filter((_, index) => !answered.has(index)), - ]; - // An artifact repeating text already in the history is the same reply arriving // twice, exactly as it is on a live stream. const shown = new Set(messages.map((message) => textOf(message.parts))); - const agent: ChatMessage[] = []; for (const artifact of task.artifacts) { const parts = toParts(artifact.parts); const body = textOf(parts); if (parts.length === 0 || (body !== "" && shown.has(body))) continue; - agent.push({ + messages.push({ // Derived, for the reason given against the message id above: an unnamed // artifact renamed on every read is an artifact the merge cannot recognise. - id: artifact.artifactId || `${task.id || "task"}-artifact-${messages.length + agent.length}`, + id: artifact.artifactId || `${task.id || "task"}-artifact-${messages.length}`, role: "agent", parts, createdAt, @@ -809,13 +787,5 @@ export function messagesFromTask(task: A2ATask): ChatMessage[] { }); } - return interleaveTaskMessages(opening, answers, agent); -} - -/** Whether a reader's turn is answering an `ask_user` rather than opening a task. */ -function isAskUserResponse(message: A2AMessage): boolean { - const carried = (message.metadata as Record | undefined)?.[ - HITL_EXTENSION_URI - ] as { type?: unknown } | undefined; - return carried?.type === "ask_user_response"; + return messages; } diff --git a/ui/src/api/chat/transcriptOrder.test.ts b/ui/src/api/chat/transcriptOrder.test.ts deleted file mode 100644 index c4da32c61..000000000 --- a/ui/src/api/chat/transcriptOrder.test.ts +++ /dev/null @@ -1,89 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { interleaveTaskMessages } from "./transcriptOrder"; -import type { ChatMessage } from "./types"; - -const text = (id: string, role: "user" | "agent", body: string): ChatMessage => ({ - id, - role, - parts: [{ kind: "text", text: body }], - createdAt: "2026-08-26T16:53:50.000Z", -}); - -const askUser = (id: string, kind: "tool_call" | "tool_result"): ChatMessage => ({ - id, - role: "agent", - parts: [{ kind: "data", dataKind: kind, data: { name: "ask_user", id: "call_1" } }], - createdAt: "2026-08-26T16:53:50.000Z", -}); - -/** - * The shape read back from a real cluster, which is what this exists for. - * - * One task, three reader turns in `history` and seven agent entries in `artifacts`: - * "ask me a question" opened it, then two `ask_user` rounds of call, pending result and - * answered result, then the closing reply. Concatenating the two lists — what this - * replaced — put both answers above every question. - */ -describe("interleaveTaskMessages", () => { - const opening = [text("u0", "user", "ask me a question")]; - const answers = [ - text("u1", "user", "Personal development"), - text("u2", "user", "none"), - ]; - const agent = [ - askUser("a0", "tool_call"), - askUser("a1", "tool_result"), - askUser("a2", "tool_result"), - askUser("a3", "tool_call"), - askUser("a4", "tool_result"), - askUser("a5", "tool_result"), - text("a6", "agent", "Thank you for sharing."), - ]; - - it("puts each answer in the round it answered", () => { - expect(interleaveTaskMessages(opening, answers, agent).map((m) => m.id)).toEqual([ - "u0", - "a0", // ask_user called - "a1", // result: pending - "u1", // the reader answers - "a2", // result: answered - "a3", - "a4", - "u2", - "a5", - "a6", // the closing reply stays last - ]); - }); - - it("leaves a task with no questions exactly as it was", () => { - const plain = [text("a0", "agent", "Hello!")]; - expect( - interleaveTaskMessages([text("u0", "user", "hi")], [], plain).map((m) => m.id), - ).toEqual(["u0", "a0"]); - }); - - it("keeps an answer it cannot place rather than dropping it", () => { - // A shape this does not recognise must degrade to the old behaviour, not lose a - // turn: a transcript missing a message is worse than one holding it out of order. - const orphan = [text("u9", "user", "answer to nothing")]; - expect( - interleaveTaskMessages([], orphan, [text("a0", "agent", "no questions here")]).map( - (m) => m.id, - ), - ).toEqual(["a0", "u9"]); - }); - - it("does not put an answer after a round's closing prose", () => { - // The last round runs to the end of the task, so "the end of the round" would put - // the answer below the agent's closing reply. It goes before the last result. - const oneRound = [ - askUser("a0", "tool_call"), - askUser("a1", "tool_result"), - askUser("a2", "tool_result"), - text("a3", "agent", "Thanks."), - ]; - expect( - interleaveTaskMessages([], [text("u1", "user", "yes")], oneRound).map((m) => m.id), - ).toEqual(["a0", "a1", "u1", "a2", "a3"]); - }); -}); diff --git a/ui/src/api/chat/transcriptOrder.ts b/ui/src/api/chat/transcriptOrder.ts deleted file mode 100644 index b9857c508..000000000 --- a/ui/src/api/chat/transcriptOrder.ts +++ /dev/null @@ -1,93 +0,0 @@ -import type { ChatMessage } from "./types"; - -/** - * Putting one task's messages back in the order they happened. - * - * `ListTasks` answers with two parallel lists: every message the reader sent is in - * `history`, every message the agent produced is in `artifacts`, and nothing in the - * response says how to interleave them. Concatenating them — which is what this used to - * do — renders every answer above the question it answers as soon as a task holds more - * than one reader turn, which `ask_user` guarantees. - * - * There is no key to do this properly with, and the gap is filed as - * https://github.com/kagent-dev/kagent/issues/2584: - * - * - every message in a task carries the task's single `status.timestamp`, so time puts - * them all in one bucket; - * - artifact ids are UUIDv7 and sort correctly, but `history` ids are minted by the - * client that sent the message and are UUIDv4, carrying no time at all; - * - the answer's own correlation id (`ask_user_response`) and the call's id (the - * model's `call_…`) do not refer to each other. - * - * So this is inference, and only from position. **Delete it when the gateway grows an - * ordering key** — one interleaved sequence, a sequence number, or a per-entry - * timestamp — rather than building on it. - * - * ## The inference - * - * A reader's answer belongs to the `ask_user` round it answered, and rounds are - * answered in the order they were asked: the runtime pairs answers to questions - * positionally, and an instance holds one non-terminal task at a time, so the *n*th - * answer answers the *n*th round. Within a round the answer goes immediately before - * that round's last `ask_user` result — the one reporting what was answered — which - * puts it after the call and after the result that was still pending. - * - * Anything left over is appended rather than dropped: a transcript missing a message is - * worse than one holding it in the wrong place, and a shape this does not recognise - * should degrade to the old behaviour rather than lose a turn. - */ -export function interleaveTaskMessages( - /** Reader turns that opened the task, in the order `history` gave them. */ - opening: readonly ChatMessage[], - /** Reader turns that answered an `ask_user`, in the order they were asked. */ - answers: readonly ChatMessage[], - /** Everything the agent produced, already time-ordered by its UUIDv7 ids. */ - agent: readonly ChatMessage[], -): ChatMessage[] { - const ordered: ChatMessage[] = [...opening]; - const unplaced = [...answers]; - - let at = 0; - while (at < agent.length) { - if (!isAskUserCall(agent[at])) { - ordered.push(agent[at]); - at += 1; - continue; - } - - // The round is this call and everything up to the next one. - let end = at + 1; - while (end < agent.length && !isAskUserCall(agent[end])) end += 1; - const round = agent.slice(at, end); - - // Before the last result of the round, which is the one carrying the answer. - // Falling back to the end of the round keeps the answer inside the round it - // belongs to even when the results are not the shape expected here. - let before = round.length; - for (let index = round.length - 1; index > 0; index -= 1) { - if (isAskUserResult(round[index])) { - before = index; - break; - } - } - - ordered.push(...round.slice(0, before)); - const answer = unplaced.shift(); - if (answer) ordered.push(answer); - ordered.push(...round.slice(before)); - at = end; - } - - // More answers than rounds recognised: keep them rather than lose them. - ordered.push(...unplaced); - return ordered; -} - -const isAskUserCall = (message: ChatMessage) => hasAskUser(message, "tool_call"); -const isAskUserResult = (message: ChatMessage) => hasAskUser(message, "tool_result"); - -function hasAskUser(message: ChatMessage, kind: "tool_call" | "tool_result"): boolean { - return message.parts.some( - (part) => part.kind === "data" && part.dataKind === kind && part.data.name === "ask_user", - ); -} From 0435b54ef3c5ff80e86331fd873627b5e6dc457b Mon Sep 17 00:00:00 2001 From: Eitan Yarmush Date: Thu, 27 Aug 2026 11:01:29 +0000 Subject: [PATCH 2/2] docs: explain ADK event workaround Signed-off-by: Eitan Yarmush --- go/adk/pkg/a2a/executor.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/go/adk/pkg/a2a/executor.go b/go/adk/pkg/a2a/executor.go index 23f26c833..7b82f80a0 100644 --- a/go/adk/pkg/a2a/executor.go +++ b/go/adk/pkg/a2a/executor.go @@ -175,6 +175,9 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.ExecutorCon update.Status.Message.TaskID = update.TaskID update.Status.Message.ContextID = update.ContextID } + // Work around upstream ADK's artifact-only event conversion: its callbacks can + // mutate an artifact but cannot replace it with another A2A event. Do this before + // a2a-go persists the update; remove when ADK exposes a general event converter. if update, ok := event.(*a2atype.TaskArtifactUpdateEvent); ok && artifactContainsToolEvent(update.Artifact) { message := a2atype.NewMessageForTask(a2atype.MessageRoleAgent, update, update.Artifact.Parts...) message.ID = string(update.Artifact.ID)