fix: preserve A2A transcript order

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,13 +175,41 @@ 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)
+				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
 			}
 		}
 	}
 }
 
+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 {

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"

go/core/v2/a2agateway/gateway.go

@@ -517,7 +517,11 @@ func (g *Gateway) prepareReply(ctx context.Context, instance *apiv1alpha1.AgentI
 	}
 	message.ContextID = stored.ContextID
 	attempt := *stored
-	attempt.History = append(append([]*a2atype.Message{}, stored.History...), message)
+	attempt.History = append([]*a2atype.Message{}, stored.History...)
+	if stored.Status.Message != nil {
+		attempt.History = append(attempt.History, stored.Status.Message)
+	}
+	attempt.History = append(attempt.History, message)
 	now := time.Now()
 	attempt.Status = a2atype.TaskStatus{State: a2atype.TaskStateSubmitted, Timestamp: &now}
 	if err := g.store.StoreAgentInstanceTaskEvent(ctx, instance.GetId(), &attempt, message, nil); err != nil {

go/core/v2/a2agateway/gateway_test.go

@@ -392,6 +392,25 @@ func TestGatewayContinuesInputRequiredTask(t *testing.T) {
 	}
 }
 
+func TestGatewayArchivesInputRequiredMessageBeforeReply(t *testing.T) {
+	question := a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.NewTextPart("Which database?"))
+	waiting := &a2atype.Task{
+		ID: "task-1", ContextID: gatewayTestID,
+		Status: a2atype.TaskStatus{State: a2atype.TaskStateInputRequired, Message: question},
+	}
+	reply := a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("PostgreSQL"))
+	reply.TaskID = waiting.ID
+	gateway := &Gateway{store: &gatewayTestStore{task: waiting}}
+
+	prepared, err := gateway.prepareReply(t.Context(), gatewayTestInstance(), &a2atype.SendMessageRequest{Message: reply})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(prepared.task.History) != 2 || prepared.task.History[0] != question || prepared.task.History[1] != reply {
+		t.Fatalf("history = %#v, want question followed by reply", prepared.task.History)
+	}
+}
+
 func TestGatewayClosesRuntimeAfterStreaming(t *testing.T) {
 	instance := gatewayTestInstance()
 	runtime := &gatewayTestRuntime{}

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

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,57 +764,28 @@ 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,
       taskId: task.id || undefined,
     });
   }
 
-  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<string, unknown> | undefined)?.[
-    HITL_EXTENSION_URI
-  ] as { type?: unknown } | undefined;
-  return carried?.type === "ask_user_response";
+  return messages;
 }